Skip to content

Commit

Permalink
feature: update workflow demo
Browse files Browse the repository at this point in the history
  • Loading branch information
@kalencaya
kalencaya committed Oct 1, 2024
1 parent d1b12c3 commit c0120e6
Show file tree
Hide file tree
Showing 16 changed files with 259 additions and 47 deletions.
115 changes: 89 additions & 26 deletions ...framework/carp-framework-dag/src/main/java/cn/sliew/carp/framework/dag/algorithm/DAG.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -18,13 +18,14 @@

package cn.sliew.carp.framework.dag.algorithm;

import com.google.common.collect.Lists;
import org.jgrapht.Graph;
import org.jgrapht.GraphPath;
import org.jgrapht.alg.shortestpath.AllDirectedPaths;
import org.jgrapht.graph.builder.GraphTypeBuilder;
import org.jgrapht.traverse.BreadthFirstIterator;
import org.jgrapht.traverse.TopologicalOrderIterator;

import java.util.List;
import java.util.Set;
import java.util.function.Consumer;
import java.util.stream.Collectors;
Expand Down Expand Up @@ -52,6 +53,30 @@ public Set<DefaultDagEdge<N>> edges() {
return jgrapht.edgeSet();
}

public DefaultDagEdge<N> getEdge(N source, N target) {
return jgrapht.getEdge(source, target);
}

public void removeEdge(N source, N target) {
jgrapht.removeEdge(source, target);
}

public Integer inDegree(N node) {
return jgrapht.inDegreeOf(node);
}

public Set<N> inDegreeOf(N node) {
return jgrapht.incomingEdgesOf(node).stream().map(DefaultDagEdge::getTarget).collect(Collectors.toSet());
}

public Integer outDegree(N node) {
return jgrapht.outDegreeOf(node);
}

public Set<N> outDegreeOf(N node) {
return jgrapht.outgoingEdgesOf(node).stream().map(DefaultDagEdge::getTarget).collect(Collectors.toSet());
}

public Set<N> getSources() {
return jgrapht.vertexSet().stream()
.filter(node -> jgrapht.inDegreeOf(node) == 0)
Expand All @@ -72,41 +97,79 @@ public Integer getMaxDepth() {
.findFirst().get();
}

public List<N> topologySort() {
List<N> queue = Lists.newArrayList();
topologyTraversal(node -> queue.add(node));
return queue;
}

public void topologyTraversal(Consumer<N> consumer) {
TopologicalOrderIterator<N, DefaultDagEdge<N>> iterator = new TopologicalOrderIterator<>(jgrapht);
while (iterator.hasNext()) {
consumer.accept(iterator.next());
}
}

public void breadthFirstTraversal(Consumer<N> consumer) {
BreadthFirstIterator<N, DefaultDagEdge<N>> iterator = new BreadthFirstIterator<>(jgrapht);
while (iterator.hasNext()) {
N node = iterator.next();
int depth = iterator.getDepth(node);
consumer.accept(node);
}
public DAG<N> copy() {
DAG<N> copy = new DAG<>();
nodes().forEach(copy::addNode);
edges().forEach(edge -> copy.addEdge(edge.getSource(), edge.getTarget()));
return copy;
}

/**
* 一层一层执行
* https://magjac.com/graphviz-visual-editor/
* <p>
* digraph {
* A -> B;
* B -> C;
* B -> D;
* B -> E;
* A -> F;
* A -> K;
* C -> G;
* D -> G;
* E -> G;
* F -> H;
* G -> H;
* H -> I;
* H -> J;
* }
*/
public void executeBFS(Consumer<N> consumer) {
doExecuteBFS(0, getMaxDepth(), consumer);
}

public void doExecuteBFS(Integer depth, Integer maxDepth, Consumer consumer) {
// 使用 BFS，计算每个节点所在的层级
BreadthFirstIterator<N, DefaultDagEdge<N>> iterator = new BreadthFirstIterator<>(jgrapht);
while (iterator.hasNext()) {
N node = iterator.next();
if (iterator.getDepth(node) == depth) {
consumer.accept(node);
}
}
if (depth <= maxDepth) {
// 开始遍历第二层
doExecuteBFS(depth + 1, maxDepth, consumer);
}
public static void main(String[] args) {
DAG<String> dag = new DAG<>();
dag.addNode("A");
dag.addNode("B");
dag.addNode("C");
dag.addNode("D");
dag.addNode("E");
dag.addNode("F");
dag.addNode("G");
dag.addNode("H");
dag.addNode("I");
dag.addNode("J");
dag.addNode("K");

dag.addEdge("A", "B");
dag.addEdge("B", "C");
dag.addEdge("B", "D");
dag.addEdge("B", "E");
dag.addEdge("A", "F");
dag.addEdge("A", "K");

dag.addEdge("C", "G");
dag.addEdge("D", "G");
dag.addEdge("E", "G");

dag.addEdge("F", "H");
dag.addEdge("G", "H");

dag.addEdge("H", "I");
dag.addEdge("H", "J");

System.out.println(dag.topologySort());
List<Set<String>> queue = Lists.newArrayList();
DagUtil.execute(dag, nodes -> queue.add(nodes));
System.out.println(queue);
}
}
70 changes: 70 additions & 0 deletions ...ework/carp-framework-dag/src/main/java/cn/sliew/carp/framework/dag/algorithm/DagUtil.java
View file
Original file line number Diff line number Diff line change
@@ -0,0 +1,70 @@
/*
* Licensed to the Apache Software Foundation (ASF) under one
* or more contributor license agreements. See the NOTICE file
* distributed with this work for additional information
* regarding copyright ownership. The ASF licenses this file
* to you under the Apache License, Version 2.0 (the
* "License"); you may not use this file except in compliance
* with the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/

package cn.sliew.carp.framework.dag.algorithm;

import com.google.common.collect.Sets;
import org.apache.commons.collections4.CollectionUtils;

import java.util.Set;
import java.util.function.BiPredicate;
import java.util.function.Consumer;
import java.util.stream.Collectors;

public enum DagUtil {
;

public static <N> void execute(DAG<N> dag, Consumer<Set<N>> consumer) {
execute(dag, (dag1, node) -> true, (dag1, edge) -> true, consumer);
}

public static <N> void execute(DAG<N> dag, BiPredicate<DAG<N>, N> nodeValidator, BiPredicate<DAG<N>, DefaultDagEdge<N>> edgeValidator, Consumer<Set<N>> consumer) {
Set<N> sources = dag.getSources();
DAG<N> state = dag.copy();
doExecute(dag, state, sources, nodeValidator, edgeValidator, consumer);
}

private static <N> void doExecute(DAG<N> dag, DAG<N> state, Set<N> set, BiPredicate<DAG<N>, N> nodeValidator, BiPredicate<DAG<N>, DefaultDagEdge<N>> edgeValidator, Consumer<Set<N>> consumer) {
if (CollectionUtils.isEmpty(set)) {
return;
}
Set<N> checkedSet = set.stream().filter(node -> nodeValidator.test(dag, node)).collect(Collectors.toSet());
if (CollectionUtils.isEmpty(checkedSet) == false) {
consumer.accept(checkedSet);
}

// 收集下一批次需要执行的节点
Set<N> nextSet = Sets.newLinkedHashSet();
set.forEach(node -> {
dag.outDegreeOf(node).forEach(outDegreeNode -> {
DefaultDagEdge<N> edge = dag.getEdge(node, outDegreeNode);
if (edgeValidator.test(dag, edge)) {
state.removeEdge(node, outDegreeNode);
}
if (state.inDegree(outDegreeNode) == 0) {
nextSet.add(outDegreeNode);
}
});
});

// 执行下一批次
if (CollectionUtils.isNotEmpty(nextSet)) {
doExecute(dag, state, nextSet, nodeValidator, edgeValidator, consumer);
}
}
}
3 changes: 3 additions & 0 deletions ...mework-dag/src/main/java/cn/sliew/carp/framework/dag/service/DagConfigComplexService.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -18,6 +18,7 @@

package cn.sliew.carp.framework.dag.service;

import cn.sliew.carp.framework.dag.algorithm.DAG;
import cn.sliew.carp.framework.dag.service.dto.DagConfigComplexDTO;
import cn.sliew.carp.framework.dag.service.dto.DagConfigDTO;
import cn.sliew.carp.framework.dag.service.dto.DagConfigStepDTO;
Expand All @@ -36,6 +37,8 @@ public interface DagConfigComplexService {

Graph<DagConfigStepDTO> getDag(Long dagId);

DAG<DagConfigStepDTO> getDagNew(Long dagId);

Long insert(DagConfigSimpleAddParam param);

boolean update(DagConfigSimpleUpdateParam param);
Expand Down
3 changes: 3 additions & 0 deletions ...work-dag/src/main/java/cn/sliew/carp/framework/dag/service/DagInstanceComplexService.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -18,6 +18,7 @@

package cn.sliew.carp.framework.dag.service;

import cn.sliew.carp.framework.dag.algorithm.DAG;
import cn.sliew.carp.framework.dag.service.dto.DagConfigStepDTO;
import cn.sliew.carp.framework.dag.service.dto.DagInstanceComplexDTO;
import cn.sliew.carp.framework.dag.service.dto.DagInstanceDTO;
Expand All @@ -32,5 +33,7 @@ public interface DagInstanceComplexService {

Graph<DagStepDTO> getDag(Long dagInstanceId, Graph<DagConfigStepDTO> configDag);

DAG<DagStepDTO> getDagNew(Long dagInstanceId);

Long initialize(Long dagConfigId);
}
2 changes: 2 additions & 0 deletions ...p-framework-dag/src/main/java/cn/sliew/carp/framework/dag/service/DagInstanceService.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -29,4 +29,6 @@ public interface DagInstanceService extends IService<DagInstance> {
Long add(DagInstanceDTO instanceDTO);

boolean update(DagInstanceDTO instanceDTO);

boolean updateStatus(Long id, String fromStatus, String toStatus);
}
2 changes: 2 additions & 0 deletions .../carp-framework-dag/src/main/java/cn/sliew/carp/framework/dag/service/DagStepService.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -33,4 +33,6 @@ public interface DagStepService extends IService<DagStep> {
boolean add(DagStepDTO stepDTO);

boolean update(DagStepDTO stepDTO);

boolean updateStatus(Long id, String fromStatus, String toStatus);
}
14 changes: 12 additions & 2 deletions ...g/src/main/java/cn/sliew/carp/framework/dag/service/impl/DagConfigComplexServiceImpl.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -18,6 +18,7 @@

package cn.sliew.carp.framework.dag.service.impl;

import cn.sliew.carp.framework.dag.algorithm.DAG;
import cn.sliew.carp.framework.dag.service.DagConfigComplexService;
import cn.sliew.carp.framework.dag.service.DagConfigLinkService;
import cn.sliew.carp.framework.dag.service.DagConfigService;
Expand Down Expand Up @@ -72,8 +73,17 @@ public DagConfigDTO selectSimpleOne(Long dagId) {

@Override
public Graph<DagConfigStepDTO> getDag(Long dagId) {
DagConfigComplexDTO dag = selectOne(dagId);
DAG<DagConfigStepDTO> dag = getDagNew(dagId);
MutableGraph<DagConfigStepDTO> graph = GraphBuilder.directed().build();
dag.nodes().forEach(graph::addNode);
dag.edges().forEach(edge -> graph.putEdge(edge.getSource(), edge.getTarget()));
return graph;
}

@Override
public DAG<DagConfigStepDTO> getDagNew(Long dagId) {
DagConfigComplexDTO dag = selectOne(dagId);
DAG<DagConfigStepDTO> graph = new DAG<>();
List<DagConfigStepDTO> steps = dag.getSteps();
List<DagConfigLinkDTO> links = dag.getLinks();
if (CollectionUtils.isEmpty(steps)) {
Expand All @@ -84,7 +94,7 @@ public Graph<DagConfigStepDTO> getDag(Long dagId) {
graph.addNode(step);
stepMap.put(step.getStepId(), step);
}
links.forEach(link -> graph.putEdge(stepMap.get(link.getFromStepId()), stepMap.get(link.getToStepId())));
links.forEach(link -> graph.addEdge(stepMap.get(link.getFromStepId()), stepMap.get(link.getToStepId())));
return graph;
}

Expand Down
28 changes: 18 additions & 10 deletions ...src/main/java/cn/sliew/carp/framework/dag/service/impl/DagInstanceComplexServiceImpl.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -19,9 +19,10 @@
package cn.sliew.carp.framework.dag.service.impl;

import cn.sliew.carp.framework.common.util.UUIDUtil;
import cn.sliew.carp.framework.dag.algorithm.DAG;
import cn.sliew.carp.framework.dag.algorithm.DefaultDagEdge;
import cn.sliew.carp.framework.dag.service.*;
import cn.sliew.carp.framework.dag.service.dto.*;
import com.google.common.graph.EndpointPair;
import com.google.common.graph.Graph;
import com.google.common.graph.GraphBuilder;
import com.google.common.graph.MutableGraph;
Expand All @@ -32,7 +33,6 @@

import java.util.Date;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

@Service
Expand Down Expand Up @@ -64,17 +64,27 @@ public DagInstanceDTO selectSimpleOne(Long dagInstanceId) {

@Override
public Graph<DagStepDTO> getDag(Long dagInstanceId, Graph<DagConfigStepDTO> configDag) {
List<DagStepDTO> dagStepDTOS = dagStepService.listSteps(dagInstanceId);
DAG<DagStepDTO> dag = getDagNew(dagInstanceId);
MutableGraph<DagStepDTO> graph = GraphBuilder.directed().build();
dag.nodes().forEach(graph::addNode);
dag.edges().forEach(edge -> graph.putEdge(edge.getSource(), edge.getTarget()));
return graph;
}

@Override
public DAG<DagStepDTO> getDagNew(Long dagInstanceId) {
DagInstanceComplexDTO dagInstanceComplexDTO = selectOne(dagInstanceId);
DAG<DagConfigStepDTO> configGraph = dagConfigComplexService.getDagNew(dagInstanceComplexDTO.getDagConfig().getId());
DAG<DagStepDTO> graph = new DAG<>();
Map<Long, DagStepDTO> stepMap = new HashMap<>();
for (DagStepDTO dagStepDTO : dagStepDTOS) {
for (DagStepDTO dagStepDTO : dagInstanceComplexDTO.getSteps()) {
stepMap.put(dagStepDTO.getDagConfigStep().getId(), dagStepDTO);
graph.addNode(dagStepDTO);
}
for (EndpointPair<DagConfigStepDTO> edge : configDag.edges()) {
DagConfigStepDTO source = edge.source();
DagConfigStepDTO target = edge.target();
graph.putEdge(stepMap.get(source.getId()), stepMap.get(target.getId()));
for (DefaultDagEdge<DagConfigStepDTO> edge : configGraph.edges()) {
DagConfigStepDTO source = edge.getSource();
DagConfigStepDTO target = edge.getTarget();
graph.addEdge(stepMap.get(source.getId()), stepMap.get(target.getId()));
}
return graph;
}
Expand All @@ -86,7 +96,6 @@ public Long initialize(Long dagConfigId) {
DagInstanceDTO dagInstanceDTO = new DagInstanceDTO();
dagInstanceDTO.setDagConfig(dagConfigComplexDTO);
dagInstanceDTO.setUuid(UUIDUtil.randomUUId());
dagInstanceDTO.setInputs(dagConfigComplexDTO.getDagAttrs());
dagInstanceDTO.setStartTime(new Date());
Long dagInstanceId = dagInstanceService.add(dagInstanceDTO);
// 插入 dag_step
Expand All @@ -96,7 +105,6 @@ public Long initialize(Long dagConfigId) {
dagStepDTO.setDagInstanceId(dagInstanceId);
dagStepDTO.setDagConfigStep(dagConfigStepDTO);
dagStepDTO.setUuid(UUIDUtil.randomUUId());
dagStepDTO.setInputs(dagConfigStepDTO.getStepAttrs());
dagStepDTO.setStartTime(new Date());
dagStepService.add(dagStepDTO);
}
Expand Down
Loading

0 comments on commit c0120e6

Please sign in to comment.