diff --git a/command/cp.go b/command/cp.go index 7ed06bfee..d2e5c59a8 100644 --- a/command/cp.go +++ b/command/cp.go @@ -419,6 +419,9 @@ increase the open file limit or try to decrease the number of workers with // Run starts copying given source objects to destination. func (c Copy) Run(ctx context.Context) error { + ctx, cancel := context.WithCancel(ctx) + defer cancel() + // override source region if set if c.srcRegion != "" { c.storageOpts.SetRegion(c.srcRegion) @@ -451,9 +454,10 @@ func (c Copy) Run(ctx context.Context) error { for err := range waiter.Err() { if strings.Contains(err.Error(), "too many open files") { fmt.Println(strings.TrimSpace(fdlimitWarning)) - fmt.Printf("ERROR %v\n", err) - - os.Exit(1) + printError(c.fullCommand, c.op, err) + merrorWaiter = multierror.Append(merrorWaiter, err) + cancel() + continue } printError(c.fullCommand, c.op, err) merrorWaiter = multierror.Append(merrorWaiter, err) @@ -558,7 +562,10 @@ func (c Copy) Run(ctx context.Context) error { } task = c.prepareUploadTask(ctx, srcurl, c.dst, isBatch, c.metadata) default: - panic("unexpected src-dst pair") + err := fmt.Errorf("unexpected src-dst pair: %v -> %v", srcurl, c.dst) + merrorObjects = multierror.Append(merrorObjects, err) + printError(c.fullCommand, c.op, err) + continue } parallel.Run(task, waiter) } diff --git a/command/sync.go b/command/sync.go index 6dede5a66..37bbff737 100644 --- a/command/sync.go +++ b/command/sync.go @@ -4,20 +4,17 @@ import ( "context" "fmt" "io" - "os" "path/filepath" "strings" "sync" "github.com/aws/aws-sdk-go/aws/awserr" - "github.com/hashicorp/go-multierror" "github.com/lanrat/extsort" "github.com/urfave/cli/v2" errorpkg "github.com/peak/s5cmd/v2/error" "github.com/peak/s5cmd/v2/log" "github.com/peak/s5cmd/v2/log/stat" - "github.com/peak/s5cmd/v2/parallel" "github.com/peak/s5cmd/v2/storage" "github.com/peak/s5cmd/v2/storage/url" ) @@ -181,6 +178,7 @@ func (s Sync) Run(c *cli.Context) error { } ctx, cancel := context.WithCancel(c.Context) + defer cancel() sourceObjects, destObjects, err := s.getSourceAndDestinationObjects(ctx, cancel, srcurl, dsturl) if err != nil { @@ -208,34 +206,18 @@ func (s Sync) Run(c *cli.Context) error { sourceObjects = nil destObjects = nil - waiter := parallel.NewWaiter() - var ( - merrorWaiter error - errDoneCh = make(chan struct{}) - ) - - go func() { - defer close(errDoneCh) - for err := range waiter.Err() { - if strings.Contains(err.Error(), "too many open files") { - fmt.Println(strings.TrimSpace(fdlimitWarning)) - fmt.Printf("ERROR %v\n", err) - - os.Exit(1) - } - printError(s.fullCommand, s.op, err) - merrorWaiter = multierror.Append(merrorWaiter, err) - } - }() - strategy := NewStrategy(s.sizeOnly) // create comparison strategy. pipeReader, pipeWriter := io.Pipe() // create a reader, writer pipe to pass commands to run + // If Run stops consuming early (e.g. the context is cancelled), closing the + // reader unblocks any planRun goroutine stuck writing to the pipe so it can + // observe the cancellation and return. + defer pipeReader.Close() + // Create commands in background. - go s.planRun(c, onlySource, onlyDest, commonObjects, dsturl, strategy, pipeWriter, isBatch) + go s.planRun(ctx, c, onlySource, onlyDest, commonObjects, dsturl, strategy, pipeWriter, isBatch) - err = NewRun(c, pipeReader).Run(ctx) - return multierror.Append(err, merrorWaiter).ErrorOrNil() + return NewRun(c, pipeReader).Run(ctx) } // compareObjects compares source and destination objects. It assumes that @@ -377,12 +359,14 @@ func (s Sync) getSourceAndDestinationObjects(ctx context.Context, cancel context sourceObjects <- &o } - // read and print the external sort errors - go func() { - for err := range srcErrCh { + // read and print the external sort errors. extsort closes the error + // channel together with the output channel, so this does not block. + // It may send a nil error when the context is cancelled mid-merge. + for err := range srcErrCh { + if err != nil { printError(s.fullCommand, s.op, err) } - }() + } }() // get destination objects. @@ -424,12 +408,14 @@ func (s Sync) getSourceAndDestinationObjects(ctx context.Context, cancel context destObjects <- &o } - // read and print the external sort errors - go func() { - for err := range dstErrCh { + // read and print the external sort errors. extsort closes the error + // channel together with the output channel, so this does not block. + // It may send a nil error when the context is cancelled mid-merge. + for err := range dstErrCh { + if err != nil { printError(s.fullCommand, s.op, err) } - }() + } }() return sourceObjects, destObjects, nil @@ -437,6 +423,7 @@ func (s Sync) getSourceAndDestinationObjects(ctx context.Context, cancel context // planRun prepares the commands and writes them to writer 'w'. func (s Sync) planRun( + ctx context.Context, c *cli.Context, onlySource, onlyDest chan *url.URL, common chan *ObjectPair, @@ -462,14 +449,22 @@ func (s Sync) planRun( wg.Add(1) go func() { defer wg.Done() - for srcurl := range onlySource { - curDestURL := generateDestinationURL(srcurl, dsturl, isBatch) - command, err := generateCommand(c, "cp", defaultFlags, srcurl, curDestURL) - if err != nil { - printDebug(s.op, err, srcurl, curDestURL) - continue + for { + select { + case <-ctx.Done(): + return + case srcurl, ok := <-onlySource: + if !ok { + return + } + curDestURL := generateDestinationURL(srcurl, dsturl, isBatch) + command, err := generateCommand(c, "cp", defaultFlags, srcurl, curDestURL) + if err != nil { + printDebug(s.op, err, srcurl, curDestURL) + continue + } + fmt.Fprintln(w, command) } - fmt.Fprintln(w, command) } }() @@ -477,21 +472,29 @@ func (s Sync) planRun( wg.Add(1) go func() { defer wg.Done() - for commonObject := range common { - sourceObject, destObject := commonObject.src, commonObject.dst - curSourceURL, curDestURL := sourceObject.URL, destObject.URL - err := strategy.ShouldSync(sourceObject, destObject) // check if object should be copied. - if err != nil { - printDebug(s.op, err, curSourceURL, curDestURL) - continue - } + for { + select { + case <-ctx.Done(): + return + case commonObject, ok := <-common: + if !ok { + return + } + sourceObject, destObject := commonObject.src, commonObject.dst + curSourceURL, curDestURL := sourceObject.URL, destObject.URL + err := strategy.ShouldSync(sourceObject, destObject) // check if object should be copied. + if err != nil { + printDebug(s.op, err, curSourceURL, curDestURL) + continue + } - command, err := generateCommand(c, "cp", defaultFlags, curSourceURL, curDestURL) - if err != nil { - printDebug(s.op, err, curSourceURL, curDestURL) - continue + command, err := generateCommand(c, "cp", defaultFlags, curSourceURL, curDestURL) + if err != nil { + printDebug(s.op, err, curSourceURL, curDestURL) + continue + } + fmt.Fprintln(w, command) } - fmt.Fprintln(w, command) } }() @@ -504,8 +507,17 @@ func (s Sync) planRun( // or rewrite generateCommand function? dstURLs := make([]*url.URL, 0, extsortChunkSize) - for d := range onlyDest { - dstURLs = append(dstURLs, d) + collect: + for { + select { + case <-ctx.Done(): + return + case d, ok := <-onlyDest: + if !ok { + break collect + } + dstURLs = append(dstURLs, d) + } } if len(dstURLs) == 0 { @@ -519,10 +531,17 @@ func (s Sync) planRun( } fmt.Fprintln(w, command) } else { - // we only need to consume them from the channel so that rest of the objects + // we only need to consume them from the channel so that rest of the objects // can be sent to channel. - for d := range onlyDest { - _ = d + for { + select { + case <-ctx.Done(): + return + case _, ok := <-onlyDest: + if !ok { + return + } + } } } }() @@ -594,7 +613,7 @@ func (s Sync) shouldStopSync(err error) bool { } if awsErr, ok := err.(awserr.Error); ok { switch awsErr.Code() { - case "AccessDenied", "NoSuchBucket": + case "AccessDenied", "NoSuchBucket", "RequestError", "SerializationError": return true } } diff --git a/command/sync_test.go b/command/sync_test.go new file mode 100644 index 000000000..df7ac1baa --- /dev/null +++ b/command/sync_test.go @@ -0,0 +1,198 @@ +package command + +import ( + "context" + "fmt" + "io" + "sort" + "strings" + "testing" + "time" + + "github.com/aws/aws-sdk-go/aws/awserr" + "github.com/google/go-cmp/cmp" + "github.com/urfave/cli/v2" + + "github.com/peak/s5cmd/v2/log" + "github.com/peak/s5cmd/v2/storage" + "github.com/peak/s5cmd/v2/storage/url" +) + +func TestShouldStopSync(t *testing.T) { + t.Parallel() + + tests := []struct { + name string + err error + exitOnError bool + want bool + }{ + { + name: "AccessDenied", + err: awserr.New("AccessDenied", "access denied", nil), + want: true, + }, + { + name: "NoSuchBucket", + err: awserr.New("NoSuchBucket", "no such bucket", nil), + want: true, + }, + { + name: "RequestError", + err: awserr.New("RequestError", "request error", nil), + want: true, + }, + { + name: "SerializationError", + err: awserr.New("SerializationError", "serialization error", nil), + want: true, + }, + { + name: "SlowDown should not stop", + err: awserr.New("SlowDown", "slow down", nil), + want: false, + }, + { + name: "non-AWS error with exitOnError true", + err: fmt.Errorf("some random error"), + exitOnError: true, + want: true, + }, + { + name: "non-AWS error with exitOnError false", + err: fmt.Errorf("some random error"), + exitOnError: false, + want: false, + }, + { + name: "nil error", + err: nil, + want: false, + }, + } + + for _, tc := range tests { + tc := tc + t.Run(tc.name, func(t *testing.T) { + t.Parallel() + + s := Sync{ + exitOnError: tc.exitOnError, + } + + got := s.shouldStopSync(tc.err) + if got != tc.want { + t.Errorf("shouldStopSync(%v) = %v, want %v", tc.err, got, tc.want) + } + }) + } +} + +// readPlan runs planRun in the background and returns everything it wrote +// before closing the writer. It fails the test if the writer is never closed. +func readPlan(t *testing.T, ctx context.Context, s Sync, c *cli.Context, onlySource, onlyDest chan *url.URL, common chan *ObjectPair, isBatch bool) []string { + t.Helper() + + dsturl := mustNewURL(t, "s3://bucket/dst/") + pr, pw := io.Pipe() + + go s.planRun(ctx, c, onlySource, onlyDest, common, dsturl, NewStrategy(false), pw, isBatch) + + type result struct { + out []byte + err error + } + resultCh := make(chan result, 1) + go func() { + out, err := io.ReadAll(pr) + resultCh <- result{out, err} + }() + + select { + case r := <-resultCh: + if r.err != nil { + t.Fatalf("unexpected read error: %v", r.err) + } + lines := strings.Split(strings.TrimSpace(string(r.out)), "\n") + if len(lines) == 1 && lines[0] == "" { + return nil + } + sort.Strings(lines) + return lines + case <-time.After(10 * time.Second): + t.Fatal("planRun did not close the writer") + return nil + } +} + +func TestPlanRunReturnsOnCancelledContext(t *testing.T) { + t.Parallel() + + for _, del := range []bool{false, true} { + del := del + t.Run(fmt.Sprintf("delete=%v", del), func(t *testing.T) { + t.Parallel() + + ctx, cancel := context.WithCancel(context.Background()) + cancel() + + // Nothing is ever sent on (or closes) these channels. Without the + // cancellation checks planRun would block on them forever. + onlySource := make(chan *url.URL) + onlyDest := make(chan *url.URL) + common := make(chan *ObjectPair) + + got := readPlan(t, ctx, Sync{delete: del}, nil, onlySource, onlyDest, common, true) + if len(got) != 0 { + t.Errorf("expected no commands, got %v", got) + } + }) + } +} + +func TestPlanRunGeneratesCommands(t *testing.T) { + t.Parallel() + + // planRun logs skipped objects through the global logger. + log.Init("error", false) + + app := cli.NewApp() + c := cli.NewContext(app, flagSet(t, "sync", nil), nil) + + now := time.Now() + older := now.Add(-time.Hour) + + onlySource := make(chan *url.URL, 1) + onlyDest := make(chan *url.URL, 2) + common := make(chan *ObjectPair, 2) + + onlySource <- mustNewURL(t, "s3://bucket/src/new") + close(onlySource) + + onlyDest <- mustNewURL(t, "s3://bucket/dst/stale1") + onlyDest <- mustNewURL(t, "s3://bucket/dst/stale2") + close(onlyDest) + + // source is newer and differs in size: should be copied. + common <- &ObjectPair{ + src: &storage.Object{URL: mustNewURL(t, "s3://bucket/src/changed"), Size: 2, ModTime: &now}, + dst: &storage.Object{URL: mustNewURL(t, "s3://bucket/dst/changed"), Size: 1, ModTime: &older}, + } + // identical: should be skipped. + common <- &ObjectPair{ + src: &storage.Object{URL: mustNewURL(t, "s3://bucket/src/same"), Size: 1, ModTime: &now}, + dst: &storage.Object{URL: mustNewURL(t, "s3://bucket/dst/same"), Size: 1, ModTime: &now}, + } + close(common) + + got := readPlan(t, context.Background(), Sync{delete: true}, c, onlySource, onlyDest, common, false) + + want := []string{ + `cp --raw='true' "s3://bucket/src/changed" "s3://bucket/dst/changed"`, + `cp --raw='true' "s3://bucket/src/new" "s3://bucket/dst/new"`, + `rm --raw='true' "s3://bucket/dst/stale1" "s3://bucket/dst/stale2"`, + } + if diff := cmp.Diff(want, got); diff != "" { + t.Errorf("(-want +got):\n%v", diff) + } +}