Enhancement: allow parallel stop, pause, unpause

[WeiZhang555]

Fixes #24724

Stop multiple containers in parallel to speed up stop process, allow
maximum 50 parallel stops.

Signed-off-by: Zhang Wei zhangwei555@huawei.com

[WeiZhang555]

ping @thaJeztah @vdemeester @cpuguy83 @tonistiigi

[AkihiroSuda]

Need a mutex for errs

[WeiZhang555]

I almost forget my PR. You're right, thanks @AkihiroSuda 😄

[WeiZhang555]

Append a commit from @abhinavdahiya, thanks again for his help! 😄

[tonistiigi]

nit: chan struct{}

[WeiZhang555]

@tonistiigi I've made some updates since your last comments, what do you think of this?

[WeiZhang555]

@tonistiigi Updated as per your suggestion, PTAL. Thanks 😄

[WeiZhang555]

What's current status? ping @tonistiigi @vdemeester @cpuguy83

[tonistiigi]

Design SGTM @vdemeester WDYT?

[vdemeester]

Design LGTM too 👼

[tonistiigi]

nit: const, also move to the beginning of the function

[WeiZhang555]

@tonistiigi Updated, PTAL.

[vdemeester]

LGTM 🐸

[WeiZhang555]

ping @tonistiigi

[vdemeester]

/cc @dnephin

[dnephin]

No reason to check err != nil here. You're also doing the same action in both branches, just send directly to the channel:

err := ...
output[container] <- err

[WeiZhang555]

Correct 👍

[dnephin]

I like the idea, I left some comments on the code.

I also wonder if it would make sense to make this more generic. I suspect we could re-use this for other operations as well (pause/unpause comes to mind). We did something similar in Compose.

Maybe something like:

const maxParallel int32 = 50

type operation func(ctx context.Context, string id)

func ParallelOperation(ctx context.Context, args []string, op operation) ([]string, []error) {
    var waitAll sync.WaitGroup
    var errs []error
    var result []string
    output := map[string]chan error{}
    var sem = make(chan struct{}, maxParallel)

    waitAll.Add(len(args))
    for _, arg := range args {
        sem <- struct{}{} // Wait for active queue sem to drain.
        go func(arg string) {
            defer waitAll.Done()
            output[arg] <- op(ctx, arg)
            <- sem
        }(arg)
    waitAll.Wait()
    for _, arg := range args {
        err = <- output[arg]
        if err != nil {
            errs = append(errs, err)
        } else {
            result = append(result, arg)
        }
    }
    return result, errs
}

(the timeout can be passed as a closure)

[AkihiroSuda]

nit: +1(non-binding) for @dnephin 's suggestion, and it'd be better if maxParallel is dynamically determined using the value of runtime.NumCPU() 😄

[WeiZhang555]

I'm also +1 on @dnephin 's non-binding suggestion 😄 . I had the plan of add parallel operations on other command, but thought it would be better that we finish this first as an example implementation so that we can do the last quickly.

I will make some modifications first as per @dnephin 's suggestions.

[WeiZhang555]

Hi all, I make a generic function as @dnephin suggested. Also add parallel operation support for pause/unpause as example although pause/unpause is usually fast and don't profit much from it.

I will add modify more time-consuming command such as start/kill/... maybe in different PR 😄

PTAL @ALL

[dnephin]

Minor nit: I think you can just pass dockerCli.Client().ContainerPause directly as the last argument, since it already has the right signature

[WeiZhang555]

Ah, right! 👍

[dnephin]

LGTM

[WeiZhang555]

Parallel start bring a failed test case, remove that first. Let's focus on parallel stop/pause/unpause first.

[vdemeester]

LGTM 🐸

[thaJeztah]

@dnephin latest version LGTY?

[dnephin]

LGTM

[thaJeztah]

Thanks @WeiZhang555!