avoid redundant blob fetching
Signed-off-by: baojiangnan <baojn1998@163.com>
This commit is contained in:
parent
02e2231e60
commit
abfc675939
2 changed files with 52 additions and 40 deletions
|
@ -76,23 +76,26 @@ func (pbs *proxyBlobStore) serveLocal(ctx context.Context, w http.ResponseWriter
|
||||||
return true, pbs.localStore.ServeBlob(ctx, w, r, dgst)
|
return true, pbs.localStore.ServeBlob(ctx, w, r, dgst)
|
||||||
}
|
}
|
||||||
|
|
||||||
func (pbs *proxyBlobStore) storeLocal(ctx context.Context, dgst digest.Digest) error {
|
func (pbs *proxyBlobStore) serveRemote(ctx context.Context, w http.ResponseWriter, dgst digest.Digest) error {
|
||||||
defer func() {
|
bw, err := pbs.localStore.Create(ctx)
|
||||||
mu.Lock()
|
|
||||||
delete(inflight, dgst)
|
|
||||||
mu.Unlock()
|
|
||||||
}()
|
|
||||||
|
|
||||||
var desc distribution.Descriptor
|
|
||||||
var err error
|
|
||||||
var bw distribution.BlobWriter
|
|
||||||
|
|
||||||
bw, err = pbs.localStore.Create(ctx)
|
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
|
|
||||||
desc, err = pbs.copyContent(ctx, dgst, bw)
|
desc, err := pbs.remoteStore.Stat(ctx, dgst)
|
||||||
|
if err != nil {
|
||||||
|
return err
|
||||||
|
}
|
||||||
|
|
||||||
|
remoteReader, err := pbs.remoteStore.Open(ctx, dgst)
|
||||||
|
if err != nil {
|
||||||
|
return err
|
||||||
|
}
|
||||||
|
defer remoteReader.Close()
|
||||||
|
|
||||||
|
// use TeeReader to avoid redundant blob fetching.
|
||||||
|
teeReader := io.TeeReader(remoteReader, bw)
|
||||||
|
_, err = io.CopyN(w, teeReader, desc.Size)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
|
@ -102,7 +105,14 @@ func (pbs *proxyBlobStore) storeLocal(ctx context.Context, dgst digest.Digest) e
|
||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
|
|
||||||
return nil
|
blobRef, err := reference.WithDigest(pbs.repositoryName, dgst)
|
||||||
|
if err != nil {
|
||||||
|
dcontext.GetLogger(ctx).Errorf("Error creating reference: %s", err)
|
||||||
|
return err
|
||||||
|
}
|
||||||
|
|
||||||
|
pbs.scheduler.AddBlob(blobRef, repositoryTTL)
|
||||||
|
return err
|
||||||
}
|
}
|
||||||
|
|
||||||
func (pbs *proxyBlobStore) ServeBlob(ctx context.Context, w http.ResponseWriter, r *http.Request, dgst digest.Digest) error {
|
func (pbs *proxyBlobStore) ServeBlob(ctx context.Context, w http.ResponseWriter, r *http.Request, dgst digest.Digest) error {
|
||||||
|
@ -130,31 +140,13 @@ func (pbs *proxyBlobStore) ServeBlob(ctx context.Context, w http.ResponseWriter,
|
||||||
inflight[dgst] = struct{}{}
|
inflight[dgst] = struct{}{}
|
||||||
mu.Unlock()
|
mu.Unlock()
|
||||||
|
|
||||||
// storeLocalCtx will be independent with ctx, because ctx it used to fetch remote image.
|
defer func() {
|
||||||
// There would be a situation, that is pulling remote bytes ends before pbs.storeLocal( 'Copy', 'Commit' ...)
|
mu.Lock()
|
||||||
// Then the registry fails to cache the layer, even though the layer had been served to client.
|
delete(inflight, dgst)
|
||||||
storeLocalCtx, cancel := context.WithCancel(context.Background())
|
mu.Unlock()
|
||||||
go func(dgst digest.Digest) {
|
}()
|
||||||
defer cancel()
|
|
||||||
if err := pbs.storeLocal(storeLocalCtx, dgst); err != nil {
|
|
||||||
dcontext.GetLogger(storeLocalCtx).Errorf("Error committing to storage: %s", err.Error())
|
|
||||||
}
|
|
||||||
|
|
||||||
blobRef, err := reference.WithDigest(pbs.repositoryName, dgst)
|
return pbs.serveRemote(ctx, w, dgst)
|
||||||
if err != nil {
|
|
||||||
dcontext.GetLogger(storeLocalCtx).Errorf("Error creating reference: %s", err)
|
|
||||||
return
|
|
||||||
}
|
|
||||||
|
|
||||||
pbs.scheduler.AddBlob(blobRef, repositoryTTL)
|
|
||||||
}(dgst)
|
|
||||||
|
|
||||||
_, err = pbs.copyContent(ctx, dgst, w)
|
|
||||||
if err != nil {
|
|
||||||
cancel()
|
|
||||||
return err
|
|
||||||
}
|
|
||||||
return nil
|
|
||||||
}
|
}
|
||||||
|
|
||||||
func (pbs *proxyBlobStore) Stat(ctx context.Context, dgst digest.Digest) (distribution.Descriptor, error) {
|
func (pbs *proxyBlobStore) Stat(ctx context.Context, dgst digest.Digest) (distribution.Descriptor, error) {
|
||||||
|
|
|
@ -329,6 +329,12 @@ func testProxyStoreServe(t *testing.T, te *testEnv, numClients int) {
|
||||||
remoteStats := te.RemoteStats()
|
remoteStats := te.RemoteStats()
|
||||||
|
|
||||||
var wg sync.WaitGroup
|
var wg sync.WaitGroup
|
||||||
|
var descHitMap = map[digest.Digest]bool{}
|
||||||
|
var hitLock sync.Mutex
|
||||||
|
|
||||||
|
for _, remoteBlob := range te.inRemote {
|
||||||
|
descHitMap[remoteBlob.Digest] = true
|
||||||
|
}
|
||||||
|
|
||||||
for i := 0; i < numClients; i++ {
|
for i := 0; i < numClients; i++ {
|
||||||
// Serveblob - pulls through blobs
|
// Serveblob - pulls through blobs
|
||||||
|
@ -355,6 +361,15 @@ func testProxyStoreServe(t *testing.T, te *testEnv, numClients int) {
|
||||||
t.Errorf("Mismatching blob fetch from proxy")
|
t.Errorf("Mismatching blob fetch from proxy")
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
|
|
||||||
|
desc, err := te.store.localStore.Stat(te.ctx, remoteBlob.Digest)
|
||||||
|
if err != nil {
|
||||||
|
continue
|
||||||
|
}
|
||||||
|
|
||||||
|
hitLock.Lock()
|
||||||
|
delete(descHitMap, desc.Digest)
|
||||||
|
hitLock.Unlock()
|
||||||
}
|
}
|
||||||
}()
|
}()
|
||||||
}
|
}
|
||||||
|
@ -364,11 +379,16 @@ func testProxyStoreServe(t *testing.T, te *testEnv, numClients int) {
|
||||||
t.FailNow()
|
t.FailNow()
|
||||||
}
|
}
|
||||||
|
|
||||||
|
if len(descHitMap) > 0 {
|
||||||
|
t.Errorf("Expected hit cache at least once, but it turns out that no caches was hit")
|
||||||
|
t.FailNow()
|
||||||
|
}
|
||||||
|
|
||||||
remoteBlobCount := len(te.inRemote)
|
remoteBlobCount := len(te.inRemote)
|
||||||
sbsMu.Lock()
|
sbsMu.Lock()
|
||||||
if (*localStats)["stat"] != remoteBlobCount*numClients && (*localStats)["create"] != te.numUnique {
|
if (*localStats)["stat"] != remoteBlobCount*numClients*2 && (*localStats)["create"] != te.numUnique {
|
||||||
sbsMu.Unlock()
|
sbsMu.Unlock()
|
||||||
t.Fatal("Expected: stat:", remoteBlobCount*numClients, "create:", remoteBlobCount)
|
t.Fatal("Expected: stat:", remoteBlobCount*numClients, "create:", remoteBlobCount, "Got: stat:", (*localStats)["stat"], "create:", (*localStats)["create"])
|
||||||
}
|
}
|
||||||
sbsMu.Unlock()
|
sbsMu.Unlock()
|
||||||
|
|
||||||
|
|
Loading…
Reference in a new issue