From 2bb2425704cc7181c5eb924400b351b3a2d9a592 Mon Sep 17 00:00:00 2001 From: Matthew Heon Date: Wed, 19 Aug 2020 16:15:35 -0400 Subject: Move pod jobs to parallel execution Make Podman pod operations that do not involve starting containers (which needs to be done in a specific order) use the same parallel operation code we use to make `podman stop` on large numbers of containers fast. We were previously stopping containers in a pod serially, which could take up to the timeout (default 15 seconds) for each container - stopping 100 containers that do not respond to SIGTERM would take 25 minutes. To do this, refactor the parallel operation code a bit to remove its dependency on libpod (damn circular import restrictions...) and use parallel functions that just re-use the standard container API operations - maximizes code reuse (previously each pod handler had a separate implementation of the container function it performed). This is a bit of a palate cleanser after fighting CI for two days - nice to be able to return to a land of sanity. Signed-off-by: Matthew Heon --- pkg/api/handlers/libpod/pods.go | 6 ++-- pkg/domain/infra/abi/containers.go | 6 ++-- pkg/domain/infra/abi/pods.go | 6 ++-- pkg/parallel/ctr/ctr.go | 40 ++++++++++++++++++++++++++ pkg/parallel/parallel.go | 30 ++++++++++++++++++++ pkg/parallel/parallel_linux.go | 57 -------------------------------------- pkg/varlinkapi/pods.go | 7 +++-- 7 files changed, 83 insertions(+), 69 deletions(-) create mode 100644 pkg/parallel/ctr/ctr.go delete mode 100644 pkg/parallel/parallel_linux.go (limited to 'pkg') diff --git a/pkg/api/handlers/libpod/pods.go b/pkg/api/handlers/libpod/pods.go index 3aa554171..5422411cf 100644 --- a/pkg/api/handlers/libpod/pods.go +++ b/pkg/api/handlers/libpod/pods.go @@ -270,7 +270,7 @@ func PodPause(w http.ResponseWriter, r *http.Request) { utils.PodNotFound(w, name, err) return } - responses, err := pod.Pause() + responses, err := pod.Pause(r.Context()) if err != nil && errors.Cause(err) != define.ErrPodPartialFail { utils.Error(w, "Something went wrong", http.StatusInternalServerError, err) return @@ -294,7 +294,7 @@ func PodUnpause(w http.ResponseWriter, r *http.Request) { utils.PodNotFound(w, name, err) return } - responses, err := pod.Unpause() + responses, err := pod.Unpause(r.Context()) if err != nil && errors.Cause(err) != define.ErrPodPartialFail { utils.Error(w, "failed to pause pod", http.StatusInternalServerError, err) return @@ -402,7 +402,7 @@ func PodKill(w http.ResponseWriter, r *http.Request) { return } - responses, err := pod.Kill(uint(sig)) + responses, err := pod.Kill(r.Context(), uint(sig)) if err != nil && errors.Cause(err) != define.ErrPodPartialFail { utils.Error(w, "failed to kill pod", http.StatusInternalServerError, err) return diff --git a/pkg/domain/infra/abi/containers.go b/pkg/domain/infra/abi/containers.go index d92911e0c..0107e18c4 100644 --- a/pkg/domain/infra/abi/containers.go +++ b/pkg/domain/infra/abi/containers.go @@ -23,7 +23,7 @@ import ( "github.com/containers/podman/v2/pkg/checkpoint" "github.com/containers/podman/v2/pkg/domain/entities" "github.com/containers/podman/v2/pkg/domain/infra/abi/terminal" - "github.com/containers/podman/v2/pkg/parallel" + parallelctr "github.com/containers/podman/v2/pkg/parallel/ctr" "github.com/containers/podman/v2/pkg/ps" "github.com/containers/podman/v2/pkg/rootless" "github.com/containers/podman/v2/pkg/signal" @@ -157,7 +157,7 @@ func (ic *ContainerEngine) ContainerStop(ctx context.Context, namesOrIds []strin if err != nil && !(options.Ignore && errors.Cause(err) == define.ErrNoSuchCtr) { return nil, err } - errMap, err := parallel.ContainerOp(ctx, ctrs, func(c *libpod.Container) error { + errMap, err := parallelctr.ContainerOp(ctx, ctrs, func(c *libpod.Container) error { var err error if options.Timeout != nil { err = c.StopWithTimeout(*options.Timeout) @@ -321,7 +321,7 @@ func (ic *ContainerEngine) ContainerRm(ctx context.Context, namesOrIds []string, return reports, nil } - errMap, err := parallel.ContainerOp(ctx, ctrs, func(c *libpod.Container) error { + errMap, err := parallelctr.ContainerOp(ctx, ctrs, func(c *libpod.Container) error { err := ic.Libpod.RemoveContainer(ctx, c, options.Force, options.Volumes) if err != nil { if options.Ignore && errors.Cause(err) == define.ErrNoSuchCtr { diff --git a/pkg/domain/infra/abi/pods.go b/pkg/domain/infra/abi/pods.go index 747da9fd4..258640a81 100644 --- a/pkg/domain/infra/abi/pods.go +++ b/pkg/domain/infra/abi/pods.go @@ -66,7 +66,7 @@ func (ic *ContainerEngine) PodKill(ctx context.Context, namesOrIds []string, opt for _, p := range pods { report := entities.PodKillReport{Id: p.ID()} - conErrs, err := p.Kill(uint(sig)) + conErrs, err := p.Kill(ctx, uint(sig)) if err != nil && errors.Cause(err) != define.ErrPodPartialFail { report.Errs = []error{err} reports = append(reports, &report) @@ -92,7 +92,7 @@ func (ic *ContainerEngine) PodPause(ctx context.Context, namesOrIds []string, op } for _, p := range pods { report := entities.PodPauseReport{Id: p.ID()} - errs, err := p.Pause() + errs, err := p.Pause(ctx) if err != nil && errors.Cause(err) != define.ErrPodPartialFail { report.Errs = []error{err} continue @@ -117,7 +117,7 @@ func (ic *ContainerEngine) PodUnpause(ctx context.Context, namesOrIds []string, } for _, p := range pods { report := entities.PodUnpauseReport{Id: p.ID()} - errs, err := p.Unpause() + errs, err := p.Unpause(ctx) if err != nil && errors.Cause(err) != define.ErrPodPartialFail { report.Errs = []error{err} continue diff --git a/pkg/parallel/ctr/ctr.go b/pkg/parallel/ctr/ctr.go new file mode 100644 index 000000000..e8c1292b8 --- /dev/null +++ b/pkg/parallel/ctr/ctr.go @@ -0,0 +1,40 @@ +package ctr + +import ( + "context" + + "github.com/containers/podman/v2/libpod" + "github.com/containers/podman/v2/pkg/parallel" + "github.com/sirupsen/logrus" +) + +// ContainerOp performs the given function on the given set of +// containers, using a number of parallel threads. +// If no error is returned, each container specified in ctrs will have an entry +// in the resulting map; containers with no error will be set to nil. +func ContainerOp(ctx context.Context, ctrs []*libpod.Container, applyFunc func(*libpod.Container) error) (map[*libpod.Container]error, error) { + // We could use a sync.Map but given Go's lack of generic I'd rather + // just use a lock on a normal map... + // The expectation is that most of the time is spent in applyFunc + // anyways. + var ( + errMap = make(map[*libpod.Container]<-chan error) + ) + + for _, ctr := range ctrs { + c := ctr + logrus.Debugf("Starting parallel job on container %s", c.ID()) + errChan := parallel.Enqueue(ctx, func() error { + return applyFunc(c) + }) + errMap[c] = errChan + } + + finalErr := make(map[*libpod.Container]error) + for ctr, errChan := range errMap { + err := <-errChan + finalErr[ctr] = err + } + + return finalErr, nil +} diff --git a/pkg/parallel/parallel.go b/pkg/parallel/parallel.go index c9e4da50d..4da7e0f89 100644 --- a/pkg/parallel/parallel.go +++ b/pkg/parallel/parallel.go @@ -1,6 +1,7 @@ package parallel import ( + "context" "sync" "github.com/pkg/errors" @@ -42,3 +43,32 @@ func SetMaxThreads(threads uint) error { func GetMaxThreads() uint { return numThreads } + +// Enqueue adds a single function to the parallel jobs queue. This function will +// be run when an unused thread is available. +// Returns a receive-only error channel that will return the error (if any) from +// the provided function fn when fn has finished executing. The channel will be +// closed after this. +func Enqueue(ctx context.Context, fn func() error) <-chan error { + retChan := make(chan error) + + go func() { + jobControlLock.RLock() + defer jobControlLock.RUnlock() + + defer close(retChan) + + if err := jobControl.Acquire(ctx, 1); err != nil { + retChan <- errors.Wrapf(err, "error acquiring job control semaphore") + return + } + + err := fn() + + jobControl.Release(1) + + retChan <- err + }() + + return retChan +} diff --git a/pkg/parallel/parallel_linux.go b/pkg/parallel/parallel_linux.go deleted file mode 100644 index 442db1502..000000000 --- a/pkg/parallel/parallel_linux.go +++ /dev/null @@ -1,57 +0,0 @@ -package parallel - -import ( - "context" - "sync" - - "github.com/containers/podman/v2/libpod" - "github.com/pkg/errors" - "github.com/sirupsen/logrus" -) - -// ContainerOp performs the given function on the given set of -// containers, using a number of parallel threads. -// If no error is returned, each container specified in ctrs will have an entry -// in the resulting map; containers with no error will be set to nil. -func ContainerOp(ctx context.Context, ctrs []*libpod.Container, applyFunc func(*libpod.Container) error) (map[*libpod.Container]error, error) { - jobControlLock.RLock() - defer jobControlLock.RUnlock() - - // We could use a sync.Map but given Go's lack of generic I'd rather - // just use a lock on a normal map... - // The expectation is that most of the time is spent in applyFunc - // anyways. - var ( - errMap = make(map[*libpod.Container]error) - errLock sync.Mutex - allDone sync.WaitGroup - ) - - for _, ctr := range ctrs { - // Block until a thread is available - if err := jobControl.Acquire(ctx, 1); err != nil { - return nil, errors.Wrapf(err, "error acquiring job control semaphore") - } - - allDone.Add(1) - - c := ctr - go func() { - logrus.Debugf("Launching job on container %s", c.ID()) - - err := applyFunc(c) - errLock.Lock() - errMap[c] = err - errLock.Unlock() - - allDone.Done() - jobControl.Release(1) - }() - } - - allDone.Wait() - - return errMap, nil -} - -// TODO: Add an Enqueue() function that returns a promise diff --git a/pkg/varlinkapi/pods.go b/pkg/varlinkapi/pods.go index 189434780..6d03afb7a 100644 --- a/pkg/varlinkapi/pods.go +++ b/pkg/varlinkapi/pods.go @@ -3,6 +3,7 @@ package varlinkapi import ( + "context" "encoding/json" "fmt" "strconv" @@ -207,7 +208,7 @@ func (i *VarlinkAPI) KillPod(call iopodman.VarlinkCall, name string, signal int6 if err != nil { return call.ReplyPodNotFound(name, err.Error()) } - ctrErrs, err := pod.Kill(killSignal) + ctrErrs, err := pod.Kill(context.TODO(), killSignal) callErr := handlePodCall(call, pod, ctrErrs, err) if callErr != nil { return err @@ -221,7 +222,7 @@ func (i *VarlinkAPI) PausePod(call iopodman.VarlinkCall, name string) error { if err != nil { return call.ReplyPodNotFound(name, err.Error()) } - ctrErrs, err := pod.Pause() + ctrErrs, err := pod.Pause(context.TODO()) callErr := handlePodCall(call, pod, ctrErrs, err) if callErr != nil { return err @@ -235,7 +236,7 @@ func (i *VarlinkAPI) UnpausePod(call iopodman.VarlinkCall, name string) error { if err != nil { return call.ReplyPodNotFound(name, err.Error()) } - ctrErrs, err := pod.Unpause() + ctrErrs, err := pod.Unpause(context.TODO()) callErr := handlePodCall(call, pod, ctrErrs, err) if callErr != nil { return err -- cgit v1.2.3-54-g00ecf