Dynamic chunk size (#122)

For chunked uploads, choose chunk size based on

* `_RESPONSE_TIME_MAX`
* `x-response-time`

Author: juanshishido

tests/test_utils.py

@@ -1,7 +1,8 @@
 import datetime
+from random import randint
 
 from twitter_ads.enum import GRANULARITY
-from twitter_ads.utils import to_time
+from twitter_ads.utils import size, to_time
 
 
 t = datetime.datetime(2006, 3, 21, 0, 0, 0)
@@ -10,3 +11,18 @@ def test_to_time_based_on_granularity():
     for g in [None, GRANULARITY.HOUR, GRANULARITY.TOTAL]:
         assert to_time(t, g) == '2006-03-21T00:00:00Z'
     assert to_time(t, GRANULARITY.DAY) == '2006-03-21'
+
+def test_sizes():
+    _DEFAULT_CHUNK_SIZE = 64
+    _RESPONSE_TIME_MAX = 5000
+    for _ in range(10):
+        response_time = randint(0, _RESPONSE_TIME_MAX)
+        assert size(_DEFAULT_CHUNK_SIZE, _RESPONSE_TIME_MAX, response_time) == _DEFAULT_CHUNK_SIZE
+    response_times = {10000 : 32,
+                      20000 : 16,
+                      40000 : 8,
+                      80000 : 4,
+                      160000 : 2,
+                      320000 : 1}
+    for rt in response_times:
+        assert size(_DEFAULT_CHUNK_SIZE, _RESPONSE_TIME_MAX, rt) == response_times[rt]

twitter_ads/http.py

@@ -19,7 +19,7 @@
 
 from requests_oauthlib import OAuth1Session
 
-from twitter_ads.utils import get_version, http_time
+from twitter_ads.utils import get_version, http_time, size
 from twitter_ads.error import Error
 
 
@@ -195,7 +195,9 @@ class TONUpload(object):
     _DEFAULT_RESOURCE = '/1.1/ton/bucket/'
     _DEFAULT_BUCKET = 'ta_partner'
     _DEFAULT_EXPIRE = datetime.now() + timedelta(days=10)
-    _MIN_FILE_SIZE = 1024 * 1024 * 64
+    _DEFAULT_CHUNK_SIZE = 64
+    _SINGLE_UPLOAD_MAX = 1024 * 1024 * _DEFAULT_CHUNK_SIZE
+    _RESPONSE_TIME_MAX = 5000
 
     def __init__(self, client, file_path, **kwargs):
         if not os.path.isfile(file_path):
@@ -240,13 +242,14 @@ def content_type(self):
     def perform(self):
         """Executes the current TONUpload object."""
 
-        if self._file_size < self._MIN_FILE_SIZE:
+        if self._file_size < self._SINGLE_UPLOAD_MAX:
             resource = "{0}{1}".format(self._DEFAULT_RESOURCE, self.bucket)
             response = self.__upload(resource, open(self._file_path, 'rb').read())
             return response.headers['location']
         else:
             response = self.__init_chunked_upload()
-            chunk_size = int(response.headers['x-ton-min-chunk-size'])
+            min_chunk_size = int(response.headers['x-ton-min-chunk-size'])
+            chunk_size = min_chunk_size * self._DEFAULT_CHUNK_SIZE
             location = response.headers['location']
 
             f = open(self._file_path, 'rb')
@@ -257,7 +260,11 @@ def perform(self):
                     break
                 bytes_start = bytes_read
                 bytes_read += len(bytes)
-                self.__upload_chunk(location, chunk_size, bytes, bytes_start, bytes_read)
+                response = self.__upload_chunk(location, chunk_size, bytes, bytes_start, bytes_read)
+                response_time = int(response.headers['x-response-time'])
+                chunk_size = min_chunk_size * size(self._DEFAULT_CHUNK_SIZE,
+                                                   self._RESPONSE_TIME_MAX,
+                                                   response_time)
             f.close()
 
             return location.split("?")[0]

twitter_ads/utils.py

@@ -1,4 +1,5 @@
 # Copyright (C) 2015 Twitter, Inc.
+from __future__ import division
 
 """Container for all helpers and utilities used throughout the Ads API SDK."""
 
@@ -45,3 +46,12 @@ def format_date(time):
 def http_time(time):
     """Formats a datetime as an RFC 1123 compliant string."""
     return formatdate(timeval=mktime(time.timetuple()), localtime=False, usegmt=True)
+
+
+def size(default_chunk_size, response_time_max, response_time_actual):
+    """Determines the chunk size based on response times."""
+    if response_time_actual == 0:
+        response_time_actual = 1
+    scale = 1 / (response_time_actual / response_time_max)
+    size = int(default_chunk_size * scale)
+    return min(max(size, 1), default_chunk_size)