Shulou(Shulou.com)06/02 Report--

This article mainly explains "DAG to achieve task scheduling and optimization", the content of the article is simple and clear, easy to learn and understand, now please follow the editor's ideas slowly in depth, together to study and learn "DAG to achieve task scheduling and optimization" bar!

Implementation of DAG Task Task scheduling algorithm

Github: https://github.com/smartxing/algorithm

1 Construction of directed graph

DAG dag = new DAG (); dag.addVertex ("A"); dag.addVertex ("B"); dag.addVertex ("C"); dag.addVertex ("D"); dag.addEdge ("A", "B"); dag.addEdge ("A", "C"); System.out.println (dag)

2 topological sorting detects whether there are rings in the graph

Public boolean isCircularity () {Set set = inDegree.keySet (); / / entry table Map inDegree = set.stream (). Collect (Collectors .toMap (k-> k, k-> new AtomicInteger (this.inDegree.get (k). Size (); / / nodes with zero entry degree Set sources = getSources (); LinkedList queue = new LinkedList (); queue.addAll (sources) While (! queue.isEmpty ()) {Object o = queue.removeFirst (); outDegree.get (o) .forEach (so-> {if (inDegree.get (so). DecrementAndGet () = = 0) {queue.add (so);}}) } return inDegree.values () .stream () .filter (x-> x.intValue () > 0) .count () > 0;}

3 stage optimization

Eg if the task has the following relationship, execute task2 after task1 execution, and task3 after task2 execution. Task1-> Task2-> Task3-> Task4 these task should be executed serially. You can package these task together to reduce thread context switching eg: more complex DAG: / * * H *\ * G *\ * A-> B *\ * C-D -E-F-> J * after optimization = = > * (H G) *\ * A-> B *\ * (C Magic DMague E)-(FMagna J) * * / see chain method for details: the key code is as follows: private void chain_ (Set sources, final LinkedHashSetMultimap foutChain, final LinkedHashSetMultimap finChain) {sources.forEach (sourceNode-> {ArrayList maxStage = Lists.newArrayList ()) FindMaxStage (sourceNode, maxStage); if (maxStage.size () > 1) {/ / there is a stage addVertex (foutChain, finChain, maxStage) to be merged; / / add a new node Object o = maxStage.get (maxStage.size ()-1); / / the last node reChain_ (foutChain, finChain, maxStage, o) } if (maxStage.size () = = 1) {/ / there is no stage addVertex (foutChain, finChain, sourceNode) to be merged; / / add a new node Set subNodes = outDegree.get (sourceNode); addSubNodeage (foutChain, finChain, sourceNode, subNodes);}}) } 4 Test DAG to execute the test program: for more information, see DAGExecTest 1 create a new task and print only one sentence: public static class Task implements Runnable {private String taskName; public Task (final String taskName) {this.taskName = taskName;} @ Override public void run () {try {Thread.sleep (2000) } catch (InterruptedException e) {e.printStackTrace ();} System.out.println ("i am running my name is" + taskName + "finish ThreadID:" + Thread.currentThread () .getId ());} public String getTaskName () {return taskName;} @ Override public String toString () {return taskName;}} 2 build DAG DAG dag = DAG.create () Task a = new Task ("a"); Task b = new Task ("b"); Task c = new Task ("c"); Task d = new Task ("d"); Task e = new Task ("e"); Task f = new Task ("f"); Task g = new Task ("g"); Task h = new Task ("h"); Task j = new Task ("j") Dag.addVertex (a); dag.addVertex (b); dag.addVertex (c); dag.addVertex (d); dag.addVertex (e); dag.addVertex (f); dag.addVertex (g); dag.addVertex (h); dag.addVertex (j); dag.addEdge (h, g); dag.addEdge (g, b); dag.addEdge (a, b) Dag.addEdge (b, f); dag.addEdge (c, d); dag.addEdge (d, e); dag.addEdge (e, f); dag.addEdge (f, j) After the construction is completed, as shown in the figure * H *\ * G *\ * A-> B *\ * C-D-E-F-> J 3 stage segmentation DAG chain = dag.chain () After the execution of the figure, go to the following: * (HMague G) * A-> B * * (Cpene Dpenny E)-(F) J) 4 the final result of executing DAG DAGExecTest is printed as follows: it can be found that there are three Stage stage1 containing three task task executing in different threads, where c-d-e gmurc Fmerj is optimized to execute in the same thread. Reduced unnecessary context switching i am running my name is a finish ThreadID: 10 i am running my name is c finish ThreadID: 11 i am running my name ish finish ThreadID: 12 i am running my name is d finish ThreadID: 11 i am running my name is g finish ThreadID: 12 i am running my name is e finish ThreadID: 11 stage end: task detached:a Task chain c-d-e task chain Hmurg-i am running my name is b finish ThreadID: 14 stage end: task detached:b -i am running my name is f finish ThreadID: 11 i am running my name is j finish ThreadID: 11 stage end: task chain fmerj test execution key code is as follows: chain.execute (col-> {Set set = (Set) col) List completableFutures = Lists.newArrayList (); StringBuilder sb = new StringBuilder (); set.stream (). ForEach (x-> {if (x instanceof Task) {CompletableFuture future = CompletableFuture.runAsync ((Task) x, executorService); completableFutures.add (future) Sb.append ("task detached:" + ((Task) x). GetTaskName ()) .append (",");} if (x instanceof List) {List taskList = (List) x; CompletableFuture future = CompletableFuture.runAsync (()-> taskList.forEach (Task::run)); completableFutures.add (future) Sb.append ("task chain" + Joiner.on ("-") .join (taskList.stream (). Map (Task::getTaskName) .join (Collectors.toList ();}}); CompletableFuture.allOf (completableFutures.toArray (new CompletableFuture [completableFutures.size ()])) .join () System.out.println ("stage end:" + sb.toString ()); System.out.println ("- -");}) Thank you for your reading, the above is the content of "DAG task scheduling and optimization". After the study of this article, I believe you have a deeper understanding of DAG task scheduling and optimization, and the specific use needs to be verified in practice. Here is, the editor will push for you more related knowledge points of the article, welcome to follow!

Welcome to subscribe "Shulou Technology Information " to get latest news, interesting things and hot topics in the IT industry, and controls the hottest and latest Internet news, technology news and IT industry trends.