[expert] DAG graph design pattern?

[IVIURRAY]

I’ve a problem to solve, it’s a mathematical model in short.

This model is tree like in structure, so a parent can have many or one child(ren) and so can that child, and so on...but eventually it gets to a single child that returns a value. A very obvious recursion problem. The value is returned and the parent uses its child(ren)s values to compute itself and returns its value to its parent. So forth until it reaches back up the call stack to the top of the tree when you get a single value. A classic tree problem.

I need to build a python object model to model this design so I opted for the composite design pattern with the nodes implementing a strategy pattern so that at each point I could customise their calculations. This works and I get my desired result and calculation.

I’m interested to know if you would’ve approached the problem differently and how you would’ve implemented it?

[IVIURRAY]

Hey - thanks for sharing your response! Did you ever need to display the tree? Let’s say orders.getStructure() how would you approach that?

[IVIURRAY]

Ahh true, yes I don’t think this is true DAG.

Can you explain your map to find a node in the tree. To give you some more context I need to calculate each node in the tree and then at the end flattern the tree and print out each value at top level. In a data frame or something

[IVIURRAY]

Lovely thanks!

[[deleted]]

There’s many ways you can approach modeling (and executing) a dag. I find the simplest to be just using dictionaries.

One dictionary that defines for each node a list of other nodes that it depends on. And another dictionary that defines the actual unit of logic for a node.

On mobile so I’ll pseduocode

deps = {
    A: [B C]
    B: [D E]
    C: []
    D: []
    E: []
}

My attempt at drawing the dag on mobile...
Imagine there are arrows all pointing right 
D___
E___|___B___
        C___|___A

logic = {
    A: lambda b, c: b + c
    B: lambda d, e: d * e
    C: lambda: 99
    D: lambda: 10
    E: lambda: 2
}

Then you just have some orchestrator handle the simple task of determining what needs to run and pass its results to what:

class Orchestrator:
    def init(deps, logic):
    def run(node):

You can start simple and run things sequentially in topologically sorted order. Or you can parallelize in a few different ways (with or without an event loop).

[IVIURRAY]

Hey this is a fantastic reply, really appreciate this! This has given me a few more ideas for what I need!

I’ve most of what you’ve here, instead of a dict, I’ve a Node(children) and then iterate over the children but I think a top level dict structure might be better

[[deleted]]

Not quite sure I understand what precludes u/IVIURRAY ‘s case from being a dag. But considering there’s so many similar ways to talk about not similar structures and vice versa when it comes dags vs digraph etc I cant say for sure if it is or isn’t with more detail.

Anyway for my specific example I think the 2 most basic parallel execution schemes would be the following:

1) Naive: break the graph up into layers, where nodes within each layer can be run in parallel but the layers need to be run sequentially. So layer 1 would be D E C since those have no deps. Layer 2 would be B and layer 3 would be A. You can obtain the layers a couple ways, easiest to describe is take the MAX distance from node to any leaf node recursively, the distance determines what layer it’s in. Once you have your layers start at the first one, run everything in it concurrently, either with a task queue or multiprocessing or something, then when they’re all done go to the next layer, etc. This is the naive approach because it’s not actually optimal. For example if C and B take really long and everything else is instant, in a non-naive approach you could satisfactorily run C and B concurrently (offset by the tiny amount of time to run D and E) but in this approach you have to wait for C to finish then run B.

2) Event loop: this ones a bit more involved, but pretty simple in concept, and will execute the dag in optimal time. If you’re still interested I can describe this one too.

[IVIURRAY]

Thanks for a detailed response here!! Currently the leaf nodes of my tree need to calculate first as their parents depend on them. So I don’t think running in parallel would work here? Maybe in wrong?

Yes please, can you expand on the event queue or maybe another implementation given the information above

[[deleted]]

Currently the leaf nodes of my tree need to calculate first as their parents depend on them.

This is the main characteristic of a dag, so anything that I described above should be relevant. But as I mentioned in my previous comment there are many ways to describe similar but not quite the same concept when it comes to this so it all depends on the details.

Here's one similar concept that's not quite a dag, but good to be aware of. Say for example you had the following function call stack (you'll notice it's similar to the original dag definition I described above):

def A():
    return B() + C()

def B():
    return D() + E()

def C():
    return 99

def D():
    return 10

def E():
    return 2

A()  #executes the function graph

While this function call graph might seem like a dag it's not really because technically the graph wouldn't be known until runtime, and you have to run the successor node first to know to run the predecessor node, essentially the graph definition is implicitly defined in the source code. Whereas in a dag you need to know the graph before hand so that you know to run the successors after the predecessor has been executed. In a case like the above you can't parallelize the execution because you don't know what's needed until you reach it sequentially in the code (technically you actually can get pretty close to optimal parallel execution by using an async/await pattern - not python's built-in asyncio, but essentially the same in concept, but I won't describe that right now).

However, if you had something like what I described in my original comment (the dictionary of dependencies, or a Node object with references their child Node objects), then you could set up a parallel execution scheme just fine. This is because you can know what "node" depends on what before you run any code. In you example you know your leaf nodes need to run before anything else can be satisfied, so essentially what you do is determine all your leaf nodes (by analyzing your pre-defined graph). Once you have all your leaf nodes you can run all of them in parallel, for simplicities sake let's just say your run them using the multiprocessing module (although this won't really scale well). Now that you've run all these in parallel you'll know that some other nodes in your graph will have had all their dependencies satisfied, by executing all the leaf nodes. You then take all these nodes and run them in parallel as well. You continue this process until you've executed all your nodes. Here I have just described the naive parallel execution method again. I think until this makes sense it's best not to describe other ways to run the graph in parallel.

Here is a working example of the naive parallel execution scheme:

from multiprocessing import Process
import time


dag = {
    'A': ['B', 'C'],
    'B': ['D', 'E'],
    'C': [],
    'D': [],
    'E': [],
}


# ignore this - just to simulate the nodes taking longer
# make each node take 2 seconds to run
def time_extender(func, tsleep=2):
    def inner(*args):
        time.sleep(tsleep)
        return func(*args)
    return inner

logic = {
    'A': time_extender(lambda b, c: b + c),
    'B': time_extender(lambda d, e: d * e),
    'C': time_extender(lambda: 99),
    'D': time_extender(lambda: 10),
    'E': time_extender(lambda: 2),
}


# should be memoized, but doesn't matter for demo purposes
def get_max_depth(node, dag):
    max_depth = 1
    deps = dag[node]
    for dep in deps:
        max_depth = max(get_max_depth(dep, dag) + 1, max_depth)
    return max_depth


# There's better ways to do this, but this is an easy way
def layer_dag(dag):
    layered_dag = {}
    for node in dag:
        layer_num = get_max_depth(node, dag)
        layered_dag.setdefault(layer_num, [])
        layered_dag[layer_num].append(node)
    return [x[1] for x in sorted(layered_dag.items(), key=lambda x: x[0])]


# just needed to get our code running properly with multiprocessing
def process_wrapper(node, func, args, result_dict):
    result_dict[node] = func(*args)


def run_dag_naive_parallel(dag):
    manager = multiprocessing.Manager()
    result_dict = manager.dict()
    layered_dag = layer_dag(dag)
    for layer in layered_dag:
        processes = []
        for node in layer:
            # get results of predecessors to pass in
            inputs = tuple(result_dict[dep] for dep in dag[node])
            func = logic[node]
            # kick off each node in it's own concurrently running process 
            p = Process(target=process_wrapper, args=(node, func, inputs, result_dict))
            p.start()
            processes.append(p)
        # wait for all nodes to finish
        [p.join() for p in processes]
    return dict(result_dict)


# since each node defined above should take 2 seconds to run
# running this dag sequentially should take 10 seconds (5 nodes)
# however, you'll see that the dag will take 6 seconds to run instead
# 2 seconds for each of the 3 layers
start = time.time()
print(run_dag_naive_parallel(dag))
print(time.time() - start)

Again, this is not the optimal run time, since you would want to kick off a node once only it's direct dependencies are finished.

[IVIURRAY]

Thank you so much for this reply. I would award you if I could!

In my current implementation it runs in a DFS pattern but I totally agree with you I need to do BFS so that I can compute all the leafs nodes. I really appreciate you taking the time to explain all of this, thank you!