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