jleben/stitch

Communication between threads in a reactive, real-time safe manner.

Stitch is a C++ library for communication and synchronization of threads and the outside world, suitable for real-time applications.

See the documentation for details.

See the comparison with related software below.

Principles

The design of Stitch is guided by the following principles:

Modularity

Instead of implementing a particular communication model (e.g. Actors, Communicating Sequential Processes, ...), Stitch provides basic building blocks. The blocks are easy to compose, and the user is free to compose them in the way most suitable for their application.

However, Stitch does provide implementations of several common communication patterns built on top of the basic blocks.

Separation of Concerns

Stitch separates two concerns:

1. Information: Producing and consuming data; addressed by shared data structures.
2. Time: Waiting for something to happen, for example data to be produced and consumed; addressed by the event system.

Despite striving to separate these concerns, Stitch makes sure the classes that represent them are easy to compose. This is demonstrated by Stitch's higher-level classes implementing common communication patterns.

Real-time safety

Stitch aims to be suitable for real-time applications. Its goal is to provide at least lock-free progress guarantees and linear or constant time complexity for as many operations as possible. Progress guarantees and complexity should be documented for each individual operation (work in progress).

Shared Data Structures

Stitch provides a number of wait-free and lock-free data structures. Here's an example of a queue and an atom:

struct Data { int x, y, z; }

Stitch::Atom<Data> atom;
Stitch::SPSC_Queue<int> queue;

thread t1([&]()
{
    Stitch::AtomWriter atom_writer(atom);
    atom_writer.store({ 1, 2, 3 }); // Lock-free
    queue.push(123); // Wait-free
});

thread t2([&]()
{
    Stitch::AtomReader atom_reader(atom);
    Data a = atom_reader.load(); // Lock-free
    int b;
    bool ok = queue.pop(b); // Wait-free
});

Events

Stitch provides an event system allowing to wait for multiple events of different types.

Stitch::Timer timer;
Stitch::Signal signal;

timer.start(chrono::milliseconds(1000));

thread notifier([&]()
{
    this_thread::sleep_for(chrono::milliseconds(500));
    signal.notify();
});

Stitch::Event_Reactor reactor;

reactor.subscribe(timer.event(), [&]()
{
    cout << "Timer" << endl;
});

reactor.subscribe(signal.event(), [&]()
{
    cout << "Signal" << endl;
});

reactor.run(Stitch::Event_Reactor::WaitUntilQuit);

Patterns

Building upon shared data structures and events, Stitch provides classes that implement common communication patterns.

Stream producers and consumers

Stream producers and consumers communicate streams of items. They can be connected in a many-to-many fashion.

struct Work { int input; };

// Create a stream consumer that allows 100 work items to be buffered.
Stitch::Stream_Consumer<Work> consumer(100);

thread consumer_thread([&]()
{
    // Consume items until at least 10 are consumed
    int count = 0;
    while(count < 10)
    {
        wait(consumer.event())

        Work item;
        while(consumer.pop(item))
        {
            ++count
            cout << item.input << endl;
        }
    }
});

// Define a producer of work
auto producer_func = [&]()
{
    Stitch::Stream_Producer<Work> producer;
    connect(producer, consumer);

    // Produce 10 work requests, one every 100 ms
    for (int v = 0; v < 10; ++v)
    {
        producer.push({v});

        this_thread::sleep_for(chrono::milliseconds(100));
    }
};

// Start 2 producers
thread producer1_thread(producer_func);
thread producer2_thread(producer_func);

State and Observers

State updater and observers communicate the latest state of a value.

Stitch::State<int> state;

thread writer_thread([&]()
{
    // Update state 10 times, every 100 ms
    for (int v = 1; v <= 10; ++v)
    {
        state.store(v);

        this_thread::sleep_for(chrono::milliseconds(100));
    }
});

auto observer_func = [&]()
{
    Stitch::State_Observer<int> observer;
    observer.connect(state);

    // React to state changes, until state equals 10
    int v;
    do {
        wait(observer.changed());

        v = observer.load();
        cout << v << endl;
    }
    while(v != 10);
};

// Start 2 observers
thread observer1_thread(observer_func);
thread observer2_thread(observer_func);

Comparison with Related Software

These are some of our observations about related software which give reasons for the existence of this library. Please contact the authors of this text if any of this is incorrect or if something else should be added.

Microsoft's Parallel Patterns Library

• Limited to Windows.
• No file events.
• No progress guarantees.

C++ Actor Framework (CAF)

Website

• A higher-level interface (the actor model) instead of basic building blocks.
• Does not specify detailed progress guarantees for each method of each class.

libevent

Website

• C instead of C++ (There exist C++ wrappers though).
• Does not provide a direct interface for waiting on events like our wait(event).
• Does not provide a direct interface for internally generated events like our Signal class.

stitch's Issues

Events: Handle system call errors

Lock-free algorithms: Yield before retrying

Optimize lock-free algorithms by using std::this_thread::yield before retrying.

Make classes non-copyable as necessary

Atom is not lockfree on x86-64

Hazard Pointers: Thread-owned pointers leak when thread ends

Set: Avoid using hazard pointers for the lifetime of an iterator

Since an iterator uses a number of hazard pointers throughout its lifetime, and it is possible that the user has multiple iterators at the same time (by nesting iterations), the number of hazard pointers potentially used by a thread is not statically bounded.

Can we use hazard pointers only during execution of iterator methods, instead of throughout an iterator's lifetime?

`Set` stress test reports false failures

Relax memory ordering constraints

All atomic operations currently use the strictest memory ordering constraints. The constraints should be relaxed to improve performance.

Publishing connection endpoints globally

Publish connections to they are globally accessible using an identifier (perhaps a string).

Never throw exceptions when sending / receiving. Provide other means to track channel full / empty.

Waiting on an Event on multiple threads is not reliable

If nothing else affects the fd, then a poll on a single fd on multiple threads would wake up all the threads when the fd is written to. However, if one of the threads wakes up and reads the fd before another wakes up, then that other one will not wake up, because the fd has turned from readable to unreadable.

The only potential solution that I see at the moment is using the edge-triggered mode of epoll. We are using poll in wait though, in contrast with Event_Reactor.

Alternatively, we could disallow using an Event on multiple threads.

Enable waiting on any sink in the reactive framework

Expose stream classes for communication without explicit connection

One problem: this could promote incorrectly using a stream without multiple writers/readers...

Implement pure event channels

Hazard pointer: Free reclaimed objects earlier

Optimize storage: Free reclaimed objects as soon as their number exceeds current number of threads.

This can be done using dynamic memory allocation. We do not support lock-free reclamation unless dynamic allocation is lock-free anyway.

Hazard pointers: Allocate number of pointers proportional to number of threads

Instead of pre-allocating a fixed number of pointers, we can take M.M.Michael's suggestion to allocate new pointers only when the total number of threads that coexisted at any point in time increases.

Can we also de-allocate pointers in a lock-free manner?

Implement lock-free shared single value