From d61190169de27bc58f1fe4a8f49b37bd2294d489 Mon Sep 17 00:00:00 2001
From: Solomon Hykes <solomon@docker.com>
Date: Sun, 4 May 2014 03:29:55 +0000
Subject: [PATCH] Engine: ensure all pipes are properly closed by Receiver and
 Sender

Docker-DCO-1.1-Signed-off-by: Solomon Hykes <solomon@docker.com> (github: shykes)

[michael@docker.com: fix stdin closing in engine.Job.Run]
[michael@docker.com: fix fd leak in engine.Receiver.Run]
Docker-DCO-1.1-Signed-off-by: Michael Crosby <michael@docker.com> (github: crosbymichael)

Docker-Tested-By: Solomon Hykes <solomon@docker.com>
Docker-Tested-by: Michael Crosby <michael@docker.com>
---
 engine/job.go         |  3 +++
 engine/remote.go      | 21 ++++++++++++++-------
 engine/remote_test.go | 28 ++++++++++++++++++++++++++++
 3 files changed, 45 insertions(+), 7 deletions(-)

diff --git a/engine/job.go b/engine/job.go
index b56155ac1c..7e655e1b12 100644
--- a/engine/job.go
+++ b/engine/job.go
@@ -72,6 +72,9 @@ func (job *Job) Run() error {
 	if err := job.Stderr.Close(); err != nil {
 		return err
 	}
+	if err := job.Stdin.Close(); err != nil {
+		return err
+	}
 	if job.status != 0 {
 		return fmt.Errorf("%s", errorMessage)
 	}
diff --git a/engine/remote.go b/engine/remote.go
index fbb9951065..1da521a3c7 100644
--- a/engine/remote.go
+++ b/engine/remote.go
@@ -36,20 +36,27 @@ func (s *Sender) Handle(job *Job) Status {
 	r := beam.NewRouter(nil)
 	r.NewRoute().KeyStartsWith("cmd", "log", "stdout").HasAttachment().Handler(func(p []byte, stdout *os.File) error {
 		tasks.Add(1)
-		io.Copy(job.Stdout, stdout)
-		tasks.Done()
+		go func() {
+			io.Copy(job.Stdout, stdout)
+			stdout.Close()
+			tasks.Done()
+		}()
 		return nil
 	})
 	r.NewRoute().KeyStartsWith("cmd", "log", "stderr").HasAttachment().Handler(func(p []byte, stderr *os.File) error {
 		tasks.Add(1)
-		io.Copy(job.Stderr, stderr)
-		tasks.Done()
+		go func() {
+			io.Copy(job.Stderr, stderr)
+			stderr.Close()
+			tasks.Done()
+		}()
 		return nil
 	})
 	r.NewRoute().KeyStartsWith("cmd", "log", "stdin").HasAttachment().Handler(func(p []byte, stdin *os.File) error {
-		tasks.Add(1)
-		io.Copy(stdin, job.Stdin)
-		tasks.Done()
+		go func() {
+			io.Copy(stdin, job.Stdin)
+			stdin.Close()
+		}()
 		return nil
 	})
 	var status int
diff --git a/engine/remote_test.go b/engine/remote_test.go
index 921a4376ce..a23830af01 100644
--- a/engine/remote_test.go
+++ b/engine/remote_test.go
@@ -5,6 +5,7 @@ import (
 	"bytes"
 	"fmt"
 	"github.com/dotcloud/docker/pkg/beam"
+	"io"
 	"strings"
 	"testing"
 	"time"
@@ -54,6 +55,33 @@ func TestHelloWorld(t *testing.T) {
 	}
 }
 
+func TestStdin(t *testing.T) {
+	testRemote(t,
+
+		func(eng *Engine) {
+			job := eng.Job("mirror")
+			job.Stdin.Add(strings.NewReader("hello world!\n"))
+			out := &bytes.Buffer{}
+			job.Stdout.Add(out)
+			if err := job.Run(); err != nil {
+				t.Fatal(err)
+			}
+			if out.String() != "hello world!\n" {
+				t.Fatalf("%#v", out.String())
+			}
+		},
+
+		func(eng *Engine) {
+			eng.Register("mirror", func(job *Job) Status {
+				if _, err := io.Copy(job.Stdout, job.Stdin); err != nil {
+					t.Fatal(err)
+				}
+				return StatusOK
+			})
+		},
+	)
+}
+
 // Helpers
 
 func testRemote(t *testing.T, senderSide, receiverSide func(*Engine)) {