proxy: stop sending chunks to objects with a Queue

During a PUT of an object, the proxy instanciates one Putter per
object-server that will store data (either the full object or a
fragment, depending on the storage policy). Each Putter is owning a
Queue that will be used to bufferize data chunks before they are
written to the socket connected to the object-server. The chunks are
moved from the queue to the socket by a greenthread. There is one
greenthread per Putter. If the client is uploading faster than the
object-servers can manage, the Queue could grow and consume a lot of
memory. To avoid that, the queue is bounded (default: 10). Having a
bounded queue also allows to ensure that all object-servers will get
the data at the same rate because if one queue is full, the
greenthread reading from the client socket will block when trying to
write to the queue. So the global rate is the one of the slowest
object-server.

The thing is, every operating system manages socket buffers for incoming
and outgoing data. Concerning the send buffer, the behavior is such that
if the buffer is full, a call to write() will block, otherwise the call
will return immediately. It behaves a lot like the Putter's Queue,
except that the size of the buffer is dynamic so it adapts itself to the
speed of the receiver.

Thus, managing a queue in addition to the socket send buffer is a
duplicate queueing/buffering that provides no interest but is, as shown
by profiling and benchmarks, very CPU costly.

This patch removes the queuing mecanism. Instead, the greenthread
reading data from the client will directly write to the socket. If an
object-server is getting slow, the buffer will fulfill, blocking the
reader greenthread. Benchmark shows a CPU consumption reduction of more
than 30% will the observed rate for an upload is increasing by about
45%.

Change-Id: Icf8f800cb25096f93d3faa1e6ec091eb29500758

doc/manpages/proxy-server.conf.5

@@ -1061,8 +1061,6 @@ recheck_account_existence before the 403s kick in.
 This is a comma separated list of account hashes that ignore the max_containers_per_account cap.
 .IP \fBdeny_host_headers\fR
 Comma separated list of Host headers to which the proxy will deny requests. The default is empty.
-.IP \fBput_queue_depth\fR
-Depth of the proxy put queue. The default is 10.
 .IP \fBsorting_method\fR
 Storage nodes can be chosen at random (shuffle - default), by using timing
 measurements (timing), or by using an explicit match (affinity).

etc/proxy-server.conf-sample

@@ -188,9 +188,6 @@ use = egg:swift#proxy
 # Prefix used when automatically creating accounts.
 # auto_create_account_prefix = .
 #
-# Depth of the proxy put queue.
-# put_queue_depth = 10
-#
 # During GET and HEAD requests, storage nodes can be chosen at random
 # (shuffle), by using timing measurements (timing), or by using an explicit
 # region/zone match (affinity). Using timing measurements may allow for lower

swift/proxy/controllers/obj.py

@@ -876,6 +876,8 @@ class ReplicatedObjectController(BaseObjectController):
                 node, part, req.swift_entity_path, headers,
                 conn_timeout=self.app.conn_timeout,
                 node_timeout=self.app.node_timeout,
+                write_timeout=self.app.node_timeout,
+                send_exception_handler=self.app.exception_occurred,
                 logger=self.app.logger,
                 need_multiphase=False)
         else:
@@ -884,6 +886,8 @@ class ReplicatedObjectController(BaseObjectController):
                 node, part, req.swift_entity_path, headers,
                 conn_timeout=self.app.conn_timeout,
                 node_timeout=self.app.node_timeout,
+                write_timeout=self.app.node_timeout,
+                send_exception_handler=self.app.exception_occurred,
                 logger=self.app.logger,
                 chunked=te.endswith(',chunked'))
         return putter
@@ -910,11 +914,6 @@ class ReplicatedObjectController(BaseObjectController):
 
         min_conns = quorum_size(len(nodes))
         try:
-            with ContextPool(len(nodes)) as pool:
-                for putter in putters:
-                    putter.spawn_sender_greenthread(
-                        pool, self.app.put_queue_depth, self.app.node_timeout,
-                        self.app.exception_occurred)
             while True:
                 with ChunkReadTimeout(self.app.client_timeout):
                     try:
@@ -940,8 +939,6 @@ class ReplicatedObjectController(BaseObjectController):
                 # send any footers set by middleware
                 putter.end_of_object_data(footer_metadata=trail_md)
 
-                for putter in putters:
-                    putter.wait()
             self._check_min_conn(
                 req, [p for p in putters if not p.failed], min_conns,
                 msg=_('Object PUT exceptions after last send, '
@@ -1576,10 +1573,14 @@ class Putter(object):
     :param resp: an HTTPResponse instance if connect() received final response
     :param path: the object path to send to the storage node
     :param connect_duration: time taken to initiate the HTTPConnection
+    :param write_timeout: time limit to write a chunk to the connection socket
+    :param send_exception_handler: callback called when an exception occured
+                                   writing to the connection socket
     :param logger: a Logger instance
     :param chunked: boolean indicating if the request encoding is chunked
     """
-    def __init__(self, conn, node, resp, path, connect_duration, logger,
+    def __init__(self, conn, node, resp, path, connect_duration,
+                 write_timeout, send_exception_handler, logger,
                  chunked=False):
         # Note: you probably want to call Putter.connect() instead of
         # instantiating one of these directly.
@@ -1588,11 +1589,12 @@ class Putter(object):
         self.resp = self.final_resp = resp
         self.path = path
         self.connect_duration = connect_duration
+        self.write_timeout = write_timeout
+        self.send_exception_handler = send_exception_handler
         # for handoff nodes node_index is None
         self.node_index = node.get('index')
 
         self.failed = False
-        self.queue = None
         self.state = NO_DATA_SENT
         self.chunked = chunked
         self.logger = logger
@@ -1624,16 +1626,6 @@ class Putter(object):
                     self.resp = self.conn.getresponse()
         return self.resp
 
-    def spawn_sender_greenthread(self, pool, queue_depth, write_timeout,
-                                 exception_handler):
-        """Call before sending the first chunk of request body"""
-        self.queue = Queue(queue_depth)
-        pool.spawn(self._send_file, write_timeout, exception_handler)
-
-    def wait(self):
-        if self.queue.unfinished_tasks:
-            self.queue.join()
-
     def _start_object_data(self):
         # Called immediately before the first chunk of object data is sent.
         # Subclasses may implement custom behaviour
@@ -1653,7 +1645,7 @@ class Putter(object):
             self._start_object_data()
             self.state = SENDING_DATA
 
-        self.queue.put(chunk)
+        self._send_chunk(chunk)
 
     def end_of_object_data(self, **kwargs):
         """
@@ -1662,33 +1654,23 @@ class Putter(object):
         if self.state == DATA_SENT:
             raise ValueError("called end_of_object_data twice")
 
-        self.queue.put(b'')
+        self._send_chunk(b'')
         self.state = DATA_SENT
 
-    def _send_file(self, write_timeout, exception_handler):
+    def _send_chunk(self, chunk):
-        """
-        Method for a file PUT coroutine. Takes chunks from a queue and sends
-        them down a socket.
-
-        If something goes wrong, the "failed" attribute will be set to true
-        and the exception handler will be called.
-        """
-        while True:
-            chunk = self.queue.get()
         if not self.failed:
             if self.chunked:
                 to_send = b"%x\r\n%s\r\n" % (len(chunk), chunk)
             else:
                 to_send = chunk
             try:
-                    with ChunkWriteTimeout(write_timeout):
+                with ChunkWriteTimeout(self.write_timeout):
                     self.conn.send(to_send)
             except (Exception, ChunkWriteTimeout):
                 self.failed = True
-                    exception_handler(self.node, _('Object'),
+                self.send_exception_handler(self.node, _('Object'),
-                                      _('Trying to write to %s') % self.path)
+                                            _('Trying to write to %s')
-
+                                            % self.path)
-            self.queue.task_done()
 
     def close(self):
         # release reference to response to ensure connection really does close,
@@ -1725,7 +1707,8 @@ class Putter(object):
 
     @classmethod
     def connect(cls, node, part, path, headers, conn_timeout, node_timeout,
-                logger=None, chunked=False, **kwargs):
+                write_timeout, send_exception_handler, logger=None,
+                chunked=False, **kwargs):
         """
         Connect to a backend node and send the headers.
 
@@ -1738,7 +1721,8 @@ class Putter(object):
         """
         conn, expect_resp, final_resp, connect_duration = cls._make_connection(
             node, part, path, headers, conn_timeout, node_timeout)
-        return cls(conn, node, final_resp, path, connect_duration, logger,
+        return cls(conn, node, final_resp, path, connect_duration,
+                   write_timeout, send_exception_handler, logger,
                    chunked=chunked)
 
 
@@ -1753,9 +1737,11 @@ class MIMEPutter(Putter):
     An HTTP PUT request that supports streaming.
     """
     def __init__(self, conn, node, resp, req, connect_duration,
-                 logger, mime_boundary, multiphase=False):
+                 write_timeout, send_exception_handler, logger, mime_boundary,
+                 multiphase=False):
         super(MIMEPutter, self).__init__(conn, node, resp, req,
-                                         connect_duration, logger)
+                                         connect_duration, write_timeout,
+                                         send_exception_handler, logger)
         # Note: you probably want to call MimePutter.connect() instead of
         # instantiating one of these directly.
         self.chunked = True  # MIME requests always send chunked body
@@ -1766,7 +1752,7 @@ class MIMEPutter(Putter):
         # We're sending the object plus other stuff in the same request
         # body, all wrapped up in multipart MIME, so we'd better start
         # off the MIME document before sending any object data.
-        self.queue.put(b"--%s\r\nX-Document: object body\r\n\r\n" %
+        self._send_chunk(b"--%s\r\nX-Document: object body\r\n\r\n" %
                          (self.mime_boundary,))
 
     def end_of_object_data(self, footer_metadata=None):
@@ -1800,9 +1786,9 @@ class MIMEPutter(Putter):
             footer_body, b"\r\n",
             tail_boundary, b"\r\n",
         ]
-        self.queue.put(b"".join(message_parts))
+        self._send_chunk(b"".join(message_parts))
 
-        self.queue.put(b'')
+        self._send_chunk(b'')
         self.state = DATA_SENT
 
     def send_commit_confirmation(self):
@@ -1827,14 +1813,15 @@ class MIMEPutter(Putter):
                 body, b"\r\n",
                 tail_boundary,
             ]
-            self.queue.put(b"".join(message_parts))
+            self._send_chunk(b"".join(message_parts))
 
-        self.queue.put(b'')
+        self._send_chunk(b'')
         self.state = COMMIT_SENT
 
     @classmethod
     def connect(cls, node, part, req, headers, conn_timeout, node_timeout,
-                logger=None, need_multiphase=True, **kwargs):
+                write_timeout, send_exception_handler, logger=None,
+                need_multiphase=True, **kwargs):
         """
         Connect to a backend node and send the headers.
 
@@ -1886,7 +1873,8 @@ class MIMEPutter(Putter):
             if need_multiphase and not can_handle_multiphase_put:
                 raise MultiphasePUTNotSupported()
 
-        return cls(conn, node, final_resp, req, connect_duration, logger,
+        return cls(conn, node, final_resp, req, connect_duration,
+                   write_timeout, send_exception_handler, logger,
                    mime_boundary, multiphase=need_multiphase)
 
 
@@ -2499,6 +2487,8 @@ class ECObjectController(BaseObjectController):
             node, part, req.swift_entity_path, headers,
             conn_timeout=self.app.conn_timeout,
             node_timeout=self.app.node_timeout,
+            write_timeout=self.app.node_timeout,
+            send_exception_handler=self.app.exception_occurred,
             logger=self.app.logger,
             need_multiphase=True)
 
@@ -2615,17 +2605,11 @@ class ECObjectController(BaseObjectController):
                       '%(conns)s/%(nodes)s required connections'))
 
         try:
-            with ContextPool(len(putters)) as pool:
-
             # build our putter_to_frag_index dict to place handoffs in the
             # same part nodes index as the primaries they are covering
             putter_to_frag_index = self._determine_chunk_destinations(
                 putters, policy)
 
-                for putter in putters:
-                    putter.spawn_sender_greenthread(
-                        pool, self.app.put_queue_depth, self.app.node_timeout,
-                        self.app.exception_occurred)
             while True:
                 with ChunkReadTimeout(self.app.client_timeout):
                     try:
@@ -2671,9 +2655,6 @@ class ECObjectController(BaseObjectController):
                 trail_md['Etag'] = frag_hashers[frag_index].hexdigest()
                 putter.end_of_object_data(footer_metadata=trail_md)
 
-                for putter in putters:
-                    putter.wait()
-
             # for storage policies requiring 2-phase commit (e.g.
             # erasure coding), enforce >= 'quorum' number of
             # 100-continue responses - this indicates successful
@@ -2713,8 +2694,6 @@ class ECObjectController(BaseObjectController):
             # a successful PUT
             for putter in putters:
                 putter.send_commit_confirmation()
-                for putter in putters:
-                    putter.wait()
         except ChunkReadTimeout as err:
             self.app.logger.warning(
                 _('ERROR Client read timeout (%ss)'), err.seconds)

swift/proxy/server.py

@@ -191,7 +191,6 @@ class Application(object):
             conf.get('recoverable_node_timeout', self.node_timeout))
         self.conn_timeout = float(conf.get('conn_timeout', 0.5))
         self.client_timeout = int(conf.get('client_timeout', 60))
-        self.put_queue_depth = int(conf.get('put_queue_depth', 10))
         self.object_chunk_size = int(conf.get('object_chunk_size', 65536))
         self.client_chunk_size = int(conf.get('client_chunk_size', 65536))
         self.trans_id_suffix = conf.get('trans_id_suffix', '')