text: // bdlcc_singleproducersingleconsumerboundedqueue.h                   -*-C++-*-
  
  #ifndef INCLUDED_BDLCC_SINGLEPRODUCERSINGLECONSUMERBOUNDEDQUEUE
  #define INCLUDED_BDLCC_SINGLEPRODUCERSINGLECONSUMERBOUNDEDQUEUE
  
  #include <bsls_ident.h>
  BSLS_IDENT("$Id: $")
  
  //@PURPOSE: Provide a thread-aware SPSC bounded queue of values.
  //
  //@CLASSES:
  //  bdlcc::SingleProducerSingleConsumerBoundedQueue: SPSCB concurrent queue
  //
  //@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' 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 'bslma_usesbslmaallocator') so that the
  // allocator of the queue is propagated to the elements contained in the queue.
  //
  ///Exception Safety
  ///----------------
  // A 'bdlcc::SingleProducerSingleConsumerBoundedQueue' is exception neutral,
  // and all of the methods of 'bdlcc::SingleProducerSingleConsumerBoundedQueue'
  // provide the strong exception safety guarantee (see 'bsldoc_glossary').
  //
  ///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
  ///-----
  // This section illustrates intended use of this component.
  //
  ///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
  // 'bdlmt_fixedthreadpool' for more information.
  //
  // First, we define a utility classes that handles a simple "work item":
  //..
  //  struct my_WorkData {
  //      // Work data...
  //  };
  //
  //  struct my_WorkRequest {
  //      enum RequestType {
  //          e_WORK = 1,
  //          e_STOP = 2
  //      };
  //
  //      RequestType d_type;
  //      my_WorkData d_data;
  //      // Work data...
  //  };
  //..
  // Next, we provide a simple function to service an individual work item.  The
  // details are unimportant for this example:
  //..
  //  void myDoWork(const my_WorkData& data)
  //      // Do some work based upon the specified 'data'.
  //  {
  //      // do some stuff...
  //      (void)data;
  //  }
  //..
  // 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:
  //..
  //  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);
  //      }
  //  }
  //..
  // 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.
  //..
  //  void myProducer()
  //      // Create a queue, start consumer thread, produce and enqueue work.
  //  {
  //      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();
  //  }
  //..
  
  #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 BloombergLP {
  namespace bdlcc {
  
        // ===============================================================
        // class SingleProducerSingleConsumerBoundedQueue_PopCompleteGuard
        // ===============================================================
  
  template <class TYPE, class NODE>
  class SingleProducerSingleConsumerBoundedQueue_PopCompleteGuard {
      // This class implements a guard that invokes 'TYPE::popComplete' on a
      // 'NODE' upon destruction.
  
      // 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
      SingleProducerSingleConsumerBoundedQueue_PopCompleteGuard(TYPE   *queue,
                                                                NODE   *node,
                                                                Uint64  index);
          // Create a guard managing the specified 'queue' and will invoke
          // 'popComplete' with the specified 'node' and 'index'.
  
      ~SingleProducerSingleConsumerBoundedQueue_PopCompleteGuard();
          // Destroy this object and invoke the 'TYPE::popComplete'.
  };
  
                // ==============================================
                // 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
      static void incrementUntil(AtomicUint *value, unsigned int bitValue);
          // 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'.
  
      // PRIVATE MANIPULATORS
      void popComplete(Node *node, Uint64 index);
          // 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.
  
      int popFrontImp(TYPE *value, bool isTry);
          // 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 pushBackImp(const 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(bslmf::MovableRef<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.
  
      void pushComplete(Node *node, Uint64 index);
          // 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'.
  
      // 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
      explicit
      SingleProducerSingleConsumerBoundedQueue(
                                           bsl::size_t       capacity,
                                           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.
  
      ~SingleProducerSingleConsumerBoundedQueue();
          // Destroy this object.
  
      // MANIPULATORS
      int popFront(TYPE *value);
          // 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 pushBack(const 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(bslmf::MovableRef<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.
  
      void removeAll();
          // 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.
  
      int tryPopFront(TYPE *value);
          // 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 tryPushBack(const 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(bslmf::MovableRef<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.
  
                         // Enqueue/Dequeue State
  
      void disablePopFront();
          // 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 disablePushBack();
          // 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 enablePushBack();
          // Enable queuing.  If the queue is not enqueue disabled, this call has
          // no effect.
  
      void enablePopFront();
          // Enable dequeueing.  If the queue is not dequeue disabled, this call
          // has no effect.
  
      // ACCESSORS
      bsl::size_t capacity() const;
          // 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.
  
      bool isEmpty() const;
          // Return 'true' if this queue is empty (has no elements), or 'false'
          // otherwise.
  
      bool isFull() const;
          // Return 'true' if this queue is full (has no available capacity), or
          // 'false' otherwise.
  
      bool isPopFrontDisabled() const;
          // Return 'true' if this queue is dequeue disabled, and 'false'
          // otherwise.  Note that the queue is created in the "dequeue enabled"
          // state.
  
      bool isPushBackDisabled() const;
          // Return 'true' if this queue is enqueue disabled, and 'false'
          // otherwise.  Note that the queue is created in the "enqueue enabled"
          // state.
  
      bsl::size_t numElements() const;
          // Returns the number of elements currently in this queue.
  
      int waitUntilEmpty() 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.
  
                                    // Aspects
  
      bslma::Allocator *allocator() const;
          // Return the allocator used by this object to supply memory.
  };
  
  // ============================================================================
  //                             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 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
  }  // close enterprise 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 ----------------------------------
  