Merge pull request #14193 from stevvooe/concurrent-pull-fix

Attempt to protect on disk image store with mutex

graph/graph.go

@@ -31,9 +31,10 @@ import (
 
 // A Graph is a store for versioned filesystem images and the relationship between them.
 type Graph struct {
-	root    string
-	idIndex *truncindex.TruncIndex
-	driver  graphdriver.Driver
+	root       string
+	idIndex    *truncindex.TruncIndex
+	driver     graphdriver.Driver
+	imageMutex imageMutex // protect images in driver.
 }
 
 var (
@@ -153,6 +154,15 @@ func (graph *Graph) Create(layerData archive.ArchiveReader, containerID, contain
 
 // Register imports a pre-existing image into the graph.
 func (graph *Graph) Register(img *image.Image, layerData archive.ArchiveReader) (err error) {
+	if err := image.ValidateID(img.ID); err != nil {
+		return err
+	}
+
+	// We need this entire operation to be atomic within the engine. Note that
+	// this doesn't mean Register is fully safe yet.
+	graph.imageMutex.Lock(img.ID)
+	defer graph.imageMutex.Unlock(img.ID)
+
 	defer func() {
 		// If any error occurs, remove the new dir from the driver.
 		// Don't check for errors since the dir might not have been created.
@@ -161,9 +171,7 @@ func (graph *Graph) Register(img *image.Image, layerData archive.ArchiveReader)
 			graph.driver.Remove(img.ID)
 		}
 	}()
-	if err := image.ValidateID(img.ID); err != nil {
-		return err
-	}
+
 	// (This is a convenience to save time. Race conditions are taken care of by os.Rename)
 	if graph.Exists(img.ID) {
 		return fmt.Errorf("Image %s already exists", img.ID)

graph/mutex.go

@@ -0,0 +1,45 @@
+package graph
+
+import "sync"
+
+// imageMutex provides a lock per image id to protect shared resources in the
+// graph. This is only used with registration but should be used when
+// manipulating the layer store.
+type imageMutex struct {
+	mus map[string]*sync.Mutex // mutexes by image id.
+	mu  sync.Mutex             // protects lock map
+
+	// NOTE(stevvooe): The map above will grow to the size of all images ever
+	// registered during a daemon run. To free these resources, we must
+	// deallocate after unlock. Doing this safely is non-trivial in the face
+	// of a very minor leak.
+}
+
+// Lock the provided id.
+func (im *imageMutex) Lock(id string) {
+	im.getImageLock(id).Lock()
+}
+
+// Unlock the provided id.
+func (im *imageMutex) Unlock(id string) {
+	im.getImageLock(id).Unlock()
+}
+
+// getImageLock returns the mutex for the given id. This method will never
+// return nil.
+func (im *imageMutex) getImageLock(id string) *sync.Mutex {
+	im.mu.Lock()
+	defer im.mu.Unlock()
+
+	if im.mus == nil { // lazy
+		im.mus = make(map[string]*sync.Mutex)
+	}
+
+	mu, ok := im.mus[id]
+	if !ok {
+		mu = new(sync.Mutex)
+		im.mus[id] = mu
+	}
+
+	return mu
+}