github/containerd
2
0
Fork 0
Code Issues Pull Requests Packages Projects Releases 2 Wiki Activity

Remove duplicated code in content Copy

Browse Source 
Signed-off-by: Jin Dong <djdongjin95@gmail.com>
This commit is contained in:
Jin Dong 2023-06-23 14:46:04 +00:00
parent 9b4ed8acc2
commit cdc90c8381
1 changed files with 13 additions and 25 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

38
content/helpers.go
View File

@ -148,32 +148,28 @@ func OpenWriter(ctx context.Context, cs Ingester, opts ...WriterOpt) (Writer, er
// //
// Copy is buffered, so no need to wrap reader in buffered io. // Copy is buffered, so no need to wrap reader in buffered io.
func Copy(ctx context.Context, cw Writer, or io.Reader, size int64, expected digest.Digest, opts ...Opt) error { func Copy(ctx context.Context, cw Writer, or io.Reader, size int64, expected digest.Digest, opts ...Opt) error {
ws, err := cw.Status()
if err != nil {
return fmt.Errorf("failed to get status: %w", err)
}
r := or r := or
if ws.Offset > 0 {
r, err = seekReader(or, ws.Offset, size)
if err != nil {
return fmt.Errorf("unable to resume write to %v: %w", ws.Ref, err)
}
}
for i := 0; i < maxResets; i++ { for i := 0; i < maxResets; i++ {
if i >= 1 { if i >= 1 {
log.G(ctx).WithField("digest", expected).Debugf("retrying copy due to reset") log.G(ctx).WithField("digest", expected).Debugf("retrying copy due to reset")
} }
copied, err := copyWithBuffer(cw, r)
if errors.Is(err, ErrReset) { ws, err := cw.Status()
ws, err := cw.Status() if err != nil {
if err != nil { return fmt.Errorf("failed to get status: %w", err)
return fmt.Errorf("failed to get status: %w", err) }
} // Reset the original reader if
// 1. there is an offset, or
// 2. this is a retry due to Reset error
if ws.Offset > 0 || i > 0 {
r, err = seekReader(or, ws.Offset, size) r, err = seekReader(or, ws.Offset, size)
if err != nil { if err != nil {
return fmt.Errorf("unable to resume write to %v: %w", ws.Ref, err) return fmt.Errorf("unable to resume write to %v: %w", ws.Ref, err)
} }
}
copied, err := copyWithBuffer(cw, r)
if errors.Is(err, ErrReset) {
continue continue
} }
if err != nil { if err != nil {
@ -185,14 +181,6 @@ func Copy(ctx context.Context, cw Writer, or io.Reader, size int64, expected dig
} }
if err := cw.Commit(ctx, size, expected, opts...); err != nil { if err := cw.Commit(ctx, size, expected, opts...); err != nil {
if errors.Is(err, ErrReset) { if errors.Is(err, ErrReset) {
ws, err := cw.Status()
if err != nil {
return fmt.Errorf("failed to get status: %w", err)
}
r, err = seekReader(or, ws.Offset, size)
if err != nil {
return fmt.Errorf("unable to resume write to %v: %w", ws.Ref, err)
}
continue continue
} }
if !errdefs.IsAlreadyExists(err) { if !errdefs.IsAlreadyExists(err) {