1
0
Fork 0
mirror of https://github.com/moby/moby.git synced 2022-11-09 12:21:53 -05:00
moby--moby/graph/load.go
Aaron Lehmann 23e68679f0 Fix race condition when waiting for a concurrent layer pull
Before, this only waited for the download to complete. There was no
guarantee that the layer had been registered in the graph and was ready
use. This is especially problematic with v2 pulls, which wait for all
downloads before extracting layers.

Change Broadcaster to allow an error value to be propagated from Close
to the waiters.

Make the wait stop when the extraction is finished, rather than just the
download.

This also fixes v2 layer downloads to prefix the pool key with "layer:"
instead of "img:". "img:" is the wrong prefix, because this is what v1
uses for entire images. A v1 pull waiting for one of these operations to
finish would only wait for that particular layer, not all its
dependencies.

Signed-off-by: Aaron Lehmann <aaron.lehmann@docker.com>
2015-09-11 10:52:34 -07:00

132 lines
3.1 KiB
Go

// +build linux windows
package graph
import (
"encoding/json"
"io"
"io/ioutil"
"os"
"path/filepath"
"github.com/Sirupsen/logrus"
"github.com/docker/docker/image"
"github.com/docker/docker/pkg/archive"
"github.com/docker/docker/pkg/chrootarchive"
)
// Load uploads a set of images into the repository. This is the complementary of ImageExport.
// The input stream is an uncompressed tar ball containing images and metadata.
func (s *TagStore) Load(inTar io.ReadCloser, outStream io.Writer) error {
tmpImageDir, err := ioutil.TempDir("", "docker-import-")
if err != nil {
return err
}
defer os.RemoveAll(tmpImageDir)
var (
repoDir = filepath.Join(tmpImageDir, "repo")
)
if err := os.Mkdir(repoDir, os.ModeDir); err != nil {
return err
}
images := s.graph.Map()
excludes := make([]string, len(images))
i := 0
for k := range images {
excludes[i] = k
i++
}
if err := chrootarchive.Untar(inTar, repoDir, &archive.TarOptions{ExcludePatterns: excludes}); err != nil {
return err
}
dirs, err := ioutil.ReadDir(repoDir)
if err != nil {
return err
}
for _, d := range dirs {
if d.IsDir() {
if err := s.recursiveLoad(d.Name(), tmpImageDir); err != nil {
return err
}
}
}
reposJSONFile, err := os.Open(filepath.Join(tmpImageDir, "repo", "repositories"))
if err != nil {
if !os.IsNotExist(err) {
return err
}
return nil
}
defer reposJSONFile.Close()
repositories := map[string]Repository{}
if err := json.NewDecoder(reposJSONFile).Decode(&repositories); err != nil {
return err
}
for imageName, tagMap := range repositories {
for tag, address := range tagMap {
if err := s.setLoad(imageName, tag, address, true, outStream); err != nil {
return err
}
}
}
return nil
}
func (s *TagStore) recursiveLoad(address, tmpImageDir string) error {
if _, err := s.LookupImage(address); err != nil {
logrus.Debugf("Loading %s", address)
imageJSON, err := ioutil.ReadFile(filepath.Join(tmpImageDir, "repo", address, "json"))
if err != nil {
logrus.Debugf("Error reading json: %v", err)
return err
}
layer, err := os.Open(filepath.Join(tmpImageDir, "repo", address, "layer.tar"))
if err != nil {
logrus.Debugf("Error reading embedded tar: %v", err)
return err
}
img, err := image.NewImgJSON(imageJSON)
if err != nil {
logrus.Debugf("Error unmarshalling json: %v", err)
return err
}
if err := image.ValidateID(img.ID); err != nil {
logrus.Debugf("Error validating ID: %v", err)
return err
}
// ensure no two downloads of the same layer happen at the same time
poolKey := "layer:" + img.ID
broadcaster, found := s.poolAdd("pull", poolKey)
if found {
logrus.Debugf("Image (id: %s) load is already running, waiting", img.ID)
return broadcaster.Wait()
}
defer s.poolRemove("pull", poolKey)
if img.Parent != "" {
if !s.graph.Exists(img.Parent) {
if err := s.recursiveLoad(img.Parent, tmpImageDir); err != nil {
return err
}
}
}
if err := s.graph.Register(img, layer); err != nil {
return err
}
}
logrus.Debugf("Completed processing %s", address)
return nil
}