fix pusher concurrent close channel

Signed-off-by: rongfu.leng <rongfu.leng@daocloud.io>
Signed-off-by: rongfu.leng <1275177125@qq.com>

Makefile

@@ -403,9 +403,9 @@ install-deps:
 	script/setup/install-cni-windows
 else
 install-deps: ## install cri dependencies
-	script/setup/install-seccomp
-	script/setup/install-runc
-	script/setup/install-critools
+	#script/setup/install-seccomp
+	#script/setup/install-runc
+	#script/setup/install-critools
 	script/setup/install-cni
 endif
 

remotes/docker/pusher.go

@@ -24,6 +24,7 @@ import (
 	"net/http"
 	"net/url"
 	"strings"
+	"sync"
 	"time"
 
 	"github.com/containerd/containerd/content"
@@ -322,6 +323,7 @@ type pushWriter struct {
 
 	pipeC chan *io.PipeWriter
 	respC chan *http.Response
+	lock  sync.Mutex
 	errC  chan error
 
 	isManifest bool
@@ -398,6 +400,7 @@ 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
+	pw.lock.Lock()
 	select {
 	case _, ok := <-pw.pipeC:
 		if ok {
@@ -406,6 +409,7 @@ func (pw *pushWriter) Close() error {
 	default:
 		close(pw.pipeC)
 	}
+	pw.lock.Unlock()
 	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))
 
 		})
 	}