Class Queue

3rdparty/libprocess/include/process/queue.hpp:29–127

template <typename T>
class Queue
{
public:
  Queue() : data(new Data()) {}

  // TODO(bmahler): Take a T&& here instead.
  void put(const T& t)
  {
    // NOTE: We need to grab the promise 'date->promises.front()' but
    // set it outside of the critical section because setting it might
    // trigger callbacks that try to reacquire the lock.
    Owned<Promise<T>> promise;

    synchronized (data->lock) {
      if (data->promises.empty()) {
        data->elements.push(t);
      } else {
        promise = data->promises.front();
        data->promises.pop_front();
      }
    }

    if (promise.get() != nullptr) {
      promise->set(t);
    }
  }

  Future<T> get()
  {
    Future<T> future;

    synchronized (data->lock) {
      if (data->elements.empty()) {
        data->promises.emplace_back(new Promise<T>());
        future = data->promises.back()->future();
      } else {
        T t = std::move(data->elements.front());
        data->elements.pop();
        return Future<T>(std::move(t));
      }
    }

    // If there were no items available, we set up a discard
    // handler. This is done here to minimize the amount of
    // work done within the critical section above.
    auto weak_data = std::weak_ptr<Data>(data);

    future.onDiscard([weak_data, future]() {
      auto data = weak_data.lock();
      if (!data) {
        return;
      }

      synchronized (data->lock) {
        for (auto it = data->promises.begin();
             it != data->promises.end();
             ++it) {
          if ((*it)->future() == future) {