Walks each node of the graph, in parallel if it can.
        The walk_func is only called when the nodes dependencies have been
        satisfied

def walk(self, dag, walk_func):
        """ Walks each node of the graph, in parallel if it can.
        The walk_func is only called when the nodes dependencies have been
        satisfied
        """

        # First, we'll topologically sort all of the nodes, with nodes that
        # have no dependencies first. We do this to ensure that we don't call
        # .join on a thread that hasn't yet been started.
        #
        # TODO(ejholmes): An alternative would be to ensure that Thread.join
        # blocks if the thread has not yet been started.
        nodes = dag.topological_sort()
        nodes.reverse()

        # This maps a node name to a thread of execution.
        threads = {}

        # Blocks until all of the given nodes have completed execution (whether
        # successfully, or errored). Returns True if all nodes returned True.
        def wait_for(nodes):
            for node in nodes:
                thread = threads[node]
                while thread.is_alive():
                    threads[node].join(0.5)

        # For each node in the graph, we're going to allocate a thread to
        # execute. The thread will block executing walk_func, until all of the
        # nodes dependencies have executed.
        for node in nodes:
            def fn(n, deps):
                if deps:
                    logger.debug(
                        "%s waiting for %s to complete",
                        n,
                        ", ".join(deps))

                # Wait for all dependencies to complete.
                wait_for(deps)

                logger.debug("%s starting", n)

                self.semaphore.acquire()
                try:
                    return walk_func(n)
                finally:
                    self.semaphore.release()

            deps = dag.all_downstreams(node)
            threads[node] = Thread(target=fn, args=(node, deps), name=node)

        # Start up all of the threads.
        for node in nodes:
            threads[node].start()

        # Wait for all threads to complete executing.
        wait_for(nodes)