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