Class AwaitProcess

include/process/collect.hpp:202–267

template <typename T>
class AwaitProcess : public Process<AwaitProcess<T>>
{
public:
  AwaitProcess(
      const std::vector<Future<T>>& _futures,
      Promise<std::vector<Future<T>>>* _promise)
    : ProcessBase(ID::generate("__await__")),
      futures(_futures),
      promise(_promise),
      ready(0) {}

  ~AwaitProcess() override
  {
    delete promise;
  }

  void initialize() override
  {
    // Stop this nonsense if nobody cares.
    promise->future().onDiscard(defer(this, &AwaitProcess::discarded));

    foreach (const Future<T>& future, futures) {
      future.onAny(defer(this, &AwaitProcess::waited, lambda::_1));
      future.onAbandoned(defer(this, &AwaitProcess::abandoned));
    }
  }

private:
  void abandoned()
  {
    // There is no use waiting because this future will never complete
    // so terminate this process which will cause `promise` to get
    // deleted and our future to also be abandoned.
    terminate(this);
  }

  void discarded()
  {
    foreach (Future<T> future, futures) {
      future.discard();
    }

    // NOTE: we discard the promise after we set discard on each of
    // the futures so that there is a happens-before relationship that
    // can be assumed by callers.
    promise->discard();

    terminate(this);
  }

  void waited(const Future<T>& future)
  {
    CHECK(!future.isPending());

    ready += 1;
    if (ready == futures.size()) {
      // It is safe to move futures at this point.
      promise->set(std::move(futures));