yumoqing/dagflow
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

bugfix

Browse Source
This commit is contained in:
yumoqing 2026-03-17 17:11:30 +08:00
parent d6289c8671
commit 21acaca857
1 changed files with 6 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

6
dagflow/dagflow.py
View File

@ -313,10 +313,14 @@ where backid=${backid}$
flg = await self.is_ok_to_step_next(sor, node, edges, inst) flg = await self.is_ok_to_step_next(sor, node, edges, inst)
if flg: if flg:
await self.start_running(sor, inst, ne, node) await self.start_running(sor, inst, ne, node)
else:
debug(f'{ne.id=}, {node=}, {edges=}不能走下一步')
elif ntype == 'subflow': elif ntype == 'subflow':
flg = await self.is_ok_to_step_next(sor, node, edges, inst) flg = await self.is_ok_to_step_next(sor, node, edges, inst)
if flg: if flg:
await self.start_subflow(sor, inst, ne, node) await self.start_subflow(sor, inst, ne, node)
else:
debug(f'{ne.id=}, {node=}, {edges=}不能走下一步')
elif status in ['done', 'cancelled']: elif status in ['done', 'cancelled']:
debug(f'节点完成，检查后续节点') debug(f'节点完成，检查后续节点')
await self.node_transfer(sor, dsl, inst, ne, node, flow_def) await self.node_transfer(sor, dsl, inst, ne, node, flow_def)
@ -496,6 +500,7 @@ where node_id in ${backnodes}$
if join == 'all': if join == 'all':
for r in recs: for r in recs:
if r.status != 'completed': if r.status != 'completed':
debug(f'前向任务未结束{r}')
return False return False
return True return True
if join == 'xor': if join == 'xor':
@ -522,6 +527,7 @@ where instance_id=${instance_id}$
return True return True
if some <= float(len(trecs))/float(len(recs)): if some <= float(len(trecs))/float(len(recs)):
return True return True
debug(f'部分成功模式：{some=},已完成：{len(trecs)=}，总数:{len(recs)=}'
return False return False
async def task_error_callback(self, ne_id, errmsg): async def task_error_callback(self, ne_id, errmsg):