Merge "Object server PUTs should respect client_timeout"

swift/obj/server.py

@@ -35,7 +35,7 @@ from swift.common.constraints import check_object_creation, \
     check_float, check_utf8
 from swift.common.exceptions import ConnectionTimeout, DiskFileQuarantined, \
     DiskFileNotExist, DiskFileCollision, DiskFileNoSpace, DiskFileDeleted, \
-    DiskFileDeviceUnavailable, DiskFileExpired
+    DiskFileDeviceUnavailable, DiskFileExpired, ChunkReadTimeout
 from swift.obj import ssync_receiver
 from swift.common.http import is_success
 from swift.common.request_helpers import split_and_validate_path, is_user_meta
@@ -398,15 +398,23 @@ class ObjectController(object):
         try:
             with disk_file.create(size=fsize) as writer:
                 upload_size = 0
-                reader = request.environ['wsgi.input'].read
-                for chunk in iter(lambda: reader(self.network_chunk_size), ''):
-                    start_time = time.time()
-                    if start_time > upload_expiration:
-                        self.logger.increment('PUT.timeouts')
-                        return HTTPRequestTimeout(request=request)
-                    etag.update(chunk)
-                    upload_size = writer.write(chunk)
-                    elapsed_time += time.time() - start_time
+
+                def timeout_reader():
+                    with ChunkReadTimeout(self.client_timeout):
+                        return request.environ['wsgi.input'].read(
+                            self.network_chunk_size)
+
+                try:
+                    for chunk in iter(lambda: timeout_reader(), ''):
+                        start_time = time.time()
+                        if start_time > upload_expiration:
+                            self.logger.increment('PUT.timeouts')
+                            return HTTPRequestTimeout(request=request)
+                        etag.update(chunk)
+                        upload_size = writer.write(chunk)
+                        elapsed_time += time.time() - start_time
+                except ChunkReadTimeout:
+                    return HTTPRequestTimeout(request=request)
                 if upload_size:
                     self.logger.transfer_rate(
                         'PUT.' + device + '.timing', elapsed_time,

test/unit/obj/test_server.py

@@ -639,6 +639,28 @@ class TestObjectController(unittest.TestCase):
                            'X-Object-Meta-1': 'One',
                            'X-Object-Meta-Two': 'Two'})
 
+    def test_PUT_client_timeout(self):
+        class FakeTimeout(BaseException):
+            def __enter__(self):
+                raise self
+
+            def __exit__(self, typ, value, tb):
+                pass
+        # This is just so the test fails when run on older object server code
+        # instead of exploding.
+        if not hasattr(object_server, 'ChunkReadTimeout'):
+            object_server.ChunkReadTimeout = None
+        with mock.patch.object(object_server, 'ChunkReadTimeout', FakeTimeout):
+            timestamp = normalize_timestamp(time())
+            req = Request.blank(
+                '/sda1/p/a/c/o', environ={'REQUEST_METHOD': 'PUT'},
+                headers={'X-Timestamp': timestamp,
+                         'Content-Type': 'text/plain',
+                         'Content-Length': '6'})
+            req.environ['wsgi.input'] = StringIO('VERIFY')
+            resp = req.get_response(self.object_controller)
+            self.assertEquals(resp.status_int, 408)
+
     def test_PUT_container_connection(self):
 
         def mock_http_connect(response, with_exc=False):