Correct a locking gap

- ensure that the read lock is held throughout the _recv_cb

Author: Tim Emiola

hyper/http20/connection.py

@@ -678,7 +678,13 @@ def _recv_cb(self, stream_id=0):
         # Concurrency
         #
         # Ignore this read if some other thread has recently read data from
-        # from the requested stream
+        # from the requested stream.
+        #
+        # The lock here looks broad, but is need to ensure correct behavior
+        # when there are multiple readers of the same stream.  It re-acquired
+        # in the calls to self._single_read.
+        #
+        # i/o occurs while the lock is held; waiting threads will see a delay.
         with self._read_lock:
             log.debug('recv for stream %d with %s already present',
                       stream_id,
@@ -687,36 +693,36 @@ def _recv_cb(self, stream_id=0):
                 self.recent_recv_streams.discard(stream_id)
                 return
 
-        # TODO: Re-evaluate this.
-        self._single_read()
-        count = 9
-        retry_wait = 0.05  # can improve responsiveness to delay the retry
-
-        while count and self._sock is not None and self._sock.can_read:
-            # If the connection has been closed, bail out, but retry
-            # on transient errors.
-            try:
-                self._single_read()
-            except ConnectionResetError:
-                break
-            except ssl.SSLError as e:  # pragma: no cover
-                # these are transient errors that can occur while reading from
-                # ssl connections.
-                if e.args[0] in TRANSIENT_SSL_ERRORS:
-                    continue
-                else:
-                    raise
-            except socket.error as e:  # pragma: no cover
-                if e.errno in (errno.EINTR, errno.EAGAIN):
-                    # if 'interrupted' or 'try again', continue
-                    time.sleep(retry_wait)
-                    continue
-                elif e.errno == errno.ECONNRESET:
+            # TODO: Re-evaluate this.
+            self._single_read()
+            count = 9
+            retry_wait = 0.05  # can improve responsiveness to delay the retry
+
+            while count and self._sock is not None and self._sock.can_read:
+                # If the connection has been closed, bail out, but retry
+                # on transient errors.
+                try:
+                    self._single_read()
+                except ConnectionResetError:
                     break
-                else:
-                    raise
+                except ssl.SSLError as e:  # pragma: no cover
+                    # these are transient errors that can occur while reading
+                    # from ssl connections.
+                    if e.args[0] in TRANSIENT_SSL_ERRORS:
+                        continue
+                    else:
+                        raise
+                except socket.error as e:  # pragma: no cover
+                    if e.errno in (errno.EINTR, errno.EAGAIN):
+                        # if 'interrupted' or 'try again', continue
+                        time.sleep(retry_wait)
+                        continue
+                    elif e.errno == errno.ECONNRESET:
+                        break
+                    else:
+                        raise
 
-            count -= 1
+                count -= 1
 
     def _send_rst_frame(self, stream_id, error_code):
         """
@@ -734,7 +740,7 @@ def _send_rst_frame(self, stream_id, error_code):
 
         # Concurrency
         #
-        # Hold _lock; the stram storage is being update. No i/o occurs, any
+        # Hold _lock; the stream storage is being updated. No i/o occurs, any
         # delay is proportional to the number of waiting threads.
         with self._lock:
             try: