Reduce the number of IO constructors

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

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
@@ -49,6 +49,7 @@ type FIFOSet struct {
 	close func() error
 }
 
+// Close the FIFOSet
 func (f *FIFOSet) Close() error {
 	if f.close != nil {
 		return f.close()
@@ -61,67 +62,72 @@ func NewFIFOSet(config Config, close func() error) *FIFOSet {
 	return &FIFOSet{Config: config, close: close}
 }
 
-// NewIO returns an Creator that will provide IO sets without a terminal
+// Streams used to configure a Creator or Attach
-func NewIO(stdin io.Reader, stdout, stderr io.Writer) Creator {
+type Streams struct {
-	return NewIOWithTerminal(stdin, stdout, stderr, false)
+	Stdin    io.Reader
+	Stdout   io.Writer
+	Stderr   io.Writer
+	Terminal bool
 }
 
-// NewIOWithTerminal creates a new io set with the provided io.Reader/Writers for use with a terminal
+// Opt customize options for creating a Creator or Attach
-func NewIOWithTerminal(stdin io.Reader, stdout, stderr io.Writer, terminal bool) Creator {
+type Opt func(*Streams)
-	return func(id string) (IO, error) {
-		fifos, err := newFIFOSetInTempDir(id)
-		if err != nil {
-			return nil, err
-		}
 
-		fifos.Terminal = terminal
+// WithStdio sets stream options to the standard input/output streams
-		set := &ioSet{in: stdin, out: stdout, err: stderr}
+func WithStdio(opt *Streams) {
-		return copyIO(fifos, set)
+	WithStreams(os.Stdin, os.Stdout, os.Stderr)(opt)
+}
+
+// WithTerminal sets the terminal option
+func WithTerminal(opt *Streams) {
+	opt.Terminal = true
+}
+
+// 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
+// NewCreator returns an IO creator from the options
-func WithAttach(stdin io.Reader, stdout, stderr io.Writer) Attach {
+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
+		}
+		return copyIO(fifos, streams)
+	}
+}
+
+// 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")
 		}
-		set := &ioSet{in: stdin, out: stdout, err: stderr}
+		return copyIO(fifos, streams)
-		return copyIO(fifos, set)
 	}
 }
 
-// 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)
-}
-
 // NullIO redirects the container's IO into /dev/null
 func NullIO(_ string) (IO, error) {
 	return &cio{}, nil
 }
 
-type ioSet struct {
-	in       io.Reader
-	out, err io.Writer
-}
-
-type pipes struct {
-	Stdin  io.WriteCloser
-	Stdout io.ReadCloser
-	Stderr io.ReadCloser
-}
-
-func (p *pipes) closers() []io.Closer {
-	return []io.Closer{p.Stdin, p.Stdout, p.Stderr}
-}
-
 // cio is a basic container IO implementation.
 type cio struct {
 	config  Config
@@ -158,3 +164,17 @@ func (c *cio) Cancel() {
 		c.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{}

cio/io_unix.go

@@ -15,11 +15,12 @@ import (
 	"github.com/pkg/errors"
 )
 
-// newFIFOSetInTempDir returns a new set of fifos for the task
+// NewFIFOSetInDir returns a new FIFOSet with paths in a temporary directory under root
-func newFIFOSetInTempDir(id string) (*FIFOSet, error) {
+func NewFIFOSetInDir(root, id string, terminal bool) (*FIFOSet, error) {
-	root := "/run/containerd/fifo"
+	if root != "" {
-	if err := os.MkdirAll(root, 0700); err != nil {
+		if err := os.MkdirAll(root, 0700); err != nil {
-		return nil, err
+			return nil, err
+		}
 	}
 	dir, err := ioutil.TempDir(root, "")
 	if err != nil {
@@ -29,18 +30,15 @@ func newFIFOSetInTempDir(id string) (*FIFOSet, error) {
 		return os.RemoveAll(dir)
 	}
 	return NewFIFOSet(Config{
-		Stdin:  filepath.Join(dir, id+"-stdin"),
+		Stdin:    filepath.Join(dir, id+"-stdin"),
-		Stdout: filepath.Join(dir, id+"-stdout"),
+		Stdout:   filepath.Join(dir, id+"-stdout"),
-		Stderr: filepath.Join(dir, id+"-stderr"),
+		Stderr:   filepath.Join(dir, id+"-stderr"),
+		Terminal: terminal,
 	}, closer), nil
 }
 
-func copyIO(fifos *FIFOSet, ioset *ioSet) (*cio, error) {
+func copyIO(fifos *FIFOSet, ioset *Streams) (*cio, error) {
-	var (
+	var ctx, cancel = context.WithCancel(context.Background())
-		ctx, cancel = context.WithCancel(context.Background())
-		wg          = &sync.WaitGroup{}
-	)
-
 	pipes, err := openFifos(ctx, fifos)
 	if err != nil {
 		cancel()
@@ -49,14 +47,15 @@ func copyIO(fifos *FIFOSet, ioset *ioSet) (*cio, error) {
 
 	if fifos.Stdin != "" {
 		go func() {
-			io.Copy(pipes.Stdin, ioset.in)
+			io.Copy(pipes.Stdin, ioset.Stdin)
 			pipes.Stdin.Close()
 		}()
 	}
 
+	var wg = &sync.WaitGroup{}
 	wg.Add(1)
 	go func() {
-		io.Copy(ioset.out, pipes.Stdout)
+		io.Copy(ioset.Stdout, pipes.Stdout)
 		pipes.Stdout.Close()
 		wg.Done()
 	}()
@@ -64,12 +63,13 @@ func copyIO(fifos *FIFOSet, ioset *ioSet) (*cio, error) {
 	if !fifos.Terminal {
 		wg.Add(1)
 		go func() {
-			io.Copy(ioset.err, pipes.Stderr)
+			io.Copy(ioset.Stderr, pipes.Stderr)
 			pipes.Stderr.Close()
 			wg.Done()
 		}()
 	}
 	return &cio{
+		config:  fifos.Config,
 		wg:      wg,
 		closers: append(pipes.closers(), fifos),
 		cancel:  cancel,
@@ -78,41 +78,38 @@ func copyIO(fifos *FIFOSet, ioset *ioSet) (*cio, error) {
 
 func openFifos(ctx context.Context, fifos *FIFOSet) (pipes, error) {
 	var err error
-	f := new(pipes)
-
 	defer func() {
 		if err != nil {
 			fifos.Close()
 		}
 	}()
 
+	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 pipes{}, errors.Wrapf(err, "failed to open stdin fifo")
+			return f, errors.Wrapf(err, "failed to open stdin fifo")
 		}
 	}
-	if f.Stdout, err = fifo.OpenFifo(ctx, fifos.Stdout, syscall.O_RDONLY|syscall.O_CREAT|syscall.O_NONBLOCK, 0700); err != nil {
+	if fifos.Stdout != "" {
-		f.Stdin.Close()
+		if f.Stdout, err = fifo.OpenFifo(ctx, fifos.Stdout, syscall.O_RDONLY|syscall.O_CREAT|syscall.O_NONBLOCK, 0700); err != nil {
-		return pipes{}, errors.Wrapf(err, "failed to open stdout fifo")
+			f.Stdin.Close()
+			return f, errors.Wrapf(err, "failed to open stdout fifo")
+		}
 	}
-	if f.Stderr, err = fifo.OpenFifo(ctx, fifos.Stderr, syscall.O_RDONLY|syscall.O_CREAT|syscall.O_NONBLOCK, 0700); err != nil {
+	if fifos.Stderr != "" {
-		f.Stdin.Close()
+		if f.Stderr, err = fifo.OpenFifo(ctx, fifos.Stderr, syscall.O_RDONLY|syscall.O_CREAT|syscall.O_NONBLOCK, 0700); err != nil {
-		f.Stdout.Close()
+			f.Stdin.Close()
-		return pipes{}, errors.Wrapf(err, "failed to open stderr fifo")
+			f.Stdout.Close()
+			return f, errors.Wrapf(err, "failed to open stderr fifo")
+		}
 	}
-	return pipes{}, nil
+	return f, nil
 }
 
 // NewDirectIO returns an IO implementation that exposes the IO streams as io.ReadCloser
-// and io.WriteCloser. FIFOs are created in /run/containerd/fifo.
+// and io.WriteCloser.
-func NewDirectIO(ctx context.Context, terminal bool) (*DirectIO, error) {
+func NewDirectIO(ctx context.Context, fifos *FIFOSet) (*DirectIO, error) {
-	fifos, err := newFIFOSetInTempDir("")
+	ctx, cancel := context.WithCancel(ctx)
-	if err != nil {
-		return nil, err
-	}
-	fifos.Terminal = terminal
-
-	ctx, cancel := context.WithCancel(context.Background())
 	pipes, err := openFifos(ctx, fifos)
 	return &DirectIO{
 		pipes: pipes,
@@ -124,10 +121,6 @@ func NewDirectIO(ctx context.Context, terminal bool) (*DirectIO, error) {
 	}, err
 }
 
-// DirectIO allows task IO to be handled externally by the caller
+func (p *pipes) closers() []io.Closer {
-type DirectIO struct {
+	return []io.Closer{p.Stdin, p.Stdout, p.Stderr}
-	pipes
-	cio
 }
-
-var _ IO = &DirectIO{}

cio/io_windows.go

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

cmd/ctr/commands/tasks/attach.go

@@ -1,8 +1,6 @@
 package tasks
 
 import (
-	"os"
-
 	"github.com/containerd/console"
 	"github.com/containerd/containerd/cio"
 	"github.com/containerd/containerd/cmd/ctr/commands"
@@ -39,7 +37,7 @@ var attachCommand = cli.Command{
 				return err
 			}
 		}
-		task, err := container.Task(ctx, cio.WithAttach(os.Stdin, os.Stdout, os.Stderr))
+		task, err := container.Task(ctx, cio.NewAttach(cio.WithStdio))
 		if err != nil {
 			return err
 		}

cmd/ctr/commands/tasks/exec.go

@@ -60,9 +60,9 @@ var execCommand = cli.Command{
 		pspec.Terminal = tty
 		pspec.Args = args
 
-		ioCreator := cio.Stdio
+		ioCreator := cio.NewCreator(cio.WithStdio)
 		if tty {
-			ioCreator = cio.StdioTerminal
+			ioCreator = cio.NewCreator(cio.WithStdio, cio.WithTerminal)
 		}
 		process, err := task.Exec(ctx, context.String("exec-id"), pspec, ioCreator)
 		if err != nil {

cmd/ctr/commands/tasks/tasks_unix.go

@@ -44,10 +44,11 @@ func HandleConsoleResize(ctx gocontext.Context, task resizer, con console.Consol
 
 // NewTask creates a new task
 func NewTask(ctx gocontext.Context, client *containerd.Client, container containerd.Container, checkpoint string, tty, nullIO bool) (containerd.Task, error) {
+	stdio := cio.NewCreator(cio.WithStdio)
 	if checkpoint == "" {
-		ioCreator := cio.Stdio
+		ioCreator := stdio
 		if tty {
-			ioCreator = cio.StdioTerminal
+			ioCreator = cio.NewCreator(cio.WithStdio, cio.WithTerminal)
 		}
 		if nullIO {
 			if tty {
@@ -61,5 +62,5 @@ func NewTask(ctx gocontext.Context, client *containerd.Client, container contain
 	if err != nil {
 		return nil, err
 	}
-	return container.NewTask(ctx, cio.Stdio, containerd.WithTaskCheckpoint(im))
+	return container.NewTask(ctx, stdio, containerd.WithTaskCheckpoint(im))
 }

cmd/ctr/commands/tasks/tasks_windows.go

@@ -42,9 +42,9 @@ func HandleConsoleResize(ctx gocontext.Context, task resizer, con console.Consol
 
 // NewTask creates a new task
 func NewTask(ctx gocontext.Context, client *containerd.Client, container containerd.Container, _ string, tty, nullIO bool) (containerd.Task, error) {
-	ioCreator := cio.Stdio
+	ioCreator := cio.NewCreator(cio.WithStdio)
 	if tty {
-		ioCreator = cio.StdioTerminal
+		ioCreator = cio.NewCreator(cio.WithStdio, cio.WithTerminal)
 	}
 	if nullIO {
 		if tty {

container_linux_test.go

@@ -373,7 +373,11 @@ func TestContainerAttach(t *testing.T) {
 }
 
 func newDirectIO(ctx context.Context) (*directIO, error) {
-	dio, err := cio.NewDirectIO(ctx, false)
+	fifos, err := cio.NewFIFOSetInDir("", "", false)
+	if err != nil {
+		return nil, err
+	}
+	dio, err := cio.NewDirectIO(ctx, fifos)
 	if err != nil {
 		return nil, err
 	}
@@ -735,7 +739,7 @@ func TestContainerKillAll(t *testing.T) {
 	defer container.Delete(ctx, WithSnapshotCleanup)
 
 	stdout := bytes.NewBuffer(nil)
-	task, err := container.NewTask(ctx, cio.NewIO(bytes.NewBuffer(nil), stdout, bytes.NewBuffer(nil)))
+	task, err := container.NewTask(ctx, cio.NewCreator(withByteBuffers(stdout)))
 	if err != nil {
 		t.Error(err)
 		return
@@ -987,7 +991,7 @@ func TestContainerKillInitPidHost(t *testing.T) {
 	defer container.Delete(ctx, WithSnapshotCleanup)
 
 	stdout := bytes.NewBuffer(nil)
-	task, err := container.NewTask(ctx, cio.NewIO(bytes.NewBuffer(nil), stdout, bytes.NewBuffer(nil)))
+	task, err := container.NewTask(ctx, cio.NewCreator(withByteBuffers(stdout)))
 	if err != nil {
 		t.Error(err)
 		return
@@ -1086,7 +1090,7 @@ func testUserNamespaces(t *testing.T, readonlyRootFS bool) {
 	}
 	defer container.Delete(ctx, WithSnapshotCleanup)
 
-	task, err := container.NewTask(ctx, cio.Stdio, func(_ context.Context, client *Client, r *TaskInfo) error {
+	task, err := container.NewTask(ctx, cio.NewCreator(cio.WithStdio), func(_ context.Context, client *Client, r *TaskInfo) error {
 		r.Options = &runctypes.CreateOptions{
 			IoUid: 1000,
 			IoGid: 1000,

container_test.go

@@ -3,6 +3,7 @@ package containerd
 import (
 	"bytes"
 	"context"
+	"io"
 	"io/ioutil"
 	"os"
 	"runtime"
@@ -27,7 +28,7 @@ func empty() cio.Creator {
 	// TODO (@mlaventure) windows searches for pipes
 	// when none are provided
 	if runtime.GOOS == "windows" {
-		return cio.Stdio
+		return cio.NewCreator(cio.WithStdio)
 	}
 	return cio.NullIO
 }
@@ -187,7 +188,7 @@ func TestContainerOutput(t *testing.T) {
 	defer container.Delete(ctx, WithSnapshotCleanup)
 
 	stdout := bytes.NewBuffer(nil)
-	task, err := container.NewTask(ctx, cio.NewIO(bytes.NewBuffer(nil), stdout, bytes.NewBuffer(nil)))
+	task, err := container.NewTask(ctx, cio.NewCreator(withByteBuffers(stdout)))
 	if err != nil {
 		t.Error(err)
 		return
@@ -223,6 +224,15 @@ func TestContainerOutput(t *testing.T) {
 	}
 }
 
+func withByteBuffers(stdout io.Writer) cio.Opt {
+	// TODO: could this use ioutil.Discard?
+	return func(streams *cio.Streams) {
+		streams.Stdin = new(bytes.Buffer)
+		streams.Stdout = stdout
+		streams.Stderr = new(bytes.Buffer)
+	}
+}
+
 func TestContainerExec(t *testing.T) {
 	t.Parallel()
 
@@ -534,7 +544,7 @@ func TestContainerCloseIO(t *testing.T) {
 		return
 	}
 
-	task, err := container.NewTask(ctx, cio.NewIO(r, stdout, ioutil.Discard))
+	task, err := container.NewTask(ctx, cio.NewCreator(cio.WithStreams(r, stdout, ioutil.Discard)))
 	if err != nil {
 		t.Error(err)
 		return
@@ -1145,7 +1155,7 @@ func TestContainerHostname(t *testing.T) {
 	defer container.Delete(ctx, WithSnapshotCleanup)
 
 	stdout := bytes.NewBuffer(nil)
-	task, err := container.NewTask(ctx, cio.NewIO(bytes.NewBuffer(nil), stdout, bytes.NewBuffer(nil)))
+	task, err := container.NewTask(ctx, cio.NewCreator(withByteBuffers(stdout)))
 	if err != nil {
 		t.Error(err)
 		return