bdlcc_singleproducersingleconsumerboundedqueue.h

Go to the documentation of this file.

/// @file bdlcc_singleproducersingleconsumerboundedqueue.h
///
/// The content of this file has been pre-processed for Doxygen.
///
 
 
// bdlcc_singleproducersingleconsumerboundedqueue.h                   -*-C++-*-
 
#ifndef INCLUDED_BDLCC_SINGLEPRODUCERSINGLECONSUMERBOUNDEDQUEUE
#define INCLUDED_BDLCC_SINGLEPRODUCERSINGLECONSUMERBOUNDEDQUEUE
 
#include <bsls_ident.h>
BSLS_IDENT("$Id: $")
 
/// @defgroup bdlcc_singleproducersingleconsumerboundedqueue bdlcc_singleproducersingleconsumerboundedqueue
/// @brief  Provide a thread-aware SPSC bounded queue of values.
/// @addtogroup bdl
/// @{
/// @addtogroup bdlcc
/// @{
/// @addtogroup bdlcc_singleproducersingleconsumerboundedqueue
/// @{
///
/// <h1> Outline </h1>
/// * <a href="#bdlcc_singleproducersingleconsumerboundedqueue-purpose"> Purpose</a>
/// * <a href="#bdlcc_singleproducersingleconsumerboundedqueue-classes"> Classes </a>
/// * <a href="#bdlcc_singleproducersingleconsumerboundedqueue-description"> Description </a>
///   * <a href="#bdlcc_singleproducersingleconsumerboundedqueue-template-requirements"> Template Requirements </a>
///   * <a href="#bdlcc_singleproducersingleconsumerboundedqueue-exception-safety"> Exception Safety </a>
///   * <a href="#bdlcc_singleproducersingleconsumerboundedqueue-move-semantics-in-c-03"> Move Semantics in C++03 </a>
///   * <a href="#bdlcc_singleproducersingleconsumerboundedqueue-usage"> Usage </a>
///     * <a href="#bdlcc_singleproducersingleconsumerboundedqueue-example-1-a-simple-thread-pool"> Example 1: A Simple Thread Pool </a>
///
/// # Purpose {#bdlcc_singleproducersingleconsumerboundedqueue-purpose}
/// Provide a thread-aware SPSC bounded queue of values.
///
/// # Classes {#bdlcc_singleproducersingleconsumerboundedqueue-classes}
///
/// -  bdlcc::SingleProducerSingleConsumerBoundedQueue: SPSCB concurrent queue
///
/// # Description {#bdlcc_singleproducersingleconsumerboundedqueue-description}
/// This component defines a type,
/// `bdlcc::SingleProducerSingleConsumerBoundedQueue`, that provides an
/// efficient, thread-aware bounded (capacity fixed at construction) queue of
/// values assuming a single producer and a single consumer.  The behavior of
/// the methods `pushBack` and `tryPushBack` is undefined unless the use is by a
/// single producer (one thread or a group of threads using external
/// synchronization).  Also, the behavior of the methods `popFront`,
/// `tryPopFront`, and `removeAll` is undefined unless the use is by a single
/// consumer.  This class is ideal for synchronization and communication between
/// threads in a producer-consumer model when a bounded queue is appropriate and
/// there is only one producer thread and one consumer thread.
///
/// The queue provides `pushBack` and `popFront` methods for pushing data into
/// the queue and popping data from the queue.  When the queue is full, the
/// `pushBack` methods block until data is removed from the queue.  When the
/// queue is empty, the `popFront` methods block until data appears in the
/// queue.  Non-blocking methods `tryPushBack` and `tryPopFront` are also
/// provided.  The `tryPushBack` method fails immediately, returning a non-zero
/// value, if the queue is full.  The `tryPopFront` method fails immediately,
/// returning a non-zero value, if the queue is empty.
///
/// The queue may be placed into a "enqueue disabled" state using the
/// `disablePushBack` method.  When disabled, `pushBack` and `tryPushBack` fail
/// immediately and return an error code.  Any threads blocked in `pushBack`
/// when the queue is enqueue disabled return from `pushBack` immediately and
/// return an error code.  The queue may be restored to normal operation with
/// the `enablePushBack` method.
///
/// The queue may be placed into a "dequeue disabled" state using the
/// `disablePopFront` method.  When dequeue disabled, `popFront`, `tryPopFront`,
/// and `waitUntilEmpty` fail immediately and return an error code.  Any threads
/// blocked in `popFront` and `waitUntilEmpty` when the queue is dequeue
/// disabled return immediately and return an error code.  The queue may be
/// restored to normal operation with the `enablePopFront` method.
///
/// ## Template Requirements {#bdlcc_singleproducersingleconsumerboundedqueue-template-requirements}
///
///
/// `bdlcc::SingleProducerSingleConsumerBoundedQueue` is a template that is
/// parameterized on the type of element contained within the queue.  The
/// supplied template argument, `TYPE`, must provide both a default constructor
/// and a copy constructor, as well as an assignment operator.  If the default
/// constructor accepts a `bslma::Allocator *`, `TYPE` must declare the uses
/// `bslma::Allocator` trait (see @ref bslma_usesbslmaallocator ) so that the
/// allocator of the queue is propagated to the elements contained in the queue.
///
/// ## Exception Safety {#bdlcc_singleproducersingleconsumerboundedqueue-exception-safety}
///
///
/// A `bdlcc::SingleProducerSingleConsumerBoundedQueue` is exception neutral,
/// and all of the methods of `bdlcc::SingleProducerSingleConsumerBoundedQueue`
/// provide the strong exception safety guarantee (see @ref bsldoc_glossary ).
///
/// ## Move Semantics in C++03 {#bdlcc_singleproducersingleconsumerboundedqueue-move-semantics-in-c-03}
///
///
/// Move-only types are supported by
/// `bdlcc::SingleProducerSingleConsumerBoundedQueue` on C++11 platforms only
/// (where `BSLMF_MOVABLEREF_USES_RVALUE_REFERENCES` is defined), and are not
/// supported on C++03 platforms.  Unfortunately, in C++03, there are user types
/// where a `bslmf::MovableRef` will not safely degrade to a lvalue reference
/// when a move constructor is not available (types providing a constructor
/// template taking any type), so `bslmf::MovableRefUtil::move` cannot be used
/// directly on a user supplied template type.  See internal bug report 99039150
/// for more information.
///
/// ## Usage {#bdlcc_singleproducersingleconsumerboundedqueue-usage}
///
///
/// This section illustrates intended use of this component.
///
/// ### Example 1: A Simple Thread Pool {#bdlcc_singleproducersingleconsumerboundedqueue-example-1-a-simple-thread-pool}
///
///
/// In the following example a `bdlcc::SingleProducerSingleConsumerBoundedQueue`
/// is used to communicate between a single "producer" thread and a single
/// "consumer" thread.  The "producer" will push work requests onto the queue,
/// and the "consumer" will iteratively take a work request from the queue and
/// service the request.  This example shows a partial, simplified
/// implementation of the `bdlmt::FixedThreadPool` class.  See component
/// @ref bdlmt_fixedthreadpool  for more information.
///
/// First, we define a utility classes that handles a simple "work item":
/// @code
/// /// Work data...
/// struct my_WorkData {
/// };
///
/// struct my_WorkRequest {
///     enum RequestType {
///         e_WORK = 1,
///         e_STOP = 2
///     };
///
///     RequestType d_type;
///     my_WorkData d_data;
///     // Work data...
/// };
/// @endcode
/// Next, we provide a simple function to service an individual work item.  The
/// details are unimportant for this example:
/// @code
/// /// Do some work based upon the specified `data`.
/// void myDoWork(const my_WorkData& data)
/// {
///     // do some stuff...
///     (void)data;
/// }
/// @endcode
/// Then, we define a `myConsumer` function that will pop elements off the queue
/// and process them.  Note that the call to `queue->popFront()` will block
/// until there is an element available on the queue:
/// @code
/// void myConsumer(
///     bdlcc::SingleProducerSingleConsumerBoundedQueue<my_WorkRequest> *queue)
///     // Pop elements from the specified `queue`.
/// {
///     while (1) {
///         // `popFront()` will wait for a `my_WorkRequest` until available.
///
///         my_WorkRequest item;
///         item.d_type = my_WorkRequest::e_WORK;
///
///         assert(0 == queue->popFront(&item));
///
///         if (item.d_type == my_WorkRequest::e_STOP) { break; }
///         myDoWork(item.d_data);
///     }
/// }
/// @endcode
/// Finally, we define a `myProducer` function that serves multiple roles: it
/// creates the `bdlcc::SingleProducerSingleConsumerBoundedQueue`, starts the
/// consumer thread, and then produces and enqueues work items.  When work
/// requests are exhausted, this function enqueues one `e_STOP` item for the
/// consumer queue.  This `e_STOP` item indicates to the consumer thread to
/// terminate its thread-handling function.
/// @code
/// /// Create a queue, start consumer thread, produce and enqueue work.
/// void myProducer()
/// {
///     enum {
///         k_MAX_QUEUE_LENGTH = 100,
///         k_NUM_WORK_ITEMS   = 1000
///     };
///
///     bdlcc::SingleProducerSingleConsumerBoundedQueue<my_WorkRequest>
///                                                  queue(k_MAX_QUEUE_LENGTH);
///
///     bslmt::ThreadGroup consumerThreads;
///     consumerThreads.addThreads(bdlf::BindUtil::bind(&myConsumer, &queue),
///                                1);
///
///     for (int i = 0; i < k_NUM_WORK_ITEMS; ++i) {
///         my_WorkRequest item;
///         item.d_type = my_WorkRequest::e_WORK;
///         item.d_data = my_WorkData(); // some stuff to do
///         queue.pushBack(item);
///     }
///
///     {
///         my_WorkRequest item;
///         item.d_type = my_WorkRequest::e_STOP;
///         queue.pushBack(item);
///     }
///
///     consumerThreads.joinAll();
/// }
/// @endcode
/// @}
/** @} */
/** @} */
 
/** @addtogroup bdl
 * @{
 */
/** @addtogroup bdlcc
 * @{
 */
/** @addtogroup bdlcc_singleproducersingleconsumerboundedqueue
 * @{
 */
 
#include <bdlscm_version.h>
 
#include <bslalg_scalarprimitives.h>
 
#include <bslma_default.h>
#include <bslma_usesbslmaallocator.h>
 
#include <bslmf_movableref.h>
#include <bslmf_nestedtraitdeclaration.h>
 
#include <bslmt_condition.h>
#include <bslmt_lockguard.h>
#include <bslmt_mutex.h>
#include <bslmt_platform.h>
#include <bslmt_threadutil.h>
 
#include <bsls_assert.h>
#include <bsls_atomicoperations.h>
#include <bsls_compilerfeatures.h>
#include <bsls_objectbuffer.h>
#include <bsls_types.h>
 
 
namespace bdlcc {
 
      // ===============================================================
      // class SingleProducerSingleConsumerBoundedQueue_PopCompleteGuard
      // ===============================================================
 
/// This class implements a guard that invokes `TYPE::popComplete` on a
/// `NODE` upon destruction.
///
/// See @ref bdlcc_singleproducersingleconsumerboundedqueue 
template <class TYPE, class NODE>
class SingleProducerSingleConsumerBoundedQueue_PopCompleteGuard {
 
    // PRIVATE TYPES
    typedef typename bsls::Types::Uint64 Uint64;
 
    // DATA
    TYPE   *d_queue_p;  // managed queue owning the managed node
    NODE   *d_node_p;   // managed node
    Uint64  d_index;    // value of 'd_queue_p->d_popIndex'
 
    // NOT IMPLEMENTED
    SingleProducerSingleConsumerBoundedQueue_PopCompleteGuard();
    SingleProducerSingleConsumerBoundedQueue_PopCompleteGuard(
             const SingleProducerSingleConsumerBoundedQueue_PopCompleteGuard&);
    SingleProducerSingleConsumerBoundedQueue_PopCompleteGuard& operator=(
             const SingleProducerSingleConsumerBoundedQueue_PopCompleteGuard&);
 
  public:
    // CREATORS
 
    /// Create a guard managing the specified `queue` and will invoke
    /// `popComplete` with the specified `node` and `index`.
    SingleProducerSingleConsumerBoundedQueue_PopCompleteGuard(TYPE   *queue,
                                                              NODE   *node,
                                                              Uint64  index);
 
    /// Destroy this object and invoke the `TYPE::popComplete`.
    ~SingleProducerSingleConsumerBoundedQueue_PopCompleteGuard();
};
 
              // ==============================================
              // class SingleProducerSingleConsumerBoundedQueue
              // ==============================================
 
template <class TYPE>
#if defined(BSLS_COMPILERFEATURES_SUPPORT_ALIGNAS)
class alignas(bslmt::Platform::e_CACHE_LINE_SIZE)
                                     SingleProducerSingleConsumerBoundedQueue {
#else
class SingleProducerSingleConsumerBoundedQueue {
#endif
    // This class provides a thread-safe bounded queue of values.
 
    // PRIVATE TYPES
    typedef          unsigned int                                Uint;
    typedef typename bsls::Types::Uint64                         Uint64;
    typedef typename bsls::AtomicOperations::AtomicTypes::Uint   AtomicUint;
    typedef typename bsls::AtomicOperations::AtomicTypes::Uint64 AtomicUint64;
    typedef typename bsls::AtomicOperations                      AtomicOp;
 
    // PRIVATE CONSTANTS
    enum {
        // These value are used as values for `d_state` in `Node`.  A node is
        // writable at creation and after a read completes (when the single
        // producer can write to the node).  A node is readable after it is
        // written (when the node can be read by the single consumer).  The
        // states in-between these two states (e.g., writing) are not needed by
        // this implementation of the queue.
 
        e_READABLE,              // node can be read
 
        e_READABLE_AND_BLOCKED,  // node can be read and has blocked producer
 
        e_WRITABLE,              // node can be written
 
        e_WRITABLE_AND_EMPTY,    // node can be written, queue is empty and
                                 // this is the *first* writable node
 
        e_WRITABLE_AND_BLOCKED   // node can be written, queue is empty, this
                                 // is the *first* writable node, and the
                                 // consumer is blocked waiting for this node
                                 // to be readable
    };
 
    // PRIVATE TYPES
    template <class DATA>
    struct QueueNode {
        // PUBLIC DATA
        bsls::ObjectBuffer<DATA> d_value;  // stored value
        AtomicUint               d_state;  // `e_READABLE`, `e_WRITABLE`, etc.
    };
 
    typedef QueueNode<TYPE> Node;
 
    // DATA
    AtomicUint64              d_popIndex;        // index of next element to
                                                 // pop
 
    Node                     *d_popElement_p;    // array of elements that
                                                 // comprise the bounded queue;
                                                 // identical to
                                                 // `d_pushElement_p`
 
    const bsl::size_t         d_popCapacity;     // the capacity of the queue;
                                                 // identical to
                                                 // `d_pushCapacity`
 
    AtomicUint                d_popDisabledGeneration;
                                                 // generation count of pop
                                                 // disablements
 
    mutable AtomicUint        d_emptyCount;      // count of threads in
                                                 // `waitUntilEmpty`
 
    AtomicUint                d_emptyGeneration; // generation count for the
                                                 // empty queue state, queue is
                                                 // empty whenever least
                                                 // significant bit is zero
 
    const char                d_popPad[  bslmt::Platform::e_CACHE_LINE_SIZE
                                       - sizeof(AtomicUint64)
                                       - sizeof(Node *)
                                       - sizeof(bsl::size_t)
                                       - sizeof(AtomicUint)
                                       - sizeof(AtomicUint)
                                       - sizeof(AtomicUint)];
                                                 // padding to prevent
                                                 // subsequent data from being
                                                 // in the same cache line as
                                                 // the prior data
 
    AtomicUint64              d_pushIndex;       // index of next target
                                                 // element for a push
 
    Node                     *d_pushElement_p;   // array of elements that
                                                 // comprise the bounded queue;
                                                 // identical to
                                                 // `d_popElement_p`
 
    const bsl::size_t         d_pushCapacity;    // the capacity of the queue;
                                                 // identical to
                                                 // `d_popCapacity`
 
    AtomicUint                d_pushDisabledGeneration;
                                                 // generation count of push
                                                 // disablements
 
    const char                d_pushPad[  bslmt::Platform::e_CACHE_LINE_SIZE
                                        - sizeof(AtomicUint64)
                                        - sizeof(Node *)
                                        - sizeof(bsl::size_t)
                                        - sizeof(AtomicUint)];
                                                 // padding to prevent
                                                 // subsequent data from being
                                                 // in the same cache line as
                                                 // the prior data
 
    bslmt::Mutex              d_popMutex;        // used with `d_popCondition`
                                                 // to block the consumer when
                                                 // the queue is empty
 
    bslmt::Condition          d_popCondition;    // condition for blocking the
                                                 // consumer when the queue is
                                                 // empty
 
    bslmt::Mutex              d_pushMutex;       // used with `d_pushCondition`
                                                 // to block the producer when
                                                 // the queue is full
 
    bslmt::Condition          d_pushCondition;   // condition for blocking the
                                                 // producer when the queue is
                                                 // full
 
    mutable bslmt::Mutex      d_emptyMutex;      // blocking point for
                                                 // `waitUntilEmpty`
 
    mutable bslmt::Condition  d_emptyCondition;  // condition variable for
                                                 // `waitUntilEmpty`
 
    bslma::Allocator         *d_allocator_p;     // allocator, held not owned
 
    // FRIENDS
    friend class SingleProducerSingleConsumerBoundedQueue_PopCompleteGuard<
                SingleProducerSingleConsumerBoundedQueue<TYPE>,
                typename SingleProducerSingleConsumerBoundedQueue<TYPE>::Node>;
 
    // PRIVATE CLASS METHODS
 
    /// If the specified `value` does not have its lowest-order bit set to
    /// the value of the specified `bitValue`, increment `value` until it
    /// does.  Note that this method is used to modify the generation counts
    /// stored in `d_popDisabledGeneration` and `d_pushDisabledGeneration`.
    static void incrementUntil(AtomicUint *value, unsigned int bitValue);
 
    // PRIVATE MANIPULATORS
 
    /// Destruct the value stored in the specified `node`, use the specified
    /// `index` in calculations to mark the `node` writable, unblock any
    /// blocked "push" threads, and if the queue is empty update the empty
    /// generation and signal the queue empty condition.  This method is
    /// used within `popFrontImp` by a guard to complete the reclamation of
    /// a node in the presence of an exception.
    void popComplete(Node *node, Uint64 index);
 
    /// If the specified `isTry` is `false`, remove the element from the
    /// front of this queue and load that element into the specified
    /// `value`; otherwise, attempt to remove the element from the front of
    /// this queue without blocking, and, if successful, load the `value`
    /// with the removed element.  If `false == isTry` and the queue is
    /// empty, block until it is not empty.  Return 0 on success, and a
    /// non-zero value otherwise.  Specifically, return `e_SUCCESS` on
    /// success, `e_DISABLED` if `isPopFrontDisabled()`, `e_EMPTY` if
    /// `true == isTry`, `!isPopFrontDisabled()`, and the queue is empty,
    /// and `e_FAILED` if an underlying mechanism returns an error.  On
    /// failure, `value` is not changed.  Threads blocked due to the queue
    /// being empty will return `e_DISABLED` if `disablePopFront` is
    /// invoked.
    int popFrontImp(TYPE *value, bool isTry);
 
    /// If the specified `isTry` is `false`, append the specified `value` to
    /// the back of this queue; otherwise, attempt to append the `value` to
    /// the back of this queue without blocking.  Return 0 on success, and a
    /// non-zero value otherwise.  Specifically, return `e_SUCCESS` on
    /// success, `e_DISABLED` if `isPushBackDisabled()`, `e_FULL` if
    /// `true == isTry`, `!isPushBackDisabled()`, and the queue is full, and
    /// `e_FAILED` if an underlying mechanism returns an error.  Threads
    /// blocked due to the queue being full will return `e_DISABLED` if
    /// `disablePushFront` is invoked.
    int pushBackImp(const TYPE& value, bool isTry);
 
    /// If the specified `isTry` is `false`, append the specified
    /// move-insertable `value` to the back of this queue; otherwise,
    /// attempt to append the `value` to the back of this queue without
    /// blocking.  `value` is left in a valid but unspecified state.  Return
    /// 0 on success, and a non-zero value otherwise.  Specifically, return
    /// `e_DISABLED` if `isPushBackDisabled()`, `e_FULL` if `true == isTry`,
    /// `!isPushBackDisabled()`, and the queue is full, and `e_FAILED` if an
    /// underlying mechanism returns an error.  On failure, `value` is not
    /// changed.  Threads blocked due to the queue being full will return
    /// `e_DISABLED` if `disablePushFront` is invoked.
    int pushBackImp(bslmf::MovableRef<TYPE> value, bool isTry);
 
    /// Mark the specified `node` readable, signal `d_popCondition` if
    /// necessary, and update `d_popIndex` to be the index value of the
    /// location to be used after specified `index` location.  This method
    /// is invoked from `pushBackImp`.
    void pushComplete(Node *node, Uint64 index);
 
    // NOT IMPLEMENTED
    SingleProducerSingleConsumerBoundedQueue(
                              const SingleProducerSingleConsumerBoundedQueue&);
    SingleProducerSingleConsumerBoundedQueue& operator=(
                              const SingleProducerSingleConsumerBoundedQueue&);
 
  public:
    // TRAITS
    BSLMF_NESTED_TRAIT_DECLARATION(SingleProducerSingleConsumerBoundedQueue,
                                   bslma::UsesBslmaAllocator);
 
    // PUBLIC TYPES
    typedef TYPE value_type;  // The type for elements.
 
    // PUBLIC CONSTANTS
    enum {
        e_SUCCESS  =  0,
        e_EMPTY    = -1,
        e_FULL     = -2,
        e_DISABLED = -3,
        e_FAILED   = -4
    };
 
    // CREATORS
 
    /// Create a thread-aware queue with at least the specified `capacity`.
    /// Optionally specify a `basicAllocator` used to supply memory.  If
    /// `basicAllocator` is 0, the currently installed default allocator is
    /// used.
    explicit
    SingleProducerSingleConsumerBoundedQueue(
                                         bsl::size_t       capacity,
                                         bslma::Allocator *basicAllocator = 0);
 
    /// Destroy this object.
    ~SingleProducerSingleConsumerBoundedQueue();
 
    // MANIPULATORS
 
    /// Remove the element from the front of this queue and load that
    /// element into the specified `value`.  If the queue is empty, block
    /// until it is not empty.  Return 0 on success, and a non-zero value
    /// otherwise.  Specifically, return `e_SUCCESS` on success,
    /// `e_DISABLED` if `isPopFrontDisabled()` and `e_FAILED` if an
    /// underlying mechanism returns an error.  On failure, `value` is not
    /// changed.  Threads blocked due to the queue being empty will return
    /// `e_DISABLED` if `disablePopFront` is invoked.  The behavior is
    /// undefined unless the invoker of this method is the single consumer.
    int popFront(TYPE *value);
 
    /// Append the specified `value` to the back of this queue.  Return 0 on
    /// success, and a non-zero value otherwise.  Specifically, return
    /// `e_SUCCESS` on success, `e_DISABLED` if `isPushBackDisabled()` and
    /// `e_FAILED` if an underlying mechanism returns an error.  Threads
    /// blocked due to the queue being full will return `e_DISABLED` if
    /// `disablePushFront` is invoked.  The behavior is undefined unless the
    /// invoker of this method is the single producer.
    int pushBack(const TYPE& value);
 
    /// Append the specified move-insertable `value` to the back of this
    /// queue.  `value` is left in a valid but unspecified state.  Return 0
    /// on success, and a non-zero value otherwise.  Specifically, return
    /// `e_SUCCESS` on success, `e_DISABLED` if `isPushBackDisabled()` and
    /// `e_FAILED` if an underlying mechanism returns an error.  On failure,
    /// `value` is not changed.  Threads blocked due to the queue being full
    /// will return `e_DISABLED` if `disablePushFront` is invoked.  The
    /// behavior is undefined unless the invoker of this method is the
    /// single producer.
    int pushBack(bslmf::MovableRef<TYPE> value);
 
    /// Remove all items currently in this queue.  Note that this operation
    /// is not atomic; if other threads are concurrently pushing items into
    /// the queue the result of `numElements()` after this function returns
    /// is not guaranteed to be 0.  The behavior is undefined unless the
    /// invoker of this method is the single consumer.
    void removeAll();
 
    /// Attempt to remove the element from the front of this queue without
    /// blocking, and, if successful, load the specified `value` with the
    /// removed element.  Return 0 on success, and a non-zero value
    /// otherwise.  Specifically, return `e_SUCCESS` on success,
    /// `e_DISABLED` if `isPopFrontDisabled()`, and `e_EMPTY` if
    /// `!isPopFrontDisabled()` and the queue was empty.  On failure,
    /// `value` is not changed.  The behavior is undefined unless the
    /// invoker of this method is the single consumer.
    int tryPopFront(TYPE *value);
 
    /// Append the specified `value` to the back of this queue.  Return 0 on
    /// success, and a non-zero value otherwise.  Specifically, return
    /// `e_SUCCESS` on success, `e_DISABLED` if `isPushBackDisabled()`, and
    /// `e_FULL` if `!isPushBackDisabled()` and the queue was full.  The
    /// behavior is undefined unless the invoker of this method is the
    /// single producer.
    int tryPushBack(const TYPE& value);
 
    /// Append the specified move-insertable `value` to the back of this
    /// queue.  `value` is left in a valid but unspecified state.  Return 0
    /// on success, and a non-zero value otherwise.  Specifically, return
    /// `e_SUCCESS` on success, `e_DISABLED` if `isPushBackDisabled()`, and
    /// `e_FULL` if `!isPushBackDisabled()` and the queue was full.  On
    /// failure, `value` is not changed.  The behavior is undefined unless
    /// the invoker of this method is the single producer.
    int tryPushBack(bslmf::MovableRef<TYPE> value);
 
                       // Enqueue/Dequeue State
 
    /// Disable dequeueing from this queue.  All subsequent invocations of
    /// `popFront` or `tryPopFront` will fail immediately.  If the single
    /// consumer is blocked in `popFront`, the invocation of `popFront` will
    /// fail immediately.  Any blocked invocations of `waitUntilEmpty` will
    /// fail immediately.  If the queue is already dequeue disabled, this
    /// method has no effect.
    void disablePopFront();
 
    /// Disable enqueueing into this queue.  All subsequent invocations of
    /// `pushBack` or `tryPushBack` will fail immediately.  If the single
    /// producer is blocked in `pushBack`, the invocation of `pushBack` will
    /// fail immediately.  If the queue is already enqueue disabled, this
    /// method has no effect.
    void disablePushBack();
 
    /// Enable queuing.  If the queue is not enqueue disabled, this call has
    /// no effect.
    void enablePushBack();
 
    /// Enable dequeueing.  If the queue is not dequeue disabled, this call
    /// has no effect.
    void enablePopFront();
 
    // ACCESSORS
 
    /// Return the maximum number of elements that may be stored in this
    /// queue.  Note that the value returned may be greater than that
    /// supplied at construction.
    bsl::size_t capacity() const;
 
    /// Return `true` if this queue is empty (has no elements), or `false`
    /// otherwise.
    bool isEmpty() const;
 
    /// Return `true` if this queue is full (has no available capacity), or
    /// `false` otherwise.
    bool isFull() const;
 
    /// Return `true` if this queue is dequeue disabled, and `false`
    /// otherwise.  Note that the queue is created in the "dequeue enabled"
    /// state.
    bool isPopFrontDisabled() const;
 
    /// Return `true` if this queue is enqueue disabled, and `false`
    /// otherwise.  Note that the queue is created in the "enqueue enabled"
    /// state.
    bool isPushBackDisabled() const;
 
    /// Returns the number of elements currently in this queue.
    bsl::size_t numElements() const;
 
    /// Block until all the elements in this queue are removed.  Return 0 on
    /// success, and a non-zero value otherwise.  Specifically, return
    /// `e_SUCCESS` on success, `e_DISABLED` if `isPopFrontDisabled()` and
    /// `e_FAILED` if an underlying mechanism returns an error.  A blocked
    /// thread waiting for the queue to empty will return a non-zero value
    /// if `disablePopFront` is invoked.
    int waitUntilEmpty() const;
 
                                  // Aspects
 
    /// Return the allocator used by this object to supply memory.
    bslma::Allocator *allocator() const;
};
 
// ============================================================================
//                             INLINE DEFINITIONS
// ============================================================================
 
     // ---------------------------------------------------------------
     // class SingleProducerSingleConsumerBoundedQueue_PopCompleteGuard
     // ---------------------------------------------------------------
 
// CREATORS
template <class TYPE, class NODE>
inline
SingleProducerSingleConsumerBoundedQueue_PopCompleteGuard<TYPE, NODE>
                   ::SingleProducerSingleConsumerBoundedQueue_PopCompleteGuard(
                                                                 TYPE   *queue,
                                                                 NODE   *node,
                                                                 Uint64  index)
: d_queue_p(queue)
, d_node_p(node)
, d_index(index)
{
}
 
template <class TYPE, class NODE>
inline
SingleProducerSingleConsumerBoundedQueue_PopCompleteGuard<TYPE, NODE>
                 ::~SingleProducerSingleConsumerBoundedQueue_PopCompleteGuard()
{
    d_queue_p->popComplete(d_node_p, d_index);
}
 
              // ----------------------------------------------
              // class SingleProducerSingleConsumerBoundedQueue
              // ----------------------------------------------
 
// PRIVATE CLASS METHODS
template <class TYPE>
void SingleProducerSingleConsumerBoundedQueue<TYPE>
                     ::incrementUntil(AtomicUint *value, unsigned int bitValue)
{
    unsigned int state = AtomicOp::getUintAcquire(value);
    if (bitValue != (state & 1)) {
        unsigned int expState;
        do {
            expState = state;
            state = AtomicOp::testAndSwapUintAcqRel(value,
                                                     state,
                                                     state + 1);
        } while (state != expState && (bitValue == (state & 1)));
    }
}
 
// PRIVATE MANIPULATORS
template <class TYPE>
inline
void SingleProducerSingleConsumerBoundedQueue<TYPE>::popComplete(Node   *node,
                                                                 Uint64  index)
{
    ++index;
    if (index == d_popCapacity) {
        index = 0;
    }
    AtomicOp::setUint64Release(&d_popIndex, index);
 
    node->d_value.object().~TYPE();
 
    Uint nodeState = AtomicOp::swapUintAcqRel(&node->d_state, e_WRITABLE);
    if (e_READABLE_AND_BLOCKED == nodeState) {
        {
            bslmt::LockGuard<bslmt::Mutex> guard(&d_pushMutex);
        }
        d_pushCondition.signal();
    }
 
    // If the node subsequent to 'node' is writable, the queue is empty and the
    // node subsequent to 'node' must be marked as 'e_WRITABLE_AND_EMPTY'.
 
    nodeState = AtomicOp::testAndSwapUintAcqRel(&d_popElement_p[index].d_state,
                                                e_WRITABLE,
                                                e_WRITABLE_AND_EMPTY);
    if (e_WRITABLE == nodeState) {
        // The queue is empty, increment the empty generation count.
 
        AtomicOp::addUintAcqRel(&d_emptyGeneration, 1);
        if (0 < AtomicOp::getUintAcquire(&d_emptyCount)) {
            {
                bslmt::LockGuard<bslmt::Mutex> guard(&d_emptyMutex);
            }
            d_emptyCondition.broadcast();
        }
    }
}
 
template <class TYPE>
int SingleProducerSingleConsumerBoundedQueue<TYPE>::popFrontImp(TYPE *value,
                                                                bool  isTry)
{
    Uint64     index = AtomicOp::getUint64Acquire(&d_popIndex);
    const Uint disabledGen =
                            AtomicOp::getUintAcquire(&d_popDisabledGeneration);
 
    if (disabledGen & 1) {
        return e_DISABLED;                                            // RETURN
    }
 
    Node& node = d_popElement_p[index];
 
    Uint nodeState = AtomicOp::getUintAcquire(&node.d_state);
 
    // If the node is not available for reading:
    //   * if this is a "try" invocation, return
    //   * otherwise, yield and check again, then block
    // Note that 'e_WRITABLE_AND_BLOCKED != nodeState' since this is the one
    // consumer.
 
    if (e_WRITABLE_AND_EMPTY == nodeState) {
        if (isTry) {
            return e_EMPTY;                                           // RETURN
        }
 
        bslmt::ThreadUtil::yield();
        nodeState = AtomicOp::getUintAcquire(&node.d_state);
        if (e_WRITABLE_AND_EMPTY == nodeState) {
            bslmt::LockGuard<bslmt::Mutex> guard(&d_popMutex);
 
            nodeState = AtomicOp::testAndSwapUintAcqRel(
                                                       &node.d_state,
                                                       nodeState,
                                                       e_WRITABLE_AND_BLOCKED);
 
            while ((   e_WRITABLE_AND_EMPTY   == nodeState
                    || e_WRITABLE_AND_BLOCKED == nodeState)
                   && disabledGen ==
                          AtomicOp::getUintAcquire(&d_popDisabledGeneration)) {
                int rv = d_popCondition.wait(&d_popMutex);
                if (rv) {
                    AtomicOp::testAndSwapUintAcqRel(&node.d_state,
                                                    e_WRITABLE_AND_BLOCKED,
                                                    e_WRITABLE_AND_EMPTY);
                    return e_FAILED;                                  // RETURN
                }
                nodeState = AtomicOp::getUint(&node.d_state);
            }
 
            // The following checks for disablement being the cause of exiting
            // the 'while' loop.
 
            if (   e_WRITABLE_AND_EMPTY   == nodeState
                || e_WRITABLE_AND_BLOCKED == nodeState) {
                AtomicOp::testAndSwapUintAcqRel(&node.d_state,
                                                e_WRITABLE_AND_BLOCKED,
                                                e_WRITABLE_AND_EMPTY);
                return e_DISABLED;                                    // RETURN
            }
        }
    }
 
    SingleProducerSingleConsumerBoundedQueue_PopCompleteGuard<
                          SingleProducerSingleConsumerBoundedQueue<TYPE>, Node>
                                                     guard(this, &node, index);
 
#if defined(BSLMF_MOVABLEREF_USES_RVALUE_REFERENCES)
    *value = bslmf::MovableRefUtil::move(node.d_value.object());
#else
    *value = node.d_value.object();
#endif
 
    return e_SUCCESS;
}
 
template <class TYPE>
int SingleProducerSingleConsumerBoundedQueue<TYPE>::pushBackImp(
                                                             const TYPE& value,
                                                             bool        isTry)
{
    Uint64     index = AtomicOp::getUint64Acquire(&d_pushIndex);
    const Uint disabledGen =
                           AtomicOp::getUintAcquire(&d_pushDisabledGeneration);
 
    if (disabledGen & 1) {
        return e_DISABLED;                                            // RETURN
    }
 
    Node& node = d_pushElement_p[index];
 
    Uint nodeState = AtomicOp::getUintAcquire(&node.d_state);
 
    // If the node is not available for writing:
    //   * if this is a "try" invocation, return
    //   * otherwise, yield and check again, then block
    // Note that 'e_READABLE_AND_BLOCKED != nodeState' since this is the one
    // producer.
 
    if (e_READABLE == nodeState) {
        if (isTry) {
            return e_FULL;                                            // RETURN
        }
 
        bslmt::ThreadUtil::yield();
        nodeState = AtomicOp::getUintAcquire(&node.d_state);
        if (e_READABLE == nodeState) {
            bslmt::LockGuard<bslmt::Mutex> guard(&d_pushMutex);
 
            nodeState = AtomicOp::testAndSwapUintAcqRel(
                                                       &node.d_state,
                                                       e_READABLE,
                                                       e_READABLE_AND_BLOCKED);
 
            while ((   e_READABLE             == nodeState
                    || e_READABLE_AND_BLOCKED == nodeState)
                   && disabledGen ==
                         AtomicOp::getUintAcquire(&d_pushDisabledGeneration)) {
                int rv = d_pushCondition.wait(&d_pushMutex);
                if (rv) {
                    AtomicOp::testAndSwapUintAcqRel(&node.d_state,
                                                    e_READABLE_AND_BLOCKED,
                                                    e_READABLE);
                    return e_FAILED;                                  // RETURN
                }
                nodeState = AtomicOp::getUint(&node.d_state);
            }
 
            // The following checks for disablement being the cause of exiting
            // the 'while' loop.
 
            if (   e_READABLE             == nodeState
                || e_READABLE_AND_BLOCKED == nodeState) {
                AtomicOp::testAndSwapUintAcqRel(&node.d_state,
                                                e_READABLE_AND_BLOCKED,
                                                e_READABLE);
                return e_DISABLED;                                    // RETURN
            }
        }
    }
 
    bslalg::ScalarPrimitives::copyConstruct(node.d_value.address(),
                                            value,
                                            d_allocator_p);
 
    pushComplete(&node, index);
 
    return e_SUCCESS;
}
 
template <class TYPE>
int SingleProducerSingleConsumerBoundedQueue<TYPE>::pushBackImp(
                                                 bslmf::MovableRef<TYPE> value,
                                                 bool                    isTry)
{
    Uint64     index = AtomicOp::getUint64Acquire(&d_pushIndex);
    const Uint disabledGen =
                           AtomicOp::getUintAcquire(&d_pushDisabledGeneration);
 
    if (disabledGen & 1) {
        return e_DISABLED;                                            // RETURN
    }
 
    Node& node = d_pushElement_p[index];
 
    Uint nodeState = AtomicOp::getUintAcquire(&node.d_state);
 
    if (e_READABLE == nodeState) {
        if (isTry) {
            return e_FULL;                                            // RETURN
        }
 
        bslmt::ThreadUtil::yield();
        nodeState = AtomicOp::getUintAcquire(&node.d_state);
        if (e_READABLE == nodeState) {
            bslmt::LockGuard<bslmt::Mutex> guard(&d_pushMutex);
 
            nodeState = AtomicOp::testAndSwapUintAcqRel(
                                                       &node.d_state,
                                                       e_READABLE,
                                                       e_READABLE_AND_BLOCKED);
 
            while ((   e_READABLE             == nodeState
                    || e_READABLE_AND_BLOCKED == nodeState)
                   && disabledGen ==
                         AtomicOp::getUintAcquire(&d_pushDisabledGeneration)) {
                int rv = d_pushCondition.wait(&d_pushMutex);
                if (rv) {
                    AtomicOp::testAndSwapUintAcqRel(&node.d_state,
                                                    e_READABLE_AND_BLOCKED,
                                                    e_READABLE);
                    return e_FAILED;                                  // RETURN
                }
                nodeState = AtomicOp::getUint(&node.d_state);
            }
 
            if (   e_READABLE             == nodeState
                || e_READABLE_AND_BLOCKED == nodeState) {
                AtomicOp::testAndSwapUintAcqRel(&node.d_state,
                                                e_READABLE_AND_BLOCKED,
                                                e_READABLE);
                return e_DISABLED;                                    // RETURN
            }
        }
    }
 
    TYPE& dummy = value;
    bslalg::ScalarPrimitives::moveConstruct(node.d_value.address(),
                                            dummy,
                                            d_allocator_p);
 
    pushComplete(&node, index);
 
    return e_SUCCESS;
}
 
template <class TYPE>
inline
void SingleProducerSingleConsumerBoundedQueue<TYPE>::pushComplete(
                                                                 Node   *node,
                                                                 Uint64  index)
{
 
    Uint nodeState = AtomicOp::swapUintAcqRel(&node->d_state, e_READABLE);
    if (e_WRITABLE_AND_BLOCKED == nodeState) {
        // Queue is no longer empty and the consumer is blocked.
 
        AtomicOp::addUintAcqRel(&d_emptyGeneration, 1);
        {
            bslmt::LockGuard<bslmt::Mutex> guard(&d_popMutex);
        }
        d_popCondition.signal();
    }
    else if (e_WRITABLE_AND_EMPTY == nodeState) {
        // Queue is no longer empty.
 
        AtomicOp::addUintAcqRel(&d_emptyGeneration, 1);
    }
 
    ++index;
    if (index == d_pushCapacity) {
        index = 0;
    }
    AtomicOp::setUint64Release(&d_pushIndex, index);
}
 
// CREATORS
template <class TYPE>
SingleProducerSingleConsumerBoundedQueue<TYPE>::
     SingleProducerSingleConsumerBoundedQueue(bsl::size_t       capacity,
                                              bslma::Allocator *basicAllocator)
: d_popElement_p(0)
, d_popCapacity(capacity > 0 ? capacity : 1)
, d_popPad()
, d_pushCapacity(capacity > 0 ? capacity : 1)
, d_pushPad()
, d_popMutex()
, d_popCondition()
, d_pushMutex()
, d_pushCondition()
, d_emptyMutex()
, d_emptyCondition()
, d_allocator_p(bslma::Default::allocator(basicAllocator))
{
    AtomicOp::initUint64(&d_popIndex,  0);
    AtomicOp::initUint64(&d_pushIndex, 0);
 
    AtomicOp::initUint(&d_popDisabledGeneration,  0);
    AtomicOp::initUint(&d_emptyCount,             0);
    AtomicOp::initUint(&d_emptyGeneration,        0);
    AtomicOp::initUint(&d_pushDisabledGeneration, 0);
 
    d_popElement_p = static_cast<Node *>(
                        d_allocator_p->allocate(d_popCapacity * sizeof(Node)));
 
    d_pushElement_p = d_popElement_p;
 
    AtomicOp::initUint(&d_popElement_p[0].d_state, e_WRITABLE_AND_EMPTY);
    for (bsl::size_t i = 1; i < d_popCapacity; ++i) {
        AtomicOp::initUint(&d_popElement_p[i].d_state, e_WRITABLE);
    }
}
 
template <class TYPE>
SingleProducerSingleConsumerBoundedQueue<TYPE>
                                  ::~SingleProducerSingleConsumerBoundedQueue()
{
    if (d_popElement_p) {
        removeAll();
        d_allocator_p->deallocate(d_popElement_p);
    }
}
 
// MANIPULATORS
template <class TYPE>
inline
int SingleProducerSingleConsumerBoundedQueue<TYPE>::popFront(TYPE *value)
{
    return popFrontImp(value, false);
}
 
template <class TYPE>
inline
int SingleProducerSingleConsumerBoundedQueue<TYPE>::pushBack(const TYPE& value)
{
    return pushBackImp(value, false);
}
 
template <class TYPE>
inline
int SingleProducerSingleConsumerBoundedQueue<TYPE>::pushBack(
                                                 bslmf::MovableRef<TYPE> value)
{
    return pushBackImp(bslmf::MovableRefUtil::move(value), false);
}
 
template <class TYPE>
void SingleProducerSingleConsumerBoundedQueue<TYPE>::removeAll()
{
    Uint64 index     = AtomicOp::getUint64Acquire(&d_popIndex);
    Uint   nodeState = AtomicOp::getUintAcquire(
                                               &d_popElement_p[index].d_state);
 
    while (e_READABLE == nodeState || e_READABLE_AND_BLOCKED == nodeState) {
        d_popElement_p[index].d_value.object().~TYPE();
 
        AtomicOp::swapUintAcqRel(&d_popElement_p[index].d_state, e_WRITABLE);
 
        ++index;
        if (index == d_popCapacity) {
            index = 0;
        }
 
        nodeState = AtomicOp::getUintAcquire(&d_popElement_p[index].d_state);
    }
 
    // If the node subsequent to the last removed element is writable, the
    // queue is empty and the node subsequent to the last removed element' must
    // be marked as 'e_WRITABLE_AND_EMPTY'.
 
    nodeState = AtomicOp::testAndSwapUintAcqRel(&d_popElement_p[index].d_state,
                                                e_WRITABLE,
                                                e_WRITABLE_AND_EMPTY);
 
    if (e_WRITABLE == nodeState) {
        // The queue is empty, increment the empty generation count.
 
        AtomicOp::addUintAcqRel(&d_emptyGeneration, 1);
    }
    else {
        // A 'removeAll' makes the queue empty.  Since there has been a
        // 'pushBack' before the queue could be marked empty ('e_WRITABLE !=
        // nodeState'), increase the empty generation by 2 to note the queue
        // was empty at some point during this method call but is no longer
        // empty (1 for becoming empty, 1 for leaving the empty state).
 
        AtomicOp::addUintAcqRel(&d_emptyGeneration, 2);
    }
 
    AtomicOp::setUint64Release(&d_popIndex, index);
 
    {
        bslmt::LockGuard<bslmt::Mutex> guard(&d_pushMutex);
    }
    d_pushCondition.signal();
 
    if (0 < AtomicOp::getUintAcquire(&d_emptyCount)) {
        {
            bslmt::LockGuard<bslmt::Mutex> guard(&d_emptyMutex);
        }
        d_emptyCondition.broadcast();
    }
}
 
template <class TYPE>
inline
int SingleProducerSingleConsumerBoundedQueue<TYPE>::tryPopFront(TYPE *value)
{
    return popFrontImp(value, true);
}
 
template <class TYPE>
inline
int SingleProducerSingleConsumerBoundedQueue<TYPE>::tryPushBack(
                                                             const TYPE& value)
{
    return pushBackImp(value, true);
}
 
template <class TYPE>
inline
int SingleProducerSingleConsumerBoundedQueue<TYPE>::tryPushBack(
                                                 bslmf::MovableRef<TYPE> value)
{
    return pushBackImp(bslmf::MovableRefUtil::move(value), true);
}
 
                       // Enqueue/Dequeue State
 
template <class TYPE>
inline
void SingleProducerSingleConsumerBoundedQueue<TYPE>::disablePopFront()
{
    incrementUntil(&d_popDisabledGeneration, 1);
 
    {
        bslmt::LockGuard<bslmt::Mutex> guard(&d_popMutex);
    }
    d_popCondition.broadcast();
 
    if (0 < AtomicOp::getUintAcquire(&d_emptyCount)) {
        {
            bslmt::LockGuard<bslmt::Mutex> guard(&d_emptyMutex);
        }
        d_emptyCondition.broadcast();
    }
}
 
template <class TYPE>
inline
void SingleProducerSingleConsumerBoundedQueue<TYPE>::disablePushBack()
{
    incrementUntil(&d_pushDisabledGeneration, 1);
 
    {
        bslmt::LockGuard<bslmt::Mutex> guard(&d_pushMutex);
    }
    d_pushCondition.broadcast();
}
 
template <class TYPE>
inline
void SingleProducerSingleConsumerBoundedQueue<TYPE>::enablePopFront()
{
    incrementUntil(&d_popDisabledGeneration, 0);
}
 
template <class TYPE>
inline
void SingleProducerSingleConsumerBoundedQueue<TYPE>::enablePushBack()
{
    incrementUntil(&d_pushDisabledGeneration, 0);
}
 
// ACCESSORS
template <class TYPE>
inline
bsl::size_t SingleProducerSingleConsumerBoundedQueue<TYPE>::capacity() const
{
    return d_popCapacity;
}
 
template <class TYPE>
inline
bool SingleProducerSingleConsumerBoundedQueue<TYPE>::isEmpty() const
{
    return 0 == (AtomicOp::getUintAcquire(&d_emptyGeneration) & 1);
}
 
template <class TYPE>
inline
bool SingleProducerSingleConsumerBoundedQueue<TYPE>::isFull() const
{
    Node& node      = d_pushElement_p[AtomicOp::getUint64Acquire(
                                                                &d_pushIndex)];
    Uint  nodeState = AtomicOp::getUintAcquire(&node.d_state);
 
    return e_READABLE == nodeState || e_READABLE_AND_BLOCKED == nodeState;
}
 
template <class TYPE>
inline
bool SingleProducerSingleConsumerBoundedQueue<TYPE>::isPopFrontDisabled() const
{
    return 1 == (AtomicOp::getUintAcquire(&d_popDisabledGeneration) & 1);
}
 
template <class TYPE>
inline
bool SingleProducerSingleConsumerBoundedQueue<TYPE>::isPushBackDisabled() const
{
    return 1 == (AtomicOp::getUintAcquire(&d_pushDisabledGeneration) & 1);
}
 
template <class TYPE>
inline
bsl::size_t SingleProducerSingleConsumerBoundedQueue<TYPE>::numElements() const
{
    Uint64 popIndex  = AtomicOp::getUint64Acquire(&d_popIndex);
    Uint64 pushIndex = AtomicOp::getUint64Acquire(&d_pushIndex);
    Node&  node      = d_pushElement_p[pushIndex];
    Uint   nodeState = AtomicOp::getUintAcquire(&node.d_state);
 
    if (e_READABLE == nodeState || e_READABLE_AND_BLOCKED == nodeState) {
        return d_popCapacity;                                         // RETURN
    }
 
    return static_cast<bsl::size_t>(  pushIndex >= popIndex
                                    ? pushIndex - popIndex
                                    : pushIndex + d_popCapacity - popIndex);
}
 
template <class TYPE>
int SingleProducerSingleConsumerBoundedQueue<TYPE>::waitUntilEmpty() const
{
    AtomicOp::addUintAcqRel(&d_emptyCount, 1);
 
    const Uint initEmptyGen = AtomicOp::getUintAcquire(&d_emptyGeneration);
 
    const Uint disabledGen =
                            AtomicOp::getUintAcquire(&d_popDisabledGeneration);
 
    if (disabledGen & 1) {
        AtomicOp::addUintAcqRel(&d_emptyCount, -1);
        return e_DISABLED;                                            // RETURN
    }
 
    if (0 == (initEmptyGen & 1)) {
        AtomicOp::addUintAcqRel(&d_emptyCount, -1);
        return e_SUCCESS;                                             // RETURN
    }
 
    bslmt::LockGuard<bslmt::Mutex> guard(&d_emptyMutex);
 
    Uint emptyGen = AtomicOp::getUintAcquire(&d_emptyGeneration);
 
    while (   initEmptyGen == emptyGen
           && disabledGen  ==
                          AtomicOp::getUintAcquire(&d_popDisabledGeneration)) {
        int rv = d_emptyCondition.wait(&d_emptyMutex);
        if (rv) {
            AtomicOp::addUintAcqRel(&d_emptyCount, -1);
            return e_FAILED;                                          // RETURN
        }
        emptyGen = AtomicOp::getUintAcquire(&d_emptyGeneration);
    }
 
    AtomicOp::addUintAcqRel(&d_emptyCount, -1);
 
    if (initEmptyGen == emptyGen) {
        return e_DISABLED;                                            // RETURN
    }
 
    return e_SUCCESS;
}
 
                                  // Aspects
 
template <class TYPE>
inline
bslma::Allocator *SingleProducerSingleConsumerBoundedQueue<TYPE>::allocator()
                                                                          const
{
    return d_allocator_p;
}
 
}  // close package namespace
 
 
#endif
 
// ----------------------------------------------------------------------------
// Copyright 2019 Bloomberg Finance L.P.
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//     http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
// ----------------------------- END-OF-FILE ----------------------------------
 
/** @} */
/** @} */
/** @} */