Building Channels in C++: Part 6 - Adding timeouts

Creating the before case

The first step is to create a new case we can pass to our select. The case should work with a standard clock, and it should store a time point. We can infer the clock by templatizing the parameters to the time point (which takes a clock and duration template parameter). We won't use the duration template parameter directly, but if we're templatizing parameters then we still need to templatize that paramter to avoid potential templating issues in the future. We'll also add a wrapper method for type inference.

Like send and receive, we'll also give it a callback so developers can print an error, clean up, etc. Unlike send and receive, we will only call the callback on timeout (or failure) - rather than success.

As for our action, we're going to compare the current time from the templatized type with the time point time. If we exceed the time point time, we will return a timeout error. We'll simply add a timeout error to our select errors, and return that error. Here's the code:

enum class SelectErrors {
  NOT_READY_YET,
  TIMEOUT,
};

template<typename Clock, typename Duration, typename Callback>
struct BeforeCase {
  std::chrono::time_point<Clock, Duration> end;
  Callback callback;

  Result<void, SelectErrors> attempt() {
    if (Clock::system_clock::now() > end) {
      callback();
      return error(SelectErrors::TIMEOUT);
    }
    return {};
  }
};

template<typename Clock, typename Duration, typename Callback>
BeforeCase<Clock, Duration, Callback> before_case(std::chrono::time_point<Clock, Duration> tp, Callback callback) {
  return BeforeCase<Clock, Duration, Callback>{tp, callback};
}

template<typename Rep, typename Period, typename Callback>
auto timeout_case(std::chrono::duration<Rep, Period> duration, Callback callback) {
  using Clock = std::chrono::system_clock;
  return before_case(Clock::now() + duration, callback);
}

Registering our Before Case

Now that we have our before case, we need to register it with our select statement. This is as simple as adding a new template specialization. However, we're going to operate a little differently.

In our previous specializations, we call the next step if our attempt method fails, but we return if our attempt method succeeds. For our before case we're going to do the opposite - call next on success and return on fail. This allows our timeout to stop execution on an error, while we permit execution on a success. This allows us to have really clear before case logic, while flipping it in the specialization. It's the whole reason we didn't abstract away earlier with just two cases - premature abstraction would ruin our case logic!

Here's the new specialization:

template<typename Clock, typename Duration, typename Callback, typename ...Args>
struct Select<BeforeCase<Clock, Duration, Callback>, Args...> {
  using Case = BeforeCase<Clock, Duration, Callback>;
  using Next = Select<Args...>;
  Case beforeCase;
  Next next;

  Select(Case beforeCase, Args... nextVals)
  : beforeCase(beforeCase), next({nextVals...}) {}

  Result<void, SelectErrors> step() {
    auto res = beforeCase.attempt();
    if (!res) {
      // since we're using SelectErrors in before,
      // we can just return our result here directly!
      return res;
    }
    return next.step();
  }

  void operator()() { while (!step()) { std::this_thread::yield(); } }
};

There's one sublte little edge case though. We're returning an error, but our operator() will keep looping so long as there's an error. What we need to do is update all of our operator() to break on a timeout. Something like the following would work:

void operator()() {
  Result<void, SelectErrors> res = step();
  while(res.is_error() && res.get_error() == SelectErrors::NOT_READY_YET) {
    std::this_thread::yield();
    res = step();
  }
}

We'll copy that to overwrite our operator() in our ReceiveCase, SendCase, and BeforeCase specializations. And with that, we've added a timeout to our select! We can use it as follows:

int main() {
  Channel<int, 1> c;
  Channel<int, 1> quit;

  auto t1 = std::thread([&]{
    int x = 1;
    bool done = false;
    using namespace std::chrono_literals;
    // make sure the entire loop takes less than 5 seconds
    auto loopDone = std::chrono::system_clock::now() + 5s;
    while (!done) {
      // our select statement
      select(
          send_case(c, x, [&](const auto&){
            x *= x + 1;
          }),
          receive_case(quit, [&](const Result<int, ChannelBlockError>& v){
            auto val = v.get_value();
            std::cout << "Ending with " << val << "\n";
            done = true;
          }),
          // Restrict each iteration to be less than 1 second
          timeout_case(1s, [&]{
            std::cerr << "LOOP ITERATION TIMED OUT!\n";
            done = true;
          }),
          // Restrict the entire loop
          before_case(loopDone, [&]{
            std::cerr << "FULL LOOP TIMED OUT!\n";
            done = true;
          })
      );
    }
  });

  for (int i = 0; i < 8; ++i) {
    std::cout << c.receive().get_value() << "\n";
  }

  quit.send(13);

  t1.join();
  return 0;
}

And we're done! Well, almost. There's on subtle bug in our select statement we'll be fixing next time - a little hidden infinite loop that could happen in just the right conditions. To see the bug (and the fix), read Building Channels in C++: Part 7 - Fixing an infinite loop.