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