contentutil: add ref lock on push

Signed-off-by: Tonis Tiigi <tonistiigi@gmail.com>

util/contentutil/refs.go

@@ -3,11 +3,16 @@ package contentutil
 import (
 	"context"
 	"net/http"
+	"sync"
 
 	"github.com/containerd/containerd/content"
+	"github.com/containerd/containerd/errdefs"
 	"github.com/containerd/containerd/remotes"
 	"github.com/containerd/containerd/remotes/docker"
+	"github.com/docker/docker/pkg/locker"
+	digest "github.com/opencontainers/go-digest"
 	ocispec "github.com/opencontainers/image-spec/specs-go/v1"
+	"github.com/pkg/errors"
 )
 
 func ProviderFromRef(ref string) (ocispec.Descriptor, content.Provider, error) {
@@ -38,11 +43,13 @@ func IngesterFromRef(ref string) (content.Ingester, error) {
 	}
 
 	return &ingester{
+		locker: locker.New(),
 		pusher: pusher,
 	}, nil
 }
 
 type ingester struct {
+	locker *locker.Locker
 	pusher remotes.Pusher
 }
 
@@ -53,5 +60,38 @@ func (w *ingester) Writer(ctx context.Context, opts ...content.WriterOpt) (conte
 			return nil, err
 		}
 	}
-	return w.pusher.Push(ctx, wo.Desc)
+	if wo.Ref == "" {
+		return nil, errors.Wrap(errdefs.ErrInvalidArgument, "ref must not be empty")
+	}
+	w.locker.Lock(wo.Ref)
+	var once sync.Once
+	unlock := func() {
+		once.Do(func() {
+			w.locker.Unlock(wo.Ref)
+		})
+	}
+	writer, err := w.pusher.Push(ctx, wo.Desc)
+	if err != nil {
+		return nil, err
+	}
+	return &lockedWriter{unlock: unlock, Writer: writer}, nil
+}
+
+type lockedWriter struct {
+	unlock func()
+	content.Writer
+}
+
+func (w *lockedWriter) Commit(ctx context.Context, size int64, expected digest.Digest, opts ...content.Opt) error {
+	err := w.Writer.Commit(ctx, size, expected, opts...)
+	if err == nil {
+		w.unlock()
+	}
+	return err
+}
+
+func (w *lockedWriter) Close() error {
+	err := w.Writer.Close()
+	w.unlock()
+	return err
 }