framework/delibs/decpp/deBlockBuffer.hpp

   1 #ifndef _DEBLOCKBUFFER_HPP
   2 #define _DEBLOCKBUFFER_HPP
   3 /*-------------------------------------------------------------------------
   4  * drawElements C++ Base Library
   5  * -----------------------------
   6  *
   7  * Copyright 2014 The Android Open Source Project
   8  *
   9  * Licensed under the Apache License, Version 2.0 (the "License");
  10  * you may not use this file except in compliance with the License.
  11  * You may obtain a copy of the License at
  12  *
  13  *      http://www.apache.org/licenses/LICENSE-2.0
  14  *
  15  * Unless required by applicable law or agreed to in writing, software
  16  * distributed under the License is distributed on an "AS IS" BASIS,
  17  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  18  * See the License for the specific language governing permissions and
  19  * limitations under the License.
  20  *
  21  *//*!
  22  * \file
  23  * \brief Block-based thread-safe queue.
  24  *//*--------------------------------------------------------------------*/
  25
  26 #include "deBlockBuffer.hpp"
  27 #include "deMutex.hpp"
  28 #include "deSemaphore.h"
  29
  30 #include <exception>
  31
  32 namespace de
  33 {
  34
  35 void BlockBuffer_selfTest (void);
  36
  37 class BufferCanceledException : public std::exception
  38 {
  39 public:
  40         inline BufferCanceledException  (void) {}
  41         inline ~BufferCanceledException (void) throw() {}
  42
  43         const char* what (void) const throw() { return "BufferCanceledException"; }
  44 };
  45
  46 template <typename T>
  47 class BlockBuffer
  48 {
  49 public:
  50         typedef BufferCanceledException CanceledException;
  51
  52                                         BlockBuffer                     (int blockSize, int numBlocks);
  53                                         ~BlockBuffer            (void);
  54
  55         void                    clear                           (void); //!< Resets buffer. Will block until pending writes and reads have completed.
  56
  57         void                    write                           (int numElements, const T* elements);
  58         int                             tryWrite                        (int numElements, const T* elements);
  59         void                    flush                           (void);
  60         bool                    tryFlush                        (void);
  61
  62         void                    read                            (int numElements, T* elements);
  63         int                             tryRead                         (int numElements, T* elements);
  64
  65         void                    cancel                          (void); //!< Sets buffer in canceled state. All (including pending) writes and reads will result in CanceledException.
  66         bool                    isCanceled                      (void) const { return !!m_canceled; }
  67
  68 private:
  69                                         BlockBuffer                     (const BlockBuffer& other);
  70         BlockBuffer&    operator=                       (const BlockBuffer& other);
  71
  72         int                             writeToCurrentBlock     (int numElements, const T* elements, bool blocking);
  73         int                             readFromCurrentBlock(int numElements, T* elements, bool blocking);
  74
  75         void                    flushWriteBlock         (void);
  76
  77         deSemaphore             m_fill;                         //!< Block fill count.
  78         deSemaphore             m_empty;                        //!< Block empty count.
  79
  80         int                             m_writeBlock;           //!< Current write block ndx.
  81         int                             m_writePos;                     //!< Position in block. 0 if block is not yet acquired.
  82
  83         int                             m_readBlock;            //!< Current read block ndx.
  84         int                             m_readPos;                      //!< Position in block. 0 if block is not yet acquired.
  85
  86         int                             m_blockSize;
  87         int                             m_numBlocks;
  88
  89         T*                              m_elements;
  90         int*                    m_numUsedInBlock;
  91
  92         Mutex                   m_writeLock;
  93         Mutex                   m_readLock;
  94
  95         volatile deUint32       m_canceled;
  96 } DE_WARN_UNUSED_TYPE;
  97
  98 template <typename T>
  99 BlockBuffer<T>::BlockBuffer (int blockSize, int numBlocks)
 100         : m_fill                        (0)
 101         , m_empty                       (0)
 102         , m_writeBlock          (0)
 103         , m_writePos            (0)
 104         , m_readBlock           (0)
 105         , m_readPos                     (0)
 106         , m_blockSize           (blockSize)
 107         , m_numBlocks           (numBlocks)
 108         , m_elements            (DE_NULL)
 109         , m_numUsedInBlock      (DE_NULL)
 110         , m_writeLock           ()
 111         , m_readLock            ()
 112         , m_canceled            (DE_FALSE)
 113 {
 114         DE_ASSERT(blockSize > 0);
 115         DE_ASSERT(numBlocks > 0);
 116
 117         try
 118         {
 119                 m_elements                      = new T[m_numBlocks*m_blockSize];
 120                 m_numUsedInBlock        = new int[m_numBlocks];
 121         }
 122         catch (...)
 123         {
 124                 delete[] m_elements;
 125                 delete[] m_numUsedInBlock;
 126                 throw;
 127         }
 128
 129         m_fill  = deSemaphore_create(0, DE_NULL);
 130         m_empty = deSemaphore_create(numBlocks, DE_NULL);
 131         DE_ASSERT(m_fill && m_empty);
 132 }
 133
 134 template <typename T>
 135 BlockBuffer<T>::~BlockBuffer (void)
 136 {
 137         delete[] m_elements;
 138         delete[] m_numUsedInBlock;
 139
 140         deSemaphore_destroy(m_fill);
 141         deSemaphore_destroy(m_empty);
 142 }
 143
 144 template <typename T>
 145 void BlockBuffer<T>::clear (void)
 146 {
 147         ScopedLock readLock             (m_readLock);
 148         ScopedLock writeLock    (m_writeLock);
 149
 150         deSemaphore_destroy(m_fill);
 151         deSemaphore_destroy(m_empty);
 152
 153         m_fill                  = deSemaphore_create(0, DE_NULL);
 154         m_empty                 = deSemaphore_create(m_numBlocks, DE_NULL);
 155         m_writeBlock    = 0;
 156         m_writePos              = 0;
 157         m_readBlock             = 0;
 158         m_readPos               = 0;
 159         m_canceled              = DE_FALSE;
 160
 161         DE_ASSERT(m_fill && m_empty);
 162 }
 163
 164 template <typename T>
 165 void BlockBuffer<T>::cancel (void)
 166 {
 167         DE_ASSERT(!m_canceled);
 168         m_canceled = DE_TRUE;
 169
 170         deSemaphore_increment(m_empty);
 171         deSemaphore_increment(m_fill);
 172 }
 173
 174 template <typename T>
 175 int BlockBuffer<T>::writeToCurrentBlock (int numElements, const T* elements, bool blocking)
 176 {
 177         DE_ASSERT(numElements > 0 && elements != DE_NULL);
 178
 179         if (m_writePos == 0)
 180         {
 181                 /* Write thread doesn't own current block - need to acquire. */
 182                 if (blocking)
 183                         deSemaphore_decrement(m_empty);
 184                 else
 185                 {
 186                         if (!deSemaphore_tryDecrement(m_empty))
 187                                 return 0;
 188                 }
 189
 190                 /* Check for canceled bit. */
 191                 if (m_canceled)
 192                 {
 193                         // \todo [2012-07-06 pyry] A bit hackish to assume that write lock is not freed if exception is thrown out here.
 194                         deSemaphore_increment(m_empty);
 195                         m_writeLock.unlock();
 196                         throw CanceledException();
 197                 }
 198         }
 199
 200         /* Write thread owns current block. */
 201         T*              block                   = m_elements + m_writeBlock*m_blockSize;
 202         int             numToWrite              = de::min(numElements, m_blockSize-m_writePos);
 203
 204         DE_ASSERT(numToWrite > 0);
 205
 206         for (int ndx = 0; ndx < numToWrite; ndx++)
 207                 block[m_writePos+ndx] = elements[ndx];
 208
 209         m_writePos += numToWrite;
 210
 211         if (m_writePos == m_blockSize)
 212                 flushWriteBlock(); /* Flush current write block. */
 213
 214         return numToWrite;
 215 }
 216
 217 template <typename T>
 218 int BlockBuffer<T>::readFromCurrentBlock (int numElements, T* elements, bool blocking)
 219 {
 220         DE_ASSERT(numElements > 0 && elements != DE_NULL);
 221
 222         if (m_readPos == 0)
 223         {
 224                 /* Read thread doesn't own current block - need to acquire. */
 225                 if (blocking)
 226                         deSemaphore_decrement(m_fill);
 227                 else
 228                 {
 229                         if (!deSemaphore_tryDecrement(m_fill))
 230                                 return 0;
 231                 }
 232
 233                 /* Check for canceled bit. */
 234                 if (m_canceled)
 235                 {
 236                         // \todo [2012-07-06 pyry] A bit hackish to assume that read lock is not freed if exception is thrown out here.
 237                         deSemaphore_increment(m_fill);
 238                         m_readLock.unlock();
 239                         throw CanceledException();
 240                 }
 241         }
 242
 243         /* Read thread now owns current block. */
 244         const T*        block                   = m_elements + m_readBlock*m_blockSize;
 245         int                     numUsedInBlock  = m_numUsedInBlock[m_readBlock];
 246         int                     numToRead               = de::min(numElements, numUsedInBlock-m_readPos);
 247
 248         DE_ASSERT(numToRead > 0);
 249
 250         for (int ndx = 0; ndx < numToRead; ndx++)
 251                 elements[ndx] = block[m_readPos+ndx];
 252
 253         m_readPos += numToRead;
 254
 255         if (m_readPos == numUsedInBlock)
 256         {
 257                 /* Free current read block and advance. */
 258                 m_readBlock             = (m_readBlock+1) % m_numBlocks;
 259                 m_readPos               = 0;
 260                 deSemaphore_increment(m_empty);
 261         }
 262
 263         return numToRead;
 264 }
 265
 266 template <typename T>
 267 int BlockBuffer<T>::tryWrite (int numElements, const T* elements)
 268 {
 269         int numWritten = 0;
 270
 271         DE_ASSERT(numElements > 0 && elements != DE_NULL);
 272
 273         if (m_canceled)
 274                 throw CanceledException();
 275
 276         if (!m_writeLock.tryLock())
 277                 return numWritten;
 278
 279         while (numWritten < numElements)
 280         {
 281                 int ret = writeToCurrentBlock(numElements-numWritten, elements+numWritten, false /* non-blocking */);
 282
 283                 if (ret == 0)
 284                         break; /* Write failed. */
 285
 286                 numWritten += ret;
 287         }
 288
 289         m_writeLock.unlock();
 290
 291         return numWritten;
 292 }
 293
 294 template <typename T>
 295 void BlockBuffer<T>::write (int numElements, const T* elements)
 296 {
 297         DE_ASSERT(numElements > 0 && elements != DE_NULL);
 298
 299         if (m_canceled)
 300                 throw CanceledException();
 301
 302         m_writeLock.lock();
 303
 304         int numWritten = 0;
 305         while (numWritten < numElements)
 306                 numWritten += writeToCurrentBlock(numElements-numWritten, elements+numWritten, true /* blocking */);
 307
 308         m_writeLock.unlock();
 309 }
 310
 311 template <typename T>
 312 void BlockBuffer<T>::flush (void)
 313 {
 314         m_writeLock.lock();
 315
 316         if (m_writePos > 0)
 317                 flushWriteBlock();
 318
 319         m_writeLock.unlock();
 320 }
 321
 322 template <typename T>
 323 bool BlockBuffer<T>::tryFlush (void)
 324 {
 325         if (!m_writeLock.tryLock())
 326                 return false;
 327
 328         if (m_writePos > 0)
 329                 flushWriteBlock();
 330
 331         m_writeLock.unlock();
 332
 333         return true;
 334 }
 335
 336 template <typename T>
 337 void BlockBuffer<T>::flushWriteBlock (void)
 338 {
 339         DE_ASSERT(de::inRange(m_writePos, 1, m_blockSize));
 340
 341         m_numUsedInBlock[m_writeBlock]  = m_writePos;
 342         m_writeBlock                                    = (m_writeBlock+1) % m_numBlocks;
 343         m_writePos                                              = 0;
 344         deSemaphore_increment(m_fill);
 345 }
 346
 347 template <typename T>
 348 int BlockBuffer<T>::tryRead (int numElements, T* elements)
 349 {
 350         int numRead = 0;
 351
 352         if (m_canceled)
 353                 throw CanceledException();
 354
 355         if (!m_readLock.tryLock())
 356                 return numRead;
 357
 358         while (numRead < numElements)
 359         {
 360                 int ret = readFromCurrentBlock(numElements-numRead, &elements[numRead], false /* non-blocking */);
 361
 362                 if (ret == 0)
 363                         break; /* Failed. */
 364
 365                 numRead += ret;
 366         }
 367
 368         m_readLock.unlock();
 369
 370         return numRead;
 371 }
 372
 373 template <typename T>
 374 void BlockBuffer<T>::read (int numElements, T* elements)
 375 {
 376         DE_ASSERT(numElements > 0 && elements != DE_NULL);
 377
 378         if (m_canceled)
 379                 throw CanceledException();
 380
 381         m_readLock.lock();
 382
 383         int numRead = 0;
 384         while (numRead < numElements)
 385                 numRead += readFromCurrentBlock(numElements-numRead, &elements[numRead], true /* blocking */);
 386
 387         m_readLock.unlock();
 388 }
 389
 390 } // de
 391
 392 #endif // _DEBLOCKBUFFER_HPP