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 #include "disk.h"
  33 #ifdef LINUX
  34 #include "execinfo.h"
  35
  36
  37 /**
  38  * Use lsof to generate file descriptor reports on select error?
  39  * (turn off for stable releases).
  40  */
  41 #define USE_LSOF GNUNET_YES
  42
  43 /**
  44  * Obtain trace information for all scheduler calls that schedule tasks.
  45  */
  46 #define EXECINFO GNUNET_NO
  47
  48 /**
  49  * Depth of the traces collected via EXECINFO.
  50  */
  51 #define MAX_TRACE_DEPTH 50
  52 #endif
  53
  54 #define DEBUG_TASKS GNUNET_NO
  55
  56 /**
  57  * Should we figure out which tasks are delayed for a while
  58  * before they are run? (Consider using in combination with EXECINFO).
  59  */
  60 #define PROFILE_DELAYS GNUNET_NO
  61
  62 /**
  63  * Task that were in the queue for longer than this are reported if
  64  * PROFILE_DELAYS is active.
  65  */
  66 #define DELAY_THRESHOLD GNUNET_TIME_UNIT_SECONDS
  67
  68 /**
  69  * Linked list of pending tasks.
  70  */
  71 struct Task
  72 {
  73   /**
  74    * This is a linked list.
  75    */
  76   struct Task *next;
  77
  78   /**
  79    * Function to run when ready.
  80    */
  81   GNUNET_SCHEDULER_Task callback;
  82
  83   /**
  84    * Closure for the callback.
  85    */
  86   void *callback_cls;
  87
  88   /**
  89    * Set of file descriptors this task is waiting
  90    * for for reading.  Once ready, this is updated
  91    * to reflect the set of file descriptors ready
  92    * for operation.
  93    */
  94   struct GNUNET_NETWORK_FDSet *read_set;
  95
  96   /**
  97    * Set of file descriptors this task is waiting for for writing.
  98    * Once ready, this is updated to reflect the set of file
  99    * descriptors ready for operation.
 100    */
 101   struct GNUNET_NETWORK_FDSet *write_set;
 102
 103   /**
 104    * Unique task identifier.
 105    */
 106   GNUNET_SCHEDULER_TaskIdentifier id;
 107
 108   /**
 109    * Identifier of a prerequisite task.
 110    */
 111   GNUNET_SCHEDULER_TaskIdentifier prereq_id;
 112
 113   /**
 114    * Absolute timeout value for the task, or
 115    * GNUNET_TIME_UNIT_FOREVER_ABS for "no timeout".
 116    */
 117   struct GNUNET_TIME_Absolute timeout;
 118
 119 #if PROFILE_DELAYS
 120   /**
 121    * When was the task scheduled?
 122    */
 123   struct GNUNET_TIME_Absolute start_time;
 124 #endif
 125
 126   /**
 127    * Why is the task ready?  Set after task is added to ready queue.
 128    * Initially set to zero.  All reasons that have already been
 129    * satisfied (i.e.  read or write ready) will be set over time.
 130    */
 131   enum GNUNET_SCHEDULER_Reason reason;
 132
 133   /**
 134    * Task priority.
 135    */
 136   enum GNUNET_SCHEDULER_Priority priority;
 137
 138   /**
 139    * Set if we only wait for reading from a single FD, otherwise -1.
 140    */
 141   int read_fd;
 142
 143   /**
 144    * Set if we only wait for writing to a single FD, otherwise -1.
 145    */
 146   int write_fd;
 147
 148 #if EXECINFO
 149   /**
 150    * Array of strings which make up a backtrace from the point when this
 151    * task was scheduled (essentially, who scheduled the task?)
 152    */
 153   char **backtrace_strings;
 154
 155   /**
 156    * Size of the backtrace_strings array
 157    */
 158   int num_backtrace_strings;
 159 #endif
 160
 161
 162 };
 163
 164
 165 /**
 166  * Handle for the scheduling service.
 167  */
 168 struct GNUNET_SCHEDULER_Handle
 169 {
 170
 171   /**
 172    * List of tasks waiting for an event.
 173    */
 174   struct Task *pending;
 175
 176   /**
 177    * List of tasks waiting ONLY for a timeout event.
 178    * Sorted by timeout (earliest first).  Used so that
 179    * we do not traverse the list of these tasks when
 180    * building select sets (we just look at the head
 181    * to determine the respective timeout ONCE).
 182    */
 183   struct Task *pending_timeout;
 184
 185   /**
 186    * Last inserted task waiting ONLY for a timeout event.
 187    * Used to (heuristically) speed up insertion.
 188    */
 189   struct Task *pending_timeout_last;
 190
 191   /**
 192    * ID of the task that is running right now.
 193    */
 194   struct Task *active_task;
 195
 196   /**
 197    * List of tasks ready to run right now,
 198    * grouped by importance.
 199    */
 200   struct Task *ready[GNUNET_SCHEDULER_PRIORITY_COUNT];
 201
 202   /**
 203    * Identity of the last task queued.  Incremented for each task to
 204    * generate a unique task ID (it is virtually impossible to start
 205    * more than 2^64 tasks during the lifetime of a process).
 206    */
 207   GNUNET_SCHEDULER_TaskIdentifier last_id;
 208
 209   /**
 210    * Highest number so that all tasks with smaller identifiers
 211    * have already completed.  Also the lowest number of a task
 212    * still waiting to be executed.
 213    */
 214   GNUNET_SCHEDULER_TaskIdentifier lowest_pending_id;
 215
 216   /**
 217    * Number of tasks on the ready list.
 218    */
 219   unsigned int ready_count;
 220
 221   /**
 222    * How many tasks have we run so far?
 223    */
 224   unsigned long long tasks_run;
 225
 226   /**
 227    * Priority of the task running right now.  Only
 228    * valid while a task is running.
 229    */
 230   enum GNUNET_SCHEDULER_Priority current_priority;
 231
 232   /**
 233    * Priority of the highest task added in the current select
 234    * iteration.
 235    */
 236   enum GNUNET_SCHEDULER_Priority max_priority_added;
 237
 238   /**
 239    * How 'nice' are we right now?
 240    */
 241   int nice_level;
 242
 243 };
 244
 245
 246 /**
 247  * Check that the given priority is legal (and return it).
 248  *
 249  * @param p priority value to check
 250  * @return p on success, 0 on error
 251  */
 252 static enum GNUNET_SCHEDULER_Priority
 253 check_priority (enum GNUNET_SCHEDULER_Priority p)
 254 {
 255   if ((p >= 0) && (p < GNUNET_SCHEDULER_PRIORITY_COUNT))
 256     return p;
 257   GNUNET_assert (0);
 258   return 0;                     /* make compiler happy */
 259 }
 260
 261
 262 /**
 263  * Is a task with this identifier still pending?  Also updates
 264  * "lowest_pending_id" as a side-effect (for faster checks in the
 265  * future), but only if the return value is "GNUNET_NO" (and
 266  * the "lowest_pending_id" check failed).
 267  *
 268  * @param sched the scheduler
 269  * @param id which task are we checking for
 270  * @return GNUNET_YES if so, GNUNET_NO if not
 271  */
 272 static int
 273 is_pending (struct GNUNET_SCHEDULER_Handle *sched,
 274             GNUNET_SCHEDULER_TaskIdentifier id)
 275 {
 276   struct Task *pos;
 277   enum GNUNET_SCHEDULER_Priority p;
 278   GNUNET_SCHEDULER_TaskIdentifier min;
 279
 280   if (id < sched->lowest_pending_id)
 281     return GNUNET_NO;
 282   min = -1;                     /* maximum value */
 283   pos = sched->pending;
 284   while (pos != NULL)
 285     {
 286       if (pos->id == id)
 287         return GNUNET_YES;
 288       if (pos->id < min)
 289         min = pos->id;
 290       pos = pos->next;
 291     }
 292   pos = sched->pending_timeout;
 293   while (pos != NULL)
 294     {
 295       if (pos->id == id)
 296         return GNUNET_YES;
 297       if (pos->id < min)
 298         min = pos->id;
 299       pos = pos->next;
 300     }
 301   for (p = 0; p < GNUNET_SCHEDULER_PRIORITY_COUNT; p++)
 302     {
 303       pos = sched->ready[p];
 304       while (pos != NULL)
 305         {
 306           if (pos->id == id)
 307             return GNUNET_YES;
 308           if (pos->id < min)
 309             min = pos->id;
 310           pos = pos->next;
 311         }
 312     }
 313   sched->lowest_pending_id = min;
 314   return GNUNET_NO;
 315 }
 316
 317
 318 /**
 319  * Update all sets and timeout for select.
 320  *
 321  * @param sched the scheduler
 322  * @param rs read-set, set to all FDs we would like to read (updated)
 323  * @param ws write-set, set to all FDs we would like to write (updated)
 324  * @param timeout next timeout (updated)
 325  */
 326 static void
 327 update_sets (struct GNUNET_SCHEDULER_Handle *sched,
 328              struct GNUNET_NETWORK_FDSet *rs,
 329              struct GNUNET_NETWORK_FDSet *ws,
 330              struct GNUNET_TIME_Relative *timeout)
 331 {
 332   struct Task *pos;
 333   struct GNUNET_TIME_Absolute now;
 334   struct GNUNET_TIME_Relative to;
 335
 336   now = GNUNET_TIME_absolute_get ();
 337   pos = sched->pending_timeout;
 338   if (pos != NULL)
 339     {
 340       to = GNUNET_TIME_absolute_get_difference (now, pos->timeout);
 341       if (timeout->rel_value > to.rel_value)
 342         *timeout = to;
 343       if (pos->reason != 0)
 344         *timeout = GNUNET_TIME_UNIT_ZERO;
 345     }
 346   pos = sched->pending;
 347   while (pos != NULL)
 348     {
 349       if ((pos->prereq_id != GNUNET_SCHEDULER_NO_TASK) &&
 350           (GNUNET_YES == is_pending (sched, pos->prereq_id)))
 351         {
 352           pos = pos->next;
 353           continue;
 354         }
 355       if (pos->timeout.abs_value != GNUNET_TIME_UNIT_FOREVER_ABS.abs_value)
 356         {
 357           to = GNUNET_TIME_absolute_get_difference (now, pos->timeout);
 358           if (timeout->rel_value > to.rel_value)
 359             *timeout = to;
 360         }
 361       if (pos->read_fd != -1)
 362         GNUNET_NETWORK_fdset_set_native (rs, pos->read_fd);
 363       if (pos->write_fd != -1)
 364         GNUNET_NETWORK_fdset_set_native (ws, pos->write_fd);
 365       if (pos->read_set != NULL)
 366         GNUNET_NETWORK_fdset_add (rs, pos->read_set);
 367       if (pos->write_set != NULL)
 368         GNUNET_NETWORK_fdset_add (ws, pos->write_set);
 369       if (pos->reason != 0)
 370         *timeout = GNUNET_TIME_UNIT_ZERO;
 371       pos = pos->next;
 372     }
 373 }
 374
 375
 376 /**
 377  * Check if the ready set overlaps with the set we want to have ready.
 378  * If so, update the want set (set all FDs that are ready).  If not,
 379  * return GNUNET_NO.
 380  *
 381  * @param ready set that is ready
 382  * @param want set that we want to be ready
 383  * @return GNUNET_YES if there was some overlap
 384  */
 385 static int
 386 set_overlaps (const struct GNUNET_NETWORK_FDSet *ready,
 387               struct GNUNET_NETWORK_FDSet *want)
 388 {
 389   if ( (NULL == want) || (NULL == ready) )
 390     return GNUNET_NO;
 391   if (GNUNET_NETWORK_fdset_overlap (ready, want))
 392     {
 393       /* copy all over (yes, there maybe unrelated bits,
 394          but this should not hurt well-written clients) */
 395       GNUNET_NETWORK_fdset_copy (want, ready);
 396       return GNUNET_YES;
 397     }
 398   return GNUNET_NO;
 399 }
 400
 401
 402 /**
 403  * Check if the given task is eligible to run now.
 404  * Also set the reason why it is eligible.
 405  *
 406  * @param sched the scheduler
 407  * @param task task to check if it is ready
 408  * @param now the current time
 409  * @param rs set of FDs ready for reading
 410  * @param ws set of FDs ready for writing
 411  * @return GNUNET_YES if we can run it, GNUNET_NO if not.
 412  */
 413 static int
 414 is_ready (struct GNUNET_SCHEDULER_Handle *sched,
 415           struct Task *task,
 416           struct GNUNET_TIME_Absolute now,
 417           const struct GNUNET_NETWORK_FDSet *rs,
 418           const struct GNUNET_NETWORK_FDSet *ws)
 419 {
 420   enum GNUNET_SCHEDULER_Reason reason;
 421
 422   reason = task->reason;
 423   if (now.abs_value >= task->timeout.abs_value)
 424     reason |= GNUNET_SCHEDULER_REASON_TIMEOUT;
 425   if ( (0 == (reason & GNUNET_SCHEDULER_REASON_READ_READY)) &&
 426        ( ( (task->read_fd != -1) &&
 427            (GNUNET_YES == GNUNET_NETWORK_fdset_test_native (rs, task->read_fd)) ) ||
 428          (set_overlaps (rs, task->read_set) ) ) )
 429     reason |= GNUNET_SCHEDULER_REASON_READ_READY;
 430   if ((0 == (reason & GNUNET_SCHEDULER_REASON_WRITE_READY)) &&
 431       ( ( (task->write_fd != -1) &&
 432           (GNUNET_YES == GNUNET_NETWORK_fdset_test_native (ws, task->write_fd)) ) ||
 433         (set_overlaps (ws, task->write_set) ) ) )
 434     reason |= GNUNET_SCHEDULER_REASON_WRITE_READY;
 435   if (reason == 0)
 436     return GNUNET_NO;           /* not ready */
 437   if (task->prereq_id != GNUNET_SCHEDULER_NO_TASK)
 438     {
 439       if (GNUNET_YES == is_pending (sched, task->prereq_id))
 440         {
 441           task->reason = reason;
 442           return GNUNET_NO;       /* prereq waiting */
 443         }
 444       reason |= GNUNET_SCHEDULER_REASON_PREREQ_DONE;
 445     }
 446   task->reason = reason;
 447   return GNUNET_YES;
 448 }
 449
 450
 451 /**
 452  * Put a task that is ready for execution into the ready queue.
 453  *
 454  * @param handle the scheduler
 455  * @param task task ready for execution
 456  */
 457 static void
 458 queue_ready_task (struct GNUNET_SCHEDULER_Handle *handle,
 459                   struct Task *task)
 460 {
 461   enum GNUNET_SCHEDULER_Priority p = task->priority;
 462   if (0 != (task->reason & GNUNET_SCHEDULER_REASON_SHUTDOWN))
 463     p = GNUNET_SCHEDULER_PRIORITY_SHUTDOWN;
 464   task->next = handle->ready[check_priority (p)];
 465   handle->ready[check_priority (p)] = task;
 466   handle->ready_count++;
 467 }
 468
 469
 470 /**
 471  * Check which tasks are ready and move them
 472  * to the respective ready queue.
 473  *
 474  * @param handle the scheduler
 475  * @param rs FDs ready for reading
 476  * @param ws FDs ready for writing
 477  */
 478 static void
 479 check_ready (struct GNUNET_SCHEDULER_Handle *handle,
 480              const struct GNUNET_NETWORK_FDSet *rs,
 481              const struct GNUNET_NETWORK_FDSet *ws)
 482 {
 483   struct Task *pos;
 484   struct Task *prev;
 485   struct Task *next;
 486   struct GNUNET_TIME_Absolute now;
 487
 488   now = GNUNET_TIME_absolute_get ();
 489   prev = NULL;
 490   pos = handle->pending_timeout;
 491   while (pos != NULL)
 492     {
 493       next = pos->next;
 494       if (now.abs_value >= pos->timeout.abs_value)
 495         pos->reason |= GNUNET_SCHEDULER_REASON_TIMEOUT;
 496       if (0 == pos->reason)
 497         break;
 498       handle->pending_timeout = next;
 499       if (handle->pending_timeout_last == pos)
 500         handle->pending_timeout_last = NULL;
 501       queue_ready_task (handle, pos);
 502       pos = next;
 503     }
 504   pos = handle->pending;
 505   while (pos != NULL)
 506     {
 507 #if DEBUG_TASKS
 508       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 509                   "Checking readiness of task: %llu / %p\n",
 510                   pos->id, pos->callback_cls);
 511 #endif
 512       next = pos->next;
 513       if (GNUNET_YES == is_ready (handle, pos, now, rs, ws))
 514         {
 515           if (prev == NULL)
 516             handle->pending = next;
 517           else
 518             prev->next = next;
 519           queue_ready_task (handle, pos);
 520           pos = next;
 521           continue;
 522         }
 523       prev = pos;
 524       pos = next;
 525     }
 526 }
 527
 528
 529 /**
 530  * Request the shutdown of a scheduler.  Marks all currently
 531  * pending tasks as ready because of shutdown.  This will
 532  * cause all tasks to run (as soon as possible, respecting
 533  * priorities and prerequisite tasks).  Note that tasks
 534  * scheduled AFTER this call may still be delayed arbitrarily.
 535  *
 536  * @param sched the scheduler
 537  */
 538 void
 539 GNUNET_SCHEDULER_shutdown (struct GNUNET_SCHEDULER_Handle *sched)
 540 {
 541   struct Task *pos;
 542   int i;
 543
 544   pos = sched->pending_timeout;
 545   while (pos != NULL)
 546     {
 547       pos->reason |= GNUNET_SCHEDULER_REASON_SHUTDOWN;
 548       /* we don't move the task into the ready queue yet; check_ready
 549          will do that later, possibly adding additional
 550          readiness-factors */
 551       pos = pos->next;
 552     }
 553   pos = sched->pending;
 554   while (pos != NULL)
 555     {
 556       pos->reason |= GNUNET_SCHEDULER_REASON_SHUTDOWN;
 557       /* we don't move the task into the ready queue yet; check_ready
 558          will do that later, possibly adding additional
 559          readiness-factors */
 560       pos = pos->next;
 561     }
 562   for (i=0;i<GNUNET_SCHEDULER_PRIORITY_COUNT;i++)
 563     {
 564       pos = sched->ready[i];
 565       while (pos != NULL)
 566         {
 567           pos->reason |= GNUNET_SCHEDULER_REASON_SHUTDOWN;
 568           /* we don't move the task into the ready queue yet; check_ready
 569              will do that later, possibly adding additional
 570              readiness-factors */
 571           pos = pos->next;
 572         }
 573     }
 574 }
 575
 576
 577 /**
 578  * Destroy a task (release associated resources)
 579  *
 580  * @param t task to destroy
 581  */
 582 static void
 583 destroy_task (struct Task *t)
 584 {
 585   if (NULL != t->read_set)
 586     GNUNET_NETWORK_fdset_destroy (t->read_set);
 587   if (NULL != t->write_set)
 588     GNUNET_NETWORK_fdset_destroy (t->write_set);
 589 #if EXECINFO
 590   GNUNET_free (t->backtrace_strings);
 591 #endif
 592   GNUNET_free (t);
 593 }
 594
 595
 596 /**
 597  * Run at least one task in the highest-priority queue that is not
 598  * empty.  Keep running tasks until we are either no longer running
 599  * "URGENT" tasks or until we have at least one "pending" task (which
 600  * may become ready, hence we should select on it).  Naturally, if
 601  * there are no more ready tasks, we also return.
 602  *
 603  * @param sched the scheduler
 604  * @param rs FDs ready for reading
 605  * @param ws FDs ready for writing
 606  */
 607 static void
 608 run_ready (struct GNUNET_SCHEDULER_Handle *sched,
 609            struct GNUNET_NETWORK_FDSet *rs,
 610            struct GNUNET_NETWORK_FDSet *ws)
 611 {
 612   enum GNUNET_SCHEDULER_Priority p;
 613   struct Task *pos;
 614   struct GNUNET_SCHEDULER_TaskContext tc;
 615
 616   sched->max_priority_added = GNUNET_SCHEDULER_PRIORITY_KEEP;
 617   do
 618     {
 619       if (sched->ready_count == 0)
 620         return;
 621       GNUNET_assert (sched->ready[GNUNET_SCHEDULER_PRIORITY_KEEP] == NULL);
 622       /* yes, p>0 is correct, 0 is "KEEP" which should
 623          always be an empty queue (see assertion)! */
 624       for (p = GNUNET_SCHEDULER_PRIORITY_COUNT - 1; p > 0; p--)
 625         {
 626           pos = sched->ready[p];
 627           if (pos != NULL)
 628             break;
 629         }
 630       GNUNET_assert (pos != NULL);      /* ready_count wrong? */
 631       sched->ready[p] = pos->next;
 632       sched->ready_count--;
 633       if (sched->current_priority != pos->priority)
 634         {
 635           sched->current_priority = pos->priority;
 636           (void) GNUNET_OS_set_process_priority (GNUNET_OS_process_current (), pos->priority);
 637         }
 638       sched->active_task = pos;
 639 #if PROFILE_DELAYS
 640       if (GNUNET_TIME_absolute_get_duration (pos->start_time).rel_value >
 641           DELAY_THRESHOLD.rel_value)
 642         {
 643           GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
 644                       "Task %llu took %llums to be scheduled\n",
 645                       pos->id,
 646                       (unsigned long long) GNUNET_TIME_absolute_get_duration (pos->start_time).rel_value);
 647         }
 648 #endif
 649       tc.sched = sched;
 650       tc.reason = pos->reason;
 651       tc.read_ready = (pos->read_set == NULL) ? rs : pos->read_set;
 652       if ( (pos->read_fd != -1) &&
 653            (0 != (pos->reason & GNUNET_SCHEDULER_REASON_READ_READY)) )
 654         GNUNET_NETWORK_fdset_set_native (rs,
 655                                          pos->read_fd);
 656       tc.write_ready = (pos->write_set == NULL) ? ws : pos->write_set;
 657       if ( (pos->write_fd != -1) &&
 658            (0 != (pos->reason & GNUNET_SCHEDULER_REASON_WRITE_READY)) )
 659         GNUNET_NETWORK_fdset_set_native (ws,
 660                                          pos->write_fd);
 661       if ( ( (tc.reason & GNUNET_SCHEDULER_REASON_WRITE_READY) != 0) &&
 662            (pos->write_fd != -1) &&
 663            (! GNUNET_NETWORK_fdset_test_native (ws,
 664                                                 pos->write_fd)))
 665         abort (); // added to ready in previous select loop!
 666 #if DEBUG_TASKS
 667       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 668                   "Running task: %llu / %p\n", pos->id, pos->callback_cls);
 669 #endif
 670       pos->callback (pos->callback_cls, &tc);
 671 #if EXECINFO
 672       int i;
 673       for (i=0;i<pos->num_backtrace_strings;i++)
 674         GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
 675                     "Task %llu trace %d: %s\n",
 676                     pos->id,
 677                     i,
 678                     pos->backtrace_strings[i]);
 679 #endif
 680       sched->active_task = NULL;
 681       destroy_task (pos);
 682       sched->tasks_run++;
 683     }
 684   while ( (sched->pending == NULL) || (p >= sched->max_priority_added) );
 685 }
 686
 687 /**
 688  * Pipe used to communicate shutdown via signal.
 689  */
 690 static struct GNUNET_DISK_PipeHandle *shutdown_pipe_handle;
 691
 692 /**
 693  * Signal handler called for SIGPIPE.
 694  */
 695 #ifndef MINGW
 696 static void
 697 sighandler_pipe ()
 698 {
 699   return;
 700 }
 701 #endif
 702 /**
 703  * Signal handler called for signals that should cause us to shutdown.
 704  */
 705 static void
 706 sighandler_shutdown ()
 707 {
 708   static char c;
 709   int old_errno = errno; /* backup errno */
 710
 711   GNUNET_DISK_file_write (GNUNET_DISK_pipe_handle
 712                           (shutdown_pipe_handle, GNUNET_DISK_PIPE_END_WRITE), &c,
 713                           sizeof (c));
 714   errno = old_errno;
 715 }
 716
 717
 718 /**
 719  * Initialize and run scheduler.  This function will return when all
 720  * tasks have completed.  On systems with signals, receiving a SIGTERM
 721  * (and other similar signals) will cause "GNUNET_SCHEDULER_shutdown"
 722  * to be run after the active task is complete.  As a result, SIGTERM
 723  * causes all active tasks to be scheduled with reason
 724  * "GNUNET_SCHEDULER_REASON_SHUTDOWN".  (However, tasks added
 725  * afterwards will execute normally!). Note that any particular signal
 726  * will only shut down one scheduler; applications should always only
 727  * create a single scheduler.
 728  *
 729  * @param task task to run immediately
 730  * @param task_cls closure of task
 731  */
 732 void
 733 GNUNET_SCHEDULER_run (GNUNET_SCHEDULER_Task task, void *task_cls)
 734 {
 735   struct GNUNET_SCHEDULER_Handle sched;
 736   struct GNUNET_NETWORK_FDSet *rs;
 737   struct GNUNET_NETWORK_FDSet *ws;
 738   struct GNUNET_TIME_Relative timeout;
 739   int ret;
 740   struct GNUNET_SIGNAL_Context *shc_int;
 741   struct GNUNET_SIGNAL_Context *shc_term;
 742 #ifndef MINGW
 743   struct GNUNET_SIGNAL_Context *shc_quit;
 744   struct GNUNET_SIGNAL_Context *shc_hup;
 745   struct GNUNET_SIGNAL_Context *shc_pipe;
 746 #endif
 747   unsigned long long last_tr;
 748   unsigned int busy_wait_warning;
 749   const struct GNUNET_DISK_FileHandle *pr;
 750   char c;
 751
 752   rs = GNUNET_NETWORK_fdset_create ();
 753   ws = GNUNET_NETWORK_fdset_create ();
 754   GNUNET_assert (shutdown_pipe_handle == NULL);
 755   shutdown_pipe_handle =  GNUNET_DISK_pipe (GNUNET_NO, GNUNET_NO, GNUNET_NO);
 756   GNUNET_assert (shutdown_pipe_handle != NULL);
 757   pr = GNUNET_DISK_pipe_handle (shutdown_pipe_handle, GNUNET_DISK_PIPE_END_READ);
 758   GNUNET_assert (pr != NULL);
 759   shc_int = GNUNET_SIGNAL_handler_install (SIGINT, &sighandler_shutdown);
 760   shc_term = GNUNET_SIGNAL_handler_install (SIGTERM, &sighandler_shutdown);
 761 #ifndef MINGW
 762   shc_pipe = GNUNET_SIGNAL_handler_install (SIGPIPE, &sighandler_pipe);
 763   shc_quit = GNUNET_SIGNAL_handler_install (SIGQUIT, &sighandler_shutdown);
 764   shc_hup = GNUNET_SIGNAL_handler_install (SIGHUP, &sighandler_shutdown);
 765 #endif
 766   memset (&sched, 0, sizeof (sched));
 767   sched.current_priority = GNUNET_SCHEDULER_PRIORITY_DEFAULT;
 768   GNUNET_SCHEDULER_add_continuation (&sched,
 769                                      task,
 770                                      task_cls,
 771                                      GNUNET_SCHEDULER_REASON_STARTUP);
 772   last_tr = 0;
 773   busy_wait_warning = 0;
 774   while ((sched.pending != NULL) ||
 775          (sched.pending_timeout != NULL) ||
 776          (sched.ready_count > 0))
 777     {
 778       GNUNET_NETWORK_fdset_zero (rs);
 779       GNUNET_NETWORK_fdset_zero (ws);
 780       timeout = GNUNET_TIME_UNIT_FOREVER_REL;
 781       update_sets (&sched, rs, ws, &timeout);
 782       GNUNET_NETWORK_fdset_handle_set (rs, pr);
 783       if (sched.ready_count > 0)
 784         {
 785           /* no blocking, more work already ready! */
 786           timeout = GNUNET_TIME_UNIT_ZERO;
 787         }
 788       ret = GNUNET_NETWORK_socket_select (rs, ws, NULL, timeout);
 789       if (ret == GNUNET_SYSERR)
 790         {
 791           if (errno == EINTR)
 792             continue;
 793
 794           GNUNET_log_strerror (GNUNET_ERROR_TYPE_ERROR, "select");
 795 #ifndef MINGW
 796 #if USE_LSOF
 797           char lsof[512];
 798           snprintf (lsof, sizeof (lsof), "lsof -p %d", getpid());
 799           close (1);
 800           dup2 (2, 1);
 801           system (lsof);
 802 #endif
 803 #endif
 804           abort ();
 805           break;
 806         }
 807       if ((ret == 0) && (timeout.rel_value == 0) && (busy_wait_warning > 16))
 808         {
 809           GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
 810                       _("Looks like we're busy waiting...\n"));
 811           sleep (1);            /* mitigate */
 812         }
 813       check_ready (&sched, rs, ws);
 814       run_ready (&sched, rs, ws);
 815       if (GNUNET_NETWORK_fdset_handle_isset (rs, pr))
 816         {
 817           /* consume the signal */
 818           GNUNET_DISK_file_read (pr, &c, sizeof (c));
 819           /* mark all active tasks as ready due to shutdown */
 820           GNUNET_SCHEDULER_shutdown (&sched);
 821         }
 822       if (last_tr == sched.tasks_run)
 823         {
 824           busy_wait_warning++;
 825         }
 826       else
 827         {
 828           last_tr = sched.tasks_run;
 829           busy_wait_warning = 0;
 830         }
 831     }
 832   GNUNET_SIGNAL_handler_uninstall (shc_int);
 833   GNUNET_SIGNAL_handler_uninstall (shc_term);
 834 #ifndef MINGW
 835   GNUNET_SIGNAL_handler_uninstall (shc_pipe);
 836   GNUNET_SIGNAL_handler_uninstall (shc_quit);
 837   GNUNET_SIGNAL_handler_uninstall (shc_hup);
 838 #endif
 839   GNUNET_DISK_pipe_close (shutdown_pipe_handle);
 840   shutdown_pipe_handle = NULL;
 841   GNUNET_NETWORK_fdset_destroy (rs);
 842   GNUNET_NETWORK_fdset_destroy (ws);
 843 }
 844
 845
 846 /**
 847  * Obtain the reason code for why the current task was
 848  * started.  Will return the same value as
 849  * the GNUNET_SCHEDULER_TaskContext's reason field.
 850  *
 851  * @param sched scheduler to query
 852  * @return reason(s) why the current task is run
 853  */
 854 enum GNUNET_SCHEDULER_Reason
 855 GNUNET_SCHEDULER_get_reason (struct GNUNET_SCHEDULER_Handle *sched)
 856 {
 857   return sched->active_task->reason;
 858 }
 859
 860
 861 /**
 862  * Get information about the current load of this scheduler.  Use this
 863  * function to determine if an elective task should be added or simply
 864  * dropped (if the decision should be made based on the number of
 865  * tasks ready to run).
 866  *
 867  * @param sched scheduler to query
 868  * @param p priority level to look at
 869  * @return number of tasks pending right now
 870  */
 871 unsigned int
 872 GNUNET_SCHEDULER_get_load (struct GNUNET_SCHEDULER_Handle *sched,
 873                            enum GNUNET_SCHEDULER_Priority p)
 874 {
 875   struct Task *pos;
 876   unsigned int ret;
 877
 878   if (p == GNUNET_SCHEDULER_PRIORITY_COUNT)
 879     return sched->ready_count;
 880   if (p == GNUNET_SCHEDULER_PRIORITY_KEEP)
 881     p = sched->current_priority;
 882   ret = 0;
 883   pos = sched->ready[check_priority (p)];
 884   while (pos != NULL)
 885     {
 886       pos = pos->next;
 887       ret++;
 888     }
 889   return ret;
 890 }
 891
 892
 893 /**
 894  * Cancel the task with the specified identifier.
 895  * The task must not yet have run.
 896  *
 897  * @param sched scheduler to use
 898  * @param task id of the task to cancel
 899  * @return original closure of the task
 900  */
 901 void *
 902 GNUNET_SCHEDULER_cancel (struct GNUNET_SCHEDULER_Handle *sched,
 903                          GNUNET_SCHEDULER_TaskIdentifier task)
 904 {
 905   struct Task *t;
 906   struct Task *prev;
 907   enum GNUNET_SCHEDULER_Priority p;
 908   int to;
 909   void *ret;
 910
 911   to = 0;
 912   prev = NULL;
 913   t = sched->pending;
 914   while (t != NULL)
 915     {
 916       if (t->id == task)
 917         break;
 918       prev = t;
 919       t = t->next;
 920     }
 921   if (t == NULL)
 922     {
 923       prev = NULL;
 924       to = 1;
 925       t = sched->pending_timeout;
 926       while (t != NULL)
 927         {
 928           if (t->id == task)
 929             break;
 930           prev = t;
 931           t = t->next;
 932         }
 933       if (sched->pending_timeout_last == t)
 934         sched->pending_timeout_last = NULL;
 935     }
 936   p = 0;
 937   while (t == NULL)
 938     {
 939       p++;
 940       GNUNET_assert (p < GNUNET_SCHEDULER_PRIORITY_COUNT);
 941       prev = NULL;
 942       t = sched->ready[p];
 943       while (t != NULL)
 944         {
 945           if (t->id == task)
 946             {
 947               sched->ready_count--;
 948               break;
 949             }
 950           prev = t;
 951           t = t->next;
 952         }
 953     }
 954   if (prev == NULL)
 955     {
 956       if (p == 0)
 957         {
 958           if (to == 0)
 959             {
 960               sched->pending = t->next;
 961             }
 962           else
 963             {
 964               sched->pending_timeout = t->next;
 965             }
 966         }
 967       else
 968         {
 969           sched->ready[p] = t->next;
 970         }
 971     }
 972   else
 973     {
 974       prev->next = t->next;
 975     }
 976   ret = t->callback_cls;
 977 #if DEBUG_TASKS
 978   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 979               "Canceling task: %llu / %p\n", task, t->callback_cls);
 980 #endif
 981   destroy_task (t);
 982   return ret;
 983 }
 984
 985
 986 /**
 987  * Continue the current execution with the given function.  This is
 988  * similar to the other "add" functions except that there is no delay
 989  * and the reason code can be specified.
 990  *
 991  * @param sched scheduler to use
 992  * @param task main function of the task
 993  * @param task_cls closure for 'main'
 994  * @param reason reason for task invocation
 995  */
 996 void
 997 GNUNET_SCHEDULER_add_continuation (struct GNUNET_SCHEDULER_Handle *sched,
 998                                    GNUNET_SCHEDULER_Task task,
 999                                    void *task_cls,
1000                                    enum GNUNET_SCHEDULER_Reason reason)
1001 {
1002   struct Task *t;
1003 #if EXECINFO
1004   void *backtrace_array[50];
1005 #endif
1006   t = GNUNET_malloc (sizeof (struct Task));
1007 #if EXECINFO
1008   t->num_backtrace_strings = backtrace(backtrace_array, 50);
1009   t->backtrace_strings = backtrace_symbols(backtrace_array, t->num_backtrace_strings);
1010 #endif
1011   t->read_fd = -1;
1012   t->write_fd = -1;
1013   t->callback = task;
1014   t->callback_cls = task_cls;
1015   t->id = ++sched->last_id;
1016 #if PROFILE_DELAYS
1017   t->start_time = GNUNET_TIME_absolute_get ();
1018 #endif
1019   t->reason = reason;
1020   t->priority = sched->current_priority;
1021 #if DEBUG_TASKS
1022   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1023               "Adding continuation task: %llu / %p\n",
1024               t->id, t->callback_cls);
1025 #endif
1026   queue_ready_task (sched, t);
1027 }
1028
1029
1030
1031 /**
1032  * Schedule a new task to be run after the specified prerequisite task
1033  * has completed. It will be run with the priority of the calling
1034  * task.
1035  *
1036  * @param sched scheduler to use
1037  * @param prerequisite_task run this task after the task with the given
1038  *        task identifier completes (and any of our other
1039  *        conditions, such as delay, read or write-readiness
1040  *        are satisfied).  Use  GNUNET_SCHEDULER_NO_TASK to not have any dependency
1041  *        on completion of other tasks (this will cause the task to run as
1042  *        soon as possible).
1043  * @param task main function of the task
1044  * @param task_cls closure of task
1045  * @return unique task identifier for the job
1046  *         only valid until "task" is started!
1047  */
1048 GNUNET_SCHEDULER_TaskIdentifier
1049 GNUNET_SCHEDULER_add_after (struct GNUNET_SCHEDULER_Handle *sched,
1050                             GNUNET_SCHEDULER_TaskIdentifier prerequisite_task,
1051                             GNUNET_SCHEDULER_Task task, void *task_cls)
1052 {
1053   return GNUNET_SCHEDULER_add_select (sched,
1054                                       GNUNET_SCHEDULER_PRIORITY_KEEP,
1055                                       prerequisite_task,
1056                                       GNUNET_TIME_UNIT_ZERO,
1057                                       NULL, NULL, task, task_cls);
1058 }
1059
1060
1061 /**
1062  * Schedule a new task to be run with a specified priority.
1063  *
1064  * @param sched scheduler to use
1065  * @param prio how important is the new task?
1066  * @param task main function of the task
1067  * @param task_cls closure of task
1068  * @return unique task identifier for the job
1069  *         only valid until "task" is started!
1070  */
1071 GNUNET_SCHEDULER_TaskIdentifier
1072 GNUNET_SCHEDULER_add_with_priority (struct GNUNET_SCHEDULER_Handle * sched,
1073                                     enum GNUNET_SCHEDULER_Priority prio,
1074                                     GNUNET_SCHEDULER_Task task,
1075                                     void *task_cls)
1076 {
1077   return GNUNET_SCHEDULER_add_select (sched,
1078                                       prio,
1079                                       GNUNET_SCHEDULER_NO_TASK,
1080                                       GNUNET_TIME_UNIT_ZERO,
1081                                       NULL, NULL, task, task_cls);
1082 }
1083
1084
1085
1086 /**
1087  * Schedule a new task to be run with a specified delay.  The task
1088  * will be scheduled for execution once the delay has expired. It
1089  * will be run with the priority of the calling task.
1090  *
1091  * @param sched scheduler to use
1092  * @param delay when should this operation time out? Use
1093  *        GNUNET_TIME_UNIT_FOREVER_REL for "on shutdown"
1094  * @param task main function of the task
1095  * @param task_cls closure of task
1096  * @return unique task identifier for the job
1097  *         only valid until "task" is started!
1098  */
1099 GNUNET_SCHEDULER_TaskIdentifier
1100 GNUNET_SCHEDULER_add_delayed (struct GNUNET_SCHEDULER_Handle * sched,
1101                               struct GNUNET_TIME_Relative delay,
1102                               GNUNET_SCHEDULER_Task task, void *task_cls)
1103 {
1104 #if 1
1105   /* new, optimized version */
1106   struct Task *t;
1107   struct Task *pos;
1108   struct Task *prev;
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   t->read_fd = -1;
1122   t->write_fd = -1;
1123   t->id = ++sched->last_id;
1124 #if PROFILE_DELAYS
1125   t->start_time = GNUNET_TIME_absolute_get ();
1126 #endif
1127   t->timeout = GNUNET_TIME_relative_to_absolute (delay);
1128   t->priority = sched->current_priority;
1129   /* try tail first (optimization in case we are
1130      appending to a long list of tasks with timeouts) */
1131   prev = sched->pending_timeout_last;
1132   if (prev != NULL)
1133     {
1134       if (prev->timeout.abs_value > t->timeout.abs_value)
1135         prev = NULL;
1136       else
1137         pos = prev->next; /* heuristic success! */
1138     }
1139   if (prev == NULL)
1140     {
1141       /* heuristic failed, do traversal of timeout list */
1142       pos = sched->pending_timeout;
1143     }
1144   while ( (pos != NULL) &&
1145           ( (pos->timeout.abs_value <= t->timeout.abs_value) ||
1146             (pos->reason != 0) ) )
1147     {
1148       prev = pos;
1149       pos = pos->next;
1150     }
1151   if (prev == NULL)
1152     sched->pending_timeout = t;
1153   else
1154     prev->next = t;
1155   t->next = pos;
1156   /* hyper-optimization... */
1157   sched->pending_timeout_last = t;
1158
1159 #if DEBUG_TASKS
1160   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1161               "Adding task: %llu / %p\n", t->id, t->callback_cls);
1162 #endif
1163 #if EXECINFO
1164   int i;
1165
1166   for (i=0;i<t->num_backtrace_strings;i++)
1167       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1168                   "Task %llu trace %d: %s\n",
1169                   t->id,
1170                   i,
1171                   t->backtrace_strings[i]);
1172 #endif
1173   return t->id;
1174
1175 #else
1176   /* unoptimized version */
1177   return GNUNET_SCHEDULER_add_select (sched,
1178                                       GNUNET_SCHEDULER_PRIORITY_KEEP,
1179                                       GNUNET_SCHEDULER_NO_TASK, delay,
1180                                       NULL, NULL, task, task_cls);
1181 #endif
1182 }
1183
1184
1185
1186 /**
1187  * Schedule a new task to be run as soon as possible. The task
1188  * will be run with the priority of the calling task.
1189  *
1190  * @param sched scheduler to use
1191  * @param task main function of the task
1192  * @param task_cls closure of task
1193  * @return unique task identifier for the job
1194  *         only valid until "task" is started!
1195  */
1196 GNUNET_SCHEDULER_TaskIdentifier
1197 GNUNET_SCHEDULER_add_now (struct GNUNET_SCHEDULER_Handle *sched,
1198                           GNUNET_SCHEDULER_Task task,
1199                           void *task_cls)
1200 {
1201   return GNUNET_SCHEDULER_add_select (sched,
1202                                       GNUNET_SCHEDULER_PRIORITY_KEEP,
1203                                       GNUNET_SCHEDULER_NO_TASK,
1204                                       GNUNET_TIME_UNIT_ZERO,
1205                                       NULL, NULL, task, task_cls);
1206 }
1207
1208
1209
1210
1211 /**
1212  * Schedule a new task to be run with a specified delay or when any of
1213  * the specified file descriptor sets is ready.  The delay can be used
1214  * as a timeout on the socket(s) being ready.  The task will be
1215  * scheduled for execution once either the delay has expired or any of
1216  * the socket operations is ready.  This is the most general
1217  * function of the "add" family.  Note that the "prerequisite_task"
1218  * must be satisfied in addition to any of the other conditions.  In
1219  * other words, the task will be started when
1220  * <code>
1221  * (prerequisite-run)
1222  * && (delay-ready
1223  *     || any-rs-ready
1224  *     || any-ws-ready
1225  *     || shutdown-active )
1226  * </code>
1227  *
1228  * @param sched scheduler to use
1229  * @param delay how long should we wait? Use GNUNET_TIME_UNIT_FOREVER_REL for "forever",
1230  *        which means that the task will only be run after we receive SIGTERM
1231  * @param rfd file descriptor we want to read (can be -1)
1232  * @param wfd file descriptors we want to write (can be -1)
1233  * @param task main function of the task
1234  * @param task_cls closure of task
1235  * @return unique task identifier for the job
1236  *         only valid until "task" is started!
1237  */
1238 GNUNET_SCHEDULER_TaskIdentifier
1239 add_without_sets (struct GNUNET_SCHEDULER_Handle * sched,
1240                   struct GNUNET_TIME_Relative delay,
1241                   int rfd,
1242                   int wfd,
1243                   GNUNET_SCHEDULER_Task task, void *task_cls)
1244 {
1245   struct Task *t;
1246 #if EXECINFO
1247   void *backtrace_array[MAX_TRACE_DEPTH];
1248 #endif
1249
1250   GNUNET_assert (NULL != task);
1251   t = GNUNET_malloc (sizeof (struct Task));
1252   t->callback = task;
1253   t->callback_cls = task_cls;
1254 #if EXECINFO
1255   t->num_backtrace_strings = backtrace(backtrace_array, MAX_TRACE_DEPTH);
1256   t->backtrace_strings = backtrace_symbols(backtrace_array, t->num_backtrace_strings);
1257 #endif
1258   t->read_fd = rfd;
1259   t->write_fd = wfd;
1260   t->id = ++sched->last_id;
1261 #if PROFILE_DELAYS
1262   t->start_time = GNUNET_TIME_absolute_get ();
1263 #endif
1264   t->prereq_id = GNUNET_SCHEDULER_NO_TASK;
1265   t->timeout = GNUNET_TIME_relative_to_absolute (delay);
1266   t->priority = check_priority (sched->current_priority);
1267   t->next = sched->pending;
1268   sched->pending = t;
1269   sched->max_priority_added = GNUNET_MAX (sched->max_priority_added,
1270                                           t->priority);
1271 #if DEBUG_TASKS
1272   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1273               "Adding task: %llu / %p\n", t->id, t->callback_cls);
1274 #endif
1275 #if EXECINFO
1276   int i;
1277
1278   for (i=0;i<t->num_backtrace_strings;i++)
1279       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1280                   "Task %llu trace %d: %s\n",
1281                   t->id,
1282                   i,
1283                   t->backtrace_strings[i]);
1284 #endif
1285   return t->id;
1286 }
1287
1288
1289
1290 /**
1291  * Schedule a new task to be run with a specified delay or when the
1292  * specified file descriptor is ready for reading.  The delay can be
1293  * used as a timeout on the socket being ready.  The task will be
1294  * scheduled for execution once either the delay has expired or the
1295  * socket operation is ready.  It will be run with the priority of
1296  * the calling task.
1297  *
1298  * @param sched scheduler to use
1299  * @param delay when should this operation time out? Use
1300  *        GNUNET_TIME_UNIT_FOREVER_REL for "on shutdown"
1301  * @param rfd read file-descriptor
1302  * @param task main function of the task
1303  * @param task_cls closure of task
1304  * @return unique task identifier for the job
1305  *         only valid until "task" is started!
1306  */
1307 GNUNET_SCHEDULER_TaskIdentifier
1308 GNUNET_SCHEDULER_add_read_net (struct GNUNET_SCHEDULER_Handle * sched,
1309                                struct GNUNET_TIME_Relative delay,
1310                                struct GNUNET_NETWORK_Handle * rfd,
1311                                GNUNET_SCHEDULER_Task task, void *task_cls)
1312 {
1313   return add_without_sets (sched,
1314                            delay,
1315                            GNUNET_NETWORK_get_fd (rfd),
1316                            -1,
1317                            task,
1318                            task_cls);
1319 }
1320
1321
1322 /**
1323  * Schedule a new task to be run with a specified delay or when the
1324  * specified file descriptor is ready for writing.  The delay can be
1325  * used as a timeout on the socket being ready.  The task will be
1326  * scheduled for execution once either the delay has expired or the
1327  * socket operation is ready.  It will be run with the priority of
1328  * the calling task.
1329  *
1330  * @param sched scheduler to use
1331  * @param delay when should this operation time out? Use
1332  *        GNUNET_TIME_UNIT_FOREVER_REL for "on shutdown"
1333  * @param wfd write file-descriptor
1334  * @param task main function of the task
1335  * @param task_cls closure of task
1336  * @return unique task identifier for the job
1337  *         only valid until "task" is started!
1338  */
1339 GNUNET_SCHEDULER_TaskIdentifier
1340 GNUNET_SCHEDULER_add_write_net (struct GNUNET_SCHEDULER_Handle * sched,
1341                                 struct GNUNET_TIME_Relative delay,
1342                                 struct GNUNET_NETWORK_Handle * wfd,
1343                                 GNUNET_SCHEDULER_Task task, void *task_cls)
1344 {
1345   return add_without_sets (sched,
1346                            delay,
1347                            -1,
1348                            GNUNET_NETWORK_get_fd (wfd),
1349                            task,
1350                            task_cls);
1351 }
1352
1353
1354 /**
1355  * Schedule a new task to be run with a specified delay or when the
1356  * specified file descriptor is ready for reading.  The delay can be
1357  * used as a timeout on the socket being ready.  The task will be
1358  * scheduled for execution once either the delay has expired or the
1359  * socket operation is ready. It will be run with the priority of
1360  * the calling task.
1361  *
1362  * @param sched scheduler to use
1363  * @param delay when should this operation time out? Use
1364  *        GNUNET_TIME_UNIT_FOREVER_REL for "on shutdown"
1365  * @param rfd read file-descriptor
1366  * @param task main function of the task
1367  * @param task_cls closure of task
1368  * @return unique task identifier for the job
1369  *         only valid until "task" is started!
1370  */
1371 GNUNET_SCHEDULER_TaskIdentifier
1372 GNUNET_SCHEDULER_add_read_file (struct GNUNET_SCHEDULER_Handle * sched,
1373                                 struct GNUNET_TIME_Relative delay,
1374                                 const struct GNUNET_DISK_FileHandle * rfd,
1375                                 GNUNET_SCHEDULER_Task task, void *task_cls)
1376 {
1377 #if MINGW
1378   struct GNUNET_NETWORK_FDSet *rs;
1379   GNUNET_SCHEDULER_TaskIdentifier ret;
1380
1381   GNUNET_assert (rfd != NULL);
1382   rs = GNUNET_NETWORK_fdset_create ();
1383   GNUNET_NETWORK_fdset_handle_set (rs, rfd);
1384   ret = GNUNET_SCHEDULER_add_select (sched,
1385                                      GNUNET_SCHEDULER_PRIORITY_KEEP,
1386                                      GNUNET_SCHEDULER_NO_TASK, delay,
1387                                      rs, NULL, task, task_cls);
1388   GNUNET_NETWORK_fdset_destroy (rs);
1389   return ret;
1390 #else
1391   int fd;
1392
1393   GNUNET_DISK_internal_file_handle_ (rfd, &fd, sizeof (int));
1394   return add_without_sets (sched,
1395                            delay,
1396                            fd,
1397                            -1,
1398                            task,
1399                            task_cls);
1400
1401 #endif
1402 }
1403
1404
1405 /**
1406  * Schedule a new task to be run with a specified delay or when the
1407  * specified file descriptor is ready for writing.  The delay can be
1408  * used as a timeout on the socket being ready.  The task will be
1409  * scheduled for execution once either the delay has expired or the
1410  * socket operation is ready. It will be run with the priority of
1411  * the calling task.
1412  *
1413  * @param sched scheduler to use
1414  * @param delay when should this operation time out? Use
1415  *        GNUNET_TIME_UNIT_FOREVER_REL for "on shutdown"
1416  * @param wfd write file-descriptor
1417  * @param task main function of the task
1418  * @param task_cls closure of task
1419  * @return unique task identifier for the job
1420  *         only valid until "task" is started!
1421  */
1422 GNUNET_SCHEDULER_TaskIdentifier
1423 GNUNET_SCHEDULER_add_write_file (struct GNUNET_SCHEDULER_Handle * sched,
1424                                  struct GNUNET_TIME_Relative delay,
1425                                  const struct GNUNET_DISK_FileHandle * wfd,
1426                                  GNUNET_SCHEDULER_Task task, void *task_cls)
1427 {
1428 #if MINGW
1429   struct GNUNET_NETWORK_FDSet *ws;
1430   GNUNET_SCHEDULER_TaskIdentifier ret;
1431
1432   GNUNET_assert (wfd != NULL);
1433   ws = GNUNET_NETWORK_fdset_create ();
1434   GNUNET_NETWORK_fdset_handle_set (ws, wfd);
1435   ret = GNUNET_SCHEDULER_add_select (sched,
1436                                      GNUNET_SCHEDULER_PRIORITY_KEEP,
1437                                      GNUNET_SCHEDULER_NO_TASK,
1438                                      delay, NULL, ws, task, task_cls);
1439   GNUNET_NETWORK_fdset_destroy (ws);
1440   return ret;
1441 #else
1442   int fd;
1443
1444   GNUNET_DISK_internal_file_handle_ (wfd, &fd, sizeof (int));
1445   return add_without_sets (sched,
1446                            delay,
1447                            -1,
1448                            fd,
1449                            task,
1450                            task_cls);
1451
1452 #endif
1453 }
1454
1455
1456
1457 /**
1458  * Schedule a new task to be run with a specified delay or when any of
1459  * the specified file descriptor sets is ready.  The delay can be used
1460  * as a timeout on the socket(s) being ready.  The task will be
1461  * scheduled for execution once either the delay has expired or any of
1462  * the socket operations is ready.  This is the most general
1463  * function of the "add" family.  Note that the "prerequisite_task"
1464  * must be satisfied in addition to any of the other conditions.  In
1465  * other words, the task will be started when
1466  * <code>
1467  * (prerequisite-run)
1468  * && (delay-ready
1469  *     || any-rs-ready
1470  *     || any-ws-ready
1471  *     || (shutdown-active && run-on-shutdown) )
1472  * </code>
1473  *
1474  * @param sched scheduler to use
1475  * @param prio how important is this task?
1476  * @param prerequisite_task run this task after the task with the given
1477  *        task identifier completes (and any of our other
1478  *        conditions, such as delay, read or write-readiness
1479  *        are satisfied).  Use GNUNET_SCHEDULER_NO_TASK to not have any dependency
1480  *        on completion of other tasks.
1481  * @param delay how long should we wait? Use GNUNET_TIME_UNIT_FOREVER_REL for "forever",
1482  *        which means that the task will only be run after we receive SIGTERM
1483  * @param rs set of file descriptors we want to read (can be NULL)
1484  * @param ws set of file descriptors we want to write (can be NULL)
1485  * @param task main function of the task
1486  * @param task_cls closure of task
1487  * @return unique task identifier for the job
1488  *         only valid until "task" is started!
1489  */
1490 GNUNET_SCHEDULER_TaskIdentifier
1491 GNUNET_SCHEDULER_add_select (struct GNUNET_SCHEDULER_Handle * sched,
1492                              enum GNUNET_SCHEDULER_Priority prio,
1493                              GNUNET_SCHEDULER_TaskIdentifier
1494                              prerequisite_task,
1495                              struct GNUNET_TIME_Relative delay,
1496                              const struct GNUNET_NETWORK_FDSet * rs,
1497                              const struct GNUNET_NETWORK_FDSet * ws,
1498                              GNUNET_SCHEDULER_Task task, void *task_cls)
1499 {
1500   struct Task *t;
1501 #if EXECINFO
1502   void *backtrace_array[MAX_TRACE_DEPTH];
1503 #endif
1504
1505   GNUNET_assert (NULL != task);
1506   t = GNUNET_malloc (sizeof (struct Task));
1507   t->callback = task;
1508   t->callback_cls = task_cls;
1509 #if EXECINFO
1510   t->num_backtrace_strings = backtrace(backtrace_array, MAX_TRACE_DEPTH);
1511   t->backtrace_strings = backtrace_symbols(backtrace_array, t->num_backtrace_strings);
1512 #endif
1513   t->read_fd = -1;
1514   t->write_fd = -1;
1515   if (rs != NULL)
1516     {
1517       t->read_set = GNUNET_NETWORK_fdset_create ();
1518       GNUNET_NETWORK_fdset_copy (t->read_set, rs);
1519     }
1520   if (ws != NULL)
1521     {
1522       t->write_set = GNUNET_NETWORK_fdset_create ();
1523       GNUNET_NETWORK_fdset_copy (t->write_set, ws);
1524     }
1525   t->id = ++sched->last_id;
1526 #if PROFILE_DELAYS
1527   t->start_time = GNUNET_TIME_absolute_get ();
1528 #endif
1529   t->prereq_id = prerequisite_task;
1530   t->timeout = GNUNET_TIME_relative_to_absolute (delay);
1531   t->priority =
1532     check_priority ((prio ==
1533                      GNUNET_SCHEDULER_PRIORITY_KEEP) ? sched->current_priority
1534                     : prio);
1535   t->next = sched->pending;
1536   sched->pending = t;
1537   sched->max_priority_added = GNUNET_MAX (sched->max_priority_added,
1538                                           t->priority);
1539 #if DEBUG_TASKS
1540   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1541               "Adding task: %llu / %p\n", t->id, t->callback_cls);
1542 #endif
1543 #if EXECINFO
1544   int i;
1545
1546   for (i=0;i<t->num_backtrace_strings;i++)
1547       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1548                   "Task %llu trace %d: %s\n",
1549                   t->id,
1550                   i,
1551                   t->backtrace_strings[i]);
1552 #endif
1553   return t->id;
1554 }
1555
1556 /* end of scheduler.c */