libcontainerd: close stdin sync if possible

Closing stdin of a container or exec (a.k.a.: task or process) has been
somewhat broken ever since support for ContainerD 1.0 was introduced
back in Docker v17.11: the error returned from the CloseIO() call was
effectively ignored due to it being assigned to a local variable which
shadowed the intended variable. Serendipitously, that oversight
prevented a data race. In my recent refactor of libcontainerd, I
corrected the variable shadowing issue and introduced the aforementioned
data race in the process.

Avoid deadlocking when closing stdin without swallowing errors or
introducing data races by calling CloseIO() synchronously if the process
handle is available, falling back to an asynchronous close-and-log
strategy otherwise. This solution is inelegant and complex, but looks to
be the best that could be done without changing the libcontainerd API.

Signed-off-by: Cory Snider <csnider@mirantis.com>

libcontainerd/remote/client.go

@@ -28,6 +28,7 @@ import (
 	"github.com/docker/docker/libcontainerd/queue"
 	libcontainerdtypes "github.com/docker/docker/libcontainerd/types"
 	"github.com/docker/docker/pkg/ioutils"
+	"github.com/hashicorp/go-multierror"
 	v1 "github.com/opencontainers/image-spec/specs-go/v1"
 	specs "github.com/opencontainers/runtime-spec/specs-go"
 	"github.com/pkg/errors"
@@ -502,27 +503,43 @@ func (c *container) createIO(fifos *cio.FIFOSet, stdinCloseSync chan containerd.
 
 	if io.Stdin != nil {
 		var (
-			err       error
+			closeErr  error
 			stdinOnce sync.Once
 		)
 		pipe := io.Stdin
 		io.Stdin = ioutils.NewWriteCloserWrapper(pipe, func() error {
 			stdinOnce.Do(func() {
-				err = pipe.Close()
-				// Do the rest in a new routine to avoid a deadlock if the
-				// Exec/Start call failed.
-				go func() {
-					p, ok := <-stdinCloseSync
+				closeErr = pipe.Close()
+
+				select {
+				case p, ok := <-stdinCloseSync:
 					if !ok {
 						return
 					}
-					err = p.CloseIO(context.Background(), containerd.WithStdinCloser)
-					if err != nil && strings.Contains(err.Error(), "transport is closing") {
-						err = nil
+					if err := closeStdin(context.Background(), p); err != nil {
+						if closeErr != nil {
+							closeErr = multierror.Append(closeErr, err)
+						} else {
+							// Avoid wrapping a single error in a multierror.
+							closeErr = err
+						}
 					}
-				}()
+				default:
+					// The process wasn't ready. Close its stdin asynchronously.
+					go func() {
+						p, ok := <-stdinCloseSync
+						if !ok {
+							return
+						}
+						if err := closeStdin(context.Background(), p); err != nil {
+							c.client.logger.WithError(err).
+								WithField("container", c.c8dCtr.ID()).
+								Error("failed to close container stdin")
+						}
+					}()
+				}
 			})
-			return err
+			return closeErr
 		})
 	}
 
@@ -534,6 +551,14 @@ func (c *container) createIO(fifos *cio.FIFOSet, stdinCloseSync chan containerd.
 	return rio, err
 }
 
+func closeStdin(ctx context.Context, p containerd.Process) error {
+	err := p.CloseIO(ctx, containerd.WithStdinCloser)
+	if err != nil && strings.Contains(err.Error(), "transport is closing") {
+		err = nil
+	}
+	return err
+}
+
 func (c *client) processEvent(ctx context.Context, et libcontainerdtypes.EventType, ei libcontainerdtypes.EventInfo) {
 	c.eventQ.Append(ei.ContainerID, func() {
 		err := c.backend.ProcessEvent(ei.ContainerID, et, ei)