Fix streaming deadlock.

Signed-off-by: Lantao Liu <lantaol@google.com>

pkg/ioutil/read_closer.go

@@ -16,39 +16,42 @@ limitations under the License.
 
 package ioutil
 
-import (
+import "io"
-	"io"
-	"sync"
-)
 
 // writeCloseInformer wraps a reader with a close function.
 type wrapReadCloser struct {
-	// TODO(random-liu): Evaluate whether the lock introduces
+	reader *io.PipeReader
-	// performance regression.
+	writer *io.PipeWriter
-	sync.RWMutex
-	r      io.Reader
-	closed bool
 }
 
 // NewWrapReadCloser creates a wrapReadCloser from a reader.
+// NOTE(random-liu): To avoid goroutine leakage, the reader passed in
+// must be eventually closed by the caller.
 func NewWrapReadCloser(r io.Reader) io.ReadCloser {
-	return &wrapReadCloser{r: r}
+	pr, pw := io.Pipe()
+	go func() {
+		_, _ = io.Copy(pw, r)
+		pr.Close()
+		pw.Close()
+	}()
+	return &wrapReadCloser{
+		reader: pr,
+		writer: pw,
+	}
 }
 
 // Read reads up to len(p) bytes into p.
 func (w *wrapReadCloser) Read(p []byte) (int, error) {
-	w.RLock()
+	n, err := w.reader.Read(p)
-	defer w.RUnlock()
+	if err == io.ErrClosedPipe {
-	if w.closed {
+		return n, io.EOF
-		return 0, io.EOF
 	}
-	return w.r.Read(p)
+	return n, err
 }
 
 // Close closes read closer.
 func (w *wrapReadCloser) Close() error {
-	w.Lock()
+	w.reader.Close()
-	defer w.Unlock()
+	w.writer.Close()
-	w.closed = true
 	return nil
 }