| OLD | NEW |
|---|
| (Empty) |
| 1 # Copyright 2010 Google Inc. | |
| 2 # | |
| 3 # Permission is hereby granted, free of charge, to any person obtaining a | |
| 4 # copy of this software and associated documentation files (the | |
| 5 # "Software"), to deal in the Software without restriction, including | |
| 6 # without limitation the rights to use, copy, modify, merge, publish, dis- | |
| 7 # tribute, sublicense, and/or sell copies of the Software, and to permit | |
| 8 # persons to whom the Software is furnished to do so, subject to the fol- | |
| 9 # lowing conditions: | |
| 10 # | |
| 11 # The above copyright notice and this permission notice shall be included | |
| 12 # in all copies or substantial portions of the Software. | |
| 13 # | |
| 14 # THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS | |
| 15 # OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABIL- | |
| 16 # ITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT | |
| 17 # SHALL THE AUTHOR BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, | |
| 18 # WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, | |
| 19 # OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS | |
| 20 # IN THE SOFTWARE. | |
| 21 | |
| 22 import cgi | |
| 23 import errno | |
| 24 import httplib | |
| 25 import os | |
| 26 import re | |
| 27 import socket | |
| 28 import time | |
| 29 import urlparse | |
| 30 import boto | |
| 31 from boto import config | |
| 32 from boto.connection import AWSAuthConnection | |
| 33 from boto.exception import InvalidUriError | |
| 34 from boto.exception import ResumableTransferDisposition | |
| 35 from boto.exception import ResumableUploadException | |
| 36 | |
| 37 """ | |
| 38 Handler for Google Storage resumable uploads. See | |
| 39 http://code.google.com/apis/storage/docs/developer-guide.html#resumable | |
| 40 for details. | |
| 41 | |
| 42 Resumable uploads will retry failed uploads, resuming at the byte | |
| 43 count completed by the last upload attempt. If too many retries happen with | |
| 44 no progress (per configurable num_retries param), the upload will be | |
| 45 aborted in the current process. | |
| 46 | |
| 47 The caller can optionally specify a tracker_file_name param in the | |
| 48 ResumableUploadHandler constructor. If you do this, that file will | |
| 49 save the state needed to allow retrying later, in a separate process | |
| 50 (e.g., in a later run of gsutil). | |
| 51 """ | |
| 52 | |
| 53 | |
| 54 class ResumableUploadHandler(object): | |
| 55 | |
| 56 BUFFER_SIZE = 8192 | |
| 57 RETRYABLE_EXCEPTIONS = (httplib.HTTPException, IOError, socket.error, | |
| 58 socket.gaierror) | |
| 59 | |
| 60 # (start, end) response indicating server has nothing (upload protocol uses | |
| 61 # inclusive numbering). | |
| 62 SERVER_HAS_NOTHING = (0, -1) | |
| 63 | |
| 64 def __init__(self, tracker_file_name=None, num_retries=None): | |
| 65 """ | |
| 66 Constructor. Instantiate once for each uploaded file. | |
| 67 | |
| 68 :type tracker_file_name: string | |
| 69 :param tracker_file_name: optional file name to save tracker URI. | |
| 70 If supplied and the current process fails the upload, it can be | |
| 71 retried in a new process. If called with an existing file containing | |
| 72 a valid tracker URI, we'll resume the upload from this URI; else | |
| 73 we'll start a new resumable upload (and write the URI to this | |
| 74 tracker file). | |
| 75 | |
| 76 :type num_retries: int | |
| 77 :param num_retries: the number of times we'll re-try a resumable upload | |
| 78 making no progress. (Count resets every time we get progress, so | |
| 79 upload can span many more than this number of retries.) | |
| 80 """ | |
| 81 self.tracker_file_name = tracker_file_name | |
| 82 self.num_retries = num_retries | |
| 83 self.server_has_bytes = 0 # Byte count at last server check. | |
| 84 self.tracker_uri = None | |
| 85 if tracker_file_name: | |
| 86 self._load_tracker_uri_from_file() | |
| 87 # Save upload_start_point in instance state so caller can find how | |
| 88 # much was transferred by this ResumableUploadHandler (across retries). | |
| 89 self.upload_start_point = None | |
| 90 | |
| 91 def _load_tracker_uri_from_file(self): | |
| 92 f = None | |
| 93 try: | |
| 94 f = open(self.tracker_file_name, 'r') | |
| 95 uri = f.readline().strip() | |
| 96 self._set_tracker_uri(uri) | |
| 97 except IOError, e: | |
| 98 # Ignore non-existent file (happens first time an upload | |
| 99 # is attempted on a file), but warn user for other errors. | |
| 100 if e.errno != errno.ENOENT: | |
| 101 # Will restart because self.tracker_uri == None. | |
| 102 print('Couldn\'t read URI tracker file (%s): %s. Restarting ' | |
| 103 'upload from scratch.' % | |
| 104 (self.tracker_file_name, e.strerror)) | |
| 105 except InvalidUriError, e: | |
| 106 # Warn user, but proceed (will restart because | |
| 107 # self.tracker_uri == None). | |
| 108 print('Invalid tracker URI (%s) found in URI tracker file ' | |
| 109 '(%s). Restarting upload from scratch.' % | |
| 110 (uri, self.tracker_file_name)) | |
| 111 finally: | |
| 112 if f: | |
| 113 f.close() | |
| 114 | |
| 115 def _save_tracker_uri_to_file(self): | |
| 116 """ | |
| 117 Saves URI to tracker file if one was passed to constructor. | |
| 118 """ | |
| 119 if not self.tracker_file_name: | |
| 120 return | |
| 121 f = None | |
| 122 try: | |
| 123 f = open(self.tracker_file_name, 'w') | |
| 124 f.write(self.tracker_uri) | |
| 125 except IOError, e: | |
| 126 raise ResumableUploadException( | |
| 127 'Couldn\'t write URI tracker file (%s): %s.\nThis can happen' | |
| 128 'if you\'re using an incorrectly configured upload tool\n' | |
| 129 '(e.g., gsutil configured to save tracker files to an ' | |
| 130 'unwritable directory)' % | |
| 131 (self.tracker_file_name, e.strerror), | |
| 132 ResumableTransferDisposition.ABORT) | |
| 133 finally: | |
| 134 if f: | |
| 135 f.close() | |
| 136 | |
| 137 def _set_tracker_uri(self, uri): | |
| 138 """ | |
| 139 Called when we start a new resumable upload or get a new tracker | |
| 140 URI for the upload. Saves URI and resets upload state. | |
| 141 | |
| 142 Raises InvalidUriError if URI is syntactically invalid. | |
| 143 """ | |
| 144 parse_result = urlparse.urlparse(uri) | |
| 145 if (parse_result.scheme.lower() not in ['http', 'https'] or | |
| 146 not parse_result.netloc or not parse_result.query): | |
| 147 raise InvalidUriError('Invalid tracker URI (%s)' % uri) | |
| 148 qdict = cgi.parse_qs(parse_result.query) | |
| 149 if not qdict or not 'upload_id' in qdict: | |
| 150 raise InvalidUriError('Invalid tracker URI (%s)' % uri) | |
| 151 self.tracker_uri = uri | |
| 152 self.tracker_uri_host = parse_result.netloc | |
| 153 self.tracker_uri_path = '%s/?%s' % (parse_result.netloc, | |
| 154 parse_result.query) | |
| 155 self.server_has_bytes = 0 | |
| 156 | |
| 157 def get_tracker_uri(self): | |
| 158 """ | |
| 159 Returns upload tracker URI, or None if the upload has not yet started. | |
| 160 """ | |
| 161 return self.tracker_uri | |
| 162 | |
| 163 def _remove_tracker_file(self): | |
| 164 if (self.tracker_file_name and | |
| 165 os.path.exists(self.tracker_file_name)): | |
| 166 os.unlink(self.tracker_file_name) | |
| 167 | |
| 168 def _build_content_range_header(self, range_spec='*', length_spec='*'): | |
| 169 return 'bytes %s/%s' % (range_spec, length_spec) | |
| 170 | |
| 171 def _query_server_state(self, conn, file_length): | |
| 172 """ | |
| 173 Queries server to find out what bytes it currently has. | |
| 174 | |
| 175 Note that this method really just makes special case use of the | |
| 176 fact that the upload server always returns the current start/end | |
| 177 state whenever a PUT doesn't complete. | |
| 178 | |
| 179 Returns (server_start, server_end), where the values are inclusive. | |
| 180 For example, (0, 2) would mean that the server has bytes 0, 1, *and* 2. | |
| 181 | |
| 182 Raises ResumableUploadException if problem querying server. | |
| 183 """ | |
| 184 # Send an empty PUT so that server replies with this resumable | |
| 185 # transfer's state. | |
| 186 put_headers = {} | |
| 187 put_headers['Content-Range'] = ( | |
| 188 self._build_content_range_header('*', file_length)) | |
| 189 put_headers['Content-Length'] = '0' | |
| 190 resp = AWSAuthConnection.make_request(conn, 'PUT', | |
| 191 path=self.tracker_uri_path, | |
| 192 auth_path=self.tracker_uri_path, | |
| 193 headers=put_headers, | |
| 194 host=self.tracker_uri_host) | |
| 195 if resp.status == 200: | |
| 196 return (0, file_length) # Completed upload. | |
| 197 if resp.status != 308: | |
| 198 # This means the server didn't have any state for the given | |
| 199 # upload ID, which can happen (for example) if the caller saved | |
| 200 # the tracker URI to a file and then tried to restart the transfer | |
| 201 # after that upload ID has gone stale. In that case we need to | |
| 202 # start a new transfer (and the caller will then save the new | |
| 203 # tracker URI to the tracker file). | |
| 204 raise ResumableUploadException( | |
| 205 'Got non-308 response (%s) from server state query' % | |
| 206 resp.status, ResumableTransferDisposition.START_OVER) | |
| 207 got_valid_response = False | |
| 208 range_spec = resp.getheader('range') | |
| 209 if range_spec: | |
| 210 # Parse 'bytes=<from>-<to>' range_spec. | |
| 211 m = re.search('bytes=(\d+)-(\d+)', range_spec) | |
| 212 if m: | |
| 213 server_start = long(m.group(1)) | |
| 214 server_end = long(m.group(2)) | |
| 215 got_valid_response = True | |
| 216 else: | |
| 217 # No Range header, which means the server does not yet have | |
| 218 # any bytes. Note that the Range header uses inclusive 'from' | |
| 219 # and 'to' values. Since Range 0-0 would mean that the server | |
| 220 # has byte 0, omitting the Range header is used to indicate that | |
| 221 # the server doesn't have any bytes. | |
| 222 return self.SERVER_HAS_NOTHING | |
| 223 if not got_valid_response: | |
| 224 raise ResumableUploadException( | |
| 225 'Couldn\'t parse upload server state query response (%s)' % | |
| 226 str(resp.getheaders()), ResumableTransferDisposition.START_OVER) | |
| 227 if conn.debug >= 1: | |
| 228 print 'Server has: Range: %d - %d.' % (server_start, server_end) | |
| 229 return (server_start, server_end) | |
| 230 | |
| 231 def _start_new_resumable_upload(self, key, headers=None): | |
| 232 """ | |
| 233 Starts a new resumable upload. | |
| 234 | |
| 235 Raises ResumableUploadException if any errors occur. | |
| 236 """ | |
| 237 conn = key.bucket.connection | |
| 238 if conn.debug >= 1: | |
| 239 print 'Starting new resumable upload.' | |
| 240 self.server_has_bytes = 0 | |
| 241 | |
| 242 # Start a new resumable upload by sending a POST request with an | |
| 243 # empty body and the "X-Goog-Resumable: start" header. Include any | |
| 244 # caller-provided headers (e.g., Content-Type) EXCEPT Content-Length | |
| 245 # (and raise an exception if they tried to pass one, since it's | |
| 246 # a semantic error to specify it at this point, and if we were to | |
| 247 # include one now it would cause the server to expect that many | |
| 248 # bytes; the POST doesn't include the actual file bytes We set | |
| 249 # the Content-Length in the subsequent PUT, based on the uploaded | |
| 250 # file size. | |
| 251 post_headers = {} | |
| 252 for k in headers: | |
| 253 if k.lower() == 'content-length': | |
| 254 raise ResumableUploadException( | |
| 255 'Attempt to specify Content-Length header (disallowed)', | |
| 256 ResumableTransferDisposition.ABORT) | |
| 257 post_headers[k] = headers[k] | |
| 258 post_headers[conn.provider.resumable_upload_header] = 'start' | |
| 259 | |
| 260 resp = conn.make_request( | |
| 261 'POST', key.bucket.name, key.name, post_headers) | |
| 262 # Get tracker URI from response 'Location' header. | |
| 263 body = resp.read() | |
| 264 | |
| 265 # Check for various status conditions. | |
| 266 if resp.status == 500 or resp.status == 503: | |
| 267 # Retry status 500 and 503 errors after a delay. | |
| 268 raise ResumableUploadException( | |
| 269 'Got status %d from attempt to start resumable upload. ' | |
| 270 'Will wait/retry' % resp.status, | |
| 271 ResumableTransferDisposition.WAIT_BEFORE_RETRY) | |
| 272 elif resp.status != 200 and resp.status != 201: | |
| 273 raise ResumableUploadException( | |
| 274 'Got status %d from attempt to start resumable upload. ' | |
| 275 'Aborting' % resp.status, | |
| 276 ResumableTransferDisposition.ABORT) | |
| 277 | |
| 278 # Else we got 200 or 201 response code, indicating the resumable | |
| 279 # upload was created. | |
| 280 tracker_uri = resp.getheader('Location') | |
| 281 if not tracker_uri: | |
| 282 raise ResumableUploadException( | |
| 283 'No resumable tracker URI found in resumable initiation ' | |
| 284 'POST response (%s)' % body, | |
| 285 ResumableTransferDisposition.WAIT_BEFORE_RETRY) | |
| 286 self._set_tracker_uri(tracker_uri) | |
| 287 self._save_tracker_uri_to_file() | |
| 288 | |
| 289 def _upload_file_bytes(self, conn, http_conn, fp, file_length, | |
| 290 total_bytes_uploaded, cb, num_cb): | |
| 291 """ | |
| 292 Makes one attempt to upload file bytes, using an existing resumable | |
| 293 upload connection. | |
| 294 | |
| 295 Returns etag from server upon success. | |
| 296 | |
| 297 Raises ResumableUploadException if any problems occur. | |
| 298 """ | |
| 299 buf = fp.read(self.BUFFER_SIZE) | |
| 300 if cb: | |
| 301 if num_cb > 2: | |
| 302 cb_count = file_length / self.BUFFER_SIZE / (num_cb-2) | |
| 303 elif num_cb < 0: | |
| 304 cb_count = -1 | |
| 305 else: | |
| 306 cb_count = 0 | |
| 307 i = 0 | |
| 308 cb(total_bytes_uploaded, file_length) | |
| 309 | |
| 310 # Build resumable upload headers for the transfer. Don't send a | |
| 311 # Content-Range header if the file is 0 bytes long, because the | |
| 312 # resumable upload protocol uses an *inclusive* end-range (so, sending | |
| 313 # 'bytes 0-0/1' would actually mean you're sending a 1-byte file). | |
| 314 put_headers = {} | |
| 315 if file_length: | |
| 316 range_header = self._build_content_range_header( | |
| 317 '%d-%d' % (total_bytes_uploaded, file_length - 1), | |
| 318 file_length) | |
| 319 put_headers['Content-Range'] = range_header | |
| 320 # Set Content-Length to the total bytes we'll send with this PUT. | |
| 321 put_headers['Content-Length'] = str(file_length - total_bytes_uploaded) | |
| 322 http_request = AWSAuthConnection.build_base_http_request( | |
| 323 conn, 'PUT', path=self.tracker_uri_path, auth_path=None, | |
| 324 headers=put_headers, host=self.tracker_uri_host) | |
| 325 http_conn.putrequest('PUT', http_request.path) | |
| 326 for k in put_headers: | |
| 327 http_conn.putheader(k, put_headers[k]) | |
| 328 http_conn.endheaders() | |
| 329 | |
| 330 # Turn off debug on http connection so upload content isn't included | |
| 331 # in debug stream. | |
| 332 http_conn.set_debuglevel(0) | |
| 333 while buf: | |
| 334 http_conn.send(buf) | |
| 335 total_bytes_uploaded += len(buf) | |
| 336 if cb: | |
| 337 i += 1 | |
| 338 if i == cb_count or cb_count == -1: | |
| 339 cb(total_bytes_uploaded, file_length) | |
| 340 i = 0 | |
| 341 buf = fp.read(self.BUFFER_SIZE) | |
| 342 if cb: | |
| 343 cb(total_bytes_uploaded, file_length) | |
| 344 if total_bytes_uploaded != file_length: | |
| 345 # Abort (and delete the tracker file) so if the user retries | |
| 346 # they'll start a new resumable uplaod rather than potentially | |
| 347 # attempting to pick back up later where we left off. | |
| 348 raise ResumableUploadException( | |
| 349 'File changed during upload: EOF at %d bytes of %d byte file.' % | |
| 350 (total_bytes_uploaded, file_length), | |
| 351 ResumableTransferDisposition.ABORT) | |
| 352 resp = http_conn.getresponse() | |
| 353 body = resp.read() | |
| 354 # Restore http connection debug level. | |
| 355 http_conn.set_debuglevel(conn.debug) | |
| 356 | |
| 357 additional_note = '' | |
| 358 if resp.status == 200: | |
| 359 return resp.getheader('etag') # Success | |
| 360 elif resp.status == 408: | |
| 361 # Request Timeout. Try again later within the current process. | |
| 362 disposition = ResumableTransferDisposition.WAIT_BEFORE_RETRY | |
| 363 elif resp.status/100 == 4: | |
| 364 # Abort for any other 4xx errors. | |
| 365 disposition = ResumableTransferDisposition.ABORT | |
| 366 # Add some more informative note for particular 4xx error codes. | |
| 367 if resp.status == 400: | |
| 368 additional_note = ('This can happen for various reasons; one ' | |
| 369 'common case is if you attempt to upload a ' | |
| 370 'different size file on a already partially ' | |
| 371 'uploaded resumable upload') | |
| 372 # Retry status 500 and 503 errors after a delay. | |
| 373 elif resp.status == 500 or resp.status == 503: | |
| 374 disposition = ResumableTransferDisposition.WAIT_BEFORE_RETRY | |
| 375 else: | |
| 376 # Catch all for any other error codes. | |
| 377 disposition = ResumableTransferDisposition.ABORT | |
| 378 raise ResumableUploadException('Got response code %d while attempting ' | |
| 379 'upload (%s)%s' % | |
| 380 (resp.status, resp.reason, | |
| 381 additional_note), disposition) | |
| 382 | |
| 383 def _attempt_resumable_upload(self, key, fp, file_length, headers, cb, | |
| 384 num_cb): | |
| 385 """ | |
| 386 Attempts a resumable upload. | |
| 387 | |
| 388 Returns etag from server upon success. | |
| 389 | |
| 390 Raises ResumableUploadException if any problems occur. | |
| 391 """ | |
| 392 (server_start, server_end) = self.SERVER_HAS_NOTHING | |
| 393 conn = key.bucket.connection | |
| 394 if self.tracker_uri: | |
| 395 # Try to resume existing resumable upload. | |
| 396 try: | |
| 397 (server_start, server_end) = ( | |
| 398 self._query_server_state(conn, file_length)) | |
| 399 self.server_has_bytes = server_start | |
| 400 key=key | |
| 401 if conn.debug >= 1: | |
| 402 print 'Resuming transfer.' | |
| 403 except ResumableUploadException, e: | |
| 404 if conn.debug >= 1: | |
| 405 print 'Unable to resume transfer (%s).' % e.message | |
| 406 self._start_new_resumable_upload(key, headers) | |
| 407 else: | |
| 408 self._start_new_resumable_upload(key, headers) | |
| 409 | |
| 410 # upload_start_point allows the code that instantiated the | |
| 411 # ResumableUploadHandler to find out the point from which it started | |
| 412 # uploading (e.g., so it can correctly compute throughput). | |
| 413 if self.upload_start_point is None: | |
| 414 self.upload_start_point = server_end | |
| 415 | |
| 416 if server_end == file_length: | |
| 417 # Boundary condition: complete file was already uploaded (e.g., | |
| 418 # user interrupted a previous upload attempt after the upload | |
| 419 # completed but before the gsutil tracker file was deleted). Set | |
| 420 # total_bytes_uploaded to server_end so we'll attempt to upload | |
| 421 # no more bytes but will still make final HTTP request and get | |
| 422 # back the response (which contains the etag we need to compare | |
| 423 # at the end). | |
| 424 total_bytes_uploaded = server_end | |
| 425 else: | |
| 426 total_bytes_uploaded = server_end + 1 | |
| 427 fp.seek(total_bytes_uploaded) | |
| 428 conn = key.bucket.connection | |
| 429 | |
| 430 # Get a new HTTP connection (vs conn.get_http_connection(), which reuses | |
| 431 # pool connections) because httplib requires a new HTTP connection per | |
| 432 # transaction. (Without this, calling http_conn.getresponse() would get | |
| 433 # "ResponseNotReady".) | |
| 434 http_conn = conn.new_http_connection(self.tracker_uri_host, | |
| 435 conn.is_secure) | |
| 436 http_conn.set_debuglevel(conn.debug) | |
| 437 | |
| 438 # Make sure to close http_conn at end so if a local file read | |
| 439 # failure occurs partway through server will terminate current upload | |
| 440 # and can report that progress on next attempt. | |
| 441 try: | |
| 442 return self._upload_file_bytes(conn, http_conn, fp, file_length, | |
| 443 total_bytes_uploaded, cb, num_cb) | |
| 444 finally: | |
| 445 http_conn.close() | |
| 446 | |
| 447 def _check_final_md5(self, key, etag): | |
| 448 """ | |
| 449 Checks that etag from server agrees with md5 computed before upload. | |
| 450 This is important, since the upload could have spanned a number of | |
| 451 hours and multiple processes (e.g., gsutil runs), and the user could | |
| 452 change some of the file and not realize they have inconsistent data. | |
| 453 """ | |
| 454 if key.bucket.connection.debug >= 1: | |
| 455 print 'Checking md5 against etag.' | |
| 456 if key.md5 != etag.strip('"\''): | |
| 457 # Call key.open_read() before attempting to delete the | |
| 458 # (incorrect-content) key, so we perform that request on a | |
| 459 # different HTTP connection. This is neededb because httplib | |
| 460 # will return a "Response not ready" error if you try to perform | |
| 461 # a second transaction on the connection. | |
| 462 key.open_read() | |
| 463 key.close() | |
| 464 key.delete() | |
| 465 raise ResumableUploadException( | |
| 466 'File changed during upload: md5 signature doesn\'t match etag ' | |
| 467 '(incorrect uploaded object deleted)', | |
| 468 ResumableTransferDisposition.ABORT) | |
| 469 | |
| 470 def send_file(self, key, fp, headers, cb=None, num_cb=10): | |
| 471 """ | |
| 472 Upload a file to a key into a bucket on GS, using GS resumable upload | |
| 473 protocol. | |
| 474 | |
| 475 :type key: :class:`boto.s3.key.Key` or subclass | |
| 476 :param key: The Key object to which data is to be uploaded | |
| 477 | |
| 478 :type fp: file-like object | |
| 479 :param fp: The file pointer to upload | |
| 480 | |
| 481 :type headers: dict | |
| 482 :param headers: The headers to pass along with the PUT request | |
| 483 | |
| 484 :type cb: function | |
| 485 :param cb: a callback function that will be called to report progress on | |
| 486 the upload. The callback should accept two integer parameters, the | |
| 487 first representing the number of bytes that have been successfully | |
| 488 transmitted to GS, and the second representing the total number of | |
| 489 bytes that need to be transmitted. | |
| 490 | |
| 491 :type num_cb: int | |
| 492 :param num_cb: (optional) If a callback is specified with the cb | |
| 493 parameter, this parameter determines the granularity of the callback | |
| 494 by defining the maximum number of times the callback will be called | |
| 495 during the file transfer. Providing a negative integer will cause | |
| 496 your callback to be called with each buffer read. | |
| 497 | |
| 498 Raises ResumableUploadException if a problem occurs during the transfer. | |
| 499 """ | |
| 500 | |
| 501 if not headers: | |
| 502 headers = {} | |
| 503 | |
| 504 fp.seek(0, os.SEEK_END) | |
| 505 file_length = fp.tell() | |
| 506 fp.seek(0) | |
| 507 debug = key.bucket.connection.debug | |
| 508 | |
| 509 # Use num-retries from constructor if one was provided; else check | |
| 510 # for a value specified in the boto config file; else default to 5. | |
| 511 if self.num_retries is None: | |
| 512 self.num_retries = config.getint('Boto', 'num_retries', 5) | |
| 513 progress_less_iterations = 0 | |
| 514 | |
| 515 while True: # Retry as long as we're making progress. | |
| 516 server_had_bytes_before_attempt = self.server_has_bytes | |
| 517 try: | |
| 518 etag = self._attempt_resumable_upload(key, fp, file_length, | |
| 519 headers, cb, num_cb) | |
| 520 # Upload succceded, so remove the tracker file (if have one). | |
| 521 self._remove_tracker_file() | |
| 522 self._check_final_md5(key, etag) | |
| 523 if debug >= 1: | |
| 524 print 'Resumable upload complete.' | |
| 525 return | |
| 526 except self.RETRYABLE_EXCEPTIONS, e: | |
| 527 if debug >= 1: | |
| 528 print('Caught exception (%s)' % e.__repr__()) | |
| 529 if isinstance(e, IOError) and e.errno == errno.EPIPE: | |
| 530 # Broken pipe error causes httplib to immediately | |
| 531 # close the socket (http://bugs.python.org/issue5542), | |
| 532 # so we need to close the connection before we resume | |
| 533 # the upload (which will cause a new connection to be | |
| 534 # opened the next time an HTTP request is sent). | |
| 535 key.bucket.connection.connection.close() | |
| 536 except ResumableUploadException, e: | |
| 537 if (e.disposition == | |
| 538 ResumableTransferDisposition.ABORT_CUR_PROCESS): | |
| 539 if debug >= 1: | |
| 540 print('Caught non-retryable ResumableUploadException ' | |
| 541 '(%s); aborting but retaining tracker file' % | |
| 542 e.message) | |
| 543 raise | |
| 544 elif (e.disposition == | |
| 545 ResumableTransferDisposition.ABORT): | |
| 546 if debug >= 1: | |
| 547 print('Caught non-retryable ResumableUploadException ' | |
| 548 '(%s); aborting and removing tracker file' % | |
| 549 e.message) | |
| 550 self._remove_tracker_file() | |
| 551 raise | |
| 552 else: | |
| 553 if debug >= 1: | |
| 554 print('Caught ResumableUploadException (%s) - will ' | |
| 555 'retry' % e.message) | |
| 556 | |
| 557 # At this point we had a re-tryable failure; see if made progress. | |
| 558 if self.server_has_bytes > server_had_bytes_before_attempt: | |
| 559 progress_less_iterations = 0 | |
| 560 else: | |
| 561 progress_less_iterations += 1 | |
| 562 | |
| 563 if progress_less_iterations > self.num_retries: | |
| 564 # Don't retry any longer in the current process. | |
| 565 raise ResumableUploadException( | |
| 566 'Too many resumable upload attempts failed without ' | |
| 567 'progress. You might try this upload again later', | |
| 568 ResumableTransferDisposition.ABORT_CUR_PROCESS) | |
| 569 | |
| 570 sleep_time_secs = 2**progress_less_iterations | |
| 571 if debug >= 1: | |
| 572 print ('Got retryable failure (%d progress-less in a row).\n' | |
| 573 'Sleeping %d seconds before re-trying' % | |
| 574 (progress_less_iterations, sleep_time_secs)) | |
| 575 time.sleep(sleep_time_secs) | |
| OLD | NEW |
|---|
Chromium Code Reviews
Issue 10199002:
Upgrade gsutil to 3.4 (Closed)
Base URL: https://dart.googlecode.com/svn/branches/bleeding_edge/dart