Method run

3rdparty/libprocess/include/process/loop.hpp:333–430

  }

  void run(Future<T> next)
  {
    auto self = shared();

    // Reset `discard` so that we're not delaying cleanup of any
    // captured futures longer than necessary.
    //
    // TODO(benh): Use `WeakFuture` in `discard` functions instead.
    synchronized (mutex) {
      discard = []() {};
    }

    while (next.isReady()) {
      Future<ControlFlow<R>> flow = body(next.get());
      if (flow.isReady()) {
        switch (flow->statement()) {
          case ControlFlow<R>::Statement::CONTINUE: {
            next = iterate();
            continue;
          }
          case ControlFlow<R>::Statement::BREAK: {
            promise.set(flow->value());
            return;
          }
        }
      } else {
        auto continuation = [self](const Future<ControlFlow<R>>& flow) {
          if (flow.isReady()) {
            switch (flow->statement()) {
              case ControlFlow<R>::Statement::CONTINUE: {
                self->run(self->iterate());
                break;
              }
              case ControlFlow<R>::Statement::BREAK: {
                self->promise.set(flow->value());
                break;
              }
            }
          } else if (flow.isFailed()) {
            self->promise.fail(flow.failure());
          } else if (flow.isDiscarded()) {
            self->promise.discard();
          }
        };

        if (pid.isSome()) {
          flow.onAny(defer(pid.get(), continuation));
        } else {
          flow.onAny(continuation);
        }

        if (!promise.future().hasDiscard()) {
          synchronized (mutex) {
            self->discard = [=]() mutable { flow.discard(); };
          }
        }

        // There's a race between when a discard occurs and the