Skip to content

fix: replace os.Exit and panic in cp and sync with error handling #882

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
terryrankine wants to merge 1 commit into
base: master
Choose a base branch
from
Open

fix: replace os.Exit and panic in cp and sync with error handling #882

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
15 changes: 11 additions & 4 deletions command/cp.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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)
Expand Down Expand Up @@ -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)
Expand Down Expand Up @@ -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)
}
Expand Down
139 changes: 79 additions & 60 deletions command/sync.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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"
)
Expand Down Expand Up @@ -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 {
Expand Down Expand Up @@ -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
Expand Down Expand Up @@ -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.
Expand Down Expand Up @@ -424,19 +408,22 @@ 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
}

// 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,
Expand All @@ -462,36 +449,52 @@ 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)
}
}()

// both in source and destination
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)
}
}()

Expand All @@ -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 {
Expand All @@ -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
}
}
}
}
}()
Expand Down Expand Up @@ -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
}
}
Expand Down
Loading
Loading