src/util/scheduler.c

   1 /*
   2       This file is part of GNUnet
   3       (C) 2009 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 #ifdef LINUX
  33 #include "execinfo.h"
  34
  35 /**
  36  * Use lsof to generate file descriptor reports on select error?
  37  * (turn off for stable releases).
  38  */
  39 #define USE_LSOF GNUNET_YES
  40
  41 /**
  42  * Obtain trace information for all scheduler calls that schedule tasks.
  43  */
  44 #define EXECINFO GNUNET_NO
  45
  46 /**
  47  * Depth of the traces collected via EXECINFO.
  48  */
  49 #define MAX_TRACE_DEPTH 50
  50 #endif
  51
  52 #define DEBUG_TASKS GNUNET_NO
  53
  54 /**
  55  * Should we figure out which tasks are delayed for a while
  56  * before they are run? (Consider using in combination with EXECINFO).
  57  */
  58 #define PROFILE_DELAYS GNUNET_NO
  59
  60 /**
  61  * Task that were in the queue for longer than this are reported if
  62  * PROFILE_DELAYS is active.
  63  */
  64 #define DELAY_THRESHOLD GNUNET_TIME_UNIT_SECONDS
  65
  66 /**
  67  * Linked list of pending tasks.
  68  */
  69 struct Task
  70 {
  71   /**
  72    * This is a linked list.
  73    */
  74   struct Task *next;
  75
  76   /**
  77    * Function to run when ready.
  78    */
  79   GNUNET_SCHEDULER_Task callback;
  80
  81   /**
  82    * Closure for the callback.
  83    */
  84   void *callback_cls;
  85
  86   /**
  87    * Set of file descriptors this task is waiting
  88    * for for reading.  Once ready, this is updated
  89    * to reflect the set of file descriptors ready
  90    * for operation.
  91    */
  92   struct GNUNET_NETWORK_FDSet *read_set;
  93
  94   /**
  95    * Set of file descriptors this task is waiting for for writing.
  96    * Once ready, this is updated to reflect the set of file
  97    * descriptors ready for operation.
  98    */
  99   struct GNUNET_NETWORK_FDSet *write_set;
 100
 101   /**
 102    * Unique task identifier.
 103    */
 104   GNUNET_SCHEDULER_TaskIdentifier id;
 105
 106   /**
 107    * Identifier of a prerequisite task.
 108    */
 109   GNUNET_SCHEDULER_TaskIdentifier prereq_id;
 110
 111   /**
 112    * Absolute timeout value for the task, or
 113    * GNUNET_TIME_UNIT_FOREVER_ABS for "no timeout".
 114    */
 115   struct GNUNET_TIME_Absolute timeout;
 116
 117 #if PROFILE_DELAYS
 118   /**
 119    * When was the task scheduled?
 120    */
 121   struct GNUNET_TIME_Absolute start_time;
 122 #endif
 123
 124   /**
 125    * Why is the task ready?  Set after task is added to ready queue.
 126    * Initially set to zero.  All reasons that have already been
 127    * satisfied (i.e.  read or write ready) will be set over time.
 128    */
 129   enum GNUNET_SCHEDULER_Reason reason;
 130
 131   /**
 132    * Task priority.
 133    */
 134   enum GNUNET_SCHEDULER_Priority priority;
 135
 136 #if EXECINFO
 137   /**
 138    * Array of strings which make up a backtrace from the point when this
 139    * task was scheduled (essentially, who scheduled the task?)
 140    */
 141   char **backtrace_strings;
 142
 143   /**
 144    * Size of the backtrace_strings array
 145    */
 146   int num_backtrace_strings;
 147 #endif
 148
 149 };
 150
 151
 152 /**
 153  * Handle for the scheduling service.
 154  */
 155 struct GNUNET_SCHEDULER_Handle
 156 {
 157
 158   /**
 159    * List of tasks waiting for an event.
 160    */
 161   struct Task *pending;
 162
 163   /**
 164    * ID of the task that is running right now.
 165    */
 166   struct Task *active_task;
 167
 168   /**
 169    * List of tasks ready to run right now,
 170    * grouped by importance.
 171    */
 172   struct Task *ready[GNUNET_SCHEDULER_PRIORITY_COUNT];
 173
 174   /**
 175    * Identity of the last task queued.  Incremented for each task to
 176    * generate a unique task ID (it is virtually impossible to start
 177    * more than 2^64 tasks during the lifetime of a process).
 178    */
 179   GNUNET_SCHEDULER_TaskIdentifier last_id;
 180
 181   /**
 182    * Highest number so that all tasks with smaller identifiers
 183    * have already completed.  Also the lowest number of a task
 184    * still waiting to be executed.
 185    */
 186   GNUNET_SCHEDULER_TaskIdentifier lowest_pending_id;
 187
 188   /**
 189    * Number of tasks on the ready list.
 190    */
 191   unsigned int ready_count;
 192
 193   /**
 194    * How many tasks have we run so far?
 195    */
 196   unsigned long long tasks_run;
 197
 198   /**
 199    * Priority of the task running right now.  Only
 200    * valid while a task is running.
 201    */
 202   enum GNUNET_SCHEDULER_Priority current_priority;
 203
 204   /**
 205    * How 'nice' are we right now?
 206    */
 207   int nice_level;
 208
 209 };
 210
 211
 212 /**
 213  * Check that the given priority is legal (and return it).
 214  *
 215  * @param p priority value to check
 216  * @return p on success, 0 on error
 217  */
 218 static enum GNUNET_SCHEDULER_Priority
 219 check_priority (enum GNUNET_SCHEDULER_Priority p)
 220 {
 221   if ((p >= 0) && (p < GNUNET_SCHEDULER_PRIORITY_COUNT))
 222     return p;
 223   GNUNET_assert (0);
 224   return 0;                     /* make compiler happy */
 225 }
 226
 227
 228 /**
 229  * Is a task with this identifier still pending?  Also updates
 230  * "lowest_pending_id" as a side-effect (for faster checks in the
 231  * future), but only if the return value is "GNUNET_NO" (and
 232  * the "lowest_pending_id" check failed).
 233  *
 234  * @param sched the scheduler
 235  * @param id which task are we checking for
 236  * @return GNUNET_YES if so, GNUNET_NO if not
 237  */
 238 static int
 239 is_pending (struct GNUNET_SCHEDULER_Handle *sched,
 240             GNUNET_SCHEDULER_TaskIdentifier id)
 241 {
 242   struct Task *pos;
 243   enum GNUNET_SCHEDULER_Priority p;
 244   GNUNET_SCHEDULER_TaskIdentifier min;
 245
 246   if (id < sched->lowest_pending_id)
 247     return GNUNET_NO;
 248   min = -1;                     /* maximum value */
 249   pos = sched->pending;
 250   while (pos != NULL)
 251     {
 252       if (pos->id == id)
 253         return GNUNET_YES;
 254       if (pos->id < min)
 255         min = pos->id;
 256       pos = pos->next;
 257     }
 258   for (p = 0; p < GNUNET_SCHEDULER_PRIORITY_COUNT; p++)
 259     {
 260       pos = sched->ready[p];
 261       while (pos != NULL)
 262         {
 263           if (pos->id == id)
 264             return GNUNET_YES;
 265           if (pos->id < min)
 266             min = pos->id;
 267           pos = pos->next;
 268         }
 269     }
 270   sched->lowest_pending_id = min;
 271   return GNUNET_NO;
 272 }
 273
 274
 275 /**
 276  * Update all sets and timeout for select.
 277  *
 278  * @param sched the scheduler
 279  * @param rs read-set, set to all FDs we would like to read (updated)
 280  * @param ws write-set, set to all FDs we would like to write (updated)
 281  * @param timeout next timeout (updated)
 282  */
 283 static void
 284 update_sets (struct GNUNET_SCHEDULER_Handle *sched,
 285              struct GNUNET_NETWORK_FDSet *rs,
 286              struct GNUNET_NETWORK_FDSet *ws,
 287              struct GNUNET_TIME_Relative *timeout)
 288 {
 289   struct Task *pos;
 290
 291   pos = sched->pending;
 292   while (pos != NULL)
 293     {
 294       if ((pos->prereq_id != GNUNET_SCHEDULER_NO_TASK) &&
 295           (GNUNET_YES == is_pending (sched, pos->prereq_id)))
 296         {
 297           pos = pos->next;
 298           continue;
 299         }
 300
 301       if (pos->timeout.value != GNUNET_TIME_UNIT_FOREVER_ABS.value)
 302         {
 303           struct GNUNET_TIME_Relative to;
 304
 305           to = GNUNET_TIME_absolute_get_remaining (pos->timeout);
 306           if (timeout->value > to.value)
 307             *timeout = to;
 308         }
 309       if (pos->read_set != NULL)
 310         GNUNET_NETWORK_fdset_add (rs, pos->read_set);
 311       if (pos->write_set != NULL)
 312         GNUNET_NETWORK_fdset_add (ws, pos->write_set);
 313       if (pos->reason != 0)
 314         *timeout = GNUNET_TIME_UNIT_ZERO;
 315       pos = pos->next;
 316     }
 317 }
 318
 319
 320 /**
 321  * Check if the ready set overlaps with the set we want to have ready.
 322  * If so, update the want set (set all FDs that are ready).  If not,
 323  * return GNUNET_NO.
 324  *
 325  * @param ready set that is ready
 326  * @param want set that we want to be ready
 327  * @return GNUNET_YES if there was some overlap
 328  */
 329 static int
 330 set_overlaps (const struct GNUNET_NETWORK_FDSet *ready,
 331               struct GNUNET_NETWORK_FDSet *want)
 332 {
 333   if (NULL == want)
 334     return GNUNET_NO;
 335   if (GNUNET_NETWORK_fdset_overlap (ready, want))
 336     {
 337       /* copy all over (yes, there maybe unrelated bits,
 338          but this should not hurt well-written clients) */
 339       GNUNET_NETWORK_fdset_copy (want, ready);
 340       return GNUNET_YES;
 341     }
 342   return GNUNET_NO;
 343 }
 344
 345
 346 /**
 347  * Check if the given task is eligible to run now.
 348  * Also set the reason why it is eligible.
 349  *
 350  * @param sched the scheduler
 351  * @param task task to check if it is ready
 352  * @param now the current time
 353  * @param rs set of FDs ready for reading
 354  * @param ws set of FDs ready for writing
 355  * @return GNUNET_YES if we can run it, GNUNET_NO if not.
 356  */
 357 static int
 358 is_ready (struct GNUNET_SCHEDULER_Handle *sched,
 359           struct Task *task,
 360           struct GNUNET_TIME_Absolute now,
 361           const struct GNUNET_NETWORK_FDSet *rs,
 362           const struct GNUNET_NETWORK_FDSet *ws)
 363 {
 364   if (now.value >= task->timeout.value)
 365     task->reason |= GNUNET_SCHEDULER_REASON_TIMEOUT;
 366   if ((0 == (task->reason & GNUNET_SCHEDULER_REASON_READ_READY)) &&
 367       (rs != NULL) && (set_overlaps (rs, task->read_set)))
 368     task->reason |= GNUNET_SCHEDULER_REASON_READ_READY;
 369   if ((0 == (task->reason & GNUNET_SCHEDULER_REASON_WRITE_READY)) &&
 370       (ws != NULL) && (set_overlaps (ws, task->write_set)))
 371     task->reason |= GNUNET_SCHEDULER_REASON_WRITE_READY;
 372   if (task->reason == 0)
 373     return GNUNET_NO;           /* not ready */
 374   if (task->prereq_id != GNUNET_SCHEDULER_NO_TASK)
 375     {
 376       if (GNUNET_YES == is_pending (sched, task->prereq_id))
 377         return GNUNET_NO;       /* prereq waiting */
 378       task->reason |= GNUNET_SCHEDULER_REASON_PREREQ_DONE;
 379     }
 380   return GNUNET_YES;
 381 }
 382
 383
 384 /**
 385  * Put a task that is ready for execution into the ready queue.
 386  *
 387  * @param handle the scheduler
 388  * @param task task ready for execution
 389  */
 390 static void
 391 queue_ready_task (struct GNUNET_SCHEDULER_Handle *handle, struct Task *task)
 392 {
 393   enum GNUNET_SCHEDULER_Priority p = task->priority;
 394   if (0 != (task->reason & GNUNET_SCHEDULER_REASON_SHUTDOWN))
 395     p = GNUNET_SCHEDULER_PRIORITY_SHUTDOWN;
 396   task->next = handle->ready[check_priority (p)];
 397   handle->ready[check_priority (p)] = task;
 398   handle->ready_count++;
 399 }
 400
 401
 402 /**
 403  * Check which tasks are ready and move them
 404  * to the respective ready queue.
 405  *
 406  * @param handle the scheduler
 407  * @param rs FDs ready for reading
 408  * @param ws FDs ready for writing
 409  */
 410 static void
 411 check_ready (struct GNUNET_SCHEDULER_Handle *handle,
 412              const struct GNUNET_NETWORK_FDSet *rs,
 413              const struct GNUNET_NETWORK_FDSet *ws)
 414 {
 415   struct Task *pos;
 416   struct Task *prev;
 417   struct Task *next;
 418   struct GNUNET_TIME_Absolute now;
 419
 420   now = GNUNET_TIME_absolute_get ();
 421   prev = NULL;
 422   pos = handle->pending;
 423   while (pos != NULL)
 424     {
 425 #if DEBUG_TASKS
 426       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 427                   "Checking readiness of task: %llu / %p\n",
 428                   pos->id, pos->callback_cls);
 429 #endif
 430       next = pos->next;
 431       if (GNUNET_YES == is_ready (handle, pos, now, rs, ws))
 432         {
 433           if (prev == NULL)
 434             handle->pending = next;
 435           else
 436             prev->next = next;
 437           queue_ready_task (handle, pos);
 438           pos = next;
 439           continue;
 440         }
 441       prev = pos;
 442       pos = next;
 443     }
 444 }
 445
 446
 447 /**
 448  * Request the shutdown of a scheduler.  Marks all currently
 449  * pending tasks as ready because of shutdown.  This will
 450  * cause all tasks to run (as soon as possible, respecting
 451  * priorities and prerequisite tasks).  Note that tasks
 452  * scheduled AFTER this call may still be delayed arbitrarily.
 453  *
 454  * @param sched the scheduler
 455  */
 456 void
 457 GNUNET_SCHEDULER_shutdown (struct GNUNET_SCHEDULER_Handle *sched)
 458 {
 459   struct Task *pos;
 460   int i;
 461
 462   pos = sched->pending;
 463   while (pos != NULL)
 464     {
 465       pos->reason |= GNUNET_SCHEDULER_REASON_SHUTDOWN;
 466       /* we don't move the task into the ready queue yet; check_ready
 467          will do that later, possibly adding additional
 468          readiness-factors */
 469       pos = pos->next;
 470     }
 471   for (i=0;i<GNUNET_SCHEDULER_PRIORITY_COUNT;i++)
 472     {
 473       pos = sched->ready[i];
 474       while (pos != NULL)
 475         {
 476           pos->reason |= GNUNET_SCHEDULER_REASON_SHUTDOWN;
 477           /* we don't move the task into the ready queue yet; check_ready
 478              will do that later, possibly adding additional
 479              readiness-factors */
 480           pos = pos->next;
 481         }
 482     }
 483 }
 484
 485
 486 /**
 487  * Destroy a task (release associated resources)
 488  *
 489  * @param t task to destroy
 490  */
 491 static void
 492 destroy_task (struct Task *t)
 493 {
 494   if (NULL != t->read_set)
 495     GNUNET_NETWORK_fdset_destroy (t->read_set);
 496   if (NULL != t->write_set)
 497     GNUNET_NETWORK_fdset_destroy (t->write_set);
 498 #if EXECINFO
 499   GNUNET_free (t->backtrace_strings);
 500 #endif
 501   GNUNET_free (t);
 502 }
 503
 504
 505 /**
 506  * Run at least one task in the highest-priority queue that is not
 507  * empty.  Keep running tasks until we are either no longer running
 508  * "URGENT" tasks or until we have at least one "pending" task (which
 509  * may become ready, hence we should select on it).  Naturally, if
 510  * there are no more ready tasks, we also return.
 511  *
 512  * @param sched the scheduler
 513  */
 514 static void
 515 run_ready (struct GNUNET_SCHEDULER_Handle *sched)
 516 {
 517   enum GNUNET_SCHEDULER_Priority p;
 518   struct Task *pos;
 519   struct GNUNET_SCHEDULER_TaskContext tc;
 520
 521   do
 522     {
 523       if (sched->ready_count == 0)
 524         return;
 525       GNUNET_assert (sched->ready[GNUNET_SCHEDULER_PRIORITY_KEEP] == NULL);
 526       /* yes, p>0 is correct, 0 is "KEEP" which should
 527          always be an empty queue (see assertion)! */
 528       for (p = GNUNET_SCHEDULER_PRIORITY_COUNT - 1; p > 0; p--)
 529         {
 530           pos = sched->ready[p];
 531           if (pos != NULL)
 532             break;
 533         }
 534       GNUNET_assert (pos != NULL);      /* ready_count wrong? */
 535       sched->ready[p] = pos->next;
 536       sched->ready_count--;
 537       if (sched->current_priority != pos->priority)
 538         {
 539           sched->current_priority = pos->priority;
 540           (void) GNUNET_OS_set_process_priority (0, pos->priority);
 541         }
 542       sched->active_task = pos;
 543 #if PROFILE_DELAYS
 544       if (GNUNET_TIME_absolute_get_duration (pos->start_time).value >
 545           DELAY_THRESHOLD.value)
 546         {
 547           GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
 548                       "Task %u took %llums to be scheduled\n",
 549                       pos->id,
 550                       (unsigned long long) GNUNET_TIME_absolute_get_duration (pos->start_time).value);
 551         }
 552 #endif
 553       tc.sched = sched;
 554       tc.reason = pos->reason;
 555       tc.read_ready = pos->read_set;
 556       tc.write_ready = pos->write_set;
 557 #if DEBUG_TASKS
 558       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 559                   "Running task: %llu / %p\n", pos->id, pos->callback_cls);
 560 #endif
 561       pos->callback (pos->callback_cls, &tc);
 562 #if EXECINFO
 563       int i;
 564       for (i=0;i<pos->num_backtrace_strings;i++)
 565         GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
 566                     "Task %u trace %d: %s\n",
 567                     pos->id,
 568                     i,
 569                     pos->backtrace_strings[i]);
 570 #endif
 571       sched->active_task = NULL;
 572       destroy_task (pos);
 573       sched->tasks_run++;
 574     }
 575   while ((sched->pending == NULL) || (p == GNUNET_SCHEDULER_PRIORITY_URGENT));
 576 }
 577
 578 /**
 579  * Pipe used to communicate shutdown via signal.
 580  */
 581 static struct GNUNET_DISK_PipeHandle *sigpipe;
 582
 583
 584 /**
 585  * Signal handler called for signals that should cause us to shutdown.
 586  */
 587 static void
 588 sighandler_shutdown ()
 589 {
 590   static char c;
 591
 592   GNUNET_DISK_file_write (GNUNET_DISK_pipe_handle
 593                           (sigpipe, GNUNET_DISK_PIPE_END_WRITE), &c,
 594                           sizeof (c));
 595 }
 596
 597
 598 /**
 599  * Initialize and run scheduler.  This function will return when all
 600  * tasks have completed.  On systems with signals, receiving a SIGTERM
 601  * (and other similar signals) will cause "GNUNET_SCHEDULER_shutdown"
 602  * to be run after the active task is complete.  As a result, SIGTERM
 603  * causes all active tasks to be scheduled with reason
 604  * "GNUNET_SCHEDULER_REASON_SHUTDOWN".  (However, tasks added
 605  * afterwards will execute normally!). Note that any particular signal
 606  * will only shut down one scheduler; applications should always only
 607  * create a single scheduler.
 608  *
 609  * @param task task to run immediately
 610  * @param task_cls closure of task
 611  */
 612 void
 613 GNUNET_SCHEDULER_run (GNUNET_SCHEDULER_Task task, void *task_cls)
 614 {
 615   struct GNUNET_SCHEDULER_Handle sched;
 616   struct GNUNET_NETWORK_FDSet *rs;
 617   struct GNUNET_NETWORK_FDSet *ws;
 618   struct GNUNET_TIME_Relative timeout;
 619   int ret;
 620   struct GNUNET_SIGNAL_Context *shc_int;
 621   struct GNUNET_SIGNAL_Context *shc_term;
 622   struct GNUNET_SIGNAL_Context *shc_quit;
 623   struct GNUNET_SIGNAL_Context *shc_hup;
 624   unsigned long long last_tr;
 625   unsigned int busy_wait_warning;
 626   const struct GNUNET_DISK_FileHandle *pr;
 627   char c;
 628
 629   rs = GNUNET_NETWORK_fdset_create ();
 630   ws = GNUNET_NETWORK_fdset_create ();
 631   GNUNET_assert (sigpipe == NULL);
 632   sigpipe = GNUNET_DISK_pipe (GNUNET_NO);
 633   GNUNET_assert (sigpipe != NULL);
 634   pr = GNUNET_DISK_pipe_handle (sigpipe, GNUNET_DISK_PIPE_END_READ);
 635   GNUNET_assert (pr != NULL);
 636   shc_int = GNUNET_SIGNAL_handler_install (SIGINT, &sighandler_shutdown);
 637   shc_term = GNUNET_SIGNAL_handler_install (SIGTERM, &sighandler_shutdown);
 638 #ifndef MINGW
 639   shc_quit = GNUNET_SIGNAL_handler_install (SIGQUIT, &sighandler_shutdown);
 640   shc_hup = GNUNET_SIGNAL_handler_install (SIGHUP, &sighandler_shutdown);
 641 #endif
 642   memset (&sched, 0, sizeof (sched));
 643   sched.current_priority = GNUNET_SCHEDULER_PRIORITY_DEFAULT;
 644   GNUNET_SCHEDULER_add_continuation (&sched,
 645                                      task,
 646                                      task_cls,
 647                                      GNUNET_SCHEDULER_REASON_STARTUP);
 648   last_tr = 0;
 649   busy_wait_warning = 0;
 650   while ((sched.pending != NULL) || (sched.ready_count > 0))
 651     {
 652       GNUNET_NETWORK_fdset_zero (rs);
 653       GNUNET_NETWORK_fdset_zero (ws);
 654       timeout = GNUNET_TIME_UNIT_FOREVER_REL;
 655       update_sets (&sched, rs, ws, &timeout);
 656       GNUNET_NETWORK_fdset_handle_set (rs, pr);
 657       if (sched.ready_count > 0)
 658         {
 659           /* no blocking, more work already ready! */
 660           timeout = GNUNET_TIME_UNIT_ZERO;
 661         }
 662       ret = GNUNET_NETWORK_socket_select (rs, ws, NULL, timeout);
 663       if (ret == GNUNET_SYSERR)
 664         {
 665           if (errno == EINTR)
 666             continue;
 667
 668           GNUNET_log_strerror (GNUNET_ERROR_TYPE_ERROR, "select");
 669 #ifndef MINGW
 670 #if USE_LSOF
 671           char lsof[512];
 672           snprintf (lsof, sizeof (lsof), "lsof -p %d", getpid());
 673           close (1);
 674           dup2 (2, 1);
 675           system (lsof);
 676 #endif
 677 #endif
 678           abort ();
 679           break;
 680         }
 681       if (GNUNET_NETWORK_fdset_handle_isset (rs, pr))
 682         {
 683           /* consume the signal */
 684           GNUNET_DISK_file_read (pr, &c, sizeof (c));
 685           /* mark all active tasks as ready due to shutdown */
 686           GNUNET_SCHEDULER_shutdown (&sched);
 687         }
 688       if (last_tr == sched.tasks_run)
 689         {
 690           busy_wait_warning++;
 691         }
 692       else
 693         {
 694           last_tr = sched.tasks_run;
 695           busy_wait_warning = 0;
 696         }
 697       if ((ret == 0) && (timeout.value == 0) && (busy_wait_warning > 16))
 698         {
 699           GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
 700                       _("Looks like we're busy waiting...\n"));
 701           sleep (1);            /* mitigate */
 702         }
 703       check_ready (&sched, rs, ws);
 704       run_ready (&sched);
 705     }
 706   GNUNET_SIGNAL_handler_uninstall (shc_int);
 707   GNUNET_SIGNAL_handler_uninstall (shc_term);
 708 #ifndef MINGW
 709   GNUNET_SIGNAL_handler_uninstall (shc_quit);
 710   GNUNET_SIGNAL_handler_uninstall (shc_hup);
 711 #endif
 712   GNUNET_DISK_pipe_close (sigpipe);
 713   sigpipe = NULL;
 714   GNUNET_NETWORK_fdset_destroy (rs);
 715   GNUNET_NETWORK_fdset_destroy (ws);
 716 }
 717
 718
 719 /**
 720  * Obtain the reason code for why the current task was
 721  * started.  Will return the same value as
 722  * the GNUNET_SCHEDULER_TaskContext's reason field.
 723  *
 724  * @param sched scheduler to query
 725  * @return reason(s) why the current task is run
 726  */
 727 enum GNUNET_SCHEDULER_Reason
 728 GNUNET_SCHEDULER_get_reason (struct GNUNET_SCHEDULER_Handle *sched)
 729 {
 730   return sched->active_task->reason;
 731 }
 732
 733
 734 /**
 735  * Get information about the current load of this scheduler.  Use this
 736  * function to determine if an elective task should be added or simply
 737  * dropped (if the decision should be made based on the number of
 738  * tasks ready to run).
 739  *
 740  * @param sched scheduler to query
 741  * @param p priority level to look at
 742  * @return number of tasks pending right now
 743  */
 744 unsigned int
 745 GNUNET_SCHEDULER_get_load (struct GNUNET_SCHEDULER_Handle *sched,
 746                            enum GNUNET_SCHEDULER_Priority p)
 747 {
 748   struct Task *pos;
 749   unsigned int ret;
 750
 751   if (p == GNUNET_SCHEDULER_PRIORITY_COUNT)
 752     return sched->ready_count;
 753   if (p == GNUNET_SCHEDULER_PRIORITY_KEEP)
 754     p = sched->current_priority;
 755   ret = 0;
 756   pos = sched->ready[p];
 757   while (pos != NULL)
 758     {
 759       pos = pos->next;
 760       ret++;
 761     }
 762   return ret;
 763 }
 764
 765
 766 /**
 767  * Cancel the task with the specified identifier.
 768  * The task must not yet have run.
 769  *
 770  * @param sched scheduler to use
 771  * @param task id of the task to cancel
 772  * @return original closure of the task
 773  */
 774 void *
 775 GNUNET_SCHEDULER_cancel (struct GNUNET_SCHEDULER_Handle *sched,
 776                          GNUNET_SCHEDULER_TaskIdentifier task)
 777 {
 778   struct Task *t;
 779   struct Task *prev;
 780   enum GNUNET_SCHEDULER_Priority p;
 781   void *ret;
 782 #if EXECINFO
 783   int i;
 784 #endif
 785   prev = NULL;
 786   t = sched->pending;
 787   while (t != NULL)
 788     {
 789       if (t->id == task)
 790         break;
 791       prev = t;
 792       t = t->next;
 793     }
 794   p = 0;
 795   while (t == NULL)
 796     {
 797       p++;
 798       GNUNET_assert (p < GNUNET_SCHEDULER_PRIORITY_COUNT);
 799       prev = NULL;
 800       t = sched->ready[p];
 801       while (t != NULL)
 802         {
 803           if (t->id == task)
 804             {
 805               sched->ready_count--;
 806               break;
 807             }
 808           prev = t;
 809           t = t->next;
 810         }
 811     }
 812   if (prev == NULL)
 813     {
 814       if (p == 0)
 815         sched->pending = t->next;
 816       else
 817         sched->ready[p] = t->next;
 818     }
 819   else
 820     prev->next = t->next;
 821   ret = t->callback_cls;
 822 #if DEBUG_TASKS
 823   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 824               "Canceling task: %llu / %p\n", task, t->callback_cls);
 825 #endif
 826   destroy_task (t);
 827   return ret;
 828 }
 829
 830
 831 /**
 832  * Continue the current execution with the given function.  This is
 833  * similar to the other "add" functions except that there is no delay
 834  * and the reason code can be specified.
 835  *
 836  * @param sched scheduler to use
 837  * @param task main function of the task
 838  * @param task_cls closure for 'main'
 839  * @param reason reason for task invocation
 840  */
 841 void
 842 GNUNET_SCHEDULER_add_continuation (struct GNUNET_SCHEDULER_Handle *sched,
 843                                    GNUNET_SCHEDULER_Task task,
 844                                    void *task_cls,
 845                                    enum GNUNET_SCHEDULER_Reason reason)
 846 {
 847   struct Task *t;
 848 #if EXECINFO
 849   void *backtrace_array[50];
 850 #endif
 851   t = GNUNET_malloc (sizeof (struct Task));
 852 #if EXECINFO
 853   t->num_backtrace_strings = backtrace(backtrace_array, 50);
 854   t->backtrace_strings = backtrace_symbols(backtrace_array, t->num_backtrace_strings);
 855 #endif
 856   t->callback = task;
 857   t->callback_cls = task_cls;
 858   t->id = ++sched->last_id;
 859 #if PROFILE_DELAYS
 860   t->start_time = GNUNET_TIME_absolute_get ();
 861 #endif
 862   t->reason = reason;
 863   t->priority = sched->current_priority;
 864 #if DEBUG_TASKS
 865   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 866               "Adding continuation task: %llu / %p\n",
 867               t->id, t->callback_cls);
 868 #endif
 869   queue_ready_task (sched, t);
 870 }
 871
 872
 873
 874 /**
 875  * Schedule a new task to be run after the specified prerequisite task
 876  * has completed. It will be run with the priority of the calling
 877  * task.
 878  *
 879  * @param sched scheduler to use
 880  * @param prerequisite_task run this task after the task with the given
 881  *        task identifier completes (and any of our other
 882  *        conditions, such as delay, read or write-readiness
 883  *        are satisfied).  Use  GNUNET_SCHEDULER_NO_TASK to not have any dependency
 884  *        on completion of other tasks (this will cause the task to run as
 885  *        soon as possible).
 886  * @param task main function of the task
 887  * @param task_cls closure of task
 888  * @return unique task identifier for the job
 889  *         only valid until "task" is started!
 890  */
 891 GNUNET_SCHEDULER_TaskIdentifier
 892 GNUNET_SCHEDULER_add_after (struct GNUNET_SCHEDULER_Handle *sched,
 893                             GNUNET_SCHEDULER_TaskIdentifier prerequisite_task,
 894                             GNUNET_SCHEDULER_Task task, void *task_cls)
 895 {
 896   return GNUNET_SCHEDULER_add_select (sched,
 897                                       GNUNET_SCHEDULER_PRIORITY_KEEP,
 898                                       prerequisite_task,
 899                                       GNUNET_TIME_UNIT_ZERO,
 900                                       NULL, NULL, task, task_cls);
 901 }
 902
 903
 904 /**
 905  * Schedule a new task to be run with a specified priority.
 906  *
 907  * @param sched scheduler to use
 908  * @param prio how important is the new task?
 909  * @param task main function of the task
 910  * @param task_cls closure of task
 911  * @return unique task identifier for the job
 912  *         only valid until "task" is started!
 913  */
 914 GNUNET_SCHEDULER_TaskIdentifier
 915 GNUNET_SCHEDULER_add_with_priority (struct GNUNET_SCHEDULER_Handle * sched,
 916                                     enum GNUNET_SCHEDULER_Priority prio,
 917                                     GNUNET_SCHEDULER_Task task,
 918                                     void *task_cls)
 919 {
 920   return GNUNET_SCHEDULER_add_select (sched,
 921                                       prio,
 922                                       GNUNET_SCHEDULER_NO_TASK,
 923                                       GNUNET_TIME_UNIT_ZERO,
 924                                       NULL, NULL, task, task_cls);
 925 }
 926
 927
 928
 929 /**
 930  * Schedule a new task to be run with a specified delay.  The task
 931  * will be scheduled for execution once the delay has expired. It
 932  * will be run with the priority of the calling task.
 933  *
 934  * @param sched scheduler to use
 935  * @param delay when should this operation time out? Use
 936  *        GNUNET_TIME_UNIT_FOREVER_REL for "on shutdown"
 937  * @param task main function of the task
 938  * @param task_cls closure of task
 939  * @return unique task identifier for the job
 940  *         only valid until "task" is started!
 941  */
 942 GNUNET_SCHEDULER_TaskIdentifier
 943 GNUNET_SCHEDULER_add_delayed (struct GNUNET_SCHEDULER_Handle * sched,
 944                               struct GNUNET_TIME_Relative delay,
 945                               GNUNET_SCHEDULER_Task task, void *task_cls)
 946 {
 947   return GNUNET_SCHEDULER_add_select (sched,
 948                                       GNUNET_SCHEDULER_PRIORITY_KEEP,
 949                                       GNUNET_SCHEDULER_NO_TASK, delay,
 950                                       NULL, NULL, task, task_cls);
 951 }
 952
 953
 954
 955 /**
 956  * Schedule a new task to be run as soon as possible. The task
 957  * will be run with the priority of the calling task.
 958  *
 959  * @param sched scheduler to use
 960  * @param task main function of the task
 961  * @param task_cls closure of task
 962  * @return unique task identifier for the job
 963  *         only valid until "task" is started!
 964  */
 965 GNUNET_SCHEDULER_TaskIdentifier
 966 GNUNET_SCHEDULER_add_now (struct GNUNET_SCHEDULER_Handle *sched,
 967                           GNUNET_SCHEDULER_Task task,
 968                           void *task_cls)
 969 {
 970   return GNUNET_SCHEDULER_add_select (sched,
 971                                       GNUNET_SCHEDULER_PRIORITY_KEEP,
 972                                       GNUNET_SCHEDULER_NO_TASK,
 973                                       GNUNET_TIME_UNIT_ZERO,
 974                                       NULL, NULL, task, task_cls);
 975 }
 976
 977
 978
 979 /**
 980  * Schedule a new task to be run with a specified delay or when the
 981  * specified file descriptor is ready for reading.  The delay can be
 982  * used as a timeout on the socket being ready.  The task will be
 983  * scheduled for execution once either the delay has expired or the
 984  * socket operation is ready.  It will be run with the priority of
 985  * the calling task.
 986  *
 987  * @param sched scheduler to use
 988  * @param delay when should this operation time out? Use
 989  *        GNUNET_TIME_UNIT_FOREVER_REL for "on shutdown"
 990  * @param rfd read file-descriptor
 991  * @param task main function of the task
 992  * @param task_cls closure of task
 993  * @return unique task identifier for the job
 994  *         only valid until "task" is started!
 995  */
 996 GNUNET_SCHEDULER_TaskIdentifier
 997 GNUNET_SCHEDULER_add_read_net (struct GNUNET_SCHEDULER_Handle * sched,
 998                                struct GNUNET_TIME_Relative delay,
 999                                struct GNUNET_NETWORK_Handle * rfd,
1000                                GNUNET_SCHEDULER_Task task, void *task_cls)
1001 {
1002   struct GNUNET_NETWORK_FDSet *rs;
1003   GNUNET_SCHEDULER_TaskIdentifier ret;
1004
1005   GNUNET_assert (rfd != NULL);
1006   rs = GNUNET_NETWORK_fdset_create ();
1007   GNUNET_NETWORK_fdset_set (rs, rfd);
1008   ret = GNUNET_SCHEDULER_add_select (sched,
1009                                      GNUNET_SCHEDULER_PRIORITY_KEEP,
1010                                      GNUNET_SCHEDULER_NO_TASK,
1011                                      delay, rs, NULL, task, task_cls);
1012   GNUNET_NETWORK_fdset_destroy (rs);
1013   return ret;
1014 }
1015
1016
1017 /**
1018  * Schedule a new task to be run with a specified delay or when the
1019  * specified file descriptor is ready for writing.  The delay can be
1020  * used as a timeout on the socket being ready.  The task will be
1021  * scheduled for execution once either the delay has expired or the
1022  * socket operation is ready.  It will be run with the priority of
1023  * the calling task.
1024  *
1025  * @param sched scheduler to use
1026  * @param delay when should this operation time out? Use
1027  *        GNUNET_TIME_UNIT_FOREVER_REL for "on shutdown"
1028  * @param wfd write file-descriptor
1029  * @param task main function of the task
1030  * @param task_cls closure of task
1031  * @return unique task identifier for the job
1032  *         only valid until "task" is started!
1033  */
1034 GNUNET_SCHEDULER_TaskIdentifier
1035 GNUNET_SCHEDULER_add_write_net (struct GNUNET_SCHEDULER_Handle * sched,
1036                                 struct GNUNET_TIME_Relative delay,
1037                                 struct GNUNET_NETWORK_Handle * wfd,
1038                                 GNUNET_SCHEDULER_Task task, void *task_cls)
1039 {
1040   struct GNUNET_NETWORK_FDSet *ws;
1041   GNUNET_SCHEDULER_TaskIdentifier ret;
1042
1043   GNUNET_assert (wfd != NULL);
1044   ws = GNUNET_NETWORK_fdset_create ();
1045   GNUNET_NETWORK_fdset_set (ws, wfd);
1046   ret = GNUNET_SCHEDULER_add_select (sched,
1047                                      GNUNET_SCHEDULER_PRIORITY_KEEP,
1048                                      GNUNET_SCHEDULER_NO_TASK, delay,
1049                                      NULL, ws, task, task_cls);
1050   GNUNET_NETWORK_fdset_destroy (ws);
1051   return ret;
1052 }
1053
1054
1055 /**
1056  * Schedule a new task to be run with a specified delay or when the
1057  * specified file descriptor is ready for reading.  The delay can be
1058  * used as a timeout on the socket being ready.  The task will be
1059  * scheduled for execution once either the delay has expired or the
1060  * socket operation is ready. It will be run with the priority of
1061  * the calling task.
1062  *
1063  * @param sched scheduler to use
1064  * @param delay when should this operation time out? Use
1065  *        GNUNET_TIME_UNIT_FOREVER_REL for "on shutdown"
1066  * @param rfd read file-descriptor
1067  * @param task main function of the task
1068  * @param task_cls closure of task
1069  * @return unique task identifier for the job
1070  *         only valid until "task" is started!
1071  */
1072 GNUNET_SCHEDULER_TaskIdentifier
1073 GNUNET_SCHEDULER_add_read_file (struct GNUNET_SCHEDULER_Handle * sched,
1074                                 struct GNUNET_TIME_Relative delay,
1075                                 const struct GNUNET_DISK_FileHandle * rfd,
1076                                 GNUNET_SCHEDULER_Task task, void *task_cls)
1077 {
1078   struct GNUNET_NETWORK_FDSet *rs;
1079   GNUNET_SCHEDULER_TaskIdentifier ret;
1080
1081   GNUNET_assert (rfd != NULL);
1082   rs = GNUNET_NETWORK_fdset_create ();
1083   GNUNET_NETWORK_fdset_handle_set (rs, rfd);
1084   ret = GNUNET_SCHEDULER_add_select (sched,
1085                                      GNUNET_SCHEDULER_PRIORITY_KEEP,
1086                                      GNUNET_SCHEDULER_NO_TASK, delay,
1087                                      rs, NULL, task, task_cls);
1088   GNUNET_NETWORK_fdset_destroy (rs);
1089   return ret;
1090 }
1091
1092
1093 /**
1094  * Schedule a new task to be run with a specified delay or when the
1095  * specified file descriptor is ready for writing.  The delay can be
1096  * used as a timeout on the socket being ready.  The task will be
1097  * scheduled for execution once either the delay has expired or the
1098  * socket operation is ready. It will be run with the priority of
1099  * the calling task.
1100  *
1101  * @param sched scheduler to use
1102  * @param delay when should this operation time out? Use
1103  *        GNUNET_TIME_UNIT_FOREVER_REL for "on shutdown"
1104  * @param wfd write file-descriptor
1105  * @param task main function of the task
1106  * @param task_cls closure of task
1107  * @return unique task identifier for the job
1108  *         only valid until "task" is started!
1109  */
1110 GNUNET_SCHEDULER_TaskIdentifier
1111 GNUNET_SCHEDULER_add_write_file (struct GNUNET_SCHEDULER_Handle * sched,
1112                                  struct GNUNET_TIME_Relative delay,
1113                                  const struct GNUNET_DISK_FileHandle * wfd,
1114                                  GNUNET_SCHEDULER_Task task, void *task_cls)
1115 {
1116   struct GNUNET_NETWORK_FDSet *ws;
1117   GNUNET_SCHEDULER_TaskIdentifier ret;
1118
1119   GNUNET_assert (wfd != NULL);
1120   ws = GNUNET_NETWORK_fdset_create ();
1121   GNUNET_NETWORK_fdset_handle_set (ws, wfd);
1122   ret = GNUNET_SCHEDULER_add_select (sched,
1123                                      GNUNET_SCHEDULER_PRIORITY_KEEP,
1124                                      GNUNET_SCHEDULER_NO_TASK,
1125                                      delay, NULL, ws, task, task_cls);
1126   GNUNET_NETWORK_fdset_destroy (ws);
1127   return ret;
1128 }
1129
1130
1131
1132 /**
1133  * Schedule a new task to be run with a specified delay or when any of
1134  * the specified file descriptor sets is ready.  The delay can be used
1135  * as a timeout on the socket(s) being ready.  The task will be
1136  * scheduled for execution once either the delay has expired or any of
1137  * the socket operations is ready.  This is the most general
1138  * function of the "add" family.  Note that the "prerequisite_task"
1139  * must be satisfied in addition to any of the other conditions.  In
1140  * other words, the task will be started when
1141  * <code>
1142  * (prerequisite-run)
1143  * && (delay-ready
1144  *     || any-rs-ready
1145  *     || any-ws-ready
1146  *     || (shutdown-active && run-on-shutdown) )
1147  * </code>
1148  *
1149  * @param sched scheduler to use
1150  * @param prio how important is this task?
1151  * @param prerequisite_task run this task after the task with the given
1152  *        task identifier completes (and any of our other
1153  *        conditions, such as delay, read or write-readiness
1154  *        are satisfied).  Use GNUNET_SCHEDULER_NO_TASK to not have any dependency
1155  *        on completion of other tasks.
1156  * @param delay how long should we wait? Use GNUNET_TIME_UNIT_FOREVER_REL for "forever",
1157  *        which means that the task will only be run after we receive SIGTERM
1158  * @param rs set of file descriptors we want to read (can be NULL)
1159  * @param ws set of file descriptors we want to write (can be NULL)
1160  * @param task main function of the task
1161  * @param task_cls closure of task
1162  * @return unique task identifier for the job
1163  *         only valid until "task" is started!
1164  */
1165 GNUNET_SCHEDULER_TaskIdentifier
1166 GNUNET_SCHEDULER_add_select (struct GNUNET_SCHEDULER_Handle * sched,
1167                              enum GNUNET_SCHEDULER_Priority prio,
1168                              GNUNET_SCHEDULER_TaskIdentifier
1169                              prerequisite_task,
1170                              struct GNUNET_TIME_Relative delay,
1171                              const struct GNUNET_NETWORK_FDSet * rs,
1172                              const struct GNUNET_NETWORK_FDSet * ws,
1173                              GNUNET_SCHEDULER_Task task, void *task_cls)
1174 {
1175   struct Task *t;
1176 #if EXECINFO
1177   void *backtrace_array[MAX_TRACE_DEPTH];
1178 #endif
1179
1180   GNUNET_assert (NULL != task);
1181   t = GNUNET_malloc (sizeof (struct Task));
1182   t->callback = task;
1183   t->callback_cls = task_cls;
1184 #if EXECINFO
1185   t->num_backtrace_strings = backtrace(backtrace_array, MAX_TRACE_DEPTH);
1186   t->backtrace_strings = backtrace_symbols(backtrace_array, t->num_backtrace_strings);
1187 #endif
1188   if (rs != NULL)
1189     {
1190       t->read_set = GNUNET_NETWORK_fdset_create ();
1191       GNUNET_NETWORK_fdset_copy (t->read_set, rs);
1192     }
1193   if (ws != NULL)
1194     {
1195       t->write_set = GNUNET_NETWORK_fdset_create ();
1196       GNUNET_NETWORK_fdset_copy (t->write_set, ws);
1197     }
1198   t->id = ++sched->last_id;
1199 #if PROFILE_DELAYS
1200   t->start_time = GNUNET_TIME_absolute_get ();
1201 #endif
1202   t->prereq_id = prerequisite_task;
1203   t->timeout = GNUNET_TIME_relative_to_absolute (delay);
1204   t->priority =
1205     check_priority ((prio ==
1206                      GNUNET_SCHEDULER_PRIORITY_KEEP) ? sched->current_priority
1207                     : prio);
1208   t->next = sched->pending;
1209   sched->pending = t;
1210 #if DEBUG_TASKS
1211   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1212               "Adding task: %llu / %p\n", t->id, t->callback_cls);
1213 #endif
1214 #if EXECINFO
1215   int i;
1216
1217   for (i=0;i<t->num_backtrace_strings;i++)
1218       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1219                   "Task %u trace %d: %s\n",
1220                   t->id,
1221                   i,
1222                   t->backtrace_strings[i]);
1223 #endif
1224   return t->id;
1225 }
1226
1227 /* end of scheduler.c */