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