irc/ArchBlackmann/QueueT.h

   1 /*
   2 ** Author: Samuel R. Blackburn
   3 ** Internet: wfc@pobox.com
   4 **
   5 ** You can use it any way you like as long as you don't try to sell it.
   6 **
   7 ** Any attempt to sell WFC in source code form must have the permission
   8 ** of the original author. You can produce commercial executables with
   9 ** WFC but you can't sell WFC.
  10 **
  11 ** Copyright, 2000, Samuel R. Blackburn
  12 **
  13 ** NOTE: I modified the info block below so it hopefully wouldn't conflict
  14 ** with the original file. Royce Mitchell III
  15 */
  16
  17 #ifndef QUEUET_CLASS_HEADER
  18 #define QUEUET_CLASS_HEADER
  19
  20 #include "ReliMT.h"
  21
  22 #ifdef WIN32
  23 #include <sys/types.h> // off_t
  24 #define HEAPCREATE(size) m_Heap = ::HeapCreate ( HEAP_NO_SERIALIZE, size, 0 )
  25 #define HEAPALLOC(size) ::HeapAlloc ( m_Heap, HEAP_NO_SERIALIZE, size )
  26 #define HEAPREALLOC(p,size) ::HeapReAlloc( m_Heap, HEAP_NO_SERIALIZE, p, size )
  27 #define HEAPFREE(p) ::HeapFree ( m_Heap, HEAP_NO_SERIALIZE, p )
  28 #define HEAPDESTROY() ::HeapDestroy ( m_Heap ); m_Heap = 0;
  29 #else
  30 #define HEAPCREATE(size)
  31 #define HEAPALLOC(size) malloc(size)
  32 #define HEAPREALLOC(p,size) realloc(p,size);
  33 #define HEAPFREE(p) free(p)
  34 #define HEAPDESTROY()
  35 #endif
  36
  37 template <class T>
  38 class CQueueT : public Uncopyable
  39 {
  40 protected:
  41
  42         // What we want to protect
  43
  44         Mutex m_AddMutex;
  45         Mutex m_GetMutex;
  46
  47         T* m_Items;
  48
  49         off_t m_AddIndex;
  50         off_t m_GetIndex;
  51         size_t m_Size;
  52
  53 #ifdef WIN32
  54         HANDLE m_Heap;
  55 #endif//WIN32
  56         inline void m_GrowBy ( size_t number_of_new_items );
  57
  58 public:
  59
  60         inline  CQueueT ( size_t initial_size = 1024 );
  61         inline ~CQueueT();
  62
  63         inline bool  Add( const T& new_item );
  64         inline void  Empty() { m_AddIndex = 0; m_GetIndex = 0; };
  65         inline bool  Get( T& item );
  66         inline size_t GetLength() const;
  67         inline size_t GetMaximumLength() const { return( m_Size ); };
  68         inline bool  AddArray ( const T* new_items, int item_count );
  69         inline int GetArray ( T* items, const int maxget, const T& tEnd );
  70         inline bool Contains ( const T& t );
  71 };
  72
  73 template <class T>
  74 inline CQueueT<T>::CQueueT ( size_t initial_size )
  75 {
  76         m_AddIndex = 0;
  77         m_GetIndex = 0;
  78         m_Items    = NULL;
  79
  80         if ( initial_size == 0 )
  81                 initial_size = 1;
  82
  83         /*
  84         ** 1999-11-05
  85         ** We create our own heap because all of the pointers used are allocated
  86         ** and freed be us. We don't have to worry about a non-serialized thread
  87         ** accessing something we allocated. Because of this, we can perform our
  88         ** memory allocations in a heap dedicated to queueing. This means when we
  89         ** have to allocate more memory, we don't have to wait for all other threads
  90         ** to pause while we allocate from the shared heap (like the C Runtime heap)
  91         */
  92
  93         HEAPCREATE( ( ( ( 2 * initial_size * sizeof(T) ) < 65536 ) ? 65536 : (2 * initial_size * sizeof(T) ) ) );
  94
  95         m_Items = (T*)HEAPALLOC ( initial_size * sizeof(T) );
  96
  97         m_Size = ( m_Items == NULL ) ? 0 : initial_size;
  98 }
  99
 100 template <class T>
 101 inline CQueueT<T>::~CQueueT()
 102 {
 103         m_AddIndex = 0;
 104         m_GetIndex = 0;
 105         m_Size     = 0;
 106
 107         if ( m_Items != NULL )
 108         {
 109                 HEAPFREE(m_Items);
 110                 m_Items = NULL;
 111         }
 112
 113         HEAPDESTROY();
 114 }
 115
 116 template <class T>
 117 inline bool CQueueT<T>::Add ( const T& item )
 118 {
 119         // Block other threads from entering Add();
 120         Mutex::Lock addlock ( m_AddMutex );
 121
 122         // Add the item
 123
 124         m_Items[ m_AddIndex ] = item;
 125
 126         // 1999-12-08
 127         // Many many thanks go to Lou Franco (lfranco@spheresoft.com)
 128         // for finding an bug here. It rare but recreatable situations,
 129         // m_AddIndex could be in an invalid state.
 130
 131         // Make sure m_AddIndex is never invalid
 132
 133         off_t new_add_index = ( ( m_AddIndex + 1 ) >= (off_t)m_Size ) ? 0 : m_AddIndex + 1;
 134
 135         if ( new_add_index == m_GetIndex )
 136         {
 137                 // The queue is full. We need to grow.
 138                 // Stop anyone from getting from the queue
 139                 Mutex::Lock getlock ( m_GetMutex );
 140
 141                 m_AddIndex = new_add_index;
 142
 143                 // One last double-check.
 144
 145                 if ( m_AddIndex == m_GetIndex )
 146                 {
 147                         m_GrowBy ( m_Size );
 148                 }
 149
 150         }
 151         else
 152         {
 153                 m_AddIndex = new_add_index;
 154         }
 155
 156         return true;
 157 }
 158
 159 template <class T>
 160 inline bool CQueueT<T>::Get( T& item )
 161 {
 162         // Prevent other threads from entering Get()
 163         Mutex::Lock getlock ( m_GetMutex );
 164
 165         if ( m_GetIndex == m_AddIndex )
 166         {
 167                 // Let's check to see if our queue has grown too big
 168                 // If it has, then shrink it
 169
 170                 if ( m_Size > 1024 )
 171                 {
 172                         // Yup, we're too big for our britches
 173                         Mutex::TryLock addlock ( m_AddMutex );
 174                         if ( addlock )
 175                         {
 176                                 // Now, no one can add to the queue
 177
 178                                 if ( m_GetIndex == m_AddIndex ) // is queue empty?
 179                                 {
 180                                         // See if we can just shrink it...
 181                                         T* return_value = (T*)HEAPREALLOC(m_Items,1024 * sizeof(T));
 182
 183                                         if ( return_value != NULL )
 184                                         {
 185                                                 m_Items = (T*) return_value;
 186                                         }
 187                                         else
 188                                         {
 189                                                 // Looks like we'll have to do it the hard way
 190                                                 HEAPFREE ( m_Items );
 191                                                 m_Items = (T*) HEAPALLOC ( 1024 * sizeof(T) );
 192                                         }
 193
 194                                         m_Size     = ( m_Items == NULL ) ? 0 : 1024;
 195                                         m_AddIndex = 0;
 196                                         m_GetIndex = 0;
 197                                 }
 198                                 else
 199                                 {
 200                                         // m_GetIndex != m_AddIndex, this means that someone added
 201                                         // to the queue between the time we checked m_Size for being
 202                                         // too big and the time we entered the add critical section.
 203                                         // If this happened then we are too busy to shrink
 204                                 }
 205                         }
 206                 }
 207                 return false;
 208         }
 209
 210         item = m_Items[ m_GetIndex ];
 211
 212         // Make sure m_GetIndex is never invalid
 213
 214         m_GetIndex = ( ( m_GetIndex + 1 ) >= (off_t)m_Size ) ? 0 : m_GetIndex + 1;
 215
 216         return true;
 217 }
 218
 219 template <class T>
 220 inline int CQueueT<T>::GetArray ( T* items, const int maxget, const T& tEnd )
 221 {
 222         // TODO - oooh baby does this need to be optimized
 223         // Prevent other threads from entering Get()
 224         Mutex::Lock getlock ( m_GetMutex ); //::EnterCriticalSection( &m_GetCriticalSection );
 225
 226         int iResult = 0;
 227         for ( int i = 0; i < maxget; i++ )
 228         {
 229                 if ( !Get(items[i]) )
 230                         break;
 231                 iResult++;
 232                 if ( items[i] == tEnd )
 233                         break;
 234         }
 235         // Let other threads call Get() now
 236         //::LeaveCriticalSection( &m_GetCriticalSection );
 237         return iResult;
 238 }
 239
 240 template <class T>
 241 inline size_t CQueueT<T>::GetLength() const
 242 {
 243         // This is a very expensive process!
 244         // No one can call Add() or Get() while we're computing this
 245
 246         size_t number_of_items_in_the_queue = 0;
 247
 248         Mutex::Lock addlock ( m_AddMutex );
 249         Mutex::Lock getlock ( m_GetMutex );
 250
 251         number_of_items_in_the_queue = ( m_AddIndex >= m_GetIndex ) ?
 252                                        ( m_AddIndex  - m_GetIndex ) :
 253                                        ( ( m_AddIndex + m_Size ) - m_GetIndex );
 254
 255         return number_of_items_in_the_queue;
 256 }
 257
 258 template <class T>
 259 inline void CQueueT<T>::m_GrowBy ( size_t number_of_new_items )
 260 {
 261         // Prevent other threads from calling Get().
 262         // We don't need to enter the AddCriticalSection because
 263         // m_GrowBy() is only called from Add();
 264
 265         T* new_array       = NULL;
 266         T* pointer_to_free = NULL;
 267
 268         size_t new_size = m_Size + number_of_new_items;
 269
 270         { // Prevent other threads from getting
 271                 Mutex::Lock getlock ( m_GetMutex );
 272
 273                 // 2000-05-16
 274                 // Thanks go to Royce Mitchell III (royce3@aim-controls.com) for finding
 275                 // a HUGE bug here. I was using HeapReAlloc as a short cut but my logic
 276                 // was flawed. In certain circumstances, queue items were being dropped.
 277
 278                 new_array = (T*)HEAPALLOC ( new_size * sizeof(T) );
 279
 280                 // Now copy all of the old items from the old queue to the new one.
 281
 282                 // Get the entries from the get-index to the end of the array
 283                 memcpy ( new_array, &m_Items[m_GetIndex], ( m_Size - m_GetIndex ) * sizeof(T) );
 284
 285                 // Get the entries from the beginning of the array to the add-index
 286                 memcpy ( &new_array[m_Size-m_GetIndex], m_Items, m_AddIndex * sizeof(T) );
 287
 288                 m_AddIndex      = (off_t)m_Size;
 289                 m_GetIndex      = 0;
 290                 m_Size          = new_size;
 291                 pointer_to_free = m_Items;
 292                 m_Items         = new_array;
 293         } // Mutex::Lock
 294         HEAPFREE ( pointer_to_free );
 295 }
 296
 297 template <class T>
 298 inline bool CQueueT<T>::Contains ( const T& t )
 299 {
 300         Mutex::Lock addlock ( m_AddMutex );
 301         Mutex::Lock getlock ( m_GetMutex );
 302
 303         for ( int i = m_GetIndex; i != m_AddIndex; i++ )
 304         {
 305                 if ( i == m_Size )
 306                         i = 0;
 307                 if ( m_Items[i] == t )
 308                         return true;
 309         }
 310         return m_Items[m_AddIndex] == t;
 311 }
 312
 313 typedef CQueueT<char> CCharQueue;
 314
 315 #endif // QUEUE_CLASS_HEADER