2 This file is part of GNUnet
3 (C) 2008--2012 Christian Grothoff (and other contributing authors)
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.
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.
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.
22 * @file testbed/testbed_api_operations.c
23 * @brief functions to manage operation queues
24 * @author Christian Grothoff
27 #include "testbed_api_operations.h"
31 * An entry in the operation queue
36 * The next DLL pointer
38 struct QueueEntry *next;
41 * The prev DLL pointer
43 struct QueueEntry *prev;
46 * The operation this entry holds
48 struct GNUNET_TESTBED_Operation *op;
51 * How many units of resources does the operation need
58 * Queue of operations where we can only support a certain
59 * number of concurrent operations of a particular type.
64 * DLL head for the wait queue. Operations which are waiting for this
65 * operation queue are put here
67 struct QueueEntry *wq_head;
70 * DLL tail for the wait queue.
72 struct QueueEntry *wq_tail;
75 * DLL head for the ready queue. Operations which are in this operation queue
76 * and are in ready state are put here
78 struct QueueEntry *rq_head;
81 * DLL tail for the ready queue
83 struct QueueEntry *rq_tail;
86 * DLL head for the active queue. Operations which are in this operation
87 * queue and are currently active are put here
89 struct QueueEntry *aq_head;
92 * DLL tail for the active queue.
94 struct QueueEntry *aq_tail;
97 * Number of operations that are currently active in this queue.
102 * Max number of operations which can be active at any time in this queue
104 unsigned int max_active;
115 * The operation is just created and is in initial state
120 * The operation is currently waiting for resources
125 * The operation is ready to be started
130 * The operation has started
137 * An entry in the ready queue (implemented as DLL)
139 struct ReadyQueueEntry
144 struct ReadyQueueEntry *next;
149 struct ReadyQueueEntry *prev;
152 * The operation associated with this entry
154 struct GNUNET_TESTBED_Operation *op;
159 * Opaque handle to an abstract operation to be executed by the testing framework.
161 struct GNUNET_TESTBED_Operation
164 * Function to call when we have the resources to begin the operation.
166 OperationStart start;
169 * Function to call to clean up after the operation (which may or may
170 * not have been started yet).
172 OperationRelease release;
175 * Closure for callbacks.
180 * Array of operation queues this Operation belongs to.
182 struct OperationQueue **queues;
185 * Array of operation queue entries corresponding to this operation in
186 * operation queues for this operation
188 struct QueueEntry **qentries;
191 * Array of number of resources an operation need from each queue. The numbers
192 * in this array should correspond to the queues array
197 * Entry corresponding to this operation in ready queue. Will be NULL if the
198 * operation is not marked as READY
200 struct ReadyQueueEntry *rq_entry;
203 * Number of queues in the operation queues array
205 unsigned int nqueues;
208 * The state of the operation
210 enum OperationState state;
215 * DLL head for the ready queue
217 struct ReadyQueueEntry *rq_head;
220 * DLL tail for the ready queue
222 struct ReadyQueueEntry *rq_tail;
225 * The id of the task to process the ready queue
227 GNUNET_SCHEDULER_TaskIdentifier process_rq_task_id;
230 remove_queue_entry (struct GNUNET_TESTBED_Operation *op, unsigned int index)
232 struct OperationQueue *opq;
233 struct QueueEntry *entry;
235 opq = op->queues[index];
236 entry = op->qentries[index];
242 case OP_STATE_WAITING:
243 GNUNET_CONTAINER_DLL_remove (opq->wq_head, opq->wq_tail, entry);
246 GNUNET_CONTAINER_DLL_remove (opq->rq_head, opq->rq_tail, entry);
248 case OP_STATE_STARTED:
249 GNUNET_CONTAINER_DLL_remove (opq->aq_head, opq->aq_tail, entry);
255 change_state (struct GNUNET_TESTBED_Operation *op, enum OperationState state)
257 struct QueueEntry *entry;
258 struct OperationQueue *opq;
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++)
269 if (OP_STATE_INIT == op->state)
271 entry = GNUNET_malloc (sizeof (struct QueueEntry));
273 entry->nres = op->nres[cnt];
275 GNUNET_array_append (op->qentries, s, entry);
279 entry = op->qentries[cnt];
280 remove_queue_entry (op, cnt);
282 opq = op->queues[cnt];
288 case OP_STATE_WAITING:
289 GNUNET_CONTAINER_DLL_insert_tail (opq->wq_head, opq->wq_tail, entry);
292 GNUNET_CONTAINER_DLL_insert_tail (opq->rq_head, opq->rq_tail, entry);
294 case OP_STATE_STARTED:
295 GNUNET_CONTAINER_DLL_insert_tail (opq->aq_head, opq->aq_tail, entry);
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.
307 * @param op the operation to be removed
310 rq_remove (struct GNUNET_TESTBED_Operation *op)
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);
316 if ( (NULL == rq_head) && (GNUNET_SCHEDULER_NO_TASK != process_rq_task_id) )
318 GNUNET_SCHEDULER_cancel (process_rq_task_id);
319 process_rq_task_id = GNUNET_SCHEDULER_NO_TASK;
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
331 * @param tc scheduler task context. Not used.
334 process_rq_task (void *cls, const struct GNUNET_SCHEDULER_TaskContext *tc)
336 struct GNUNET_TESTBED_Operation *op;
338 process_rq_task_id = GNUNET_SCHEDULER_NO_TASK;
339 GNUNET_assert (NULL != rq_head);
340 GNUNET_assert (NULL != (op = rq_head->op));
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);
351 * Adds the operation to the ready queue and starts the 'process_rq_task'
353 * @param op the operation to be queued
356 rq_add (struct GNUNET_TESTBED_Operation *op)
358 struct ReadyQueueEntry *rq_entry;
360 GNUNET_assert (NULL == op->rq_entry);
361 rq_entry = GNUNET_malloc (sizeof (struct ReadyQueueEntry));
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);
371 is_queue_empty (struct OperationQueue *opq)
373 if ( (NULL != opq->wq_head)
374 || (NULL != opq->rq_head)
375 || (NULL != opq->aq_head) )
381 * Checks for the readiness of an operation and schedules a operation start task
383 * @param op the operation
386 check_readiness (struct GNUNET_TESTBED_Operation *op)
390 GNUNET_assert (NULL == op->rq_entry);
391 GNUNET_assert (OP_STATE_WAITING == op->state);
392 for (i = 0; i < op->nqueues; i++)
394 GNUNET_assert (0 < op->nres[i]);
395 if ((op->queues[i]->active + op->nres[i]) > op->queues[i]->max_active)
398 for (i = 0; i < op->nqueues; i++)
399 op->queues[i]->active += op->nres[i];
400 change_state (op, OP_STATE_READY);
406 * Defers a ready to be executed operation back to waiting
408 * @param op the operation to defer
411 defer (struct GNUNET_TESTBED_Operation *op)
415 GNUNET_assert (OP_STATE_READY == op->state);
417 for (i = 0; i < op->nqueues; i++)
418 op->queues[i]->active--;
419 change_state (op, OP_STATE_WAITING);
424 * Create an 'operation' to be performed.
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
431 struct GNUNET_TESTBED_Operation *
432 GNUNET_TESTBED_operation_create_ (void *cls, OperationStart start,
433 OperationRelease release)
435 struct GNUNET_TESTBED_Operation *op;
437 op = GNUNET_malloc (sizeof (struct GNUNET_TESTBED_Operation));
439 op->state = OP_STATE_INIT;
440 op->release = release;
447 * Create an operation queue.
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
453 struct OperationQueue *
454 GNUNET_TESTBED_operation_queue_create_ (unsigned int max_active)
456 struct OperationQueue *queue;
458 queue = GNUNET_malloc (sizeof (struct OperationQueue));
459 queue->max_active = max_active;
465 * Destroy an operation queue. The queue MUST be empty
468 * @param queue queue to destroy
471 GNUNET_TESTBED_operation_queue_destroy_ (struct OperationQueue *queue)
473 GNUNET_break (GNUNET_YES == is_queue_empty (queue));
479 * Destroys the operation queue if it is empty. If not empty return GNUNET_NO.
481 * @param queue the queue to destroy if empty
482 * @return GNUNET_YES if the queue is destroyed. GNUNET_NO if not (because it
486 GNUNET_TESTBED_operation_queue_destroy_empty_ (struct OperationQueue *queue)
488 if (GNUNET_NO == is_queue_empty (queue))
490 GNUNET_TESTBED_operation_queue_destroy_ (queue);
496 recheck_waiting (struct OperationQueue *opq)
498 struct QueueEntry *entry;
499 struct QueueEntry *entry2;
501 entry = opq->wq_head;
502 while ( (NULL != entry) && (opq->active < opq->max_active) )
504 entry2 = entry->next;
505 check_readiness (entry->op);
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.
516 * @param queue the operation queue which has to be modified
517 * @param max_active the new maximum number of active operations
520 GNUNET_TESTBED_operation_queue_reset_max_active_ (struct OperationQueue *queue,
521 unsigned int max_active)
523 struct QueueEntry *entry;
525 queue->max_active = max_active;
526 while ( (queue->active > queue->max_active)
527 && (NULL != (entry = queue->rq_head)) )
529 recheck_waiting (queue);
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.
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.
545 GNUNET_TESTBED_operation_queue_insert2_ (struct OperationQueue *queue,
546 struct GNUNET_TESTBED_Operation *op,
551 GNUNET_assert (0 < nres);
553 GNUNET_array_append (op->queues, op->nqueues, queue);
554 GNUNET_array_append (op->nres, qsize, nres);
555 GNUNET_assert (qsize == op->nqueues);
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
567 * @param queue queue to add the operation to
568 * @param op operation to add to the queue
571 GNUNET_TESTBED_operation_queue_insert_ (struct OperationQueue *queue,
572 struct GNUNET_TESTBED_Operation *op)
574 return GNUNET_TESTBED_operation_queue_insert2_ (queue, op, 1);
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).
585 * @param op the operation to marks as waiting
588 GNUNET_TESTBED_operation_begin_wait_ (struct GNUNET_TESTBED_Operation *op)
590 GNUNET_assert (NULL == op->rq_entry);
591 change_state (op, OP_STATE_WAITING);
592 check_readiness (op);
597 * An operation is 'done' (was cancelled or finished); remove
598 * it from the queues and release associated resources.
600 * @param op operation that finished
603 GNUNET_TESTBED_operation_release_ (struct GNUNET_TESTBED_Operation *op)
605 struct QueueEntry *entry;
606 struct OperationQueue *opq;
609 if (OP_STATE_INIT == op->state)
614 if (OP_STATE_READY == op->state)
616 GNUNET_assert (NULL != op->queues);
617 GNUNET_assert (NULL != op->qentries);
618 for (i = 0; i < op->nqueues; i++)
620 entry = op->qentries[i];
621 remove_queue_entry (op, i);
626 case OP_STATE_WAITING:
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);
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);
647 /* end of testbed_api_operations.c */