src/util/scheduler.c

   1 /*
   2       This file is part of GNUnet
   3       (C) 2009, 2011 Christian Grothoff (and other contributing authors)
   4
   5       GNUnet is free software; you can redistribute it and/or modify
   6       it under the terms of the GNU General Public License as published
   7       by the Free Software Foundation; either version 2, or (at your
   8       option) any later version.
   9
  10       GNUnet is distributed in the hope that it will be useful, but
  11       WITHOUT ANY WARRANTY; without even the implied warranty of
  12       MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
  13       General Public License for more details.
  14
  15       You should have received a copy of the GNU General Public License
  16       along with GNUnet; see the file COPYING.  If not, write to the
  17       Free Software Foundation, Inc., 59 Temple Place - Suite 330,
  18       Boston, MA 02111-1307, USA.
  19  */
  20
  21 /**
  22  * @file util/scheduler.c
  23  * @brief schedule computations using continuation passing style
  24  * @author Christian Grothoff
  25  */
  26 #include "platform.h"
  27 #include "gnunet_common.h"
  28 #include "gnunet_os_lib.h"
  29 #include "gnunet_scheduler_lib.h"
  30 #include "gnunet_signal_lib.h"
  31 #include "gnunet_time_lib.h"
  32 #include "disk.h"
  33 #ifdef LINUX
  34 #include "execinfo.h"
  35
  36
  37 /**
  38  * Use lsof to generate file descriptor reports on select error?
  39  * (turn off for stable releases).
  40  */
  41 #define USE_LSOF GNUNET_NO
  42
  43 /**
  44  * Obtain trace information for all scheduler calls that schedule tasks.
  45  */
  46 #define EXECINFO GNUNET_NO
  47
  48 /**
  49  * Check each file descriptor before adding
  50  */
  51 #define DEBUG_FDS GNUNET_NO
  52
  53 /**
  54  * Depth of the traces collected via EXECINFO.
  55  */
  56 #define MAX_TRACE_DEPTH 50
  57 #endif
  58
  59 #define DEBUG_TASKS GNUNET_NO
  60
  61 /**
  62  * Should we figure out which tasks are delayed for a while
  63  * before they are run? (Consider using in combination with EXECINFO).
  64  */
  65 #define PROFILE_DELAYS GNUNET_NO
  66
  67 /**
  68  * Task that were in the queue for longer than this are reported if
  69  * PROFILE_DELAYS is active.
  70  */
  71 #define DELAY_THRESHOLD GNUNET_TIME_UNIT_SECONDS
  72
  73 /**
  74  * Linked list of pending tasks.
  75  */
  76 struct Task
  77 {
  78   /**
  79    * This is a linked list.
  80    */
  81   struct Task *next;
  82
  83   /**
  84    * Function to run when ready.
  85    */
  86   GNUNET_SCHEDULER_Task callback;
  87
  88   /**
  89    * Closure for the callback.
  90    */
  91   void *callback_cls;
  92
  93   /**
  94    * Set of file descriptors this task is waiting
  95    * for for reading.  Once ready, this is updated
  96    * to reflect the set of file descriptors ready
  97    * for operation.
  98    */
  99   struct GNUNET_NETWORK_FDSet *read_set;
 100
 101   /**
 102    * Set of file descriptors this task is waiting for for writing.
 103    * Once ready, this is updated to reflect the set of file
 104    * descriptors ready for operation.
 105    */
 106   struct GNUNET_NETWORK_FDSet *write_set;
 107
 108   /**
 109    * Unique task identifier.
 110    */
 111   GNUNET_SCHEDULER_TaskIdentifier id;
 112
 113   /**
 114    * Identifier of a prerequisite task.
 115    */
 116   GNUNET_SCHEDULER_TaskIdentifier prereq_id;
 117
 118   /**
 119    * Absolute timeout value for the task, or
 120    * GNUNET_TIME_UNIT_FOREVER_ABS for "no timeout".
 121    */
 122   struct GNUNET_TIME_Absolute timeout;
 123
 124 #if PROFILE_DELAYS
 125   /**
 126    * When was the task scheduled?
 127    */
 128   struct GNUNET_TIME_Absolute start_time;
 129 #endif
 130
 131   /**
 132    * Why is the task ready?  Set after task is added to ready queue.
 133    * Initially set to zero.  All reasons that have already been
 134    * satisfied (i.e.  read or write ready) will be set over time.
 135    */
 136   enum GNUNET_SCHEDULER_Reason reason;
 137
 138   /**
 139    * Task priority.
 140    */
 141   enum GNUNET_SCHEDULER_Priority priority;
 142
 143   /**
 144    * Set if we only wait for reading from a single FD, otherwise -1.
 145    */
 146   int read_fd;
 147
 148   /**
 149    * Set if we only wait for writing to a single FD, otherwise -1.
 150    */
 151   int write_fd;
 152
 153   /**
 154    * Should the existence of this task in the queue be counted as
 155    * reason to not shutdown the scheduler?
 156    */
 157   int lifeness;
 158
 159 #if EXECINFO
 160   /**
 161    * Array of strings which make up a backtrace from the point when this
 162    * task was scheduled (essentially, who scheduled the task?)
 163    */
 164   char **backtrace_strings;
 165
 166   /**
 167    * Size of the backtrace_strings array
 168    */
 169   int num_backtrace_strings;
 170 #endif
 171
 172
 173 };
 174
 175
 176 /**
 177  * List of tasks waiting for an event.
 178  */
 179 static struct Task *pending;
 180
 181 /**
 182  * List of tasks waiting ONLY for a timeout event.
 183  * Sorted by timeout (earliest first).  Used so that
 184  * we do not traverse the list of these tasks when
 185  * building select sets (we just look at the head
 186  * to determine the respective timeout ONCE).
 187  */
 188 static struct Task *pending_timeout;
 189
 190 /**
 191  * Last inserted task waiting ONLY for a timeout event.
 192  * Used to (heuristically) speed up insertion.
 193  */
 194 static struct Task *pending_timeout_last;
 195
 196 /**
 197  * ID of the task that is running right now.
 198  */
 199 static struct Task *active_task;
 200
 201 /**
 202  * List of tasks ready to run right now,
 203  * grouped by importance.
 204  */
 205 static struct Task *ready[GNUNET_SCHEDULER_PRIORITY_COUNT];
 206
 207 /**
 208  * Identity of the last task queued.  Incremented for each task to
 209  * generate a unique task ID (it is virtually impossible to start
 210  * more than 2^64 tasks during the lifetime of a process).
 211  */
 212 static GNUNET_SCHEDULER_TaskIdentifier last_id;
 213
 214 /**
 215  * Highest number so that all tasks with smaller identifiers
 216  * have already completed.  Also the lowest number of a task
 217  * still waiting to be executed.
 218  */
 219 static GNUNET_SCHEDULER_TaskIdentifier lowest_pending_id;
 220
 221 /**
 222  * Number of tasks on the ready list.
 223  */
 224 static unsigned int ready_count;
 225
 226 /**
 227  * How many tasks have we run so far?
 228  */
 229 static unsigned long long tasks_run;
 230
 231 /**
 232  * Priority of the task running right now.  Only
 233  * valid while a task is running.
 234  */
 235 static enum GNUNET_SCHEDULER_Priority current_priority;
 236
 237 /**
 238  * Priority of the highest task added in the current select
 239  * iteration.
 240  */
 241 static enum GNUNET_SCHEDULER_Priority max_priority_added;
 242
 243 /**
 244  * Value of the 'lifeness' flag for the current task.
 245  */
 246 static int current_lifeness;
 247
 248 /**
 249  * Function to use as a select() in the scheduler.
 250  * If NULL, we use GNUNET_NETWORK_socket_select ().
 251  */
 252 static GNUNET_SCHEDULER_select scheduler_select;
 253
 254 /**
 255  * Closure for 'scheduler_select'.
 256  */
 257 static void *scheduler_select_cls;
 258
 259 /**
 260  * Sets the select function to use in the scheduler (scheduler_select).
 261  *
 262  * @param new_select new select function to use
 263  * @return previously used select function, NULL for default
 264  */
 265 void
 266 GNUNET_SCHEDULER_set_select (GNUNET_SCHEDULER_select new_select,
 267                              void *new_select_cls)
 268 {
 269   scheduler_select = new_select;
 270   scheduler_select_cls = new_select_cls;
 271 }
 272
 273
 274 /**
 275  * Check that the given priority is legal (and return it).
 276  *
 277  * @param p priority value to check
 278  * @return p on success, 0 on error
 279  */
 280 static enum GNUNET_SCHEDULER_Priority
 281 check_priority (enum GNUNET_SCHEDULER_Priority p)
 282 {
 283   if ((p >= 0) && (p < GNUNET_SCHEDULER_PRIORITY_COUNT))
 284     return p;
 285   GNUNET_assert (0);
 286   return 0;                     /* make compiler happy */
 287 }
 288
 289
 290 /**
 291  * Is a task with this identifier still pending?  Also updates
 292  * "lowest_pending_id" as a side-effect (for faster checks in the
 293  * future), but only if the return value is "GNUNET_NO" (and
 294  * the "lowest_pending_id" check failed).
 295  *
 296  * @param id which task are we checking for
 297  * @return GNUNET_YES if so, GNUNET_NO if not
 298  */
 299 static int
 300 is_pending (GNUNET_SCHEDULER_TaskIdentifier id)
 301 {
 302   struct Task *pos;
 303   enum GNUNET_SCHEDULER_Priority p;
 304   GNUNET_SCHEDULER_TaskIdentifier min;
 305
 306   if (id < lowest_pending_id)
 307     return GNUNET_NO;
 308   min = -1;                     /* maximum value */
 309   pos = pending;
 310   while (pos != NULL)
 311     {
 312       if (pos->id == id)
 313         return GNUNET_YES;
 314       if (pos->id < min)
 315         min = pos->id;
 316       pos = pos->next;
 317     }
 318   pos = pending_timeout;
 319   while (pos != NULL)
 320     {
 321       if (pos->id == id)
 322         return GNUNET_YES;
 323       if (pos->id < min)
 324         min = pos->id;
 325       pos = pos->next;
 326     }
 327   for (p = 0; p < GNUNET_SCHEDULER_PRIORITY_COUNT; p++)
 328     {
 329       pos = ready[p];
 330       while (pos != NULL)
 331         {
 332           if (pos->id == id)
 333             return GNUNET_YES;
 334           if (pos->id < min)
 335             min = pos->id;
 336           pos = pos->next;
 337         }
 338     }
 339   lowest_pending_id = min;
 340   return GNUNET_NO;
 341 }
 342
 343
 344 /**
 345  * Update all sets and timeout for select.
 346  *
 347  * @param rs read-set, set to all FDs we would like to read (updated)
 348  * @param ws write-set, set to all FDs we would like to write (updated)
 349  * @param timeout next timeout (updated)
 350  */
 351 static void
 352 update_sets (struct GNUNET_NETWORK_FDSet *rs,
 353              struct GNUNET_NETWORK_FDSet *ws,
 354              struct GNUNET_TIME_Relative *timeout)
 355 {
 356   struct Task *pos;
 357   struct GNUNET_TIME_Absolute now;
 358   struct GNUNET_TIME_Relative to;
 359
 360   now = GNUNET_TIME_absolute_get ();
 361   pos = pending_timeout;
 362   if (pos != NULL)
 363     {
 364       to = GNUNET_TIME_absolute_get_difference (now, pos->timeout);
 365       if (timeout->rel_value > to.rel_value)
 366         *timeout = to;
 367       if (pos->reason != 0)
 368         *timeout = GNUNET_TIME_UNIT_ZERO;
 369     }
 370   pos = pending;
 371   while (pos != NULL)
 372     {
 373       if ((pos->prereq_id != GNUNET_SCHEDULER_NO_TASK) &&
 374           (GNUNET_YES == is_pending (pos->prereq_id)))
 375         {
 376           pos = pos->next;
 377           continue;
 378         }
 379       if (pos->timeout.abs_value != GNUNET_TIME_UNIT_FOREVER_ABS.abs_value)
 380         {
 381           to = GNUNET_TIME_absolute_get_difference (now, pos->timeout);
 382           if (timeout->rel_value > to.rel_value)
 383             *timeout = to;
 384         }
 385       if (pos->read_fd != -1)
 386         GNUNET_NETWORK_fdset_set_native (rs, pos->read_fd);
 387       if (pos->write_fd != -1)
 388         GNUNET_NETWORK_fdset_set_native (ws, pos->write_fd);
 389       if (pos->read_set != NULL)
 390         GNUNET_NETWORK_fdset_add (rs, pos->read_set);
 391       if (pos->write_set != NULL)
 392         GNUNET_NETWORK_fdset_add (ws, pos->write_set);
 393       if (pos->reason != 0)
 394         *timeout = GNUNET_TIME_UNIT_ZERO;
 395       pos = pos->next;
 396     }
 397 }
 398
 399
 400 /**
 401  * Check if the ready set overlaps with the set we want to have ready.
 402  * If so, update the want set (set all FDs that are ready).  If not,
 403  * return GNUNET_NO.
 404  *
 405  * @param ready set that is ready
 406  * @param want set that we want to be ready
 407  * @return GNUNET_YES if there was some overlap
 408  */
 409 static int
 410 set_overlaps (const struct GNUNET_NETWORK_FDSet *ready,
 411               struct GNUNET_NETWORK_FDSet *want)
 412 {
 413   if ( (NULL == want) || (NULL == ready) )
 414     return GNUNET_NO;
 415   if (GNUNET_NETWORK_fdset_overlap (ready, want))
 416     {
 417       /* copy all over (yes, there maybe unrelated bits,
 418          but this should not hurt well-written clients) */
 419       GNUNET_NETWORK_fdset_copy (want, ready);
 420       return GNUNET_YES;
 421     }
 422   return GNUNET_NO;
 423 }
 424
 425
 426 /**
 427  * Check if the given task is eligible to run now.
 428  * Also set the reason why it is eligible.
 429  *
 430  * @param task task to check if it is ready
 431  * @param now the current time
 432  * @param rs set of FDs ready for reading
 433  * @param ws set of FDs ready for writing
 434  * @return GNUNET_YES if we can run it, GNUNET_NO if not.
 435  */
 436 static int
 437 is_ready (struct Task *task,
 438           struct GNUNET_TIME_Absolute now,
 439           const struct GNUNET_NETWORK_FDSet *rs,
 440           const struct GNUNET_NETWORK_FDSet *ws)
 441 {
 442   enum GNUNET_SCHEDULER_Reason reason;
 443
 444   reason = task->reason;
 445   if (now.abs_value >= task->timeout.abs_value)
 446     reason |= GNUNET_SCHEDULER_REASON_TIMEOUT;
 447   if ( (0 == (reason & GNUNET_SCHEDULER_REASON_READ_READY)) &&
 448        ( ( (task->read_fd != -1) &&
 449            (GNUNET_YES == GNUNET_NETWORK_fdset_test_native (rs, task->read_fd)) ) ||
 450          (set_overlaps (rs, task->read_set) ) ) )
 451     reason |= GNUNET_SCHEDULER_REASON_READ_READY;
 452   if ((0 == (reason & GNUNET_SCHEDULER_REASON_WRITE_READY)) &&
 453       ( ( (task->write_fd != -1) &&
 454           (GNUNET_YES == GNUNET_NETWORK_fdset_test_native (ws, task->write_fd)) ) ||
 455         (set_overlaps (ws, task->write_set) ) ) )
 456     reason |= GNUNET_SCHEDULER_REASON_WRITE_READY;
 457   if (reason == 0)
 458     return GNUNET_NO;           /* not ready */
 459   if (task->prereq_id != GNUNET_SCHEDULER_NO_TASK)
 460     {
 461       if (GNUNET_YES == is_pending (task->prereq_id))
 462         {
 463           task->reason = reason;
 464           return GNUNET_NO;       /* prereq waiting */
 465         }
 466       reason |= GNUNET_SCHEDULER_REASON_PREREQ_DONE;
 467     }
 468   task->reason = reason;
 469   return GNUNET_YES;
 470 }
 471
 472
 473 /**
 474  * Put a task that is ready for execution into the ready queue.
 475  *
 476  * @param task task ready for execution
 477  */
 478 static void
 479 queue_ready_task (struct Task *task)
 480 {
 481   enum GNUNET_SCHEDULER_Priority p = task->priority;
 482   if (0 != (task->reason & GNUNET_SCHEDULER_REASON_SHUTDOWN))
 483     p = GNUNET_SCHEDULER_PRIORITY_SHUTDOWN;
 484   task->next = ready[check_priority (p)];
 485   ready[check_priority (p)] = task;
 486   ready_count++;
 487 }
 488
 489
 490 /**
 491  * Check which tasks are ready and move them
 492  * to the respective ready queue.
 493  *
 494  * @param rs FDs ready for reading
 495  * @param ws FDs ready for writing
 496  */
 497 static void
 498 check_ready (const struct GNUNET_NETWORK_FDSet *rs,
 499              const struct GNUNET_NETWORK_FDSet *ws)
 500 {
 501   struct Task *pos;
 502   struct Task *prev;
 503   struct Task *next;
 504   struct GNUNET_TIME_Absolute now;
 505
 506   now = GNUNET_TIME_absolute_get ();
 507   prev = NULL;
 508   pos = pending_timeout;
 509   while (pos != NULL)
 510     {
 511       next = pos->next;
 512       if (now.abs_value >= pos->timeout.abs_value)
 513         pos->reason |= GNUNET_SCHEDULER_REASON_TIMEOUT;
 514       if (0 == pos->reason)
 515         break;
 516       pending_timeout = next;
 517       if (pending_timeout_last == pos)
 518         pending_timeout_last = NULL;
 519       queue_ready_task (pos);
 520       pos = next;
 521     }
 522   pos = pending;
 523   while (pos != NULL)
 524     {
 525 #if DEBUG_TASKS
 526       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 527                   "Checking readiness of task: %llu / %p\n",
 528                   pos->id, pos->callback_cls);
 529 #endif
 530       next = pos->next;
 531       if (GNUNET_YES == is_ready (pos, now, rs, ws))
 532         {
 533           if (prev == NULL)
 534             pending = next;
 535           else
 536             prev->next = next;
 537           queue_ready_task (pos);
 538           pos = next;
 539           continue;
 540         }
 541       prev = pos;
 542       pos = next;
 543     }
 544 }
 545
 546
 547 /**
 548  * Request the shutdown of a scheduler.  Marks all currently
 549  * pending tasks as ready because of shutdown.  This will
 550  * cause all tasks to run (as soon as possible, respecting
 551  * priorities and prerequisite tasks).  Note that tasks
 552  * scheduled AFTER this call may still be delayed arbitrarily.
 553  */
 554 void
 555 GNUNET_SCHEDULER_shutdown ()
 556 {
 557   struct Task *pos;
 558   int i;
 559
 560   pos = pending_timeout;
 561   while (pos != NULL)
 562     {
 563       pos->reason |= GNUNET_SCHEDULER_REASON_SHUTDOWN;
 564       /* we don't move the task into the ready queue yet; check_ready
 565          will do that later, possibly adding additional
 566          readiness-factors */
 567       pos = pos->next;
 568     }
 569   pos = pending;
 570   while (pos != NULL)
 571     {
 572       pos->reason |= GNUNET_SCHEDULER_REASON_SHUTDOWN;
 573       /* we don't move the task into the ready queue yet; check_ready
 574          will do that later, possibly adding additional
 575          readiness-factors */
 576       pos = pos->next;
 577     }
 578   for (i=0;i<GNUNET_SCHEDULER_PRIORITY_COUNT;i++)
 579     {
 580       pos = ready[i];
 581       while (pos != NULL)
 582         {
 583           pos->reason |= GNUNET_SCHEDULER_REASON_SHUTDOWN;
 584           /* we don't move the task into the ready queue yet; check_ready
 585              will do that later, possibly adding additional
 586              readiness-factors */
 587           pos = pos->next;
 588         }
 589     }
 590 }
 591
 592
 593 /**
 594  * Destroy a task (release associated resources)
 595  *
 596  * @param t task to destroy
 597  */
 598 static void
 599 destroy_task (struct Task *t)
 600 {
 601   if (NULL != t->read_set)
 602     GNUNET_NETWORK_fdset_destroy (t->read_set);
 603   if (NULL != t->write_set)
 604     GNUNET_NETWORK_fdset_destroy (t->write_set);
 605 #if EXECINFO
 606   GNUNET_free (t->backtrace_strings);
 607 #endif
 608   GNUNET_free (t);
 609 }
 610
 611
 612 /**
 613  * Run at least one task in the highest-priority queue that is not
 614  * empty.  Keep running tasks until we are either no longer running
 615  * "URGENT" tasks or until we have at least one "pending" task (which
 616  * may become ready, hence we should select on it).  Naturally, if
 617  * there are no more ready tasks, we also return.
 618  *
 619  * @param rs FDs ready for reading
 620  * @param ws FDs ready for writing
 621  */
 622 static void
 623 run_ready (struct GNUNET_NETWORK_FDSet *rs,
 624            struct GNUNET_NETWORK_FDSet *ws)
 625 {
 626   enum GNUNET_SCHEDULER_Priority p;
 627   struct Task *pos;
 628   struct GNUNET_SCHEDULER_TaskContext tc;
 629
 630   max_priority_added = GNUNET_SCHEDULER_PRIORITY_KEEP;
 631   do
 632     {
 633       if (ready_count == 0)
 634         return;
 635       GNUNET_assert (ready[GNUNET_SCHEDULER_PRIORITY_KEEP] == NULL);
 636       /* yes, p>0 is correct, 0 is "KEEP" which should
 637          always be an empty queue (see assertion)! */
 638       for (p = GNUNET_SCHEDULER_PRIORITY_COUNT - 1; p > 0; p--)
 639         {
 640           pos = ready[p];
 641           if (pos != NULL)
 642             break;
 643         }
 644       GNUNET_assert (pos != NULL);      /* ready_count wrong? */
 645       ready[p] = pos->next;
 646       ready_count--;
 647       if (current_priority != pos->priority)
 648         {
 649           current_priority = pos->priority;
 650           (void) GNUNET_OS_set_process_priority (GNUNET_OS_process_current (),
 651                                                  pos->priority);
 652         }
 653       current_lifeness = pos->lifeness;
 654       active_task = pos;
 655 #if PROFILE_DELAYS
 656       if (GNUNET_TIME_absolute_get_duration (pos->start_time).rel_value >
 657           DELAY_THRESHOLD.rel_value)
 658         {
 659           GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
 660                       "Task %llu took %llums to be scheduled\n",
 661                       pos->id,
 662                       (unsigned long long) GNUNET_TIME_absolute_get_duration (pos->start_time).rel_value);
 663         }
 664 #endif
 665       tc.reason = pos->reason;
 666       tc.read_ready = (pos->read_set == NULL) ? rs : pos->read_set;
 667       if ( (pos->read_fd != -1) &&
 668            (0 != (pos->reason & GNUNET_SCHEDULER_REASON_READ_READY)) )
 669         GNUNET_NETWORK_fdset_set_native (rs,
 670                                          pos->read_fd);
 671       tc.write_ready = (pos->write_set == NULL) ? ws : pos->write_set;
 672       if ( (pos->write_fd != -1) &&
 673            (0 != (pos->reason & GNUNET_SCHEDULER_REASON_WRITE_READY)) )
 674         GNUNET_NETWORK_fdset_set_native (ws,
 675                                          pos->write_fd);
 676       if ( ( (tc.reason & GNUNET_SCHEDULER_REASON_WRITE_READY) != 0) &&
 677            (pos->write_fd != -1) &&
 678            (! GNUNET_NETWORK_fdset_test_native (ws,
 679                                                 pos->write_fd)))
 680         abort (); // added to ready in previous select loop!
 681 #if DEBUG_TASKS
 682       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 683                   "Running task: %llu / %p\n", pos->id, pos->callback_cls);
 684 #endif
 685       pos->callback (pos->callback_cls, &tc);
 686 #if EXECINFO
 687       int i;
 688       for (i=0;i<pos->num_backtrace_strings;i++)
 689         GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
 690                     "Task %llu trace %d: %s\n",
 691                     pos->id,
 692                     i,
 693                     pos->backtrace_strings[i]);
 694 #endif
 695       active_task = NULL;
 696       destroy_task (pos);
 697       tasks_run++;
 698     }
 699   while ( (pending == NULL) || (p >= max_priority_added) );
 700 }
 701
 702 /**
 703  * Pipe used to communicate shutdown via signal.
 704  */
 705 static struct GNUNET_DISK_PipeHandle *shutdown_pipe_handle;
 706
 707 /**
 708  * Signal handler called for SIGPIPE.
 709  */
 710 #ifndef MINGW
 711 static void
 712 sighandler_pipe ()
 713 {
 714   return;
 715 }
 716 #endif
 717 /**
 718  * Signal handler called for signals that should cause us to shutdown.
 719  */
 720 static void
 721 sighandler_shutdown ()
 722 {
 723   static char c;
 724   int old_errno = errno; /* backup errno */
 725
 726   GNUNET_DISK_file_write (GNUNET_DISK_pipe_handle
 727                           (shutdown_pipe_handle, GNUNET_DISK_PIPE_END_WRITE), &c,
 728                           sizeof (c));
 729   errno = old_errno;
 730 }
 731
 732
 733 /**
 734  * Check if the system is still life. Trigger shutdown if we
 735  * have tasks, but none of them give us lifeness.
 736  *
 737  * @return GNUNET_OK to continue the main loop,
 738  *         GNUNET_NO to exit
 739  */
 740 static int
 741 check_lifeness()
 742 {
 743   struct Task *t;
 744   if (ready_count > 0)
 745     return GNUNET_OK;
 746   for (t = pending; NULL != t; t = t->next)
 747     if (t->lifeness == GNUNET_YES)
 748       return GNUNET_OK;
 749   for (t = pending_timeout; NULL != t; t = t->next)
 750     if (t->lifeness == GNUNET_YES)
 751         return GNUNET_OK;
 752   if ( (NULL != pending) || (NULL != pending_timeout) )
 753     {
 754       GNUNET_SCHEDULER_shutdown ();
 755       return GNUNET_OK;
 756     }
 757   return GNUNET_NO;
 758 }
 759
 760
 761 /**
 762  * Initialize and run scheduler.  This function will return when all
 763  * tasks have completed.  On systems with signals, receiving a SIGTERM
 764  * (and other similar signals) will cause "GNUNET_SCHEDULER_shutdown"
 765  * to be run after the active task is complete.  As a result, SIGTERM
 766  * causes all active tasks to be scheduled with reason
 767  * "GNUNET_SCHEDULER_REASON_SHUTDOWN".  (However, tasks added
 768  * afterwards will execute normally!). Note that any particular signal
 769  * will only shut down one scheduler; applications should always only
 770  * create a single scheduler.
 771  *
 772  * @param task task to run immediately
 773  * @param task_cls closure of task
 774  */
 775 void
 776 GNUNET_SCHEDULER_run (GNUNET_SCHEDULER_Task task, void *task_cls)
 777 {
 778   struct GNUNET_NETWORK_FDSet *rs;
 779   struct GNUNET_NETWORK_FDSet *ws;
 780   struct GNUNET_TIME_Relative timeout;
 781   int ret;
 782   struct GNUNET_SIGNAL_Context *shc_int;
 783   struct GNUNET_SIGNAL_Context *shc_term;
 784 #ifndef MINGW
 785   struct GNUNET_SIGNAL_Context *shc_quit;
 786   struct GNUNET_SIGNAL_Context *shc_hup;
 787   struct GNUNET_SIGNAL_Context *shc_pipe;
 788 #endif
 789   unsigned long long last_tr;
 790   unsigned int busy_wait_warning;
 791   const struct GNUNET_DISK_FileHandle *pr;
 792   char c;
 793
 794   GNUNET_assert (active_task == NULL);
 795   rs = GNUNET_NETWORK_fdset_create ();
 796   ws = GNUNET_NETWORK_fdset_create ();
 797   GNUNET_assert (shutdown_pipe_handle == NULL);
 798   shutdown_pipe_handle =  GNUNET_DISK_pipe (GNUNET_NO, GNUNET_NO, GNUNET_NO);
 799   GNUNET_assert (shutdown_pipe_handle != NULL);
 800   pr = GNUNET_DISK_pipe_handle (shutdown_pipe_handle, GNUNET_DISK_PIPE_END_READ);
 801   GNUNET_assert (pr != NULL);
 802   shc_int = GNUNET_SIGNAL_handler_install (SIGINT, &sighandler_shutdown);
 803   shc_term = GNUNET_SIGNAL_handler_install (SIGTERM, &sighandler_shutdown);
 804 #ifndef MINGW
 805   shc_pipe = GNUNET_SIGNAL_handler_install (SIGPIPE, &sighandler_pipe);
 806   shc_quit = GNUNET_SIGNAL_handler_install (SIGQUIT, &sighandler_shutdown);
 807   shc_hup = GNUNET_SIGNAL_handler_install (SIGHUP, &sighandler_shutdown);
 808 #endif
 809   current_priority = GNUNET_SCHEDULER_PRIORITY_DEFAULT;
 810   current_lifeness = GNUNET_YES;
 811   GNUNET_SCHEDULER_add_continuation (task,
 812                                      task_cls,
 813                                      GNUNET_SCHEDULER_REASON_STARTUP);
 814 #if ENABLE_WINDOWS_WORKAROUNDS
 815   active_task = (void*) (long) -1; /* force passing of sanity check */
 816   GNUNET_SCHEDULER_add_now_with_lifeness (GNUNET_NO,
 817                                           &GNUNET_OS_install_parent_control_handler,
 818                                           NULL);
 819   active_task = NULL;
 820 #endif
 821   last_tr = 0;
 822   busy_wait_warning = 0;
 823   while (GNUNET_OK == check_lifeness ())
 824     {
 825       GNUNET_NETWORK_fdset_zero (rs);
 826       GNUNET_NETWORK_fdset_zero (ws);
 827       timeout = GNUNET_TIME_UNIT_FOREVER_REL;
 828       update_sets (rs, ws, &timeout);
 829       GNUNET_NETWORK_fdset_handle_set (rs, pr);
 830       if (ready_count > 0)
 831         {
 832           /* no blocking, more work already ready! */
 833           timeout = GNUNET_TIME_UNIT_ZERO;
 834         }
 835       if (NULL == scheduler_select)
 836         ret = GNUNET_NETWORK_socket_select (rs, ws, NULL, timeout);
 837       else
 838         ret = scheduler_select (scheduler_select_cls,
 839                                 rs, ws, NULL,
 840                                 timeout);
 841       if (ret == GNUNET_SYSERR)
 842         {
 843           if (errno == EINTR)
 844             continue;
 845
 846           GNUNET_log_strerror (GNUNET_ERROR_TYPE_ERROR, "select");
 847 #ifndef MINGW
 848 #if USE_LSOF
 849           char lsof[512];
 850           snprintf (lsof, sizeof (lsof), "lsof -p %d", getpid());
 851           (void) close (1);
 852           (void) dup2 (2, 1);
 853           if (0 != system (lsof))
 854             GNUNET_log_strerror (GNUNET_ERROR_TYPE_WARNING, "system");
 855 #endif
 856 #endif
 857           abort ();
 858           break;
 859         }
 860       if ((ret == 0) && (timeout.rel_value == 0) && (busy_wait_warning > 16))
 861         {
 862           GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
 863                       _("Looks like we're busy waiting...\n"));
 864           sleep (1);            /* mitigate */
 865         }
 866       check_ready (rs, ws);
 867       run_ready (rs, ws);
 868       if (GNUNET_NETWORK_fdset_handle_isset (rs, pr))
 869         {
 870           /* consume the signal */
 871           GNUNET_DISK_file_read (pr, &c, sizeof (c));
 872           /* mark all active tasks as ready due to shutdown */
 873           GNUNET_SCHEDULER_shutdown ();
 874         }
 875       if (last_tr == tasks_run)
 876         {
 877           busy_wait_warning++;
 878         }
 879       else
 880         {
 881           last_tr = tasks_run;
 882           busy_wait_warning = 0;
 883         }
 884     }
 885   GNUNET_SIGNAL_handler_uninstall (shc_int);
 886   GNUNET_SIGNAL_handler_uninstall (shc_term);
 887 #ifndef MINGW
 888   GNUNET_SIGNAL_handler_uninstall (shc_pipe);
 889   GNUNET_SIGNAL_handler_uninstall (shc_quit);
 890   GNUNET_SIGNAL_handler_uninstall (shc_hup);
 891 #endif
 892   GNUNET_DISK_pipe_close (shutdown_pipe_handle);
 893   shutdown_pipe_handle = NULL;
 894   GNUNET_NETWORK_fdset_destroy (rs);
 895   GNUNET_NETWORK_fdset_destroy (ws);
 896 }
 897
 898
 899 /**
 900  * Obtain the reason code for why the current task was
 901  * started.  Will return the same value as
 902  * the GNUNET_SCHEDULER_TaskContext's reason field.
 903  *
 904  * @return reason(s) why the current task is run
 905  */
 906 enum GNUNET_SCHEDULER_Reason
 907 GNUNET_SCHEDULER_get_reason ()
 908 {
 909   GNUNET_assert (active_task != NULL);
 910   return active_task->reason;
 911 }
 912
 913
 914 /**
 915  * Get information about the current load of this scheduler.  Use this
 916  * function to determine if an elective task should be added or simply
 917  * dropped (if the decision should be made based on the number of
 918  * tasks ready to run).
 919  *
 920  * @param p priority level to look at
 921  * @return number of tasks pending right now
 922  */
 923 unsigned int
 924 GNUNET_SCHEDULER_get_load (enum GNUNET_SCHEDULER_Priority p)
 925 {
 926   struct Task *pos;
 927   unsigned int ret;
 928
 929   GNUNET_assert (active_task != NULL);
 930   if (p == GNUNET_SCHEDULER_PRIORITY_COUNT)
 931     return ready_count;
 932   if (p == GNUNET_SCHEDULER_PRIORITY_KEEP)
 933     p = current_priority;
 934   ret = 0;
 935   pos = ready[check_priority (p)];
 936   while (pos != NULL)
 937     {
 938       pos = pos->next;
 939       ret++;
 940     }
 941   return ret;
 942 }
 943
 944
 945 /**
 946  * Cancel the task with the specified identifier.
 947  * The task must not yet have run.
 948  *
 949  * @param task id of the task to cancel
 950  * @return original closure of the task
 951  */
 952 void *
 953 GNUNET_SCHEDULER_cancel (GNUNET_SCHEDULER_TaskIdentifier task)
 954 {
 955   struct Task *t;
 956   struct Task *prev;
 957   enum GNUNET_SCHEDULER_Priority p;
 958   int to;
 959   void *ret;
 960
 961   GNUNET_assert (active_task != NULL);
 962   to = 0;
 963   prev = NULL;
 964   t = pending;
 965   while (t != NULL)
 966     {
 967       if (t->id == task)
 968         break;
 969       prev = t;
 970       t = t->next;
 971     }
 972   if (t == NULL)
 973     {
 974       prev = NULL;
 975       to = 1;
 976       t = pending_timeout;
 977       while (t != NULL)
 978         {
 979           if (t->id == task)
 980             break;
 981           prev = t;
 982           t = t->next;
 983         }
 984       if (pending_timeout_last == t)
 985         pending_timeout_last = NULL;
 986     }
 987   p = 0;
 988   while (t == NULL)
 989     {
 990       p++;
 991       if (p >= GNUNET_SCHEDULER_PRIORITY_COUNT)
 992         {
 993           GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
 994                       _("Attempt to cancel dead task %llu!\n"),
 995                       (unsigned long long) task);
 996           GNUNET_assert (0);
 997         }
 998       prev = NULL;
 999       t = ready[p];
1000       while (t != NULL)
1001         {
1002           if (t->id == task)
1003             {
1004               ready_count--;
1005               break;
1006             }
1007           prev = t;
1008           t = t->next;
1009         }
1010     }
1011   if (prev == NULL)
1012     {
1013       if (p == 0)
1014         {
1015           if (to == 0)
1016             {
1017               pending = t->next;
1018             }
1019           else
1020             {
1021               pending_timeout = t->next;
1022             }
1023         }
1024       else
1025         {
1026           ready[p] = t->next;
1027         }
1028     }
1029   else
1030     {
1031       prev->next = t->next;
1032     }
1033   ret = t->callback_cls;
1034 #if DEBUG_TASKS
1035   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1036               "Canceling task: %llu / %p\n", task, t->callback_cls);
1037 #endif
1038   destroy_task (t);
1039   return ret;
1040 }
1041
1042
1043 /**
1044  * Continue the current execution with the given function.  This is
1045  * similar to the other "add" functions except that there is no delay
1046  * and the reason code can be specified.
1047  *
1048  * @param task main function of the task
1049  * @param task_cls closure for 'main'
1050  * @param reason reason for task invocation
1051  */
1052 void
1053 GNUNET_SCHEDULER_add_continuation (GNUNET_SCHEDULER_Task task,
1054                                    void *task_cls,
1055                                    enum GNUNET_SCHEDULER_Reason reason)
1056 {
1057   struct Task *t;
1058 #if EXECINFO
1059   void *backtrace_array[50];
1060 #endif
1061
1062   GNUNET_assert (NULL != task);
1063   GNUNET_assert ( (active_task != NULL) ||
1064                   (reason == GNUNET_SCHEDULER_REASON_STARTUP) );
1065   t = GNUNET_malloc (sizeof (struct Task));
1066 #if EXECINFO
1067   t->num_backtrace_strings = backtrace(backtrace_array, 50);
1068   t->backtrace_strings = backtrace_symbols(backtrace_array, t->num_backtrace_strings);
1069 #endif
1070   t->read_fd = -1;
1071   t->write_fd = -1;
1072   t->callback = task;
1073   t->callback_cls = task_cls;
1074   t->id = ++last_id;
1075 #if PROFILE_DELAYS
1076   t->start_time = GNUNET_TIME_absolute_get ();
1077 #endif
1078   t->reason = reason;
1079   t->priority = current_priority;
1080   t->lifeness = current_lifeness;
1081 #if DEBUG_TASKS
1082   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1083               "Adding continuation task: %llu / %p\n",
1084               t->id, t->callback_cls);
1085 #endif
1086   queue_ready_task (t);
1087 }
1088
1089
1090
1091 /**
1092  * Schedule a new task to be run after the specified prerequisite task
1093  * has completed. It will be run with the priority of the calling
1094  * task.
1095  *
1096  * @param prerequisite_task run this task after the task with the given
1097  *        task identifier completes (and any of our other
1098  *        conditions, such as delay, read or write-readiness
1099  *        are satisfied).  Use  GNUNET_SCHEDULER_NO_TASK to not have any dependency
1100  *        on completion of other tasks (this will cause the task to run as
1101  *        soon as possible).
1102  * @param task main function of the task
1103  * @param task_cls closure of task
1104  * @return unique task identifier for the job
1105  *         only valid until "task" is started!
1106  */
1107 GNUNET_SCHEDULER_TaskIdentifier
1108 GNUNET_SCHEDULER_add_after (GNUNET_SCHEDULER_TaskIdentifier prerequisite_task,
1109                             GNUNET_SCHEDULER_Task task, void *task_cls)
1110 {
1111   return GNUNET_SCHEDULER_add_select (GNUNET_SCHEDULER_PRIORITY_KEEP,
1112                                       prerequisite_task,
1113                                       GNUNET_TIME_UNIT_ZERO,
1114                                       NULL, NULL, task, task_cls);
1115 }
1116
1117
1118 /**
1119  * Schedule a new task to be run with a specified priority.
1120  *
1121  * @param prio how important is the new task?
1122  * @param task main function of the task
1123  * @param task_cls closure of task
1124  * @return unique task identifier for the job
1125  *         only valid until "task" is started!
1126  */
1127 GNUNET_SCHEDULER_TaskIdentifier
1128 GNUNET_SCHEDULER_add_with_priority (enum GNUNET_SCHEDULER_Priority prio,
1129                                     GNUNET_SCHEDULER_Task task,
1130                                     void *task_cls)
1131 {
1132   return GNUNET_SCHEDULER_add_select (prio,
1133                                       GNUNET_SCHEDULER_NO_TASK,
1134                                       GNUNET_TIME_UNIT_ZERO,
1135                                       NULL, NULL, task, task_cls);
1136 }
1137
1138
1139
1140 /**
1141  * Schedule a new task to be run with a specified delay.  The task
1142  * will be scheduled for execution once the delay has expired. It
1143  * will be run with the priority of the calling task.
1144  *
1145  * @param delay when should this operation time out? Use
1146  *        GNUNET_TIME_UNIT_FOREVER_REL for "on shutdown"
1147  * @param task main function of the task
1148  * @param task_cls closure of task
1149  * @return unique task identifier for the job
1150  *         only valid until "task" is started!
1151  */
1152 GNUNET_SCHEDULER_TaskIdentifier
1153 GNUNET_SCHEDULER_add_delayed (struct GNUNET_TIME_Relative delay,
1154                               GNUNET_SCHEDULER_Task task, void *task_cls)
1155 {
1156 #if 1
1157   /* new, optimized version */
1158   struct Task *t;
1159   struct Task *pos;
1160   struct Task *prev;
1161 #if EXECINFO
1162   void *backtrace_array[MAX_TRACE_DEPTH];
1163 #endif
1164
1165   GNUNET_assert (active_task != NULL);
1166   GNUNET_assert (NULL != task);
1167   t = GNUNET_malloc (sizeof (struct Task));
1168   t->callback = task;
1169   t->callback_cls = task_cls;
1170 #if EXECINFO
1171   t->num_backtrace_strings = backtrace(backtrace_array, MAX_TRACE_DEPTH);
1172   t->backtrace_strings = backtrace_symbols(backtrace_array, t->num_backtrace_strings);
1173 #endif
1174   t->read_fd = -1;
1175   t->write_fd = -1;
1176   t->id = ++last_id;
1177 #if PROFILE_DELAYS
1178   t->start_time = GNUNET_TIME_absolute_get ();
1179 #endif
1180   t->timeout = GNUNET_TIME_relative_to_absolute (delay);
1181   t->priority = current_priority;
1182   t->lifeness = current_lifeness;
1183   /* try tail first (optimization in case we are
1184      appending to a long list of tasks with timeouts) */
1185   prev = pending_timeout_last;
1186   if (prev != NULL)
1187     {
1188       if (prev->timeout.abs_value > t->timeout.abs_value)
1189         prev = NULL;
1190       else
1191         pos = prev->next; /* heuristic success! */
1192     }
1193   if (prev == NULL)
1194     {
1195       /* heuristic failed, do traversal of timeout list */
1196       pos = pending_timeout;
1197     }
1198   while ( (pos != NULL) &&
1199           ( (pos->timeout.abs_value <= t->timeout.abs_value) ||
1200             (pos->reason != 0) ) )
1201     {
1202       prev = pos;
1203       pos = pos->next;
1204     }
1205   if (prev == NULL)
1206     pending_timeout = t;
1207   else
1208     prev->next = t;
1209   t->next = pos;
1210   /* hyper-optimization... */
1211   pending_timeout_last = t;
1212
1213 #if DEBUG_TASKS
1214   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1215               "Adding task: %llu / %p\n", t->id, t->callback_cls);
1216 #endif
1217 #if EXECINFO
1218   int i;
1219
1220   for (i=0;i<t->num_backtrace_strings;i++)
1221       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1222                   "Task %llu trace %d: %s\n",
1223                   t->id,
1224                   i,
1225                   t->backtrace_strings[i]);
1226 #endif
1227   return t->id;
1228
1229 #else
1230   /* unoptimized version */
1231   return GNUNET_SCHEDULER_add_select (GNUNET_SCHEDULER_PRIORITY_KEEP,
1232                                       GNUNET_SCHEDULER_NO_TASK, delay,
1233                                       NULL, NULL, task, task_cls);
1234 #endif
1235 }
1236
1237
1238
1239 /**
1240  * Schedule a new task to be run as soon as possible. The task
1241  * will be run with the priority of the calling task.
1242  *
1243  * @param task main function of the task
1244  * @param task_cls closure of task
1245  * @return unique task identifier for the job
1246  *         only valid until "task" is started!
1247  */
1248 GNUNET_SCHEDULER_TaskIdentifier
1249 GNUNET_SCHEDULER_add_now (GNUNET_SCHEDULER_Task task,
1250                           void *task_cls)
1251 {
1252   return GNUNET_SCHEDULER_add_select (GNUNET_SCHEDULER_PRIORITY_KEEP,
1253                                       GNUNET_SCHEDULER_NO_TASK,
1254                                       GNUNET_TIME_UNIT_ZERO,
1255                                       NULL, NULL, task, task_cls);
1256 }
1257
1258
1259 /**
1260  * Schedule a new task to be run as soon as possible with the
1261  * (transitive) ignore-shutdown flag either explicitly set or
1262  * explicitly enabled.  This task (and all tasks created from it,
1263  * other than by another call to this function) will either count or
1264  * not count for the 'lifeness' of the process.  This API is only
1265  * useful in a few special cases.
1266  *
1267  * @param lifeness GNUNET_YES if the task counts for lifeness, GNUNET_NO if not.
1268  * @param task main function of the task
1269  * @param task_cls closure of task
1270  * @return unique task identifier for the job
1271  *         only valid until "task" is started!
1272  */
1273 GNUNET_SCHEDULER_TaskIdentifier
1274 GNUNET_SCHEDULER_add_now_with_lifeness (int lifeness,
1275                                         GNUNET_SCHEDULER_Task task,
1276                                         void *task_cls)
1277 {
1278   GNUNET_SCHEDULER_TaskIdentifier ret;
1279
1280   ret = GNUNET_SCHEDULER_add_select (GNUNET_SCHEDULER_PRIORITY_KEEP,
1281                                      GNUNET_SCHEDULER_NO_TASK,
1282                                      GNUNET_TIME_UNIT_ZERO,
1283                                      NULL, NULL, task, task_cls);
1284   GNUNET_assert (pending->id == ret);
1285   pending->lifeness = lifeness;
1286   return ret;
1287 }
1288
1289
1290
1291
1292 /**
1293  * Schedule a new task to be run with a specified delay or when any of
1294  * the specified file descriptor sets is ready.  The delay can be used
1295  * as a timeout on the socket(s) being ready.  The task will be
1296  * scheduled for execution once either the delay has expired or any of
1297  * the socket operations is ready.  This is the most general
1298  * function of the "add" family.  Note that the "prerequisite_task"
1299  * must be satisfied in addition to any of the other conditions.  In
1300  * other words, the task will be started when
1301  * <code>
1302  * (prerequisite-run)
1303  * && (delay-ready
1304  *     || any-rs-ready
1305  *     || any-ws-ready
1306  *     || shutdown-active )
1307  * </code>
1308  *
1309  * @param delay how long should we wait? Use GNUNET_TIME_UNIT_FOREVER_REL for "forever",
1310  *        which means that the task will only be run after we receive SIGTERM
1311  * @param rfd file descriptor we want to read (can be -1)
1312  * @param wfd file descriptors we want to write (can be -1)
1313  * @param task main function of the task
1314  * @param task_cls closure of task
1315  * @return unique task identifier for the job
1316  *         only valid until "task" is started!
1317  */
1318 #ifndef MINGW
1319 GNUNET_SCHEDULER_TaskIdentifier
1320 add_without_sets (struct GNUNET_TIME_Relative delay,
1321                   int rfd,
1322                   int wfd,
1323                   GNUNET_SCHEDULER_Task task, void *task_cls)
1324 {
1325   struct Task *t;
1326 #if EXECINFO
1327   void *backtrace_array[MAX_TRACE_DEPTH];
1328 #endif
1329
1330   GNUNET_assert (active_task != NULL);
1331   GNUNET_assert (NULL != task);
1332   t = GNUNET_malloc (sizeof (struct Task));
1333   t->callback = task;
1334   t->callback_cls = task_cls;
1335 #if EXECINFO
1336   t->num_backtrace_strings = backtrace(backtrace_array, MAX_TRACE_DEPTH);
1337   t->backtrace_strings = backtrace_symbols(backtrace_array, t->num_backtrace_strings);
1338 #endif
1339 #if DEBUG_FDS
1340   if (-1 != rfd)
1341     {
1342       int flags = fcntl(rfd, F_GETFD);
1343       if (flags == -1 && errno == EBADF)
1344         {
1345           GNUNET_log(GNUNET_ERROR_TYPE_ERROR, "Got invalid file descriptor %d!\n", rfd);
1346 #if EXECINFO
1347           int i;
1348
1349           for (i=0;i<t->num_backtrace_strings;i++)
1350             GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1351                         "Trace: %s\n",
1352                         t->backtrace_strings[i]);
1353 #endif
1354           GNUNET_assert(0);
1355         }
1356     }
1357   if (-1 != wfd)
1358     {
1359       int flags = fcntl(wfd, F_GETFD);
1360       if (flags == -1 && errno == EBADF)
1361         {
1362           GNUNET_log(GNUNET_ERROR_TYPE_ERROR, "Got invalid file descriptor %d!\n", wfd);
1363 #if EXECINFO
1364           int i;
1365
1366           for (i=0;i<t->num_backtrace_strings;i++)
1367             GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1368                         "Trace: %s\n",
1369                         t->backtrace_strings[i]);
1370 #endif
1371           GNUNET_assert(0);
1372         }
1373     }
1374 #endif
1375   t->read_fd = rfd;
1376   GNUNET_assert(wfd >= -1);
1377   t->write_fd = wfd;
1378   t->id = ++last_id;
1379 #if PROFILE_DELAYS
1380   t->start_time = GNUNET_TIME_absolute_get ();
1381 #endif
1382   t->prereq_id = GNUNET_SCHEDULER_NO_TASK;
1383   t->timeout = GNUNET_TIME_relative_to_absolute (delay);
1384   t->priority = check_priority (current_priority);
1385   t->lifeness = current_lifeness;
1386   t->next = pending;
1387   pending = t;
1388   max_priority_added = GNUNET_MAX (max_priority_added,
1389                                           t->priority);
1390 #if DEBUG_TASKS
1391   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1392               "Adding task: %llu / %p\n", t->id, t->callback_cls);
1393 #endif
1394 #if EXECINFO
1395   int i;
1396
1397   for (i=0;i<t->num_backtrace_strings;i++)
1398       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1399                   "Task %llu trace %d: %s\n",
1400                   t->id,
1401                   i,
1402                   t->backtrace_strings[i]);
1403 #endif
1404   return t->id;
1405 }
1406 #endif
1407
1408
1409
1410 /**
1411  * Schedule a new task to be run with a specified delay or when the
1412  * specified file descriptor is ready for reading.  The delay can be
1413  * used as a timeout on the socket being ready.  The task will be
1414  * scheduled for execution once either the delay has expired or the
1415  * socket operation is ready.  It will be run with the priority of
1416  * the calling task.
1417  *
1418  * @param delay when should this operation time out? Use
1419  *        GNUNET_TIME_UNIT_FOREVER_REL for "on shutdown"
1420  * @param rfd read file-descriptor
1421  * @param task main function of the task
1422  * @param task_cls closure of task
1423  * @return unique task identifier for the job
1424  *         only valid until "task" is started!
1425  */
1426 GNUNET_SCHEDULER_TaskIdentifier
1427 GNUNET_SCHEDULER_add_read_net (struct GNUNET_TIME_Relative delay,
1428                                struct GNUNET_NETWORK_Handle * rfd,
1429                                GNUNET_SCHEDULER_Task task, void *task_cls)
1430 {
1431 #if MINGW
1432   struct GNUNET_NETWORK_FDSet *rs;
1433   GNUNET_SCHEDULER_TaskIdentifier ret;
1434
1435   GNUNET_assert (rfd != NULL);
1436   rs = GNUNET_NETWORK_fdset_create ();
1437   GNUNET_NETWORK_fdset_set (rs, rfd);
1438   ret = GNUNET_SCHEDULER_add_select (check_priority (current_priority),
1439                                      GNUNET_SCHEDULER_NO_TASK, delay,
1440                                      rs, NULL, task, task_cls);
1441   GNUNET_NETWORK_fdset_destroy (rs);
1442   return ret;
1443 #else
1444   return add_without_sets (delay,
1445                            GNUNET_NETWORK_get_fd (rfd),
1446                            -1,
1447                            task,
1448                            task_cls);
1449 #endif
1450 }
1451
1452
1453 /**
1454  * Schedule a new task to be run with a specified delay or when the
1455  * specified file descriptor is ready for writing.  The delay can be
1456  * used as a timeout on the socket being ready.  The task will be
1457  * scheduled for execution once either the delay has expired or the
1458  * socket operation is ready.  It will be run with the priority of
1459  * the calling task.
1460  *
1461  * @param delay when should this operation time out? Use
1462  *        GNUNET_TIME_UNIT_FOREVER_REL for "on shutdown"
1463  * @param wfd write file-descriptor
1464  * @param task main function of the task
1465  * @param task_cls closure of task
1466  * @return unique task identifier for the job
1467  *         only valid until "task" is started!
1468  */
1469 GNUNET_SCHEDULER_TaskIdentifier
1470 GNUNET_SCHEDULER_add_write_net (struct GNUNET_TIME_Relative delay,
1471                                 struct GNUNET_NETWORK_Handle * wfd,
1472                                 GNUNET_SCHEDULER_Task task, void *task_cls)
1473 {
1474 #if MINGW
1475   struct GNUNET_NETWORK_FDSet *ws;
1476   GNUNET_SCHEDULER_TaskIdentifier ret;
1477
1478   GNUNET_assert (wfd != NULL);
1479   ws = GNUNET_NETWORK_fdset_create ();
1480   GNUNET_NETWORK_fdset_set (ws, wfd);
1481   ret = GNUNET_SCHEDULER_add_select (check_priority (current_priority),
1482                                      GNUNET_SCHEDULER_NO_TASK, delay,
1483                                      NULL, ws, task, task_cls);
1484   GNUNET_NETWORK_fdset_destroy (ws);
1485   return ret;
1486 #else
1487   GNUNET_assert(GNUNET_NETWORK_get_fd(wfd) >= 0);
1488   return add_without_sets (delay,
1489                            -1,
1490                            GNUNET_NETWORK_get_fd (wfd),
1491                            task,
1492                            task_cls);
1493 #endif
1494 }
1495
1496
1497 /**
1498  * Schedule a new task to be run with a specified delay or when the
1499  * specified file descriptor is ready for reading.  The delay can be
1500  * used as a timeout on the socket being ready.  The task will be
1501  * scheduled for execution once either the delay has expired or the
1502  * socket operation is ready. It will be run with the priority of
1503  * the calling task.
1504  *
1505  * @param delay when should this operation time out? Use
1506  *        GNUNET_TIME_UNIT_FOREVER_REL for "on shutdown"
1507  * @param rfd read file-descriptor
1508  * @param task main function of the task
1509  * @param task_cls closure of task
1510  * @return unique task identifier for the job
1511  *         only valid until "task" is started!
1512  */
1513 GNUNET_SCHEDULER_TaskIdentifier
1514 GNUNET_SCHEDULER_add_read_file (struct GNUNET_TIME_Relative delay,
1515                                 const struct GNUNET_DISK_FileHandle * rfd,
1516                                 GNUNET_SCHEDULER_Task task, void *task_cls)
1517 {
1518 #if MINGW
1519   struct GNUNET_NETWORK_FDSet *rs;
1520   GNUNET_SCHEDULER_TaskIdentifier ret;
1521
1522   GNUNET_assert (rfd != NULL);
1523   rs = GNUNET_NETWORK_fdset_create ();
1524   GNUNET_NETWORK_fdset_handle_set (rs, rfd);
1525   ret = GNUNET_SCHEDULER_add_select (check_priority (current_priority),
1526                                      GNUNET_SCHEDULER_NO_TASK, delay,
1527                                      rs, NULL, task, task_cls);
1528   GNUNET_NETWORK_fdset_destroy (rs);
1529   return ret;
1530 #else
1531   int fd;
1532
1533   GNUNET_DISK_internal_file_handle_ (rfd, &fd, sizeof (int));
1534   return add_without_sets (delay,
1535                            fd,
1536                            -1,
1537                            task,
1538                            task_cls);
1539
1540 #endif
1541 }
1542
1543
1544 /**
1545  * Schedule a new task to be run with a specified delay or when the
1546  * specified file descriptor is ready for writing.  The delay can be
1547  * used as a timeout on the socket being ready.  The task will be
1548  * scheduled for execution once either the delay has expired or the
1549  * socket operation is ready. It will be run with the priority of
1550  * the calling task.
1551  *
1552  * @param delay when should this operation time out? Use
1553  *        GNUNET_TIME_UNIT_FOREVER_REL for "on shutdown"
1554  * @param wfd write file-descriptor
1555  * @param task main function of the task
1556  * @param task_cls closure of task
1557  * @return unique task identifier for the job
1558  *         only valid until "task" is started!
1559  */
1560 GNUNET_SCHEDULER_TaskIdentifier
1561 GNUNET_SCHEDULER_add_write_file (struct GNUNET_TIME_Relative delay,
1562                                  const struct GNUNET_DISK_FileHandle * wfd,
1563                                  GNUNET_SCHEDULER_Task task, void *task_cls)
1564 {
1565 #if MINGW
1566   struct GNUNET_NETWORK_FDSet *ws;
1567   GNUNET_SCHEDULER_TaskIdentifier ret;
1568
1569   GNUNET_assert (wfd != NULL);
1570   ws = GNUNET_NETWORK_fdset_create ();
1571   GNUNET_NETWORK_fdset_handle_set (ws, wfd);
1572   ret = GNUNET_SCHEDULER_add_select (check_priority (current_priority),
1573                                      GNUNET_SCHEDULER_NO_TASK,
1574                                      delay, NULL, ws, task, task_cls);
1575   GNUNET_NETWORK_fdset_destroy (ws);
1576   return ret;
1577 #else
1578   int fd;
1579
1580   GNUNET_DISK_internal_file_handle_ (wfd, &fd, sizeof (int));
1581   GNUNET_assert(fd >= 0);
1582   return add_without_sets (delay,
1583                            -1,
1584                            fd,
1585                            task,
1586                            task_cls);
1587
1588 #endif
1589 }
1590
1591
1592
1593 /**
1594  * Schedule a new task to be run with a specified delay or when any of
1595  * the specified file descriptor sets is ready.  The delay can be used
1596  * as a timeout on the socket(s) being ready.  The task will be
1597  * scheduled for execution once either the delay has expired or any of
1598  * the socket operations is ready.  This is the most general
1599  * function of the "add" family.  Note that the "prerequisite_task"
1600  * must be satisfied in addition to any of the other conditions.  In
1601  * other words, the task will be started when
1602  * <code>
1603  * (prerequisite-run)
1604  * && (delay-ready
1605  *     || any-rs-ready
1606  *     || any-ws-ready
1607  *     || (shutdown-active && run-on-shutdown) )
1608  * </code>
1609  *
1610  * @param prio how important is this task?
1611  * @param prerequisite_task run this task after the task with the given
1612  *        task identifier completes (and any of our other
1613  *        conditions, such as delay, read or write-readiness
1614  *        are satisfied).  Use GNUNET_SCHEDULER_NO_TASK to not have any dependency
1615  *        on completion of other tasks.
1616  * @param delay how long should we wait? Use GNUNET_TIME_UNIT_FOREVER_REL for "forever",
1617  *        which means that the task will only be run after we receive SIGTERM
1618  * @param rs set of file descriptors we want to read (can be NULL)
1619  * @param ws set of file descriptors we want to write (can be NULL)
1620  * @param task main function of the task
1621  * @param task_cls closure of task
1622  * @return unique task identifier for the job
1623  *         only valid until "task" is started!
1624  */
1625 GNUNET_SCHEDULER_TaskIdentifier
1626 GNUNET_SCHEDULER_add_select (enum GNUNET_SCHEDULER_Priority prio,
1627                              GNUNET_SCHEDULER_TaskIdentifier
1628                              prerequisite_task,
1629                              struct GNUNET_TIME_Relative delay,
1630                              const struct GNUNET_NETWORK_FDSet * rs,
1631                              const struct GNUNET_NETWORK_FDSet * ws,
1632                              GNUNET_SCHEDULER_Task task, void *task_cls)
1633 {
1634   struct Task *t;
1635 #if EXECINFO
1636   void *backtrace_array[MAX_TRACE_DEPTH];
1637 #endif
1638
1639   GNUNET_assert (active_task != NULL);
1640   GNUNET_assert (NULL != task);
1641   t = GNUNET_malloc (sizeof (struct Task));
1642   t->callback = task;
1643   t->callback_cls = task_cls;
1644 #if EXECINFO
1645   t->num_backtrace_strings = backtrace(backtrace_array, MAX_TRACE_DEPTH);
1646   t->backtrace_strings = backtrace_symbols(backtrace_array, t->num_backtrace_strings);
1647 #endif
1648   t->read_fd = -1;
1649   t->write_fd = -1;
1650   if (rs != NULL)
1651     {
1652       t->read_set = GNUNET_NETWORK_fdset_create ();
1653       GNUNET_NETWORK_fdset_copy (t->read_set, rs);
1654     }
1655   if (ws != NULL)
1656     {
1657       t->write_set = GNUNET_NETWORK_fdset_create ();
1658       GNUNET_NETWORK_fdset_copy (t->write_set, ws);
1659     }
1660   t->id = ++last_id;
1661 #if PROFILE_DELAYS
1662   t->start_time = GNUNET_TIME_absolute_get ();
1663 #endif
1664   t->prereq_id = prerequisite_task;
1665   t->timeout = GNUNET_TIME_relative_to_absolute (delay);
1666   t->priority =
1667     check_priority ((prio ==
1668                      GNUNET_SCHEDULER_PRIORITY_KEEP) ? current_priority
1669                     : prio);
1670   t->lifeness = current_lifeness;
1671   t->next = pending;
1672   pending = t;
1673   max_priority_added = GNUNET_MAX (max_priority_added,
1674                                    t->priority);
1675 #if DEBUG_TASKS
1676   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1677               "Adding task: %llu / %p\n", t->id, t->callback_cls);
1678 #endif
1679 #if EXECINFO
1680   int i;
1681
1682   for (i=0;i<t->num_backtrace_strings;i++)
1683       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1684                   "Task %llu trace %d: %s\n",
1685                   t->id,
1686                   i,
1687                   t->backtrace_strings[i]);
1688 #endif
1689   return t->id;
1690 }
1691
1692 /* end of scheduler.c */