From SO...

paradag01.d

// Code taken from the following SO thread:
// http://stackoverflow.com/questions/19528562/need-help-parallel-traversing-a-dag-in-d

import std.stdio;
import std.conv;
import std.parallelism;
 
class Node {
  string name;
  Node[] clients;
  Node[] masters;
  bool dirty = true;
 
  this(string inname){
    this.name = inname;
  }
 
  void doProcess(){
    writeln("Cleaning ", this.name);
    this.dirty = false;
  }
 
  override string toString() const {
    return name;
  }
 
  void addMasters(Node[] inmasters ...){
    this.masters ~= inmasters;
    foreach (master; inmasters){
      master.clients ~= this;
    }
  }
 
  // Dejan: always returns false, except for the root node!
  bool canProcess() const {
    foreach (node; this.masters){
      if (node.dirty){
        return false;
      }
    }
    return true;
  }
}
 
Node[] mkLinearDag(const int depth){
  // makes a simple linked list of named nodes
  Node[] nodes;  
  auto root = new Node("RootNode");
  nodes ~= root;
  foreach (i; 0 .. depth){
    nodes ~= new Node("test" ~ to!string(i));
    nodes[$-1].addMasters(nodes[$-2]);
  }
  return nodes;
}
 
// Dejan: will be called only once!
// Because canProcess() returns false for all children.
void cleanNodeSimple(Node node, TaskPool pool){
  node.doProcess();
  writeln("cleanNodeSimple() called for node `", node, "`"); 
  foreach (cli; node.clients){
    writeln("I am ", cli, " and I am executed only once, because I am only RootNode's client.");
    if (cli.canProcess()) {
      writeln("I will be executed only once because because ", cli, " has ", cli.clients.length, " clients.");
      writeln("And its master #1 is ", cli.clients[0].masters[0].dirty ? "dirty" : "clean");
      pool.put( task!cleanNodeSimple(cli, pool) );
    }
  }
}

  // Even this one will be executed only once.
  // Second time it is executed, the node0.clients has a master that is dirty
  // and the loop will not be executed at all...
  void cleanNodeSimple(Node node, TaskPool pool){
    node.doProcess();
    foreach (cli; node.clients) {
    pool.put( task!cleanNodeSimple(cli, pool) );
    }
  }
 
void main(){
  auto dag = mkLinearDag(5);
  auto pool = taskPool();
 
  pool.put( task!cleanNodeSimple(dag[0], pool));
  pool.finish(true);
 
  writeln("\n\nOutput:");
  foreach (d;dag){
    writeln(d);
    writeln(d.dirty ? "dirty" : "clean","\n");
  }
}