Controlling parallelism with large number of exec.Run tasks

[cueckoo]

Originally opened by @svend in cuelang/cue#640

cue version: cue version 0.3.0-beta.1 darwin/amd64

I using a custom cue command to start a large number (100s) of independent tasks using exec.Run. The number of process tasks is too large to run them all in parallel, so I am trying to find a way to limit how many tasks are run at a time.

My current attempt involves grouping tasks, and making each group depend on the previous group completing. However, adding the $after dependencies makes the cue command very slow.

Here's an simplfied example of what I am trying:

test_tool.cue

package test

import (
	"tool/exec"
)

command: test: {
	let numTasks = 40
	let parallelism = 10

	_tasks: [ for i, _ in numTasks * [0] {"task-\(i)"}]

        // Number of task groups, each should have parallelism or less tasks
	_numGroups: __div(len(_tasks)+parallelism-1, parallelism)

        // Group tasks into numGroups groups
	_taskGroups: [
		for tg in {
			for i, t in _tasks {"\(__mod(i, _numGroups))": "\(t)": t}
		} {tg},
	]

	for i, tg in _taskGroups
	for k, t in tg {
		let shellCommand = """
echo $(date +%!s(MISSING)): group:\( i ) in-group:\( k ) task:\( t ) starting
/bin/sleep 1
echo $(date +%!s(MISSING)): group:\( i ) in-group:\( k ) task:\( t ) finished
"""

		task: "\( i )": "\( k )": exec.Run & {
			// Run each group serially
			if i > 0 {
				$after: task["\( i-1 )"]
			}
			cmd: ["sh", "-e", "-c", shellCommand]
		}
	}
}

This should start four groups of tasks serially. The tasks in a group run in parallel. I would expect the tasks to take about four seconds, but it takes over a minute on my computer, and causes high CPU. time cue test shows:

real	1m12.176s
user	1m37.900s
sys	0m3.872s

If I comment out the $after: task["\( i-1 )"], time cue test shows:

real	0m1.340s
user	0m0.667s
sys	0m0.494s

Is there a reason adding these dependencies is so expensive? Is there a better way to control how many tasks run at once?

[cueckoo]

Original reply by @svend in cuelang/cue#640 (comment)

Diving up the work into parallelism number of groups, then running the tasks in the group serially scales better.

test_tool.go

package test

import (
	"tool/exec"
)

command: test: {
	let numTasks = 40
	let parallelism = 10

	_tasks: [ for i, _ in numTasks * [0] {"task-\(i)"}]

	// Group tasks into groups
	_taskGroups: [
		for tg in {
			for i, t in _tasks {"\(__mod(i, parallelism))": "\(t)": t}
		} {
			[ for t in tg {t}]
		},
	]

	for i, tg in _taskGroups
	for j, t in tg {
		let shellCommand = """
echo $(date +%!s(MISSING)): group:\( i ) in-group:\( j ) task:\( t ) starting
/bin/sleep 1
echo $(date +%!s(MISSING)): group:\( i ) in-group:\( j ) task:\( t ) finished
"""

		task: "\( i )": "\( j )": exec.Run & {
			// Run each task in group serially
			if j > 0 {
				$after: task["\(i)"]["\( j-1 )"]
			}
			cmd: ["sh", "-e", "-c", shellCommand]
		}
	}
}

time cue test:

real	0m4.394s
user	0m1.070s
sys	0m0.511s

[cueckoo]

Original reply by @mpvl in cuelang/cue#640 (comment)

It would not be an unreasonable feature for cue to limit the amount of concurrency.

I've thought about doing this when writing tools/flow, but realized it may be hard to come up with an upper bound for the general case. But it makes sense to have this as an option, at least, with some reasonable default.

[cueckoo]

Original reply by @myitcv in cuelang/cue#640 (comment)

I've just hit a similar use case where I need to effectively hold a mutex whilst running a specific task (that writes to a part of a file), a number of instances of which are created, much like above, using a comprehension based on a file.Glob.

How about a solution that uses the equivalent of a lengthn buffered channel? Not sure what that would look like in terms of a declaration (using a tool/sync builtin of sorts?), but a task would effectively be able to depend on such a limiter.

Initialising the value of such a limiter (via a builtin) would take a positive integer value. If we were to support the injection of special values like os, we might extend that to include the CUE equivalent of Go's runtime.NumCPU().

[cueckoo]

Original reply by @mpvl in cuelang/cue#640 (comment)

https://cue-review.googlesource.com/c/cue/+/8283

[cueckoo]

Original reply by @myitcv in cuelang/cue#640 (comment)

On the basis this seems like a good idea (per @mpvl) I've just created cuelang/cue#709 on the back of this discussion.

[svend]

Is there a reason adding these dependencies is so expensive?

I think the reason using $after like this is expensive is that setting a task in $after stores the whole task, including its $after and so on.

I switched from $after to a hidden field that is set to an output of the previous task. The tasks run serially, and the performance is much better. I ran a test with 30 tasks which should each take one second. Here are the results:

Dependency	Duration	maximum resident set size
$after: task["\( i-1 )"]	325 seconds	170 MiB
_after: task["\( i-1 )"].success	31.13 seconds	25 MiB

Here's the code:

$ cat serial-1/test_tool.cue 
package test

import (
	"tool/exec"
)

let numTasks = 30
let tasks = [ for i, _ in numTasks * [0] {"task-\(i)"}]
let duration = 1 // each task takes N seconds

command: test: {
	for i, t in tasks {
		let shellCommand = """
echo $(date +%s): group:\( i ) task:\( t ) starting
/bin/sleep \( duration )
echo $(date +%s): group:\( i ) task:\( t ) finished
"""

		task: "\( i )": exec.Run & {
			// Run each task serially
			if i > 0 {
				$after: task["\( i-1 )"]
			}
			cmd: ["sh", "-e", "-c", shellCommand]
		}
	}
}

$ cat serial-2/test_tool.cue 
package test

import (
	"tool/exec"
)

let numTasks = 30
let tasks = [ for i, _ in numTasks * [0] {"task-\(i)"}]
let duration = 1 // each task takes N seconds

command: test: {
	for i, t in tasks {
		let shellCommand = """
echo $(date +%s): group:\( i ) task:\( t ) starting
/bin/sleep \( duration )
echo $(date +%s): group:\( i ) task:\( t ) finished
"""

		task: "\( i )": exec.Run & {
			// Run each task serially
			if i > 0 {
				_after: task["\( i-1 )"].success
			}
			cmd: ["sh", "-e", "-c", shellCommand]
		}
	}
}

$ diff serial-{1,2}/test_tool.cue 
23c23
< 				$after: task["\( i-1 )"]
---
> 				_after: task["\( i-1 )"].success

There is a comment in the tool documentation that seems relevant.

// TODO: consider these options:
// $success: bool
// $runif: a.b.$success or $guard: a.b.$success
// With this $after: a.b would just be a shorthand for $guard: a.b.$success.