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