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