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