BACKPORT: Resolve race conditions in attach API call

Upstream reference: moby#30446

Signed-off-by: Jim Minter <jminter@redhat.com>

Move attach code to stream package

This cleans up attach a little bit, and moves it out of the container
package.
Really `AttachStream` is a method on `*stream.Config`, so moved if from
a package level function to one bound to `Config`.
In addition, uses a config struct rather than passing around tons and
tons of arguments.

Signed-off-by: Brian Goff <cpuguy83@gmail.com>

Refactor attaches `copyEscapable`

`copyEscapable` is a copy/paste of io.Copy with some added handling for
checking for the attach escape sequence.

This removes the copy/paste and uses `io.Copy` directly. To be able to
do this, it now implements an `io.Reader` which proxies to the main
reader but looks for the escape sequence.

Signed-off-by: Brian Goff <cpuguy83@gmail.com>
Signed-off-by: Antonio Murdaca <runcom@redhat.com>

Author: cpuguy83

container/container.go

@@ -30,7 +30,6 @@ import (
 	"github.com/docker/docker/libcontainerd"
 	"github.com/docker/docker/pkg/idtools"
 	"github.com/docker/docker/pkg/ioutils"
-	"github.com/docker/docker/pkg/promise"
 	"github.com/docker/docker/pkg/signal"
 	"github.com/docker/docker/pkg/symlink"
 	"github.com/docker/docker/restartmanager"
@@ -58,13 +57,6 @@ var (
 	errInvalidNetwork  = fmt.Errorf("invalid network settings while building port map info")
 )
 
-// DetachError is special error which returned in case of container detach.
-type DetachError struct{}
-
-func (DetachError) Error() string {
-	return "detached from container"
-}
-
 // CommonContainer holds the fields for a container which are
 // applicable across all platforms supported by the daemon.
 type CommonContainer struct {
@@ -366,185 +358,6 @@ func (container *Container) GetExecIDs() []string {
 	return container.ExecCommands.List()
 }
 
-// Attach connects to the container's TTY, delegating to standard
-// streams or websockets depending on the configuration.
-func (container *Container) Attach(stdin io.ReadCloser, stdout io.Writer, stderr io.Writer, keys []byte) chan error {
-	ctx := container.InitAttachContext()
-	return AttachStreams(ctx, container.StreamConfig, container.Config.OpenStdin, container.Config.StdinOnce, container.Config.Tty, stdin, stdout, stderr, keys)
-}
-
-// AttachStreams connects streams to a TTY.
-// Used by exec too. Should this move somewhere else?
-func AttachStreams(ctx context.Context, streamConfig *stream.Config, openStdin, stdinOnce, tty bool, stdin io.ReadCloser, stdout io.Writer, stderr io.Writer, keys []byte) chan error {
-	var (
-		cStdout, cStderr io.ReadCloser
-		cStdin           io.WriteCloser
-		wg               sync.WaitGroup
-		errors           = make(chan error, 3)
-	)
-
-	if stdin != nil && openStdin {
-		cStdin = streamConfig.StdinPipe()
-		wg.Add(1)
-	}
-
-	if stdout != nil {
-		cStdout = streamConfig.StdoutPipe()
-		wg.Add(1)
-	}
-
-	if stderr != nil {
-		cStderr = streamConfig.StderrPipe()
-		wg.Add(1)
-	}
-
-	// Connect stdin of container to the http conn.
-	go func() {
-		if stdin == nil || !openStdin {
-			return
-		}
-		logrus.Debug("attach: stdin: begin")
-
-		var err error
-		if tty {
-			_, err = copyEscapable(cStdin, stdin, keys)
-		} else {
-			_, err = io.Copy(cStdin, stdin)
-		}
-		if err == io.ErrClosedPipe {
-			err = nil
-		}
-		if err != nil {
-			logrus.Errorf("attach: stdin: %s", err)
-			errors <- err
-		}
-		if stdinOnce && !tty {
-			cStdin.Close()
-		} else {
-			// No matter what, when stdin is closed (io.Copy unblock), close stdout and stderr
-			if cStdout != nil {
-				cStdout.Close()
-			}
-			if cStderr != nil {
-				cStderr.Close()
-			}
-		}
-		logrus.Debug("attach: stdin: end")
-		wg.Done()
-	}()
-
-	attachStream := func(name string, stream io.Writer, streamPipe io.ReadCloser) {
-		if stream == nil {
-			return
-		}
-
-		logrus.Debugf("attach: %s: begin", name)
-		_, err := io.Copy(stream, streamPipe)
-		if err == io.ErrClosedPipe {
-			err = nil
-		}
-		if err != nil {
-			logrus.Errorf("attach: %s: %v", name, err)
-			errors <- err
-		}
-		// Make sure stdin gets closed
-		if stdin != nil {
-			stdin.Close()
-		}
-		streamPipe.Close()
-		logrus.Debugf("attach: %s: end", name)
-		wg.Done()
-	}
-
-	go attachStream("stdout", stdout, cStdout)
-	go attachStream("stderr", stderr, cStderr)
-
-	return promise.Go(func() error {
-		done := make(chan struct{})
-		go func() {
-			wg.Wait()
-			close(done)
-		}()
-		select {
-		case <-done:
-		case <-ctx.Done():
-			// close all pipes
-			if cStdin != nil {
-				cStdin.Close()
-			}
-			if cStdout != nil {
-				cStdout.Close()
-			}
-			if cStderr != nil {
-				cStderr.Close()
-			}
-			<-done
-		}
-		close(errors)
-		for err := range errors {
-			if err != nil {
-				return err
-			}
-		}
-		return nil
-	})
-}
-
-// Code c/c from io.Copy() modified to handle escape sequence
-func copyEscapable(dst io.Writer, src io.ReadCloser, keys []byte) (written int64, err error) {
-	if len(keys) == 0 {
-		// Default keys : ctrl-p ctrl-q
-		keys = []byte{16, 17}
-	}
-	buf := make([]byte, 32*1024)
-	for {
-		nr, er := src.Read(buf)
-		if nr > 0 {
-			// ---- Docker addition
-			preservBuf := []byte{}
-			for i, key := range keys {
-				preservBuf = append(preservBuf, buf[0:nr]...)
-				if nr != 1 || buf[0] != key {
-					break
-				}
-				if i == len(keys)-1 {
-					src.Close()
-					return 0, DetachError{}
-				}
-				nr, er = src.Read(buf)
-			}
-			var nw int
-			var ew error
-			if len(preservBuf) > 0 {
-				nw, ew = dst.Write(preservBuf)
-				nr = len(preservBuf)
-			} else {
-				// ---- End of docker
-				nw, ew = dst.Write(buf[0:nr])
-			}
-			if nw > 0 {
-				written += int64(nw)
-			}
-			if ew != nil {
-				err = ew
-				break
-			}
-			if nr != nw {
-				err = io.ErrShortWrite
-				break
-			}
-		}
-		if er == io.EOF {
-			break
-		}
-		if er != nil {
-			err = er
-			break
-		}
-	}
-	return written, err
-}
-
 // ShouldRestart decides whether the daemon should restart the container or not.
 // This is based on the container's restart policy.
 func (container *Container) ShouldRestart() bool {