857be9459e
- allows all methods to be called on a `shared(Channel!T)` instance. - `Channel` pre-defines the `buffer_size` argument to 100, matching `createChannel`
261 lines
6.9 KiB
D
261 lines
6.9 KiB
D
/** Implements a thread-safe, typed producer-consumer queue.
|
|
|
|
Copyright: © 2017-2019 RejectedSoftware e.K.
|
|
Authors: Sönke Ludwig
|
|
License: Subject to the terms of the MIT license, as written in the included LICENSE.txt file.
|
|
*/
|
|
module vibe.core.channel;
|
|
|
|
import vibe.core.sync : TaskCondition;
|
|
import vibe.internal.array : FixedRingBuffer;
|
|
|
|
import std.algorithm.mutation : move, swap;
|
|
import std.exception : enforce;
|
|
import core.sync.mutex;
|
|
|
|
// multiple producers allowed, multiple consumers allowed - Q: should this be restricted to allow higher performance? maybe configurable?
|
|
// currently always buffered - TODO: implement blocking non-buffered mode
|
|
// TODO: implement a multi-channel wait, e.g.
|
|
// TaggedAlgebraic!(...) consumeAny(ch1, ch2, ch3); - requires a waitOnMultipleConditions function
|
|
|
|
|
|
/** Creates a new channel suitable for cross-task and cross-thread communication.
|
|
*/
|
|
Channel!(T, buffer_size) createChannel(T, size_t buffer_size = 100)()
|
|
{
|
|
Channel!(T, buffer_size) ret;
|
|
ret.m_impl = new shared ChannelImpl!(T, buffer_size);
|
|
return ret;
|
|
}
|
|
|
|
|
|
/** Thread-safe typed data channel implementation.
|
|
|
|
The implementation supports multiple-reader-multiple-writer operation across
|
|
multiple tasks in multiple threads.
|
|
*/
|
|
struct Channel(T, size_t buffer_size = 100) {
|
|
enum bufferSize = buffer_size;
|
|
|
|
private shared ChannelImpl!(T, buffer_size) m_impl;
|
|
|
|
/** Determines whether there is more data to read.
|
|
|
|
This property is empty $(I iff) no more elements are in the internal
|
|
buffer and `close()` has been called. Once the channel is empty,
|
|
subsequent calls to `consumeOne` or `consumeAll` will throw an
|
|
exception.
|
|
|
|
Note that relying on the return value to determine whether another
|
|
element can be read is only safe in a single-reader scenario. Use
|
|
`tryConsumeOne` in a multiple-reader scenario instead.
|
|
*/
|
|
@property bool empty() { return m_impl.empty; }
|
|
/// ditto
|
|
@property bool empty() shared { return m_impl.empty; }
|
|
|
|
/** Closes the channel.
|
|
|
|
A closed channel does not accept any new items enqueued using `put` and
|
|
causes `empty` to return `fals` as soon as all preceeding elements have
|
|
been consumed.
|
|
*/
|
|
void close() { m_impl.close(); }
|
|
/// ditto
|
|
void close() shared { m_impl.close(); }
|
|
|
|
/** Consumes a single element off the queue.
|
|
|
|
This function will block if no elements are available. If the `empty`
|
|
property is `true`, an exception will be thrown.
|
|
*/
|
|
T consumeOne() { return m_impl.consumeOne(); }
|
|
/// ditto
|
|
T consumeOne() shared { return m_impl.consumeOne(); }
|
|
|
|
/** Attempts to consume a single element.
|
|
|
|
If no more elements are available and the channel has been closed,
|
|
`false` is returned and `dst` is left untouched.
|
|
*/
|
|
bool tryConsumeOne(ref T dst) { return m_impl.tryConsumeOne(dst); }
|
|
/// ditto
|
|
bool tryConsumeOne(ref T dst) shared { return m_impl.tryConsumeOne(dst); }
|
|
|
|
/** Attempts to consume all elements currently in the queue.
|
|
|
|
This function will block if no elements are available. Once at least one
|
|
element is available, the contents of `dst` will be replaced with all
|
|
available elements.
|
|
|
|
If the `empty` property is or becomes `true` before data becomes
|
|
avaiable, `dst` will be left untouched and `false` is returned.
|
|
*/
|
|
bool consumeAll(ref FixedRingBuffer!(T, buffer_size) dst)
|
|
in { assert(dst.empty); }
|
|
body { return m_impl.consumeAll(dst); }
|
|
/// ditto
|
|
bool consumeAll(ref FixedRingBuffer!(T, buffer_size) dst) shared
|
|
in { assert(dst.empty); }
|
|
body { return m_impl.consumeAll(dst); }
|
|
|
|
/** Enqueues an element.
|
|
|
|
This function may block the the event that the internal buffer is full.
|
|
*/
|
|
void put(T item) { m_impl.put(item.move); }
|
|
/// ditto
|
|
void put(T item) shared { m_impl.put(item.move); }
|
|
}
|
|
|
|
|
|
private final class ChannelImpl(T, size_t buffer_size) {
|
|
import vibe.core.concurrency : isWeaklyIsolated;
|
|
static assert(isWeaklyIsolated!T, "Channel data type "~T.stringof~" is not safe to pass between threads.");
|
|
|
|
private {
|
|
Mutex m_mutex;
|
|
TaskCondition m_condition;
|
|
FixedRingBuffer!(T, buffer_size) m_items;
|
|
bool m_closed = false;
|
|
}
|
|
|
|
this()
|
|
shared {
|
|
m_mutex = cast(shared)new Mutex;
|
|
m_condition = cast(shared)new TaskCondition(cast(Mutex)m_mutex);
|
|
}
|
|
|
|
@property bool empty()
|
|
shared {
|
|
synchronized (m_mutex) {
|
|
auto thisus = () @trusted { return cast(ChannelImpl)this; } ();
|
|
return thisus.m_closed && thisus.m_items.empty;
|
|
}
|
|
}
|
|
|
|
void close()
|
|
shared {
|
|
synchronized (m_mutex) {
|
|
auto thisus = () @trusted { return cast(ChannelImpl)this; } ();
|
|
thisus.m_closed = true;
|
|
thisus.m_condition.notifyAll();
|
|
}
|
|
}
|
|
|
|
bool tryConsumeOne(ref T dst)
|
|
shared {
|
|
auto thisus = () @trusted { return cast(ChannelImpl)this; } ();
|
|
bool was_full = false;
|
|
|
|
synchronized (m_mutex) {
|
|
while (thisus.m_items.empty) {
|
|
if (m_closed) return false;
|
|
thisus.m_condition.wait();
|
|
}
|
|
was_full = thisus.m_items.full;
|
|
move(thisus.m_items.front, dst);
|
|
thisus.m_items.popFront();
|
|
}
|
|
|
|
if (was_full) thisus.m_condition.notifyAll();
|
|
|
|
return true;
|
|
}
|
|
|
|
T consumeOne()
|
|
shared {
|
|
auto thisus = () @trusted { return cast(ChannelImpl)this; } ();
|
|
T ret;
|
|
bool was_full = false;
|
|
|
|
synchronized (m_mutex) {
|
|
while (thisus.m_items.empty) {
|
|
if (m_closed) throw new Exception("Attempt to consume from an empty channel.");
|
|
thisus.m_condition.wait();
|
|
}
|
|
was_full = thisus.m_items.full;
|
|
move(thisus.m_items.front, ret);
|
|
thisus.m_items.popFront();
|
|
}
|
|
|
|
if (was_full) thisus.m_condition.notifyAll();
|
|
|
|
return ret.move;
|
|
}
|
|
|
|
bool consumeAll(ref FixedRingBuffer!(T, buffer_size) dst)
|
|
shared {
|
|
auto thisus = () @trusted { return cast(ChannelImpl)this; } ();
|
|
bool was_full = false;
|
|
|
|
synchronized (m_mutex) {
|
|
while (thisus.m_items.empty) {
|
|
if (m_closed) return false;
|
|
thisus.m_condition.wait();
|
|
}
|
|
|
|
was_full = thisus.m_items.full;
|
|
swap(thisus.m_items, dst);
|
|
}
|
|
|
|
if (was_full) thisus.m_condition.notifyAll();
|
|
|
|
return true;
|
|
}
|
|
|
|
void put(T item)
|
|
shared {
|
|
auto thisus = () @trusted { return cast(ChannelImpl)this; } ();
|
|
bool need_notify = false;
|
|
|
|
synchronized (m_mutex) {
|
|
enforce(!m_closed, "Sending on closed channel.");
|
|
while (thisus.m_items.full)
|
|
thisus.m_condition.wait();
|
|
need_notify = thisus.m_items.empty;
|
|
thisus.m_items.put(item.move);
|
|
}
|
|
|
|
if (need_notify) thisus.m_condition.notifyAll();
|
|
}
|
|
}
|
|
|
|
unittest { // test basic operation and non-copyable struct compatiblity
|
|
static struct S {
|
|
int i;
|
|
@disable this(this);
|
|
}
|
|
|
|
auto ch = createChannel!S;
|
|
ch.put(S(1));
|
|
assert(ch.consumeOne().i == 1);
|
|
ch.put(S(4));
|
|
ch.put(S(5));
|
|
{
|
|
FixedRingBuffer!(S, 100) buf;
|
|
ch.consumeAll(buf);
|
|
assert(buf.length == 2);
|
|
assert(buf[0].i == 4);
|
|
assert(buf[1].i == 5);
|
|
}
|
|
ch.put(S(2));
|
|
assert(!ch.empty);
|
|
ch.close();
|
|
assert(!ch.empty);
|
|
S v;
|
|
assert(ch.tryConsumeOne(v));
|
|
assert(v.i == 2);
|
|
assert(ch.empty);
|
|
assert(!ch.tryConsumeOne(v));
|
|
}
|
|
|
|
unittest { // make sure shared(Channel!T) can also be used
|
|
shared ch = createChannel!int;
|
|
ch.put(1);
|
|
assert(!ch.empty);
|
|
assert(ch.consumeOne == 1);
|
|
ch.close();
|
|
assert(ch.empty);
|
|
}
|