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