async-queue.cpp

#include <iostream>
#include <optional>
#include <queue>
#include <thread>

template <typename T> class AsyncQueue {
  struct Node {
    std::mutex mutex;
    std::queue<T> queue;
    Node *next;
    bool done;

    Node() : next(nullptr), done(false) {}
  };

  Node *past;
  Node *present;

  void assert_nonnull() {
    if (present == nullptr || past == nullptr) {
      throw std::runtime_error("node is null");
    }
  }

  void advance_present() {
    assert_nonnull();
    past->mutex.lock();
    while (past != nullptr && past->queue.empty() && past->done) {
      Node *new_past = past;
      past = new_past->next;
      if (new_past->next != nullptr) {
        new_past->next->mutex.lock();
      }
      new_past->mutex.unlock();
      delete new_past;
    }
    past->mutex.unlock();
  }

public:
  AsyncQueue() : past(new Node), present(past) {}
  ~AsyncQueue() {
    if (present != nullptr) {
      std::lock_guard<std::mutex> guard(present->mutex);
      present->done = true;
    }
  }

  AsyncQueue subtask() {
    assert_nonnull();
    std::mutex *m0 = &past->mutex;
    std::mutex *m1 = &present->mutex;
    m0->lock();
    if (m0 != m1) {
      m1->lock();
    }

    Node *n0 = present;  // our events that already happened
    Node *n1 = new Node; // the events of the subtask
    Node *n2 = new Node; // our events that have yet to happen
    n0->next = n1;
    n1->next = n2;

    n0->done = true;
    present = n2;

    AsyncQueue queue;
    queue.past = past;
    queue.present = n1;

    m0->unlock();
    if (m0 != m1) {
      m1->unlock();
    }

    return queue;
  };

  AsyncQueue(const AsyncQueue &other) = delete;
  AsyncQueue &operator=(const AsyncQueue &) = delete;
  AsyncQueue(AsyncQueue &&other) : past(other.past), present(other.present) {
    other.past = nullptr;
    other.present = nullptr;
  }
  AsyncQueue &operator=(AsyncQueue &&other) = delete;

  void send(T elt) {
    if (present->done) {
      throw std::runtime_error("illegal state (done flag set too early)");
    }
    assert_nonnull();
    std::lock_guard<std::mutex> guard(present->mutex);
    present->queue.push(elt);
  }

  bool empty() {
    assert_nonnull();
    advance_present();
    std::lock_guard<std::mutex> guard(present->mutex);
    return past == nullptr || past->queue.empty();
  }

  std::optional<T> recv() {
    assert_nonnull();
    advance_present();
    std::lock_guard<std::mutex> guard(present->mutex);
    if (past == nullptr || past->queue.empty()) {
      return std::nullopt;
    }
    T val = std::move(past->queue.front());
    past->queue.pop();
    return val;
  }

  void debug() {
    Node *n = past;
    bool first = true;
    while (n != nullptr) {
      if (!first) {
        std::cerr << " --> ";
      }
      if (n == present) {
        std::cerr << "[";
      }
      std::cerr << n->queue.size();
      if (n == present) {
        std::cerr << "]";
      }
      if (!n->done) {
        std::cerr << "+";
      }
      first = false;
      n = n->next;
    }
    std::cerr << std::endl;
  }
};

void task(AsyncQueue<int> &&queue, int begin, int end) {
  for (int i = begin; i < end; i++) {
    queue.send(i);
    std::this_thread::sleep_for(std::chrono::milliseconds(100));
  }
}

int main() {
  AsyncQueue<int> queue;

  std::thread t1(task, queue.subtask(), 0, 50);
  std::thread t2(task, queue.subtask(), 50, 100);
  std::thread t3(task, queue.subtask(), 100, 150);
  std::thread t4(task, queue.subtask(), 150, 200);
  std::thread t5(task, queue.subtask(), 200, 250);

  int old_value = -1;
  while (true) {
    std::optional<int> value = queue.recv();
    if (value.has_value()) {
      queue.debug();
      std::cerr << "RECV " << value.value() << std::endl;
      if (value.value() != old_value + 1) {
        throw std::runtime_error("desynchronized");
      }
      old_value = value.value();
    } else {
      std::this_thread::sleep_for(std::chrono::milliseconds(10));
    }
  }
}