Merge pull request #35748 from dnephin/try-new-containerd-cio

Update vendor of containerd, use new containerd/cio

container/container.go

@@ -27,7 +27,6 @@ import (
 	"github.com/docker/docker/daemon/network"
 	"github.com/docker/docker/image"
 	"github.com/docker/docker/layer"
-	"github.com/docker/docker/libcontainerd"
 	"github.com/docker/docker/opts"
 	"github.com/docker/docker/pkg/containerfs"
 	"github.com/docker/docker/pkg/idtools"
@@ -1004,7 +1003,7 @@ func (container *Container) CloseStreams() error {
 }
 
 // InitializeStdio is called by libcontainerd to connect the stdio.
-func (container *Container) InitializeStdio(iop *libcontainerd.IOPipe) (cio.IO, error) {
+func (container *Container) InitializeStdio(iop *cio.DirectIO) (cio.IO, error) {
 	if err := container.startLogging(); err != nil {
 		container.Reset(false)
 		return nil, err

container/stream/streams.go

@@ -7,7 +7,7 @@ import (
 	"strings"
 	"sync"
 
-	"github.com/docker/docker/libcontainerd"
+	"github.com/containerd/containerd/cio"
 	"github.com/docker/docker/pkg/broadcaster"
 	"github.com/docker/docker/pkg/ioutils"
 	"github.com/docker/docker/pkg/pools"
@@ -114,7 +114,7 @@ func (c *Config) CloseStreams() error {
 }
 
 // CopyToPipe connects streamconfig with a libcontainerd.IOPipe
-func (c *Config) CopyToPipe(iop *libcontainerd.IOPipe) {
+func (c *Config) CopyToPipe(iop *cio.DirectIO) {
 	copyFunc := func(w io.Writer, r io.ReadCloser) {
 		c.Add(1)
 		go func() {

daemon/exec/exec.go

@@ -6,7 +6,6 @@ import (
 
 	"github.com/containerd/containerd/cio"
 	"github.com/docker/docker/container/stream"
-	"github.com/docker/docker/libcontainerd"
 	"github.com/docker/docker/pkg/stringid"
 	"github.com/sirupsen/logrus"
 )
@@ -63,7 +62,7 @@ func (i *rio) Wait() {
 }
 
 // InitializeStdio is called by libcontainerd to connect the stdio.
-func (c *Config) InitializeStdio(iop *libcontainerd.IOPipe) (cio.IO, error) {
+func (c *Config) InitializeStdio(iop *cio.DirectIO) (cio.IO, error) {
 	c.StreamConfig.CopyToPipe(iop)
 
 	if c.StreamConfig.Stdin() == nil && !c.Tty && runtime.GOOS == "windows" {

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,8 +121,12 @@ func (c *client) Restore(ctx context.Context, id string, attachStdio StdioCallba
 	c.Lock()
 	defer c.Unlock()
 
-	var rio cio.IO
+	var dio *cio.DirectIO
 	defer func() {
+		if err != nil && dio != nil {
+			dio.Cancel()
+			dio.Close()
+		}
 		err = wrapError(err)
 	}()
 
@@ -131,22 +135,16 @@ func (c *client) Restore(ctx context.Context, id string, attachStdio StdioCallba
 		return false, -1, errors.WithStack(err)
 	}
 
-	defer func() {
-		if err != nil && rio != nil {
-			rio.Cancel()
-			rio.Close()
-		}
-	}()
-
-	t, err := ctr.Task(ctx, func(fifos *cio.FIFOSet) (cio.IO, error) {
-		io, err := newIOPipe(fifos)
+	attachIO := func(fifos *cio.FIFOSet) (cio.IO, error) {
+		// dio must be assigned to the previously defined dio for the defer above
+		// to handle cleanup
+		dio, err = cio.NewDirectIO(ctx, fifos)
 		if err != nil {
 			return nil, err
 		}
-
-		rio, err = attachStdio(io)
-		return rio, err
-	})
+		return attachStdio(dio)
+	}
+	t, err := ctr.Task(ctx, attachIO)
 	if err != nil && !errdefs.IsNotFound(errors.Cause(err)) {
 		return false, -1, err
 	}
@@ -255,7 +253,7 @@ func (c *client) Start(ctx context.Context, id, checkpointDir string, withStdin
 	uid, gid := getSpecUser(spec)
 	t, err = ctr.ctr.NewTask(ctx,
 		func(id string) (cio.IO, error) {
-			fifos := newFIFOSet(ctr.bundleDir, id, InitProcessName, withStdin, spec.Process.Terminal)
+			fifos := newFIFOSet(ctr.bundleDir, InitProcessName, withStdin, spec.Process.Terminal)
 			rio, err = c.createIO(fifos, id, InitProcessName, stdinCloseSync, attachStdio)
 			return rio, err
 		},
@@ -315,7 +313,7 @@ func (c *client) Exec(ctx context.Context, containerID, processID string, spec *
 		stdinCloseSync = make(chan struct{})
 	)
 
-	fifos := newFIFOSet(ctr.bundleDir, containerID, processID, withStdin, spec.Terminal)
+	fifos := newFIFOSet(ctr.bundleDir, processID, withStdin, spec.Terminal)
 
 	defer func() {
 		if err != nil {
@@ -323,7 +321,6 @@ func (c *client) Exec(ctx context.Context, containerID, processID string, spec *
 				rio.Cancel()
 				rio.Close()
 			}
-			rmFIFOSet(fifos)
 		}
 	}()
 
@@ -333,10 +330,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
 	}
 
@@ -612,7 +605,7 @@ func (c *client) getProcess(containerID, processID string) (containerd.Process,
 // createIO creates the io to be used by a process
 // This needs to get a pointer to interface as upon closure the process may not have yet been registered
 func (c *client) createIO(fifos *cio.FIFOSet, containerID, processID string, stdinCloseSync chan struct{}, attachStdio StdioCallback) (cio.IO, error) {
-	io, err := newIOPipe(fifos)
+	io, err := cio.NewDirectIO(context.Background(), fifos)
 	if err != nil {
 		return nil, err
 	}
@@ -687,7 +680,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.ContainerID, ei.ProcessID, true, false))
+				newFIFOSet(ctr.bundleDir, ei.ProcessID, true, false).Close()
 			}
 		}
 	})
@@ -851,11 +844,9 @@ func (c *client) writeContent(ctx context.Context, mediaType, ref string, r io.R
 }
 
 func wrapError(err error) error {
-	if err == nil {
-		return nil
-	}
-
 	switch {
+	case err == nil:
+		return nil
 	case errdefs.IsNotFound(err):
 		return wrapNotFoundError(err)
 	}

libcontainerd/client_daemon_linux.go

@@ -80,29 +80,29 @@ func prepareBundleDir(bundleDir string, ociSpec *specs.Spec) (string, error) {
 	return p, nil
 }
 
-func newFIFOSet(bundleDir, containerID, processID string, withStdin, withTerminal bool) *cio.FIFOSet {
-	fifos := &cio.FIFOSet{
+func newFIFOSet(bundleDir, processID string, withStdin, withTerminal bool) *cio.FIFOSet {
+	config := cio.Config{
 		Terminal: withTerminal,
-		Out:      filepath.Join(bundleDir, processID+"-stdout"),
+		Stdout:   filepath.Join(bundleDir, processID+"-stdout"),
 	}
+	paths := []string{config.Stdout}
 
 	if withStdin {
-		fifos.In = filepath.Join(bundleDir, processID+"-stdin")
+		config.Stdin = filepath.Join(bundleDir, processID+"-stdin")
+		paths = append(paths, config.Stdin)
 	}
-
-	if !fifos.Terminal {
-		fifos.Err = 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.Out, fset.In, fset.Err} {
-		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"
@@ -18,6 +17,7 @@ import (
 	"github.com/Microsoft/hcsshim"
 	opengcs "github.com/Microsoft/opengcs/client"
 	"github.com/containerd/containerd"
+	"github.com/containerd/containerd/cio"
 	"github.com/docker/docker/pkg/sysinfo"
 	"github.com/docker/docker/pkg/system"
 	specs "github.com/opencontainers/runtime-spec/specs-go"
@@ -670,28 +670,12 @@ func (c *client) Start(_ context.Context, id, _ string, withStdin bool, attachSt
 		return p.pid, nil
 	}
 
-	var (
-		stdout, stderr io.ReadCloser
-		stdin          io.WriteCloser
-	)
-	stdin, stdout, stderr, err = newProcess.Stdio()
+	dio, err := newIOFromProcess(newProcess, ctr.ociSpec.Process.Terminal)
 	if err != nil {
 		logger.WithError(err).Error("failed to get stdio pipes")
 		return -1, err
 	}
-
-	iopipe := &IOPipe{Terminal: ctr.ociSpec.Process.Terminal}
-	iopipe.Stdin = createStdInCloser(stdin, newProcess)
-
-	// Convert io.ReadClosers to io.Readers
-	if stdout != nil {
-		iopipe.Stdout = ioutil.NopCloser(&autoClosingReader{ReadCloser: stdout})
-	}
-	if stderr != nil {
-		iopipe.Stderr = ioutil.NopCloser(&autoClosingReader{ReadCloser: stderr})
-	}
-
-	_, err = attachStdio(iopipe)
+	_, err = attachStdio(dio)
 	if err != nil {
 		logger.WithError(err).Error("failed to attache stdio")
 		return -1, err
@@ -727,6 +711,24 @@ func (c *client) Start(_ context.Context, id, _ string, withStdin bool, attachSt
 	return p.pid, nil
 }
 
+func newIOFromProcess(newProcess hcsshim.Process, terminal bool) (*cio.DirectIO, error) {
+	stdin, stdout, stderr, err := newProcess.Stdio()
+	if err != nil {
+		return nil, err
+	}
+
+	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})
+	}
+	if stderr != nil {
+		dio.Stderr = ioutil.NopCloser(&autoClosingReader{ReadCloser: stderr})
+	}
+	return dio, nil
+}
+
 // Exec adds a process in an running container
 func (c *client) Exec(ctx context.Context, containerID, processID string, spec *specs.Process, withStdin bool, attachStdio StdioCallback) (int, error) {
 	ctr := c.getContainer(containerID)
@@ -781,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")
@@ -807,25 +805,13 @@ func (c *client) Exec(ctx context.Context, containerID, processID string, spec *
 		}
 	}()
 
-	stdin, stdout, stderr, err = newProcess.Stdio()
+	dio, err := newIOFromProcess(newProcess, spec.Terminal)
 	if err != nil {
-		logger.WithError(err).Error("getting std pipes failed")
+		logger.WithError(err).Error("failed to get stdio pipes")
 		return -1, err
 	}
-
-	iopipe := &IOPipe{Terminal: spec.Terminal}
-	iopipe.Stdin = createStdInCloser(stdin, newProcess)
-
-	// Convert io.ReadClosers to io.Readers
-	if stdout != nil {
-		iopipe.Stdout = ioutil.NopCloser(&autoClosingReader{ReadCloser: stdout})
-	}
-	if stderr != nil {
-		iopipe.Stderr = ioutil.NopCloser(&autoClosingReader{ReadCloser: stderr})
-	}
-
 	// Tell the engine to attach streams back to the client
-	_, err = attachStdio(iopipe)
+	_, err = attachStdio(dio)
 	if err != nil {
 		return -1, err
 	}

libcontainerd/io.go

@@ -1,36 +0,0 @@
-package libcontainerd
-
-import "github.com/containerd/containerd/cio"
-
-// Config returns the containerd.IOConfig of this pipe set
-func (p *IOPipe) Config() cio.Config {
-	return p.config
-}
-
-// Cancel aborts ongoing operations if they have not completed yet
-func (p *IOPipe) Cancel() {
-	p.cancel()
-}
-
-// Wait waits for io operations to finish
-func (p *IOPipe) Wait() {
-}
-
-// Close closes the underlying pipes
-func (p *IOPipe) Close() error {
-	p.cancel()
-
-	if p.Stdin != nil {
-		p.Stdin.Close()
-	}
-
-	if p.Stdout != nil {
-		p.Stdout.Close()
-	}
-
-	if p.Stderr != nil {
-		p.Stderr.Close()
-	}
-
-	return nil
-}

libcontainerd/io_unix.go

@@ -1,60 +0,0 @@
-// +build !windows
-
-package libcontainerd
-
-import (
-	"context"
-	"io"
-	"syscall"
-
-	"github.com/containerd/containerd/cio"
-	"github.com/containerd/fifo"
-	"github.com/pkg/errors"
-)
-
-func newIOPipe(fifos *cio.FIFOSet) (*IOPipe, error) {
-	var (
-		err         error
-		ctx, cancel = context.WithCancel(context.Background())
-		f           io.ReadWriteCloser
-		iop         = &IOPipe{
-			Terminal: fifos.Terminal,
-			cancel:   cancel,
-			config: cio.Config{
-				Terminal: fifos.Terminal,
-				Stdin:    fifos.In,
-				Stdout:   fifos.Out,
-				Stderr:   fifos.Err,
-			},
-		}
-	)
-	defer func() {
-		if err != nil {
-			cancel()
-			iop.Close()
-		}
-	}()
-
-	if fifos.In != "" {
-		if f, err = fifo.OpenFifo(ctx, fifos.In, syscall.O_WRONLY|syscall.O_CREAT|syscall.O_NONBLOCK, 0700); err != nil {
-			return nil, errors.WithStack(err)
-		}
-		iop.Stdin = f
-	}
-
-	if fifos.Out != "" {
-		if f, err = fifo.OpenFifo(ctx, fifos.Out, syscall.O_RDONLY|syscall.O_CREAT|syscall.O_NONBLOCK, 0700); err != nil {
-			return nil, errors.WithStack(err)
-		}
-		iop.Stdout = f
-	}
-
-	if fifos.Err != "" {
-		if f, err = fifo.OpenFifo(ctx, fifos.Err, syscall.O_RDONLY|syscall.O_CREAT|syscall.O_NONBLOCK, 0700); err != nil {
-			return nil, errors.WithStack(err)
-		}
-		iop.Stderr = f
-	}
-
-	return iop, nil
-}

libcontainerd/io_windows.go

@@ -1,138 +0,0 @@
-package libcontainerd
-
-import (
-	"context"
-	"io"
-	"net"
-	"sync"
-
-	winio "github.com/Microsoft/go-winio"
-	"github.com/containerd/containerd/cio"
-	"github.com/pkg/errors"
-)
-
-type winpipe struct {
-	sync.Mutex
-
-	ctx      context.Context
-	listener net.Listener
-	readyCh  chan struct{}
-	readyErr error
-
-	client net.Conn
-}
-
-func newWinpipe(ctx context.Context, pipe string) (*winpipe, error) {
-	l, err := winio.ListenPipe(pipe, nil)
-	if err != nil {
-		return nil, errors.Wrapf(err, "%q pipe creation failed", pipe)
-	}
-	wp := &winpipe{
-		ctx:      ctx,
-		listener: l,
-		readyCh:  make(chan struct{}),
-	}
-	go func() {
-		go func() {
-			defer close(wp.readyCh)
-			defer wp.listener.Close()
-			c, err := wp.listener.Accept()
-			if err != nil {
-				wp.Lock()
-				if wp.readyErr == nil {
-					wp.readyErr = err
-				}
-				wp.Unlock()
-				return
-			}
-			wp.client = c
-		}()
-
-		select {
-		case <-wp.readyCh:
-		case <-ctx.Done():
-			wp.Lock()
-			if wp.readyErr == nil {
-				wp.listener.Close()
-				wp.readyErr = ctx.Err()
-			}
-			wp.Unlock()
-		}
-	}()
-
-	return wp, nil
-}
-
-func (wp *winpipe) Read(b []byte) (int, error) {
-	select {
-	case <-wp.ctx.Done():
-		return 0, wp.ctx.Err()
-	case <-wp.readyCh:
-		return wp.client.Read(b)
-	}
-}
-
-func (wp *winpipe) Write(b []byte) (int, error) {
-	select {
-	case <-wp.ctx.Done():
-		return 0, wp.ctx.Err()
-	case <-wp.readyCh:
-		return wp.client.Write(b)
-	}
-}
-
-func (wp *winpipe) Close() error {
-	select {
-	case <-wp.readyCh:
-		return wp.client.Close()
-	default:
-		return nil
-	}
-}
-
-func newIOPipe(fifos *cio.FIFOSet) (*IOPipe, error) {
-	var (
-		err         error
-		ctx, cancel = context.WithCancel(context.Background())
-		p           io.ReadWriteCloser
-		iop         = &IOPipe{
-			Terminal: fifos.Terminal,
-			cancel:   cancel,
-			config: cio.Config{
-				Terminal: fifos.Terminal,
-				Stdin:    fifos.In,
-				Stdout:   fifos.Out,
-				Stderr:   fifos.Err,
-			},
-		}
-	)
-	defer func() {
-		if err != nil {
-			cancel()
-			iop.Close()
-		}
-	}()
-
-	if fifos.In != "" {
-		if p, err = newWinpipe(ctx, fifos.In); err != nil {
-			return nil, err
-		}
-		iop.Stdin = p
-	}
-
-	if fifos.Out != "" {
-		if p, err = newWinpipe(ctx, fifos.Out); err != nil {
-			return nil, err
-		}
-		iop.Stdout = p
-	}
-
-	if fifos.Err != "" {
-		if p, err = newWinpipe(ctx, fifos.Err); err != nil {
-			return nil, err
-		}
-		iop.Stderr = p
-	}
-
-	return iop, nil
-}

libcontainerd/remote_daemon_process.go

@@ -1,56 +0,0 @@
-// +build !windows
-
-package libcontainerd
-
-import "github.com/pkg/errors"
-
-// process represents the state for the main container process or an exec.
-type process struct {
-	// id is the logical name of the process
-	id string
-
-	// cid is the container id to which this process belongs
-	cid string
-
-	// pid is the identifier of the process
-	pid uint32
-
-	// io holds the io reader/writer associated with the process
-	io *IOPipe
-
-	// root is the state directory for the process
-	root string
-}
-
-func (p *process) ID() string {
-	return p.id
-}
-
-func (p *process) Pid() uint32 {
-	return p.pid
-}
-
-func (p *process) SetPid(pid uint32) error {
-	if p.pid != 0 {
-		return errors.Errorf("pid is already set to %d", pid)
-	}
-
-	p.pid = pid
-	return nil
-}
-
-func (p *process) IOPipe() *IOPipe {
-	return p.io
-}
-
-func (p *process) CloseIO() {
-	if p.io.Stdin != nil {
-		p.io.Stdin.Close()
-	}
-	if p.io.Stdout != nil {
-		p.io.Stdout.Close()
-	}
-	if p.io.Stderr != nil {
-		p.io.Stderr.Close()
-	}
-}

libcontainerd/remote_daemon_process_linux.go

@@ -1,59 +0,0 @@
-package libcontainerd
-
-import (
-	"os"
-	"path/filepath"
-
-	"github.com/pkg/errors"
-	"golang.org/x/sys/unix"
-)
-
-var fdNames = map[int]string{
-	unix.Stdin:  "stdin",
-	unix.Stdout: "stdout",
-	unix.Stderr: "stderr",
-}
-
-func (p *process) pipeName(index int) string {
-	return filepath.Join(p.root, p.id+"-"+fdNames[index])
-}
-
-func (p *process) IOPaths() (string, string, string) {
-	var (
-		stdin  = p.pipeName(unix.Stdin)
-		stdout = p.pipeName(unix.Stdout)
-		stderr = p.pipeName(unix.Stderr)
-	)
-	// TODO: debug why we're having zombies when I don't unset those
-	if p.io.Stdin == nil {
-		stdin = ""
-	}
-	if p.io.Stderr == nil {
-		stderr = ""
-	}
-	return stdin, stdout, stderr
-}
-
-func (p *process) Cleanup() error {
-	var retErr error
-
-	// Ensure everything was closed
-	p.CloseIO()
-
-	for _, i := range [3]string{
-		p.pipeName(unix.Stdin),
-		p.pipeName(unix.Stdout),
-		p.pipeName(unix.Stderr),
-	} {
-		err := os.Remove(i)
-		if err != nil {
-			if retErr == nil {
-				retErr = errors.Wrapf(err, "failed to remove %s", i)
-			} else {
-				retErr = errors.Wrapf(retErr, "failed to remove %s", i)
-			}
-		}
-	}
-
-	return retErr
-}

libcontainerd/types.go

@@ -2,7 +2,6 @@ package libcontainerd
 
 import (
 	"context"
-	"io"
 	"time"
 
 	"github.com/containerd/containerd"
@@ -107,20 +106,4 @@ type Client interface {
 }
 
 // StdioCallback is called to connect a container or process stdio.
-type StdioCallback func(*IOPipe) (cio.IO, error)
-
-// IOPipe contains the stdio streams.
-type IOPipe struct {
-	Stdin    io.WriteCloser
-	Stdout   io.ReadCloser
-	Stderr   io.ReadCloser
-	Terminal bool // Whether stderr is connected on Windows
-
-	cancel context.CancelFunc
-	config cio.Config
-}
-
-// ServerVersion contains version information as retrieved from the
-// server
-type ServerVersion struct {
-}
+type StdioCallback func(io *cio.DirectIO) (cio.IO, error)

plugin/executor/containerd/containerd.go

@@ -122,7 +122,7 @@ func (c *rio) Wait() {
 }
 
 func attachStreamsFunc(stdout, stderr io.WriteCloser) libcontainerd.StdioCallback {
-	return func(iop *libcontainerd.IOPipe) (cio.IO, error) {
+	return func(iop *cio.DirectIO) (cio.IO, error) {
 		if iop.Stdin != nil {
 			iop.Stdin.Close()
 			// closing stdin shouldn't be needed here, it should never be open

vendor.conf

@@ -66,8 +66,8 @@ google.golang.org/grpc v1.3.0
 
 # When updating, also update RUNC_COMMIT in hack/dockerfile/binaries-commits accordingly
 github.com/opencontainers/runc b2567b37d7b75eb4cf325b77297b140ea686ce8f
-github.com/opencontainers/runtime-spec v1.0.0
-github.com/opencontainers/image-spec v1.0.0
+github.com/opencontainers/runtime-spec v1.0.1
+github.com/opencontainers/image-spec v1.0.1
 github.com/seccomp/libseccomp-golang 32f571b70023028bd57d9288c20efbcb237f3ce0
 
 # libcontainer deps (see src/github.com/opencontainers/runc/Godeps/Godeps.json)
@@ -103,7 +103,7 @@ github.com/googleapis/gax-go da06d194a00e19ce00d9011a13931c3f6f6887c7
 google.golang.org/genproto d80a6e20e776b0b17a324d0ba1ab50a39c8e8944
 
 # containerd
-github.com/containerd/containerd 89623f28b87a6004d4b785663257362d1658a729 # v1.0.0
+github.com/containerd/containerd 3fa104f843ec92328912e042b767d26825f202aa
 github.com/containerd/fifo fbfb6a11ec671efbe94ad1c12c2e98773f19e1e6
 github.com/containerd/continuity 35d55c5e8dd23b32037d56cf97174aff3efdfa83
 github.com/containerd/cgroups 29da22c6171a4316169f9205ab6c49f59b5b852f

vendor/github.com/containerd/containerd/archive/tar.go

@@ -381,8 +381,10 @@ func (cw *changeWriter) HandleChange(k fs.ChangeKind, p string, f os.FileInfo, e
 				additionalLinks = cw.inodeRefs[inode]
 				delete(cw.inodeRefs, inode)
 			}
-		} else if k == fs.ChangeKindUnmodified {
+		} else if k == fs.ChangeKindUnmodified && !f.IsDir() {
 			// Nothing to write to diff
+			// Unmodified directories should still be written to keep
+			// directory permissions correct on direct unpack
 			return nil
 		}
 

vendor/github.com/containerd/containerd/cio/io.go

@@ -8,7 +8,7 @@ import (
 	"sync"
 )
 
-// Config holds the io configurations.
+// Config holds the IO configurations.
 type Config struct {
 	// Terminal is true if one has been allocated
 	Terminal bool
@@ -24,173 +24,157 @@ type Config struct {
 type IO interface {
 	// Config returns the IO configuration.
 	Config() Config
-	// Cancel aborts all current io operations
+	// Cancel aborts all current io operations.
 	Cancel()
-	// Wait blocks until all io copy operations have completed
+	// Wait blocks until all io copy operations have completed.
 	Wait()
-	// Close cleans up all open io resources
+	// Close cleans up all open io resources. Cancel() is always called before
+	// Close()
 	Close() error
 }
 
-// cio is a basic container IO implementation.
-type cio struct {
-	config Config
+// Creator creates new IO sets for a task
+type Creator func(id string) (IO, error)
 
-	closer *wgCloser
-}
+// Attach allows callers to reattach to running tasks
+//
+// There should only be one reader for a task's IO set
+// because fifo's can only be read from one reader or the output
+// will be sent only to the first reads
+type Attach func(*FIFOSet) (IO, error)
 
-func (c *cio) Config() Config {
-	return c.config
+// FIFOSet is a set of file paths to FIFOs for a task's standard IO streams
+type FIFOSet struct {
+	Config
+	close func() error
 }
 
-func (c *cio) Cancel() {
-	if c.closer == nil {
-		return
+// Close the FIFOSet
+func (f *FIFOSet) Close() error {
+	if f.close != nil {
+		return f.close()
 	}
-	c.closer.Cancel()
+	return nil
 }
 
-func (c *cio) Wait() {
-	if c.closer == nil {
-		return
-	}
-	c.closer.Wait()
+// NewFIFOSet returns a new FIFOSet from a Config and a close function
+func NewFIFOSet(config Config, close func() error) *FIFOSet {
+	return &FIFOSet{Config: config, close: close}
 }
 
-func (c *cio) Close() error {
-	if c.closer == nil {
-		return nil
-	}
-	return c.closer.Close()
+// Streams used to configure a Creator or Attach
+type Streams struct {
+	Stdin    io.Reader
+	Stdout   io.Writer
+	Stderr   io.Writer
+	Terminal bool
 }
 
-// Creation creates new IO sets for a task
-type Creation func(id string) (IO, error)
+// Opt customize options for creating a Creator or Attach
+type Opt func(*Streams)
 
-// Attach allows callers to reattach to running tasks
-//
-// There should only be one reader for a task's IO set
-// because fifo's can only be read from one reader or the output
-// will be sent only to the first reads
-type Attach func(*FIFOSet) (IO, error)
+// WithStdio sets stream options to the standard input/output streams
+func WithStdio(opt *Streams) {
+	WithStreams(os.Stdin, os.Stdout, os.Stderr)(opt)
+}
 
-// NewIO returns an Creation that will provide IO sets without a terminal
-func NewIO(stdin io.Reader, stdout, stderr io.Writer) Creation {
-	return NewIOWithTerminal(stdin, stdout, stderr, false)
+// WithTerminal sets the terminal option
+func WithTerminal(opt *Streams) {
+	opt.Terminal = true
 }
 
-// NewIOWithTerminal creates a new io set with the provied io.Reader/Writers for use with a terminal
-func NewIOWithTerminal(stdin io.Reader, stdout, stderr io.Writer, terminal bool) Creation {
-	return func(id string) (_ IO, err error) {
-		paths, err := NewFifos(id)
-		if err != nil {
-			return nil, err
-		}
-		defer func() {
-			if err != nil && paths.Dir != "" {
-				os.RemoveAll(paths.Dir)
-			}
-		}()
-		cfg := Config{
-			Terminal: terminal,
-			Stdout:   paths.Out,
-			Stderr:   paths.Err,
-			Stdin:    paths.In,
-		}
-		i := &cio{config: cfg}
-		set := &ioSet{
-			in:  stdin,
-			out: stdout,
-			err: stderr,
-		}
-		closer, err := copyIO(paths, set, cfg.Terminal)
-		if err != nil {
-			return nil, err
-		}
-		i.closer = closer
-		return i, nil
+// WithStreams sets the stream options to the specified Reader and Writers
+func WithStreams(stdin io.Reader, stdout, stderr io.Writer) Opt {
+	return func(opt *Streams) {
+		opt.Stdin = stdin
+		opt.Stdout = stdout
+		opt.Stderr = stderr
 	}
 }
 
-// WithAttach attaches the existing io for a task to the provided io.Reader/Writers
-func WithAttach(stdin io.Reader, stdout, stderr io.Writer) Attach {
-	return func(paths *FIFOSet) (IO, error) {
-		if paths == nil {
-			return nil, fmt.Errorf("cannot attach to existing fifos")
-		}
-		cfg := Config{
-			Terminal: paths.Terminal,
-			Stdout:   paths.Out,
-			Stderr:   paths.Err,
-			Stdin:    paths.In,
-		}
-		i := &cio{config: cfg}
-		set := &ioSet{
-			in:  stdin,
-			out: stdout,
-			err: stderr,
-		}
-		closer, err := copyIO(paths, set, cfg.Terminal)
+// NewCreator returns an IO creator from the options
+func NewCreator(opts ...Opt) Creator {
+	streams := &Streams{}
+	for _, opt := range opts {
+		opt(streams)
+	}
+	return func(id string) (IO, error) {
+		// TODO: accept root as a param
+		root := "/run/containerd/fifo"
+		fifos, err := NewFIFOSetInDir(root, id, streams.Terminal)
 		if err != nil {
 			return nil, err
 		}
-		i.closer = closer
-		return i, nil
+		return copyIO(fifos, streams)
 	}
 }
 
-// Stdio returns an IO set to be used for a task
-// that outputs the container's IO as the current processes Stdio
-func Stdio(id string) (IO, error) {
-	return NewIO(os.Stdin, os.Stdout, os.Stderr)(id)
-}
-
-// StdioTerminal will setup the IO for the task to use a terminal
-func StdioTerminal(id string) (IO, error) {
-	return NewIOWithTerminal(os.Stdin, os.Stdout, os.Stderr, true)(id)
+// NewAttach attaches the existing io for a task to the provided io.Reader/Writers
+func NewAttach(opts ...Opt) Attach {
+	streams := &Streams{}
+	for _, opt := range opts {
+		opt(streams)
+	}
+	return func(fifos *FIFOSet) (IO, error) {
+		if fifos == nil {
+			return nil, fmt.Errorf("cannot attach, missing fifos")
+		}
+		return copyIO(fifos, streams)
+	}
 }
 
 // NullIO redirects the container's IO into /dev/null
-func NullIO(id string) (IO, error) {
+func NullIO(_ string) (IO, error) {
 	return &cio{}, nil
 }
 
-// FIFOSet is a set of fifos for use with tasks
-type FIFOSet struct {
-	// Dir is the directory holding the task fifos
-	Dir string
-	// In, Out, and Err fifo paths
-	In, Out, Err string
-	// Terminal returns true if a terminal is being used for the task
-	Terminal bool
+// cio is a basic container IO implementation.
+type cio struct {
+	config  Config
+	wg      *sync.WaitGroup
+	closers []io.Closer
+	cancel  context.CancelFunc
 }
 
-type ioSet struct {
-	in       io.Reader
-	out, err io.Writer
+func (c *cio) Config() Config {
+	return c.config
 }
 
-type wgCloser struct {
-	wg     *sync.WaitGroup
-	dir    string
-	set    []io.Closer
-	cancel context.CancelFunc
+func (c *cio) Wait() {
+	if c.wg != nil {
+		c.wg.Wait()
+	}
 }
 
-func (g *wgCloser) Wait() {
-	g.wg.Wait()
+func (c *cio) Close() error {
+	var lastErr error
+	for _, closer := range c.closers {
+		if closer == nil {
+			continue
+		}
+		if err := closer.Close(); err != nil {
+			lastErr = err
+		}
+	}
+	return lastErr
 }
 
-func (g *wgCloser) Close() error {
-	for _, f := range g.set {
-		f.Close()
-	}
-	if g.dir != "" {
-		return os.RemoveAll(g.dir)
+func (c *cio) Cancel() {
+	if c.cancel != nil {
+		c.cancel()
 	}
-	return nil
 }
 
-func (g *wgCloser) Cancel() {
-	g.cancel()
+type pipes struct {
+	Stdin  io.WriteCloser
+	Stdout io.ReadCloser
+	Stderr io.ReadCloser
 }
+
+// DirectIO allows task IO to be handled externally by the caller
+type DirectIO struct {
+	pipes
+	cio
+}
+
+var _ IO = &DirectIO{}

vendor/github.com/containerd/containerd/cio/io_unix.go

@@ -12,173 +12,115 @@ import (
 	"syscall"
 
 	"github.com/containerd/fifo"
+	"github.com/pkg/errors"
 )
 
-// NewFifos returns a new set of fifos for the task
-func NewFifos(id string) (*FIFOSet, error) {
-	root := "/run/containerd/fifo"
-	if err := os.MkdirAll(root, 0700); err != nil {
-		return nil, err
+// NewFIFOSetInDir returns a new FIFOSet with paths in a temporary directory under root
+func NewFIFOSetInDir(root, id string, terminal bool) (*FIFOSet, error) {
+	if root != "" {
+		if err := os.MkdirAll(root, 0700); err != nil {
+			return nil, err
+		}
 	}
 	dir, err := ioutil.TempDir(root, "")
 	if err != nil {
 		return nil, err
 	}
-	return &FIFOSet{
-		Dir: dir,
-		In:  filepath.Join(dir, id+"-stdin"),
-		Out: filepath.Join(dir, id+"-stdout"),
-		Err: filepath.Join(dir, id+"-stderr"),
-	}, nil
+	closer := func() error {
+		return os.RemoveAll(dir)
+	}
+	return NewFIFOSet(Config{
+		Stdin:    filepath.Join(dir, id+"-stdin"),
+		Stdout:   filepath.Join(dir, id+"-stdout"),
+		Stderr:   filepath.Join(dir, id+"-stderr"),
+		Terminal: terminal,
+	}, closer), nil
 }
 
-func copyIO(fifos *FIFOSet, ioset *ioSet, tty bool) (_ *wgCloser, err error) {
-	var (
-		f           io.ReadWriteCloser
-		set         []io.Closer
-		ctx, cancel = context.WithCancel(context.Background())
-		wg          = &sync.WaitGroup{}
-	)
-	defer func() {
-		if err != nil {
-			for _, f := range set {
-				f.Close()
-			}
-			cancel()
-		}
-	}()
-
-	if f, err = fifo.OpenFifo(ctx, fifos.In, syscall.O_WRONLY|syscall.O_CREAT|syscall.O_NONBLOCK, 0700); err != nil {
+func copyIO(fifos *FIFOSet, ioset *Streams) (*cio, error) {
+	var ctx, cancel = context.WithCancel(context.Background())
+	pipes, err := openFifos(ctx, fifos)
+	if err != nil {
+		cancel()
 		return nil, err
 	}
-	set = append(set, f)
-	go func(w io.WriteCloser) {
-		io.Copy(w, ioset.in)
-		w.Close()
-	}(f)
 
-	if f, err = fifo.OpenFifo(ctx, fifos.Out, syscall.O_RDONLY|syscall.O_CREAT|syscall.O_NONBLOCK, 0700); err != nil {
-		return nil, err
+	if fifos.Stdin != "" {
+		go func() {
+			io.Copy(pipes.Stdin, ioset.Stdin)
+			pipes.Stdin.Close()
+		}()
 	}
-	set = append(set, f)
+
+	var wg = &sync.WaitGroup{}
 	wg.Add(1)
-	go func(r io.ReadCloser) {
-		io.Copy(ioset.out, r)
-		r.Close()
+	go func() {
+		io.Copy(ioset.Stdout, pipes.Stdout)
+		pipes.Stdout.Close()
 		wg.Done()
-	}(f)
-
-	if f, err = fifo.OpenFifo(ctx, fifos.Err, syscall.O_RDONLY|syscall.O_CREAT|syscall.O_NONBLOCK, 0700); err != nil {
-		return nil, err
-	}
-	set = append(set, f)
+	}()
 
-	if !tty {
+	if !fifos.Terminal {
 		wg.Add(1)
-		go func(r io.ReadCloser) {
-			io.Copy(ioset.err, r)
-			r.Close()
+		go func() {
+			io.Copy(ioset.Stderr, pipes.Stderr)
+			pipes.Stderr.Close()
 			wg.Done()
-		}(f)
+		}()
 	}
-	return &wgCloser{
-		wg:     wg,
-		dir:    fifos.Dir,
-		set:    set,
-		cancel: cancel,
+	return &cio{
+		config:  fifos.Config,
+		wg:      wg,
+		closers: append(pipes.closers(), fifos),
+		cancel:  cancel,
 	}, nil
 }
 
-// NewDirectIO returns an IO implementation that exposes the pipes directly
-func NewDirectIO(ctx context.Context, terminal bool) (*DirectIO, error) {
-	set, err := NewFifos("")
-	if err != nil {
-		return nil, err
-	}
-	f := &DirectIO{
-		set:      set,
-		terminal: terminal,
-	}
+func openFifos(ctx context.Context, fifos *FIFOSet) (pipes, error) {
+	var err error
 	defer func() {
 		if err != nil {
-			f.Delete()
+			fifos.Close()
 		}
 	}()
-	if f.Stdin, err = fifo.OpenFifo(ctx, set.In, syscall.O_WRONLY|syscall.O_CREAT|syscall.O_NONBLOCK, 0700); err != nil {
-		return nil, err
+
+	var f pipes
+	if fifos.Stdin != "" {
+		if f.Stdin, err = fifo.OpenFifo(ctx, fifos.Stdin, syscall.O_WRONLY|syscall.O_CREAT|syscall.O_NONBLOCK, 0700); err != nil {
+			return f, errors.Wrapf(err, "failed to open stdin fifo")
+		}
 	}
-	if f.Stdout, err = fifo.OpenFifo(ctx, set.Out, syscall.O_RDONLY|syscall.O_CREAT|syscall.O_NONBLOCK, 0700); err != nil {
-		f.Stdin.Close()
-		return nil, err
+	if fifos.Stdout != "" {
+		if f.Stdout, err = fifo.OpenFifo(ctx, fifos.Stdout, syscall.O_RDONLY|syscall.O_CREAT|syscall.O_NONBLOCK, 0700); err != nil {
+			f.Stdin.Close()
+			return f, errors.Wrapf(err, "failed to open stdout fifo")
+		}
 	}
-	if f.Stderr, err = fifo.OpenFifo(ctx, set.Err, syscall.O_RDONLY|syscall.O_CREAT|syscall.O_NONBLOCK, 0700); err != nil {
-		f.Stdin.Close()
-		f.Stdout.Close()
-		return nil, err
+	if fifos.Stderr != "" {
+		if f.Stderr, err = fifo.OpenFifo(ctx, fifos.Stderr, syscall.O_RDONLY|syscall.O_CREAT|syscall.O_NONBLOCK, 0700); err != nil {
+			f.Stdin.Close()
+			f.Stdout.Close()
+			return f, errors.Wrapf(err, "failed to open stderr fifo")
+		}
 	}
 	return f, nil
 }
 
-// DirectIO allows task IO to be handled externally by the caller
-type DirectIO struct {
-	Stdin  io.WriteCloser
-	Stdout io.ReadCloser
-	Stderr io.ReadCloser
-
-	set      *FIFOSet
-	terminal bool
-}
-
-// IOCreate returns IO avaliable for use with task creation
-func (f *DirectIO) IOCreate(id string) (IO, error) {
-	return f, nil
-}
-
-// IOAttach returns IO avaliable for use with task attachment
-func (f *DirectIO) IOAttach(set *FIFOSet) (IO, error) {
-	return f, nil
+// NewDirectIO returns an IO implementation that exposes the IO streams as io.ReadCloser
+// and io.WriteCloser.
+func NewDirectIO(ctx context.Context, fifos *FIFOSet) (*DirectIO, error) {
+	ctx, cancel := context.WithCancel(ctx)
+	pipes, err := openFifos(ctx, fifos)
+	return &DirectIO{
+		pipes: pipes,
+		cio: cio{
+			config:  fifos.Config,
+			closers: append(pipes.closers(), fifos),
+			cancel:  cancel,
+		},
+	}, err
 }
 
-// Config returns the Config
-func (f *DirectIO) Config() Config {
-	return Config{
-		Terminal: f.terminal,
-		Stdin:    f.set.In,
-		Stdout:   f.set.Out,
-		Stderr:   f.set.Err,
-	}
-}
-
-// Cancel stops any IO copy operations
-//
-// Not applicable for DirectIO
-func (f *DirectIO) Cancel() {
-	// nothing to cancel as all operations are handled externally
-}
-
-// Wait on any IO copy operations
-//
-// Not applicable for DirectIO
-func (f *DirectIO) Wait() {
-	// nothing to wait on as all operations are handled externally
-}
-
-// Close closes all open fds
-func (f *DirectIO) Close() error {
-	err := f.Stdin.Close()
-	if err2 := f.Stdout.Close(); err == nil {
-		err = err2
-	}
-	if err2 := f.Stderr.Close(); err == nil {
-		err = err2
-	}
-	return err
-}
-
-// Delete removes the underlying directory containing fifos
-func (f *DirectIO) Delete() error {
-	if f.set.Dir == "" {
-		return nil
-	}
-	return os.RemoveAll(f.set.Dir)
+func (p *pipes) closers() []io.Closer {
+	return []io.Closer{p.Stdin, p.Stdout, p.Stderr}
 }

vendor/github.com/containerd/containerd/cio/io_windows.go

@@ -13,25 +13,26 @@ import (
 
 const pipeRoot = `\\.\pipe`
 
-// NewFifos returns a new set of fifos for the task
-func NewFifos(id string) (*FIFOSet, error) {
-	return &FIFOSet{
-		In:  fmt.Sprintf(`%s\ctr-%s-stdin`, pipeRoot, id),
-		Out: fmt.Sprintf(`%s\ctr-%s-stdout`, pipeRoot, id),
-		Err: fmt.Sprintf(`%s\ctr-%s-stderr`, pipeRoot, id),
-	}, nil
+// NewFIFOSetInDir returns a new set of fifos for the task
+func NewFIFOSetInDir(_, id string, terminal bool) (*FIFOSet, error) {
+	return NewFIFOSet(Config{
+		Terminal: terminal,
+		Stdin:    fmt.Sprintf(`%s\ctr-%s-stdin`, pipeRoot, id),
+		Stdout:   fmt.Sprintf(`%s\ctr-%s-stdout`, pipeRoot, id),
+		Stderr:   fmt.Sprintf(`%s\ctr-%s-stderr`, pipeRoot, id),
+	}, nil), nil
 }
 
-func copyIO(fifos *FIFOSet, ioset *ioSet, tty bool) (_ *wgCloser, err error) {
+func copyIO(fifos *FIFOSet, ioset *Streams) (*cio, error) {
 	var (
 		wg  sync.WaitGroup
 		set []io.Closer
 	)
 
-	if fifos.In != "" {
-		l, err := winio.ListenPipe(fifos.In, nil)
+	if fifos.Stdin != "" {
+		l, err := winio.ListenPipe(fifos.Stdin, nil)
 		if err != nil {
-			return nil, errors.Wrapf(err, "failed to create stdin pipe %s", fifos.In)
+			return nil, errors.Wrapf(err, "failed to create stdin pipe %s", fifos.Stdin)
 		}
 		defer func(l net.Listener) {
 			if err != nil {
@@ -43,19 +44,19 @@ func copyIO(fifos *FIFOSet, ioset *ioSet, tty bool) (_ *wgCloser, err error) {
 		go func() {
 			c, err := l.Accept()
 			if err != nil {
-				log.L.WithError(err).Errorf("failed to accept stdin connection on %s", fifos.In)
+				log.L.WithError(err).Errorf("failed to accept stdin connection on %s", fifos.Stdin)
 				return
 			}
-			io.Copy(c, ioset.in)
+			io.Copy(c, ioset.Stdin)
 			c.Close()
 			l.Close()
 		}()
 	}
 
-	if fifos.Out != "" {
-		l, err := winio.ListenPipe(fifos.Out, nil)
+	if fifos.Stdout != "" {
+		l, err := winio.ListenPipe(fifos.Stdout, nil)
 		if err != nil {
-			return nil, errors.Wrapf(err, "failed to create stdin pipe %s", fifos.Out)
+			return nil, errors.Wrapf(err, "failed to create stdin pipe %s", fifos.Stdout)
 		}
 		defer func(l net.Listener) {
 			if err != nil {
@@ -69,19 +70,19 @@ func copyIO(fifos *FIFOSet, ioset *ioSet, tty bool) (_ *wgCloser, err error) {
 			defer wg.Done()
 			c, err := l.Accept()
 			if err != nil {
-				log.L.WithError(err).Errorf("failed to accept stdout connection on %s", fifos.Out)
+				log.L.WithError(err).Errorf("failed to accept stdout connection on %s", fifos.Stdout)
 				return
 			}
-			io.Copy(ioset.out, c)
+			io.Copy(ioset.Stdout, c)
 			c.Close()
 			l.Close()
 		}()
 	}
 
-	if !tty && fifos.Err != "" {
-		l, err := winio.ListenPipe(fifos.Err, nil)
+	if !fifos.Terminal && fifos.Stderr != "" {
+		l, err := winio.ListenPipe(fifos.Stderr, nil)
 		if err != nil {
-			return nil, errors.Wrapf(err, "failed to create stderr pipe %s", fifos.Err)
+			return nil, errors.Wrapf(err, "failed to create stderr pipe %s", fifos.Stderr)
 		}
 		defer func(l net.Listener) {
 			if err != nil {
@@ -95,23 +96,29 @@ func copyIO(fifos *FIFOSet, ioset *ioSet, tty bool) (_ *wgCloser, err error) {
 			defer wg.Done()
 			c, err := l.Accept()
 			if err != nil {
-				log.L.WithError(err).Errorf("failed to accept stderr connection on %s", fifos.Err)
+				log.L.WithError(err).Errorf("failed to accept stderr connection on %s", fifos.Stderr)
 				return
 			}
-			io.Copy(ioset.err, c)
+			io.Copy(ioset.Stderr, c)
 			c.Close()
 			l.Close()
 		}()
 	}
 
-	return &wgCloser{
-		wg:  &wg,
-		dir: fifos.Dir,
-		set: set,
-		cancel: func() {
-			for _, l := range set {
-				l.Close()
-			}
+	return &cio{config: fifos.Config, closers: set}, nil
+}
+
+// NewDirectIO returns an IO implementation that exposes the IO streams as io.ReadCloser
+// and io.WriteCloser.
+func NewDirectIO(stdin io.WriteCloser, stdout, stderr io.ReadCloser, terminal bool) *DirectIO {
+	return &DirectIO{
+		pipes: pipes{
+			Stdin:  stdin,
+			Stdout: stdout,
+			Stderr: stderr,
+		},
+		cio: cio{
+			config: Config{Terminal: terminal},
 		},
-	}, nil
+	}
 }

vendor/github.com/containerd/containerd/container.go

@@ -3,6 +3,7 @@ package containerd
 import (
 	"context"
 	"encoding/json"
+	"os"
 	"path/filepath"
 	"strings"
 
@@ -26,7 +27,7 @@ type Container interface {
 	// Delete removes the container
 	Delete(context.Context, ...DeleteOpts) error
 	// NewTask creates a new task based on the container metadata
-	NewTask(context.Context, cio.Creation, ...NewTaskOpts) (Task, error)
+	NewTask(context.Context, cio.Creator, ...NewTaskOpts) (Task, error)
 	// Spec returns the OCI runtime specification
 	Spec(context.Context) (*specs.Spec, error)
 	// Task returns the current task for the container
@@ -162,7 +163,7 @@ func (c *container) Image(ctx context.Context) (Image, error) {
 	}, nil
 }
 
-func (c *container) NewTask(ctx context.Context, ioCreate cio.Creation, opts ...NewTaskOpts) (_ Task, err error) {
+func (c *container) NewTask(ctx context.Context, ioCreate cio.Creator, opts ...NewTaskOpts) (_ Task, err error) {
 	i, err := ioCreate(c.id)
 	if err != nil {
 		return nil, err
@@ -288,20 +289,23 @@ func (c *container) get(ctx context.Context) (containers.Container, error) {
 	return c.client.ContainerService().Get(ctx, c.id)
 }
 
+// get the existing fifo paths from the task information stored by the daemon
 func attachExistingIO(response *tasks.GetResponse, ioAttach cio.Attach) (cio.IO, error) {
-	// get the existing fifo paths from the task information stored by the daemon
-	paths := &cio.FIFOSet{
-		Dir: getFifoDir([]string{
-			response.Process.Stdin,
-			response.Process.Stdout,
-			response.Process.Stderr,
-		}),
-		In:       response.Process.Stdin,
-		Out:      response.Process.Stdout,
-		Err:      response.Process.Stderr,
-		Terminal: response.Process.Terminal,
+	path := getFifoDir([]string{
+		response.Process.Stdin,
+		response.Process.Stdout,
+		response.Process.Stderr,
+	})
+	closer := func() error {
+		return os.RemoveAll(path)
 	}
-	return ioAttach(paths)
+	fifoSet := cio.NewFIFOSet(cio.Config{
+		Stdin:    response.Process.Stdin,
+		Stdout:   response.Process.Stdout,
+		Stderr:   response.Process.Stderr,
+		Terminal: response.Process.Terminal,
+	}, closer)
+	return ioAttach(fifoSet)
 }
 
 // getFifoDir looks for any non-empty path for a stdio fifo

vendor/github.com/containerd/containerd/content/helpers.go

@@ -77,7 +77,7 @@ func Copy(ctx context.Context, cw Writer, r io.Reader, size int64, expected dige
 		r, err = seekReader(r, ws.Offset, size)
 		if err != nil {
 			if !isUnseekable(err) {
-				return errors.Wrapf(err, "unabled to resume write to %v", ws.Ref)
+				return errors.Wrapf(err, "unable to resume write to %v", ws.Ref)
 			}
 
 			// reader is unseekable, try to move the writer back to the start.

vendor/github.com/containerd/containerd/content/local/writer.go

@@ -2,6 +2,7 @@ package local
 
 import (
 	"context"
+	"io"
 	"os"
 	"path/filepath"
 	"runtime"
@@ -167,5 +168,8 @@ func (w *writer) Truncate(size int64) error {
 	}
 	w.offset = 0
 	w.digester.Hash().Reset()
+	if _, err := w.fp.Seek(0, io.SeekStart); err != nil {
+		return err
+	}
 	return w.fp.Truncate(0)
 }

vendor/github.com/containerd/containerd/linux/shim/client/client.go

@@ -47,7 +47,10 @@ func WithStart(binary, address, daemonAddress, cgroup string, debug bool, exitHa
 		}
 		defer f.Close()
 
-		cmd := newCommand(binary, daemonAddress, debug, config, f)
+		cmd, err := newCommand(binary, daemonAddress, debug, config, f)
+		if err != nil {
+			return nil, nil, err
+		}
 		ec, err := reaper.Default.Start(cmd)
 		if err != nil {
 			return nil, nil, errors.Wrapf(err, "failed to start shim")
@@ -87,10 +90,10 @@ func WithStart(binary, address, daemonAddress, cgroup string, debug bool, exitHa
 	}
 }
 
-func newCommand(binary, daemonAddress string, debug bool, config shim.Config, socket *os.File) *exec.Cmd {
+func newCommand(binary, daemonAddress string, debug bool, config shim.Config, socket *os.File) (*exec.Cmd, error) {
 	selfExe, err := os.Executable()
 	if err != nil {
-		panic(err)
+		return nil, err
 	}
 	args := []string{
 		"-namespace", config.Namespace,
@@ -123,7 +126,7 @@ func newCommand(binary, daemonAddress string, debug bool, config shim.Config, so
 		cmd.Stdout = os.Stdout
 		cmd.Stderr = os.Stderr
 	}
-	return cmd
+	return cmd, nil
 }
 
 func newSocket(address string) (*net.UnixListener, error) {

vendor/github.com/containerd/containerd/linux/shim/service.go

@@ -235,8 +235,8 @@ func (s *Service) ResizePty(ctx context.Context, r *shimapi.ResizePtyRequest) (*
 // State returns runtime state information for a process
 func (s *Service) State(ctx context.Context, r *shimapi.StateRequest) (*shimapi.StateResponse, error) {
 	s.mu.Lock()
-	defer s.mu.Unlock()
 	p := s.processes[r.ID]
+	s.mu.Unlock()
 	if p == nil {
 		return nil, errdefs.ToGRPCf(errdefs.ErrNotFound, "process id %s not found", r.ID)
 	}

vendor/github.com/containerd/containerd/linux/task.go

@@ -282,11 +282,14 @@ func (t *Task) Update(ctx context.Context, resources *types.Any) error {
 
 // Process returns a specific process inside the task by the process id
 func (t *Task) Process(ctx context.Context, id string) (runtime.Process, error) {
-	// TODO: verify process exists for container
-	return &Process{
+	p := &Process{
 		id: id,
 		t:  t,
-	}, nil
+	}
+	if _, err := p.State(ctx); err != nil {
+		return nil, err
+	}
+	return p, nil
 }
 
 // Metrics returns runtime specific system level metric information for the task

vendor/github.com/containerd/containerd/process.go

@@ -191,6 +191,7 @@ func (p *process) Delete(ctx context.Context, opts ...ProcessDeleteOpts) (*ExitS
 		return nil, errdefs.FromGRPC(err)
 	}
 	if p.io != nil {
+		p.io.Cancel()
 		p.io.Wait()
 		p.io.Close()
 	}

vendor/github.com/containerd/containerd/remotes/hints.go

@@ -1,31 +0,0 @@
-package remotes
-
-import "strings"
-
-// HintExists returns true if a hint of the provided kind and values exists in
-// the set of provided hints.
-func HintExists(kind, value string, hints ...string) bool {
-	for _, hint := range hints {
-		if strings.HasPrefix(hint, kind) && strings.HasSuffix(hint, value) {
-			return true
-		}
-	}
-
-	return false
-}
-
-// HintValues returns a slice of the values of the hints that match kind.
-func HintValues(kind string, hints ...string) []string {
-	var values []string
-	for _, hint := range hints {
-		if strings.HasPrefix(hint, kind) {
-			parts := strings.SplitN(hint, ":", 2)
-			if len(parts) < 2 {
-				continue
-			}
-			values = append(values, parts[1])
-		}
-	}
-
-	return values
-}

vendor/github.com/containerd/containerd/server/server_linux.go

@@ -19,9 +19,9 @@ func apply(ctx context.Context, config *Config) error {
 		}
 	}
 	if config.OOMScore != 0 {
-		log.G(ctx).Infof("changing OOM score to %d", config.OOMScore)
+		log.G(ctx).Debugf("changing OOM score to %d", config.OOMScore)
 		if err := sys.SetOOMScore(os.Getpid(), config.OOMScore); err != nil {
-			return err
+			log.G(ctx).WithError(err).Errorf("failed to change OOM score to %d", config.OOMScore)
 		}
 	}
 	if config.Cgroup.Path != "" {

vendor/github.com/containerd/containerd/task.go

@@ -123,7 +123,7 @@ type Task interface {
 	// Resume the execution of the task
 	Resume(context.Context) error
 	// Exec creates a new process inside the task
-	Exec(context.Context, string, *specs.Process, cio.Creation) (Process, error)
+	Exec(context.Context, string, *specs.Process, cio.Creator) (Process, error)
 	// Pids returns a list of system specific process ids inside the task
 	Pids(context.Context) ([]ProcessInfo, error)
 	// Checkpoint serializes the runtime and memory information of a task into an
@@ -163,6 +163,7 @@ func (t *task) Start(ctx context.Context) error {
 		ContainerID: t.id,
 	})
 	if err != nil {
+		t.io.Cancel()
 		t.io.Close()
 		return errdefs.FromGRPC(err)
 	}
@@ -277,7 +278,7 @@ func (t *task) Delete(ctx context.Context, opts ...ProcessDeleteOpts) (*ExitStat
 	return &ExitStatus{code: r.ExitStatus, exitedAt: r.ExitedAt}, nil
 }
 
-func (t *task) Exec(ctx context.Context, id string, spec *specs.Process, ioCreate cio.Creation) (_ Process, err error) {
+func (t *task) Exec(ctx context.Context, id string, spec *specs.Process, ioCreate cio.Creator) (_ Process, err error) {
 	if id == "" {
 		return nil, errors.Wrapf(errdefs.ErrInvalidArgument, "exec id must not be empty")
 	}

vendor/github.com/containerd/containerd/vendor.conf

@@ -15,8 +15,8 @@ github.com/matttproud/golang_protobuf_extensions v1.0.0
 github.com/docker/go-units v0.3.1
 github.com/gogo/protobuf v0.5
 github.com/golang/protobuf 1643683e1b54a9e88ad26d98f81400c8c9d9f4f9
-github.com/opencontainers/runtime-spec v1.0.0
-github.com/opencontainers/runc 74a17296470088de3805e138d3d87c62e613dfc4
+github.com/opencontainers/runtime-spec v1.0.1
+github.com/opencontainers/runc 7f24b40cc5423969b4554ef04ba0b00e2b4ba010
 github.com/sirupsen/logrus v1.0.0
 github.com/containerd/btrfs cc52c4dea2ce11a44e6639e561bb5c2af9ada9e3
 github.com/stretchr/testify v1.1.4
@@ -25,18 +25,17 @@ github.com/pmezard/go-difflib v1.0.0
 github.com/containerd/fifo fbfb6a11ec671efbe94ad1c12c2e98773f19e1e6
 github.com/urfave/cli 7bc6a0acffa589f415f88aca16cc1de5ffd66f9c
 golang.org/x/net 7dcfb8076726a3fdd9353b6b8a1f1b6be6811bd6
-google.golang.org/grpc v1.7.2
+google.golang.org/grpc v1.7.4
 github.com/pkg/errors v0.8.0
 github.com/opencontainers/go-digest 21dfd564fd89c944783d00d069f33e3e7123c448
 golang.org/x/sys 314a259e304ff91bd6985da2a7149bbf91237993 https://github.com/golang/sys
-github.com/opencontainers/image-spec v1.0.0
+github.com/opencontainers/image-spec v1.0.1
 github.com/containerd/continuity cf279e6ac893682272b4479d4c67fd3abf878b4e
 golang.org/x/sync 450f422ab23cf9881c94e2db30cac0eb1b7cf80c
 github.com/BurntSushi/toml v0.2.0-21-g9906417
 github.com/grpc-ecosystem/go-grpc-prometheus 6b7015e65d366bf3f19b2b2a000a831940f0f7e0
 github.com/Microsoft/go-winio v0.4.4
 github.com/Microsoft/hcsshim v0.6.7
-github.com/Microsoft/opengcs v0.3.2
 github.com/boltdb/bolt e9cf4fae01b5a8ff89d0ec6b32f0d9c9f79aefdd
 google.golang.org/genproto d80a6e20e776b0b17a324d0ba1ab50a39c8e8944
 golang.org/x/text 19e51611da83d6be54ddafce4a4af510cb3e9ea4

vendor/github.com/opencontainers/image-spec/README.md

@@ -51,7 +51,7 @@ Find more [FAQ on the OCI site](https://www.opencontainers.org/faq).
 
 ## Roadmap
 
-The [GitHub milestones](https://github.com/opencontainers/image-spec/milestones) lay out the path to the OCI v1.0.0 release in late 2016.
+The [GitHub milestones](https://github.com/opencontainers/image-spec/milestones) lay out the path to the future improvements.
 
 # Contributing
 

vendor/github.com/opencontainers/image-spec/specs-go/version.go

@@ -22,7 +22,7 @@ const (
 	// VersionMinor is for functionality in a backwards-compatible manner
 	VersionMinor = 0
 	// VersionPatch is for backwards-compatible bug fixes
-	VersionPatch = 0
+	VersionPatch = 1
 
 	// VersionDev indicates development branch. Releases will be empty string.
 	VersionDev = ""

vendor/github.com/opencontainers/runtime-spec/README.md

@@ -52,17 +52,12 @@ It also guarantees that the design is sound before code is written; a GitHub pul
 Typos and grammatical errors can go straight to a pull-request.
 When in doubt, start on the [mailing-list](#mailing-list).
 
-### Weekly Call
-
-The contributors and maintainers of all OCI projects have a weekly meeting on Wednesdays at:
-
-* 8:00 AM (USA Pacific), during [odd weeks][iso-week].
-* 2:00 PM (USA Pacific), during [even weeks][iso-week].
+### Meetings
 
+The contributors and maintainers of all OCI projects have monthly meetings at 2:00 PM (USA Pacific) on the first Wednesday of every month.
 There is an [iCalendar][rfc5545] format for the meetings [here](meeting.ics).
-
 Everyone is welcome to participate via [UberConference web][uberconference] or audio-only: +1 415 968 0849 (no PIN needed).
-An initial agenda will be posted to the [mailing list](#mailing-list) earlier in the week, and everyone is welcome to propose additional topics or suggest other agenda alterations there.
+An initial agenda will be posted to the [mailing list](#mailing-list) in the week before each meeting, and everyone is welcome to propose additional topics or suggest other agenda alterations there.
 Minutes are posted to the [mailing list](#mailing-list) and minutes from past calls are archived [here][minutes], with minutes from especially old meetings (September 2015 and earlier) archived [here][runtime-wiki].
 
 ### Mailing List

vendor/github.com/opencontainers/runtime-spec/specs-go/config.go

@@ -4,7 +4,7 @@ import "os"
 
 // Spec is the base configuration for the container.
 type Spec struct {
-	// Version of the Open Container Runtime Specification with which the bundle complies.
+	// Version of the Open Container Initiative Runtime Specification with which the bundle complies.
 	Version string `json:"ociVersion"`
 	// Process configures the container process.
 	Process *Process `json:"process,omitempty"`

vendor/github.com/opencontainers/runtime-spec/specs-go/version.go

@@ -8,7 +8,7 @@ const (
 	// VersionMinor is for functionality in a backwards-compatible manner
 	VersionMinor = 0
 	// VersionPatch is for backwards-compatible bug fixes
-	VersionPatch = 0
+	VersionPatch = 1
 
 	// VersionDev indicates development branch. Releases will be empty string.
 	VersionDev = ""