Use cio.FIFOSet.Close() to cleanup fifos

Signed-off-by: Daniel Nephin <dnephin@docker.com>

hack/dockerfile/binaries-commits

@@ -4,6 +4,10 @@ TOMLV_COMMIT=9baf8a8a9f2ed20a8e54160840c492f937eeaf9a
 
 # When updating RUNC_COMMIT, also update runc in vendor.conf accordingly
 RUNC_COMMIT=b2567b37d7b75eb4cf325b77297b140ea686ce8f
+
+# containerd is also pinned in vendor.conf. When updating the binary
+# version you may also need to update the vendor version to pick up bug
+# fixes or new APIs.
 CONTAINERD_COMMIT=89623f28b87a6004d4b785663257362d1658a729 # v1.0.0
 TINI_COMMIT=949e6facb77383876aeff8a6944dde66b3089574
 LIBNETWORK_COMMIT=7b2b1feb1de4817d522cc372af149ff48d25028e

libcontainerd/client_daemon.go

@@ -121,7 +121,7 @@ func (c *client) Restore(ctx context.Context, id string, attachStdio StdioCallba
 	c.Lock()
 	defer c.Unlock()
 
-	var rio *cio.DirectIO
+	var rio cio.IO
 	defer func() {
 		err = wrapError(err)
 	}()
@@ -139,12 +139,13 @@ func (c *client) Restore(ctx context.Context, id string, attachStdio StdioCallba
 	}()
 
 	t, err := ctr.Task(ctx, func(fifos *cio.FIFOSet) (cio.IO, error) {
-		rio, err = cio.NewDirectIO(ctx, fifos)
+		io, err := cio.NewDirectIO(ctx, fifos)
 		if err != nil {
 			return nil, err
 		}
 
-		return attachStdio(rio)
+		rio, err = attachStdio(io)
+		return rio, err
 	})
 	if err != nil && !errdefs.IsNotFound(errors.Cause(err)) {
 		return false, -1, err
@@ -322,7 +323,6 @@ func (c *client) Exec(ctx context.Context, containerID, processID string, spec *
 				rio.Cancel()
 				rio.Close()
 			}
-			rmFIFOSet(fifos)
 		}
 	}()
 
@@ -332,10 +332,6 @@ func (c *client) Exec(ctx context.Context, containerID, processID string, spec *
 	})
 	if err != nil {
 		close(stdinCloseSync)
-		if rio != nil {
-			rio.Cancel()
-			rio.Close()
-		}
 		return -1, err
 	}
 
@@ -686,7 +682,7 @@ func (c *client) processEvent(ctr *container, et EventType, ei EventInfo) {
 					"container": ei.ContainerID,
 				}).Error("failed to find container")
 			} else {
-				rmFIFOSet(newFIFOSet(ctr.bundleDir, ei.ProcessID, true, false))
+				newFIFOSet(ctr.bundleDir, ei.ProcessID, true, false).Close()
 			}
 		}
 	})

libcontainerd/client_daemon_linux.go

@@ -81,30 +81,28 @@ func prepareBundleDir(bundleDir string, ociSpec *specs.Spec) (string, error) {
 }
 
 func newFIFOSet(bundleDir, processID string, withStdin, withTerminal bool) *cio.FIFOSet {
-	fifos := &cio.FIFOSet{
-		Config: cio.Config{
-			Terminal: withTerminal,
-			Stdout:   filepath.Join(bundleDir, processID+"-stdout"),
-		},
+	config := cio.Config{
+		Terminal: withTerminal,
+		Stdout:   filepath.Join(bundleDir, processID+"-stdout"),
 	}
+	paths := []string{config.Stdout}
 
 	if withStdin {
-		fifos.Stdin = filepath.Join(bundleDir, processID+"-stdin")
+		config.Stdin = filepath.Join(bundleDir, processID+"-stdin")
+		paths = append(paths, config.Stdin)
 	}
-
-	if !fifos.Terminal {
-		fifos.Stderr = filepath.Join(bundleDir, processID+"-stderr")
+	if !withTerminal {
+		config.Stderr = filepath.Join(bundleDir, processID+"-stderr")
+		paths = append(paths, config.Stderr)
 	}
-
-	return fifos
-}
-
-func rmFIFOSet(fset *cio.FIFOSet) {
-	for _, fn := range []string{fset.Stdout, fset.Stdin, fset.Stderr} {
-		if fn != "" {
-			if err := os.RemoveAll(fn); err != nil {
-				logrus.Warnf("libcontainerd: failed to remove fifo %v: %v", fn, err)
+	closer := func() error {
+		for _, path := range paths {
+			if err := os.RemoveAll(path); err != nil {
+				logrus.Warnf("libcontainerd: failed to remove fifo %v: %v", path, err)
 			}
 		}
+		return nil
 	}
+
+	return cio.NewFIFOSet(config, closer)
 }

libcontainerd/client_daemon_windows.go

@@ -2,6 +2,7 @@ package libcontainerd
 
 import (
 	"fmt"
+	"path/filepath"
 
 	"github.com/containerd/containerd/cio"
 	"github.com/containerd/containerd/windows/hcsshimtypes"
@@ -35,19 +36,20 @@ func pipeName(containerID, processID, name string) string {
 	return fmt.Sprintf(`\\.\pipe\containerd-%s-%s-%s`, containerID, processID, name)
 }
 
-func newFIFOSet(bundleDir, containerID, processID string, withStdin, withTerminal bool) *cio.FIFOSet {
-	fifos := &cio.FIFOSet{
+func newFIFOSet(bundleDir, processID string, withStdin, withTerminal bool) *cio.FIFOSet {
+	containerID := filepath.Base(bundleDir)
+	config := cio.Config{
 		Terminal: withTerminal,
-		Out:      pipeName(containerID, processID, "stdout"),
+		Stdout:   pipeName(containerID, processID, "stdout"),
 	}
 
 	if withStdin {
-		fifos.In = pipeName(containerID, processID, "stdin")
+		config.Stdin = pipeName(containerID, processID, "stdin")
 	}
 
-	if !fifos.Terminal {
-		fifos.Err = pipeName(containerID, processID, "stderr")
+	if !config.Terminal {
+		config.Stderr = pipeName(containerID, processID, "stderr")
 	}
 
-	return fifos
+	return cio.NewFIFOSet(config, nil)
 }

libcontainerd/client_local_windows.go

@@ -4,7 +4,6 @@ import (
 	"context"
 	"encoding/json"
 	"fmt"
-	"io"
 	"io/ioutil"
 	"os"
 	"path"
@@ -671,7 +670,7 @@ func (c *client) Start(_ context.Context, id, _ string, withStdin bool, attachSt
 		return p.pid, nil
 	}
 
-	dio, err := newIOFromProcess(newProcess)
+	dio, err := newIOFromProcess(newProcess, ctr.ociSpec.Process.Terminal)
 	if err != nil {
 		logger.WithError(err).Error("failed to get stdio pipes")
 		return -1, err
@@ -712,16 +711,14 @@ func (c *client) Start(_ context.Context, id, _ string, withStdin bool, attachSt
 	return p.pid, nil
 }
 
-func newIOFromProcess(newProcess process) (*cio.DirectIO, error) {
+func newIOFromProcess(newProcess hcsshim.Process, terminal bool) (*cio.DirectIO, error) {
 	stdin, stdout, stderr, err := newProcess.Stdio()
 	if err != nil {
 		return nil, err
 	}
 
-	dio := cio.DirectIO{
-		Terminal: ctr.ociSpec.Process.Terminal,
-		Stdin:    createStdInCloser(stdin, newProcess),
-	}
+	dio := cio.NewDirectIO(createStdInCloser(stdin, newProcess), nil, nil, terminal)
+
 	// Convert io.ReadClosers to io.Readers
 	if stdout != nil {
 		dio.Stdout = ioutil.NopCloser(&autoClosingReader{ReadCloser: stdout})
@@ -786,10 +783,6 @@ func (c *client) Exec(ctx context.Context, containerID, processID string, spec *
 	logger.Debugf("exec commandLine: %s", createProcessParms.CommandLine)
 
 	// Start the command running in the container.
-	var (
-		stdout, stderr io.ReadCloser
-		stdin          io.WriteCloser
-	)
 	newProcess, err := ctr.hcsContainer.CreateProcess(&createProcessParms)
 	if err != nil {
 		logger.WithError(err).Errorf("exec's CreateProcess() failed")
@@ -812,12 +805,11 @@ func (c *client) Exec(ctx context.Context, containerID, processID string, spec *
 		}
 	}()
 
-	dio, err := newIOFromProcess(newProcess)
+	dio, err := newIOFromProcess(newProcess, spec.Terminal)
 	if err != nil {
 		logger.WithError(err).Error("failed to get stdio pipes")
 		return -1, err
 	}
-	dio.Termainl = spec.Terminal
 	// Tell the engine to attach streams back to the client
 	_, err = attachStdio(dio)
 	if err != nil {