Skip to content

Parallel workers to cp and mv command. Backward python 2.5.2 compatibility. #2

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
cyno wants to merge 6 commits into
base: master
Choose a base branch
from
Open

Parallel workers to cp and mv command. Backward python 2.5.2 compatibility. #2

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
6 commits
Select commit Hold shift + click to select a range
1b22cd7
Add parallel workers to cp and mv command
Aug 2, 2011
ced3963
sync bucket to bucket and ls --directory option
Aug 4, 2011
b3debba
Add sequence label in cp and mv commands
Aug 5, 2011
88869ab
Add --max-retries and --retry-delay
Oct 31, 2011
01c9ec3
Remove delay message if there is no retry delay
Dec 8, 2011
e265e11
Fix bug in remote2remote sync
Feb 16, 2012
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
3 changes: 3 additions & 0 deletions S3/Config.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -77,6 +77,9 @@ class Config(object):
parallel = False
workers = 10
follow_symlinks=False
select_dir = False
max_retries = 5
retry_delay = 3

## Creating a singleton
def __new__(self, configfile = None):
Expand Down
70 changes: 47 additions & 23 deletions S3/S3.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -47,7 +47,7 @@ def update_timestamp(self):
def format_param_str(self):
"""
Format URL parameters from self.params and returns
?parm1=val1&parm2=val2 or an empty string if there
?parm1=val1&parm2=val2 or an empty string if there
are no parameters. Output of this function should
be appended directly to self.resource['uri']
"""
Expand Down Expand Up @@ -119,9 +119,6 @@ class S3(object):
## S3 sometimes sends HTTP-307 response
redir_map = {}

## Maximum attempts of re-issuing failed requests
_max_retries = 5

def __init__(self, config):
self.config = config

Expand Down Expand Up @@ -458,9 +455,11 @@ def create_request(self, operation, uri = None, bucket = None, object = None, he

def _fail_wait(self, retries):
# Wait a few seconds. The more it fails the more we wait.
return (self._max_retries - retries + 1) * 3
return (self.config.max_retries - retries + 1) * self.config.retry_delay

def send_request(self, request, body = None, retries = _max_retries):
def send_request(self, request, body = None, retries = -1):
if retries == -1:
retries = self.config.max_retries
method_string, resource, headers = request.get_triplet()
debug("Processing request, please wait...")
if not headers.has_key('content-length'):
Expand All @@ -479,8 +478,11 @@ def send_request(self, request, body = None, retries = _max_retries):
except Exception, e:
if retries:
warning("Retrying failed request: %s (%s)" % (resource['uri'], e))
warning("Waiting %d sec..." % self._fail_wait(retries))
time.sleep(self._fail_wait(retries))

if self._fail_wait(retries) > 0:
warning("Waiting %d sec..." % self._fail_wait(retries))
time.sleep(self._fail_wait(retries))

return self.send_request(request, body, retries - 1)
else:
raise S3RequestError("Request failed for: %s" % resource['uri'])
Expand All @@ -498,8 +500,11 @@ def send_request(self, request, body = None, retries = _max_retries):
if retries:
warning(u"Retrying failed request: %s" % resource['uri'])
warning(unicode(e))
warning("Waiting %d sec..." % self._fail_wait(retries))
time.sleep(self._fail_wait(retries))

if self._fail_wait(retries) > 0:
warning("Waiting %d sec..." % self._fail_wait(retries))
time.sleep(self._fail_wait(retries))

return self.send_request(request, body, retries - 1)
else:
raise e
Expand All @@ -509,7 +514,9 @@ def send_request(self, request, body = None, retries = _max_retries):

return response

def send_file(self, request, file, labels, throttle = 0, retries = _max_retries):
def send_file(self, request, file, labels, throttle = 0, retries = -1):
if retries == -1:
retries = self.config.max_retries
method_string, resource, headers = request.get_triplet()
size_left = size_total = headers.get("content-length")
if self.config.progress_meter:
Expand All @@ -529,8 +536,11 @@ def send_file(self, request, file, labels, throttle = 0, retries = _max_retries)
progress.done("failed")
if retries:
warning("Retrying failed request: %s (%s)" % (resource['uri'], e))
warning("Waiting %d sec..." % self._fail_wait(retries))
time.sleep(self._fail_wait(retries))

if self._fail_wait(retries) > 0:
warning("Waiting %d sec..." % self._fail_wait(retries))
time.sleep(self._fail_wait(retries))

# Connection error -> same throttle value
return self.send_file(request, file, labels, throttle, retries - 1)
else:
Expand Down Expand Up @@ -562,12 +572,15 @@ def send_file(self, request, file, labels, throttle = 0, retries = _max_retries)
if self.config.progress_meter:
progress.done("failed")
if retries:
if retries < self._max_retries:
if retries < self.config.max_retries:
throttle = throttle and throttle * 5 or 0.01
warning("Upload failed: %s (%s)" % (resource['uri'], e))
warning("Retrying on lower speed (throttle=%0.2f)" % throttle)
warning("Waiting %d sec..." % self._fail_wait(retries))
time.sleep(self._fail_wait(retries))

if self._fail_wait(retries) > 0:
warning("Waiting %d sec..." % self._fail_wait(retries))
time.sleep(self._fail_wait(retries))

# Connection error -> same throttle value
return self.send_file(request, file, labels, throttle, retries - 1)
else:
Expand Down Expand Up @@ -612,8 +625,11 @@ def send_file(self, request, file, labels, throttle = 0, retries = _max_retries)
if try_retry:
if retries:
warning("Upload failed: %s (%s)" % (resource['uri'], S3Error(response)))
warning("Waiting %d sec..." % self._fail_wait(retries))
time.sleep(self._fail_wait(retries))

if self._fail_wait(retries) > 0:
warning("Waiting %d sec..." % self._fail_wait(retries))
time.sleep(self._fail_wait(retries))

return self.send_file(request, file, labels, throttle, retries - 1)
else:
warning("Too many failures. Giving up on '%s'" % (file.name))
Expand All @@ -634,7 +650,9 @@ def send_file(self, request, file, labels, throttle = 0, retries = _max_retries)

return response

def recv_file(self, request, stream, labels, start_position = 0, retries = _max_retries):
def recv_file(self, request, stream, labels, start_position = 0, retries = -1):
if retries == -1:
retries = self.config.max_retries
method_string, resource, headers = request.get_triplet()
if self.config.progress_meter:
progress = self.config.progress_class(labels, 0)
Expand Down Expand Up @@ -662,8 +680,11 @@ def recv_file(self, request, stream, labels, start_position = 0, retries = _max_
progress.done("failed")
if retries:
warning("Retrying failed request: %s (%s)" % (resource['uri'], e))
warning("Waiting %d sec..." % self._fail_wait(retries))
time.sleep(self._fail_wait(retries))

if self._fail_wait(retries) > 0:
warning("Waiting %d sec..." % self._fail_wait(retries))
time.sleep(self._fail_wait(retries))

# Connection error -> same throttle value
return self.recv_file(request, stream, labels, start_position, retries - 1)
else:
Expand Down Expand Up @@ -711,8 +732,11 @@ def recv_file(self, request, stream, labels, start_position = 0, retries = _max_
progress.done("failed")
if retries:
warning("Retrying failed request: %s (%s)" % (resource['uri'], e))
warning("Waiting %d sec..." % self._fail_wait(retries))
time.sleep(self._fail_wait(retries))

if self._fail_wait(retries) > 0:
warning("Waiting %d sec..." % self._fail_wait(retries))
time.sleep(self._fail_wait(retries))

# Connection error -> same throttle value
return self.recv_file(request, stream, labels, current_position, retries - 1)
else:
Expand Down
Loading