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