Repository: tornadoweb/tornado
Pull request #1935 (2017-01-25 11:48:03)

TCPClient: connect using specific source IP address and port
Hello,

I'm opening this PoC to start discussing about a potential solution for #1931.

Please let me know what you think!
Thanks,
Mircea


Patch:
diff --git a/docs/locks.rst b/docs/locks.rst
index 94184db2..7115b0da 100644
--- a/docs/locks.rst
+++ b/docs/locks.rst
@@ -6,12 +6,10 @@
 Coordinate coroutines with synchronization primitives analogous to those the
 standard library provides to threads.
 
-.. warning::
-
-   Note that these primitives are not actually thread-safe and cannot be used in
-   place of those from the standard library--they are meant to coordinate Tornado
-   coroutines in a single-threaded app, not to protect shared objects in a
-   multithreaded app.
+*(Note that these primitives are not actually thread-safe and cannot be used in
+place of those from the standard library--they are meant to coordinate Tornado
+coroutines in a single-threaded app, not to protect shared objects in a
+multithreaded app.)*
 
 .. automodule:: tornado.locks
 
diff --git a/tornado/queues.py b/tornado/queues.py
index 77eda98b..b8e9b569 100644
--- a/tornado/queues.py
+++ b/tornado/queues.py
@@ -12,16 +12,6 @@
 # License for the specific language governing permissions and limitations
 # under the License.
 
-"""Asynchronous queues for coroutines.
-
-.. warning::
-
-   Unlike the standard library's `queue` module, the classes defined here
-   are *not* thread-safe. To use these queues from another thread,
-   use `.IOLoop.add_callback` to transfer control to the `.IOLoop` thread
-   before calling any queue methods.
-"""
-
 from __future__ import absolute_import, division, print_function, with_statement
 
 import collections
diff --git a/tornado/tcpclient.py b/tornado/tcpclient.py
index 11146860..1e289174 100644
--- a/tornado/tcpclient.py
+++ b/tornado/tcpclient.py
@@ -155,16 +155,27 @@ class TCPClient(object):
 
     @gen.coroutine
     def connect(self, host, port, af=socket.AF_UNSPEC, ssl_options=None,
-                max_buffer_size=None):
+                max_buffer_size=None, source_ip=None, source_port=None):
         """Connect to the given host and port.
 
         Asynchronously returns an `.IOStream` (or `.SSLIOStream` if
         ``ssl_options`` is not None).
+
+        Using the ``source_ip`` kwarg, one can specify the source
+        IP address to use when establishing the connection.
+        In case the user needs to resolve and
+        use a specific interface, it has to be handled outside
+        of Tornado as this depends very much on the platform.
+
+        Similarly, when the user requires a certain source port, it can
+        be specified using the ``source_port`` arg.
         """
         addrinfo = yield self.resolver.resolve(host, port, af)
         connector = _Connector(
             addrinfo, self.io_loop,
-            functools.partial(self._create_stream, max_buffer_size))
+            functools.partial(self._create_stream, max_buffer_size,
+                              source_ip=source_ip, source_port=source_port)
+        )
         af, addr, stream = yield connector.start()
         # TODO: For better performance we could cache the (af, addr)
         # information here and re-use it on subsequent connections to
@@ -174,13 +185,28 @@ class TCPClient(object):
                                             server_hostname=host)
         raise gen.Return(stream)
 
-    def _create_stream(self, max_buffer_size, af, addr):
+    def _create_stream(self, max_buffer_size, af, addr, source_ip=None,
+                       source_port=None):
         # Always connect in plaintext; we'll convert to ssl if necessary
         # after one connection has completed.
+        source_port_bind = source_port if isinstance(source_port, int) else 0
+        source_ip_bind = source_ip
+        if source_port_bind and not source_ip:
+            # User required a specific port, but did not specify
+            # a certain source IP, will bind to the default loopback.
+            source_ip_bind = '::1' if af == socket.AF_INET6 else '127.0.0.1'
+            # Trying to use the same address family as the requested af socket:
+            # - 127.0.0.1 for IPv4
+            # - ::1 for IPv6
+        socket_obj = socket.socket(af)
+        if source_port_bind or source_ip_bind:
+            # If the user requires binding also to a specific IP/port.
+            socket_obj.bind((source_ip_bind, source_port_bind))
+            # Fail loudly if unable to use the IP/port.
         try:
-            stream = IOStream(socket.socket(af),
-                            io_loop=self.io_loop,
-                            max_buffer_size=max_buffer_size)
+            stream = IOStream(socket_obj,
+                              io_loop=self.io_loop,
+                              max_buffer_size=max_buffer_size)
         except socket.error as e:
             fu = Future()
             fu.set_exception(e)
diff --git a/tornado/web.py b/tornado/web.py
index 21cc25f9..9557a6f3 100644
--- a/tornado/web.py
+++ b/tornado/web.py
@@ -1672,6 +1672,10 @@ def stream_request_body(cls):
     * The regular HTTP method (``post``, ``put``, etc) will be called after
       the entire body has been read.
 
+    There is a subtle interaction between ``data_received`` and asynchronous
+    ``prepare``: The first call to ``data_received`` may occur at any point
+    after the call to ``prepare`` has returned *or yielded*.
+
     See the `file receiver demo <https://github.com/tornadoweb/tornado/tree/master/demos/file_upload/>`_
     for example usage.
     """


Test patch:
diff --git a/tornado/test/tcpclient_test.py b/tornado/test/tcpclient_test.py
index 1a4201e6..702e0308 100644
--- a/tornado/test/tcpclient_test.py
+++ b/tornado/test/tcpclient_test.py
@@ -25,7 +25,7 @@ from tornado.netutil import bind_sockets, Resolver
 from tornado.tcpclient import TCPClient, _Connector
 from tornado.tcpserver import TCPServer
 from tornado.testing import AsyncTestCase, gen_test
-from tornado.test.util import skipIfNoIPv6, unittest, refusing_port
+from tornado.test.util import skipIfNoIPv6, unittest, refusing_port, skipIfNonUnix
 
 # Fake address families for testing.  Used in place of AF_INET
 # and AF_INET6 because some installations do not have AF_INET6.
@@ -81,9 +81,11 @@ class TCPClientTest(AsyncTestCase):
             self.skipTest("localhost does not resolve to ipv6")
 
     @gen_test
-    def do_test_connect(self, family, host):
+    def do_test_connect(self, family, host, source_ip=None, source_port=None):
         port = self.start_server(family)
-        stream = yield self.client.connect(host, port)
+        stream = yield self.client.connect(host, port,
+                                           source_ip=source_ip,
+                                           source_port=source_port)
         with closing(stream):
             stream.write(b"hello")
             data = yield self.server.streams[0].read_bytes(5)
@@ -125,6 +127,33 @@ class TCPClientTest(AsyncTestCase):
         with self.assertRaises(IOError):
             yield self.client.connect('127.0.0.1', port)
 
+    def test_source_ip_fail(self):
+        '''
+        Fail when trying to use the source IP Address '8.8.8.8'.
+        '''
+        self.assertRaises(socket.error,
+                          self.do_test_connect,
+                          socket.AF_INET,
+                          '127.0.0.1',
+                          source_ip='8.8.8.8')
+
+    def test_source_ip_success(self):
+        '''
+        Success when trying to use the source IP Address '127.0.0.1'
+        '''
+        self.do_test_connect(socket.AF_INET, '127.0.0.1', source_ip='127.0.0.1')
+
+    @skipIfNonUnix
+    def test_source_port_fail(self):
+        '''
+        Fail when trying to use source port 1.
+        '''
+        self.assertRaises(socket.error,
+                          self.do_test_connect,
+                          socket.AF_INET,
+                          '127.0.0.1',
+                          source_port=1)
+
 
 class TestConnectorSplit(unittest.TestCase):
     def test_one_family(self):