From f31db98144bb0338ddab5ece76ac6a192db1d541 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?S=C3=B6nke=20Ludwig?= Date: Mon, 14 Jan 2019 22:16:45 +0100 Subject: [PATCH] Add documentation, basic unit test, createChannel(), close() and tryConsumeOne(). --- source/vibe/core/channel.d | 186 +++++++++++++++++++++++++++++++++---- 1 file changed, 168 insertions(+), 18 deletions(-) diff --git a/source/vibe/core/channel.d b/source/vibe/core/channel.d index e66c295..cb7ed37 100644 --- a/source/vibe/core/channel.d +++ b/source/vibe/core/channel.d @@ -1,27 +1,103 @@ /** Implements a thread-safe, typed producer-consumer queue. - Copyright: © 2017 RejectedSoftware e.K. + 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 -// TODO: implement close() -// TODO: fully support non-copyable types using swap/move where appropriate -// TODO: add unit tests -// Q: Should this be exposed as a class, or as an RC struct? -private final class Channel(T, size_t buffer_size = 100) { + +/** 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) { + 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; } + + /** 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(); } + + /** 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(); } + + /** Attempts to consume a single element. + + If no more elements are available and the channel has been closed, + a null value is returned. + */ + bool tryConsumeOne(ref T dst) { return m_impl.tryConsumeOne(dst); } + + /** Attempts to consume all elements currently in the queue. + + This function will block if no elements are available. If the `empty` + property is `true`, an exception will be thrown. + */ + void consumeAll(ref FixedRingBuffer!(T, buffer_size) dst) + { return m_impl.consumeAll(dst); } + + /** Enqueues an element. + */ + void put(T item) { 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."); - Mutex m_mutex; - TaskCondition m_condition; - FixedRingBuffer!(T, buffer_size) m_items; + private { + Mutex m_mutex; + TaskCondition m_condition; + FixedRingBuffer!(T, buffer_size) m_items; + bool m_closed = false; + } this() shared { @@ -29,50 +105,124 @@ private final class Channel(T, size_t buffer_size = 100) { m_condition = cast(shared)new TaskCondition(cast(Mutex)m_mutex); } - bool empty() + @property bool empty() shared { - synchronized (m_mutex) - return (cast(Channel)this).m_items.empty; + 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 = cast(Channel)this; + auto thisus = () @trusted { return cast(ChannelImpl)this; } (); T ret; bool was_full = false; + synchronized (m_mutex) { - while (thisus.m_items.empty) + 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; - swap(thisus.m_items.front, ret); + move(thisus.m_items.front, ret); + thisus.m_items.popFront(); } + if (was_full) thisus.m_condition.notifyAll(); + return ret.move; } void consumeAll(ref FixedRingBuffer!(T, buffer_size) dst) shared { - auto thisus = cast(Channel)this; + auto thisus = () @trusted { return cast(ChannelImpl)this; } (); bool was_full = false; + synchronized (m_mutex) { - while (thisus.m_items.empty) + 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; swap(thisus.m_items, dst); } + if (was_full) thisus.m_condition.notifyAll(); } void put(T item) shared { - auto thisus = cast(Channel)this; + 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)); +}