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