4 #ifndef SQUID_IPC_QUEUE_H
5 #define SQUID_IPC_QUEUE_H
7 #include "base/InstanceId.h"
8 #include "base/Vector.h"
10 #include "ipc/AtomicWord.h"
11 #include "ipc/mem/FlexibleArray.h"
12 #include "ipc/mem/Pointer.h"
20 /// State of the reading end of a queue (i.e., of the code calling pop()).
21 /// Multiple queues attached to one reader share this state.
25 QueueReader(); // the initial state is "blocked without a signal"
27 /// whether the reader is waiting for a notification signal
28 bool blocked() const { return popBlocked
== 1; }
30 /// marks the reader as blocked, waiting for a notification signal
31 void block() { popBlocked
.swap_if(0, 1); }
33 /// removes the block() effects
34 void unblock() { popBlocked
.swap_if(1, 0); }
36 /// if reader is blocked and not notified, marks the notification signal
37 /// as sent and not received, returning true; otherwise, returns false
38 bool raiseSignal() { return blocked() && popSignal
.swap_if(0,1); }
40 /// marks sent reader notification as received (also removes pop blocking)
41 void clearSignal() { unblock(); popSignal
.swap_if(1,0); }
44 Atomic::Word popBlocked
; ///< whether the reader is blocked on pop()
45 Atomic::Word popSignal
; ///< whether writer has sent and reader has not received notification
48 typedef Atomic::Word Rate
; ///< pop()s per second
49 Rate rateLimit
; ///< pop()s per second limit if positive
51 // we need a signed atomic type because balance may get negative
52 typedef Atomic::WordT
<int> AtomicSignedMsec
;
53 typedef AtomicSignedMsec Balance
;
54 /// how far ahead the reader is compared to a perfect read/sec event rate
57 /// unique ID for debugging which reader is used (works across processes)
58 const InstanceId
<QueueReader
> id
;
61 /// shared array of QueueReaders
65 QueueReaders(const int aCapacity
);
66 size_t sharedMemorySize() const;
67 static size_t SharedMemorySize(const int capacity
);
69 const int theCapacity
; /// number of readers
70 Ipc::Mem::FlexibleArray
<QueueReader
> theReaders
; /// readers
74 * Lockless fixed-capacity queue for a single writer and a single reader.
76 * If the queue is empty, the reader is considered "blocked" and needs
77 * an out-of-band notification message to notice the next pushed item.
79 * Current implementation assumes that the writer cannot get blocked: if the
80 * queue is full, the writer will just not push and come back later (with a
81 * different value). We can add support for blocked writers if needed.
83 class OneToOneUniQueue
86 // pop() and push() exceptions; TODO: use TextException instead
88 class ItemTooLarge
{};
90 OneToOneUniQueue(const unsigned int aMaxItemSize
, const int aCapacity
);
92 unsigned int maxItemSize() const { return theMaxItemSize
; }
93 int size() const { return theSize
; }
94 int capacity() const { return theCapacity
; }
95 int sharedMemorySize() const { return Items2Bytes(theMaxItemSize
, theCapacity
); }
97 bool empty() const { return !theSize
; }
98 bool full() const { return theSize
== theCapacity
; }
100 static int Bytes2Items(const unsigned int maxItemSize
, int size
);
101 static int Items2Bytes(const unsigned int maxItemSize
, const int size
);
103 /// returns true iff the value was set; [un]blocks the reader as needed
104 template<class Value
> bool pop(Value
&value
, QueueReader
*const reader
= NULL
);
106 /// returns true iff the caller must notify the reader of the pushed item
107 template<class Value
> bool push(const Value
&value
, QueueReader
*const reader
= NULL
);
109 /// returns true iff the value was set; the value may be stale!
110 template<class Value
> bool peek(Value
&value
) const;
114 unsigned int theIn
; ///< input index, used only in push()
115 unsigned int theOut
; ///< output index, used only in pop()
117 Atomic::Word theSize
; ///< number of items in the queue
118 const unsigned int theMaxItemSize
; ///< maximum item size
119 const int theCapacity
; ///< maximum number of items, i.e. theBuffer size
124 /// shared array of OneToOneUniQueues
125 class OneToOneUniQueues
128 OneToOneUniQueues(const int aCapacity
, const unsigned int maxItemSize
, const int queueCapacity
);
130 size_t sharedMemorySize() const;
131 static size_t SharedMemorySize(const int capacity
, const unsigned int maxItemSize
, const int queueCapacity
);
133 const OneToOneUniQueue
&operator [](const int index
) const;
134 inline OneToOneUniQueue
&operator [](const int index
);
137 inline const OneToOneUniQueue
&front() const;
140 const int theCapacity
; /// number of OneToOneUniQueues
144 * Base class for lockless fixed-capacity bidirectional queues for a
145 * limited number processes.
150 BaseMultiQueue(const int aLocalProcessId
);
151 virtual ~BaseMultiQueue() {}
153 /// clears the reader notification received by the local process from the remote process
154 void clearReaderSignal(const int remoteProcessId
);
156 /// picks a process and calls OneToOneUniQueue::pop() using its queue
157 template <class Value
> bool pop(int &remoteProcessId
, Value
&value
);
159 /// calls OneToOneUniQueue::push() using the given process queue
160 template <class Value
> bool push(const int remoteProcessId
, const Value
&value
);
162 /// peeks at the item likely to be pop()ed next
163 template<class Value
> bool peek(int &remoteProcessId
, Value
&value
) const;
165 /// returns local reader's balance
166 QueueReader::Balance
&localBalance() { return localReader().balance
; }
168 /// returns reader's balance for a given remote process
169 const QueueReader::Balance
&balance(const int remoteProcessId
) const;
171 /// returns local reader's rate limit
172 QueueReader::Rate
&localRateLimit() { return localReader().rateLimit
; }
174 /// returns reader's rate limit for a given remote process
175 const QueueReader::Rate
&rateLimit(const int remoteProcessId
) const;
177 /// number of items in incoming queue from a given remote process
178 int inSize(const int remoteProcessId
) const { return inQueue(remoteProcessId
).size(); }
180 /// number of items in outgoing queue to a given remote process
181 int outSize(const int remoteProcessId
) const { return outQueue(remoteProcessId
).size(); }
184 /// incoming queue from a given remote process
185 virtual const OneToOneUniQueue
&inQueue(const int remoteProcessId
) const = 0;
186 OneToOneUniQueue
&inQueue(const int remoteProcessId
);
188 /// outgoing queue to a given remote process
189 virtual const OneToOneUniQueue
&outQueue(const int remoteProcessId
) const = 0;
190 OneToOneUniQueue
&outQueue(const int remoteProcessId
);
192 virtual const QueueReader
&localReader() const = 0;
193 QueueReader
&localReader();
195 virtual const QueueReader
&remoteReader(const int remoteProcessId
) const = 0;
196 QueueReader
&remoteReader(const int remoteProcessId
);
198 virtual int remotesCount() const = 0;
199 virtual int remotesIdOffset() const = 0;
202 const int theLocalProcessId
; ///< process ID of this queue
205 int theLastPopProcessId
; ///< the ID of the last process we tried to pop() from
209 * Lockless fixed-capacity bidirectional queue for a limited number
210 * processes. Allows communication between two groups of processes:
211 * any process in one group may send data to and receive from any
212 * process in another group, but processes in the same group can not
213 * communicate. Process in each group has a unique integer ID in
214 * [groupIdOffset, groupIdOffset + groupSize) range.
216 class FewToFewBiQueue
: public BaseMultiQueue
219 typedef OneToOneUniQueue::Full Full
;
220 typedef OneToOneUniQueue::ItemTooLarge ItemTooLarge
;
223 /// Shared metadata for FewToFewBiQueue
225 Metadata(const int aGroupASize
, const int aGroupAIdOffset
, const int aGroupBSize
, const int aGroupBIdOffset
);
226 size_t sharedMemorySize() const { return sizeof(*this); }
227 static size_t SharedMemorySize(const int, const int, const int, const int) { return sizeof(Metadata
); }
229 const int theGroupASize
;
230 const int theGroupAIdOffset
;
231 const int theGroupBSize
;
232 const int theGroupBIdOffset
;
239 Owner(const String
&id
, const int groupASize
, const int groupAIdOffset
, const int groupBSize
, const int groupBIdOffset
, const unsigned int maxItemSize
, const int capacity
);
243 Mem::Owner
<Metadata
> *const metadataOwner
;
244 Mem::Owner
<OneToOneUniQueues
> *const queuesOwner
;
245 Mem::Owner
<QueueReaders
> *const readersOwner
;
248 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
);
250 enum Group
{ groupA
= 0, groupB
= 1 };
251 FewToFewBiQueue(const String
&id
, const Group aLocalGroup
, const int aLocalProcessId
);
253 /// maximum number of items in the queue
254 static int MaxItemsCount(const int groupASize
, const int groupBSize
, const int capacity
);
256 /// finds the oldest item in incoming and outgoing queues between
257 /// us and the given remote process
258 template<class Value
> bool findOldest(const int remoteProcessId
, Value
&value
) const;
261 virtual const OneToOneUniQueue
&inQueue(const int remoteProcessId
) const;
262 virtual const OneToOneUniQueue
&outQueue(const int remoteProcessId
) const;
263 virtual const QueueReader
&localReader() const;
264 virtual const QueueReader
&remoteReader(const int processId
) const;
265 virtual int remotesCount() const;
266 virtual int remotesIdOffset() const;
269 bool validProcessId(const Group group
, const int processId
) const;
270 int oneToOneQueueIndex(const Group fromGroup
, const int fromProcessId
, const Group toGroup
, const int toProcessId
) const;
271 const OneToOneUniQueue
&oneToOneQueue(const Group fromGroup
, const int fromProcessId
, const Group toGroup
, const int toProcessId
) const;
272 int readerIndex(const Group group
, const int processId
) const;
273 Group
localGroup() const { return theLocalGroup
; }
274 Group
remoteGroup() const { return theLocalGroup
== groupA
? groupB
: groupA
; }
277 const Mem::Pointer
<Metadata
> metadata
; ///< shared metadata
278 const Mem::Pointer
<OneToOneUniQueues
> queues
; ///< unidirection one-to-one queues
279 const Mem::Pointer
<QueueReaders
> readers
; ///< readers array
281 const Group theLocalGroup
; ///< group of this queue
285 * Lockless fixed-capacity bidirectional queue for a limited number
286 * processes. Any process may send data to and receive from any other
287 * process (including itself). Each process has a unique integer ID in
288 * [processIdOffset, processIdOffset + processCount) range.
290 class MultiQueue
: public BaseMultiQueue
293 typedef OneToOneUniQueue::Full Full
;
294 typedef OneToOneUniQueue::ItemTooLarge ItemTooLarge
;
297 /// Shared metadata for MultiQueue
299 Metadata(const int aProcessCount
, const int aProcessIdOffset
);
300 size_t sharedMemorySize() const { return sizeof(*this); }
301 static size_t SharedMemorySize(const int, const int) { return sizeof(Metadata
); }
303 const int theProcessCount
;
304 const int theProcessIdOffset
;
311 Owner(const String
&id
, const int processCount
, const int processIdOffset
, const unsigned int maxItemSize
, const int capacity
);
315 Mem::Owner
<Metadata
> *const metadataOwner
;
316 Mem::Owner
<OneToOneUniQueues
> *const queuesOwner
;
317 Mem::Owner
<QueueReaders
> *const readersOwner
;
320 static Owner
*Init(const String
&id
, const int processCount
, const int processIdOffset
, const unsigned int maxItemSize
, const int capacity
);
322 MultiQueue(const String
&id
, const int localProcessId
);
325 virtual const OneToOneUniQueue
&inQueue(const int remoteProcessId
) const;
326 virtual const OneToOneUniQueue
&outQueue(const int remoteProcessId
) const;
327 virtual const QueueReader
&localReader() const;
328 virtual const QueueReader
&remoteReader(const int remoteProcessId
) const;
329 virtual int remotesCount() const;
330 virtual int remotesIdOffset() const;
333 bool validProcessId(const int processId
) const;
334 const OneToOneUniQueue
&oneToOneQueue(const int fromProcessId
, const int toProcessId
) const;
335 const QueueReader
&reader(const int processId
) const;
338 const Mem::Pointer
<Metadata
> metadata
; ///< shared metadata
339 const Mem::Pointer
<OneToOneUniQueues
> queues
; ///< unidirection one-to-one queues
340 const Mem::Pointer
<QueueReaders
> readers
; ///< readers array
345 template <class Value
>
347 OneToOneUniQueue::pop(Value
&value
, QueueReader
*const reader
)
349 if (sizeof(value
) > theMaxItemSize
)
350 throw ItemTooLarge();
352 // A writer might push between the empty test and block() below, so we do
353 // not return false right after calling block(), but test again.
359 // A writer might push between the empty test and block() below,
360 // so we must test again as such a writer will not signal us.
368 const unsigned int pos
= (theOut
++ % theCapacity
) * theMaxItemSize
;
369 memcpy(&value
, theBuffer
+ pos
, sizeof(value
));
375 template <class Value
>
377 OneToOneUniQueue::peek(Value
&value
) const
379 if (sizeof(value
) > theMaxItemSize
)
380 throw ItemTooLarge();
385 // the reader may pop() before we copy; making this method imprecise
386 const unsigned int pos
= (theOut
% theCapacity
) * theMaxItemSize
;
387 memcpy(&value
, theBuffer
+ pos
, sizeof(value
));
391 template <class Value
>
393 OneToOneUniQueue::push(const Value
&value
, QueueReader
*const reader
)
395 if (sizeof(value
) > theMaxItemSize
)
396 throw ItemTooLarge();
401 const unsigned int pos
= theIn
++ % theCapacity
* theMaxItemSize
;
402 memcpy(theBuffer
+ pos
, &value
, sizeof(value
));
403 const bool wasEmpty
= !theSize
++;
405 return wasEmpty
&& (!reader
|| reader
->raiseSignal());
410 inline OneToOneUniQueue
&
411 OneToOneUniQueues::operator [](const int index
)
413 return const_cast<OneToOneUniQueue
&>((*const_cast<const OneToOneUniQueues
*>(this))[index
]);
416 inline const OneToOneUniQueue
&
417 OneToOneUniQueues::front() const
419 const char *const queue
=
420 reinterpret_cast<const char *>(this) + sizeof(*this);
421 return *reinterpret_cast<const OneToOneUniQueue
*>(queue
);
426 template <class Value
>
428 BaseMultiQueue::pop(int &remoteProcessId
, Value
&value
)
430 // iterate all remote processes, starting after the one we visited last
431 for (int i
= 0; i
< remotesCount(); ++i
) {
432 if (++theLastPopProcessId
>= remotesIdOffset() + remotesCount())
433 theLastPopProcessId
= remotesIdOffset();
434 OneToOneUniQueue
&queue
= inQueue(theLastPopProcessId
);
435 if (queue
.pop(value
, &localReader())) {
436 remoteProcessId
= theLastPopProcessId
;
437 debugs(54, 7, HERE
<< "popped from " << remoteProcessId
<< " to " << theLocalProcessId
<< " at " << queue
.size());
441 return false; // no process had anything to pop
444 template <class Value
>
446 BaseMultiQueue::push(const int remoteProcessId
, const Value
&value
)
448 OneToOneUniQueue
&remoteQueue
= outQueue(remoteProcessId
);
449 QueueReader
&reader
= remoteReader(remoteProcessId
);
450 debugs(54, 7, HERE
<< "pushing from " << theLocalProcessId
<< " to " << remoteProcessId
<< " at " << remoteQueue
.size());
451 return remoteQueue
.push(value
, &reader
);
454 template <class Value
>
456 BaseMultiQueue::peek(int &remoteProcessId
, Value
&value
) const
458 // mimic FewToFewBiQueue::pop() but quit just before popping
459 int popProcessId
= theLastPopProcessId
; // preserve for future pop()
460 for (int i
= 0; i
< remotesCount(); ++i
) {
461 if (++popProcessId
>= remotesIdOffset() + remotesCount())
462 popProcessId
= remotesIdOffset();
463 const OneToOneUniQueue
&queue
= inQueue(popProcessId
);
464 if (queue
.peek(value
)) {
465 remoteProcessId
= popProcessId
;
469 return false; // most likely, no process had anything to pop
474 template <class Value
>
476 FewToFewBiQueue::findOldest(const int remoteProcessId
, Value
&value
) const
478 // we may be called before remote process configured its queue end
479 if (!validProcessId(remoteGroup(), remoteProcessId
))
482 // we need the oldest value, so start with the incoming, them-to-us queue:
483 const OneToOneUniQueue
&in
= inQueue(remoteProcessId
);
484 debugs(54, 2, HERE
<< "peeking from " << remoteProcessId
<< " to " <<
485 theLocalProcessId
<< " at " << in
.size());
489 // if the incoming queue is empty, check the outgoing, us-to-them queue:
490 const OneToOneUniQueue
&out
= outQueue(remoteProcessId
);
491 debugs(54, 2, HERE
<< "peeking from " << theLocalProcessId
<< " to " <<
492 remoteProcessId
<< " at " << out
.size());
493 return out
.peek(value
);
498 #endif // SQUID_IPC_QUEUE_H