Merge pull request #7562 from dcantah/cp-concurrent-closech

[release/1.6] fix pusher concurrent close channel

remotes/docker/pusher.go

@@ -24,6 +24,7 @@ import (
 	"net/http"
 	"net/url"
 	"strings"
+	"sync"
 	"time"
 
 	"github.com/containerd/containerd/content"
@@ -317,9 +318,10 @@ type pushWriter struct {
 
 	pipe *io.PipeWriter
 
-	pipeC chan *io.PipeWriter
-	respC chan *http.Response
-	errC  chan error
+	pipeC     chan *io.PipeWriter
+	respC     chan *http.Response
+	closeOnce sync.Once
+	errC      chan error
 
 	isManifest bool
 
@@ -395,14 +397,9 @@ func (pw *pushWriter) Write(p []byte) (n int, err error) {
 func (pw *pushWriter) Close() error {
 	// Ensure pipeC is closed but handle `Close()` being
 	// called multiple times without panicking
-	select {
-	case _, ok := <-pw.pipeC:
-		if ok {
-			close(pw.pipeC)
-		}
-	default:
+	pw.closeOnce.Do(func() {
 		close(pw.pipeC)
-	}
+	})
 	if pw.pipe != nil {
 		status, err := pw.tracker.GetStatus(pw.ref)
 		if err == nil && !status.Committed {

remotes/docker/pusher_test.go

@@ -293,7 +293,7 @@ func Test_dockerPusher_push(t *testing.T) {
 		dp               dockerPusher
 		dockerBaseObject string
 		args             args
-		checkerFunc      func(writer pushWriter) bool
+		checkerFunc      func(writer *pushWriter) bool
 		wantErr          error
 	}{
 		{
@@ -306,7 +306,7 @@ func Test_dockerPusher_push(t *testing.T) {
 				ref:               fmt.Sprintf("manifest-%s", manifestContentDigest.String()),
 				unavailableOnFail: false,
 			},
-			checkerFunc: func(writer pushWriter) bool {
+			checkerFunc: func(writer *pushWriter) bool {
 				select {
 				case resp := <-writer.respC:
 					// 201 should be the response code when uploading a new manifest
@@ -340,7 +340,7 @@ func Test_dockerPusher_push(t *testing.T) {
 				ref:               fmt.Sprintf("layer-%s", layerContentDigest.String()),
 				unavailableOnFail: false,
 			},
-			checkerFunc: func(writer pushWriter) bool {
+			checkerFunc: func(writer *pushWriter) bool {
 				select {
 				case resp := <-writer.respC:
 					// 201 should be the response code when uploading a new blob
@@ -379,7 +379,7 @@ func Test_dockerPusher_push(t *testing.T) {
 			}
 
 			// test whether a proper response has been received after the push operation
-			assert.True(t, test.checkerFunc(*pw))
+			assert.True(t, test.checkerFunc(pw))
 
 		})
 	}