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