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