bdlcc_singleconsumerqueueimpl.h

Go to the documentation of this file.

/// @file bdlcc_singleconsumerqueueimpl.h
///
/// The content of this file has been pre-processed for Doxygen.
///
 
 
// bdlcc_singleconsumerqueueimpl.h                                    -*-C++-*-
 
#ifndef INCLUDED_BDLCC_SINGLECONSUMERQUEUEIMPL
#define INCLUDED_BDLCC_SINGLECONSUMERQUEUEIMPL
 
#include <bsls_ident.h>
BSLS_IDENT("$Id: $")
 
/// @defgroup bdlcc_singleconsumerqueueimpl bdlcc_singleconsumerqueueimpl
/// @brief  Provide a testable thread-aware single consumer queue of values.
/// @addtogroup bdl
/// @{
/// @addtogroup bdlcc
/// @{
/// @addtogroup bdlcc_singleconsumerqueueimpl
/// @{
///
/// <h1> Outline </h1>
/// * <a href="#bdlcc_singleconsumerqueueimpl-purpose"> Purpose</a>
/// * <a href="#bdlcc_singleconsumerqueueimpl-classes"> Classes </a>
/// * <a href="#bdlcc_singleconsumerqueueimpl-description"> Description </a>
///   * <a href="#bdlcc_singleconsumerqueueimpl-allocator-requirements"> Allocator Requirements </a>
///   * <a href="#bdlcc_singleconsumerqueueimpl-exception-safety"> Exception Safety </a>
///   * <a href="#bdlcc_singleconsumerqueueimpl-move-semantics-in-c-03"> Move Semantics in C++03 </a>
///   * <a href="#bdlcc_singleconsumerqueueimpl-memory-usage"> Memory Usage </a>
///   * <a href="#bdlcc_singleconsumerqueueimpl-warning-synchronization-required-on-destruction"> WARNING: Synchronization Required on Destruction </a>
///   * <a href="#bdlcc_singleconsumerqueueimpl-usage"> Usage </a>
///
/// # Purpose {#bdlcc_singleconsumerqueueimpl-purpose}
/// Provide a testable thread-aware single consumer queue of values.
///
/// # Classes {#bdlcc_singleconsumerqueueimpl-classes}
///
/// -  bdlcc::SingleConsumerQueueImpl: thread-aware single consumer `TYPE` queue
///
/// # Description {#bdlcc_singleconsumerqueueimpl-description}
/// This component defines a type,
/// `bdlcc::SingleConsumerQueueImpl`, that provides an efficient, thread-aware
/// queue of values assuming a single consumer (the use of `popFront`,
/// `tryPopFront`, and `removeAll` is done by one thread or a group of threads
/// using external synchronization).  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 there is only one consumer thread.
///
/// The queue provides `pushBack` and `popFront` methods for pushing data into
/// the queue and popping data from the queue.  The queue will allocate memory
/// as necessary to accommodate `pushBack` invocations (`pushBack` will never
/// block and is provided for consistency with other containers).  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 `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.  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` and
/// `tryPopFront` fail immediately and return an error code.  Any threads
/// blocked in `popFront` when the queue is dequeue disabled return from
/// `popFront` immediately and return an error code.
///
/// ## Allocator Requirements {#bdlcc_singleconsumerqueueimpl-allocator-requirements}
///
///
/// Access to the allocator supplied to the constructor is internally
/// synchronized by this component.  If allocations performed by this component
/// must be synchronized with external allocations (performed outside of this
/// component), that synchronization must be guaranteed by the user.  Using a
/// thread-safe allocator is the common way to satisfy this requirement.
///
/// ## Exception Safety {#bdlcc_singleconsumerqueueimpl-exception-safety}
///
///
/// A `bdlcc::SingleConsumerQueueImpl` is exception neutral, and all of the
/// methods of `bdlcc::SingleConsumerQueueImpl` provide the basic exception
/// safety guarantee (see @ref bsldoc_glossary ).
///
/// ## Move Semantics in C++03 {#bdlcc_singleconsumerqueueimpl-move-semantics-in-c-03}
///
///
/// Move-only types are supported by `bdlcc::SingleConsumerQueueImpl` 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.
///
/// ## Memory Usage {#bdlcc_singleconsumerqueueimpl-memory-usage}
///
///
/// `bdlcc::SingleConsumerQueueImpl` is most efficient when dealing with small
/// objects or fundamental types (as a thread-safe container, its methods pass
/// objects *by* *value*).  We recommend large objects be stored as
/// shared-pointers (or possibly raw pointers).
///
/// ## WARNING: Synchronization Required on Destruction {#bdlcc_singleconsumerqueueimpl-warning-synchronization-required-on-destruction}
///
///
/// The behavior for the destructor is undefined unless all access or
/// modification of the object is completed prior to its destruction.  Some form
/// of synchronization, external to the component, is required to ensure the
/// precondition on the destructor is met.  For example, if two (or more)
/// threads are manipulating a queue, it is *not* safe to anticipate the number
/// of elements added to the queue, and destroy that queue immediately after the
/// last element is popped (without additional synchronization) because one of
/// the corresponding push functions may not have completed (push may, for
/// instance, signal waiting threads after the element is considered added to
/// the container).
///
/// ## Usage {#bdlcc_singleconsumerqueueimpl-usage}
///
///
/// There is no usage example for this component since it is not meant for
/// direct client use.
/// @}
/** @} */
/** @} */
 
/** @addtogroup bdl
 * @{
 */
/** @addtogroup bdlcc
 * @{
 */
/** @addtogroup bdlcc_singleconsumerqueueimpl
 * @{
 */
 
#include <bdlscm_version.h>
 
#include <bdlma_infrequentdeleteblocklist.h>
 
#include <bslalg_scalarprimitives.h>
 
#include <bslma_usesbslmaallocator.h>
 
#include <bslmf_movableref.h>
#include <bslmf_nestedtraitdeclaration.h>
 
#include <bslmt_lockguard.h>
#include <bslmt_threadutil.h>
 
#include <bsls_assert.h>
#include <bsls_objectbuffer.h>
#include <bsls_types.h>
 
#include <bsl_cstddef.h>
 
 
namespace bdlcc {
 
             // ================================================
             // class SingleConsumerQueueImpl_MarkReclaimProctor
             // ================================================
 
/// This class implements a proctor that, unless its `release` method has
/// previously been invoked, automatically invokes `markReclaim` on a `NODE`
/// upon destruction.
///
/// See @ref bdlcc_singleconsumerqueueimpl 
template <class TYPE, class NODE>
class SingleConsumerQueueImpl_MarkReclaimProctor {
 
    // DATA
    TYPE *d_queue_p;  // managed queue owning the managed node
    NODE *d_node_p;   // managed node
 
  private:
    // NOT IMPLEMENTED
    SingleConsumerQueueImpl_MarkReclaimProctor();
    SingleConsumerQueueImpl_MarkReclaimProctor(
                            const SingleConsumerQueueImpl_MarkReclaimProctor&);
    SingleConsumerQueueImpl_MarkReclaimProctor& operator=(
                            const SingleConsumerQueueImpl_MarkReclaimProctor&);
 
  public:
    // CREATORS
 
    /// Create a `markReclaim` proctor managing the specified `node` of the
    /// specified `queue`.
    SingleConsumerQueueImpl_MarkReclaimProctor(TYPE *queue, NODE *node);
 
    /// Destroy this object and, if `release` has not been invoked, invoke
    /// the managed queue's `markReclaim` method with the managed node.
    ~SingleConsumerQueueImpl_MarkReclaimProctor();
 
    // MANIPULATORS
 
    /// Release from management the queue and node currently managed by this
    /// proctor.  If no queue, this method has no effect.
    void release();
};
 
              // ==============================================
              // class SingleConsumerQueueImpl_PopCompleteGuard
              // ==============================================
 
/// This class implements a guard that automatically invokes `popComplete`
/// on the managed queue upon destruction.
///
/// See @ref bdlcc_singleconsumerqueueimpl 
template <class TYPE>
class SingleConsumerQueueImpl_PopCompleteGuard {
 
    // DATA
    TYPE *d_queue_p;  // managed queue
 
  private:
    // NOT IMPLEMENTED
    SingleConsumerQueueImpl_PopCompleteGuard();
    SingleConsumerQueueImpl_PopCompleteGuard(
                              const SingleConsumerQueueImpl_PopCompleteGuard&);
    SingleConsumerQueueImpl_PopCompleteGuard& operator=(
                              const SingleConsumerQueueImpl_PopCompleteGuard&);
 
  public:
    // CREATORS
 
    /// Create a `popComplete` guard managing the specified `queue`.
    explicit
    SingleConsumerQueueImpl_PopCompleteGuard(TYPE *queue);
 
    /// Destroy this object and invoke the `popComplete` method on the
    /// managed queue.
    ~SingleConsumerQueueImpl_PopCompleteGuard();
};
 
             // ===============================================
             // class SingleConsumerQueueImpl_AllocateLockGuard
             // ===============================================
 
/// This class implements a guard that automatically invokes
/// `releaseAllocateLock` on the managed queue upon destruction.
///
/// See @ref bdlcc_singleconsumerqueueimpl 
template <class TYPE>
class SingleConsumerQueueImpl_AllocateLockGuard {
 
    // DATA
    TYPE *d_queue_p;  // managed queue
 
  private:
    // NOT IMPLEMENTED
    SingleConsumerQueueImpl_AllocateLockGuard();
    SingleConsumerQueueImpl_AllocateLockGuard(
                             const SingleConsumerQueueImpl_AllocateLockGuard&);
    SingleConsumerQueueImpl_AllocateLockGuard& operator=(
                             const SingleConsumerQueueImpl_AllocateLockGuard&);
 
  public:
    // CREATORS
 
    /// Create a `releaseAllocateLock` guard managing the specified `queue`.
    explicit SingleConsumerQueueImpl_AllocateLockGuard(TYPE *queue);
 
    /// Destroy this object and invoke the managed queue's
    ~SingleConsumerQueueImpl_AllocateLockGuard();
       // 'releaseAllocateLock' method.
};
 
                      // =============================
                      // class SingleConsumerQueueImpl
                      // =============================
 
/// This class provides a thread-safe unbounded queue of values that assumes
/// a single consumer thread.
///
/// The types `ATOMIC_OP`, `MUTEX`, and `CONDITION` are exposed for testing.
/// Typical usage is with `bsls::AtomicOperations` for `ATOMIC_OP`,
/// `bslmt::Mutex` for `MUTEX`, and `bslmt::Condition` for `CONDITION`.
///
/// See @ref bdlcc_singleconsumerqueueimpl 
template <class TYPE, class ATOMIC_OP, class MUTEX, class CONDITION>
class SingleConsumerQueueImpl {
 
    // 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 producers
        // 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_WRITABLE,              // node can be written
        e_WRITABLE_AND_BLOCKED,  // node can be written and has blocked reader
        e_RECLAIM                // node suffered exception while being written
    };
 
    static const bsl::size_t k_ALLOCATION_BATCH_SIZE = 8;
                                 // number of nodes to allocate during a
                                 // 'pushBack' when no nodes are available for
                                 // reuse
 
    // The queus's state is maintained in 'd_state' whose bits have the
    // following meaning (and can be maintained by the constants below):
    //..
    //  63          19 18   0
    // +-----------+--+-----+
    // | available |a | use |
    // +-----------+--+-----+
    //..
    //
    //: * available:    the number of nodes available for use, which becomes
    //:                 negative when an allocation is needed
    //:
    //: * a (allocate): a bit indicating a thread is holding the allocation
    //:                 lock
    //:
    //: * use:          number of threads attempting to use existing nodes
    //
    // The 'k_*_MASK' constants define the layout of the attributes, the
    // 'k_*_INC' constants are used to modify the 'd_state' attributes, and the
    // 'k_*_SHIFT' constants allow recovery of the stored value.
    //
    // See *Implementation* *Note* for further details.
 
    static const bsls::Types::Int64 k_ALLOCATE_MASK     = 0x0000000000080000LL;
    static const bsls::Types::Int64 k_ALLOCATE_INC      = 0x0000000000080000LL;
    static const bsls::Types::Int64 k_USE_MASK          = 0x000000000007ffffLL;
    static const bsls::Types::Int64 k_USE_INC           = 0x0000000000000001LL;
    static const bsls::Types::Int64 k_AVAILABLE_MASK    = 0xfffffffffff00000LL;
    static const bsls::Types::Int64 k_AVAILABLE_INC     = 0x0000000000100000LL;
    static const int                k_AVAILABLE_SHIFT   = 20;
 
    // PRIVATE TYPES
    typedef typename ATOMIC_OP::AtomicTypes::Int     AtomicInt;
    typedef typename ATOMIC_OP::AtomicTypes::Int64   AtomicInt64;
    typedef typename ATOMIC_OP::AtomicTypes::Uint    AtomicUint;
    typedef typename ATOMIC_OP::AtomicTypes::Pointer AtomicPointer;
 
    template <class T>
    struct QueueNode {
        // PUBLIC DATA
        bsls::ObjectBuffer<T> d_value;  // stored value
        AtomicInt             d_state;  // 'e_READABLE', 'e_WRITABLE', etc.
        AtomicPointer         d_next;   // pointer to next node
    };
 
    typedef QueueNode<TYPE>                  Node;
    typedef bdlma::InfrequentDeleteBlockList Allocator;
 
    // DATA
    AtomicPointer     d_nextWrite;         // pointer to next write to node
 
    AtomicPointer     d_nextRead;          // pointer to next read from node
 
    MUTEX             d_readMutex;         // used with 'd_readCondition' to
                                           // block until an element is
                                           // available for popping
 
    CONDITION         d_readCondition;     // condition variable for popping
                                           // thread
 
    MUTEX             d_writeMutex;        // during allocation, used to
                                           // synchronize threads access to
                                           // 'd_nextWrite'
 
    AtomicInt64       d_capacity;          // capacity of this queue
 
    mutable MUTEX     d_emptyMutex;        // blocking point for consumer
                                           // during 'waitUntilEmpty'
 
    mutable CONDITION d_emptyCondition;    // condition variable for consumer
                                           // during 'waitUntilEmpty'
 
    AtomicInt64       d_state;             // bit pattern representing the
                                           // state of the queue (see
                                           // implementation notes)
 
    AtomicUint        d_popFrontDisabled;  // is queue pop disabled and
                                           // generation count; see
                                           // *Implementation* *Note*
 
    AtomicUint        d_pushBackDisabled;  // is queue push disabled and
                                           // generation count; see
                                           // *Implementation* *Note*
 
    Allocator         d_allocator;         // allocator
 
    // FRIENDS
    friend class SingleConsumerQueueImpl_MarkReclaimProctor<
                           SingleConsumerQueueImpl<TYPE,
                                                   ATOMIC_OP,
                                                   MUTEX,
                                                   CONDITION>,
                           typename SingleConsumerQueueImpl<TYPE,
                                                            ATOMIC_OP,
                                                            MUTEX,
                                                            CONDITION>::Node >;
 
    friend class SingleConsumerQueueImpl_PopCompleteGuard<
                                          SingleConsumerQueueImpl<TYPE,
                                                                  ATOMIC_OP,
                                                                  MUTEX,
                                                                  CONDITION> >;
 
    friend class SingleConsumerQueueImpl_AllocateLockGuard<
                                          SingleConsumerQueueImpl<TYPE,
                                                                  ATOMIC_OP,
                                                                  MUTEX,
                                                                  CONDITION> >;
 
    // PRIVATE CLASS METHODS
 
    /// Return the available attribute from the specified `state`.
    static bsls::Types::Int64 available(bsls::Types::Int64 state);
 
    // PRIVATE MANIPULATORS
 
    /// 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 Note that   this method is used to modify the generation counts
    /// stored in `d_popFrontDisabled` and `d_pushBackDisabled`.  See
    /// *Implementation* *Note* for further details.
    void incrementUntil(AtomicUint *value, unsigned int bitValue);
 
    /// Mark the specified `node` as a node to be reclaimed.
    void markReclaim(Node *node);
 
    /// If the specified `destruct` is true, destruct the value stored in
    /// `d_nextRead`.  Mark `d_nextRead` writable, and if the queue is empty
    /// then signal the queue empty condition.  This method is used to
    /// complete the reclamation of a node in the presence of an exception.
    void popComplete(bool destruct);
 
    /// Return a pointer to the node to assign the value being pushed into
    /// this queue, or 0 if `isPushBackDisabled()`.
    Node *pushBackHelper();
 
    /// Remove the allocation lock indicator from `d_state`.  This method is
    /// intended to be used to remove the allocation lock indicator from
    /// `d_state` when there is an exception during allocation and the
    /// locked state is set (i.e., `pushBackHelper`).
    void releaseAllocateLock();
 
  private:
    // NOT IMPLEMENTED
    SingleConsumerQueueImpl(const SingleConsumerQueueImpl&);
    SingleConsumerQueueImpl& operator=(const SingleConsumerQueueImpl&);
 
  public:
    // TRAITS
    BSLMF_NESTED_TRAIT_DECLARATION(SingleConsumerQueueImpl,
                                   bslma::UsesBslmaAllocator);
 
    // PUBLIC TYPES
    typedef TYPE value_type;  // The type for elements.
 
    // PUBLIC CONSTANTS
    enum {
        e_SUCCESS  =  0,  // must be 0
        e_EMPTY    = -1,
        e_DISABLED = -2
    };
 
    // CREATORS
 
    /// Create a thread-aware queue.  Optionally specify a `basicAllocator`
    /// used to supply memory.  If `basicAllocator` is 0, the currently
    /// installed default allocator is used.
    explicit
    SingleConsumerQueueImpl(bslma::Allocator *basicAllocator = 0);
 
    /// 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
    SingleConsumerQueueImpl(bsl::size_t       capacity,
                            bslma::Allocator *basicAllocator = 0);
 
    /// Destroy this container.  
    /// \pre The behavior is undefined  unless all access
    /// or modification of the container has completed prior to this call.
    ~SingleConsumerQueueImpl();
 
    // 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_DISABLED` if
    /// `isPopFrontDisabled()`.  On failure, `value` is not changed.
    /// Threads blocked due to the queue being empty will return
    /// `e_DISABLED` if `disablePopFront` is invoked.  
    /// 
    /// \pre 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_DISABLED` if `isPushBackDisabled()`.
    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_DISABLED` if `isPushBackDisabled()`.  On failure, `value` is not
    /// changed.
    int pushBack(bslmf::MovableRef<TYPE> value);
 
    /// Remove all items currently in this queue.  
    /// \note 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.  
    /// 
    /// \pre 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_DISABLED` if
    /// `isPopFrontDisabled()`, and `e_EMPTY` if `!isPopFrontDisabled()` and
    /// the queue was empty.  On failure, `value` is not changed.  
    /// 
    /// \pre 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, retun
    /// `e_DISABLED` if `isPushBackDisabled()`.
    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_DISABLED` if `isPushBackDisabled()`.  On failure, `value` is not
    /// changed.
    int tryPushBack(bslmf::MovableRef<TYPE> value);
 
                       // Enqueue/Dequeue State
 
    /// Disable dequeueing from this queue.  All subsequent invocations of
    /// `popFront` or `tryPopFront` will fail immediately.  All blocked
    /// invocations of `popFront` and `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.  All blocked
    /// invocations of `pushBack` will fail immediately.  If the queue is
    /// already enqueue disabled, this method has no effect.
    void disablePushBack();
 
    /// Enable dequeueing.  If the queue is not dequeue disabled, this call
    /// has no effect.
    void enablePopFront();
 
    /// Enable queuing.  If the queue is not enqueue disabled, this call has
    /// no effect.
    void enablePushBack();
 
    // ACCESSORS
 
    /// 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.  
    /// 
    /// \note Note that   for unbounded queues, this method
    /// always returns `false`.
    bool isFull() const;
 
    /// Return `true` if this queue is dequeue disabled, and `false` otherwise.  
    /// 
    /// \note 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 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_DISABLED` if `!isEmpty() && isPopFrontDisabled()`.  A blocked
    /// thread waiting for the queue to empty will return `e_DISABLED` 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 SingleConsumerQueueImpl_MarkReclaimProctor
             // ------------------------------------------------
 
// CREATORS
template <class TYPE, class NODE>
SingleConsumerQueueImpl_MarkReclaimProctor<TYPE, NODE>::
            SingleConsumerQueueImpl_MarkReclaimProctor(TYPE *queue, NODE *node)
: d_queue_p(queue)
, d_node_p(node)
{
}
 
template <class TYPE, class NODE>
SingleConsumerQueueImpl_MarkReclaimProctor<TYPE, NODE>::
                                  ~SingleConsumerQueueImpl_MarkReclaimProctor()
{
    if (d_queue_p) {
        d_queue_p->markReclaim(d_node_p);
    }
}
 
// MANIPULATORS
template <class TYPE, class NODE>
void SingleConsumerQueueImpl_MarkReclaimProctor<TYPE, NODE>::release()
{
    d_queue_p = 0;
}
 
              // ----------------------------------------------
              // class SingleConsumerQueueImpl_PopCompleteGuard
              // ----------------------------------------------
 
// CREATORS
template <class TYPE>
SingleConsumerQueueImpl_PopCompleteGuard<TYPE>::
                          SingleConsumerQueueImpl_PopCompleteGuard(TYPE *queue)
: d_queue_p(queue)
{
}
 
template <class TYPE>
SingleConsumerQueueImpl_PopCompleteGuard<TYPE>::
                                    ~SingleConsumerQueueImpl_PopCompleteGuard()
{
    d_queue_p->popComplete(true);
}
 
          // ------------------------------------------------------
          // class SingleConsumerQueueImpl_AllocateLockGuardProctor
          // ------------------------------------------------------
 
// CREATORS
template <class TYPE>
SingleConsumerQueueImpl_AllocateLockGuard<TYPE>::
                         SingleConsumerQueueImpl_AllocateLockGuard(TYPE *queue)
: d_queue_p(queue)
{
}
 
template <class TYPE>
SingleConsumerQueueImpl_AllocateLockGuard<TYPE>::
                                   ~SingleConsumerQueueImpl_AllocateLockGuard()
{
    d_queue_p->releaseAllocateLock();
}
 
                      // -----------------------------
                      // class SingleConsumerQueueImpl
                      // -----------------------------
 
// PRIVATE CLASS METHODS
template <class TYPE, class ATOMIC_OP, class MUTEX, class CONDITION>
bsls::Types::Int64 SingleConsumerQueueImpl<TYPE, ATOMIC_OP, MUTEX, CONDITION>
                                          ::available(bsls::Types::Int64 state)
{
    return state >> k_AVAILABLE_SHIFT;
}
 
// PRIVATE MANIPULATORS
template <class TYPE, class ATOMIC_OP, class MUTEX, class CONDITION>
void SingleConsumerQueueImpl<TYPE, ATOMIC_OP, MUTEX, CONDITION>
                     ::incrementUntil(AtomicUint *value, unsigned int bitValue)
{
    unsigned int state = ATOMIC_OP::getUintAcquire(value);
    if (bitValue != (state & 1)) {
        unsigned int expState;
        do {
            expState = state;
            state = ATOMIC_OP::testAndSwapUintAcqRel(value,
                                                     state,
                                                     state + 1);
        } while (state != expState && (bitValue != (state & 1)));
    }
}
 
template <class TYPE, class ATOMIC_OP, class MUTEX, class CONDITION>
void SingleConsumerQueueImpl<TYPE, ATOMIC_OP, MUTEX, CONDITION>
                                                      ::markReclaim(Node *node)
{
    // A reclaimed node is used to denote an exception occurred and the node is
    // currently invalid.  A reclaimed node is considered removed and not
    // counted as part of the capacity of the queue (thus ensuring
    // 'numElements' is correct).
 
    ATOMIC_OP::addInt64AcqRel(&d_capacity, -1);
 
    int nodeState = ATOMIC_OP::swapIntAcqRel(&node->d_state, e_RECLAIM);
    if (e_WRITABLE_AND_BLOCKED == nodeState) {
        {
            bslmt::LockGuard<MUTEX> guard(&d_readMutex);
        }
        d_readCondition.signal();
    }
}
 
template <class TYPE, class ATOMIC_OP, class MUTEX, class CONDITION>
void SingleConsumerQueueImpl<TYPE, ATOMIC_OP, MUTEX, CONDITION>
                                                   ::popComplete(bool destruct)
{
    Node *nextRead =
                    static_cast<Node *>(ATOMIC_OP::getPtrAcquire(&d_nextRead));
 
    if (destruct) {
        nextRead->d_value.object().~TYPE();
    }
 
    ATOMIC_OP::setIntRelease(&nextRead->d_state, e_WRITABLE);
 
    ATOMIC_OP::setPtrRelease(&d_nextRead,
                             ATOMIC_OP::getPtrAcquire(&nextRead->d_next));
 
    bsls::Types::Int64 state = ATOMIC_OP::addInt64NvAcqRel(&d_state,
                                                           k_AVAILABLE_INC);
 
    if (ATOMIC_OP::getInt64Acquire(&d_capacity) == available(state)) {
        {
            bslmt::LockGuard<MUTEX> guard(&d_emptyMutex);
        }
        d_emptyCondition.broadcast();
    }
}
 
template <class TYPE, class ATOMIC_OP, class MUTEX, class CONDITION>
typename SingleConsumerQueueImpl<TYPE, ATOMIC_OP, MUTEX, CONDITION>::Node *
                     SingleConsumerQueueImpl<TYPE, ATOMIC_OP, MUTEX, CONDITION>
                                                             ::pushBackHelper()
{
    if (1 == (ATOMIC_OP::getUintAcquire(&d_pushBackDisabled) & 1)) {
        return 0;                                                     // RETURN
    }
 
    // Fast path requires an available node ('-k_AVAILABLE_INC') and needs to
    // indicate a thread is intending to use an existing node ('k_USE_INC').
 
    bsls::Types::Int64 state = ATOMIC_OP::addInt64NvAcqRel(
                                                  &d_state,
                                                  k_USE_INC - k_AVAILABLE_INC);
 
    if (state < 0 || 0 != (state & k_ALLOCATE_MASK)) {
        // The determination to use an existing node was premature, undo the
        // indication.
 
        state = ATOMIC_OP::addInt64NvAcqRel(&d_state,
                                            k_AVAILABLE_INC - k_USE_INC);
 
        bsls::Types::Int64 expState;
 
        do {
            expState = state;
            if (state >= k_AVAILABLE_INC && 0 == (state & k_ALLOCATE_MASK)) {
                // The are now sufficient available nodes to reserve one.  This
                // can be due to an allocation completing or a node being made
                // available by the consumer.
 
                state = ATOMIC_OP::testAndSwapInt64AcqRel(
                                          &d_state,
                                          state,
                                          state + k_USE_INC - k_AVAILABLE_INC);
            }
            else if (   0 == (  (state >> k_AVAILABLE_SHIFT)
                              + (state & k_USE_MASK))
                     && 0 == (state & k_ALLOCATE_MASK)) {
                // '-AVAILABLE == USE' indicates all threads will wait for the
                // allocation, so attempt to become the allocating thread.
                // Note that 'AVAILABLE < 0 && -AVAILABLE != USE' indicates the
                // temporary state where threads are still completing the
                // re-use of an existing node or there are available nodes to
                // be re-used (nodes were allocated or made available by the
                // consumer).  If threads are still completing the re-use of an
                // existing node, ownership of 'd_nextWrite' can not be
                // guaranteed by setting the allocation lock.
 
                state = ATOMIC_OP::testAndSwapInt64AcqRel(
                                                       &d_state,
                                                       state,
                                                       state + k_ALLOCATE_INC);
                if (expState == state) {
                    // This thread is the only thread acccessing 'd_nextWrite'.
                    // Allocate new nodes and insert them.  The variables 'a'
                    // and 'b' in the below are pointers to 'Node' as per the
                    // following diagram.
                    //..
                    //  d_nextWrite
                    //       |
                    //       V
                    //     +---+     +---+
                    // --> |   | --> |   | -->
                    //     +---+     +---+
                    //       ^         ^
                    //       |         |
                    //       a         b
                    //..
 
                    SingleConsumerQueueImpl_AllocateLockGuard<
                              SingleConsumerQueueImpl<TYPE,
                                                      ATOMIC_OP,
                                                      MUTEX,
                                                      CONDITION> > guard(this);
 
                    Node *a = static_cast<Node *>(
                                       ATOMIC_OP::getPtrAcquire(&d_nextWrite));
 
                    Node *b = static_cast<Node *>(
                                         ATOMIC_OP::getPtrAcquire(&a->d_next));
 
                    Node *nodes = static_cast<Node *>(
                              d_allocator.allocate(  sizeof(Node)
                                                   * k_ALLOCATION_BATCH_SIZE));
                    for (bsl::size_t i = 0;
                         i < k_ALLOCATION_BATCH_SIZE - 1;
                         ++i) {
                        Node *n = nodes + i;
                        ATOMIC_OP::initInt(&n->d_state, e_WRITABLE);
                        ATOMIC_OP::initPointer(&n->d_next, n + 1);
                    }
                    {
                        Node *n = nodes + k_ALLOCATION_BATCH_SIZE - 1;
                        ATOMIC_OP::initInt(&n->d_state, e_WRITABLE);
                        ATOMIC_OP::initPointer(&n->d_next, b);
                    }
 
                    ATOMIC_OP::setPtrRelease(&a->d_next,   nodes);
                    ATOMIC_OP::setPtrRelease(&d_nextWrite, nodes);
 
                    ATOMIC_OP::addInt64AcqRel(&d_capacity,
                                              k_ALLOCATION_BATCH_SIZE);
 
                    // Reserve one node for this thread and make the other
                    // nodes available to other threads.
 
                    ATOMIC_OP::addInt64AcqRel(
                              &d_state,
                              k_AVAILABLE_INC * (k_ALLOCATION_BATCH_SIZE - 1));
 
                    return a;                                         // RETURN
                }
 
                expState = ~state;  // cause the 'while' to fail and remain in
                                    // this loop
            }
            else {
                bslmt::ThreadUtil::yield();
                state = ATOMIC_OP::getInt64Acquire(&d_state);
 
                expState = ~state;  // cause the 'while' to fail and remain in
                                    // this loop
            }
        } while (state != expState);
    }
 
    Node *nextWrite =
                   static_cast<Node *>(ATOMIC_OP::getPtrAcquire(&d_nextWrite));
 
    Node *expNextWrite;
    do {
        expNextWrite = nextWrite;
        Node *next = static_cast<Node *>(ATOMIC_OP::getPtrAcquire(
                                                          &nextWrite->d_next));
 
        nextWrite = static_cast<Node *>(ATOMIC_OP::testAndSwapPtrAcqRel(
                                                                  &d_nextWrite,
                                                                  nextWrite,
                                                                  next));
    } while (nextWrite != expNextWrite);
 
    ATOMIC_OP::addInt64AcqRel(&d_state, -k_USE_INC);
 
    return nextWrite;
}
 
template <class TYPE, class ATOMIC_OP, class MUTEX, class CONDITION>
inline
void SingleConsumerQueueImpl<TYPE, ATOMIC_OP, MUTEX, CONDITION>
                                                        ::releaseAllocateLock()
{
    ATOMIC_OP::addInt64AcqRel(&d_state, -k_ALLOCATE_INC);
}
 
// CREATORS
template <class TYPE, class ATOMIC_OP, class MUTEX, class CONDITION>
SingleConsumerQueueImpl<TYPE, ATOMIC_OP, MUTEX, CONDITION>::
                      SingleConsumerQueueImpl(bslma::Allocator *basicAllocator)
: d_readMutex()
, d_readCondition()
, d_writeMutex()
, d_emptyMutex()
, d_emptyCondition()
, d_allocator(basicAllocator)
{
    ATOMIC_OP::initInt64(&d_capacity, 0);
    ATOMIC_OP::initInt64(&d_state,    0);
 
    ATOMIC_OP::initUint(&d_popFrontDisabled, 0);
    ATOMIC_OP::initUint(&d_pushBackDisabled, 0);
 
    ATOMIC_OP::initPointer(&d_nextWrite, 0);
 
    Node *n = static_cast<Node *>(d_allocator.allocate(sizeof(Node)));
    ATOMIC_OP::initInt(&n->d_state, e_WRITABLE);
    ATOMIC_OP::initPointer(&n->d_next, n);
 
    ATOMIC_OP::setPtrRelease(&d_nextWrite, n);
    ATOMIC_OP::setPtrRelease(&d_nextRead,  n);
}
 
template <class TYPE, class ATOMIC_OP, class MUTEX, class CONDITION>
SingleConsumerQueueImpl<TYPE, ATOMIC_OP, MUTEX, CONDITION>::
                      SingleConsumerQueueImpl(bsl::size_t       capacity,
                                              bslma::Allocator *basicAllocator)
: d_readMutex()
, d_readCondition()
, d_writeMutex()
, d_emptyMutex()
, d_emptyCondition()
, d_allocator(basicAllocator)
{
    ATOMIC_OP::initInt64(&d_capacity, 0);
    ATOMIC_OP::initInt64(&d_state,    0);
 
    ATOMIC_OP::initUint(&d_popFrontDisabled, 0);
    ATOMIC_OP::initUint(&d_pushBackDisabled, 0);
 
    ATOMIC_OP::initPointer(&d_nextWrite, 0);
 
    Node *nodes = static_cast<Node *>(d_allocator.allocate(  sizeof(Node)
                                                           * (capacity + 1)));
    for (bsl::size_t i = 0; i < capacity; ++i) {
        Node *n = nodes + i;
        ATOMIC_OP::initInt(&n->d_state, e_WRITABLE);
        ATOMIC_OP::initPointer(&n->d_next, n + 1);
    }
    {
        Node *n = nodes + capacity;
        ATOMIC_OP::initInt(&n->d_state, e_WRITABLE);
        ATOMIC_OP::initPointer(&n->d_next, nodes);
    }
 
    ATOMIC_OP::setPtrRelease(&d_nextWrite, nodes);
    ATOMIC_OP::setPtrRelease(&d_nextRead,  nodes);
 
    ATOMIC_OP::addInt64AcqRel(&d_capacity, capacity);
    ATOMIC_OP::addInt64AcqRel(&d_state, k_AVAILABLE_INC * capacity);
}
 
template <class TYPE, class ATOMIC_OP, class MUTEX, class CONDITION>
SingleConsumerQueueImpl<TYPE, ATOMIC_OP, MUTEX, CONDITION>::
                                                     ~SingleConsumerQueueImpl()
{
    Node *end = static_cast<Node *>(ATOMIC_OP::getPtrAcquire(&d_nextWrite));
 
    if (end) {
        Node *at = static_cast<Node *>(ATOMIC_OP::getPtrAcquire(&end->d_next));
 
        while (at != end) {
            Node *next =
                    static_cast<Node *>(ATOMIC_OP::getPtrAcquire(&at->d_next));
 
            if (e_READABLE == ATOMIC_OP::getIntAcquire(&at->d_state)) {
                at->d_value.object().~TYPE();
            }
 
            at = next;
        }
 
        if (e_READABLE == ATOMIC_OP::getIntAcquire(&at->d_state)) {
            at->d_value.object().~TYPE();
        }
    }
}
 
// MANIPULATORS
template <class TYPE, class ATOMIC_OP, class MUTEX, class CONDITION>
int SingleConsumerQueueImpl<TYPE, ATOMIC_OP, MUTEX, CONDITION>::popFront(
                                                                   TYPE *value)
{
    unsigned int generation = ATOMIC_OP::getUintAcquire(&d_popFrontDisabled);
    if (1 == (generation & 1)) {
        return e_DISABLED;                                            // RETURN
    }
 
    Node *nextRead =
                    static_cast<Node *>(ATOMIC_OP::getPtrAcquire(&d_nextRead));
    int nodeState = ATOMIC_OP::getIntAcquire(&nextRead->d_state);
    do {
        // Note that 'e_WRITABLE_AND_BLOCKED != nodeState' since if the one
        // consumer sets this state, the one consumer waits until the node is
        // readable, and either the producer that signalled the consumer
        // changed the node state already, or the consumer will change the node
        // state in 'popComplete'.
 
        if (e_WRITABLE == nodeState) {
            bslmt::ThreadUtil::yield();
            nodeState = ATOMIC_OP::getIntAcquire(&nextRead->d_state);
            if (e_WRITABLE == nodeState) {
                bslmt::LockGuard<MUTEX> guard(&d_readMutex);
                nodeState = ATOMIC_OP::swapIntAcqRel(&nextRead->d_state,
                                                     e_WRITABLE_AND_BLOCKED);
                while (e_READABLE != nodeState && e_RECLAIM != nodeState) {
                    if (generation !=
                              ATOMIC_OP::getUintAcquire(&d_popFrontDisabled)) {
                        ATOMIC_OP::testAndSwapIntAcqRel(&nextRead->d_state,
                                                        e_WRITABLE_AND_BLOCKED,
                                                        e_WRITABLE);
                        return e_DISABLED;                            // RETURN
                    }
                    d_readCondition.wait(&d_readMutex);
                    nodeState = ATOMIC_OP::getIntAcquire(&nextRead->d_state);
                }
            }
        }
        if (e_RECLAIM == nodeState) {
            ATOMIC_OP::addInt64AcqRel(&d_capacity, 1);
            popComplete(false);
            nextRead =
                    static_cast<Node *>(ATOMIC_OP::getPtrAcquire(&d_nextRead));
            nodeState = ATOMIC_OP::getIntAcquire(&nextRead->d_state);
        }
    } while (e_RECLAIM == nodeState);
 
    SingleConsumerQueueImpl_PopCompleteGuard<
                              SingleConsumerQueueImpl<TYPE,
                                                      ATOMIC_OP,
                                                      MUTEX,
                                                      CONDITION> > guard(this);
 
#if defined(BSLMF_MOVABLEREF_USES_RVALUE_REFERENCES)
    *value = bslmf::MovableRefUtil::move(nextRead->d_value.object());
#else
    *value = nextRead->d_value.object();
#endif
 
    return 0;
}
 
template <class TYPE, class ATOMIC_OP, class MUTEX, class CONDITION>
int SingleConsumerQueueImpl<TYPE, ATOMIC_OP, MUTEX, CONDITION>::pushBack(
                                                             const TYPE& value)
{
    Node *target = pushBackHelper();
 
    if (0 == target) {
        return e_DISABLED;                                            // RETURN
    }
 
    SingleConsumerQueueImpl_MarkReclaimProctor<
                                            SingleConsumerQueueImpl<TYPE,
                                                                    ATOMIC_OP,
                                                                    MUTEX,
                                                                    CONDITION>,
                                            Node> proctor(this, target);
 
    bslalg::ScalarPrimitives::copyConstruct(target->d_value.address(),
                                            value,
                                            allocator());
 
    proctor.release();
 
    int nodeState = ATOMIC_OP::swapIntAcqRel(&target->d_state, e_READABLE);
    if (e_WRITABLE_AND_BLOCKED == nodeState) {
        {
            bslmt::LockGuard<MUTEX> guard(&d_readMutex);
        }
        d_readCondition.signal();
    }
 
    return 0;
}
 
template <class TYPE, class ATOMIC_OP, class MUTEX, class CONDITION>
int SingleConsumerQueueImpl<TYPE, ATOMIC_OP, MUTEX, CONDITION>::pushBack(
                                                 bslmf::MovableRef<TYPE> value)
{
    Node *target = pushBackHelper();
 
    if (0 == target) {
        return e_DISABLED;                                            // RETURN
    }
 
    SingleConsumerQueueImpl_MarkReclaimProctor<
                                            SingleConsumerQueueImpl<TYPE,
                                                                    ATOMIC_OP,
                                                                    MUTEX,
                                                                    CONDITION>,
                                            Node> proctor(this, target);
 
    TYPE& dummy = value;
    bslalg::ScalarPrimitives::moveConstruct(target->d_value.address(),
                                            dummy,
                                            allocator());
 
    proctor.release();
 
    int nodeState = ATOMIC_OP::swapIntAcqRel(&target->d_state, e_READABLE);
    if (e_WRITABLE_AND_BLOCKED == nodeState) {
        {
            bslmt::LockGuard<MUTEX> guard(&d_readMutex);
        }
        d_readCondition.signal();
    }
 
    return 0;
}
 
template <class TYPE, class ATOMIC_OP, class MUTEX, class CONDITION>
void SingleConsumerQueueImpl<TYPE, ATOMIC_OP, MUTEX, CONDITION>::removeAll()
{
    int count = 0;
 
    bsls::Types::Int64 reclaim = 0;
 
    Node *nextRead =
                    static_cast<Node *>(ATOMIC_OP::getPtrAcquire(&d_nextRead));
    int nodeState = ATOMIC_OP::getIntAcquire(&nextRead->d_state);
    while (e_READABLE == nodeState || e_RECLAIM == nodeState) {
        if (e_READABLE == nodeState) {
            nextRead->d_value.object().~TYPE();
        }
        else {
            ++reclaim;
        }
        ATOMIC_OP::setIntRelease(&nextRead->d_state, e_WRITABLE);
        nextRead =
              static_cast<Node *>(ATOMIC_OP::getPtrAcquire(&nextRead->d_next));
        ATOMIC_OP::setPtrRelease(&d_nextRead, nextRead);
        nodeState = ATOMIC_OP::getIntAcquire(&nextRead->d_state);
        ++count;
    }
 
    ATOMIC_OP::addInt64AcqRel(&d_capacity, reclaim);
    ATOMIC_OP::addInt64AcqRel(&d_state, k_AVAILABLE_INC * count);
 
    {
        bslmt::LockGuard<MUTEX> guard(&d_emptyMutex);
    }
    d_emptyCondition.broadcast();
}
 
template <class TYPE, class ATOMIC_OP, class MUTEX, class CONDITION>
int SingleConsumerQueueImpl<TYPE, ATOMIC_OP, MUTEX, CONDITION>::tryPopFront(
                                                                   TYPE *value)
{
    unsigned int generation = ATOMIC_OP::getUintAcquire(&d_popFrontDisabled);
    if (1 == (generation & 1)) {
        return e_DISABLED;                                            // RETURN
    }
 
    Node *nextRead =
                    static_cast<Node *>(ATOMIC_OP::getPtrAcquire(&d_nextRead));
    int nodeState = ATOMIC_OP::getIntAcquire(&nextRead->d_state);
 
    while (e_RECLAIM == nodeState) {
        ATOMIC_OP::addInt64AcqRel(&d_capacity, 1);
        popComplete(false);
        nextRead = static_cast<Node *>(ATOMIC_OP::getPtrAcquire(&d_nextRead));
        nodeState = ATOMIC_OP::getIntAcquire(&nextRead->d_state);
    }
 
    if (e_READABLE != nodeState) {
        return e_EMPTY;                                               // RETURN
    }
 
    SingleConsumerQueueImpl_PopCompleteGuard<
                              SingleConsumerQueueImpl<TYPE,
                                                      ATOMIC_OP,
                                                      MUTEX,
                                                      CONDITION> > guard(this);
 
#if defined(BSLMF_MOVABLEREF_USES_RVALUE_REFERENCES)
    *value = bslmf::MovableRefUtil::move(nextRead->d_value.object());
#else
    *value = nextRead->d_value.object();
#endif
 
    return 0;
}
 
template <class TYPE, class ATOMIC_OP, class MUTEX, class CONDITION>
int SingleConsumerQueueImpl<TYPE, ATOMIC_OP, MUTEX, CONDITION>::tryPushBack(
                                                             const TYPE& value)
{
    return pushBack(value);
}
 
template <class TYPE, class ATOMIC_OP, class MUTEX, class CONDITION>
int SingleConsumerQueueImpl<TYPE, ATOMIC_OP, MUTEX, CONDITION>::tryPushBack(
                                                 bslmf::MovableRef<TYPE> value)
{
    return pushBack(bslmf::MovableRefUtil::move(value));
}
 
                       // Enqueue/Dequeue State
 
template <class TYPE, class ATOMIC_OP, class MUTEX, class CONDITION>
void SingleConsumerQueueImpl<TYPE, ATOMIC_OP, MUTEX, CONDITION>
                                                            ::disablePopFront()
{
    incrementUntil(&d_popFrontDisabled, 1);
 
    {
        bslmt::LockGuard<MUTEX> guard(&d_readMutex);
    }
    d_readCondition.signal();
 
    {
        bslmt::LockGuard<MUTEX> guard(&d_emptyMutex);
    }
    d_emptyCondition.broadcast();
}
 
template <class TYPE, class ATOMIC_OP, class MUTEX, class CONDITION>
void SingleConsumerQueueImpl<TYPE, ATOMIC_OP, MUTEX, CONDITION>
                                                            ::disablePushBack()
{
    incrementUntil(&d_pushBackDisabled, 1);
}
 
template <class TYPE, class ATOMIC_OP, class MUTEX, class CONDITION>
void SingleConsumerQueueImpl<TYPE, ATOMIC_OP, MUTEX, CONDITION>
                                                             ::enablePopFront()
{
    incrementUntil(&d_popFrontDisabled, 0);
}
 
template <class TYPE, class ATOMIC_OP, class MUTEX, class CONDITION>
void SingleConsumerQueueImpl<TYPE, ATOMIC_OP, MUTEX, CONDITION>
                                                             ::enablePushBack()
{
    incrementUntil(&d_pushBackDisabled, 0);
}
 
// ACCESSORS
template <class TYPE, class ATOMIC_OP, class MUTEX, class CONDITION>
bool SingleConsumerQueueImpl<TYPE, ATOMIC_OP, MUTEX, CONDITION>
                                                              ::isEmpty() const
{
    return ATOMIC_OP::getInt64Acquire(&d_capacity) ==
                               available(ATOMIC_OP::getInt64Acquire(&d_state));
}
 
template <class TYPE, class ATOMIC_OP, class MUTEX, class CONDITION>
bool SingleConsumerQueueImpl<TYPE, ATOMIC_OP, MUTEX, CONDITION>::isFull() const
{
    return false;
}
 
template <class TYPE, class ATOMIC_OP, class MUTEX, class CONDITION>
bool SingleConsumerQueueImpl<TYPE, ATOMIC_OP, MUTEX, CONDITION>
                                                   ::isPopFrontDisabled() const
{
    return 1 == (ATOMIC_OP::getUintAcquire(&d_popFrontDisabled) & 1);
}
 
template <class TYPE, class ATOMIC_OP, class MUTEX, class CONDITION>
bool SingleConsumerQueueImpl<TYPE, ATOMIC_OP, MUTEX, CONDITION>
                                                   ::isPushBackDisabled() const
{
    return 1 == (ATOMIC_OP::getUintAcquire(&d_pushBackDisabled) & 1);
}
 
template <class TYPE, class ATOMIC_OP, class MUTEX, class CONDITION>
bsl::size_t SingleConsumerQueueImpl<TYPE, ATOMIC_OP, MUTEX, CONDITION>
                                                          ::numElements() const
{
    bsls::Types::Int64 avail = available(ATOMIC_OP::getInt64Acquire(&d_state));
    return static_cast<bsl::size_t>(
                                avail > 0
                              ? ATOMIC_OP::getInt64Acquire(&d_capacity) - avail
                              : ATOMIC_OP::getInt64Acquire(&d_capacity));
}
 
template <class TYPE, class ATOMIC_OP, class MUTEX, class CONDITION>
int SingleConsumerQueueImpl<TYPE, ATOMIC_OP, MUTEX, CONDITION>
                                                       ::waitUntilEmpty() const
{
    unsigned int generation = ATOMIC_OP::getUintAcquire(&d_popFrontDisabled);
    if (1 == (generation & 1)) {
        return e_DISABLED;                                            // RETURN
    }
 
    bslmt::LockGuard<MUTEX> guard(&d_emptyMutex);
 
    bsls::Types::Int64 state = ATOMIC_OP::getInt64Acquire(&d_state);
    while (ATOMIC_OP::getInt64Acquire(&d_capacity) != available(state)) {
        if (generation != ATOMIC_OP::getUintAcquire(&d_popFrontDisabled)) {
            return e_DISABLED;                                        // RETURN
        }
        d_emptyCondition.wait(&d_emptyMutex);
        state = ATOMIC_OP::getInt64Acquire(&d_state);
    }
 
    return 0;
}
 
                                  // Aspects
 
template <class TYPE, class ATOMIC_OP, class MUTEX, class CONDITION>
bslma::Allocator *SingleConsumerQueueImpl<TYPE, ATOMIC_OP, MUTEX, CONDITION>::
                                                              allocator() const
{
    return d_allocator.allocator();
}
 
}  // 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 ----------------------------------
 
/** @} */
/** @} */
/** @} */