Merge pull request #26778 from WeiZhang555/parallel-operations

Support parallel kill,rm

cli/command/container/kill.go

@@ -39,8 +39,11 @@ func NewKillCommand(dockerCli *command.DockerCli) *cobra.Command {
 func runKill(dockerCli *command.DockerCli, opts *killOptions) error {
 	var errs []string
 	ctx := context.Background()
+	errChan := parallelOperation(ctx, opts.containers, func(ctx context.Context, container string) error {
+		return dockerCli.Client().ContainerKill(ctx, container, opts.signal)
+	})
 	for _, name := range opts.containers {
-		if err := dockerCli.Client().ContainerKill(ctx, name, opts.signal); err != nil {
+		if err := <-errChan; err != nil {
 			errs = append(errs, err.Error())
 		} else {
 			fmt.Fprintf(dockerCli.Out(), "%s\n", name)

cli/command/container/rm.go

@@ -45,13 +45,22 @@ func runRm(dockerCli *command.DockerCli, opts *rmOptions) error {
 	ctx := context.Background()
 
 	var errs []string
-	for _, name := range opts.containers {
-		if name == "" {
+	options := types.ContainerRemoveOptions{
+		RemoveVolumes: opts.rmVolumes,
+		RemoveLinks:   opts.rmLink,
+		Force:         opts.force,
+	}
+
+	errChan := parallelOperation(ctx, opts.containers, func(ctx context.Context, container string) error {
+		if container == "" {
 			return fmt.Errorf("Container name cannot be empty")
 		}
-		name = strings.Trim(name, "/")
+		container = strings.Trim(container, "/")
+		return dockerCli.Client().ContainerRemove(ctx, container, options)
+	})
 
-		if err := removeContainer(dockerCli, ctx, name, opts.rmVolumes, opts.rmLink, opts.force); err != nil {
+	for _, name := range opts.containers {
+		if err := <-errChan; err != nil {
 			errs = append(errs, err.Error())
 		} else {
 			fmt.Fprintf(dockerCli.Out(), "%s\n", name)
@@ -62,15 +71,3 @@ func runRm(dockerCli *command.DockerCli, opts *rmOptions) error {
 	}
 	return nil
 }
-
-func removeContainer(dockerCli *command.DockerCli, ctx context.Context, container string, removeVolumes, removeLinks, force bool) error {
-	options := types.ContainerRemoveOptions{
-		RemoveVolumes: removeVolumes,
-		RemoveLinks:   removeLinks,
-		Force:         force,
-	}
-	if err := dockerCli.Client().ContainerRemove(ctx, container, options); err != nil {
-		return err
-	}
-	return nil
-}

cli/command/container/utils.go

@@ -99,8 +99,8 @@ func getExitCode(dockerCli *command.DockerCli, ctx context.Context, containerID
 	return c.State.Running, c.State.ExitCode, nil
 }
 
-func parallelOperation(ctx context.Context, cids []string, op func(ctx context.Context, id string) error) chan error {
-	if len(cids) == 0 {
+func parallelOperation(ctx context.Context, containers []string, op func(ctx context.Context, container string) error) chan error {
+	if len(containers) == 0 {
 		return nil
 	}
 	const defaultParallel int = 50
@@ -109,18 +109,18 @@ func parallelOperation(ctx context.Context, cids []string, op func(ctx context.C
 
 	// make sure result is printed in correct order
 	output := map[string]chan error{}
-	for _, c := range cids {
+	for _, c := range containers {
 		output[c] = make(chan error, 1)
 	}
 	go func() {
-		for _, c := range cids {
+		for _, c := range containers {
 			err := <-output[c]
 			errChan <- err
 		}
 	}()
 
 	go func() {
-		for _, c := range cids {
+		for _, c := range containers {
 			sem <- struct{}{} // Wait for active queue sem to drain.
 			go func(container string) {
 				output[container] <- op(ctx, container)