Queues are not used when DAG is called as sub dag

[privat-eco]

Hi everyone ;
The following has been tested against dagu 2.6.1.

I have defined a queue in ~/.config/dagu/config.yaml :

queues:
  enabled: true
  config:
    - name: "my-queue"
      max_concurrency: 1

I have a main DAG which starts sub DAG :

type: graph
steps:
  - id: my_step
    call: my_sub_dag
    parallel:
      items:
        - item1
        - item2
        - item3
        - item4
        - item5
        - item6
        - item7
    params:
       SCOPE: "${ITEM}"
---
name: my_sub_dag
type: graph
queue: my-queue
steps:
  - id: run_this
    command: echo "calling this with scope ${SCOPE}" && /bin/bash -c 'sleep "$((1 + $RANDOM % 10))"'

Expected behavior : my_sub_dag should run one at a time, regarding the max_concurrency specified by the queue definition
Current behavior : my_sub_dag is all run at the same time (7 occurrences in the example above)

[yohamta0]

Cross-posting from Discord:

Currently, sub-DAG runs do not use a queue for the sake of simplicity. If you need to limit concurrency, you may need to trigger them as separate DAG runs. If you have a better solution or any suggestions, I’d really appreciate your input.

[privat-eco]

maybe we can have a mechanism where calling a subgraph always puts this child graph on a queue (and using the default queue if a more specific one is not provided ?) and the parent graph waits for all scheduled subgraphs to terminate before going further if there is dependencies on the subgraph step ?

[yohamta0]

That is a possibility. Another solution would be to implement a feature equivalent to Airflow's Pool.

[struffel]

maybe we can have a mechanism where calling a subgraph always puts this child graph on a queue (and using the default queue if a more specific one is not provided ?) and the parent graph waits for all scheduled subgraphs to terminate before going further if there is dependencies on the subgraph step ?

To maybe add a little bit to this, in my use case I don't actually want the main graph to wait for all the children, instead it only acts as a "dispatcher". I was able to solve this by using the dagu CLI inside the DAG itself to just enqueue more DAG runs (using dagu history to check that there isn't a duplicate run already). This certainly works and could maybe be adjusted to wait for all child-runs to finish, but it is probably not the most elegant solution to this problem.

Therefore, my use case would also profit from an "official" implementation that works like call: but actually spawns entirely new runs.

[privat-eco]

To maybe add a little bit to this, in my use case I don't actually want the main graph to wait for all the children, instead it only acts as a "dispatcher". I was able to solve this by using the dagu CLI inside the DAG itself to just enqueue more DAG runs (using dagu history to check that there isn't a duplicate run already). This certainly works and could maybe be adjusted to wait for all child-runs to finish, but it is probably not the most elegant solution to this problem.

Maybe not using dependencies between steps can satisfy the need to not wait for child DAGs. In case we need to wait for sub DAGs to end we could use depends (as I wrote this I think we should also track which DAG scheduled those sub DAGs to ensure multiple main DAG run can find out when its childs has finished to run)

[kriyanshii]

if you want @yottahmd i can work on this.
my two cents: cant you create workers and manage the number of executions?

[yohamta0]

You're right. Worker might be the solution. Sub DAG can be dispatched to a worker with concurrency limit.

[kriyanshii]

we need to think more using building blocks dagu provide -> eg. queues, workers.
This can gives us more grip and control over architecture than we think.

[kriyanshii]

Dagu already limits how many parallel branches of one step can run at the same time:

parallel.go, Lines 8-21
// ParallelConfig contains the configuration for parallel execution of a step.
// MVP version supports basic parallel execution with max_concurrent control.
type ParallelConfig struct {
	// ...
	// MaxConcurrent is the maximum number of parallel executions.
	// Default is 10 if not specified.
	MaxConcurrent int `json:"max_concurrent,omitempty"`
}

So on the main DAG, on the step that uses parallel: + call: my_sub_dag, add:

parallel:
  max_concurrent: 1   # or 2, 3, …
  items:
    - item1
    # ...

That serializes (or caps) concurrent sub-DAG invocations from that fan-out, which matches “only one my_sub_dag at a time” for this parent workflow, using only YAML.

When to use: Throttling is only needed inside this graph’s parallel loop. You do not need the global queue for that.