src/util/scheduler.c

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