GitHub - andreiavrammsd/cpp-channel: C++11 thread-safe container for sharing data between threads (synchronized queue) (original) (raw)

Channel

Thread-safe container for sharing data between threads (synchronized queue). Header-only. Compatible with C++11 and newer.

About

msd::channel

• A synchronized queue that can be easily and safely shared between multiple threads.
• Tested with GCC, Clang, and MSVC.
• Uses std::mutex for synchronization.
• Uses a customizable storage to store elements.

It's a class that can be constructed in several ways:

• Buffered:
  • The channel accepts a specified number of elements, after which it blocks the writer threads and waits for a reader thread to read an element.
  • It blocks the reader threads when channel is empty until a writer thread writes elements.
  • msd::channel<int> chan{2};
• Unbuffered:
  • Never blocks writes.
  • It blocks the reader threads when channel is empty until a writer thread writes elements.
  • msd::channel<int> chan{};
• Heap- or stack-allocated: pass a custom storage or choose a built-in storage:
  • msd::queue_storage (default): uses std::queue
  • msd::vector_storage: uses std::vector (if cache locality is important)
    * msd::channel<int, msd::vector_storage<int>> chan{2};
  • msd::array_storage (always buffered): uses std::array (if you want stack allocation)
    * msd::channel<int, msd::array_storage<int, 10>> chan{};
    * msd::channel<int, msd::array_storage<int, 10>> chan{10}; // does not compile because capacity is already passed as template argument
    * aka msd::static_channel<int, 10>

A storage is:

• A class with a specific interface for storing elements.
• Must implement FIFO logic.
• See built-in storages.

Exceptions:

• msd::operator<< throws msd::closed_channel if channel is closed.
• msd::channel::write returns bool status instead of throwing.
• Heap-allocated storages could throw.
• Static-allocated storage does not throw.
• Throws if stored elements throw.

Features

• Thread-safe push and fetch.
• Use stream operators to push (<<) and fetch (>>) items.
• Value type must be default constructible, move constructible, move assignable, and destructible.
• Blocking (forever waiting to fetch).
• Range-based for loop supported.
• Close to prevent pushing and stop waiting to fetch.
• Integrates with some of the STL algorithms. Eg:
  • std::move(ch.begin(), ch.end(), ...)
  • std::transform(input_chan.begin(), input_chan.end(), msd::back_inserter(output_chan)).
  • std::copy_if(chan.begin(), chan.end(), ...);

Installation

Choose one of the methods:

• Copy the include directory into your project and add it to your include path.
• CMake FetchContent
• CMake install - choose a version, then run:

VERSION=X.Y.Z
&& wget https://github.com/andreiavrammsd/cpp-channel/archive/refs/tags/v$VERSION.zip
&& unzip v$VERSION.zip
&& cd cpp-channel-$VERSION
&& mkdir build && cd build
&& cmake .. -DCMAKE_INSTALL_PREFIX=/usr/local
&& sudo cmake --install .

• Bazel

Usage

// Unbuffered channel

#include <msd/channel.hpp>

int main() { msd::channel chan;

chan << 1 << 2;  // Send

int first_value{};
int second_value{};
chan >> first_value >> second_value;  // Receive
chan.read(first_value);  // Returns channel close status (true/false), blocks thread when channel is empty

}

// Buffered channel with custom storage

#include <msd/channel.hpp>

int main() { msd::channel<int, msd::vector_storage> chan{2};

chan << 1;      // Throws if channel is closed
chan.write(2);  // Non-throwing write, returns channel close status (true/false)
chan << 3;      // Blocks thread (no space, no reader)

}

// Range-based iteration

#include <msd/channel.hpp>

#include

int main() { msd::channel chan{2};

chan << 1 << 2;
for (int value : chan) {
    if (chan.closed()) {
        // You can break before it's empty
        break;
    }

    std::cout << value << '\n';  // Blocks thread until there is data to read or channel is closed and empty
}

}

// Channel with statically-allocated storage (always buffered)

#include <msd/static_channel.hpp>

#include

int main() { msd::static_channel<int, 2> src{}; msd::static_channel<int, 2> dst{};

src.write(1);
src.write(2);
src.close();

std::copy_if(src.begin(), src.end(), msd::back_inserter(dst), [](int value) { return value % 2 == 0; });

dst.size();  // 1

}

See examples and tests. Read the documentation for full API reference.

Known limitations

• In some cases, the integration with some STL algorithms does not compile with MSVC. See the Transform test.

Developed with CLion and Visual Studio Code.