src/testbed/testbed_api_operations.c

   1 /*
   2       This file is part of GNUnet
   3       (C) 2008--2012 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 3, 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 testbed/testbed_api_operations.c
  23  * @brief functions to manage operation queues
  24  * @author Christian Grothoff
  25  */
  26 #include "platform.h"
  27 #include "testbed_api_operations.h"
  28
  29
  30 /**
  31  * An entry in the operation queue
  32  */
  33 struct QueueEntry
  34 {
  35   /**
  36    * The next DLL pointer
  37    */
  38   struct QueueEntry *next;
  39
  40   /**
  41    * The prev DLL pointer
  42    */
  43   struct QueueEntry *prev;
  44
  45   /**
  46    * The operation this entry holds
  47    */
  48   struct GNUNET_TESTBED_Operation *op;
  49
  50   /**
  51    * How many units of resources does the operation need
  52    */
  53   unsigned int nres;
  54 };
  55
  56
  57 /**
  58  * Queue of operations where we can only support a certain
  59  * number of concurrent operations of a particular type.
  60  */
  61 struct OperationQueue
  62 {
  63   /**
  64    * DLL head for the wait queue.  Operations which are waiting for this
  65    * operation queue are put here
  66    */
  67   struct QueueEntry *wq_head;
  68
  69   /**
  70    * DLL tail for the wait queue.
  71    */
  72   struct QueueEntry *wq_tail;
  73
  74   /**
  75    * DLL head for the ready queue.  Operations which are in this operation queue
  76    * and are in ready state are put here
  77    */
  78   struct QueueEntry *rq_head;
  79
  80   /**
  81    * DLL tail for the ready queue
  82    */
  83   struct QueueEntry *rq_tail;
  84
  85   /**
  86    * DLL head for the active queue.  Operations which are in this operation
  87    * queue and are currently active are put here
  88    */
  89   struct QueueEntry *aq_head;
  90
  91   /**
  92    * DLL tail for the active queue.
  93    */
  94   struct QueueEntry *aq_tail;
  95
  96   /**
  97    * Number of operations that are currently active in this queue.
  98    */
  99   unsigned int active;
 100
 101   /**
 102    * Max number of operations which can be active at any time in this queue
 103    */
 104   unsigned int max_active;
 105
 106 };
 107
 108
 109 /**
 110  * Operation state
 111  */
 112 enum OperationState
 113 {
 114   /**
 115    * The operation is just created and is in initial state
 116    */
 117   OP_STATE_INIT,
 118
 119   /**
 120    * The operation is currently waiting for resources
 121    */
 122   OP_STATE_WAITING,
 123
 124   /**
 125    * The operation is ready to be started
 126    */
 127   OP_STATE_READY,
 128
 129   /**
 130    * The operation has started
 131    */
 132   OP_STATE_STARTED
 133 };
 134
 135
 136 /**
 137  * An entry in the ready queue (implemented as DLL)
 138  */
 139 struct ReadyQueueEntry
 140 {
 141   /**
 142    * next ptr for DLL
 143    */
 144   struct ReadyQueueEntry *next;
 145
 146   /**
 147    * prev ptr for DLL
 148    */
 149   struct ReadyQueueEntry *prev;
 150
 151   /**
 152    * The operation associated with this entry
 153    */
 154   struct GNUNET_TESTBED_Operation *op;
 155 };
 156
 157
 158 /**
 159  * Opaque handle to an abstract operation to be executed by the testing framework.
 160  */
 161 struct GNUNET_TESTBED_Operation
 162 {
 163   /**
 164    * Function to call when we have the resources to begin the operation.
 165    */
 166   OperationStart start;
 167
 168   /**
 169    * Function to call to clean up after the operation (which may or may
 170    * not have been started yet).
 171    */
 172   OperationRelease release;
 173
 174   /**
 175    * Closure for callbacks.
 176    */
 177   void *cb_cls;
 178
 179   /**
 180    * Array of operation queues this Operation belongs to.
 181    */
 182   struct OperationQueue **queues;
 183
 184   /**
 185    * Array of operation queue entries corresponding to this operation in
 186    * operation queues for this operation
 187    */
 188   struct QueueEntry **qentries;
 189
 190   /**
 191    * Array of number of resources an operation need from each queue. The numbers
 192    * in this array should correspond to the queues array
 193    */
 194   unsigned int *nres;
 195
 196   /**
 197    * Entry corresponding to this operation in ready queue.  Will be NULL if the
 198    * operation is not marked as READY
 199    */
 200   struct ReadyQueueEntry *rq_entry;
 201
 202   /**
 203    * Number of queues in the operation queues array
 204    */
 205   unsigned int nqueues;
 206
 207   /**
 208    * The state of the operation
 209    */
 210   enum OperationState state;
 211
 212 };
 213
 214 /**
 215  * DLL head for the ready queue
 216  */
 217 struct ReadyQueueEntry *rq_head;
 218
 219 /**
 220  * DLL tail for the ready queue
 221  */
 222 struct ReadyQueueEntry *rq_tail;
 223
 224 /**
 225  * The id of the task to process the ready queue
 226  */
 227 GNUNET_SCHEDULER_TaskIdentifier process_rq_task_id;
 228
 229 void
 230 remove_queue_entry (struct GNUNET_TESTBED_Operation *op, unsigned int index)
 231 {
 232   struct OperationQueue *opq;
 233   struct QueueEntry *entry;
 234
 235   opq = op->queues[index];
 236   entry = op->qentries[index];
 237   switch (op->state)
 238   {
 239   case OP_STATE_INIT:
 240     GNUNET_assert (0);
 241     break;
 242   case OP_STATE_WAITING:
 243     GNUNET_CONTAINER_DLL_remove (opq->wq_head, opq->wq_tail, entry);
 244     break;
 245   case OP_STATE_READY:
 246     GNUNET_CONTAINER_DLL_remove (opq->rq_head, opq->rq_tail, entry);
 247     break;
 248   case OP_STATE_STARTED:
 249     GNUNET_CONTAINER_DLL_remove (opq->aq_head, opq->aq_tail, entry);
 250     break;
 251   }
 252 }
 253
 254 void
 255 change_state (struct GNUNET_TESTBED_Operation *op, enum OperationState state)
 256 {
 257   struct QueueEntry *entry;
 258   struct OperationQueue *opq;
 259   unsigned int cnt;
 260   unsigned int s;
 261
 262   GNUNET_assert (OP_STATE_INIT != state);
 263   GNUNET_assert (NULL != op->queues);
 264   GNUNET_assert (NULL != op->nres);
 265   GNUNET_assert ((OP_STATE_INIT == op->state) || (NULL != op->qentries));
 266   GNUNET_assert (op->state != state);
 267   for (cnt = 0; cnt < op->nqueues; cnt++)
 268   {
 269     if (OP_STATE_INIT == op->state)
 270     {
 271       entry = GNUNET_malloc (sizeof (struct QueueEntry));
 272       entry->op = op;
 273       entry->nres = op->nres[cnt];
 274       s = cnt;
 275       GNUNET_array_append (op->qentries, s, entry);
 276     }
 277     else
 278     {
 279       entry = op->qentries[cnt];
 280       remove_queue_entry (op, cnt);
 281     }
 282     opq = op->queues[cnt];
 283     switch (state)
 284     {
 285     case OP_STATE_INIT:
 286       GNUNET_assert (0);
 287       break;
 288     case OP_STATE_WAITING:
 289       GNUNET_CONTAINER_DLL_insert_tail (opq->wq_head, opq->wq_tail, entry);
 290       break;
 291     case OP_STATE_READY:
 292       GNUNET_CONTAINER_DLL_insert_tail (opq->rq_head, opq->rq_tail, entry);
 293       break;
 294     case OP_STATE_STARTED:
 295       GNUNET_CONTAINER_DLL_insert_tail (opq->aq_head, opq->aq_tail, entry);
 296       break;
 297     }
 298   }
 299   op->state = state;
 300 }
 301
 302
 303 /**
 304  * Removes an operation from the ready queue.  Also stops the 'process_rq_task'
 305  * if the given operation is the last one in the queue.
 306  *
 307  * @param op the operation to be removed
 308  */
 309 static void
 310 rq_remove (struct GNUNET_TESTBED_Operation *op)
 311 {
 312   GNUNET_assert (NULL != op->rq_entry);
 313   GNUNET_CONTAINER_DLL_remove (rq_head, rq_tail, op->rq_entry);
 314   GNUNET_free (op->rq_entry);
 315   op->rq_entry = NULL;
 316   if ( (NULL == rq_head) && (GNUNET_SCHEDULER_NO_TASK != process_rq_task_id) )
 317   {
 318     GNUNET_SCHEDULER_cancel (process_rq_task_id);
 319     process_rq_task_id = GNUNET_SCHEDULER_NO_TASK;
 320   }
 321 }
 322
 323
 324 /**
 325  * Processes the ready queue by calling the operation start callback of the
 326  * operation at the head.  The operation is then removed from the queue.  The
 327  * task is scheduled to run again immediately until no more operations are in
 328  * the ready queue.
 329  *
 330  * @param cls NULL
 331  * @param tc scheduler task context.  Not used.
 332  */
 333 static void
 334 process_rq_task (void *cls, const struct GNUNET_SCHEDULER_TaskContext *tc)
 335 {
 336   struct GNUNET_TESTBED_Operation *op;
 337
 338   process_rq_task_id = GNUNET_SCHEDULER_NO_TASK;
 339   GNUNET_assert (NULL != rq_head);
 340   GNUNET_assert (NULL != (op = rq_head->op));
 341   rq_remove (op);
 342   if (NULL != rq_head)
 343     process_rq_task_id = GNUNET_SCHEDULER_add_now (&process_rq_task, NULL);
 344   change_state (op, OP_STATE_STARTED);
 345   if (NULL != op->start)
 346     op->start (op->cb_cls);
 347 }
 348
 349
 350 /**
 351  * Adds the operation to the ready queue and starts the 'process_rq_task'
 352  *
 353  * @param op the operation to be queued
 354  */
 355 static void
 356 rq_add (struct GNUNET_TESTBED_Operation *op)
 357 {
 358   struct ReadyQueueEntry *rq_entry;
 359
 360   GNUNET_assert (NULL == op->rq_entry);
 361   rq_entry = GNUNET_malloc (sizeof (struct ReadyQueueEntry));
 362   rq_entry->op = op;
 363   GNUNET_CONTAINER_DLL_insert_tail (rq_head, rq_tail, rq_entry);
 364   op->rq_entry = rq_entry;
 365   if (GNUNET_SCHEDULER_NO_TASK == process_rq_task_id)
 366     process_rq_task_id = GNUNET_SCHEDULER_add_now (&process_rq_task, NULL);
 367 }
 368
 369
 370 static int
 371 is_queue_empty (struct OperationQueue *opq)
 372 {
 373   if ( (NULL != opq->wq_head)
 374        || (NULL != opq->rq_head)
 375        || (NULL != opq->aq_head) )
 376     return GNUNET_NO;
 377   return GNUNET_YES;
 378 }
 379
 380 /**
 381  * Checks for the readiness of an operation and schedules a operation start task
 382  *
 383  * @param op the operation
 384  */
 385 static void
 386 check_readiness (struct GNUNET_TESTBED_Operation *op)
 387 {
 388   unsigned int i;
 389
 390   GNUNET_assert (NULL == op->rq_entry);
 391   GNUNET_assert (OP_STATE_WAITING == op->state);
 392   for (i = 0; i < op->nqueues; i++)
 393   {
 394     GNUNET_assert (0 < op->nres[i]);
 395     if ((op->queues[i]->active + op->nres[i]) > op->queues[i]->max_active)
 396       return;
 397   }
 398   for (i = 0; i < op->nqueues; i++)
 399     op->queues[i]->active += op->nres[i];
 400   change_state (op, OP_STATE_READY);
 401   rq_add (op);
 402 }
 403
 404
 405 /**
 406  * Defers a ready to be executed operation back to waiting
 407  *
 408  * @param op the operation to defer
 409  */
 410 static void
 411 defer (struct GNUNET_TESTBED_Operation *op)
 412 {
 413   unsigned int i;
 414
 415   GNUNET_assert (OP_STATE_READY == op->state);
 416   rq_remove (op);
 417   for (i = 0; i < op->nqueues; i++)
 418     op->queues[i]->active--;
 419   change_state (op, OP_STATE_WAITING);
 420 }
 421
 422
 423 /**
 424  * Create an 'operation' to be performed.
 425  *
 426  * @param cls closure for the callbacks
 427  * @param start function to call to start the operation
 428  * @param release function to call to close down the operation
 429  * @return handle to the operation
 430  */
 431 struct GNUNET_TESTBED_Operation *
 432 GNUNET_TESTBED_operation_create_ (void *cls, OperationStart start,
 433                                   OperationRelease release)
 434 {
 435   struct GNUNET_TESTBED_Operation *op;
 436
 437   op = GNUNET_malloc (sizeof (struct GNUNET_TESTBED_Operation));
 438   op->start = start;
 439   op->state = OP_STATE_INIT;
 440   op->release = release;
 441   op->cb_cls = cls;
 442   return op;
 443 }
 444
 445
 446 /**
 447  * Create an operation queue.
 448  *
 449  * @param max_active maximum number of operations in this
 450  *        queue that can be active in parallel at the same time
 451  * @return handle to the queue
 452  */
 453 struct OperationQueue *
 454 GNUNET_TESTBED_operation_queue_create_ (unsigned int max_active)
 455 {
 456   struct OperationQueue *queue;
 457
 458   queue = GNUNET_malloc (sizeof (struct OperationQueue));
 459   queue->max_active = max_active;
 460   return queue;
 461 }
 462
 463
 464 /**
 465  * Destroy an operation queue.  The queue MUST be empty
 466  * at this time.
 467  *
 468  * @param queue queue to destroy
 469  */
 470 void
 471 GNUNET_TESTBED_operation_queue_destroy_ (struct OperationQueue *queue)
 472 {
 473   GNUNET_break (GNUNET_YES == is_queue_empty (queue));
 474   GNUNET_free (queue);
 475 }
 476
 477
 478 /**
 479  * Destroys the operation queue if it is empty.  If not empty return GNUNET_NO.
 480  *
 481  * @param queue the queue to destroy if empty
 482  * @return GNUNET_YES if the queue is destroyed.  GNUNET_NO if not (because it
 483  *           is not empty)
 484  */
 485 int
 486 GNUNET_TESTBED_operation_queue_destroy_empty_ (struct OperationQueue *queue)
 487 {
 488   if (GNUNET_NO == is_queue_empty (queue))
 489     return GNUNET_NO;
 490   GNUNET_TESTBED_operation_queue_destroy_ (queue);
 491   return GNUNET_YES;
 492 }
 493
 494
 495 void
 496 recheck_waiting (struct OperationQueue *opq)
 497 {
 498   struct QueueEntry *entry;
 499   struct QueueEntry *entry2;
 500
 501   entry = opq->wq_head;
 502   while ( (NULL != entry) && (opq->active < opq->max_active) )
 503   {
 504     entry2 = entry->next;
 505     check_readiness (entry->op);
 506     entry = entry2;
 507   }
 508 }
 509
 510
 511 /**
 512  * Function to reset the maximum number of operations in the given queue. If
 513  * max_active is lesser than the number of currently active operations, the
 514  * active operations are not stopped immediately.
 515  *
 516  * @param queue the operation queue which has to be modified
 517  * @param max_active the new maximum number of active operations
 518  */
 519 void
 520 GNUNET_TESTBED_operation_queue_reset_max_active_ (struct OperationQueue *queue,
 521                                                   unsigned int max_active)
 522 {
 523   struct QueueEntry *entry;
 524
 525   queue->max_active = max_active;
 526   while ( (queue->active > queue->max_active)
 527           && (NULL != (entry = queue->rq_head)) )
 528     defer (entry->op);
 529   recheck_waiting (queue);
 530 }
 531
 532
 533 /**
 534  * Add an operation to a queue.  An operation can be in multiple queues at
 535  * once. Once the operation is inserted into all the queues
 536  * GNUNET_TESTBED_operation_begin_wait_() has to be called to actually start
 537  * waiting for the operation to become active.
 538  *
 539  * @param queue queue to add the operation to
 540  * @param op operation to add to the queue
 541  * @param nres the number of units of the resources of queue needed by the
 542  *          operation. Should be greater than 0.
 543  */
 544 void
 545 GNUNET_TESTBED_operation_queue_insert2_ (struct OperationQueue *queue,
 546                                          struct GNUNET_TESTBED_Operation *op,
 547                                          unsigned int nres)
 548 {
 549   unsigned int qsize;
 550
 551   GNUNET_assert (0 < nres);
 552   qsize = op->nqueues;
 553   GNUNET_array_append (op->queues, op->nqueues, queue);
 554   GNUNET_array_append (op->nres, qsize, nres);
 555   GNUNET_assert (qsize == op->nqueues);
 556 }
 557
 558
 559 /**
 560  * Add an operation to a queue.  An operation can be in multiple queues at
 561  * once. Once the operation is inserted into all the queues
 562  * GNUNET_TESTBED_operation_begin_wait_() has to be called to actually start
 563  * waiting for the operation to become active. The operation is assumed to take
 564  * 1 queue resource. Use GNUNET_TESTBED_operation_queue_insert2_() if it
 565  * requires more than 1
 566  *
 567  * @param queue queue to add the operation to
 568  * @param op operation to add to the queue
 569  */
 570 void
 571 GNUNET_TESTBED_operation_queue_insert_ (struct OperationQueue *queue,
 572                                         struct GNUNET_TESTBED_Operation *op)
 573 {
 574   return GNUNET_TESTBED_operation_queue_insert2_ (queue, op, 1);
 575 }
 576
 577
 578 /**
 579  * Marks the given operation as waiting on the queues.  Once all queues permit
 580  * the operation to become active, the operation will be activated.  The actual
 581  * activation will occur in a separate task (thus allowing multiple queue
 582  * insertions to be made without having the first one instantly trigger the
 583  * operation if the first queue has sufficient resources).
 584  *
 585  * @param op the operation to marks as waiting
 586  */
 587 void
 588 GNUNET_TESTBED_operation_begin_wait_ (struct GNUNET_TESTBED_Operation *op)
 589 {
 590   GNUNET_assert (NULL == op->rq_entry);
 591   change_state (op, OP_STATE_WAITING);
 592   check_readiness (op);
 593 }
 594
 595
 596 /**
 597  * An operation is 'done' (was cancelled or finished); remove
 598  * it from the queues and release associated resources.
 599  *
 600  * @param op operation that finished
 601  */
 602 void
 603 GNUNET_TESTBED_operation_release_ (struct GNUNET_TESTBED_Operation *op)
 604 {
 605   struct QueueEntry *entry;
 606   struct OperationQueue *opq;
 607   unsigned int i;
 608
 609   if (OP_STATE_INIT == op->state)
 610   {
 611     GNUNET_free (op);
 612     return;
 613   }
 614   if (OP_STATE_READY == op->state)
 615     rq_remove (op);
 616   GNUNET_assert (NULL != op->queues);
 617   GNUNET_assert (NULL != op->qentries);
 618   for (i = 0; i < op->nqueues; i++)
 619   {
 620     entry = op->qentries[i];
 621     remove_queue_entry (op, i);
 622     opq = op->queues[i];
 623     switch (op->state)
 624     {
 625     case OP_STATE_INIT:
 626     case OP_STATE_WAITING:
 627       break;
 628     case OP_STATE_READY:
 629     case OP_STATE_STARTED:
 630       GNUNET_assert (0 != opq->active);
 631       GNUNET_assert (opq->active >= entry->nres);
 632       opq->active -= entry->nres;
 633       recheck_waiting (opq);
 634       break;
 635     }
 636     GNUNET_free (entry);
 637   }
 638   GNUNET_free_non_null (op->qentries);
 639   GNUNET_free (op->queues);
 640   GNUNET_free (op->nres);
 641   if (NULL != op->release)
 642     op->release (op->cb_cls);
 643   GNUNET_free (op);
 644 }
 645
 646
 647 /* end of testbed_api_operations.c */