mirror of
https://github.com/intrepidcs/libicsneo.git
synced 2026-08-05 09:28:40 +02:00
Initial commit
This commit is contained in:
Paul Hollinsky
@@ -0,0 +1,26 @@
|
||||
*.ipch
|
||||
*.suo
|
||||
*.user
|
||||
*.sdf
|
||||
*.opensdf
|
||||
*.exe
|
||||
*.pdb
|
||||
*.vs
|
||||
*.VC.db
|
||||
build/bin/
|
||||
build/*.log
|
||||
build/msvc14/*.log
|
||||
build/msvc14/obj/
|
||||
build/msvc12/*.log
|
||||
build/msvc12/obj/
|
||||
build/msvc11/*.log
|
||||
build/msvc11/obj/
|
||||
build/xcode/build/
|
||||
tests/fuzztests/fuzztests.log
|
||||
benchmarks/benchmarks.log
|
||||
tests/CDSChecker/*.o
|
||||
tests/CDSChecker/*.log
|
||||
tests/CDSChecker/model-checker/
|
||||
tests/relacy/freelist.exe
|
||||
tests/relacy/spmchash.exe
|
||||
tests/relacy/log.txt
|
||||
+61
@@ -0,0 +1,61 @@
|
||||
This license file applies to everything in this repository except that which
|
||||
is explicitly annotated as being written by other authors, i.e. the Boost
|
||||
queue (included in the benchmarks for comparison), Intel's TBB library (ditto),
|
||||
the CDSChecker tool (used for verification), the Relacy model checker (ditto),
|
||||
and Jeff Preshing's semaphore implementation (used in the blocking queue) which
|
||||
has a zlib license (embedded in blockingconcurrentqueue.h).
|
||||
|
||||
---
|
||||
|
||||
Simplified BSD License:
|
||||
|
||||
Copyright (c) 2013-2016, Cameron Desrochers.
|
||||
All rights reserved.
|
||||
|
||||
Redistribution and use in source and binary forms, with or without modification,
|
||||
are permitted provided that the following conditions are met:
|
||||
|
||||
- Redistributions of source code must retain the above copyright notice, this list of
|
||||
conditions and the following disclaimer.
|
||||
- Redistributions in binary form must reproduce the above copyright notice, this list of
|
||||
conditions and the following disclaimer in the documentation and/or other materials
|
||||
provided with the distribution.
|
||||
|
||||
THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" AND ANY
|
||||
EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES OF
|
||||
MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL
|
||||
THE COPYRIGHT HOLDER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
|
||||
SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT
|
||||
OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION)
|
||||
HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR
|
||||
TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE,
|
||||
EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
|
||||
|
||||
---
|
||||
|
||||
I have also chosen to dual-license under the Boost Software License as an alternative to
|
||||
the Simplified BSD license above:
|
||||
|
||||
Boost Software License - Version 1.0 - August 17th, 2003
|
||||
|
||||
Permission is hereby granted, free of charge, to any person or organization
|
||||
obtaining a copy of the software and accompanying documentation covered by
|
||||
this license (the "Software") to use, reproduce, display, distribute,
|
||||
execute, and transmit the Software, and to prepare derivative works of the
|
||||
Software, and to permit third-parties to whom the Software is furnished to
|
||||
do so, all subject to the following:
|
||||
|
||||
The copyright notices in the Software and this entire statement, including
|
||||
the above license grant, this restriction and the following disclaimer,
|
||||
must be included in all copies of the Software, in whole or in part, and
|
||||
all derivative works of the Software, unless such copies or derivative
|
||||
works are solely in the form of machine-executable object code generated by
|
||||
a source language processor.
|
||||
|
||||
THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
|
||||
IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
|
||||
FITNESS FOR A PARTICULAR PURPOSE, TITLE AND NON-INFRINGEMENT. IN NO EVENT
|
||||
SHALL THE COPYRIGHT HOLDERS OR ANYONE DISTRIBUTING THE SOFTWARE BE LIABLE
|
||||
FOR ANY DAMAGES OR OTHER LIABILITY, WHETHER IN CONTRACT, TORT OR OTHERWISE,
|
||||
ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER
|
||||
DEALINGS IN THE SOFTWARE.
|
||||
+486
@@ -0,0 +1,486 @@
|
||||
# moodycamel::ConcurrentQueue<T>
|
||||
|
||||
An industrial-strength lock-free queue for C++.
|
||||
|
||||
Note: If all you need is a single-producer, single-consumer queue, I have [one of those too][spsc].
|
||||
|
||||
## Features
|
||||
|
||||
- Knock-your-socks-off [blazing fast performance][benchmarks].
|
||||
- Single-header implementation. Just drop it in your project.
|
||||
- Fully thread-safe lock-free queue. Use concurrently from any number of threads.
|
||||
- C++11 implementation -- elements are moved (instead of copied) where possible.
|
||||
- Templated, obviating the need to deal exclusively with pointers -- memory is managed for you.
|
||||
- No artificial limitations on element types or maximum count.
|
||||
- Memory can be allocated once up-front, or dynamically as needed.
|
||||
- Fully portable (no assembly; all is done through standard C++11 primitives).
|
||||
- Supports super-fast bulk operations.
|
||||
- Includes a low-overhead blocking version (BlockingConcurrentQueue).
|
||||
- Exception safe.
|
||||
|
||||
## Reasons to use
|
||||
|
||||
There are not that many full-fledged lock-free queues for C++. Boost has one, but it's limited to objects with trivial
|
||||
assignment operators and trivial destructors, for example.
|
||||
Intel's TBB queue isn't lock-free, and requires trivial constructors too.
|
||||
There're many academic papers that implement lock-free queues in C++, but usable source code is
|
||||
hard to find, and tests even more so.
|
||||
|
||||
This queue not only has less limitations than others (for the most part), but [it's also faster][benchmarks].
|
||||
It's been fairly well-tested, and offers advanced features like **bulk enqueueing/dequeueing**
|
||||
(which, with my new design, is much faster than one element at a time, approaching and even surpassing
|
||||
the speed of a non-concurrent queue even under heavy contention).
|
||||
|
||||
In short, there was a lock-free queue shaped hole in the C++ open-source universe, and I set out
|
||||
to fill it with the fastest, most complete, and well-tested design and implementation I could.
|
||||
The result is `moodycamel::ConcurrentQueue` :-)
|
||||
|
||||
## Reasons *not* to use
|
||||
|
||||
The fastest synchronization of all is the kind that never takes place. Fundamentally,
|
||||
concurrent data structures require some synchronization, and that takes time. Every effort
|
||||
was made, of course, to minimize the overhead, but if you can avoid sharing data between
|
||||
threads, do so!
|
||||
|
||||
Why use concurrent data structures at all, then? Because they're gosh darn convenient! (And, indeed,
|
||||
sometimes sharing data concurrently is unavoidable.)
|
||||
|
||||
My queue is **not linearizable** (see the next section on high-level design). The foundations of
|
||||
its design assume that producers are independent; if this is not the case, and your producers
|
||||
co-ordinate amongst themselves in some fashion, be aware that the elements won't necessarily
|
||||
come out of the queue in the same order they were put in *relative to the ordering formed by that co-ordination*
|
||||
(but they will still come out in the order they were put in by any *individual* producer). If this affects
|
||||
your use case, you may be better off with another implementation; either way, it's an important limitation
|
||||
to be aware of.
|
||||
|
||||
My queue is also **not NUMA aware**, and does a lot of memory re-use internally, meaning it probably doesn't
|
||||
scale particularly well on NUMA architectures; however, I don't know of any other lock-free queue that *is*
|
||||
NUMA aware (except for [SALSA][salsa], which is very cool, but has no publicly available implementation that I know of).
|
||||
|
||||
Finally, the queue is **not sequentially consistent**; there *is* a happens-before relationship between when an element is put
|
||||
in the queue and when it comes out, but other things (such as pumping the queue until it's empty) require more thought
|
||||
to get right in all eventualities, because explicit memory ordering may have to be done to get the desired effect. In other words,
|
||||
it can sometimes be difficult to use the queue correctly. This is why it's a good idea to follow the [samples][samples.md] where possible.
|
||||
On the other hand, the upside of this lack of sequential consistency is better performance.
|
||||
|
||||
## High-level design
|
||||
|
||||
Elements are stored internally using contiguous blocks instead of linked lists for better performance.
|
||||
The queue is made up of a collection of sub-queues, one for each producer. When a consumer
|
||||
wants to dequeue an element, it checks all the sub-queues until it finds one that's not empty.
|
||||
All of this is largely transparent to the user of the queue, however -- it mostly just works<sup>TM</sup>.
|
||||
|
||||
One particular consequence of this design, however, (which seems to be non-intuitive) is that if two producers
|
||||
enqueue at the same time, there is no defined ordering between the elements when they're later dequeued.
|
||||
Normally this is fine, because even with a fully linearizable queue there'd be a race between the producer
|
||||
threads and so you couldn't rely on the ordering anyway. However, if for some reason you do extra explicit synchronization
|
||||
between the two producer threads yourself, thus defining a total order between enqueue operations, you might expect
|
||||
that the elements would come out in the same total order, which is a guarantee my queue does not offer. At that
|
||||
point, though, there semantically aren't really two separate producers, but rather one that happens to be spread
|
||||
across multiple threads. In this case, you can still establish a total ordering with my queue by creating
|
||||
a single producer token, and using that from both threads to enqueue (taking care to synchronize access to the token,
|
||||
of course, but there was already extra synchronization involved anyway).
|
||||
|
||||
I've written a more detailed [overview of the internal design][blog], as well as [the full
|
||||
nitty-gritty details of the design][design], on my blog. Finally, the
|
||||
[source][source] itself is available for perusal for those interested in its implementation.
|
||||
|
||||
## Basic use
|
||||
|
||||
The entire queue's implementation is contained in **one header**, [`concurrentqueue.h`][concurrentqueue.h].
|
||||
Simply download and include that to use the queue. The blocking version is in a separate header,
|
||||
[`blockingconcurrentqueue.h`][blockingconcurrentqueue.h], that depends on the first.
|
||||
The implementation makes use of certain key C++11 features, so it requires a fairly recent compiler
|
||||
(e.g. VS2012+ or g++ 4.8; note that g++ 4.6 has a known bug with `std::atomic` and is thus not supported).
|
||||
The algorithm implementations themselves are platform independent.
|
||||
|
||||
Use it like you would any other templated queue, with the exception that you can use
|
||||
it from many threads at once :-)
|
||||
|
||||
Simple example:
|
||||
|
||||
#include "concurrentqueue.h"
|
||||
|
||||
moodycamel::ConcurrentQueue<int> q;
|
||||
q.enqueue(25);
|
||||
|
||||
int item;
|
||||
bool found = q.try_dequeue(item);
|
||||
assert(found && item == 25);
|
||||
|
||||
Description of basic methods:
|
||||
- `ConcurrentQueue(size_t initialSizeEstimate)`
|
||||
Constructor which optionally accepts an estimate of the number of elements the queue will hold
|
||||
- `enqueue(T&& item)`
|
||||
Enqueues one item, allocating extra space if necessary
|
||||
- `try_enqueue(T&& item)`
|
||||
Enqueues one item, but only if enough memory is already allocated
|
||||
- `try_dequeue(T& item)`
|
||||
Dequeues one item, returning true if an item was found or false if the queue appeared empty
|
||||
|
||||
Note that it is up to the user to ensure that the queue object is completely constructed before
|
||||
being used by any other threads (this includes making the memory effects of construction
|
||||
visible, possibly via a memory barrier). Similarly, it's important that all threads have
|
||||
finished using the queue (and the memory effects have fully propagated) before it is
|
||||
destructed.
|
||||
|
||||
There's usually two versions of each method, one "explicit" version that takes a user-allocated per-producer or
|
||||
per-consumer token, and one "implicit" version that works without tokens. Using the explicit methods is almost
|
||||
always faster (though not necessarily by a huge factor). Apart from performance, the primary distinction between them
|
||||
is their sub-queue allocation behaviour for enqueue operations: Using the implicit enqueue methods causes an
|
||||
automatically-allocated thread-local producer sub-queue to be allocated (it is marked for reuse once the thread exits).
|
||||
Explicit producers, on the other hand, are tied directly to their tokens' lifetimes (and are also recycled as needed).
|
||||
|
||||
Full API (pseudocode):
|
||||
|
||||
# Allocates more memory if necessary
|
||||
enqueue(item) : bool
|
||||
enqueue(prod_token, item) : bool
|
||||
enqueue_bulk(item_first, count) : bool
|
||||
enqueue_bulk(prod_token, item_first, count) : bool
|
||||
|
||||
# Fails if not enough memory to enqueue
|
||||
try_enqueue(item) : bool
|
||||
try_enqueue(prod_token, item) : bool
|
||||
try_enqueue_bulk(item_first, count) : bool
|
||||
try_enqueue_bulk(prod_token, item_first, count) : bool
|
||||
|
||||
# Attempts to dequeue from the queue (never allocates)
|
||||
try_dequeue(item&) : bool
|
||||
try_dequeue(cons_token, item&) : bool
|
||||
try_dequeue_bulk(item_first, max) : size_t
|
||||
try_dequeue_bulk(cons_token, item_first, max) : size_t
|
||||
|
||||
# If you happen to know which producer you want to dequeue from
|
||||
try_dequeue_from_producer(prod_token, item&) : bool
|
||||
try_dequeue_bulk_from_producer(prod_token, item_first, max) : size_t
|
||||
|
||||
# A not-necessarily-accurate count of the total number of elements
|
||||
size_approx() : size_t
|
||||
|
||||
## Blocking version
|
||||
|
||||
As mentioned above, a full blocking wrapper of the queue is provided that adds
|
||||
`wait_dequeue` and `wait_dequeue_bulk` methods in addition to the regular interface.
|
||||
This wrapper is extremely low-overhead, but slightly less fast than the non-blocking
|
||||
queue (due to the necessary bookkeeping involving a lightweight semaphore).
|
||||
|
||||
There are also timed versions that allow a timeout to be specified (either in microseconds
|
||||
or with a `std::chrono` object).
|
||||
|
||||
The only major caveat with the blocking version is that you must be careful not to
|
||||
destroy the queue while somebody is waiting on it. This generally means you need to
|
||||
know for certain that another element is going to come along before you call one of
|
||||
the blocking methods. (To be fair, the non-blocking version cannot be destroyed while
|
||||
in use either, but it can be easier to coordinate the cleanup.)
|
||||
|
||||
Blocking example:
|
||||
|
||||
#include "blockingconcurrentqueue.h"
|
||||
|
||||
moodycamel::BlockingConcurrentQueue<int> q;
|
||||
std::thread producer([&]() {
|
||||
for (int i = 0; i != 100; ++i) {
|
||||
std::this_thread::sleep_for(std::chrono::milliseconds(i % 10));
|
||||
q.enqueue(i);
|
||||
}
|
||||
});
|
||||
std::thread consumer([&]() {
|
||||
for (int i = 0; i != 100; ++i) {
|
||||
int item;
|
||||
q.wait_dequeue(item);
|
||||
assert(item == i);
|
||||
|
||||
if (q.wait_dequeue_timed(item, std::chrono::milliseconds(5))) {
|
||||
++i;
|
||||
assert(item == i);
|
||||
}
|
||||
}
|
||||
});
|
||||
producer.join();
|
||||
consumer.join();
|
||||
|
||||
assert(q.size_approx() == 0);
|
||||
|
||||
## Advanced features
|
||||
|
||||
#### Tokens
|
||||
|
||||
The queue can take advantage of extra per-producer and per-consumer storage if
|
||||
it's available to speed up its operations. This takes the form of "tokens":
|
||||
You can create a consumer token and/or a producer token for each thread or task
|
||||
(tokens themselves are not thread-safe), and use the methods that accept a token
|
||||
as their first parameter:
|
||||
|
||||
moodycamel::ConcurrentQueue<int> q;
|
||||
|
||||
moodycamel::ProducerToken ptok(q);
|
||||
q.enqueue(ptok, 17);
|
||||
|
||||
moodycamel::ConsumerToken ctok(q);
|
||||
int item;
|
||||
q.try_dequeue(ctok, item);
|
||||
assert(item == 17);
|
||||
|
||||
If you happen to know which producer you want to consume from (e.g. in
|
||||
a single-producer, multi-consumer scenario), you can use the `try_dequeue_from_producer`
|
||||
methods, which accept a producer token instead of a consumer token, and cut some overhead.
|
||||
|
||||
Note that tokens work with the blocking version of the queue too.
|
||||
|
||||
When producing or consuming many elements, the most efficient way is to:
|
||||
|
||||
1. Use the bulk methods of the queue with tokens
|
||||
2. Failing that, use the bulk methods without tokens
|
||||
3. Failing that, use the single-item methods with tokens
|
||||
4. Failing that, use the single-item methods without tokens
|
||||
|
||||
Having said that, don't create tokens willy-nilly -- ideally there would be
|
||||
one token (of each kind) per thread. The queue will work with what it is
|
||||
given, but it performs best when used with tokens.
|
||||
|
||||
Note that tokens aren't actually tied to any given thread; it's not technically
|
||||
required that they be local to the thread, only that they be used by a single
|
||||
producer/consumer at a time.
|
||||
|
||||
#### Bulk operations
|
||||
|
||||
Thanks to the [novel design][blog] of the queue, it's just as easy to enqueue/dequeue multiple
|
||||
items as it is to do one at a time. This means that overhead can be cut drastically for
|
||||
bulk operations. Example syntax:
|
||||
|
||||
moodycamel::ConcurrentQueue<int> q;
|
||||
|
||||
int items[] = { 1, 2, 3, 4, 5 };
|
||||
q.enqueue_bulk(items, 5);
|
||||
|
||||
int results[5]; // Could also be any iterator
|
||||
size_t count = q.try_dequeue_bulk(results, 5);
|
||||
for (size_t i = 0; i != count; ++i) {
|
||||
assert(results[i] == items[i]);
|
||||
}
|
||||
|
||||
#### Preallocation (correctly using `try_enqueue`)
|
||||
|
||||
`try_enqueue`, unlike just plain `enqueue`, will never allocate memory. If there's not enough room in the
|
||||
queue, it simply returns false. The key to using this method properly, then, is to ensure enough space is
|
||||
pre-allocated for your desired maximum element count.
|
||||
|
||||
The constructor accepts a count of the number of elements that it should reserve space for. Because the
|
||||
queue works with blocks of elements, however, and not individual elements themselves, the value to pass
|
||||
in order to obtain an effective number of pre-allocated element slots is non-obvious.
|
||||
|
||||
First, be aware that the count passed is rounded up to the next multiple of the block size. Note that the
|
||||
default block size is 32 (this can be changed via the traits). Second, once a slot in a block has been
|
||||
enqueued to, that slot cannot be re-used until the rest of the block has completely been completely filled
|
||||
up and then completely emptied. This affects the number of blocks you need in order to account for the
|
||||
overhead of partially-filled blocks. Third, each producer (whether implicit or explicit) claims and recycles
|
||||
blocks in a different manner, which again affects the number of blocks you need to account for a desired number of
|
||||
usable slots.
|
||||
|
||||
Suppose you want the queue to be able to hold at least `N` elements at any given time. Without delving too
|
||||
deep into the rather arcane implementation details, here are some simple formulas for the number of elements
|
||||
to request for pre-allocation in such a case. Note the division is intended to be arithmetic division and not
|
||||
integer division (in order for `ceil()` to work).
|
||||
|
||||
For explicit producers (using tokens to enqueue):
|
||||
|
||||
(ceil(N / BLOCK_SIZE) + 1) * MAX_NUM_PRODUCERS * BLOCK_SIZE
|
||||
|
||||
For implicit producers (no tokens):
|
||||
|
||||
(ceil(N / BLOCK_SIZE) - 1 + 2 * MAX_NUM_PRODUCERS) * BLOCK_SIZE
|
||||
|
||||
When using mixed producer types:
|
||||
|
||||
((ceil(N / BLOCK_SIZE) - 1) * (MAX_EXPLICIT_PRODUCERS + 1) + 2 * (MAX_IMPLICIT_PRODUCERS + MAX_EXPLICIT_PRODUCERS)) * BLOCK_SIZE
|
||||
|
||||
If these formulas seem rather inconvenient, you can use the constructor overload that accepts the minimum
|
||||
number of elements (`N`) and the maximum number of explicit and implicit producers directly, and let it do the
|
||||
computation for you.
|
||||
|
||||
Finally, it's important to note that because the queue is only eventually consistent and takes advantage of
|
||||
weak memory ordering for speed, there's always a possibility that under contention `try_enqueue` will fail
|
||||
even if the queue is correctly pre-sized for the desired number of elements. (e.g. A given thread may think that
|
||||
the queue's full even when that's no longer the case.) So no matter what, you still need to handle the failure
|
||||
case (perhaps looping until it succeeds), unless you don't mind dropping elements.
|
||||
|
||||
#### Exception safety
|
||||
|
||||
The queue is exception safe, and will never become corrupted if used with a type that may throw exceptions.
|
||||
The queue itself never throws any exceptions (operations fail gracefully (return false) if memory allocation
|
||||
fails instead of throwing `std::bad_alloc`).
|
||||
|
||||
It is important to note that the guarantees of exception safety only hold if the element type never throws
|
||||
from its destructor, and that any iterators passed into the queue (for bulk operations) never throw either.
|
||||
Note that in particular this means `std::back_inserter` iterators must be used with care, since the vector
|
||||
being inserted into may need to allocate and throw a `std::bad_alloc` exception from inside the iterator;
|
||||
so be sure to reserve enough capacity in the target container first if you do this.
|
||||
|
||||
The guarantees are presently as follows:
|
||||
- Enqueue operations are rolled back completely if an exception is thrown from an element's constructor.
|
||||
For bulk enqueue operations, this means that elements are copied instead of moved (in order to avoid
|
||||
having only some of the objects be moved in the event of an exception). Non-bulk enqueues always use
|
||||
the move constructor if one is available.
|
||||
- If the assignment operator throws during a dequeue operation (both single and bulk), the element(s) are
|
||||
considered dequeued regardless. In such a case, the dequeued elements are all properly destructed before
|
||||
the exception is propagated, but there's no way to get the elements themselves back.
|
||||
- Any exception that is thrown is propagated up the call stack, at which point the queue is in a consistent
|
||||
state.
|
||||
|
||||
Note: If any of your type's copy constructors/move constructors/assignment operators don't throw, be sure
|
||||
to annotate them with `noexcept`; this will avoid the exception-checking overhead in the queue where possible
|
||||
(even with zero-cost exceptions, there's still a code size impact that has to be taken into account).
|
||||
|
||||
#### Traits
|
||||
|
||||
The queue also supports a traits template argument which defines various types, constants,
|
||||
and the memory allocation and deallocation functions that are to be used by the queue. The typical pattern
|
||||
to providing your own traits is to create a class that inherits from the default traits
|
||||
and override only the values you wish to change. Example:
|
||||
|
||||
struct MyTraits : public moodycamel::ConcurrentQueueDefaultTraits
|
||||
{
|
||||
static const size_t BLOCK_SIZE = 256; // Use bigger blocks
|
||||
};
|
||||
|
||||
moodycamel::ConcurrentQueue<int, MyTraits> q;
|
||||
|
||||
#### How to dequeue types without calling the constructor
|
||||
|
||||
The normal way to dequeue an item is to pass in an existing object by reference, which
|
||||
is then assigned to internally by the queue (using the move-assignment operator if possible).
|
||||
This can pose a problem for types that are
|
||||
expensive to construct or don't have a default constructor; fortunately, there is a simple
|
||||
workaround: Create a wrapper class that copies the memory contents of the object when it
|
||||
is assigned by the queue (a poor man's move, essentially). Note that this only works if
|
||||
the object contains no internal pointers. Example:
|
||||
|
||||
struct MyObjectMover {
|
||||
inline void operator=(MyObject&& obj)
|
||||
{
|
||||
std::memcpy(data, &obj, sizeof(MyObject));
|
||||
|
||||
// TODO: Cleanup obj so that when it's destructed by the queue
|
||||
// it doesn't corrupt the data of the object we just moved it into
|
||||
}
|
||||
|
||||
inline MyObject& obj() { return *reinterpret_cast<MyObject*>(data); }
|
||||
|
||||
private:
|
||||
align(alignof(MyObject)) char data[sizeof(MyObject)];
|
||||
};
|
||||
|
||||
A less dodgy alternative, if moves are cheap but default construction is not, is to use a
|
||||
wrapper that defers construction until the object is assigned, enabling use of the move
|
||||
constructor:
|
||||
|
||||
struct MyObjectMover {
|
||||
inline void operator=(MyObject&& x) {
|
||||
new (data) MyObject(std::move(x));
|
||||
created = true;
|
||||
}
|
||||
|
||||
inline MyObject& obj() {
|
||||
assert(created);
|
||||
return *reinterpret_cast<MyObject*>(data);
|
||||
}
|
||||
|
||||
~MyObjectMover() {
|
||||
if (created)
|
||||
obj().~MyObject();
|
||||
}
|
||||
|
||||
private:
|
||||
align(alignof(MyObject)) char data[sizeof(MyObject)];
|
||||
bool created = false;
|
||||
};
|
||||
|
||||
|
||||
## Samples
|
||||
|
||||
There are some more detailed samples [here][samples.md]. The source of
|
||||
the [unit tests][unittest-src] and [benchmarks][benchmark-src] are available for reference as well.
|
||||
|
||||
## Benchmarks
|
||||
|
||||
See my blog post for some [benchmark results][benchmarks] (including versus `boost::lockfree::queue` and `tbb::concurrent_queue`),
|
||||
or run the benchmarks yourself (requires MinGW and certain GnuWin32 utilities to build on Windows, or a recent
|
||||
g++ on Linux):
|
||||
|
||||
cd build
|
||||
make benchmarks
|
||||
bin/benchmarks
|
||||
|
||||
The short version of the benchmarks is that it's so fast (especially the bulk methods), that if you're actually
|
||||
using the queue to *do* anything, the queue won't be your bottleneck.
|
||||
|
||||
## Tests (and bugs)
|
||||
|
||||
I've written quite a few unit tests as well as a randomized long-running fuzz tester. I also ran the
|
||||
core queue algorithm through the [CDSChecker][cdschecker] C++11 memory model model checker. Some of the
|
||||
inner algorithms were tested separately using the [Relacy][relacy] model checker, and full integration
|
||||
tests were also performed with Relacy.
|
||||
I've tested
|
||||
on Linux (Fedora 19) and Windows (7), but only on x86 processors so far (Intel and AMD). The code was
|
||||
written to be platform-independent, however, and should work across all processors and OSes.
|
||||
|
||||
Due to the complexity of the implementation and the difficult-to-test nature of lock-free code in general,
|
||||
there may still be bugs. If anyone is seeing buggy behaviour, I'd like to hear about it! (Especially if
|
||||
a unit test for it can be cooked up.) Just open an issue on GitHub.
|
||||
|
||||
## License
|
||||
|
||||
I'm releasing the source of this repository (with the exception of third-party code, i.e. the Boost queue
|
||||
(used in the benchmarks for comparison), Intel's TBB library (ditto), CDSChecker, Relacy, and Jeff Preshing's
|
||||
cross-platform semaphore, which all have their own licenses)
|
||||
under a simplified BSD license. I'm also dual-licensing under the Boost Software License.
|
||||
See the [LICENSE.md][license] file for more details.
|
||||
|
||||
Note that lock-free programming is a patent minefield, and this code may very
|
||||
well violate a pending patent (I haven't looked), though it does not to my present knowledge.
|
||||
I did design and implement this queue from scratch.
|
||||
|
||||
## Diving into the code
|
||||
|
||||
If you're interested in the source code itself, it helps to have a rough idea of how it's laid out. This
|
||||
section attempts to describe that.
|
||||
|
||||
The queue is formed of several basic parts (listed here in roughly the order they appear in the source). There's the
|
||||
helper functions (e.g. for rounding to a power of 2). There's the default traits of the queue, which contain the
|
||||
constants and malloc/free functions used by the queue. There's the producer and consumer tokens. Then there's the queue's
|
||||
public API itself, starting with the constructor, destructor, and swap/assignment methods. There's the public enqueue methods,
|
||||
which are all wrappers around a small set of private enqueue methods found later on. There's the dequeue methods, which are
|
||||
defined inline and are relatively straightforward.
|
||||
|
||||
Then there's all the main internal data structures. First, there's a lock-free free list, used for recycling spent blocks (elements
|
||||
are enqueued to blocks internally). Then there's the block structure itself, which has two different ways of tracking whether
|
||||
it's fully emptied or not (remember, given two parallel consumers, there's no way to know which one will finish first) depending on where it's used.
|
||||
Then there's a small base class for the two types of internal SPMC producer queues (one for explicit producers that holds onto memory
|
||||
but attempts to be faster, and one for implicit ones which attempt to recycle more memory back into the parent but is a little slower).
|
||||
The explicit producer is defined first, then the implicit one. They both contain the same general four methods: One to enqueue, one to
|
||||
dequeue, one to enqueue in bulk, and one to dequeue in bulk. (Obviously they have constructors and destructors too, and helper methods.)
|
||||
The main difference between them is how the block handling is done (they both use the same blocks, but in different ways, and map indices
|
||||
to them in different ways).
|
||||
|
||||
Finally, there's the miscellaneous internal methods: There's the ones that handle the initial block pool (populated when the queue is constructed),
|
||||
and an abstract block pool that comprises the initial pool and any blocks on the free list. There's ones that handle the producer list
|
||||
(a lock-free add-only linked list of all the producers in the system). There's ones that handle the implicit producer lookup table (which
|
||||
is really a sort of specialized TLS lookup). And then there's some helper methods for allocating and freeing objects, and the data members
|
||||
of the queue itself, followed lastly by the free-standing swap functions.
|
||||
|
||||
|
||||
[blog]: http://moodycamel.com/blog/2014/a-fast-general-purpose-lock-free-queue-for-c++
|
||||
[design]: http://moodycamel.com/blog/2014/detailed-design-of-a-lock-free-queue
|
||||
[samples.md]: https://github.com/cameron314/concurrentqueue/blob/master/samples.md
|
||||
[source]: https://github.com/cameron314/concurrentqueue
|
||||
[concurrentqueue.h]: https://github.com/cameron314/concurrentqueue/blob/master/concurrentqueue.h
|
||||
[blockingconcurrentqueue.h]: https://github.com/cameron314/concurrentqueue/blob/master/blockingconcurrentqueue.h
|
||||
[unittest-src]: https://github.com/cameron314/concurrentqueue/tree/master/tests/unittests
|
||||
[benchmarks]: http://moodycamel.com/blog/2014/a-fast-general-purpose-lock-free-queue-for-c++#benchmarks
|
||||
[benchmark-src]: https://github.com/cameron314/concurrentqueue/tree/master/benchmarks
|
||||
[license]: https://github.com/cameron314/concurrentqueue/blob/master/LICENSE.md
|
||||
[cdschecker]: http://demsky.eecs.uci.edu/c11modelchecker.html
|
||||
[relacy]: http://www.1024cores.net/home/relacy-race-detector
|
||||
[spsc]: https://github.com/cameron314/readerwriterqueue
|
||||
[salsa]: http://webee.technion.ac.il/~idish/ftp/spaa049-gidron.pdf
|
||||
@@ -0,0 +1,981 @@
|
||||
// Provides an efficient blocking version of moodycamel::ConcurrentQueue.
|
||||
// ©2015-2016 Cameron Desrochers. Distributed under the terms of the simplified
|
||||
// BSD license, available at the top of concurrentqueue.h.
|
||||
// Uses Jeff Preshing's semaphore implementation (under the terms of its
|
||||
// separate zlib license, embedded below).
|
||||
|
||||
#pragma once
|
||||
|
||||
#include "concurrentqueue.h"
|
||||
#include <type_traits>
|
||||
#include <cerrno>
|
||||
#include <memory>
|
||||
#include <chrono>
|
||||
#include <ctime>
|
||||
|
||||
#if defined(_WIN32)
|
||||
// Avoid including windows.h in a header; we only need a handful of
|
||||
// items, so we'll redeclare them here (this is relatively safe since
|
||||
// the API generally has to remain stable between Windows versions).
|
||||
// I know this is an ugly hack but it still beats polluting the global
|
||||
// namespace with thousands of generic names or adding a .cpp for nothing.
|
||||
extern "C" {
|
||||
struct _SECURITY_ATTRIBUTES;
|
||||
__declspec(dllimport) void* __stdcall CreateSemaphoreW(_SECURITY_ATTRIBUTES* lpSemaphoreAttributes, long lInitialCount, long lMaximumCount, const wchar_t* lpName);
|
||||
__declspec(dllimport) int __stdcall CloseHandle(void* hObject);
|
||||
__declspec(dllimport) unsigned long __stdcall WaitForSingleObject(void* hHandle, unsigned long dwMilliseconds);
|
||||
__declspec(dllimport) int __stdcall ReleaseSemaphore(void* hSemaphore, long lReleaseCount, long* lpPreviousCount);
|
||||
}
|
||||
#elif defined(__MACH__)
|
||||
#include <mach/mach.h>
|
||||
#elif defined(__unix__)
|
||||
#include <semaphore.h>
|
||||
#endif
|
||||
|
||||
namespace moodycamel
|
||||
{
|
||||
namespace details
|
||||
{
|
||||
// Code in the mpmc_sema namespace below is an adaptation of Jeff Preshing's
|
||||
// portable + lightweight semaphore implementations, originally from
|
||||
// https://github.com/preshing/cpp11-on-multicore/blob/master/common/sema.h
|
||||
// LICENSE:
|
||||
// Copyright (c) 2015 Jeff Preshing
|
||||
//
|
||||
// This software is provided 'as-is', without any express or implied
|
||||
// warranty. In no event will the authors be held liable for any damages
|
||||
// arising from the use of this software.
|
||||
//
|
||||
// Permission is granted to anyone to use this software for any purpose,
|
||||
// including commercial applications, and to alter it and redistribute it
|
||||
// freely, subject to the following restrictions:
|
||||
//
|
||||
// 1. The origin of this software must not be misrepresented; you must not
|
||||
// claim that you wrote the original software. If you use this software
|
||||
// in a product, an acknowledgement in the product documentation would be
|
||||
// appreciated but is not required.
|
||||
// 2. Altered source versions must be plainly marked as such, and must not be
|
||||
// misrepresented as being the original software.
|
||||
// 3. This notice may not be removed or altered from any source distribution.
|
||||
namespace mpmc_sema
|
||||
{
|
||||
#if defined(_WIN32)
|
||||
class Semaphore
|
||||
{
|
||||
private:
|
||||
void* m_hSema;
|
||||
|
||||
Semaphore(const Semaphore& other) MOODYCAMEL_DELETE_FUNCTION;
|
||||
Semaphore& operator=(const Semaphore& other) MOODYCAMEL_DELETE_FUNCTION;
|
||||
|
||||
public:
|
||||
Semaphore(int initialCount = 0)
|
||||
{
|
||||
assert(initialCount >= 0);
|
||||
const long maxLong = 0x7fffffff;
|
||||
m_hSema = CreateSemaphoreW(nullptr, initialCount, maxLong, nullptr);
|
||||
}
|
||||
|
||||
~Semaphore()
|
||||
{
|
||||
CloseHandle(m_hSema);
|
||||
}
|
||||
|
||||
void wait()
|
||||
{
|
||||
const unsigned long infinite = 0xffffffff;
|
||||
WaitForSingleObject(m_hSema, infinite);
|
||||
}
|
||||
|
||||
bool try_wait()
|
||||
{
|
||||
const unsigned long RC_WAIT_TIMEOUT = 0x00000102;
|
||||
return WaitForSingleObject(m_hSema, 0) != RC_WAIT_TIMEOUT;
|
||||
}
|
||||
|
||||
bool timed_wait(std::uint64_t usecs)
|
||||
{
|
||||
const unsigned long RC_WAIT_TIMEOUT = 0x00000102;
|
||||
return WaitForSingleObject(m_hSema, (unsigned long)(usecs / 1000)) != RC_WAIT_TIMEOUT;
|
||||
}
|
||||
|
||||
void signal(int count = 1)
|
||||
{
|
||||
ReleaseSemaphore(m_hSema, count, nullptr);
|
||||
}
|
||||
};
|
||||
#elif defined(__MACH__)
|
||||
//---------------------------------------------------------
|
||||
// Semaphore (Apple iOS and OSX)
|
||||
// Can't use POSIX semaphores due to http://lists.apple.com/archives/darwin-kernel/2009/Apr/msg00010.html
|
||||
//---------------------------------------------------------
|
||||
class Semaphore
|
||||
{
|
||||
private:
|
||||
semaphore_t m_sema;
|
||||
|
||||
Semaphore(const Semaphore& other) MOODYCAMEL_DELETE_FUNCTION;
|
||||
Semaphore& operator=(const Semaphore& other) MOODYCAMEL_DELETE_FUNCTION;
|
||||
|
||||
public:
|
||||
Semaphore(int initialCount = 0)
|
||||
{
|
||||
assert(initialCount >= 0);
|
||||
semaphore_create(mach_task_self(), &m_sema, SYNC_POLICY_FIFO, initialCount);
|
||||
}
|
||||
|
||||
~Semaphore()
|
||||
{
|
||||
semaphore_destroy(mach_task_self(), m_sema);
|
||||
}
|
||||
|
||||
void wait()
|
||||
{
|
||||
semaphore_wait(m_sema);
|
||||
}
|
||||
|
||||
bool try_wait()
|
||||
{
|
||||
return timed_wait(0);
|
||||
}
|
||||
|
||||
bool timed_wait(std::uint64_t timeout_usecs)
|
||||
{
|
||||
mach_timespec_t ts;
|
||||
ts.tv_sec = static_cast<unsigned int>(timeout_usecs / 1000000);
|
||||
ts.tv_nsec = (timeout_usecs % 1000000) * 1000;
|
||||
|
||||
// added in OSX 10.10: https://developer.apple.com/library/prerelease/mac/documentation/General/Reference/APIDiffsMacOSX10_10SeedDiff/modules/Darwin.html
|
||||
kern_return_t rc = semaphore_timedwait(m_sema, ts);
|
||||
|
||||
return rc != KERN_OPERATION_TIMED_OUT && rc != KERN_ABORTED;
|
||||
}
|
||||
|
||||
void signal()
|
||||
{
|
||||
semaphore_signal(m_sema);
|
||||
}
|
||||
|
||||
void signal(int count)
|
||||
{
|
||||
while (count-- > 0)
|
||||
{
|
||||
semaphore_signal(m_sema);
|
||||
}
|
||||
}
|
||||
};
|
||||
#elif defined(__unix__)
|
||||
//---------------------------------------------------------
|
||||
// Semaphore (POSIX, Linux)
|
||||
//---------------------------------------------------------
|
||||
class Semaphore
|
||||
{
|
||||
private:
|
||||
sem_t m_sema;
|
||||
|
||||
Semaphore(const Semaphore& other) MOODYCAMEL_DELETE_FUNCTION;
|
||||
Semaphore& operator=(const Semaphore& other) MOODYCAMEL_DELETE_FUNCTION;
|
||||
|
||||
public:
|
||||
Semaphore(int initialCount = 0)
|
||||
{
|
||||
assert(initialCount >= 0);
|
||||
sem_init(&m_sema, 0, initialCount);
|
||||
}
|
||||
|
||||
~Semaphore()
|
||||
{
|
||||
sem_destroy(&m_sema);
|
||||
}
|
||||
|
||||
void wait()
|
||||
{
|
||||
// http://stackoverflow.com/questions/2013181/gdb-causes-sem-wait-to-fail-with-eintr-error
|
||||
int rc;
|
||||
do {
|
||||
rc = sem_wait(&m_sema);
|
||||
} while (rc == -1 && errno == EINTR);
|
||||
}
|
||||
|
||||
bool try_wait()
|
||||
{
|
||||
int rc;
|
||||
do {
|
||||
rc = sem_trywait(&m_sema);
|
||||
} while (rc == -1 && errno == EINTR);
|
||||
return !(rc == -1 && errno == EAGAIN);
|
||||
}
|
||||
|
||||
bool timed_wait(std::uint64_t usecs)
|
||||
{
|
||||
struct timespec ts;
|
||||
const int usecs_in_1_sec = 1000000;
|
||||
const int nsecs_in_1_sec = 1000000000;
|
||||
clock_gettime(CLOCK_REALTIME, &ts);
|
||||
ts.tv_sec += usecs / usecs_in_1_sec;
|
||||
ts.tv_nsec += (usecs % usecs_in_1_sec) * 1000;
|
||||
// sem_timedwait bombs if you have more than 1e9 in tv_nsec
|
||||
// so we have to clean things up before passing it in
|
||||
if (ts.tv_nsec >= nsecs_in_1_sec) {
|
||||
ts.tv_nsec -= nsecs_in_1_sec;
|
||||
++ts.tv_sec;
|
||||
}
|
||||
|
||||
int rc;
|
||||
do {
|
||||
rc = sem_timedwait(&m_sema, &ts);
|
||||
} while (rc == -1 && errno == EINTR);
|
||||
return !(rc == -1 && errno == ETIMEDOUT);
|
||||
}
|
||||
|
||||
void signal()
|
||||
{
|
||||
sem_post(&m_sema);
|
||||
}
|
||||
|
||||
void signal(int count)
|
||||
{
|
||||
while (count-- > 0)
|
||||
{
|
||||
sem_post(&m_sema);
|
||||
}
|
||||
}
|
||||
};
|
||||
#else
|
||||
#error Unsupported platform! (No semaphore wrapper available)
|
||||
#endif
|
||||
|
||||
//---------------------------------------------------------
|
||||
// LightweightSemaphore
|
||||
//---------------------------------------------------------
|
||||
class LightweightSemaphore
|
||||
{
|
||||
public:
|
||||
typedef std::make_signed<std::size_t>::type ssize_t;
|
||||
|
||||
private:
|
||||
std::atomic<ssize_t> m_count;
|
||||
Semaphore m_sema;
|
||||
|
||||
bool waitWithPartialSpinning(std::int64_t timeout_usecs = -1)
|
||||
{
|
||||
ssize_t oldCount;
|
||||
// Is there a better way to set the initial spin count?
|
||||
// If we lower it to 1000, testBenaphore becomes 15x slower on my Core i7-5930K Windows PC,
|
||||
// as threads start hitting the kernel semaphore.
|
||||
int spin = 10000;
|
||||
while (--spin >= 0)
|
||||
{
|
||||
oldCount = m_count.load(std::memory_order_relaxed);
|
||||
if ((oldCount > 0) && m_count.compare_exchange_strong(oldCount, oldCount - 1, std::memory_order_acquire, std::memory_order_relaxed))
|
||||
return true;
|
||||
std::atomic_signal_fence(std::memory_order_acquire); // Prevent the compiler from collapsing the loop.
|
||||
}
|
||||
oldCount = m_count.fetch_sub(1, std::memory_order_acquire);
|
||||
if (oldCount > 0)
|
||||
return true;
|
||||
if (timeout_usecs < 0)
|
||||
{
|
||||
m_sema.wait();
|
||||
return true;
|
||||
}
|
||||
if (m_sema.timed_wait((std::uint64_t)timeout_usecs))
|
||||
return true;
|
||||
// At this point, we've timed out waiting for the semaphore, but the
|
||||
// count is still decremented indicating we may still be waiting on
|
||||
// it. So we have to re-adjust the count, but only if the semaphore
|
||||
// wasn't signaled enough times for us too since then. If it was, we
|
||||
// need to release the semaphore too.
|
||||
while (true)
|
||||
{
|
||||
oldCount = m_count.load(std::memory_order_acquire);
|
||||
if (oldCount >= 0 && m_sema.try_wait())
|
||||
return true;
|
||||
if (oldCount < 0 && m_count.compare_exchange_strong(oldCount, oldCount + 1, std::memory_order_relaxed, std::memory_order_relaxed))
|
||||
return false;
|
||||
}
|
||||
}
|
||||
|
||||
ssize_t waitManyWithPartialSpinning(ssize_t max, std::int64_t timeout_usecs = -1)
|
||||
{
|
||||
assert(max > 0);
|
||||
ssize_t oldCount;
|
||||
int spin = 10000;
|
||||
while (--spin >= 0)
|
||||
{
|
||||
oldCount = m_count.load(std::memory_order_relaxed);
|
||||
if (oldCount > 0)
|
||||
{
|
||||
ssize_t newCount = oldCount > max ? oldCount - max : 0;
|
||||
if (m_count.compare_exchange_strong(oldCount, newCount, std::memory_order_acquire, std::memory_order_relaxed))
|
||||
return oldCount - newCount;
|
||||
}
|
||||
std::atomic_signal_fence(std::memory_order_acquire);
|
||||
}
|
||||
oldCount = m_count.fetch_sub(1, std::memory_order_acquire);
|
||||
if (oldCount <= 0)
|
||||
{
|
||||
if (timeout_usecs < 0)
|
||||
m_sema.wait();
|
||||
else if (!m_sema.timed_wait((std::uint64_t)timeout_usecs))
|
||||
{
|
||||
while (true)
|
||||
{
|
||||
oldCount = m_count.load(std::memory_order_acquire);
|
||||
if (oldCount >= 0 && m_sema.try_wait())
|
||||
break;
|
||||
if (oldCount < 0 && m_count.compare_exchange_strong(oldCount, oldCount + 1, std::memory_order_relaxed, std::memory_order_relaxed))
|
||||
return 0;
|
||||
}
|
||||
}
|
||||
}
|
||||
if (max > 1)
|
||||
return 1 + tryWaitMany(max - 1);
|
||||
return 1;
|
||||
}
|
||||
|
||||
public:
|
||||
LightweightSemaphore(ssize_t initialCount = 0) : m_count(initialCount)
|
||||
{
|
||||
assert(initialCount >= 0);
|
||||
}
|
||||
|
||||
bool tryWait()
|
||||
{
|
||||
ssize_t oldCount = m_count.load(std::memory_order_relaxed);
|
||||
while (oldCount > 0)
|
||||
{
|
||||
if (m_count.compare_exchange_weak(oldCount, oldCount - 1, std::memory_order_acquire, std::memory_order_relaxed))
|
||||
return true;
|
||||
}
|
||||
return false;
|
||||
}
|
||||
|
||||
void wait()
|
||||
{
|
||||
if (!tryWait())
|
||||
waitWithPartialSpinning();
|
||||
}
|
||||
|
||||
bool wait(std::int64_t timeout_usecs)
|
||||
{
|
||||
return tryWait() || waitWithPartialSpinning(timeout_usecs);
|
||||
}
|
||||
|
||||
// Acquires between 0 and (greedily) max, inclusive
|
||||
ssize_t tryWaitMany(ssize_t max)
|
||||
{
|
||||
assert(max >= 0);
|
||||
ssize_t oldCount = m_count.load(std::memory_order_relaxed);
|
||||
while (oldCount > 0)
|
||||
{
|
||||
ssize_t newCount = oldCount > max ? oldCount - max : 0;
|
||||
if (m_count.compare_exchange_weak(oldCount, newCount, std::memory_order_acquire, std::memory_order_relaxed))
|
||||
return oldCount - newCount;
|
||||
}
|
||||
return 0;
|
||||
}
|
||||
|
||||
// Acquires at least one, and (greedily) at most max
|
||||
ssize_t waitMany(ssize_t max, std::int64_t timeout_usecs)
|
||||
{
|
||||
assert(max >= 0);
|
||||
ssize_t result = tryWaitMany(max);
|
||||
if (result == 0 && max > 0)
|
||||
result = waitManyWithPartialSpinning(max, timeout_usecs);
|
||||
return result;
|
||||
}
|
||||
|
||||
ssize_t waitMany(ssize_t max)
|
||||
{
|
||||
ssize_t result = waitMany(max, -1);
|
||||
assert(result > 0);
|
||||
return result;
|
||||
}
|
||||
|
||||
void signal(ssize_t count = 1)
|
||||
{
|
||||
assert(count >= 0);
|
||||
ssize_t oldCount = m_count.fetch_add(count, std::memory_order_release);
|
||||
ssize_t toRelease = -oldCount < count ? -oldCount : count;
|
||||
if (toRelease > 0)
|
||||
{
|
||||
m_sema.signal((int)toRelease);
|
||||
}
|
||||
}
|
||||
|
||||
ssize_t availableApprox() const
|
||||
{
|
||||
ssize_t count = m_count.load(std::memory_order_relaxed);
|
||||
return count > 0 ? count : 0;
|
||||
}
|
||||
};
|
||||
} // end namespace mpmc_sema
|
||||
} // end namespace details
|
||||
|
||||
|
||||
// This is a blocking version of the queue. It has an almost identical interface to
|
||||
// the normal non-blocking version, with the addition of various wait_dequeue() methods
|
||||
// and the removal of producer-specific dequeue methods.
|
||||
template<typename T, typename Traits = ConcurrentQueueDefaultTraits>
|
||||
class BlockingConcurrentQueue
|
||||
{
|
||||
private:
|
||||
typedef ::moodycamel::ConcurrentQueue<T, Traits> ConcurrentQueue;
|
||||
typedef details::mpmc_sema::LightweightSemaphore LightweightSemaphore;
|
||||
|
||||
public:
|
||||
typedef typename ConcurrentQueue::producer_token_t producer_token_t;
|
||||
typedef typename ConcurrentQueue::consumer_token_t consumer_token_t;
|
||||
|
||||
typedef typename ConcurrentQueue::index_t index_t;
|
||||
typedef typename ConcurrentQueue::size_t size_t;
|
||||
typedef typename std::make_signed<size_t>::type ssize_t;
|
||||
|
||||
static const size_t BLOCK_SIZE = ConcurrentQueue::BLOCK_SIZE;
|
||||
static const size_t EXPLICIT_BLOCK_EMPTY_COUNTER_THRESHOLD = ConcurrentQueue::EXPLICIT_BLOCK_EMPTY_COUNTER_THRESHOLD;
|
||||
static const size_t EXPLICIT_INITIAL_INDEX_SIZE = ConcurrentQueue::EXPLICIT_INITIAL_INDEX_SIZE;
|
||||
static const size_t IMPLICIT_INITIAL_INDEX_SIZE = ConcurrentQueue::IMPLICIT_INITIAL_INDEX_SIZE;
|
||||
static const size_t INITIAL_IMPLICIT_PRODUCER_HASH_SIZE = ConcurrentQueue::INITIAL_IMPLICIT_PRODUCER_HASH_SIZE;
|
||||
static const std::uint32_t EXPLICIT_CONSUMER_CONSUMPTION_QUOTA_BEFORE_ROTATE = ConcurrentQueue::EXPLICIT_CONSUMER_CONSUMPTION_QUOTA_BEFORE_ROTATE;
|
||||
static const size_t MAX_SUBQUEUE_SIZE = ConcurrentQueue::MAX_SUBQUEUE_SIZE;
|
||||
|
||||
public:
|
||||
// Creates a queue with at least `capacity` element slots; note that the
|
||||
// actual number of elements that can be inserted without additional memory
|
||||
// allocation depends on the number of producers and the block size (e.g. if
|
||||
// the block size is equal to `capacity`, only a single block will be allocated
|
||||
// up-front, which means only a single producer will be able to enqueue elements
|
||||
// without an extra allocation -- blocks aren't shared between producers).
|
||||
// This method is not thread safe -- it is up to the user to ensure that the
|
||||
// queue is fully constructed before it starts being used by other threads (this
|
||||
// includes making the memory effects of construction visible, possibly with a
|
||||
// memory barrier).
|
||||
explicit BlockingConcurrentQueue(size_t capacity = 6 * BLOCK_SIZE)
|
||||
: inner(capacity), sema(create<LightweightSemaphore>(), &BlockingConcurrentQueue::template destroy<LightweightSemaphore>)
|
||||
{
|
||||
assert(reinterpret_cast<ConcurrentQueue*>((BlockingConcurrentQueue*)1) == &((BlockingConcurrentQueue*)1)->inner && "BlockingConcurrentQueue must have ConcurrentQueue as its first member");
|
||||
if (!sema) {
|
||||
MOODYCAMEL_THROW(std::bad_alloc());
|
||||
}
|
||||
}
|
||||
|
||||
BlockingConcurrentQueue(size_t minCapacity, size_t maxExplicitProducers, size_t maxImplicitProducers)
|
||||
: inner(minCapacity, maxExplicitProducers, maxImplicitProducers), sema(create<LightweightSemaphore>(), &BlockingConcurrentQueue::template destroy<LightweightSemaphore>)
|
||||
{
|
||||
assert(reinterpret_cast<ConcurrentQueue*>((BlockingConcurrentQueue*)1) == &((BlockingConcurrentQueue*)1)->inner && "BlockingConcurrentQueue must have ConcurrentQueue as its first member");
|
||||
if (!sema) {
|
||||
MOODYCAMEL_THROW(std::bad_alloc());
|
||||
}
|
||||
}
|
||||
|
||||
// Disable copying and copy assignment
|
||||
BlockingConcurrentQueue(BlockingConcurrentQueue const&) MOODYCAMEL_DELETE_FUNCTION;
|
||||
BlockingConcurrentQueue& operator=(BlockingConcurrentQueue const&) MOODYCAMEL_DELETE_FUNCTION;
|
||||
|
||||
// Moving is supported, but note that it is *not* a thread-safe operation.
|
||||
// Nobody can use the queue while it's being moved, and the memory effects
|
||||
// of that move must be propagated to other threads before they can use it.
|
||||
// Note: When a queue is moved, its tokens are still valid but can only be
|
||||
// used with the destination queue (i.e. semantically they are moved along
|
||||
// with the queue itself).
|
||||
BlockingConcurrentQueue(BlockingConcurrentQueue&& other) MOODYCAMEL_NOEXCEPT
|
||||
: inner(std::move(other.inner)), sema(std::move(other.sema))
|
||||
{ }
|
||||
|
||||
inline BlockingConcurrentQueue& operator=(BlockingConcurrentQueue&& other) MOODYCAMEL_NOEXCEPT
|
||||
{
|
||||
return swap_internal(other);
|
||||
}
|
||||
|
||||
// Swaps this queue's state with the other's. Not thread-safe.
|
||||
// Swapping two queues does not invalidate their tokens, however
|
||||
// the tokens that were created for one queue must be used with
|
||||
// only the swapped queue (i.e. the tokens are tied to the
|
||||
// queue's movable state, not the object itself).
|
||||
inline void swap(BlockingConcurrentQueue& other) MOODYCAMEL_NOEXCEPT
|
||||
{
|
||||
swap_internal(other);
|
||||
}
|
||||
|
||||
private:
|
||||
BlockingConcurrentQueue& swap_internal(BlockingConcurrentQueue& other)
|
||||
{
|
||||
if (this == &other) {
|
||||
return *this;
|
||||
}
|
||||
|
||||
inner.swap(other.inner);
|
||||
sema.swap(other.sema);
|
||||
return *this;
|
||||
}
|
||||
|
||||
public:
|
||||
// Enqueues a single item (by copying it).
|
||||
// Allocates memory if required. Only fails if memory allocation fails (or implicit
|
||||
// production is disabled because Traits::INITIAL_IMPLICIT_PRODUCER_HASH_SIZE is 0,
|
||||
// or Traits::MAX_SUBQUEUE_SIZE has been defined and would be surpassed).
|
||||
// Thread-safe.
|
||||
inline bool enqueue(T const& item)
|
||||
{
|
||||
if ((details::likely)(inner.enqueue(item))) {
|
||||
sema->signal();
|
||||
return true;
|
||||
}
|
||||
return false;
|
||||
}
|
||||
|
||||
// Enqueues a single item (by moving it, if possible).
|
||||
// Allocates memory if required. Only fails if memory allocation fails (or implicit
|
||||
// production is disabled because Traits::INITIAL_IMPLICIT_PRODUCER_HASH_SIZE is 0,
|
||||
// or Traits::MAX_SUBQUEUE_SIZE has been defined and would be surpassed).
|
||||
// Thread-safe.
|
||||
inline bool enqueue(T&& item)
|
||||
{
|
||||
if ((details::likely)(inner.enqueue(std::move(item)))) {
|
||||
sema->signal();
|
||||
return true;
|
||||
}
|
||||
return false;
|
||||
}
|
||||
|
||||
// Enqueues a single item (by copying it) using an explicit producer token.
|
||||
// Allocates memory if required. Only fails if memory allocation fails (or
|
||||
// Traits::MAX_SUBQUEUE_SIZE has been defined and would be surpassed).
|
||||
// Thread-safe.
|
||||
inline bool enqueue(producer_token_t const& token, T const& item)
|
||||
{
|
||||
if ((details::likely)(inner.enqueue(token, item))) {
|
||||
sema->signal();
|
||||
return true;
|
||||
}
|
||||
return false;
|
||||
}
|
||||
|
||||
// Enqueues a single item (by moving it, if possible) using an explicit producer token.
|
||||
// Allocates memory if required. Only fails if memory allocation fails (or
|
||||
// Traits::MAX_SUBQUEUE_SIZE has been defined and would be surpassed).
|
||||
// Thread-safe.
|
||||
inline bool enqueue(producer_token_t const& token, T&& item)
|
||||
{
|
||||
if ((details::likely)(inner.enqueue(token, std::move(item)))) {
|
||||
sema->signal();
|
||||
return true;
|
||||
}
|
||||
return false;
|
||||
}
|
||||
|
||||
// Enqueues several items.
|
||||
// Allocates memory if required. Only fails if memory allocation fails (or
|
||||
// implicit production is disabled because Traits::INITIAL_IMPLICIT_PRODUCER_HASH_SIZE
|
||||
// is 0, or Traits::MAX_SUBQUEUE_SIZE has been defined and would be surpassed).
|
||||
// Note: Use std::make_move_iterator if the elements should be moved instead of copied.
|
||||
// Thread-safe.
|
||||
template<typename It>
|
||||
inline bool enqueue_bulk(It itemFirst, size_t count)
|
||||
{
|
||||
if ((details::likely)(inner.enqueue_bulk(std::forward<It>(itemFirst), count))) {
|
||||
sema->signal((LightweightSemaphore::ssize_t)(ssize_t)count);
|
||||
return true;
|
||||
}
|
||||
return false;
|
||||
}
|
||||
|
||||
// Enqueues several items using an explicit producer token.
|
||||
// Allocates memory if required. Only fails if memory allocation fails
|
||||
// (or Traits::MAX_SUBQUEUE_SIZE has been defined and would be surpassed).
|
||||
// Note: Use std::make_move_iterator if the elements should be moved
|
||||
// instead of copied.
|
||||
// Thread-safe.
|
||||
template<typename It>
|
||||
inline bool enqueue_bulk(producer_token_t const& token, It itemFirst, size_t count)
|
||||
{
|
||||
if ((details::likely)(inner.enqueue_bulk(token, std::forward<It>(itemFirst), count))) {
|
||||
sema->signal((LightweightSemaphore::ssize_t)(ssize_t)count);
|
||||
return true;
|
||||
}
|
||||
return false;
|
||||
}
|
||||
|
||||
// Enqueues a single item (by copying it).
|
||||
// Does not allocate memory. Fails if not enough room to enqueue (or implicit
|
||||
// production is disabled because Traits::INITIAL_IMPLICIT_PRODUCER_HASH_SIZE
|
||||
// is 0).
|
||||
// Thread-safe.
|
||||
inline bool try_enqueue(T const& item)
|
||||
{
|
||||
if (inner.try_enqueue(item)) {
|
||||
sema->signal();
|
||||
return true;
|
||||
}
|
||||
return false;
|
||||
}
|
||||
|
||||
// Enqueues a single item (by moving it, if possible).
|
||||
// Does not allocate memory (except for one-time implicit producer).
|
||||
// Fails if not enough room to enqueue (or implicit production is
|
||||
// disabled because Traits::INITIAL_IMPLICIT_PRODUCER_HASH_SIZE is 0).
|
||||
// Thread-safe.
|
||||
inline bool try_enqueue(T&& item)
|
||||
{
|
||||
if (inner.try_enqueue(std::move(item))) {
|
||||
sema->signal();
|
||||
return true;
|
||||
}
|
||||
return false;
|
||||
}
|
||||
|
||||
// Enqueues a single item (by copying it) using an explicit producer token.
|
||||
// Does not allocate memory. Fails if not enough room to enqueue.
|
||||
// Thread-safe.
|
||||
inline bool try_enqueue(producer_token_t const& token, T const& item)
|
||||
{
|
||||
if (inner.try_enqueue(token, item)) {
|
||||
sema->signal();
|
||||
return true;
|
||||
}
|
||||
return false;
|
||||
}
|
||||
|
||||
// Enqueues a single item (by moving it, if possible) using an explicit producer token.
|
||||
// Does not allocate memory. Fails if not enough room to enqueue.
|
||||
// Thread-safe.
|
||||
inline bool try_enqueue(producer_token_t const& token, T&& item)
|
||||
{
|
||||
if (inner.try_enqueue(token, std::move(item))) {
|
||||
sema->signal();
|
||||
return true;
|
||||
}
|
||||
return false;
|
||||
}
|
||||
|
||||
// Enqueues several items.
|
||||
// Does not allocate memory (except for one-time implicit producer).
|
||||
// Fails if not enough room to enqueue (or implicit production is
|
||||
// disabled because Traits::INITIAL_IMPLICIT_PRODUCER_HASH_SIZE is 0).
|
||||
// Note: Use std::make_move_iterator if the elements should be moved
|
||||
// instead of copied.
|
||||
// Thread-safe.
|
||||
template<typename It>
|
||||
inline bool try_enqueue_bulk(It itemFirst, size_t count)
|
||||
{
|
||||
if (inner.try_enqueue_bulk(std::forward<It>(itemFirst), count)) {
|
||||
sema->signal((LightweightSemaphore::ssize_t)(ssize_t)count);
|
||||
return true;
|
||||
}
|
||||
return false;
|
||||
}
|
||||
|
||||
// Enqueues several items using an explicit producer token.
|
||||
// Does not allocate memory. Fails if not enough room to enqueue.
|
||||
// Note: Use std::make_move_iterator if the elements should be moved
|
||||
// instead of copied.
|
||||
// Thread-safe.
|
||||
template<typename It>
|
||||
inline bool try_enqueue_bulk(producer_token_t const& token, It itemFirst, size_t count)
|
||||
{
|
||||
if (inner.try_enqueue_bulk(token, std::forward<It>(itemFirst), count)) {
|
||||
sema->signal((LightweightSemaphore::ssize_t)(ssize_t)count);
|
||||
return true;
|
||||
}
|
||||
return false;
|
||||
}
|
||||
|
||||
|
||||
// Attempts to dequeue from the queue.
|
||||
// Returns false if all producer streams appeared empty at the time they
|
||||
// were checked (so, the queue is likely but not guaranteed to be empty).
|
||||
// Never allocates. Thread-safe.
|
||||
template<typename U>
|
||||
inline bool try_dequeue(U& item)
|
||||
{
|
||||
if (sema->tryWait()) {
|
||||
while (!inner.try_dequeue(item)) {
|
||||
continue;
|
||||
}
|
||||
return true;
|
||||
}
|
||||
return false;
|
||||
}
|
||||
|
||||
// Attempts to dequeue from the queue using an explicit consumer token.
|
||||
// Returns false if all producer streams appeared empty at the time they
|
||||
// were checked (so, the queue is likely but not guaranteed to be empty).
|
||||
// Never allocates. Thread-safe.
|
||||
template<typename U>
|
||||
inline bool try_dequeue(consumer_token_t& token, U& item)
|
||||
{
|
||||
if (sema->tryWait()) {
|
||||
while (!inner.try_dequeue(token, item)) {
|
||||
continue;
|
||||
}
|
||||
return true;
|
||||
}
|
||||
return false;
|
||||
}
|
||||
|
||||
// Attempts to dequeue several elements from the queue.
|
||||
// Returns the number of items actually dequeued.
|
||||
// Returns 0 if all producer streams appeared empty at the time they
|
||||
// were checked (so, the queue is likely but not guaranteed to be empty).
|
||||
// Never allocates. Thread-safe.
|
||||
template<typename It>
|
||||
inline size_t try_dequeue_bulk(It itemFirst, size_t max)
|
||||
{
|
||||
size_t count = 0;
|
||||
max = (size_t)sema->tryWaitMany((LightweightSemaphore::ssize_t)(ssize_t)max);
|
||||
while (count != max) {
|
||||
count += inner.template try_dequeue_bulk<It&>(itemFirst, max - count);
|
||||
}
|
||||
return count;
|
||||
}
|
||||
|
||||
// Attempts to dequeue several elements from the queue using an explicit consumer token.
|
||||
// Returns the number of items actually dequeued.
|
||||
// Returns 0 if all producer streams appeared empty at the time they
|
||||
// were checked (so, the queue is likely but not guaranteed to be empty).
|
||||
// Never allocates. Thread-safe.
|
||||
template<typename It>
|
||||
inline size_t try_dequeue_bulk(consumer_token_t& token, It itemFirst, size_t max)
|
||||
{
|
||||
size_t count = 0;
|
||||
max = (size_t)sema->tryWaitMany((LightweightSemaphore::ssize_t)(ssize_t)max);
|
||||
while (count != max) {
|
||||
count += inner.template try_dequeue_bulk<It&>(token, itemFirst, max - count);
|
||||
}
|
||||
return count;
|
||||
}
|
||||
|
||||
|
||||
|
||||
// Blocks the current thread until there's something to dequeue, then
|
||||
// dequeues it.
|
||||
// Never allocates. Thread-safe.
|
||||
template<typename U>
|
||||
inline void wait_dequeue(U& item)
|
||||
{
|
||||
sema->wait();
|
||||
while (!inner.try_dequeue(item)) {
|
||||
continue;
|
||||
}
|
||||
}
|
||||
|
||||
// Blocks the current thread until either there's something to dequeue
|
||||
// or the timeout (specified in microseconds) expires. Returns false
|
||||
// without setting `item` if the timeout expires, otherwise assigns
|
||||
// to `item` and returns true.
|
||||
// Using a negative timeout indicates an indefinite timeout,
|
||||
// and is thus functionally equivalent to calling wait_dequeue.
|
||||
// Never allocates. Thread-safe.
|
||||
template<typename U>
|
||||
inline bool wait_dequeue_timed(U& item, std::int64_t timeout_usecs)
|
||||
{
|
||||
if (!sema->wait(timeout_usecs)) {
|
||||
return false;
|
||||
}
|
||||
while (!inner.try_dequeue(item)) {
|
||||
continue;
|
||||
}
|
||||
return true;
|
||||
}
|
||||
|
||||
// Blocks the current thread until either there's something to dequeue
|
||||
// or the timeout expires. Returns false without setting `item` if the
|
||||
// timeout expires, otherwise assigns to `item` and returns true.
|
||||
// Never allocates. Thread-safe.
|
||||
template<typename U, typename Rep, typename Period>
|
||||
inline bool wait_dequeue_timed(U& item, std::chrono::duration<Rep, Period> const& timeout)
|
||||
{
|
||||
return wait_dequeue_timed(item, std::chrono::duration_cast<std::chrono::microseconds>(timeout).count());
|
||||
}
|
||||
|
||||
// Blocks the current thread until there's something to dequeue, then
|
||||
// dequeues it using an explicit consumer token.
|
||||
// Never allocates. Thread-safe.
|
||||
template<typename U>
|
||||
inline void wait_dequeue(consumer_token_t& token, U& item)
|
||||
{
|
||||
sema->wait();
|
||||
while (!inner.try_dequeue(token, item)) {
|
||||
continue;
|
||||
}
|
||||
}
|
||||
|
||||
// Blocks the current thread until either there's something to dequeue
|
||||
// or the timeout (specified in microseconds) expires. Returns false
|
||||
// without setting `item` if the timeout expires, otherwise assigns
|
||||
// to `item` and returns true.
|
||||
// Using a negative timeout indicates an indefinite timeout,
|
||||
// and is thus functionally equivalent to calling wait_dequeue.
|
||||
// Never allocates. Thread-safe.
|
||||
template<typename U>
|
||||
inline bool wait_dequeue_timed(consumer_token_t& token, U& item, std::int64_t timeout_usecs)
|
||||
{
|
||||
if (!sema->wait(timeout_usecs)) {
|
||||
return false;
|
||||
}
|
||||
while (!inner.try_dequeue(token, item)) {
|
||||
continue;
|
||||
}
|
||||
return true;
|
||||
}
|
||||
|
||||
// Blocks the current thread until either there's something to dequeue
|
||||
// or the timeout expires. Returns false without setting `item` if the
|
||||
// timeout expires, otherwise assigns to `item` and returns true.
|
||||
// Never allocates. Thread-safe.
|
||||
template<typename U, typename Rep, typename Period>
|
||||
inline bool wait_dequeue_timed(consumer_token_t& token, U& item, std::chrono::duration<Rep, Period> const& timeout)
|
||||
{
|
||||
return wait_dequeue_timed(token, item, std::chrono::duration_cast<std::chrono::microseconds>(timeout).count());
|
||||
}
|
||||
|
||||
// Attempts to dequeue several elements from the queue.
|
||||
// Returns the number of items actually dequeued, which will
|
||||
// always be at least one (this method blocks until the queue
|
||||
// is non-empty) and at most max.
|
||||
// Never allocates. Thread-safe.
|
||||
template<typename It>
|
||||
inline size_t wait_dequeue_bulk(It itemFirst, size_t max)
|
||||
{
|
||||
size_t count = 0;
|
||||
max = (size_t)sema->waitMany((LightweightSemaphore::ssize_t)(ssize_t)max);
|
||||
while (count != max) {
|
||||
count += inner.template try_dequeue_bulk<It&>(itemFirst, max - count);
|
||||
}
|
||||
return count;
|
||||
}
|
||||
|
||||
// Attempts to dequeue several elements from the queue.
|
||||
// Returns the number of items actually dequeued, which can
|
||||
// be 0 if the timeout expires while waiting for elements,
|
||||
// and at most max.
|
||||
// Using a negative timeout indicates an indefinite timeout,
|
||||
// and is thus functionally equivalent to calling wait_dequeue_bulk.
|
||||
// Never allocates. Thread-safe.
|
||||
template<typename It>
|
||||
inline size_t wait_dequeue_bulk_timed(It itemFirst, size_t max, std::int64_t timeout_usecs)
|
||||
{
|
||||
size_t count = 0;
|
||||
max = (size_t)sema->waitMany((LightweightSemaphore::ssize_t)(ssize_t)max, timeout_usecs);
|
||||
while (count != max) {
|
||||
count += inner.template try_dequeue_bulk<It&>(itemFirst, max - count);
|
||||
}
|
||||
return count;
|
||||
}
|
||||
|
||||
// Attempts to dequeue several elements from the queue.
|
||||
// Returns the number of items actually dequeued, which can
|
||||
// be 0 if the timeout expires while waiting for elements,
|
||||
// and at most max.
|
||||
// Never allocates. Thread-safe.
|
||||
template<typename It, typename Rep, typename Period>
|
||||
inline size_t wait_dequeue_bulk_timed(It itemFirst, size_t max, std::chrono::duration<Rep, Period> const& timeout)
|
||||
{
|
||||
return wait_dequeue_bulk_timed<It&>(itemFirst, max, std::chrono::duration_cast<std::chrono::microseconds>(timeout).count());
|
||||
}
|
||||
|
||||
// Attempts to dequeue several elements from the queue using an explicit consumer token.
|
||||
// Returns the number of items actually dequeued, which will
|
||||
// always be at least one (this method blocks until the queue
|
||||
// is non-empty) and at most max.
|
||||
// Never allocates. Thread-safe.
|
||||
template<typename It>
|
||||
inline size_t wait_dequeue_bulk(consumer_token_t& token, It itemFirst, size_t max)
|
||||
{
|
||||
size_t count = 0;
|
||||
max = (size_t)sema->waitMany((LightweightSemaphore::ssize_t)(ssize_t)max);
|
||||
while (count != max) {
|
||||
count += inner.template try_dequeue_bulk<It&>(token, itemFirst, max - count);
|
||||
}
|
||||
return count;
|
||||
}
|
||||
|
||||
// Attempts to dequeue several elements from the queue using an explicit consumer token.
|
||||
// Returns the number of items actually dequeued, which can
|
||||
// be 0 if the timeout expires while waiting for elements,
|
||||
// and at most max.
|
||||
// Using a negative timeout indicates an indefinite timeout,
|
||||
// and is thus functionally equivalent to calling wait_dequeue_bulk.
|
||||
// Never allocates. Thread-safe.
|
||||
template<typename It>
|
||||
inline size_t wait_dequeue_bulk_timed(consumer_token_t& token, It itemFirst, size_t max, std::int64_t timeout_usecs)
|
||||
{
|
||||
size_t count = 0;
|
||||
max = (size_t)sema->waitMany((LightweightSemaphore::ssize_t)(ssize_t)max, timeout_usecs);
|
||||
while (count != max) {
|
||||
count += inner.template try_dequeue_bulk<It&>(token, itemFirst, max - count);
|
||||
}
|
||||
return count;
|
||||
}
|
||||
|
||||
// Attempts to dequeue several elements from the queue using an explicit consumer token.
|
||||
// Returns the number of items actually dequeued, which can
|
||||
// be 0 if the timeout expires while waiting for elements,
|
||||
// and at most max.
|
||||
// Never allocates. Thread-safe.
|
||||
template<typename It, typename Rep, typename Period>
|
||||
inline size_t wait_dequeue_bulk_timed(consumer_token_t& token, It itemFirst, size_t max, std::chrono::duration<Rep, Period> const& timeout)
|
||||
{
|
||||
return wait_dequeue_bulk_timed<It&>(token, itemFirst, max, std::chrono::duration_cast<std::chrono::microseconds>(timeout).count());
|
||||
}
|
||||
|
||||
|
||||
// Returns an estimate of the total number of elements currently in the queue. This
|
||||
// estimate is only accurate if the queue has completely stabilized before it is called
|
||||
// (i.e. all enqueue and dequeue operations have completed and their memory effects are
|
||||
// visible on the calling thread, and no further operations start while this method is
|
||||
// being called).
|
||||
// Thread-safe.
|
||||
inline size_t size_approx() const
|
||||
{
|
||||
return (size_t)sema->availableApprox();
|
||||
}
|
||||
|
||||
|
||||
// Returns true if the underlying atomic variables used by
|
||||
// the queue are lock-free (they should be on most platforms).
|
||||
// Thread-safe.
|
||||
static bool is_lock_free()
|
||||
{
|
||||
return ConcurrentQueue::is_lock_free();
|
||||
}
|
||||
|
||||
|
||||
private:
|
||||
template<typename U>
|
||||
static inline U* create()
|
||||
{
|
||||
auto p = (Traits::malloc)(sizeof(U));
|
||||
return p != nullptr ? new (p) U : nullptr;
|
||||
}
|
||||
|
||||
template<typename U, typename A1>
|
||||
static inline U* create(A1&& a1)
|
||||
{
|
||||
auto p = (Traits::malloc)(sizeof(U));
|
||||
return p != nullptr ? new (p) U(std::forward<A1>(a1)) : nullptr;
|
||||
}
|
||||
|
||||
template<typename U>
|
||||
static inline void destroy(U* p)
|
||||
{
|
||||
if (p != nullptr) {
|
||||
p->~U();
|
||||
}
|
||||
(Traits::free)(p);
|
||||
}
|
||||
|
||||
private:
|
||||
ConcurrentQueue inner;
|
||||
std::unique_ptr<LightweightSemaphore, void (*)(LightweightSemaphore*)> sema;
|
||||
};
|
||||
|
||||
|
||||
template<typename T, typename Traits>
|
||||
inline void swap(BlockingConcurrentQueue<T, Traits>& a, BlockingConcurrentQueue<T, Traits>& b) MOODYCAMEL_NOEXCEPT
|
||||
{
|
||||
a.swap(b);
|
||||
}
|
||||
|
||||
} // end namespace moodycamel
|
||||
+3635
File diff suppressed because it is too large
Load Diff
@@ -0,0 +1,87 @@
|
||||
#pragma once
|
||||
|
||||
//#define MCDBGQ_TRACKMEM 1
|
||||
//#define MCDBGQ_NOLOCKFREE_FREELIST 1
|
||||
//#define MCDBGQ_USEDEBUGFREELIST 1
|
||||
//#define MCDBGQ_NOLOCKFREE_IMPLICITPRODBLOCKINDEX 1
|
||||
//#define MCDBGQ_NOLOCKFREE_IMPLICITPRODHASH 1
|
||||
|
||||
#if defined(_WIN32) || defined(__WINDOWS__) || defined(__WIN32__)
|
||||
#define WIN32_LEAN_AND_MEAN
|
||||
#include <windows.h>
|
||||
namespace moodycamel { namespace debug {
|
||||
struct DebugMutex {
|
||||
DebugMutex() { InitializeCriticalSectionAndSpinCount(&cs, 0x400); }
|
||||
~DebugMutex() { DeleteCriticalSection(&cs); }
|
||||
|
||||
void lock() { EnterCriticalSection(&cs); }
|
||||
void unlock() { LeaveCriticalSection(&cs); }
|
||||
|
||||
private:
|
||||
CRITICAL_SECTION cs;
|
||||
};
|
||||
} }
|
||||
#else
|
||||
#include <mutex>
|
||||
namespace moodycamel { namespace debug {
|
||||
struct DebugMutex {
|
||||
void lock() { m.lock(); }
|
||||
void unlock() { m.unlock(); }
|
||||
|
||||
private:
|
||||
std::mutex m;
|
||||
};
|
||||
} }
|
||||
#define
|
||||
#endif
|
||||
|
||||
namespace moodycamel { namespace debug {
|
||||
struct DebugLock {
|
||||
explicit DebugLock(DebugMutex& mutex)
|
||||
: mutex(mutex)
|
||||
{
|
||||
mutex.lock();
|
||||
}
|
||||
|
||||
~DebugLock()
|
||||
{
|
||||
mutex.unlock();
|
||||
}
|
||||
|
||||
private:
|
||||
DebugMutex& mutex;
|
||||
};
|
||||
|
||||
|
||||
template<typename N>
|
||||
struct DebugFreeList {
|
||||
DebugFreeList() : head(nullptr) { }
|
||||
DebugFreeList(DebugFreeList&& other) : head(other.head) { other.head = nullptr; }
|
||||
void swap(DebugFreeList& other) { std::swap(head, other.head); }
|
||||
|
||||
inline void add(N* node)
|
||||
{
|
||||
DebugLock lock(mutex);
|
||||
node->freeListNext = head;
|
||||
head = node;
|
||||
}
|
||||
|
||||
inline N* try_get()
|
||||
{
|
||||
DebugLock lock(mutex);
|
||||
if (head == nullptr) {
|
||||
return nullptr;
|
||||
}
|
||||
|
||||
auto prevHead = head;
|
||||
head = head->freeListNext;
|
||||
return prevHead;
|
||||
}
|
||||
|
||||
N* head_unsafe() const { return head; }
|
||||
|
||||
private:
|
||||
N* head;
|
||||
DebugMutex mutex;
|
||||
};
|
||||
} }
|
||||
+375
@@ -0,0 +1,375 @@
|
||||
# Samples for moodycamel::ConcurrentQueue
|
||||
|
||||
Here are some example usage scenarios with sample code. Note that most
|
||||
use the simplest version of each available method for demonstration purposes,
|
||||
but they can all be adapted to use tokens and/or the corresponding bulk methods for
|
||||
extra speed.
|
||||
|
||||
|
||||
## Hello queue
|
||||
|
||||
ConcurrentQueue<int> q;
|
||||
|
||||
for (int i = 0; i != 123; ++i)
|
||||
q.enqueue(i);
|
||||
|
||||
int item;
|
||||
for (int i = 0; i != 123; ++i) {
|
||||
q.try_dequeue(item);
|
||||
assert(item == i);
|
||||
}
|
||||
|
||||
|
||||
## Hello concurrency
|
||||
|
||||
Basic example of how to use the queue from multiple threads, with no
|
||||
particular goal (i.e. it does nothing, but in an instructive way).
|
||||
|
||||
ConcurrentQueue<int> q;
|
||||
int dequeued[100] = { 0 };
|
||||
std::thread threads[20];
|
||||
|
||||
// Producers
|
||||
for (int i = 0; i != 10; ++i) {
|
||||
threads[i] = std::thread([&](int i) {
|
||||
for (int j = 0; j != 10; ++j) {
|
||||
q.enqueue(i * 10 + j);
|
||||
}
|
||||
}, i);
|
||||
}
|
||||
|
||||
// Consumers
|
||||
for (int i = 10; i != 20; ++i) {
|
||||
threads[i] = std::thread([&]() {
|
||||
int item;
|
||||
for (int j = 0; j != 20; ++j) {
|
||||
if (q.try_dequeue(item)) {
|
||||
++dequeued[item];
|
||||
}
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
// Wait for all threads
|
||||
for (int i = 0; i != 20; ++i) {
|
||||
threads[i].join();
|
||||
}
|
||||
|
||||
// Collect any leftovers (could be some if e.g. consumers finish before producers)
|
||||
int item;
|
||||
while (q.try_dequeue(item)) {
|
||||
++dequeued[item];
|
||||
}
|
||||
|
||||
// Make sure everything went in and came back out!
|
||||
for (int i = 0; i != 100; ++i) {
|
||||
assert(dequeued[i] == 1);
|
||||
}
|
||||
|
||||
|
||||
## Bulk up
|
||||
|
||||
Same as previous example, but runs faster.
|
||||
|
||||
ConcurrentQueue<int> q;
|
||||
int dequeued[100] = { 0 };
|
||||
std::thread threads[20];
|
||||
|
||||
// Producers
|
||||
for (int i = 0; i != 10; ++i) {
|
||||
threads[i] = std::thread([&](int i) {
|
||||
int items[10];
|
||||
for (int j = 0; j != 10; ++j) {
|
||||
items[j] = i * 10 + j;
|
||||
}
|
||||
q.enqueue_bulk(items, 10);
|
||||
}, i);
|
||||
}
|
||||
|
||||
// Consumers
|
||||
for (int i = 10; i != 20; ++i) {
|
||||
threads[i] = std::thread([&]() {
|
||||
int items[20];
|
||||
for (std::size_t count = q.try_dequeue_bulk(items, 20); count != 0; --count) {
|
||||
++dequeued[items[count - 1]];
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
// Wait for all threads
|
||||
for (int i = 0; i != 20; ++i) {
|
||||
threads[i].join();
|
||||
}
|
||||
|
||||
// Collect any leftovers (could be some if e.g. consumers finish before producers)
|
||||
int items[10];
|
||||
std::size_t count;
|
||||
while ((count = q.try_dequeue_bulk(items, 10)) != 0) {
|
||||
for (std::size_t i = 0; i != count; ++i) {
|
||||
++dequeued[items[i]];
|
||||
}
|
||||
}
|
||||
|
||||
// Make sure everything went in and came back out!
|
||||
for (int i = 0; i != 100; ++i) {
|
||||
assert(dequeued[i] == 1);
|
||||
}
|
||||
|
||||
|
||||
## Producer/consumer model (simultaneous)
|
||||
|
||||
In this model, one set of threads is producing items,
|
||||
and the other is consuming them concurrently until all of
|
||||
them have been consumed. The counters are required to
|
||||
ensure that all items eventually get consumed.
|
||||
|
||||
ConcurrentQueue<Item> q;
|
||||
const int ProducerCount = 8;
|
||||
const int ConsumerCount = 8;
|
||||
std::thread producers[ProducerCount];
|
||||
std::thread consumers[ConsumerCount];
|
||||
std::atomic<int> doneProducers(0);
|
||||
std::atomic<int> doneConsumers(0);
|
||||
for (int i = 0; i != ProducerCount; ++i) {
|
||||
producers[i] = std::thread([&]() {
|
||||
while (produce) {
|
||||
q.enqueue(produceItem());
|
||||
}
|
||||
doneProducers.fetch_add(1, std::memory_order_release);
|
||||
});
|
||||
}
|
||||
for (int i = 0; i != ConsumerCount; ++i) {
|
||||
consumers[i] = std::thread([&]() {
|
||||
Item item;
|
||||
bool itemsLeft;
|
||||
do {
|
||||
// It's important to fence (if the producers have finished) *before* dequeueing
|
||||
itemsLeft = doneProducers.load(std::memory_order_acquire) != ProducerCount;
|
||||
while (q.try_dequeue(item)) {
|
||||
itemsLeft = true;
|
||||
consumeItem(item);
|
||||
}
|
||||
} while (itemsLeft || doneConsumers.fetch_add(1, std::memory_order_acq_rel) + 1 == ConsumerCount);
|
||||
// The condition above is a bit tricky, but it's necessary to ensure that the
|
||||
// last consumer sees the memory effects of all the other consumers before it
|
||||
// calls try_dequeue for the last time
|
||||
});
|
||||
}
|
||||
for (int i = 0; i != ProducerCount; ++i) {
|
||||
producers[i].join();
|
||||
}
|
||||
for (int i = 0; i != ConsumerCount; ++i) {
|
||||
consumers[i].join();
|
||||
}
|
||||
|
||||
## Producer/consumer model (simultaneous, blocking)
|
||||
|
||||
The blocking version is different, since either the number of elements being produced needs
|
||||
to be known ahead of time, or some other coordination is required to tell the consumers when
|
||||
to stop calling wait_dequeue (not shown here). This is necessary because otherwise a consumer
|
||||
could end up blocking forever -- and destroying a queue while a consumer is blocking on it leads
|
||||
to undefined behaviour.
|
||||
|
||||
BlockingConcurrentQueue<Item> q;
|
||||
const int ProducerCount = 8;
|
||||
const int ConsumerCount = 8;
|
||||
std::thread producers[ProducerCount];
|
||||
std::thread consumers[ConsumerCount];
|
||||
std::atomic<int> promisedElementsRemaining(ProducerCount * 1000);
|
||||
for (int i = 0; i != ProducerCount; ++i) {
|
||||
producers[i] = std::thread([&]() {
|
||||
for (int j = 0; j != 1000; ++j) {
|
||||
q.enqueue(produceItem());
|
||||
}
|
||||
});
|
||||
}
|
||||
for (int i = 0; i != ConsumerCount; ++i) {
|
||||
consumers[i] = std::thread([&]() {
|
||||
Item item;
|
||||
while (promisedElementsRemaining.fetch_sub(1, std::memory_order_relaxed)) {
|
||||
q.wait_dequeue(item);
|
||||
consumeItem(item);
|
||||
}
|
||||
});
|
||||
}
|
||||
for (int i = 0; i != ProducerCount; ++i) {
|
||||
producers[i].join();
|
||||
}
|
||||
for (int i = 0; i != ConsumerCount; ++i) {
|
||||
consumers[i].join();
|
||||
}
|
||||
|
||||
|
||||
## Producer/consumer model (separate stages)
|
||||
|
||||
ConcurrentQueue<Item> q;
|
||||
|
||||
// Production stage
|
||||
std::thread threads[8];
|
||||
for (int i = 0; i != 8; ++i) {
|
||||
threads[i] = std::thread([&]() {
|
||||
while (produce) {
|
||||
q.enqueue(produceItem());
|
||||
}
|
||||
});
|
||||
}
|
||||
for (int i = 0; i != 8; ++i) {
|
||||
threads[i].join();
|
||||
}
|
||||
|
||||
// Consumption stage
|
||||
std::atomic<int> doneConsumers(0);
|
||||
for (int i = 0; i != 8; ++i) {
|
||||
threads[i] = std::thread([&]() {
|
||||
Item item;
|
||||
do {
|
||||
while (q.try_dequeue(item)) {
|
||||
consumeItem(item);
|
||||
}
|
||||
// Loop again one last time if we're the last producer (with the acquired
|
||||
// memory effects of the other producers):
|
||||
} while (doneConsumers.fetch_add(1, std::memory_order_acq_rel) + 1 == 8);
|
||||
});
|
||||
}
|
||||
for (int i = 0; i != 8; ++i) {
|
||||
threads[i].join();
|
||||
}
|
||||
|
||||
Note that there's no point trying to use the blocking queue with this model, since
|
||||
there's no need to use the `wait` methods (all the elements are produced before any
|
||||
are consumed), and hence the complexity would be the same but with additional overhead.
|
||||
|
||||
|
||||
## Object pool
|
||||
|
||||
If you don't know what threads will be using the queue in advance,
|
||||
you can't really declare any long-term tokens. The obvious solution
|
||||
is to use the implicit methods (that don't take any tokens):
|
||||
|
||||
// A pool of 'Something' objects that can be safely accessed
|
||||
// from any thread
|
||||
class SomethingPool
|
||||
{
|
||||
public:
|
||||
Something getSomething()
|
||||
{
|
||||
Something obj;
|
||||
queue.try_dequeue(obj);
|
||||
|
||||
// If the dequeue succeeded, obj will be an object from the
|
||||
// thread pool, otherwise it will be the default-constructed
|
||||
// object as declared above
|
||||
return obj;
|
||||
}
|
||||
|
||||
void recycleSomething(Something&& obj)
|
||||
{
|
||||
queue.enqueue(std::move(obj));
|
||||
}
|
||||
};
|
||||
|
||||
|
||||
## Threadpool task queue
|
||||
|
||||
BlockingConcurrentQueue<Task> q;
|
||||
|
||||
// To create a task from any thread:
|
||||
q.enqueue(...);
|
||||
|
||||
// On threadpool threads:
|
||||
Task task;
|
||||
while (true) {
|
||||
q.wait_dequeue(task);
|
||||
|
||||
// Process task...
|
||||
}
|
||||
|
||||
|
||||
## Multithreaded game loop
|
||||
|
||||
BlockingConcurrentQueue<Task> q;
|
||||
std::atomic<int> pendingTasks(0);
|
||||
|
||||
// On threadpool threads:
|
||||
Task task;
|
||||
while (true) {
|
||||
q.wait_dequeue(task);
|
||||
|
||||
// Process task...
|
||||
|
||||
pendingTasks.fetch_add(-1, std::memory_order_release);
|
||||
}
|
||||
|
||||
// Whenever a new task needs to be processed for the frame:
|
||||
pendingTasks.fetch_add(1, std::memory_order_release);
|
||||
q.enqueue(...);
|
||||
|
||||
// To wait for all the frame's tasks to complete before rendering:
|
||||
while (pendingTasks.load(std::memory_order_acquire) != 0)
|
||||
continue;
|
||||
|
||||
// Alternatively you could help out the thread pool while waiting:
|
||||
while (pendingTasks.load(std::memory_order_acquire) != 0) {
|
||||
if (!q.try_dequeue(task)) {
|
||||
continue;
|
||||
}
|
||||
|
||||
// Process task...
|
||||
|
||||
pendingTasks.fetch_add(-1, std::memory_order_release);
|
||||
}
|
||||
|
||||
|
||||
## Pump until empty
|
||||
|
||||
This might be useful if, for example, you want to process any remaining items
|
||||
in the queue before it's destroyed. Note that it is your responsibility
|
||||
to ensure that the memory effects of any enqueue operations you wish to see on
|
||||
the dequeue thread are visible (i.e. if you're waiting for a certain set of elements,
|
||||
you need to use memory fences to ensure that those elements are visible to the dequeue
|
||||
thread after they've been enqueued).
|
||||
|
||||
ConcurrentQueue<Item> q;
|
||||
|
||||
// Single-threaded pumping:
|
||||
Item item;
|
||||
while (q.try_dequeue(item)) {
|
||||
// Process item...
|
||||
}
|
||||
// q is guaranteed to be empty here, unless there is another thread enqueueing still or
|
||||
// there was another thread dequeueing at one point and its memory effects have not
|
||||
// yet been propagated to this thread.
|
||||
|
||||
// Multi-threaded pumping:
|
||||
std::thread threads[8];
|
||||
std::atomic<int> doneConsumers(0);
|
||||
for (int i = 0; i != 8; ++i) {
|
||||
threads[i] = std::thread([&]() {
|
||||
Item item;
|
||||
do {
|
||||
while (q.try_dequeue(item)) {
|
||||
// Process item...
|
||||
}
|
||||
} while (doneConsumers.fetch_add(1, std::memory_order_acq_rel) + 1 == 8);
|
||||
// If there are still enqueue operations happening on other threads,
|
||||
// then the queue may not be empty at this point. However, if all enqueue
|
||||
// operations completed before we finished pumping (and the propagation of
|
||||
// their memory effects too), and all dequeue operations apart from those
|
||||
// our threads did above completed before we finished pumping (and the
|
||||
// propagation of their memory effects too), then the queue is guaranteed
|
||||
// to be empty at this point.
|
||||
});
|
||||
}
|
||||
for (int i = 0; i != 8; ++i) {
|
||||
threads[i].join();
|
||||
}
|
||||
|
||||
|
||||
## Wait for a queue to become empty (without dequeueing)
|
||||
|
||||
You can't (robustly) :-) However, you can set up your own atomic counter and
|
||||
poll that instead (see the game loop example). If you're satisfied with merely an estimate, you can use
|
||||
`size_approx()`. Note that `size_approx()` may return 0 even if the queue is
|
||||
not completely empty, unless the queue has already stabilized first (no threads
|
||||
are enqueueing or dequeueing, and all memory effects of any previous operations
|
||||
have been propagated to the thread before it calls `size_approx()`).
|
||||
Reference in New Issue
Block a user