Building Channels in C++: Part 1 - The Adventure Begins

Planning the channel

For now, we're just going to focus on bufferred channels. Unbufferred channels work in the coroutine/fiber space, but they start to have issues in the thread-only space (which we'll get to in a future article). Plus, bufferred channels give us a much simpler and easier to understand starting model. So, that's where we'll start.

This means we'll need some sort of buffer or queue to hold the items in. Since we're accessing this from multiple threads, we'll also need a lock. At some point, there may be a thread waiting for a message, so we'll need a signal. This means we'll need a queue, lock, and signal. In the C++ standard, we can map these to std::queue, std::mutex, and std::condition_variable. This is a good starting point, and it will let us get the basics figured out before we start enhancing it. It also means we'll be building a dynamic channel rather than a static channel. In otherwords, operations on our channel will result in memory allocations - which could be problematic in memory-constrained or safety critical environments. Again though, this is just a starting point. We'll explore how to make the memory fixed and static in a future post.

As for operations, we need to send and receive messages. For now, we'll just let these block indefinitely (though we'll have to fix this later on). We also want the message type to be templatized, that way we can change what we're sending. So with the planning done, let's get started!

Outline the Channel

We'le start off making the channel. We'll start off making a simple templatized class with all of our member variables. For this post, I'm using C++20 (though C++17 should be pretty similar).

#include <queue>
#include <thread>

template <typename T>
class Channel {
  std::queue<T> buffer = {};
  std::mutex mux;
  std::condition_variable signal;

public:
  void send(const T&) {
    // TODO: implement
  }

  T receive() {
    // TODO: implement
  }
};

With the channel outline built we can start with the actual implementation of it.

Sending Data

Let's start off with sending data. This is fairly straightforward. We first acquire the lock, then we add to the queue, and then we tell the signal that a message is ready. We need to notify the signal, that way anyone waiting for a message will know they can read it. Below is the body of our send method.

void send(const T& elem) {
  auto lock = std::unique_lock{mux};
  buffer.push(elem);
  signal.notify_one();
}

We notify one thread with our channel since we only published a single message, and we only want one thread to consume that message. If we were doing something that could impact all waiting threads then we'd notify all threads instead.

Receiving Data

Receiving data is a little more complicated, but not much. Unlike send, where we can always add items, with receive we may not be able to remove items, such as when the queue is empty. In this case, we could return an error saying the channel is empty, but then the user of our channel would have to loop and keep locking the queue checking for items. Ideally, if there aren't new items, instead of returning we just wait for a message, and then when it comes we "wake up" and grab the message. The "wait until x happens" is what the condition variable is. It allows us to specify a condition to wake up ("a new message appeared") and then wait until that condition is met (which happens when send does a notify).

One little side note about condition variables, is that when waiting for one to be met you need to already have the associated lock. That lock will then be passed to the condition variable, which will handle unlocking and re-locking automatically. By the time the condition variable returns, the lock will be reobtained. This allows writing code as if the lock was always in place, while simultaneously allowing other threads to run in the locked section and meet the condition.

With this knowledge, our receive function looks like the following:

T receive() {
  auto lock = std::unique_lock{mux};

  if (buffer.empty()) {
    // wait for a new message
    signal.wait(lock);
  }

  auto res = buffer.front();
  buffer.pop();
  return res;
}

And with that we've finished our simple channel. We can use it as follows:

Channel<int> ch;

ch.send(5);
assert(ch.receive() == 5);

Of course, our channels are overly simplistic and lack a lot of features that "real" channels have (such as the ability to close them). They also continuously allocate memory on the heap with each send, which could lead to memory exhaustion (or at least, take up so much memory we start using swap space). A lot of channel implementations have ways to set limits on memory consumption (which can also make it so channels could live completely in stack memory too). Additionally, we're far away from being able to implement a select statement properly.

Select statements will try a series of channel operations until one succeeds. But our channel always blocks until success, so that won't really work. Finally, there's the issue of timeouts. Soft real-time systems have limits on how long an operation can run before a success or failure is reported. But, right now we have no way to limit how long we block, so we could block forever. We'll need some way to add timeouts too.

All of this to say, we've just started scratching the surface.

To continue the adventure, go to Building Channels in C++: Part 2 Limiting Size.