src/testbed/testbed_api_operations.c

   1 /*
   2       This file is part of GNUnet
   3       Copyright (C) 2008--2013 GNUnet e.V.
   4
   5       GNUnet is free software: you can redistribute it and/or modify it
   6       under the terms of the GNU Affero General Public License as published
   7       by the Free Software Foundation, either version 3 of the License,
   8       or (at your 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       Affero General Public License for more details.
  14
  15       You should have received a copy of the GNU Affero General Public License
  16       along with this program.  If not, see <http://www.gnu.org/licenses/>.
  17  */
  18
  19 /**
  20  * @file testbed/testbed_api_operations.c
  21  * @brief functions to manage operation queues
  22  * @author Christian Grothoff
  23  * @author Sree Harsha Totakura
  24  */
  25
  26 #include "platform.h"
  27 #include "testbed_api_operations.h"
  28 #include "testbed_api_sd.h"
  29
  30 /**
  31  * The number of readings containing past operation's timing information that we
  32  * keep track of for adaptive queues
  33  */
  34 #define ADAPTIVE_QUEUE_DEFAULT_HISTORY 40
  35
  36 /**
  37  * The number of parallel opeartions we start with by default for adaptive
  38  * queues
  39  */
  40 #define ADAPTIVE_QUEUE_DEFAULT_MAX_ACTIVE 4
  41
  42 /**
  43  * An entry in the operation queue
  44  */
  45 struct QueueEntry
  46 {
  47   /**
  48    * The next DLL pointer
  49    */
  50   struct QueueEntry *next;
  51
  52   /**
  53    * The prev DLL pointer
  54    */
  55   struct QueueEntry *prev;
  56
  57   /**
  58    * The operation this entry holds
  59    */
  60   struct GNUNET_TESTBED_Operation *op;
  61
  62   /**
  63    * How many units of resources does the operation need
  64    */
  65   unsigned int nres;
  66 };
  67
  68
  69 /**
  70  * Queue of operations where we can only support a certain
  71  * number of concurrent operations of a particular type.
  72  */
  73 struct OperationQueue;
  74
  75
  76 /**
  77  * A slot to record time taken by an operation
  78  */
  79 struct TimeSlot
  80 {
  81   /**
  82    * DLL next pointer
  83    */
  84   struct TimeSlot *next;
  85
  86   /**
  87    * DLL prev pointer
  88    */
  89   struct TimeSlot *prev;
  90
  91   /**
  92    * This operation queue to which this time slot belongs to
  93    */
  94   struct OperationQueue *queue;
  95
  96   /**
  97    * The operation to which this timeslot is currently allocated to
  98    */
  99   struct GNUNET_TESTBED_Operation *op;
 100
 101   /**
 102    * Accumulated time
 103    */
 104   struct GNUNET_TIME_Relative tsum;
 105
 106   /**
 107    * Number of timing values accumulated
 108    */
 109   unsigned int nvals;
 110 };
 111
 112
 113 /**
 114  * Context for operation queues of type OPERATION_QUEUE_TYPE_ADAPTIVE
 115  */
 116 struct FeedbackCtx
 117 {
 118   /**
 119    * Handle for calculating standard deviation
 120    */
 121   struct SDHandle *sd;
 122
 123   /**
 124    * Head for DLL of time slots which are free to be allocated to operations
 125    */
 126   struct TimeSlot *alloc_head;
 127
 128   /**
 129    * Tail for DLL of time slots which are free to be allocated to operations
 130    */
 131   struct TimeSlot *alloc_tail;
 132
 133   /**
 134    * Pointer to the chunk of time slots.  Free all time slots at a time using
 135    * this pointer.
 136    */
 137   struct TimeSlot *tslots_freeptr;
 138
 139   /**
 140    * Number of time slots filled so far
 141    */
 142   unsigned int tslots_filled;
 143
 144   /**
 145    * Bound on the maximum number of operations which can be active
 146    */
 147   unsigned int max_active_bound;
 148
 149   /**
 150    * Number of operations that have failed
 151    */
 152   unsigned int nfailed;
 153 };
 154
 155
 156 /**
 157  * Queue of operations where we can only support a certain
 158  * number of concurrent operations of a particular type.
 159  */
 160 struct OperationQueue
 161 {
 162   /**
 163    * DLL head for the wait queue.  Operations which are waiting for this
 164    * operation queue are put here
 165    */
 166   struct QueueEntry *wq_head;
 167
 168   /**
 169    * DLL tail for the wait queue.
 170    */
 171   struct QueueEntry *wq_tail;
 172
 173   /**
 174    * DLL head for the ready queue.  Operations which are in this operation queue
 175    * and are in ready state are put here
 176    */
 177   struct QueueEntry *rq_head;
 178
 179   /**
 180    * DLL tail for the ready queue
 181    */
 182   struct QueueEntry *rq_tail;
 183
 184   /**
 185    * DLL head for the active queue.  Operations which are in this operation
 186    * queue and are currently active are put here
 187    */
 188   struct QueueEntry *aq_head;
 189
 190   /**
 191    * DLL tail for the active queue.
 192    */
 193   struct QueueEntry *aq_tail;
 194
 195   /**
 196    * DLL head for the inactive queue.  Operations which are inactive and can be
 197    * evicted if the queues it holds are maxed out and another operation begins
 198    * to wait on them.
 199    */
 200   struct QueueEntry *nq_head;
 201
 202   /**
 203    * DLL tail for the inactive queue.
 204    */
 205   struct QueueEntry *nq_tail;
 206
 207   /**
 208    * Feedback context; only relevant for adaptive operation queues.  NULL for
 209    * fixed operation queues
 210    */
 211   struct FeedbackCtx *fctx;
 212
 213   /**
 214    * The type of this opeartion queue
 215    */
 216   enum OperationQueueType type;
 217
 218   /**
 219    * Number of operations that are currently active in this queue.
 220    */
 221   unsigned int active;
 222
 223   /**
 224    * Max number of operations which can be active at any time in this queue.
 225    * This value can be changed either by calling
 226    * GNUNET_TESTBED_operation_queue_reset_max_active_() or by the adaptive
 227    * algorithm if this operation queue is of type #OPERATION_QUEUE_TYPE_ADAPTIVE
 228    */
 229   unsigned int max_active;
 230
 231   /**
 232    * The number of resources occupied by failed operations in the current shot.
 233    * This is only relavant if the operation queue is of type
 234    * #OPERATION_QUEUE_TYPE_ADAPTIVE
 235    */
 236   unsigned int overload;
 237
 238   /**
 239    * Is this queue marked for expiry?
 240    */
 241   unsigned int expired;
 242 };
 243
 244
 245 /**
 246  * Operation state
 247  */
 248 enum OperationState
 249 {
 250   /**
 251    * The operation is just created and is in initial state
 252    */
 253   OP_STATE_INIT,
 254
 255   /**
 256    * The operation is currently waiting for resources
 257    */
 258   OP_STATE_WAITING,
 259
 260   /**
 261    * The operation is ready to be started
 262    */
 263   OP_STATE_READY,
 264
 265   /**
 266    * The operation has started and is active
 267    */
 268   OP_STATE_ACTIVE,
 269
 270   /**
 271    * The operation is inactive.  It still holds resources on the operation
 272    * queues.  However, this operation will be evicted when another operation
 273    * requires resources from the maxed out queues this operation is holding
 274    * resources from.
 275    */
 276   OP_STATE_INACTIVE
 277 };
 278
 279
 280 /**
 281  * An entry in the ready queue (implemented as DLL)
 282  */
 283 struct ReadyQueueEntry
 284 {
 285   /**
 286    * next ptr for DLL
 287    */
 288   struct ReadyQueueEntry *next;
 289
 290   /**
 291    * prev ptr for DLL
 292    */
 293   struct ReadyQueueEntry *prev;
 294
 295   /**
 296    * The operation associated with this entry
 297    */
 298   struct GNUNET_TESTBED_Operation *op;
 299 };
 300
 301
 302 /**
 303  * Opaque handle to an abstract operation to be executed by the testing framework.
 304  */
 305 struct GNUNET_TESTBED_Operation
 306 {
 307   /**
 308    * Function to call when we have the resources to begin the operation.
 309    */
 310   OperationStart start;
 311
 312   /**
 313    * Function to call to clean up after the operation (which may or may
 314    * not have been started yet).
 315    */
 316   OperationRelease release;
 317
 318   /**
 319    * Closure for callbacks.
 320    */
 321   void *cb_cls;
 322
 323   /**
 324    * Array of operation queues this Operation belongs to.
 325    */
 326   struct OperationQueue **queues;
 327
 328   /**
 329    * Array of operation queue entries corresponding to this operation in
 330    * operation queues for this operation
 331    */
 332   struct QueueEntry **qentries;
 333
 334   /**
 335    * Array of number of resources an operation need from each queue. The numbers
 336    * in this array should correspond to the queues array
 337    */
 338   unsigned int *nres;
 339
 340   /**
 341    * Entry corresponding to this operation in ready queue.  Will be NULL if the
 342    * operation is not marked as READY
 343    */
 344   struct ReadyQueueEntry *rq_entry;
 345
 346   /**
 347    * Head pointer for DLL of tslots allocated to this operation
 348    */
 349   struct TimeSlot *tslots_head;
 350
 351   /**
 352    * Tail pointer for DLL of tslots allocated to this operation
 353    */
 354   struct TimeSlot *tslots_tail;
 355
 356   /**
 357    * The time at which the operation is started
 358    */
 359   struct GNUNET_TIME_Absolute tstart;
 360
 361   /**
 362    * Number of queues in the operation queues array
 363    */
 364   unsigned int nqueues;
 365
 366   /**
 367    * The state of the operation
 368    */
 369   enum OperationState state;
 370
 371   /**
 372    * Is this a failed operation?
 373    */
 374   int failed;
 375
 376 };
 377
 378 /**
 379  * DLL head for the ready queue
 380  */
 381 static struct ReadyQueueEntry *rq_head;
 382
 383 /**
 384  * DLL tail for the ready queue
 385  */
 386 static struct ReadyQueueEntry *rq_tail;
 387
 388 /**
 389  * Array of operation queues which are to be destroyed
 390  */
 391 static struct OperationQueue **expired_opqs;
 392
 393 /**
 394  * Number of expired operation queues in the above array
 395  */
 396 static unsigned int n_expired_opqs;
 397
 398 /**
 399  * The id of the task to process the ready queue
 400  */
 401 struct GNUNET_SCHEDULER_Task *process_rq_task_id;
 402
 403
 404 /**
 405  * Assigns the given operation a time slot from the given operation queue
 406  *
 407  * @param op the operation
 408  * @param queue the operation queue
 409  * @return the timeslot
 410  */
 411 static void
 412 assign_timeslot (struct GNUNET_TESTBED_Operation *op,
 413                  struct OperationQueue *queue)
 414 {
 415   struct FeedbackCtx *fctx = queue->fctx;
 416   struct TimeSlot *tslot;
 417
 418   GNUNET_assert (OPERATION_QUEUE_TYPE_ADAPTIVE == queue->type);
 419   tslot = fctx->alloc_head;
 420   GNUNET_assert (NULL != tslot);
 421   GNUNET_CONTAINER_DLL_remove (fctx->alloc_head, fctx->alloc_tail, tslot);
 422   GNUNET_CONTAINER_DLL_insert_tail (op->tslots_head, op->tslots_tail, tslot);
 423   tslot->op = op;
 424 }
 425
 426
 427 /**
 428  * Removes a queue entry of an operation from one of the operation queues' lists
 429  * depending on the state of the operation
 430  *
 431  * @param op the operation whose entry has to be removed
 432  * @param index the index of the entry in the operation's array of queue entries
 433  */
 434 static void
 435 remove_queue_entry (struct GNUNET_TESTBED_Operation *op, unsigned int index)
 436 {
 437   struct OperationQueue *opq;
 438   struct QueueEntry *entry;
 439
 440   opq = op->queues[index];
 441   entry = op->qentries[index];
 442   switch (op->state)
 443   {
 444   case OP_STATE_INIT:
 445     GNUNET_assert (0);
 446     break;
 447   case OP_STATE_WAITING:
 448     GNUNET_CONTAINER_DLL_remove (opq->wq_head, opq->wq_tail, entry);
 449     break;
 450   case OP_STATE_READY:
 451     GNUNET_CONTAINER_DLL_remove (opq->rq_head, opq->rq_tail, entry);
 452     break;
 453   case OP_STATE_ACTIVE:
 454     GNUNET_CONTAINER_DLL_remove (opq->aq_head, opq->aq_tail, entry);
 455     break;
 456   case OP_STATE_INACTIVE:
 457     GNUNET_CONTAINER_DLL_remove (opq->nq_head, opq->nq_tail, entry);
 458     break;
 459   }
 460 }
 461
 462
 463 /**
 464  * Changes the state of the operation while moving its associated queue entries
 465  * in the operation's operation queues
 466  *
 467  * @param op the operation whose state has to be changed
 468  * @param state the state the operation should have.  It cannot be OP_STATE_INIT
 469  */
 470 static void
 471 change_state (struct GNUNET_TESTBED_Operation *op, enum OperationState state)
 472 {
 473   struct QueueEntry *entry;
 474   struct OperationQueue *opq;
 475   unsigned int cnt;
 476   unsigned int s;
 477
 478   GNUNET_assert (OP_STATE_INIT != state);
 479   GNUNET_assert (NULL != op->queues);
 480   GNUNET_assert (NULL != op->nres);
 481   GNUNET_assert ((OP_STATE_INIT == op->state) || (NULL != op->qentries));
 482   GNUNET_assert (op->state != state);
 483   for (cnt = 0; cnt < op->nqueues; cnt++)
 484   {
 485     if (OP_STATE_INIT == op->state)
 486     {
 487       entry = GNUNET_new (struct QueueEntry);
 488       entry->op = op;
 489       entry->nres = op->nres[cnt];
 490       s = cnt;
 491       GNUNET_array_append (op->qentries, s, entry);
 492     }
 493     else
 494     {
 495       entry = op->qentries[cnt];
 496       remove_queue_entry (op, cnt);
 497     }
 498     opq = op->queues[cnt];
 499     switch (state)
 500     {
 501     case OP_STATE_INIT:
 502       GNUNET_assert (0);
 503       break;
 504     case OP_STATE_WAITING:
 505       GNUNET_CONTAINER_DLL_insert_tail (opq->wq_head, opq->wq_tail, entry);
 506       break;
 507     case OP_STATE_READY:
 508       GNUNET_CONTAINER_DLL_insert_tail (opq->rq_head, opq->rq_tail, entry);
 509       break;
 510     case OP_STATE_ACTIVE:
 511       GNUNET_CONTAINER_DLL_insert_tail (opq->aq_head, opq->aq_tail, entry);
 512       break;
 513     case OP_STATE_INACTIVE:
 514       GNUNET_CONTAINER_DLL_insert_tail (opq->nq_head, opq->nq_tail, entry);
 515       break;
 516     }
 517   }
 518   op->state = state;
 519 }
 520
 521
 522 /**
 523  * Removes an operation from the ready queue.  Also stops the 'process_rq_task'
 524  * if the given operation is the last one in the queue.
 525  *
 526  * @param op the operation to be removed
 527  */
 528 static void
 529 rq_remove (struct GNUNET_TESTBED_Operation *op)
 530 {
 531   GNUNET_assert (NULL != op->rq_entry);
 532   GNUNET_CONTAINER_DLL_remove (rq_head, rq_tail, op->rq_entry);
 533   GNUNET_free (op->rq_entry);
 534   op->rq_entry = NULL;
 535   if ( (NULL == rq_head) && (NULL != process_rq_task_id) )
 536   {
 537     GNUNET_SCHEDULER_cancel (process_rq_task_id);
 538     process_rq_task_id = NULL;
 539   }
 540 }
 541
 542
 543 /**
 544  * Processes the ready queue by calling the operation start callback of the
 545  * operation at the head.  The operation is then removed from the queue.  The
 546  * task is scheduled to run again immediately until no more operations are in
 547  * the ready queue.
 548  *
 549  * @param cls NULL
 550  */
 551 static void
 552 process_rq_task (void *cls)
 553 {
 554   struct GNUNET_TESTBED_Operation *op;
 555   struct OperationQueue *queue;
 556   unsigned int cnt;
 557
 558   process_rq_task_id = NULL;
 559   GNUNET_assert (NULL != rq_head);
 560   GNUNET_assert (NULL != (op = rq_head->op));
 561   rq_remove (op);
 562   if (NULL != rq_head)
 563     process_rq_task_id = GNUNET_SCHEDULER_add_now (&process_rq_task, NULL);
 564   change_state (op, OP_STATE_ACTIVE);
 565   for (cnt = 0; cnt < op->nqueues; cnt++)
 566   {
 567     queue = op->queues[cnt];
 568     if (OPERATION_QUEUE_TYPE_ADAPTIVE == queue->type)
 569       assign_timeslot (op, queue);
 570   }
 571   op->tstart = GNUNET_TIME_absolute_get ();
 572   if (NULL != op->start)
 573     op->start (op->cb_cls);
 574 }
 575
 576
 577 /**
 578  * Adds the operation to the ready queue and starts the 'process_rq_task'
 579  *
 580  * @param op the operation to be queued
 581  */
 582 static void
 583 rq_add (struct GNUNET_TESTBED_Operation *op)
 584 {
 585   struct ReadyQueueEntry *rq_entry;
 586
 587   GNUNET_assert (NULL == op->rq_entry);
 588   rq_entry = GNUNET_new (struct ReadyQueueEntry);
 589   rq_entry->op = op;
 590   GNUNET_CONTAINER_DLL_insert_tail (rq_head, rq_tail, rq_entry);
 591   op->rq_entry = rq_entry;
 592   if (NULL == process_rq_task_id)
 593     process_rq_task_id = GNUNET_SCHEDULER_add_now (&process_rq_task, NULL);
 594 }
 595
 596
 597 /**
 598  * Checks if the given operation queue is empty or not
 599  *
 600  * @param opq the operation queue
 601  * @return GNUNET_YES if the given operation queue has no operations; GNUNET_NO
 602  *           otherwise
 603  */
 604 static int
 605 is_queue_empty (struct OperationQueue *opq)
 606 {
 607   if ( (NULL != opq->wq_head)
 608        || (NULL != opq->rq_head)
 609        || (NULL != opq->aq_head)
 610        || (NULL != opq->nq_head) )
 611     return GNUNET_NO;
 612   return GNUNET_YES;
 613 }
 614
 615
 616 /**
 617  * Checks if the given operation queue has enough resources to provide for the
 618  * operation of the given queue entry.  It also checks if any inactive
 619  * operations are to be released in order to accommodate the needed resources
 620  * and returns them as an array.
 621  *
 622  * @param opq the operation queue to check for resource accommodation
 623  * @param entry the operation queue entry whose operation's resources are to be
 624  *          accommodated
 625  * @param ops_ pointer to return the array of operations which are to be released
 626  *          in order to accommodate the new operation.  Can be NULL
 627  * @param n_ops_ the number of operations in ops_
 628  * @return GNUNET_YES if the given entry's operation can be accommodated in this
 629  *           queue. GNUNET_NO if it cannot be accommodated; ops_ and n_ops_ will
 630  *           be set to NULL and 0 respectively.
 631  */
 632 static int
 633 decide_capacity (struct OperationQueue *opq,
 634                  struct QueueEntry *entry,
 635                  struct GNUNET_TESTBED_Operation ***ops_,
 636                  unsigned int *n_ops_)
 637 {
 638   struct QueueEntry **evict_entries;
 639   struct GNUNET_TESTBED_Operation **ops;
 640   struct GNUNET_TESTBED_Operation *op;
 641   unsigned int n_ops;
 642   unsigned int n_evict_entries;
 643   unsigned int need;
 644   unsigned int max;
 645   int deficit;
 646   int rval;
 647
 648   GNUNET_assert (NULL != (op = entry->op));
 649   GNUNET_assert (0 < (need = entry->nres));
 650   ops = NULL;
 651   n_ops = 0;
 652   evict_entries = NULL;
 653   n_evict_entries = 0;
 654   rval = GNUNET_YES;
 655   if (OPERATION_QUEUE_TYPE_ADAPTIVE == opq->type)
 656   {
 657     GNUNET_assert (NULL != opq->fctx);
 658     GNUNET_assert (opq->max_active >= opq->overload);
 659     max = opq->max_active - opq->overload;
 660   }
 661   else
 662     max = opq->max_active;
 663   if (opq->active > max)
 664   {
 665     rval = GNUNET_NO;
 666     goto ret;
 667   }
 668   if ((opq->active + need) <= max)
 669     goto ret;
 670   deficit = need - (max - opq->active);
 671   for (entry = opq->nq_head;
 672        (0 < deficit) && (NULL != entry);
 673        entry = entry->next)
 674   {
 675     GNUNET_array_append (evict_entries, n_evict_entries, entry);
 676     deficit -= entry->nres;
 677   }
 678   if (0 < deficit)
 679   {
 680     rval = GNUNET_NO;
 681     goto ret;
 682   }
 683   for (n_ops = 0; n_ops < n_evict_entries;)
 684   {
 685     op = evict_entries[n_ops]->op;
 686     GNUNET_array_append (ops, n_ops, op); /* increments n-ops */
 687   }
 688
 689  ret:
 690   GNUNET_free_non_null (evict_entries);
 691   if (NULL != ops_)
 692     *ops_ = ops;
 693   else
 694     GNUNET_free (ops);
 695   if (NULL != n_ops_)
 696     *n_ops_ = n_ops;
 697   return rval;
 698 }
 699
 700
 701 /**
 702  * Merges an array of operations into another, eliminating duplicates.  No
 703  * ordering is guaranteed.
 704  *
 705  * @param old the array into which the merging is done.
 706  * @param n_old the number of operations in old array
 707  * @param new the array from which operations are to be merged
 708  * @param n_new the number of operations in new array
 709  */
 710 static void
 711 merge_ops (struct GNUNET_TESTBED_Operation ***old,
 712            unsigned int *n_old,
 713            struct GNUNET_TESTBED_Operation **new,
 714            unsigned int n_new)
 715 {
 716   struct GNUNET_TESTBED_Operation **cur;
 717   unsigned int i;
 718   unsigned int j;
 719   unsigned int n_cur;
 720
 721   GNUNET_assert (NULL != old);
 722   n_cur = *n_old;
 723   cur = *old;
 724   for (i = 0; i < n_new; i++)
 725   {
 726     for (j = 0; j < *n_old; j++)
 727     {
 728       if (new[i] == cur[j])
 729         break;
 730     }
 731     if (j < *n_old)
 732       continue;
 733     GNUNET_array_append (cur, n_cur, new[j]);
 734   }
 735   *old = cur;
 736   *n_old = n_cur;
 737 }
 738
 739
 740
 741 /**
 742  * Checks for the readiness of an operation and schedules a operation start task
 743  *
 744  * @param op the operation
 745  */
 746 static int
 747 check_readiness (struct GNUNET_TESTBED_Operation *op)
 748 {
 749   struct GNUNET_TESTBED_Operation **evict_ops;
 750   struct GNUNET_TESTBED_Operation **ops;
 751   unsigned int n_ops;
 752   unsigned int n_evict_ops;
 753   unsigned int i;
 754
 755   GNUNET_assert (NULL == op->rq_entry);
 756   GNUNET_assert (OP_STATE_WAITING == op->state);
 757   evict_ops = NULL;
 758   n_evict_ops = 0;
 759   for (i = 0; i < op->nqueues; i++)
 760   {
 761     ops = NULL;
 762     n_ops = 0;
 763     if (GNUNET_NO == decide_capacity (op->queues[i], op->qentries[i],
 764                                       &ops, &n_ops))
 765     {
 766       GNUNET_free_non_null (evict_ops);
 767       return GNUNET_NO;
 768     }
 769     if (NULL == ops)
 770       continue;
 771     merge_ops (&evict_ops, &n_evict_ops, ops, n_ops);
 772     GNUNET_free (ops);
 773   }
 774   if (NULL != evict_ops)
 775   {
 776     for (i = 0; i < n_evict_ops; i++)
 777       GNUNET_TESTBED_operation_release_ (evict_ops[i]);
 778     GNUNET_free (evict_ops);
 779     evict_ops = NULL;
 780     /* Evicting the operations should schedule this operation */
 781     GNUNET_assert (OP_STATE_READY == op->state);
 782     return GNUNET_YES;
 783   }
 784   for (i = 0; i < op->nqueues; i++)
 785     op->queues[i]->active += op->nres[i];
 786   change_state (op, OP_STATE_READY);
 787   rq_add (op);
 788   return GNUNET_YES;
 789 }
 790
 791
 792 /**
 793  * Defers a ready to be executed operation back to waiting
 794  *
 795  * @param op the operation to defer
 796  */
 797 static void
 798 defer (struct GNUNET_TESTBED_Operation *op)
 799 {
 800   unsigned int i;
 801
 802   GNUNET_assert (OP_STATE_READY == op->state);
 803   rq_remove (op);
 804   for (i = 0; i < op->nqueues; i++)
 805   {
 806     GNUNET_assert (op->queues[i]->active >= op->nres[i]);
 807     op->queues[i]->active -= op->nres[i];
 808   }
 809   change_state (op, OP_STATE_WAITING);
 810 }
 811
 812
 813 /**
 814  * Cleanups the array of timeslots of an operation queue.  For each time slot in
 815  * the array, if it is allocated to an operation, it will be deallocated from
 816  * the operation
 817  *
 818  * @param queue the operation queue
 819  */
 820 static void
 821 cleanup_tslots (struct OperationQueue *queue)
 822 {
 823   struct FeedbackCtx *fctx = queue->fctx;
 824   struct TimeSlot *tslot;
 825   struct GNUNET_TESTBED_Operation *op;
 826   unsigned int cnt;
 827
 828   GNUNET_assert (NULL != fctx);
 829   for (cnt = 0; cnt < queue->max_active; cnt++)
 830   {
 831     tslot = &fctx->tslots_freeptr[cnt];
 832     op = tslot->op;
 833     if (NULL == op)
 834       continue;
 835     GNUNET_CONTAINER_DLL_remove (op->tslots_head, op->tslots_tail, tslot);
 836   }
 837   GNUNET_free_non_null (fctx->tslots_freeptr);
 838   fctx->tslots_freeptr = NULL;
 839   fctx->alloc_head = NULL;
 840   fctx->alloc_tail = NULL;
 841   fctx->tslots_filled = 0;
 842 }
 843
 844
 845 /**
 846  * Cleansup the existing timing slots and sets new timing slots in the given
 847  * queue to accommodate given number of max active operations.
 848  *
 849  * @param queue the queue
 850  * @param n the number of maximum active operations.  If n is greater than the
 851  *   maximum limit set while creating the queue, then the minimum of these two
 852  *   will be selected as n
 853  */
 854 static void
 855 adaptive_queue_set_max_active (struct OperationQueue *queue, unsigned int n)
 856 {
 857   struct FeedbackCtx *fctx = queue->fctx;
 858   struct TimeSlot *tslot;
 859   unsigned int cnt;
 860
 861   cleanup_tslots (queue);
 862   n = GNUNET_MIN (n ,fctx->max_active_bound);
 863   fctx->tslots_freeptr = GNUNET_malloc (n * sizeof (struct TimeSlot));
 864   fctx->nfailed = 0;
 865   for (cnt = 0; cnt < n; cnt++)
 866   {
 867     tslot = &fctx->tslots_freeptr[cnt];
 868     tslot->queue = queue;
 869     GNUNET_CONTAINER_DLL_insert_tail (fctx->alloc_head, fctx->alloc_tail, tslot);
 870   }
 871   GNUNET_TESTBED_operation_queue_reset_max_active_ (queue, n);
 872 }
 873
 874
 875 /**
 876  * Adapts parallelism in an adaptive queue by using the statistical data from
 877  * the feedback context.
 878  *
 879  * @param queue the queue
 880  */
 881 static void
 882 adapt_parallelism (struct OperationQueue *queue)
 883 {
 884   struct GNUNET_TIME_Relative avg;
 885   struct FeedbackCtx *fctx;
 886   struct TimeSlot *tslot;
 887   int sd;
 888   unsigned int nvals;
 889   unsigned int cnt;
 890   unsigned int parallelism;
 891
 892   avg = GNUNET_TIME_UNIT_ZERO;
 893   nvals = 0;
 894   fctx = queue->fctx;
 895   for (cnt = 0; cnt < queue->max_active; cnt++)
 896   {
 897     tslot = &fctx->tslots_freeptr[cnt];
 898     avg = GNUNET_TIME_relative_add (avg, tslot->tsum);
 899     nvals += tslot->nvals;
 900   }
 901   GNUNET_assert (nvals >= queue->max_active);
 902   GNUNET_assert (fctx->nfailed <= nvals);
 903   nvals -= fctx->nfailed;
 904   if (0 == nvals)
 905   {
 906     if (1 == queue->max_active)
 907       adaptive_queue_set_max_active (queue, 1);
 908     else
 909       adaptive_queue_set_max_active (queue, queue->max_active / 2);
 910     return;
 911   }
 912   avg = GNUNET_TIME_relative_divide (avg, nvals);
 913   GNUNET_TESTBED_SD_add_data_ (fctx->sd, (unsigned int) avg.rel_value_us);
 914   if (GNUNET_SYSERR ==
 915       GNUNET_TESTBED_SD_deviation_factor_ (fctx->sd,
 916                                            (unsigned int) avg.rel_value_us,
 917                                            &sd))
 918   {
 919     adaptive_queue_set_max_active (queue, queue->max_active); /* no change */
 920     return;
 921   }
 922
 923   parallelism = 0;
 924   if (-1 == sd)
 925     parallelism = queue->max_active + 1;
 926   if (sd <= -2)
 927     parallelism = queue->max_active * 2;
 928   if (1 == sd)
 929     parallelism = queue->max_active - 1;
 930   if (2 <= sd)
 931     parallelism = queue->max_active / 2;
 932   parallelism = GNUNET_MAX (parallelism, ADAPTIVE_QUEUE_DEFAULT_MAX_ACTIVE);
 933   adaptive_queue_set_max_active (queue, parallelism);
 934
 935 #if 0
 936   /* old algorithm */
 937   if (sd < 0)
 938     sd = 0;
 939   GNUNET_assert (0 <= sd);
 940   //GNUNET_TESTBED_SD_add_data_ (fctx->sd, (unsigned int) avg.rel_value_us);
 941   if (0 == sd)
 942   {
 943     adaptive_queue_set_max_active (queue, queue->max_active * 2);
 944     return;
 945   }
 946   if (1 == sd)
 947   {
 948     adaptive_queue_set_max_active (queue, queue->max_active + 1);
 949     return;
 950   }
 951   if (1 == queue->max_active)
 952   {
 953     adaptive_queue_set_max_active (queue, 1);
 954     return;
 955   }
 956   if (2 == sd)
 957   {
 958     adaptive_queue_set_max_active (queue, queue->max_active - 1);
 959     return;
 960   }
 961   adaptive_queue_set_max_active (queue, queue->max_active / 2);
 962 #endif
 963 }
 964
 965
 966 /**
 967  * update tslots with the operation's completion time.  Additionally, if
 968  * updating a timeslot makes all timeslots filled in an adaptive operation
 969  * queue, call adapt_parallelism() for that queue.
 970  *
 971  * @param op the operation
 972  */
 973 static void
 974 update_tslots (struct GNUNET_TESTBED_Operation *op)
 975 {
 976   struct OperationQueue *queue;
 977   struct GNUNET_TIME_Relative t;
 978   struct TimeSlot *tslot;
 979   struct FeedbackCtx *fctx;
 980   unsigned int i;
 981
 982   t = GNUNET_TIME_absolute_get_duration (op->tstart);
 983   while (NULL != (tslot = op->tslots_head)) /* update time slots */
 984   {
 985     queue = tslot->queue;
 986     fctx = queue->fctx;
 987     GNUNET_CONTAINER_DLL_remove (op->tslots_head, op->tslots_tail, tslot);
 988     tslot->op = NULL;
 989     GNUNET_CONTAINER_DLL_insert_tail (fctx->alloc_head, fctx->alloc_tail,
 990                                       tslot);
 991     if (op->failed)
 992     {
 993       fctx->nfailed++;
 994       for (i = 0; i < op->nqueues; i++)
 995         if (queue == op->queues[i])
 996             break;
 997       GNUNET_assert (i != op->nqueues);
 998       op->queues[i]->overload += op->nres[i];
 999     }
1000     tslot->tsum = GNUNET_TIME_relative_add (tslot->tsum, t);
1001     if (0 != tslot->nvals++)
1002       continue;
1003     fctx->tslots_filled++;
1004     if (queue->max_active == fctx->tslots_filled)
1005       adapt_parallelism (queue);
1006   }
1007 }
1008
1009
1010 /**
1011  * Create an 'operation' to be performed.
1012  *
1013  * @param cls closure for the callbacks
1014  * @param start function to call to start the operation
1015  * @param release function to call to close down the operation
1016  * @return handle to the operation
1017  */
1018 struct GNUNET_TESTBED_Operation *
1019 GNUNET_TESTBED_operation_create_ (void *cls, OperationStart start,
1020                                   OperationRelease release)
1021 {
1022   struct GNUNET_TESTBED_Operation *op;
1023
1024   op = GNUNET_new (struct GNUNET_TESTBED_Operation);
1025   op->start = start;
1026   op->state = OP_STATE_INIT;
1027   op->release = release;
1028   op->cb_cls = cls;
1029   return op;
1030 }
1031
1032
1033 /**
1034  * Create an operation queue.
1035  *
1036  * @param type the type of operation queue
1037  * @param max_active maximum number of operations in this
1038  *        queue that can be active in parallel at the same time
1039  * @return handle to the queue
1040  */
1041 struct OperationQueue *
1042 GNUNET_TESTBED_operation_queue_create_ (enum OperationQueueType type,
1043                                         unsigned int max_active)
1044 {
1045   struct OperationQueue *queue;
1046   struct FeedbackCtx *fctx;
1047
1048   queue = GNUNET_new (struct OperationQueue);
1049   queue->type = type;
1050   if (OPERATION_QUEUE_TYPE_FIXED == type)
1051   {
1052     queue->max_active = max_active;
1053   }
1054   else
1055   {
1056     fctx = GNUNET_new (struct FeedbackCtx);
1057     fctx->max_active_bound = max_active;
1058     fctx->sd = GNUNET_TESTBED_SD_init_ (ADAPTIVE_QUEUE_DEFAULT_HISTORY);
1059     queue->fctx = fctx;
1060     adaptive_queue_set_max_active (queue, ADAPTIVE_QUEUE_DEFAULT_MAX_ACTIVE);
1061   }
1062   return queue;
1063 }
1064
1065
1066 /**
1067  * Cleanup the given operation queue.
1068  *
1069  * @param queue the operation queue to destroy
1070  */
1071 static void
1072 queue_destroy (struct OperationQueue *queue)
1073 {
1074   struct FeedbackCtx *fctx;
1075
1076   if (OPERATION_QUEUE_TYPE_ADAPTIVE == queue->type)
1077   {
1078     cleanup_tslots (queue);
1079     fctx = queue->fctx;
1080     GNUNET_TESTBED_SD_destroy_ (fctx->sd);
1081     GNUNET_free (fctx);
1082   }
1083   GNUNET_free (queue);
1084 }
1085
1086
1087 /**
1088  * Destroys an operation queue.  If the queue is still in use by operations it
1089  * is marked as expired and its resources are released in the destructor
1090  * GNUNET_TESTBED_operations_fini().
1091  *
1092  * @param queue queue to destroy
1093  */
1094 void
1095 GNUNET_TESTBED_operation_queue_destroy_ (struct OperationQueue *queue)
1096 {
1097   if (GNUNET_YES != is_queue_empty (queue))
1098   {
1099     GNUNET_assert (0 == queue->expired); /* Are you calling twice on same queue? */
1100     queue->expired = 1;
1101     GNUNET_array_append (expired_opqs, n_expired_opqs, queue);
1102     return;
1103   }
1104   queue_destroy (queue);
1105 }
1106
1107
1108 /**
1109  * Destroys the operation queue if it is empty.  If not empty return GNUNET_NO.
1110  *
1111  * @param queue the queue to destroy if empty
1112  * @return GNUNET_YES if the queue is destroyed.  GNUNET_NO if not (because it
1113  *           is not empty)
1114  */
1115 int
1116 GNUNET_TESTBED_operation_queue_destroy_empty_ (struct OperationQueue *queue)
1117 {
1118   if (GNUNET_NO == is_queue_empty (queue))
1119     return GNUNET_NO;
1120   GNUNET_TESTBED_operation_queue_destroy_ (queue);
1121   return GNUNET_YES;
1122 }
1123
1124
1125 /**
1126  * Rechecks if any of the operations in the given operation queue's waiting list
1127  * can be made active
1128  *
1129  * @param opq the operation queue
1130  */
1131 static void
1132 recheck_waiting (struct OperationQueue *opq)
1133 {
1134   struct QueueEntry *entry;
1135   struct QueueEntry *entry2;
1136
1137   entry = opq->wq_head;
1138   while (NULL != entry)
1139   {
1140     entry2 = entry->next;
1141     if (GNUNET_NO == check_readiness (entry->op))
1142       break;
1143     entry = entry2;
1144   }
1145 }
1146
1147
1148 /**
1149  * Function to reset the maximum number of operations in the given queue. If
1150  * max_active is lesser than the number of currently active operations, the
1151  * active operations are not stopped immediately.
1152  *
1153  * @param queue the operation queue which has to be modified
1154  * @param max_active the new maximum number of active operations
1155  */
1156 void
1157 GNUNET_TESTBED_operation_queue_reset_max_active_ (struct OperationQueue *queue,
1158                                                   unsigned int max_active)
1159 {
1160   struct QueueEntry *entry;
1161
1162   queue->max_active = max_active;
1163   queue->overload = 0;
1164   while ( (queue->active > queue->max_active)
1165           && (NULL != (entry = queue->rq_head)) )
1166     defer (entry->op);
1167   recheck_waiting (queue);
1168 }
1169
1170
1171 /**
1172  * Add an operation to a queue.  An operation can be in multiple queues at
1173  * once. Once the operation is inserted into all the queues
1174  * GNUNET_TESTBED_operation_begin_wait_() has to be called to actually start
1175  * waiting for the operation to become active.
1176  *
1177  * @param queue queue to add the operation to
1178  * @param op operation to add to the queue
1179  * @param nres the number of units of the resources of queue needed by the
1180  *          operation. Should be greater than 0.
1181  */
1182 void
1183 GNUNET_TESTBED_operation_queue_insert2_ (struct OperationQueue *queue,
1184                                          struct GNUNET_TESTBED_Operation *op,
1185                                          unsigned int nres)
1186 {
1187   unsigned int qsize;
1188
1189   GNUNET_assert (0 < nres);
1190   qsize = op->nqueues;
1191   GNUNET_array_append (op->queues, op->nqueues, queue);
1192   GNUNET_array_append (op->nres, qsize, nres);
1193   GNUNET_assert (qsize == op->nqueues);
1194 }
1195
1196
1197 /**
1198  * Add an operation to a queue.  An operation can be in multiple queues at
1199  * once. Once the operation is inserted into all the queues
1200  * GNUNET_TESTBED_operation_begin_wait_() has to be called to actually start
1201  * waiting for the operation to become active. The operation is assumed to take
1202  * 1 queue resource. Use GNUNET_TESTBED_operation_queue_insert2_() if it
1203  * requires more than 1
1204  *
1205  * @param queue queue to add the operation to
1206  * @param op operation to add to the queue
1207  */
1208 void
1209 GNUNET_TESTBED_operation_queue_insert_ (struct OperationQueue *queue,
1210                                         struct GNUNET_TESTBED_Operation *op)
1211 {
1212   return GNUNET_TESTBED_operation_queue_insert2_ (queue, op, 1);
1213 }
1214
1215
1216 /**
1217  * Marks the given operation as waiting on the queues.  Once all queues permit
1218  * the operation to become active, the operation will be activated.  The actual
1219  * activation will occur in a separate task (thus allowing multiple queue
1220  * insertions to be made without having the first one instantly trigger the
1221  * operation if the first queue has sufficient resources).
1222  *
1223  * @param op the operation to marks as waiting
1224  */
1225 void
1226 GNUNET_TESTBED_operation_begin_wait_ (struct GNUNET_TESTBED_Operation *op)
1227 {
1228   GNUNET_assert (NULL == op->rq_entry);
1229   change_state (op, OP_STATE_WAITING);
1230   (void) check_readiness (op);
1231 }
1232
1233
1234 /**
1235  * Marks an active operation as inactive - the operation will be kept in a
1236  * ready-to-be-released state and continues to hold resources until another
1237  * operation contents for them.
1238  *
1239  * @param op the operation to be marked as inactive.  The operation start
1240  *          callback should have been called before for this operation to mark
1241  *          it as inactive.
1242  */
1243 void
1244 GNUNET_TESTBED_operation_inactivate_ (struct GNUNET_TESTBED_Operation *op)
1245 {
1246   struct OperationQueue **queues;
1247   size_t ms;
1248   unsigned int nqueues;
1249   unsigned int i;
1250
1251   GNUNET_assert (OP_STATE_ACTIVE == op->state);
1252   change_state (op, OP_STATE_INACTIVE);
1253   nqueues = op->nqueues;
1254   ms = sizeof (struct OperationQueue *) * nqueues;
1255   queues = GNUNET_malloc (ms);
1256   /* Cloning is needed as the operation be released by waiting operations and
1257      hence its nqueues memory ptr will be freed */
1258   GNUNET_memcpy (queues, op->queues, ms);
1259   for (i = 0; i < nqueues; i++)
1260     recheck_waiting (queues[i]);
1261   GNUNET_free (queues);
1262 }
1263
1264
1265 /**
1266  * Marks and inactive operation as active.  This fuction should be called to
1267  * ensure that the oprelease callback will not be called until it is either
1268  * marked as inactive or released.
1269  *
1270  * @param op the operation to be marked as active
1271  */
1272 void
1273 GNUNET_TESTBED_operation_activate_ (struct GNUNET_TESTBED_Operation *op)
1274 {
1275
1276   GNUNET_assert (OP_STATE_INACTIVE == op->state);
1277   change_state (op, OP_STATE_ACTIVE);
1278 }
1279
1280
1281 /**
1282  * An operation is 'done' (was cancelled or finished); remove
1283  * it from the queues and release associated resources.
1284  *
1285  * @param op operation that finished
1286  */
1287 void
1288 GNUNET_TESTBED_operation_release_ (struct GNUNET_TESTBED_Operation *op)
1289 {
1290   struct QueueEntry *entry;
1291   struct OperationQueue *opq;
1292   unsigned int i;
1293
1294   if (OP_STATE_INIT == op->state)
1295   {
1296     GNUNET_free (op);
1297     return;
1298   }
1299   if (OP_STATE_READY == op->state)
1300     rq_remove (op);
1301   if (OP_STATE_INACTIVE == op->state) /* Activate the operation if inactive */
1302     GNUNET_TESTBED_operation_activate_ (op);
1303   if (OP_STATE_ACTIVE == op->state)
1304     update_tslots (op);
1305   GNUNET_assert (NULL != op->queues);
1306   GNUNET_assert (NULL != op->qentries);
1307   for (i = 0; i < op->nqueues; i++)
1308   {
1309     entry = op->qentries[i];
1310     remove_queue_entry (op, i);
1311     opq = op->queues[i];
1312     switch (op->state)
1313     {
1314     case OP_STATE_INIT:
1315     case OP_STATE_INACTIVE:
1316       GNUNET_assert (0);
1317       break;
1318     case OP_STATE_WAITING:
1319       break;
1320     case OP_STATE_ACTIVE:
1321     case OP_STATE_READY:
1322       GNUNET_assert (0 != opq->active);
1323       GNUNET_assert (opq->active >= entry->nres);
1324       opq->active -= entry->nres;
1325       recheck_waiting (opq);
1326       break;
1327     }
1328     GNUNET_free (entry);
1329   }
1330   GNUNET_free_non_null (op->qentries);
1331   GNUNET_free (op->queues);
1332   GNUNET_free (op->nres);
1333   if (NULL != op->release)
1334     op->release (op->cb_cls);
1335   GNUNET_free (op);
1336 }
1337
1338
1339 /**
1340  * Marks an operation as failed
1341  *
1342  * @param op the operation to be marked as failed
1343  */
1344 void
1345 GNUNET_TESTBED_operation_mark_failed (struct GNUNET_TESTBED_Operation *op)
1346 {
1347   op->failed = GNUNET_YES;
1348 }
1349
1350
1351 /**
1352  * Cleanup expired operation queues.  While doing so, also check for any
1353  * operations which are not completed and warn about them.
1354  */
1355 void __attribute__ ((destructor))
1356 GNUNET_TESTBED_operations_fini ()
1357 {
1358   struct OperationQueue *queue;
1359   unsigned int i;
1360   int warn = 0;
1361
1362   for (i=0; i < n_expired_opqs; i++)
1363   {
1364     queue = expired_opqs[i];
1365     if (GNUNET_NO == is_queue_empty (queue))
1366       warn = 1;
1367     queue_destroy (queue);
1368   }
1369   GNUNET_free_non_null (expired_opqs);
1370   n_expired_opqs = 0;
1371   if (warn)
1372     GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
1373                 "Be disciplined.  Some operations were not marked as done.\n");
1374
1375 }
1376 /* end of testbed_api_operations.c */