BDE 4.39.x Production Release
Loading...
Searching...
No Matches
bdlcc_multipriorityqueue.h
Go to the documentation of this file.
1/// @file bdlcc_multipriorityqueue.h
2///
3/// The content of this file has been pre-processed for Doxygen.
4///
5
6
7// bdlcc_multipriorityqueue.h -*-C++-*-
8#ifndef INCLUDED_BDLCC_MULTIPRIORITYQUEUE
9#define INCLUDED_BDLCC_MULTIPRIORITYQUEUE
10
11#include <bsls_ident.h>
12BSLS_IDENT("$Id: $")
13
14/// @defgroup bdlcc_multipriorityqueue bdlcc_multipriorityqueue
15/// @brief Provide a thread-enabled parameterized multi-priority queue.
16/// @addtogroup bdl
17/// @{
18/// @addtogroup bdlcc
19/// @{
20/// @addtogroup bdlcc_multipriorityqueue
21/// @{
22///
23/// <h1> Outline </h1>
24/// * <a href="#bdlcc_multipriorityqueue-purpose"> Purpose</a>
25/// * <a href="#bdlcc_multipriorityqueue-classes"> Classes </a>
26/// * <a href="#bdlcc_multipriorityqueue-description"> Description </a>
27/// * <a href="#bdlcc_multipriorityqueue-thread-enabled-idioms-in-the-bdlcc-multipriorityqueue-interface"> Thread-Enabled Idioms in the bdlcc::MultipriorityQueue Interface </a>
28/// * <a href="#bdlcc_multipriorityqueue-possible-future-enhancements"> Possible Future Enhancements </a>
29/// * <a href="#bdlcc_multipriorityqueue-warning-synchronization-required-on-destruction"> WARNING: Synchronization Required on Destruction </a>
30/// * <a href="#bdlcc_multipriorityqueue-usage"> Usage </a>
31/// * <a href="#bdlcc_multipriorityqueue-example-1-simple-thread-pool"> Example 1: Simple Thread Pool </a>
32/// * <a href="#bdlcc_multipriorityqueue-example-2-multi-threaded-observer"> Example 2: Multi-Threaded Observer </a>
33///
34/// # Purpose {#bdlcc_multipriorityqueue-purpose}
35/// Provide a thread-enabled parameterized multi-priority queue.
36///
37/// # Classes {#bdlcc_multipriorityqueue-classes}
38///
39/// - bdlcc::MultipriorityQueue: thread-enabled, multi-priority queue
40///
41/// @see
42///
43/// # Description {#bdlcc_multipriorityqueue-description}
44/// This component provides a thread-enabled mechanism,
45/// `bdlcc::MultipriorityQueue`, implementing a special-purpose priority queue
46/// container of items of parameterized `TYPE`. Each item has a priority which,
47/// for efficiency of implementation, is limited to a relatively small number
48/// `N` of contiguous integers `[ 0 .. N - 1 ]`, with `N` indicated at
49/// construction, and 0 being the most urgent priority. This queue also takes
50/// an optional allocator, supplied at construction. Once configured, these
51/// instance parameters remain unchanged for the life of each multi-priority
52/// queue.
53///
54/// ## Thread-Enabled Idioms in the bdlcc::MultipriorityQueue Interface {#bdlcc_multipriorityqueue-thread-enabled-idioms-in-the-bdlcc-multipriorityqueue-interface}
55///
56///
57/// The thread-enabled `bdlcc::MultipriorityQueue` is, in many regards, similar
58/// to a value-semantic type in that there is an obvious abstract notion of
59/// "value" that can be described in terms of salient attributes, which for this
60/// type is a sequence of priority/element pairs, constrained to be in
61/// increasing order of priority. There are, however, several differences in
62/// method behavior and signature that arise due to the thread-enabled nature of
63/// the queue and its anticipated usage pattern.
64///
65/// For example, if a queue object is empty, `popFront` will block indefinitely
66/// until an element is added to the queue. Also, since dynamic instance
67/// information, such as the number of elements currently in a queue, can be
68/// out-of-date by the time it is returned, some manipulators (e.g.,
69/// `tryPopFront`) are deliberately combined with an accessor operation (e.g.,
70/// `isEmpty`) in order to guarantee proper behavior.
71///
72/// Finally, note that although the parameterized `TYPE` is expected to at least
73/// support copy construction and assignment, the
74/// `bdlcc::MultipriorityQueue<TYPE>` type currently does not support any
75/// value-semantic operations, since different queues could have different
76/// numbers of priorities, making comparison, assignment and copy construction
77/// awkward.
78///
79/// ## Possible Future Enhancements {#bdlcc_multipriorityqueue-possible-future-enhancements}
80///
81///
82/// In addition to `popFront` and `tryPopFront`, a `bdlcc::MultipriorityQueue`
83/// may some day also provide a `timedPopFront` method. This method would block
84/// until it is able to complete successfully or until the specified time limit
85/// expires.
86///
87/// ## WARNING: Synchronization Required on Destruction {#bdlcc_multipriorityqueue-warning-synchronization-required-on-destruction}
88///
89///
90/// The behavior for the destructor is undefined unless all access or
91/// modification of the object is completed prior to its destruction. Some form
92/// of synchronization, external to the component, is required to ensure the
93/// precondition on the destructor is met. For example, if two (or more)
94/// threads are manipulating a queue, it is *not* safe to anticipate the number
95/// of elements added to the queue, and destroy that queue immediately after the
96/// last element is popped (without additional synchronization) because one of
97/// the corresponding push functions may not have completed (push may, for
98/// instance, signal waiting threads after the element is considered added to
99/// the queue).
100///
101/// ## Usage {#bdlcc_multipriorityqueue-usage}
102///
103///
104/// This section illustrates intended use of this component.
105///
106/// ### Example 1: Simple Thread Pool {#bdlcc_multipriorityqueue-example-1-simple-thread-pool}
107///
108///
109/// This example demonstrates how we might use a `bdlcc::MultipriorityQueue` to
110/// communicate between a single "producer" thread and multiple "consumer"
111/// threads. The "producer" pushes work requests of varying priority onto the
112/// queue, and each "consumer" iteratively takes the highest priority work
113/// request from the queue and services it.
114///
115/// We begin our example with some utility classes that define a simple "work
116/// item":
117/// @code
118/// enum {
119/// k_MAX_CONSUMER_THREADS = 10
120/// };
121///
122/// struct MyWorkData {
123/// int d_i; // input to work to be done
124///
125/// // Work data...
126/// };
127///
128/// struct MyWorkRequest {
129/// enum RequestType {
130/// e_WORK = 1,
131/// e_STOP = 2
132/// };
133///
134/// RequestType d_type;
135/// MyWorkData d_data;
136///
137/// // Work data...
138/// };
139/// @endcode
140/// Next, we provide a simple function to service an individual work item, and a
141/// function to get a work item. The details are unimportant for this example:
142/// @code
143/// void myDoWork(MyWorkData& data)
144/// {
145/// // Do work...
146/// (void)data;
147/// }
148///
149/// int getWorkData(MyWorkData *result)
150/// {
151/// static int count = 0;
152/// result->d_i = rand(); // Only one thread runs this routine, so it
153/// // does not matter that 'rand()' is not
154/// // thread-safe, or that 'count' is 'static'.
155///
156/// return ++count >= 100;
157/// }
158/// @endcode
159/// The `myConsumer` function (below) will pop elements off the queue in
160/// priority order and process them. As discussed above, note that the call to
161/// `queue->popFront(&item)` will block until there is an element available on
162/// the queue. This function will be executed in multiple threads, so that each
163/// thread waits in `queue->popFront()`; `bdlcc::MultipriorityQueue` guarantees
164/// that each thread gets a unique element from the queue:
165/// @code
166/// void myConsumer(bdlcc::MultipriorityQueue<MyWorkRequest> *queue)
167/// {
168/// MyWorkRequest item;
169/// while (1) {
170///
171/// // The 'popFront' function will wait for a 'MyWorkRequest' until
172/// // one is available.
173///
174/// queue->popFront(&item);
175///
176/// if (MyWorkRequest::e_STOP == item.d_type) {
177/// break;
178/// }
179///
180/// myDoWork(item.d_data);
181/// }
182/// }
183/// @endcode
184/// The `myConsumerThread` function below is a callback for `bslmt::ThreadUtil`,
185/// which requires a "C" signature. `bslmt::ThreadUtil::create()` expects a
186/// pointer to this function, and provides that function pointer to the
187/// newly-created thread. The new thread then executes this function.
188///
189/// Since `bslmt::ThreadUtil::create()` uses the familiar "C" convention of
190/// passing a `void` pointer, our function simply casts that pointer to our
191/// required type (`bdlcc::MultipriorityQueue<MyWorkRequest> *`), and then
192/// delegates to the queue-specific function `myConsumer` (above):
193/// @code
194/// extern "C" void *myConsumerThread(void *queuePtr)
195/// {
196/// myConsumer ((bdlcc::MultipriorityQueue<MyWorkRequest>*) queuePtr);
197/// return queuePtr;
198/// }
199/// @endcode
200/// In this simple example, the `myProducer` function (below) serves multiple
201/// roles: it creates the `bdlcc::MultipriorityQueue`, starts the consumer
202/// threads, and then produces and queues work items. When work requests are
203/// exhausted, this function queues one `e_STOP` item for each consumer thread.
204///
205/// When each consumer thread reads a `e_STOP`, it terminates its
206/// thread-handling function. Note that, although the producer cannot control
207/// which thread pops a particular work item, it can rely on the knowledge that
208/// each consumer thread will read a single `e_STOP` item and then terminate.
209///
210/// Finally, the `myProducer` function "joins" each consumer thread, which
211/// ensures that the thread itself will terminate correctly (see the
212/// @ref bslmt_threadutil component-level documentation for details):
213/// @code
214/// void myProducer()
215/// {
216/// enum {
217/// k_NUM_PRIORITIES = 8,
218/// k_NUM_THREADS = 8
219/// };
220///
221/// MyWorkRequest item;
222/// MyWorkData workData;
223///
224/// // Create multi-priority queue with specified number of priorities.
225///
226/// bdlcc::MultipriorityQueue<MyWorkRequest> queue(k_NUM_PRIORITIES);
227///
228/// // Start the specified number of threads.
229///
230/// assert(0 < k_NUM_THREADS
231/// && k_NUM_THREADS <= static_cast<int>(k_MAX_CONSUMER_THREADS));
232/// bslmt::ThreadUtil::Handle consumerHandles[k_MAX_CONSUMER_THREADS];
233///
234/// for (int i = 0; i < k_NUM_THREADS; ++i) {
235/// bslmt::ThreadUtil::create(&consumerHandles[i],
236/// myConsumerThread,
237/// &queue);
238/// }
239///
240/// // Load work data into work requests and push them onto the queue with
241/// // varying priority until all work data has been exhausted.
242///
243/// int count = 0; // used to generate priorities
244///
245/// while (!getWorkData(&workData)) { // see declaration (above)
246/// item.d_type = MyWorkRequest::e_WORK;
247/// item.d_data = workData;
248/// queue.pushBack(item, count % k_NUM_PRIORITIES); // mixed
249/// // priorities
250/// ++count;
251/// }
252///
253/// // Load as many stop requests as there are active consumer threads.
254///
255/// for (int i = 0; i < k_NUM_THREADS; ++i) {
256/// item.d_type = MyWorkRequest::e_STOP;
257/// queue.pushBack(item, k_NUM_PRIORITIES - 1); // lowest priority
258/// }
259///
260/// // Join all of the consumer threads back with the main thread.
261///
262/// for (int i = 0; i < k_NUM_THREADS; ++i) {
263/// bslmt::ThreadUtil::join(consumerHandles[i]);
264/// }
265/// }
266/// @endcode
267///
268/// ### Example 2: Multi-Threaded Observer {#bdlcc_multipriorityqueue-example-2-multi-threaded-observer}
269///
270///
271/// The previous example shows a simple mechanism for distributing work requests
272/// over multiple threads. This approach works well for large tasks that can be
273/// decomposed into discrete, independent tasks that can benefit from parallel
274/// execution. Note also that the various threads are synchronized only at the
275/// end of execution, when the producer "joins" the various consumer threads.
276///
277/// The simple strategy used in the first example works well for tasks that
278/// share no state, and are completely independent of one another. For
279/// instance, a web server might use a similar strategy to distribute `http`
280/// requests across multiple worker threads.
281///
282/// In more complicated examples, it is often necessary or desirable to
283/// synchronize the separate tasks during execution. The second example below
284/// shows a single "Observer" mechanism that receives event notification from
285/// the various worker threads.
286///
287/// We first create a simple `MyEvent` data type. Worker threads will use this
288/// type to report information about their work. In our example, we will report
289/// the "worker Id", the event number, and some arbitrary text.
290///
291/// As with the previous example, class `MyEvent` also contains an `EventType`,
292/// an enumeration that indicates whether the worker has completed all work.
293/// The "Observer" will use this enumerated value to note when a worker thread
294/// has completed its work:
295/// @code
296/// enum {
297/// k_MAX_CONSUMER_THREADS = 10,
298/// k_MAX_EVENT_TEXT = 80
299/// };
300///
301/// struct MyEvent {
302/// enum EventType {
303/// e_IN_PROGRESS = 1,
304/// e_TASK_COMPLETE = 2
305/// };
306///
307/// EventType d_type;
308/// int d_workerId;
309/// int d_eventNumber;
310/// char d_eventText[k_MAX_EVENT_TEXT];
311/// };
312/// @endcode
313/// As noted in the previous example, `bslmt::ThreadUtil::create()` spawns a new
314/// thread, which invokes a simple "C" function taking a `void` pointer. In the
315/// previous example, we simply converted that `void` pointer into a pointer to
316/// `bdlcc::MultipriorityQueue<MyWorkRequest>`.
317///
318/// In this example, however, we want to pass an additional data item. Each
319/// worker thread is initialized with a unique integer value ("worker Id"),
320/// which identifies that thread. We therefore create a simple `struct` that
321/// contains both of these values:
322/// @code
323/// struct MyWorkerData {
324/// int d_workerId;
325/// bdlcc::MultipriorityQueue<MyEvent> *d_queue;
326/// };
327/// @endcode
328/// Function `myWorker` (below) simulates a working thread by enqueuing multiple
329/// `MyEvent` events during execution. In a realistic application, each
330/// `MyEvent` structure would likely contain different textual information. For
331/// the sake of simplicity, however, our loop uses a constant value for the text
332/// field. Note that various priorities are generated to illustrate the
333/// multi-priority aspect of this particular queue:
334/// @code
335/// void myWorker(int workerId, bdlcc::MultipriorityQueue<MyEvent> *queue)
336/// {
337/// const int N = queue->numPriorities();
338/// const int NUM_EVENTS = 5;
339/// int eventNumber; // used also to generate mixed priorities
340///
341/// // First push 'NUM_EVENTS' events onto 'queue' with mixed priorities.
342///
343/// for (eventNumber = 0; eventNumber < NUM_EVENTS; ++eventNumber) {
344/// MyEvent ev = {
345/// MyEvent::e_IN_PROGRESS,
346/// workerId,
347/// eventNumber,
348/// "In-Progress Event" // constant (for simplicity)
349/// };
350/// queue->pushBack(ev, eventNumber % N); // mixed priorities
351/// }
352///
353/// // Now push an event to end this task.
354///
355/// MyEvent ev = {
356/// MyEvent::e_TASK_COMPLETE,
357/// workerId,
358/// eventNumber,
359/// "Task Complete"
360/// };
361/// queue->pushBack(ev, N - 1); // lowest priority
362/// }
363/// @endcode
364/// The callback function `myWorkerThread` (below) invoked by
365/// `bslmt::ThreadUtil::create` takes the traditional `void` pointer. The
366/// expected data is the composite structure `MyWorkerData`. The callback
367/// function casts the `void` pointer to the application-specific data type and
368/// then uses the referenced object to construct a call to the `myWorker`
369/// function:
370/// @code
371/// extern "C" void *myWorkerThread(void *vWorkerPtr)
372/// {
373/// MyWorkerData *workerPtr = (MyWorkerData *)vWorkerPtr;
374/// myWorker(workerPtr->d_workerId, workerPtr->d_queue);
375/// return vWorkerPtr;
376/// }
377/// @endcode
378/// For the sake of simplicity, we will implement the Observer behavior (below)
379/// in the main thread. The `void` function `myObserver` starts multiple
380/// threads running the `myWorker` function, reads `MyEvent` values from the
381/// queue, and logs all messages in the order of arrival.
382///
383/// As each `myWorker` thread terminates, it sends a `e_TASK_COMPLETE` event.
384/// Upon receiving this event, the `myObserver` function uses the `d_workerId`
385/// to find the relevant thread, and then "joins" that thread.
386///
387/// The `myObserver` function determines when all tasks have completed simply by
388/// counting the number of `e_TASK_COMPLETE` messages received:
389/// @code
390/// void myObserver()
391/// {
392/// const int k_NUM_THREADS = 10;
393/// const int k_NUM_PRIORITIES = 4;
394///
395/// bdlcc::MultipriorityQueue<MyEvent> queue(k_NUM_PRIORITIES);
396///
397/// assert(0 < k_NUM_THREADS
398/// && k_NUM_THREADS <= static_cast<int>(k_MAX_CONSUMER_THREADS));
399/// bslmt::ThreadUtil::Handle workerHandles[k_MAX_CONSUMER_THREADS];
400///
401/// // Create `k_NUM_THREADS` threads, each having a unique "worker id".
402///
403/// MyWorkerData workerData[k_NUM_THREADS];
404/// for (int i = 0; i < k_NUM_THREADS; ++i) {
405/// workerData[i].d_queue = &queue;
406/// workerData[i].d_workerId = i;
407/// bslmt::ThreadUtil::create(&workerHandles[i],
408/// myWorkerThread,
409/// &workerData[i]);
410/// }
411///
412/// // Now print out each of the `MyEvent` values as the threads complete.
413/// // This function ends after a total of `k_NUM_THREADS`
414/// // `MyEvent::e_TASK_COMPLETE` events have been printed.
415///
416/// int nStop = 0;
417/// while (nStop < k_NUM_THREADS) {
418/// MyEvent ev;
419/// queue.popFront(&ev);
420/// bsl::cout << "[" << ev.d_workerId << "] "
421/// << ev.d_eventNumber << ". "
422/// << ev.d_eventText << bsl::endl;
423/// if (MyEvent::e_TASK_COMPLETE == ev.d_type) {
424/// ++nStop;
425/// bslmt::ThreadUtil::join(workerHandles[ev.d_workerId]);
426/// }
427/// }
428/// }
429/// @endcode
430/// @}
431/** @} */
432/** @} */
433
434/** @addtogroup bdl
435 * @{
436 */
437/** @addtogroup bdlcc
438 * @{
439 */
440/** @addtogroup bdlcc_multipriorityqueue
441 * @{
442 */
443
444#include <bdlscm_version.h>
445
446#include <bdlb_bitutil.h>
447
448#include <bdlma_concurrentpool.h>
449
451
452#include <bslma_allocator.h>
454#include <bslma_default.h>
455#include <bslma_managedptr.h>
457
458#include <bslmf_movableref.h>
460
461#include <bslmt_condition.h>
462#include <bslmt_lockguard.h>
463#include <bslmt_mutex.h>
464#include <bslmt_threadutil.h>
465
466#include <bsls_assert.h>
467#include <bsls_atomic.h>
468
469#include <bsl_climits.h>
470#include <bsl_cstdint.h>
471#include <bsl_new.h>
472#include <bsl_vector.h>
473
474#ifndef BDE_DONT_ALLOW_TRANSITIVE_INCLUDES
475#include <bslalg_typetraits.h>
476#endif // BDE_DONT_ALLOW_TRANSITIVE_INCLUDES
477
478
479namespace bdlcc {
480
481 // =========================================
482 // local class MultipriorityQueue_Node<TYPE>
483 // =========================================
484
485/// This class handles storage of one item of parameterized `TYPE` as a node
486/// in a linked list of items stored in a multipriority queue for a given
487/// priority. This class is not to be used from outside this component.
488///
489/// See @ref bdlcc_multipriorityqueue
490template <class TYPE>
492
493 // DATA
494 bslalg::ConstructorProxy<TYPE> d_item; // object stored in node
495 MultipriorityQueue_Node<TYPE> *d_next_p; // next node on linked list
496
497 private:
498 // NOT IMPLEMENTED
501
502 public:
503 // TRAITS
506
507 // CREATORS
508
509 /// Create a node containing a copy of the specified `item` and having
510 /// the specified `next` pointer. Use the specified `basicAllocator` to supply memory.
511 ///
512 /// \pre The behavior is undefined unless `basicAllocator` is non-null.
513 ///
514 /// \note Note that `item` must be copyable and assignable.
515 MultipriorityQueue_Node(const TYPE& item,
516 bslma::Allocator *basicAllocator);
517
518 /// Create a node containing the value of the specified `item` and
519 /// having the specified `next` pointer. `item` is left in a valid but
520 /// unspecified state. Use the specified `basicAllocator` to supply memory.
521 ///
522 /// \pre The behavior is undefined unless `basicAllocator` is
523 /// non-null.
525 bslma::Allocator *basicAllocator);
526
527 /// Destroy this node and free all memory that was allocated on its
528 /// behalf, if any.
530
531 // MANIPULATORS
532
533 /// Return a reference to the non-modifiable item stored in this node.
534 TYPE& item();
535
536 /// Return a reference to the modifiable pointer to the node following
537 /// this node on the linked list.
539
540 // ACCESSORS
541
542 /// Return a pointer to the non-modifiable node following this node on
543 /// the linked list, or 0 if this node has no successor.
544 const MultipriorityQueue_Node *nextPtr() const;
545};
546
547 // ==============================
548 // class MultipriorityQueue<TYPE>
549 // ==============================
550
551/// This class implements a thread-enabled multipriority queue whose
552/// priorities are restricted to a (small) set of contiguous `N` integer
553/// values, `[ 0 .. N - 1 ]`, with 0 being the most urgent.
554///
555/// This class does have a notion of value, namely the sequence of
556/// priority/element pairs, constrained to be in decreasing order of urgency
557/// (i.e., monotonically increasing priority values). However, no value-semantic operations are implemented.
558///
559/// \note Note that elements having
560/// the same priority are maintained in First-In-First-Out (FIFO) order.
561///
562/// \note Note that the current implementation supports up to a maximum of
563/// `sizeof(int) * CHAR_BIT` priorities.
564///
565/// This class is implemented as a set of linked lists, one for each
566/// priority. Two vectors are used to maintain head and tail pointers for
567/// the lists.
568///
569/// See @ref bdlcc_multipriorityqueue
570template <class TYPE>
572
573 // PRIVATE CONSTANTS
574 enum {
575 k_BITS_PER_INT = sizeof(int) * CHAR_BIT,
576 k_DEFAULT_NUM_PRIORITIES = k_BITS_PER_INT,
577 k_MAX_NUM_PRIORITIES = k_BITS_PER_INT
578 };
579
580 // PRIVATE TYPES
581
582 /// The type of the elements on the linked lists of items that are
583 /// maintained for the `N` priorities handled by this multipriority
584 /// queue.
586
587 /// The type of the vectors of list head and tail pointers.
589
590 // DATA
591 mutable bslmt::Mutex d_mutex; // used to synchronize access
592 // (including 'const' access)
593
594 bslmt::Condition d_notEmptyCondition;
595 // signaled on each push
596
597 NodePtrVector d_heads; // pointers to heads of linked lists
598 // -- one for each priority
599
600 NodePtrVector d_tails; // pointers to tails of linked lists
601 // -- one for each priority
602
603 int d_notEmptyFlags; // bit mask indicating priorities
604 // for which there is data, where
605 // bit 0 is the lowest order bit,
606 // representing most urgent priority
607
608 bdlma::ConcurrentPool d_pool; // memory pool used for node storage
609
610 bsls::AtomicInt d_length; // total number of items in this
611 // multipriority queue
612
613 bool d_enabledFlag; // enabled/disabled state of pushes
614 // to the multipriority queue (does
615 // not affect pops)
616
617 bslma::Allocator *d_allocator_p; // memory allocator (held)
618
619 private:
620 // NOT IMPLEMENTED
622 MultipriorityQueue& operator=(const MultipriorityQueue&);
623
624 private:
625 // PRIVATE MANIPULATORS
626
627 /// Attempt to remove (immediately) the least-recently added item having
628 /// the most urgent priority (lowest value) from this multipriority
629 /// queue. If the specified `blockFlag` is `true`, this method blocks
630 /// the calling thread until an item becomes available. On success,
631 /// load the value of the popped item into the specified `item`; if the
632 /// specified `itemPriority` is non-null, load the priority of the
633 /// popped item into `itemPriority`; and return 0. Otherwise, leave
634 /// `item` and `itemPriority` unmodified, and return a non-zero value
635 /// indicating that this multipriority queue was empty.
636 ///
637 /// \pre The behavior is undefined unless `item` is non-null.
638 /// \note Note that a non-zero value can
639 /// be returned only if `blockFlag` is `false`.
640 int tryPopFrontImpl(TYPE *item, int *itemPriority, bool blockFlag);
641
642 public:
643 // TRAITS
646
647 // CREATORS
648
649 /// Create a multipriority queue. Optionally specify `numPriorities`,
650 /// the number of distinct priorities supported by the multipriority
651 /// queue. If `numPriorities` is not specified, the
652 /// (implementation-imposed maximum) number 32 is used. Optionally
653 /// specify a `basicAllocator` used to supply memory. If
654 /// `basicAllocator` is 0, the currently installed default allocator is used.
655 ///
656 /// \pre The behavior is undefined unless `1 <= numPriorities <= 32`
657 /// (if specified).
658 explicit MultipriorityQueue(bslma::Allocator *basicAllocator = 0);
660 bslma::Allocator *basicAllocator = 0);
661
662 /// Destroy this container.
663 /// \pre The behavior is undefined unless all access
664 /// or modification of the container has completed prior to this call.
666
667 // MANIPULATORS
668
669 /// Remove the least-recently added item having the most urgent priority
670 /// (lowest value) from this multi-priority queue and load its value
671 /// into the specified `item`. If this queue is empty, this method
672 /// blocks the calling thread until an item becomes available. If the
673 /// optionally specified `itemPriority` is non-null, load the priority
674 /// of the popped item into `itemPriority`.
675 ///
676 /// \pre The behavior is undefined unless `item` is non-null. Note this is unaffected by the enabled /
677 /// disabled state of the queue.
678 void popFront(TYPE *item, int *itemPriority = 0);
679
680 /// Insert the value of the specified `item` with the specified
681 /// `itemPriority` into this multipriority queue before any queued items
682 /// having a less urgent priority (higher value) than `itemPriority`,
683 /// and after any items having the same or more urgent priority (lower
684 /// value) than `itemPriority`. If the multipriority queue is enabled,
685 /// the push succeeds and `0` is returned, otherwise the push fails, the
686 /// queue remains unchanged, and a nonzero value is returned.
687 ///
688 /// \pre The behavior is undefined unless `0 <= itemPriority < numPriorities()`.
689 int pushBack(const TYPE& item, int itemPriority);
690
691 /// Insert the value of the specified `item` with the specified
692 /// `itemPriority` into this multipriority queue before any queued items
693 /// having a less urgent priority (higher value) than `itemPriority`,
694 /// and after any items having the same or more urgent priority (lower
695 /// value) than `itemPriority`. `item` is left in a valid but
696 /// unspecified state. If the multipriority queue is enabled, the push
697 /// succeeds and `0` is returned, otherwise the push fails, the queue
698 /// remains unchanged, and a nonzero value is returned.
699 ///
700 /// \pre The behavior is undefined unless `0 <= itemPriority < numPriorities()`.
701 int pushBack(bslmf::MovableRef<TYPE> item, int itemPriority);
702
703 /// Insert the value of the specified `item` with the specified
704 /// `itemPriority` onto the back of this multipriority queue before any
705 /// queued items having a less urgent priority (higher value) than
706 /// `itemPriority`, and after any items having the same or more urgent
707 /// priority (lower value) than `itemPriority`. All of the specified
708 /// `numItems` items are pushed as a single atomic action, unless the
709 /// copy constructor for one of them throws an exception, in which case
710 /// a possibly empty subset of the pushes will have completed and no
711 /// memory will be leaked. `Raw` means that the push will succeed even if the multipriority queue is disabled.
712 ///
713 /// \note Note that this method is
714 /// targeted for specific use by the class `bdlmt::MultipriorityThreadPool`.
715 ///
716 /// \pre The behavior is undefined unless
717 /// `0 <= itemPriority < numPriorities()`.
718 void pushBackMultipleRaw(const TYPE& item, int itemPriority, int numItems);
719
720 /// Insert the value of the specified `item` with the specified
721 /// `itemPriority` into the front of this multipriority queue the
722 /// specified `numItems` times, before any queued items having the same
723 /// or less urgent priority (higher value) than `itemPriority`, and
724 /// after any items having more urgent priority (lower value) than
725 /// `itemPriority`. All `numItems` items are pushed as a single atomic
726 /// action, unless the copy constructor throws while creating one of
727 /// them, in which case a possibly empty subset of the pushes will have
728 /// completed and no memory will be leaked. `Raw` means that the push
729 /// will succeed even if the multipriority queue is disabled.
730 ///
731 /// \pre The behavior is undefined unless `0 <= itemPriority < numPriorities()`.
732 ///
733 /// \note Note that this method is targeted at specific uses by the class
734 /// `bdlmt::MultipriorityThreadPool`.
735 void pushFrontMultipleRaw(const TYPE& item,
736 int itemPriority,
737 int numItems);
738
739 /// Attempt to remove (immediately) the least-recently added item having
740 /// the most urgent priority (lowest value) from this multi-priority
741 /// queue. On success, load the value of the popped item into the
742 /// specified `item`; if the optionally specified `itemPriority` is
743 /// non-null, load the priority of the popped item into `itemPriority`;
744 /// and return 0. Otherwise, leave `item` and `itemPriority`
745 /// unmodified, and return a non-zero value indicating that this queue was empty.
746 ///
747 /// \pre The behavior is undefined unless `item` is non-null.
748 /// Note this is unaffected by the enabled / disabled state of the
749 /// queue.
750 int tryPopFront(TYPE *item, int *itemPriority = 0);
751
752 /// Remove and destroy all items from this multi-priority queue.
753 void removeAll();
754
755 /// Enable pushes to this multipriority queue. This method has no
756 /// effect unless the queue was disabled.
757 void enable();
758
759 /// Disable pushes to this multipriority queue. This method has no
760 /// effect unless the queue was enabled.
761 void disable();
762
763 // ACCESSORS
764
765 /// Return the number of distinct priorities (indicated at construction)
766 /// that are supported by this multi-priority queue.
767 int numPriorities() const;
768
769 /// Return the total number of items in this multi-priority queue.
770 int length() const;
771
772 /// Return `true` if there are no items in this multi-priority queue,
773 /// and `false` otherwise.
774 bool isEmpty() const;
775
776 /// Return `true` if this multipriority queue is enable and `false`
777 /// otherwise.
778 bool isEnabled() const;
779};
780
781// ============================================================================
782// INLINE DEFINITIONS
783// ============================================================================
784
785 // -----------------------------------------
786 // local class MultipriorityQueue_Node<TYPE>
787 // -----------------------------------------
788
789// CREATORS
790template <class TYPE>
791inline
793 const TYPE& item,
794 bslma::Allocator *basicAllocator)
795: d_item(item, basicAllocator)
796, d_next_p(0)
797{}
798
799template <class TYPE>
800inline
803 bslma::Allocator *basicAllocator)
804: d_item(bslmf::MovableRefUtil::move(item), basicAllocator)
805, d_next_p(0)
806{}
807
808template <class TYPE>
809inline
812
813// MANIPULATORS
814template <class TYPE>
815inline
817{
818 return d_item.object();
819}
820
821template <class TYPE>
822inline
827
828// ACCESSORS
829template <class TYPE>
830inline
833{
834 return d_next_p;
835}
836
837 // ------------------------------
838 // class MultipriorityQueue<TYPE>
839 // ------------------------------
840
841// PRIVATE MANIPULATORS
842template <class TYPE>
844 int *itemPriority,
845 bool blockFlag)
846{
847 enum { e_SUCCESS = 0, e_FAILURE = -1 };
848
849 Node *condemned;
850 int priority;
851
852 BSLS_ASSERT(item);
853
854 {
855 bslmt::LockGuard<bslmt::Mutex> lock(&d_mutex);
856
857 while (0 == d_length) {
858 // Note that if we get a spurious signal, we will check the
859 // 'blockFlag' unnecessarily, but that will typically be a rare
860 // occurrence. This arrangement minimizes the time taken in the
861 // case where '0 != d_length', which will typically be a frequent
862 // occurrence.
863
864 if (blockFlag) {
865 d_notEmptyCondition.wait(&d_mutex);
866 }
867 else {
868 return e_FAILURE; // RETURN
869 }
870 }
871
873 (bsl::uint32_t)d_notEmptyFlags);
874 BSLS_ASSERT(priority < k_MAX_NUM_PRIORITIES);
875 // verifies there is at least one priority bit set. Note that
876 // 'numTrailingUnsetBits' cannot return a negative value.
877
878 Node *& head = d_heads[priority];
879 condemned = head;
880
881 *item = bslmf::MovableRefUtil::move(condemned->item()); // might throw
882
883 head = head->nextPtr();
884 if (0 == head) {
885 // The last item with this priority was just popped.
886
887 BSLS_ASSERT(d_tails[priority] == condemned);
888 d_notEmptyFlags &= ~(1 << priority);
889 }
890
891 --d_length;
892 }
893
894 if (itemPriority) {
895 *itemPriority = priority;
896 }
897
898 condemned->~Node();
899 d_pool.deallocate(condemned);
900
901 return e_SUCCESS;
902}
903
904// CREATORS
905template <class TYPE>
907: d_heads((typename NodePtrVector::size_type)k_DEFAULT_NUM_PRIORITIES, 0,
908 basicAllocator)
909, d_tails((typename NodePtrVector::size_type)k_DEFAULT_NUM_PRIORITIES, 0,
910 basicAllocator)
911, d_notEmptyFlags(0)
912, d_pool(sizeof(Node), bslma::Default::allocator(basicAllocator))
913, d_length(0)
914, d_enabledFlag(true)
915, d_allocator_p(bslma::Default::allocator(basicAllocator))
916{
917}
918
919template <class TYPE>
921 bslma::Allocator *basicAllocator)
922: d_heads((typename NodePtrVector::size_type)numPriorities, 0, basicAllocator)
923, d_tails((typename NodePtrVector::size_type)numPriorities, 0, basicAllocator)
924, d_notEmptyFlags(0)
925, d_pool(sizeof(Node), bslma::Default::allocator(basicAllocator))
926, d_length(0)
927, d_enabledFlag(true)
928, d_allocator_p(bslma::Default::allocator(basicAllocator))
929{
931 BSLS_ASSERT(k_MAX_NUM_PRIORITIES >= numPriorities);
932}
933
934template <class TYPE>
936{
937 removeAll();
938
939 typename NodePtrVector::iterator it;
940 typename NodePtrVector::iterator endIt;
941
942 for (it = d_heads.begin(), endIt = d_heads.end(); endIt != it; ++it) {
943 BSLS_ASSERT(!*it);
944 }
945
946 // Tails do not get set to null by 'removeAll', so are indeterminate.
947
948 BSLS_ASSERT(isEmpty());
949 BSLS_ASSERT(0 == d_notEmptyFlags);
950}
951
952// MANIPULATORS
953template <class TYPE>
954inline
955void MultipriorityQueue<TYPE>::popFront(TYPE *item, int *itemPriority)
956{
957 tryPopFrontImpl(item, itemPriority, true);
958}
959
960template <class TYPE>
961int MultipriorityQueue<TYPE>::pushBack(const TYPE& item, int itemPriority)
962{
963 enum { e_SUCCESS = 0, e_FAILURE = -1 };
964
965 BSLS_ASSERT((unsigned)itemPriority < d_heads.size());
966
967 // Allocate and copy construct. Note we are doing this work outside the
968 // mutex, which is advantageous in that no one is waiting on us, but it has
969 // the disadvantage that we haven't checked whether this multipriority
970 // queue is disabled, in which case we'll throw the new node away.
971
972 // Note the queue being disabled is not the usual case. Note a race
973 // condition occurs if we check d_enabledFlag outside the mutex.
974
975 Node *newNode = (Node *)d_pool.allocate();
977 &d_pool);
978
979 ::new (newNode) Node(item, d_allocator_p); // might throw
980 deallocator.release();
981 bslma::ManagedPtr<Node> deleter(newNode, &d_pool);
982
983 {
984 bslmt::LockGuard<bslmt::Mutex> lock(&d_mutex);
985
986 if (!d_enabledFlag) {
987 return e_FAILURE; // RETURN
988 }
989
990 deleter.release();
991
992 const int mask = 1 << itemPriority;
993 if (d_notEmptyFlags & mask) {
994 d_tails[itemPriority]->nextPtr() = newNode;
995 }
996 else {
997 d_heads[itemPriority] = newNode;
998 d_notEmptyFlags |= mask;
999 }
1000 d_tails[itemPriority] = newNode;
1001
1002 ++d_length;
1003 }
1004
1005 d_notEmptyCondition.signal();
1006
1007 return e_SUCCESS;
1008}
1009
1010template <class TYPE>
1012 int itemPriority)
1013{
1014 enum { e_SUCCESS = 0, e_FAILURE = -1 };
1015
1016 BSLS_ASSERT((unsigned)itemPriority < d_heads.size());
1017
1018 // Allocate and copy construct. Note we are doing this work outside the
1019 // mutex, which is advantageous in that no one is waiting on us, but it has
1020 // the disadvantage that we haven't checked whether this multipriority
1021 // queue is disabled, in which case we'll throw the new node away.
1022 //
1023 // Note the queue being disabled is not the usual case. Note a race
1024 // condition occurs if we check d_enabledFlag outside the mutex.
1025
1026 Node *newNode = static_cast<Node *>(d_pool.allocate());
1028 &d_pool);
1029
1030 {
1031 bslmt::LockGuard<bslmt::Mutex> lock(&d_mutex);
1032
1033 // Do the enable check before the move, since if it is a move and not a
1034 // copy, there's no backing out after that.
1035
1036 if (!d_enabledFlag) {
1037 return e_FAILURE; // RETURN
1038 }
1039
1040 ::new (newNode) Node(bslmf::MovableRefUtil::move(item), // might throw
1041 d_allocator_p);
1042 deallocator.release();
1043
1044 const int mask = 1 << itemPriority;
1045 if (d_notEmptyFlags & mask) {
1046 d_tails[itemPriority]->nextPtr() = newNode;
1047 }
1048 else {
1049 d_heads[itemPriority] = newNode;
1050 d_notEmptyFlags |= mask;
1051 }
1052 d_tails[itemPriority] = newNode;
1053
1054 ++d_length;
1055 }
1056
1057 d_notEmptyCondition.signal();
1058
1059 return e_SUCCESS;
1060}
1061
1062template <class TYPE>
1064 int itemPriority,
1065 int numItems)
1066{
1067 BSLS_ASSERT((unsigned)itemPriority < d_heads.size());
1068
1069 const int mask = 1 << itemPriority;
1070
1071 {
1072 bslmt::LockGuard<bslmt::Mutex> lock(&d_mutex);
1073
1074 for (int ii = 0; ii < numItems; ++ii) {
1075 Node *newNode = (Node *)d_pool.allocate();
1077 newNode, &d_pool);
1078
1079 ::new (newNode) Node(item, d_allocator_p); // might throw
1080 deallocator.release();
1081
1082 if (d_notEmptyFlags & mask) {
1083 d_tails[itemPriority]->nextPtr() = newNode;
1084 }
1085 else {
1086 d_heads[itemPriority] = newNode;
1087 d_notEmptyFlags |= mask;
1088 }
1089 d_tails[itemPriority] = newNode;
1090
1091 ++d_length;
1092 }
1093 }
1094
1095 for (int ii = 0; ii < numItems; ++ii) {
1096 d_notEmptyCondition.signal();
1097 }
1098}
1099
1100template <class TYPE>
1102 int itemPriority,
1103 int numItems)
1104{
1105 BSLS_ASSERT((unsigned)itemPriority < d_heads.size());
1106
1107 const int mask = 1 << itemPriority;
1108
1109 {
1110 bslmt::LockGuard<bslmt::Mutex> lock(&d_mutex);
1111
1112 for (int ii = 0; ii < numItems; ++ii) {
1113 Node *newNode = (Node *)d_pool.allocate();
1115 newNode, &d_pool);
1116
1117 ::new (newNode) Node(item, d_allocator_p); // might throw
1118 deallocator.release();
1119
1120 Node *& head = d_heads[itemPriority];
1121 if (!head) {
1122 d_tails[itemPriority] = newNode;
1123 d_notEmptyFlags |= mask;
1124 }
1125 newNode->nextPtr() = head;
1126 head = newNode;
1127
1128 ++d_length;
1129 }
1130 }
1131
1132 for (int ii = 0; ii < numItems; ++ii) {
1133 d_notEmptyCondition.signal();
1134 }
1135}
1136
1137template <class TYPE>
1138inline
1139int MultipriorityQueue<TYPE>::tryPopFront(TYPE *item, int *itemPriority)
1140{
1141 return tryPopFrontImpl(item, itemPriority, false);
1142}
1143
1144template <class TYPE>
1146{
1147 Node *condemnedList = 0;
1148
1149 {
1150 bslmt::LockGuard<bslmt::Mutex> lock(&d_mutex);
1151
1152 while (d_notEmptyFlags) {
1153 const int priority = bdlb::BitUtil::numTrailingUnsetBits(
1154 static_cast<bsl::uint32_t>(d_notEmptyFlags));
1155
1156 Node *& head = d_heads[priority];
1157 BSLS_ASSERT(head);
1158
1159 d_tails[priority]->nextPtr() = condemnedList;
1160 condemnedList = head;
1161
1162 head = 0;
1163
1164 d_notEmptyFlags &= ~(1 << priority);
1165 }
1166
1167 BSLS_ASSERT(0 == d_notEmptyFlags);
1168
1169 d_length = 0;
1170 }
1171
1172 Node *node = condemnedList;
1173 while (node) {
1174 Node *condemnedNode = node;
1175 node = node->nextPtr();
1176
1177 condemnedNode->~Node();
1178 d_pool.deallocate(condemnedNode);
1179 }
1180}
1181
1182template <class TYPE>
1183inline
1185{
1186 bslmt::LockGuard<bslmt::Mutex> lock(&d_mutex);
1187
1188 d_enabledFlag = true;
1189}
1190
1191template <class TYPE>
1192inline
1194{
1195 bslmt::LockGuard<bslmt::Mutex> lock(&d_mutex);
1196
1197 d_enabledFlag = false;
1198}
1199
1200// ACCESSORS
1201template <class TYPE>
1202inline
1204{
1205 return static_cast<int>(d_heads.size());
1206}
1207
1208template <class TYPE>
1209inline
1211{
1212 return d_length;
1213}
1214
1215template <class TYPE>
1216inline
1218{
1219 return 0 == d_length;
1220}
1221
1222template <class TYPE>
1223inline
1225{
1226 return d_enabledFlag;
1227}
1228
1229} // close package namespace
1230
1231
1232#endif
1233
1234// ----------------------------------------------------------------------------
1235// Copyright 2015 Bloomberg Finance L.P.
1236//
1237// Licensed under the Apache License, Version 2.0 (the "License");
1238// you may not use this file except in compliance with the License.
1239// You may obtain a copy of the License at
1240//
1241// http://www.apache.org/licenses/LICENSE-2.0
1242//
1243// Unless required by applicable law or agreed to in writing, software
1244// distributed under the License is distributed on an "AS IS" BASIS,
1245// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
1246// See the License for the specific language governing permissions and
1247// limitations under the License.
1248// ----------------------------- END-OF-FILE ----------------------------------
1249
1250/** @} */
1251/** @} */
1252/** @} */
Definition bdlcc_multipriorityqueue.h:491
MultipriorityQueue_Node *& nextPtr()
Definition bdlcc_multipriorityqueue.h:823
~MultipriorityQueue_Node()
Definition bdlcc_multipriorityqueue.h:810
BSLMF_NESTED_TRAIT_DECLARATION(MultipriorityQueue_Node, bslma::UsesBslmaAllocator)
TYPE & item()
Return a reference to the non-modifiable item stored in this node.
Definition bdlcc_multipriorityqueue.h:816
Definition bdlcc_multipriorityqueue.h:571
void pushBackMultipleRaw(const TYPE &item, int itemPriority, int numItems)
Definition bdlcc_multipriorityqueue.h:1063
int pushBack(bslmf::MovableRef< TYPE > item, int itemPriority)
Definition bdlcc_multipriorityqueue.h:1011
int pushBack(const TYPE &item, int itemPriority)
Definition bdlcc_multipriorityqueue.h:961
void enable()
Definition bdlcc_multipriorityqueue.h:1184
void disable()
Definition bdlcc_multipriorityqueue.h:1193
MultipriorityQueue(int numPriorities, bslma::Allocator *basicAllocator=0)
Definition bdlcc_multipriorityqueue.h:920
void popFront(TYPE *item, int *itemPriority=0)
Definition bdlcc_multipriorityqueue.h:955
void pushFrontMultipleRaw(const TYPE &item, int itemPriority, int numItems)
Definition bdlcc_multipriorityqueue.h:1101
bool isEmpty() const
Definition bdlcc_multipriorityqueue.h:1217
int length() const
Return the total number of items in this multi-priority queue.
Definition bdlcc_multipriorityqueue.h:1210
int tryPopFront(TYPE *item, int *itemPriority=0)
Definition bdlcc_multipriorityqueue.h:1139
void removeAll()
Remove and destroy all items from this multi-priority queue.
Definition bdlcc_multipriorityqueue.h:1145
int numPriorities() const
Definition bdlcc_multipriorityqueue.h:1203
bool isEnabled() const
Definition bdlcc_multipriorityqueue.h:1224
BSLMF_NESTED_TRAIT_DECLARATION(MultipriorityQueue, bslma::UsesBslmaAllocator)
MultipriorityQueue(bslma::Allocator *basicAllocator=0)
Definition bdlcc_multipriorityqueue.h:906
~MultipriorityQueue()
Definition bdlcc_multipriorityqueue.h:935
Definition bdlma_concurrentpool.h:332
Definition bslstl_vector.h:1120
Node * * iterator
Definition bslstl_vector.h:1152
Definition bslalg_constructorproxy.h:376
Definition bslma_allocator.h:545
Definition bslma_deallocatorproctor.h:312
void release()
Definition bslma_deallocatorproctor.h:389
Definition bslma_managedptr.h:1173
ManagedPtr_PairProxy< TARGET_TYPE, ManagedPtrDeleter > release()
Definition bslma_managedptr.h:2490
Definition bslmf_movableref.h:752
Definition bslmt_condition.h:220
Definition bslmt_lockguard.h:234
Definition bslmt_mutex.h:317
Definition bsls_atomic.h:744
#define BSLS_ASSERT(X)
Definition bsls_assert.h:1976
#define BSLS_IDENT(str)
BSLS_IDENT() - insert string into .comment binary segment (if supported)
Definition bsls_ident.h:238
Definition bdlcc_boundedqueue.h:270
Definition baljsn_encoder_testtypes.h:76
Definition bdlbb_blob.h:579
static int numTrailingUnsetBits(unsigned int value)
Definition bdlb_bitutil.h:456
Definition bslma_usesbslmaallocator.h:344
static MovableRef< t_TYPE > move(t_TYPE &reference) BSLS_KEYWORD_NOEXCEPT
Definition bslmf_movableref.h:1067