queue/array_lock_free_queue_single_producer_impl.h

   1 // ============================================================================
   2 // Copyright (c) 2010 Faustino Frechilla
   3 // All rights reserved.
   4 //
   5 // Redistribution and use in source and binary forms, with or without
   6 // modification, are permitted provided that the following conditions are met:
   7 //
   8 //  1. Redistributions of source code must retain the above copyright notice,
   9 //     this list of conditions and the following disclaimer.
  10 //  2. Redistributions in binary form must reproduce the above copyright
  11 //     notice, this list of conditions and the following disclaimer in the
  12 //     documentation and/or other materials provided with the distribution.
  13 //  3. The name of the author may not be used to endorse or promote products
  14 //     derived from this software without specific prior written permission.
  15 //
  16 // THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
  17 // AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
  18 // IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
  19 // ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE
  20 // LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR
  21 // CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF
  22 // SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS
  23 // INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN
  24 // CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE)
  25 // ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
  26 // POSSIBILITY OF SUCH DAMAGE.
  27 //
  28 /// @file array_lock_free_queue_single_producer_impl.h
  29 /// @brief Implementation of a circular array based lock-free queue
  30 ///
  31 /// @author Faustino Frechilla
  32 /// @history
  33 /// Ref  Who                 When         What
  34 ///      Faustino Frechilla  11-Jul-2010  Original development
  35 /// @endhistory
  36 ///
  37 // ============================================================================
  38
  39 #ifndef __ARRAY_LOCK_FREE_QUEUE_SINGLE_PRODUCER_IMPL_H__
  40 #define __ARRAY_LOCK_FREE_QUEUE_SINGLE_PRODUCER_IMPL_H__
  41
  42 #include <assert.h> // assert()
  43
  44 template <typename ELEM_T, uint32_t Q_SIZE>
  45 ArrayLockFreeQueueSingleProducer<ELEM_T, Q_SIZE>::ArrayLockFreeQueueSingleProducer() :
  46     m_writeIndex(0),
  47     m_readIndex(0)
  48 {
  49 #ifdef ARRAY_LOCK_FREE_Q_KEEP_REAL_SIZE
  50     m_count = 0;
  51 #endif
  52 }
  53
  54 template <typename ELEM_T, uint32_t Q_SIZE>
  55 ArrayLockFreeQueueSingleProducer<ELEM_T, Q_SIZE>::~ArrayLockFreeQueueSingleProducer()
  56 {
  57 }
  58
  59 template <typename ELEM_T, uint32_t Q_SIZE>
  60 inline
  61 uint32_t ArrayLockFreeQueueSingleProducer<ELEM_T, Q_SIZE>::countToIndex(uint32_t a_count)
  62 {
  63     // if Q_SIZE is a power of 2 this statement could be also written as
  64     // return (a_count & (Q_SIZE - 1));
  65     return (a_count % Q_SIZE);
  66 }
  67
  68 template <typename ELEM_T, uint32_t Q_SIZE>
  69 uint32_t ArrayLockFreeQueueSingleProducer<ELEM_T, Q_SIZE>::size()
  70 {
  71 #ifdef ARRAY_LOCK_FREE_Q_KEEP_REAL_SIZE
  72     return m_count;
  73 #else
  74     uint32_t currentWriteIndex = m_writeIndex;
  75     uint32_t currentReadIndex  = m_readIndex;
  76
  77     // let's think of a scenario where this function returns bogus data
  78     // 1. when the statement 'currentWriteIndex = m_writeIndex' is run
  79     // m_writeIndex is 3 and m_readIndex is 2. Real size is 1
  80     // 2. afterwards this thread is preemted. While this thread is inactive 2
  81     // elements are inserted and removed from the queue, so m_writeIndex is 5
  82     // m_readIndex 4. Real size is still 1
  83     // 3. Now the current thread comes back from preemption and reads m_readIndex.
  84     // currentReadIndex is 4
  85     // 4. currentReadIndex is bigger than currentWriteIndex, so
  86     // m_totalSize + currentWriteIndex - currentReadIndex is returned, that is,
  87     // it returns that the queue is almost full, when it is almost empty
  88
  89     if (currentWriteIndex >= currentReadIndex)
  90     {
  91         return (currentWriteIndex - currentReadIndex);
  92     }
  93     else
  94     {
  95         return (Q_SIZE + currentWriteIndex - currentReadIndex);
  96     }
  97 #endif // ARRAY_LOCK_FREE_Q_KEEP_REAL_SIZE
  98 }
  99
 100 template <typename ELEM_T, uint32_t Q_SIZE>
 101 bool ArrayLockFreeQueueSingleProducer<ELEM_T, Q_SIZE>::push(const ELEM_T &a_data)
 102 {
 103     uint32_t currentReadIndex;
 104     uint32_t currentWriteIndex;
 105
 106     currentWriteIndex = m_writeIndex;
 107     currentReadIndex  = m_readIndex;
 108     if (countToIndex(currentWriteIndex + 1) ==
 109         countToIndex(currentReadIndex))
 110     {
 111         // the queue is full
 112         return false;
 113     }
 114
 115     // save the date into the q
 116     m_theQueue[countToIndex(currentWriteIndex)] = a_data;
 117     // increment atomically write index. Now a consumer thread can read
 118     // the piece of data that was just stored
 119     AtomicAdd(&m_writeIndex, 1);
 120
 121     // The value was successfully inserted into the queue
 122 #ifdef ARRAY_LOCK_FREE_Q_KEEP_REAL_SIZE
 123     AtomicAdd(&m_count, 1);
 124 #endif
 125
 126     return true;
 127 }
 128
 129 template <typename ELEM_T, uint32_t Q_SIZE>
 130 bool ArrayLockFreeQueueSingleProducer<ELEM_T, Q_SIZE>::pop(ELEM_T &a_data)
 131 {
 132     uint32_t currentMaximumReadIndex;
 133     uint32_t currentReadIndex;
 134
 135     do
 136     {
 137         // m_maximumReadIndex doesn't exist when the queue is set up as
 138         // single-producer. The maximum read index is described by the current
 139         // write index
 140         currentReadIndex        = m_readIndex;
 141         currentMaximumReadIndex = m_writeIndex;
 142
 143         if (countToIndex(currentReadIndex) ==
 144             countToIndex(currentMaximumReadIndex))
 145         {
 146             // the queue is empty or
 147             // a producer thread has allocate space in the queue but is
 148             // waiting to commit the data into it
 149             return false;
 150         }
 151
 152         // retrieve the data from the queue
 153         a_data = m_theQueue[countToIndex(currentReadIndex)];
 154
 155         // try to perfrom now the CAS operation on the read index. If we succeed
 156         // a_data already contains what m_readIndex pointed to before we
 157         // increased it
 158         if (CAS(&m_readIndex, currentReadIndex, (currentReadIndex + 1)))
 159         {
 160             // got here. The value was retrieved from the queue. Note that the
 161             // data inside the m_queue array is not deleted nor reseted
 162 #ifdef ARRAY_LOCK_FREE_Q_KEEP_REAL_SIZE
 163             AtomicSub(&m_count, 1);
 164 #endif
 165             return true;
 166         }
 167
 168         // it failed retrieving the element off the queue. Someone else must
 169         // have read the element stored at countToIndex(currentReadIndex)
 170         // before we could perform the CAS operation
 171
 172     } while(1); // keep looping to try again!
 173
 174     // Something went wrong. it shouldn't be possible to reach here
 175     assert(0);
 176
 177     // Add this return statement to avoid compiler warnings
 178     return false;
 179 }
 180
 181 #endif // __ARRAY_LOCK_FREE_QUEUE_SINGLE_PRODUCER_IMPL_H__
 182