Core Library 1.7.0.0 Library containing core utilities and tools for threading, networking, logging, INI and CSV file management etc.

ConcurrentQueue.h
Go to the documentation of this file.
1 // This file is part of CoreLibrary containing useful reusable utility
2 // classes.
3 //
4 // Copyright (C) 2014 to present, Duncan Crutchley
5 // Contact <dac1976github@outlook.com>
6 //
7 // This program is free software: you can redistribute it and/or modify
8 // it under the terms of the GNU Lesser General Public License as published
9 // by the Free Software Foundation, either version 3 of the License, or
10 // (at your option) any later version.
11 //
12 // This program is distributed in the hope that it will be useful,
13 // but WITHOUT ANY WARRANTY; without even the implied warranty of
14 // MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
15 // GNU General Public License and GNU Lesser General Public License
16 // for more details.
17 //
18 // You should have received a copy of the GNU General Public License
19 // and GNU Lesser General Public License along with this program. If
20 // not, see <http://www.gnu.org/licenses/>.
21
27 #ifndef CONCURRENTQUEUE
28 #define CONCURRENTQUEUE
29
30 #include <deque>
31 #include <algorithm>
32 #include <utility>
33 #include <stdexcept>
34 #include <boost/throw_exception.hpp>
35 #include "SyncEvent.h"
36
38 namespace core_lib
39 {
41 namespace threads
42 {
43
49 template <typename P> struct SingleItemDeleter
50 {
55 void operator()(P* p) const
56 {
57 delete p;
58 }
59 };
60
66 template <typename P> struct ArrayDeleter
67 {
72 void operator()(P* p) const
73 {
74 delete[] p;
75 }
76 };
77
93 template <typename T> class ConcurrentQueue final
94 {
95 public:
99 ConcurrentQueue() = default;
101 ConcurrentQueue(const ConcurrentQueue&) = delete;
103 ConcurrentQueue& operator=(const ConcurrentQueue&) = delete;
105 ConcurrentQueue(ConcurrentQueue&&) = delete;
107 ConcurrentQueue& operator=(ConcurrentQueue&&) = delete;
116 ~ConcurrentQueue() = default;
121 size_t Size() const
122 {
123 std::lock_guard<std::mutex> lock{m_mutex};
124 return m_queue.size();
125 }
130 bool Empty() const
131 {
132 std::lock_guard<std::mutex> lock{m_mutex};
133 return m_queue.empty();
134 }
147 void Push(T&& item)
148 {
149 {
150 std::lock_guard<std::mutex> lock{m_mutex};
151 m_queue.emplace_back(std::move(item));
152 }
153
154 m_itemEvent.Signal();
155 }
167 void Push(const T& item)
168 {
169 {
170 std::lock_guard<std::mutex> lock{m_mutex};
171 m_queue.emplace_back(item);
172 }
173
174 m_itemEvent.Signal();
175 }
182 void BreakPopWait()
183 {
184 m_itemEvent.Signal();
185 }
194 bool Pop(T& item)
195 {
196 m_itemEvent.Wait();
197 return PopNow(item);
198 }
209 void PopThrow(T& item)
210 {
211 m_itemEvent.Wait();
212
213 if (!PopNow(item))
214 {
215 BOOST_THROW_EXCEPTION(std::runtime_error("no item to pop"));
216 }
217 }
223 bool TryPop(T& item)
224 {
225 return PopNow(item);
226 }
234 void TryPopThrow(T& item)
235 {
236 if (!PopNow(item))
237 {
238 BOOST_THROW_EXCEPTION(std::runtime_error("no item to pop"));
239 }
240 }
247 bool TimedPop(unsigned int timeoutMilliseconds, T& item)
248 {
249 bool popSuccess{false};
250
251 if (m_itemEvent.WaitForTime(timeoutMilliseconds))
252 {
253 popSuccess = PopNow(item);
254 }
255
256 return popSuccess;
257 }
266 void TimedPopThrow(unsigned int timeoutMilliseconds, T& item)
267 {
268 if (!m_itemEvent.WaitForTime(timeoutMilliseconds))
269 {
270 BOOST_THROW_EXCEPTION(std::runtime_error("item event timed out"));
271 }
272
273 if (!PopNow(item))
274 {
275 BOOST_THROW_EXCEPTION(std::runtime_error("no item to pop"));
276 }
277 }
283 bool TrySteal(T& item)
284 {
285 return PopNow(item, eQueueEnd::back);
286 }
294 void TryStealThrow(T& item)
295 {
296 if (!PopNow(item, eQueueEnd::back))
297 {
298 BOOST_THROW_EXCEPTION(std::runtime_error("no item to pop"));
299 }
300 }
311 const T* Peek(size_t index) const
312 {
313 const T* pItem{};
314 std::lock_guard<std::mutex> lock{m_mutex};
315
316 if (!m_queue.empty() && (index < m_queue.size()))
317 {
318 pItem = &m_queue[index];
319 }
320
321 return pItem;
322 }
334 void Clear()
335 {
336 std::lock_guard<std::mutex> lock{m_mutex};
337 m_queue.clear();
338 }
355 template <typename F> void Clear(F deleter)
356 {
357 std::lock_guard<std::mutex> lock{m_mutex};
358
359 for (auto& qi : m_queue)
360 {
361 deleter(qi);
362 }
363
364 m_queue.clear();
365 }
367 using container_type = std::deque<T>;
379 container_type TakeAll()
380 {
381 container_type q{};
382
383 {
384 std::lock_guard<std::mutex> lock{m_mutex};
385 q.swap(m_queue);
386 }
387
388 return q;
389 }
390
391 private:
393 enum class eQueueEnd
394 {
396 front,
398 back
399 };
406 bool PopNow(T& item, eQueueEnd whichEnd = eQueueEnd::front)
407 {
408 std::lock_guard<std::mutex> lock{m_mutex};
409 auto isEmpty = m_queue.empty();
410
411 if (!isEmpty)
412 {
413 if (whichEnd == eQueueEnd::front)
414 {
415 PopFront(item);
416 }
417 else
418 {
419 PopBack(item);
420 }
421 }
422
423 if (m_queue.empty())
424 {
425 m_itemEvent.Reset();
426 }
427
428 return !isEmpty;
429 }
430
435 void PopFront(T& item)
436 {
437 item = std::move(m_queue.front());
438 m_queue.pop_front();
439 }
440
445 void PopBack(T& item)
446 {
447 item = std::move(m_queue.back());
448 m_queue.pop_back();
449 }
450
451 private:
453 mutable std::mutex m_mutex;
455 SyncEvent m_itemEvent{
456 eNotifyType::signalOneThread, eResetCondition::manualReset, eIntialCondition::notSignalled};
458 container_type m_queue{};
459 };
460
461 } // namespace threads
462 } // namespace core_lib
463
464 #endif // CONCURRENTQUEUE
core_lib::threads::ConcurrentQueue< MessageType >::eQueueEnd
eQueueEnd
Enumeration controlling end of queue to pop from.
Definition: ConcurrentQueue.h:393
core_lib::threads::ConcurrentQueue::TimedPop
bool TimedPop(unsigned int timeoutMilliseconds, T &item)
Pop an item off the queue but only wait for a given amount of time.
Definition: ConcurrentQueue.h:247
core_lib::threads::ConcurrentQueue::Push
void Push(T &&item)
Push an item onto the queue.
Definition: ConcurrentQueue.h:147
core_lib::threads::ConcurrentQueue::Peek
const T * Peek(size_t index) const
Take a peek at an item at a given index on the queue.
Definition: ConcurrentQueue.h:311
core_lib::threads::ConcurrentQueue::TimedPopThrow
void TimedPopThrow(unsigned int timeoutMilliseconds, T &item)
Pop an item off the queue but only wait for a given amount of time.
Definition: ConcurrentQueue.h:266
core_lib::threads::ConcurrentQueue::Size
size_t Size() const
Size of the queue.
Definition: ConcurrentQueue.h:121
core_lib::threads::ConcurrentQueue::Pop
bool Pop(T &item)
Pop an item off the queue if there are any else wait.
Definition: ConcurrentQueue.h:194
core_lib::threads::SyncEvent
Class defining a thread synchronisation event.
Definition: SyncEvent.h:70
core_lib::threads::ConcurrentQueue::PopThrow
void PopThrow(T &item)
Pop an item off the queue if there are any else wait.
Definition: ConcurrentQueue.h:209
core_lib::threads::ConcurrentQueue::TakeAll
container_type TakeAll()
Take all items from the queue and return them, thus clearing down the internal queue.
Definition: ConcurrentQueue.h:379
core_lib::threads::ConcurrentQueue::BreakPopWait
void BreakPopWait()
Break out of waiting on a Pop method.
Definition: ConcurrentQueue.h:182
core_lib::threads::ConcurrentQueue::Clear
void Clear()
Clear the queue.
Definition: ConcurrentQueue.h:334
core_lib::threads::ConcurrentQueue::TryPopThrow
void TryPopThrow(T &item)
Pop an item off the queue.
Definition: ConcurrentQueue.h:234
core_lib
The core_lib namespace.
Definition: AsioDefines.h:59
core_lib::threads::ConcurrentQueue::m_mutex
std::mutex m_mutex
Synchronization mutex.
Definition: ConcurrentQueue.h:453
core_lib::threads::ConcurrentQueue::Clear
void Clear(F deleter)
Clear the queue.
Definition: ConcurrentQueue.h:355
core_lib::threads::SingleItemDeleter::operator()
void operator()(P *p) const
Function operator.
Definition: ConcurrentQueue.h:55
core_lib::threads::ConcurrentQueue::TrySteal
bool TrySteal(T &item)
Steal an item from the back of the queue if there are any else return.
Definition: ConcurrentQueue.h:283
core_lib::threads::ConcurrentQueue::PopNow
bool PopNow(T &item, eQueueEnd whichEnd=eQueueEnd::front)
Pop an item off the queue.
Definition: ConcurrentQueue.h:406
core_lib::threads::ConcurrentQueue::PopBack
void PopBack(T &item)
Pop an item off the back of the queue.
Definition: ConcurrentQueue.h:445
core_lib::threads::ArrayDeleter::operator()
void operator()(P *p) const
Function operator.
Definition: ConcurrentQueue.h:72
core_lib::threads::ConcurrentQueue::Empty
bool Empty() const
Is the queue empty.
Definition: ConcurrentQueue.h:130
SyncEvent.h
File containing declaration of SyncEvent class.
core_lib::threads::ConcurrentQueue::PopFront
void PopFront(T &item)
Pop an item off the front of the queue.
Definition: ConcurrentQueue.h:435
core_lib::threads::ConcurrentQueue::Push
void Push(const T &item)
Push an item onto the queue.
Definition: ConcurrentQueue.h:167
core_lib::threads::SingleItemDeleter
Single item deleter for queue item.
Definition: ConcurrentQueue.h:49
core_lib::threads::ConcurrentQueue
Class defining a concurrent queue.
Definition: ConcurrentQueue.h:93
core_lib::threads::ConcurrentQueue< MessageType >::container_type
std::deque< MessageType > container_type
Typedef for container type.
Definition: ConcurrentQueue.h:367
core_lib::threads::ArrayDeleter
Array deleter for queue item.
Definition: ConcurrentQueue.h:66
core_lib::threads::ConcurrentQueue::TryStealThrow
void TryStealThrow(T &item)
Steal an item from the back of the queue.
Definition: ConcurrentQueue.h:294
core_lib::threads::ConcurrentQueue::TryPop
bool TryPop(T &item)
Pop an item off the queue if there are any else return.
Definition: ConcurrentQueue.h:223