Merge remote-tracking branch 'upstream/development' into secure_proxy

# Conflicts:
#	hyper/http20/connection.py

Author: KostyaEsmukov

hyper/http11/connection.py

@@ -451,7 +451,8 @@ def close(self):
         .. warning:: This method should absolutely only be called when you are
                      certain the connection object is no longer needed.
         """
-        self._sock.close()
+        if self._sock is not None:
+            self._sock.close()
         self._sock = None
 
     # The following two methods are the implementation of the context manager

hyper/http20/connection.py

@@ -143,36 +143,9 @@ def __init__(self, host, port=None, secure=None, window_manager=None,
 
         # Concurrency
         #
-        # Use one lock (_lock) to synchronize any interaction with global
-        # connection state, e.g. stream creation/deletion.
-        #
-        # It's ok to use the same in lock all these cases as they occur at
-        # different/linked points in the connection's lifecycle.
-        #
-        # Use another 2 locks (_write_lock, _read_lock) to synchronize
-        # - _send_cb
-        # - _recv_cb
-        # respectively.
-        #
-        # I.e, send/recieve on the connection and its streams are serialized
-        # separately across the threads accessing the connection.  This is a
-        # simple way of providing thread-safety.
-        #
-        # _write_lock and _read_lock synchronize all interactions between
-        # streams and the connnection.  There is a third I/O callback,
-        # _close_stream, passed to a stream's constructor.  It does not need to
-        # be synchronized, it uses _send_cb internally (which is serialized);
-        # its other activity (safe deletion of the stream from self.streams)
-        # does not require synchronization.
-        #
-        # _read_lock may be acquired when already holding the _write_lock,
-        # when they both held it is always by acquiring _write_lock first.
-        #
-        # Either _read_lock or _write_lock may be acquired whilst holding _lock
-        # which should always be acquired before either of the other two.
+        # Use one universal lock (_lock) to synchronize all interaction
+        # with global connection state, _send_cb and _recv_cb.
         self._lock = threading.RLock()
-        self._write_lock = threading.RLock()
-        self._read_lock = threading.RLock()
 
         # Create the mutable state.
         self.__wm_class = window_manager or FlowControlManager
@@ -236,7 +209,7 @@ def ping(self, opaque_data):
         :returns: Nothing
         """
         self.connect()
-        with self._write_lock:
+        with self._lock:
             with self._conn as conn:
                 conn.ping(to_bytestring(opaque_data))
             self._send_outstanding_data()
@@ -275,7 +248,7 @@ def request(self, method, url, body=None, headers=None):
         # If threads interleave these operations, it could result in messages
         # being sent in the wrong order, which can lead to the out-of-order
         # messages with lower stream IDs being closed prematurely.
-        with self._write_lock:
+        with self._lock:
             # Unlike HTTP/1.1, HTTP/2 (according to RFC 7540) doesn't require
             # to use absolute URI when proxying.
 
@@ -483,10 +456,10 @@ def _send_outstanding_data(self, tolerate_peer_gone=False,
                                send_empty=True):
         # Concurrency
         #
-        # Hold _write_lock; getting and writing data from _conn is synchronized
+        # Hold _lock; getting and writing data from _conn is synchronized
         #
         # I/O occurs while the lock is held; waiting threads will see a delay.
-        with self._write_lock:
+        with self._lock:
             with self._conn as conn:
                 data = conn.data_to_send()
             if data or send_empty:
@@ -576,9 +549,9 @@ def endheaders(self, message_body=None, final=False, stream_id=None):
 
         # Concurrency:
         #
-        # Hold _write_lock: synchronize access to the connection's HPACK
+        # Hold _lock: synchronize access to the connection's HPACK
         # encoder and decoder and the subsquent write to the connection
-        with self._write_lock:
+        with self._lock:
             stream.send_headers(headers_only)
 
             # Send whatever data we have.
@@ -641,10 +614,10 @@ def _send_cb(self, data, tolerate_peer_gone=False):
         """
         # Concurrency
         #
-        # Hold _write_lock: ensures only writer at a time
+        # Hold _lock: ensures only writer at a time
         #
         # I/O occurs while the lock is held; waiting threads will see a delay.
-        with self._write_lock:
+        with self._lock:
             try:
                 self._sock.sendall(data)
             except socket.error as e:
@@ -659,12 +632,12 @@ def _adjust_receive_window(self, frame_len):
         """
         # Concurrency
         #
-        # Hold _write_lock; synchronize the window manager update and the
+        # Hold _lock; synchronize the window manager update and the
         # subsequent potential write to the connection
         #
         # I/O may occur while the lock is held; waiting threads may see a
         # delay.
-        with self._write_lock:
+        with self._lock:
             increment = self.window_manager._handle_frame(frame_len)
 
             if increment:
@@ -686,7 +659,7 @@ def _single_read(self):
         # Synchronizes reading the data
         #
         # I/O occurs while the lock is held; waiting threads will see a delay.
-        with self._read_lock:
+        with self._lock:
             if self._sock is None:
                 raise ConnectionError('tried to read after connection close')
             self._sock.fill()
@@ -780,7 +753,7 @@ def _recv_cb(self, stream_id=0):
         # 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:
+        with self._lock:
             log.debug('recv for stream %d with %s already present',
                       stream_id,
                       self.recent_recv_streams)
@@ -831,11 +804,11 @@ def _send_rst_frame(self, stream_id, error_code):
         """
         # Concurrency
         #
-        # Hold _write_lock; synchronize generating the reset frame and writing
+        # Hold _lock; synchronize generating the reset frame and writing
         # it
         #
         # I/O occurs while the lock is held; waiting threads will see a delay.
-        with self._write_lock:
+        with self._lock:
             with self._conn as conn:
                 conn.reset_stream(stream_id, error_code=error_code)
             self._send_outstanding_data()

test/test_http11.py

@@ -588,6 +588,10 @@ def read(self, size):
         assert 'File-like bodies must return bytestrings. ' \
                'Got: {}'.format(int) in str(exc_info)
 
+    def test_close_with_uninitialized_socket(self):
+        c = HTTP11Connection('httpbin.org')
+        c.close()
+
 
 class TestHTTP11Response(object):
     def test_short_circuit_read(self):

test/test_integration.py

@@ -15,6 +15,7 @@
 import pytest
 from contextlib import contextmanager
 from mock import patch
+from concurrent.futures import ThreadPoolExecutor, TimeoutError
 from h2.frame_buffer import FrameBuffer
 from hyper.compat import ssl
 from hyper.contrib import HTTP20Adapter
@@ -1142,7 +1143,6 @@ def test_version_after_http_upgrade(self):
 
         def socket_handler(listener):
             sock = listener.accept()[0]
-
             # We should get the initial request.
             data = b''
             while not data.endswith(b'\r\n\r\n'):
@@ -1192,6 +1192,44 @@ def socket_handler(listener):
 
         self.tear_down()
 
+    def test_connection_and_send_simultaneously(self):
+        # Since deadlock occurs probabilistic,
+        # It still has deadlock probability
+        # even the testcase is passed.
+        self.set_up()
+
+        recv_event = threading.Event()
+
+        def socket_handler(listener):
+            sock = listener.accept()[0]
+
+            receive_preamble(sock)
+            sock.recv(65535)
+
+            recv_event.set()
+            sock.close()
+
+        def do_req(conn):
+            conn.request('GET', '/')
+            recv_event.wait()
+
+        def do_connect(conn):
+            conn.connect()
+
+        self._start_server(socket_handler)
+        conn = self.get_connection()
+
+        pool = ThreadPoolExecutor(max_workers=2)
+        pool.submit(do_connect, conn)
+        f = pool.submit(do_req, conn)
+
+        try:
+            f.result(timeout=10)
+        except TimeoutError:
+            assert False
+
+        self.tear_down()
+
 
 @patch('hyper.http20.connection.H2_NPN_PROTOCOLS', PROTOCOLS)
 class TestRequestsAdapter(SocketLevelTest):

test_requirements.txt

@@ -1,4 +1,4 @@
-pytest
+pytest>=2.7
 pytest-xdist
 pytest-cov
 requests