Object server PUTs should respect client_timeout

It seems the object server never respected the client_timeout value
since the beginning of Swift. This is normally fine since the proxy does
and will normally hang up on the backends. But if the proxy has a bug or
if there's network issues or whatever, the object server should be smart
enough to enforce this timeout as well.

Our operations guys noticed this problem when older processes would
never exit after a reload. They started investigating and saw that the
object server had open tmp files that hadn't been touched in quite some
time. Sometimes the tmp files didn't even exist anymore since the
reclaimer deletes really old untouched tmp files.

Change-Id: Iba0397203a2dccca4a28a8c8cbfc5a60e429837a

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):