folly::FutureDAG Class Reference

#include <FutureDAG.h>

Inheritance diagram for folly::FutureDAG:

Classes
struct	Node

Public Types
typedef size_t	Handle
typedef std::function< Future< Unit >)>	FutureFunc

Public Member Functions
Handle	add (FutureFunc func, Executor *executor=nullptr)
void	remove (Handle a)
void	reset ()
void	dependency (Handle a, Handle b)
void	clean_state (Handle source, Handle sink)
Future< Unit >	go ()

Static Public Member Functions
static std::shared_ptr< FutureDAG >	create ()

Private Member Functions
	FutureDAG ()=default
bool	hasCycle ()

Private Attributes
std::vector< Node >	nodes

Detailed Description

Definition at line 23 of file FutureDAG.h.

Member Typedef Documentation

typedef std::function<Future<Unit>)> folly::FutureDAG::FutureFunc

Definition at line 30 of file FutureDAG.h.

typedef size_t folly::FutureDAG::Handle

Definition at line 29 of file FutureDAG.h.

Constructor & Destructor Documentation

folly::FutureDAG::FutureDAG ( )

privatedefault

Referenced by create(), and go().

Member Function Documentation

Handle folly::FutureDAG::add ( FutureFunc  func, Executor *  executor = nullptr  )

inline

Definition at line 32 of file FutureDAG.h.

References folly::pushmi::executor, folly::gen::move, and nodes.

Referenced by go().

                                                            {
    nodes.emplace_back(std::move(func), executor);
    return nodes.size() - 1;
  }

void folly::FutureDAG::clean_state ( Handle  source, Handle  sink  )

inline

Definition at line 84 of file FutureDAG.h.

References nodes.

                                               {
    for (auto handle : nodes[sink].dependencies) {
      nodes[handle].hasDependents = false;
    }
    nodes[0].hasDependents = false;
    remove(source);
    remove(sink);
  }

static std::shared_ptr<FutureDAG> folly::FutureDAG::create ( )

inlinestatic

Definition at line 25 of file FutureDAG.h.

References FutureDAG().

Referenced by TEST_F().

                                           {
    return std::shared_ptr<FutureDAG>(new FutureDAG());
  }

void folly::FutureDAG::dependency ( Handle  a, Handle  b  )

inline

Definition at line 79 of file FutureDAG.h.

References a, b, and nodes.

Referenced by go(), and hasCycle().

                                      {
    nodes[b].dependencies.push_back(a);
    nodes[a].hasDependents = true;
  }

Future<Unit> folly::FutureDAG::go ( )

inline

Definition at line 93 of file FutureDAG.h.

References add(), folly::collect(), dependency(), folly::pushmi::executor, FutureDAG(), hasCycle(), folly::gen::move, nodes, and folly::pushmi::detail::t.

                    {
    if (hasCycle()) {
      return makeFuture<Unit>(std::runtime_error("Cycle in FutureDAG graph"));
    }
    std::vector<Handle> rootNodes;
    std::vector<Handle> leafNodes;
    for (Handle handle = 0; handle < nodes.size(); handle++) {
      if (nodes[handle].dependencies.empty()) {
        rootNodes.push_back(handle);
      }
      if (!nodes[handle].hasDependents) {
        leafNodes.push_back(handle);
      }
    }

    auto sinkHandle = add([] { return Future<Unit>(); });
    for (auto handle : leafNodes) {
      dependency(handle, sinkHandle);
    }

    auto sourceHandle = add(nullptr);
    for (auto handle : rootNodes) {
      dependency(sourceHandle, handle);
    }

    for (Handle handle = 0; handle < nodes.size() - 1; handle++) {
      std::vector<Future<Unit>> dependencies;
      for (auto depHandle : nodes[handle].dependencies) {
        dependencies.push_back(nodes[depHandle].promise.getFuture());
      }

      collect(dependencies)
          .via(nodes[handle].executor)
          .thenValue([this, handle](std::vector<Unit>&&) {
            nodes[handle].func().then([this, handle](Try<Unit>&& t) {
              nodes[handle].promise.setTry(std::move(t));
            });
          })
          .onError([this, handle](exception_wrapper ew) {
            nodes[handle].promise.setException(std::move(ew));
          });
    }

    nodes[sourceHandle].promise.setValue();
    return nodes[sinkHandle].promise.getFuture().thenValue(
        [that = shared_from_this(), sourceHandle, sinkHandle](Unit) {
          that->clean_state(sourceHandle, sinkHandle);
        });
  }

bool folly::FutureDAG::hasCycle ( )

inlineprivate

Definition at line 146 of file FutureDAG.h.

References dependency(), folly::empty(), and nodes.

Referenced by go().

                  {
    // Perform a modified topological sort to detect cycles
    std::vector<std::vector<Handle>> dependencies;
    for (auto& node : nodes) {
      dependencies.push_back(node.dependencies);
    }

    std::vector<size_t> dependents(nodes.size());
    for (auto& dependencyEdges : dependencies) {
      for (auto handle : dependencyEdges) {
        dependents[handle]++;
      }
    }

    std::vector<Handle> handles;
    for (Handle handle = 0; handle < nodes.size(); handle++) {
      if (!nodes[handle].hasDependents) {
        handles.push_back(handle);
      }
    }

    while (!handles.empty()) {
      auto handle = handles.back();
      handles.pop_back();
      while (!dependencies[handle].empty()) {
        auto dependency = dependencies[handle].back();
        dependencies[handle].pop_back();
        if (--dependents[dependency] == 0) {
          handles.push_back(dependency);
        }
      }
    }

    for (auto& dependencyEdges : dependencies) {
      if (!dependencyEdges.empty()) {
        return true;
      }
    }

    return false;
  }

void folly::FutureDAG::remove ( Handle  a )

inline

Definition at line 37 of file FutureDAG.h.

References a, folly::test::begin(), folly::test::end(), and nodes.

                        {
    if (a >= nodes.size()) {
      return;
    }

    if (nodes[a].hasDependents) {
      for (auto& node : nodes) {
        auto& deps = node.dependencies;
        deps.erase(
            std::remove(std::begin(deps), std::end(deps), a), std::end(deps));
        for (Handle& handle : deps) {
          if (handle > a) {
            handle--;
          }
        }
      }
    }

    nodes.erase(nodes.begin() + a);
  }

void folly::FutureDAG::reset ( )

inline

Definition at line 58 of file FutureDAG.h.

References nodes.

               {
    // Delete all but source node, and reset dependency properties
    Handle source_node;
    std::unordered_set<Handle> memo;
    for (auto& node : nodes) {
      for (Handle handle : node.dependencies) {
        memo.insert(handle);
      }
    }
    for (Handle handle = 0; handle < nodes.size(); handle++) {
      if (memo.find(handle) == memo.end()) {
        source_node = handle;
      }
    }

    nodes.erase(nodes.begin(), nodes.begin() + source_node);
    nodes.erase(nodes.begin() + 1, nodes.end());
    nodes[0].hasDependents = false;
    nodes[0].dependencies.clear();
  }

Member Data Documentation

std::vector<Node> folly::FutureDAG::nodes

private

Definition at line 200 of file FutureDAG.h.

Referenced by add(), clean_state(), dependency(), go(), hasCycle(), remove(), and reset().

---

The documentation for this class was generated from the following file:

• proxygen/folly/folly/experimental/FutureDAG.h