4 #ifndef SQUID_IPC_QUEUE_H
5 #define SQUID_IPC_QUEUE_H
7 #include "base/InstanceId.h"
9 #include "ipc/AtomicWord.h"
10 #include "ipc/mem/FlexibleArray.h"
11 #include "ipc/mem/Pointer.h"
19 /// State of the reading end of a queue (i.e., of the code calling pop()).
20 /// Multiple queues attached to one reader share this state.
24 QueueReader(); // the initial state is "blocked without a signal"
26 /// whether the reader is waiting for a notification signal
27 bool blocked() const { return popBlocked
== 1; }
29 /// marks the reader as blocked, waiting for a notification signal
30 void block() { popBlocked
.swap_if(0, 1); }
32 /// removes the block() effects
33 void unblock() { popBlocked
.swap_if(1, 0); }
35 /// if reader is blocked and not notified, marks the notification signal
36 /// as sent and not received, returning true; otherwise, returns false
37 bool raiseSignal() { return blocked() && popSignal
.swap_if(0,1); }
39 /// marks sent reader notification as received (also removes pop blocking)
40 void clearSignal() { unblock(); popSignal
.swap_if(1,0); }
43 Atomic::Word popBlocked
; ///< whether the reader is blocked on pop()
44 Atomic::Word popSignal
; ///< whether writer has sent and reader has not received notification
47 typedef Atomic::Word Rate
; ///< pop()s per second
48 Rate rateLimit
; ///< pop()s per second limit if positive
50 // we need a signed atomic type because balance may get negative
51 typedef Atomic::WordT
<int> AtomicSignedMsec
;
52 typedef AtomicSignedMsec Balance
;
53 /// how far ahead the reader is compared to a perfect read/sec event rate
56 /// unique ID for debugging which reader is used (works across processes)
57 const InstanceId
<QueueReader
> id
;
60 /// shared array of QueueReaders
64 QueueReaders(const int aCapacity
);
65 size_t sharedMemorySize() const;
66 static size_t SharedMemorySize(const int capacity
);
68 const int theCapacity
; /// number of readers
69 Ipc::Mem::FlexibleArray
<QueueReader
> theReaders
; /// readers
73 * Lockless fixed-capacity queue for a single writer and a single reader.
75 * If the queue is empty, the reader is considered "blocked" and needs
76 * an out-of-band notification message to notice the next pushed item.
78 * Current implementation assumes that the writer cannot get blocked: if the
79 * queue is full, the writer will just not push and come back later (with a
80 * different value). We can add support for blocked writers if needed.
82 class OneToOneUniQueue
85 // pop() and push() exceptions; TODO: use TextException instead
87 class ItemTooLarge
{};
89 OneToOneUniQueue(const unsigned int aMaxItemSize
, const int aCapacity
);
91 unsigned int maxItemSize() const { return theMaxItemSize
; }
92 int size() const { return theSize
; }
93 int capacity() const { return theCapacity
; }
94 int sharedMemorySize() const { return Items2Bytes(theMaxItemSize
, theCapacity
); }
96 bool empty() const { return !theSize
; }
97 bool full() const { return theSize
== theCapacity
; }
99 static int Bytes2Items(const unsigned int maxItemSize
, int size
);
100 static int Items2Bytes(const unsigned int maxItemSize
, const int size
);
102 /// returns true iff the value was set; [un]blocks the reader as needed
103 template<class Value
> bool pop(Value
&value
, QueueReader
*const reader
= NULL
);
105 /// returns true iff the caller must notify the reader of the pushed item
106 template<class Value
> bool push(const Value
&value
, QueueReader
*const reader
= NULL
);
108 /// returns true iff the value was set; the value may be stale!
109 template<class Value
> bool peek(Value
&value
) const;
113 unsigned int theIn
; ///< input index, used only in push()
114 unsigned int theOut
; ///< output index, used only in pop()
116 Atomic::Word theSize
; ///< number of items in the queue
117 const unsigned int theMaxItemSize
; ///< maximum item size
118 const int theCapacity
; ///< maximum number of items, i.e. theBuffer size
123 /// shared array of OneToOneUniQueues
124 class OneToOneUniQueues
127 OneToOneUniQueues(const int aCapacity
, const unsigned int maxItemSize
, const int queueCapacity
);
129 size_t sharedMemorySize() const;
130 static size_t SharedMemorySize(const int capacity
, const unsigned int maxItemSize
, const int queueCapacity
);
132 const OneToOneUniQueue
&operator [](const int index
) const;
133 inline OneToOneUniQueue
&operator [](const int index
);
136 inline const OneToOneUniQueue
&front() const;
139 const int theCapacity
; /// number of OneToOneUniQueues
143 * Base class for lockless fixed-capacity bidirectional queues for a
144 * limited number processes.
149 BaseMultiQueue(const int aLocalProcessId
);
150 virtual ~BaseMultiQueue() {}
152 /// clears the reader notification received by the local process from the remote process
153 void clearReaderSignal(const int remoteProcessId
);
155 /// picks a process and calls OneToOneUniQueue::pop() using its queue
156 template <class Value
> bool pop(int &remoteProcessId
, Value
&value
);
158 /// calls OneToOneUniQueue::push() using the given process queue
159 template <class Value
> bool push(const int remoteProcessId
, const Value
&value
);
161 /// peeks at the item likely to be pop()ed next
162 template<class Value
> bool peek(int &remoteProcessId
, Value
&value
) const;
164 /// returns local reader's balance
165 QueueReader::Balance
&localBalance() { return localReader().balance
; }
167 /// returns reader's balance for a given remote process
168 const QueueReader::Balance
&balance(const int remoteProcessId
) const;
170 /// returns local reader's rate limit
171 QueueReader::Rate
&localRateLimit() { return localReader().rateLimit
; }
173 /// returns reader's rate limit for a given remote process
174 const QueueReader::Rate
&rateLimit(const int remoteProcessId
) const;
176 /// number of items in incoming queue from a given remote process
177 int inSize(const int remoteProcessId
) const { return inQueue(remoteProcessId
).size(); }
179 /// number of items in outgoing queue to a given remote process
180 int outSize(const int remoteProcessId
) const { return outQueue(remoteProcessId
).size(); }
183 /// incoming queue from a given remote process
184 virtual const OneToOneUniQueue
&inQueue(const int remoteProcessId
) const = 0;
185 OneToOneUniQueue
&inQueue(const int remoteProcessId
);
187 /// outgoing queue to a given remote process
188 virtual const OneToOneUniQueue
&outQueue(const int remoteProcessId
) const = 0;
189 OneToOneUniQueue
&outQueue(const int remoteProcessId
);
191 virtual const QueueReader
&localReader() const = 0;
192 QueueReader
&localReader();
194 virtual const QueueReader
&remoteReader(const int remoteProcessId
) const = 0;
195 QueueReader
&remoteReader(const int remoteProcessId
);
197 virtual int remotesCount() const = 0;
198 virtual int remotesIdOffset() const = 0;
201 const int theLocalProcessId
; ///< process ID of this queue
204 int theLastPopProcessId
; ///< the ID of the last process we tried to pop() from
208 * Lockless fixed-capacity bidirectional queue for a limited number
209 * processes. Allows communication between two groups of processes:
210 * any process in one group may send data to and receive from any
211 * process in another group, but processes in the same group can not
212 * communicate. Process in each group has a unique integer ID in
213 * [groupIdOffset, groupIdOffset + groupSize) range.
215 class FewToFewBiQueue
: public BaseMultiQueue
218 typedef OneToOneUniQueue::Full Full
;
219 typedef OneToOneUniQueue::ItemTooLarge ItemTooLarge
;
222 /// Shared metadata for FewToFewBiQueue
224 Metadata(const int aGroupASize
, const int aGroupAIdOffset
, const int aGroupBSize
, const int aGroupBIdOffset
);
225 size_t sharedMemorySize() const { return sizeof(*this); }
226 static size_t SharedMemorySize(const int, const int, const int, const int) { return sizeof(Metadata
); }
228 const int theGroupASize
;
229 const int theGroupAIdOffset
;
230 const int theGroupBSize
;
231 const int theGroupBIdOffset
;
238 Owner(const String
&id
, const int groupASize
, const int groupAIdOffset
, const int groupBSize
, const int groupBIdOffset
, const unsigned int maxItemSize
, const int capacity
);
242 Mem::Owner
<Metadata
> *const metadataOwner
;
243 Mem::Owner
<OneToOneUniQueues
> *const queuesOwner
;
244 Mem::Owner
<QueueReaders
> *const readersOwner
;
247 static Owner
*Init(const String
&id
, const int groupASize
, const int groupAIdOffset
, const int groupBSize
, const int groupBIdOffset
, const unsigned int maxItemSize
, const int capacity
);
249 enum Group
{ groupA
= 0, groupB
= 1 };
250 FewToFewBiQueue(const String
&id
, const Group aLocalGroup
, const int aLocalProcessId
);
252 /// maximum number of items in the queue
253 static int MaxItemsCount(const int groupASize
, const int groupBSize
, const int capacity
);
255 /// finds the oldest item in incoming and outgoing queues between
256 /// us and the given remote process
257 template<class Value
> bool findOldest(const int remoteProcessId
, Value
&value
) const;
260 virtual const OneToOneUniQueue
&inQueue(const int remoteProcessId
) const;
261 virtual const OneToOneUniQueue
&outQueue(const int remoteProcessId
) const;
262 virtual const QueueReader
&localReader() const;
263 virtual const QueueReader
&remoteReader(const int processId
) const;
264 virtual int remotesCount() const;
265 virtual int remotesIdOffset() const;
268 bool validProcessId(const Group group
, const int processId
) const;
269 int oneToOneQueueIndex(const Group fromGroup
, const int fromProcessId
, const Group toGroup
, const int toProcessId
) const;
270 const OneToOneUniQueue
&oneToOneQueue(const Group fromGroup
, const int fromProcessId
, const Group toGroup
, const int toProcessId
) const;
271 int readerIndex(const Group group
, const int processId
) const;
272 Group
localGroup() const { return theLocalGroup
; }
273 Group
remoteGroup() const { return theLocalGroup
== groupA
? groupB
: groupA
; }
276 const Mem::Pointer
<Metadata
> metadata
; ///< shared metadata
277 const Mem::Pointer
<OneToOneUniQueues
> queues
; ///< unidirection one-to-one queues
278 const Mem::Pointer
<QueueReaders
> readers
; ///< readers array
280 const Group theLocalGroup
; ///< group of this queue
284 * Lockless fixed-capacity bidirectional queue for a limited number
285 * processes. Any process may send data to and receive from any other
286 * process (including itself). Each process has a unique integer ID in
287 * [processIdOffset, processIdOffset + processCount) range.
289 class MultiQueue
: public BaseMultiQueue
292 typedef OneToOneUniQueue::Full Full
;
293 typedef OneToOneUniQueue::ItemTooLarge ItemTooLarge
;
296 /// Shared metadata for MultiQueue
298 Metadata(const int aProcessCount
, const int aProcessIdOffset
);
299 size_t sharedMemorySize() const { return sizeof(*this); }
300 static size_t SharedMemorySize(const int, const int) { return sizeof(Metadata
); }
302 const int theProcessCount
;
303 const int theProcessIdOffset
;
310 Owner(const String
&id
, const int processCount
, const int processIdOffset
, const unsigned int maxItemSize
, const int capacity
);
314 Mem::Owner
<Metadata
> *const metadataOwner
;
315 Mem::Owner
<OneToOneUniQueues
> *const queuesOwner
;
316 Mem::Owner
<QueueReaders
> *const readersOwner
;
319 static Owner
*Init(const String
&id
, const int processCount
, const int processIdOffset
, const unsigned int maxItemSize
, const int capacity
);
321 MultiQueue(const String
&id
, const int localProcessId
);
324 virtual const OneToOneUniQueue
&inQueue(const int remoteProcessId
) const;
325 virtual const OneToOneUniQueue
&outQueue(const int remoteProcessId
) const;
326 virtual const QueueReader
&localReader() const;
327 virtual const QueueReader
&remoteReader(const int remoteProcessId
) const;
328 virtual int remotesCount() const;
329 virtual int remotesIdOffset() const;
332 bool validProcessId(const int processId
) const;
333 const OneToOneUniQueue
&oneToOneQueue(const int fromProcessId
, const int toProcessId
) const;
334 const QueueReader
&reader(const int processId
) const;
337 const Mem::Pointer
<Metadata
> metadata
; ///< shared metadata
338 const Mem::Pointer
<OneToOneUniQueues
> queues
; ///< unidirection one-to-one queues
339 const Mem::Pointer
<QueueReaders
> readers
; ///< readers array
344 template <class Value
>
346 OneToOneUniQueue::pop(Value
&value
, QueueReader
*const reader
)
348 if (sizeof(value
) > theMaxItemSize
)
349 throw ItemTooLarge();
351 // A writer might push between the empty test and block() below, so we do
352 // not return false right after calling block(), but test again.
358 // A writer might push between the empty test and block() below,
359 // so we must test again as such a writer will not signal us.
367 const unsigned int pos
= (theOut
++ % theCapacity
) * theMaxItemSize
;
368 memcpy(&value
, theBuffer
+ pos
, sizeof(value
));
374 template <class Value
>
376 OneToOneUniQueue::peek(Value
&value
) const
378 if (sizeof(value
) > theMaxItemSize
)
379 throw ItemTooLarge();
384 // the reader may pop() before we copy; making this method imprecise
385 const unsigned int pos
= (theOut
% theCapacity
) * theMaxItemSize
;
386 memcpy(&value
, theBuffer
+ pos
, sizeof(value
));
390 template <class Value
>
392 OneToOneUniQueue::push(const Value
&value
, QueueReader
*const reader
)
394 if (sizeof(value
) > theMaxItemSize
)
395 throw ItemTooLarge();
400 const unsigned int pos
= theIn
++ % theCapacity
* theMaxItemSize
;
401 memcpy(theBuffer
+ pos
, &value
, sizeof(value
));
402 const bool wasEmpty
= !theSize
++;
404 return wasEmpty
&& (!reader
|| reader
->raiseSignal());
409 inline OneToOneUniQueue
&
410 OneToOneUniQueues::operator [](const int index
)
412 return const_cast<OneToOneUniQueue
&>((*const_cast<const OneToOneUniQueues
*>(this))[index
]);
415 inline const OneToOneUniQueue
&
416 OneToOneUniQueues::front() const
418 const char *const queue
=
419 reinterpret_cast<const char *>(this) + sizeof(*this);
420 return *reinterpret_cast<const OneToOneUniQueue
*>(queue
);
425 template <class Value
>
427 BaseMultiQueue::pop(int &remoteProcessId
, Value
&value
)
429 // iterate all remote processes, starting after the one we visited last
430 for (int i
= 0; i
< remotesCount(); ++i
) {
431 if (++theLastPopProcessId
>= remotesIdOffset() + remotesCount())
432 theLastPopProcessId
= remotesIdOffset();
433 OneToOneUniQueue
&queue
= inQueue(theLastPopProcessId
);
434 if (queue
.pop(value
, &localReader())) {
435 remoteProcessId
= theLastPopProcessId
;
436 debugs(54, 7, HERE
<< "popped from " << remoteProcessId
<< " to " << theLocalProcessId
<< " at " << queue
.size());
440 return false; // no process had anything to pop
443 template <class Value
>
445 BaseMultiQueue::push(const int remoteProcessId
, const Value
&value
)
447 OneToOneUniQueue
&remoteQueue
= outQueue(remoteProcessId
);
448 QueueReader
&reader
= remoteReader(remoteProcessId
);
449 debugs(54, 7, HERE
<< "pushing from " << theLocalProcessId
<< " to " << remoteProcessId
<< " at " << remoteQueue
.size());
450 return remoteQueue
.push(value
, &reader
);
453 template <class Value
>
455 BaseMultiQueue::peek(int &remoteProcessId
, Value
&value
) const
457 // mimic FewToFewBiQueue::pop() but quit just before popping
458 int popProcessId
= theLastPopProcessId
; // preserve for future pop()
459 for (int i
= 0; i
< remotesCount(); ++i
) {
460 if (++popProcessId
>= remotesIdOffset() + remotesCount())
461 popProcessId
= remotesIdOffset();
462 const OneToOneUniQueue
&queue
= inQueue(popProcessId
);
463 if (queue
.peek(value
)) {
464 remoteProcessId
= popProcessId
;
468 return false; // most likely, no process had anything to pop
473 template <class Value
>
475 FewToFewBiQueue::findOldest(const int remoteProcessId
, Value
&value
) const
477 // we may be called before remote process configured its queue end
478 if (!validProcessId(remoteGroup(), remoteProcessId
))
481 // we need the oldest value, so start with the incoming, them-to-us queue:
482 const OneToOneUniQueue
&in
= inQueue(remoteProcessId
);
483 debugs(54, 2, HERE
<< "peeking from " << remoteProcessId
<< " to " <<
484 theLocalProcessId
<< " at " << in
.size());
488 // if the incoming queue is empty, check the outgoing, us-to-them queue:
489 const OneToOneUniQueue
&out
= outQueue(remoteProcessId
);
490 debugs(54, 2, HERE
<< "peeking from " << theLocalProcessId
<< " to " <<
491 remoteProcessId
<< " at " << out
.size());
492 return out
.peek(value
);
497 #endif // SQUID_IPC_QUEUE_H