Merge branch 'master' into mash-merge-master

.travis.yml

@@ -19,6 +19,7 @@ env:
     - TOX_ENV=cover
     - TOX_ENV=flake8
     - TOX_ENV=docs
+    - TOX_ENV=benchmark
 
 install:
   - make install

CHANGES.rst

@@ -19,11 +19,65 @@ Changes by Version
   connection was previously terminated.
 
 
-0.21.5 (unreleased)
+0.21.11 (unreleased)
+--------------------
+
+- Peer selection is now constant time instead of linear time. This should
+  significantly reduce CPU load per request.
+- Fixed a bug where certain errors while reading requests would propagate as
+  TimeoutErrors.
+- Attempting to register endpoints against a synchronous TChannel now logs an
+  INFO level message.
+- Reduced default advertisement interval to 3 minutes.
+
+
+0.21.10 (2016-03-17)
+--------------------
+
+- Zipkin traces now include a server-side 'cn' annotation to identify callers.
+- Reduced "unconsumed message" warnings to INFO. These are typically generated
+  when Hyperbahn garbage collects your process due to a timed-out
+  advertisement.
+- Handshake timeouts were incorrectly being surfaced as StreamClosedError but
+  are now raised as NetworkError.
+- Reduced default tracing sample rate from 100% to 1%.
+
+
+0.21.9 (2016-03-14)
+-------------------
+
+- Fixed a bug that caused silent failures when a write attempt was made to a
+  closed connection.
+- Reduce ``StreamClosedError`` log noisiness for certain scenarios.
+- Make ``TChannel.advertise`` idempotent and thread-safe.
+
+
+0.21.8 (2016-03-10)
+-------------------
+
+- Reduce read errors due to clients disconnecting to INFO from ERROR.
+
+
+0.21.7 (2016-03-08)
+-------------------
+
+- Fixed an unhelpful stack trace on failed reads.
+
+
+0.21.6 (2016-03-08)
+-------------------
+
+- Fixed a logging error on failed reads.
+
+
+0.21.5 (2016-03-08)
 -------------------
 
 - Tornado 4.2 was listed as a requirement but this was corrected to be 4.3
   which introduced the locks module.
+- Fixed in issue where clients could incorrectly time out when reading large
+  response bodies. This was due to response fragments being dropped due to
+  out-of-order writes; writes are now serialized on a per-connection basis.
 
 
 0.21.4 (2016-02-15)

Makefile

@@ -48,6 +48,10 @@ test: clean
 test_ci: clean
 	tox -e $(TOX_ENV) -- tests
 
+.PHONY: benchmark
+benchmark:
+	py.test benchmarks --benchmark-autosave --benchmark-save-data --benchmark-warmup --benchmark-disable-gc --benchmark-histogram
+
 .PHONY: testhtml
 testhtml: clean
 	$(pytest) $(html_report) && open htmlcov/index.html

benchmarks/test_concurrent_requests.py

@@ -40,6 +40,7 @@ def hello(request):
 
         server.listen()
         servers.append(server)
+
     return servers
 
 

benchmarks/test_roundtrip.py

@@ -0,0 +1,64 @@
+# Copyright (c) 2015 Uber Technologies, Inc.
+#
+# Permission is hereby granted, free of charge, to any person obtaining a copy
+# of this software and associated documentation files (the "Software"), to deal
+# in the Software without restriction, including without limitation the rights
+# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+# copies of the Software, and to permit persons to whom the Software is
+# furnished to do so, subject to the following conditions:
+#
+# The above copyright notice and this permission notice shall be included in
+# all copies or substantial portions of the Software.
+#
+# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
+# THE SOFTWARE.
+
+from tornado import ioloop, gen
+
+from tchannel import TChannel, thrift
+
+
+service = thrift.load(
+    path='examples/guide/keyvalue/service.thrift',
+    service='benchmark-server',
+)
+
+
+def test_roundtrip(benchmark):
+    loop = ioloop.IOLoop.current()
+
+    server = TChannel('benchmark-server')
+    server.listen()
+
+    clients = [TChannel('benchmark-client') for _ in range(10)]
+
+    @server.thrift.register(service.KeyValue)
+    def getValue(request):
+        return 'bar'
+
+    def roundtrip():
+        @gen.coroutine
+        def doit():
+            futures = []
+            # 10 clients send 10 requests concurrently
+            for client in clients:
+                for _ in range(10):
+                    futures.append(
+                        client.thrift(
+                            service.KeyValue.getValue("foo"),
+                            hostport=server.hostport,
+                        )
+                    )
+            yield futures
+
+        return loop.run_sync(doit)
+
+    # Establish initial connection
+    roundtrip()
+
+    benchmark(roundtrip)

examples/sync/fanout/client.py

@@ -0,0 +1,59 @@
+# Copyright (c) 2015 Uber Technologies, Inc.
+#
+# Permission is hereby granted, free of charge, to any person obtaining a copy
+# of this software and associated documentation files (the "Software"), to deal
+# in the Software without restriction, including without limitation the rights
+# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+# copies of the Software, and to permit persons to whom the Software is
+# furnished to do so, subject to the following conditions:
+#
+# The above copyright notice and this permission notice shall be included in
+# all copies or substantial portions of the Software.
+#
+# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
+# THE SOFTWARE.
+
+import json
+
+from tchannel import thrift
+from tchannel.sync import TChannel
+
+tchannel = TChannel('thrift-client')
+service = thrift.load(
+    path='tests/data/idls/ThriftTest.thrift',
+    service='thrift-server',
+    hostport='localhost:54498',
+)
+
+
+def make_requests():
+
+    # Fan-out
+    futures = [tchannel.thrift(
+        request=service.ThriftTest.testString(thing="req"),
+        headers={
+            'req': 'header',
+        },
+    ) for _ in xrange(20)]
+
+    # Fan-in
+    for future in futures:
+        response = future.result()
+
+    return response
+
+
+resp = make_requests()
+
+assert resp.headers == {
+    'resp': 'header',
+}
+assert resp.body == 'resp' * 100000
+
+print resp.body[:4]
+print json.dumps(resp.headers)

examples/sync/fanout/server.py

@@ -0,0 +1,44 @@
+# Copyright (c) 2015 Uber Technologies, Inc.
+#
+# Permission is hereby granted, free of charge, to any person obtaining a copy
+# of this software and associated documentation files (the "Software"), to deal
+# in the Software without restriction, including without limitation the rights
+# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+# copies of the Software, and to permit persons to whom the Software is
+# furnished to do so, subject to the following conditions:
+#
+# The above copyright notice and this permission notice shall be included in
+# all copies or substantial portions of the Software.
+#
+# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
+# THE SOFTWARE.
+
+from __future__ import absolute_import
+
+from tornado import gen, ioloop
+from tchannel import TChannel, Response, thrift
+
+tchannel = TChannel('thrift-server', hostport='localhost:54498')
+service = thrift.load('tests/data/idls/ThriftTest.thrift')
+
+
+@tchannel.thrift.register(service.ThriftTest)
+@gen.coroutine
+def testString(request):
+
+    assert request.headers == {'req': 'header'}
+    assert request.body.thing == 'req'
+
+    return Response('resp' * 100000, headers={'resp': 'header'})
+
+
+tchannel.listen()
+
+print tchannel.hostport
+
+ioloop.IOLoop.current().start()

tchannel/__init__.py

@@ -22,7 +22,7 @@
     absolute_import, division, print_function, unicode_literals
 )
 
-__version__ = '0.21.5.dev0'
+__version__ = '0.21.11.dev0'
 # Update setup.py when changing this. zest.releaser doesn't support updating
 # both of them yet.
 

tchannel/_future.py

@@ -0,0 +1,78 @@
+# Copyright (c) 2015 Uber Technologies, Inc.
+#
+# Permission is hereby granted, free of charge, to any person obtaining a copy
+# of this software and associated documentation files (the "Software"), to deal
+# in the Software without restriction, including without limitation the rights
+# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+# copies of the Software, and to permit persons to whom the Software is
+# furnished to do so, subject to the following conditions:
+#
+# The above copyright notice and this permission notice shall be included in
+# all copies or substantial portions of the Software.
+#
+# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
+# THE SOFTWARE.
+
+from __future__ import (
+    absolute_import, unicode_literals, division, print_function
+)
+
+import sys
+from functools import wraps
+
+from tornado.gen import is_future
+
+
+def fail_to(future):
+    """A decorator for function callbacks to catch uncaught non-async
+    exceptions and forward them to the given future.
+
+    The primary use for this is to catch exceptions in async callbacks and
+    propagate them to futures. For example, consider,
+
+    .. code-block:: python
+
+        answer = Future()
+
+        def on_done(future):
+            foo = bar()
+            answer.set_result(foo)
+
+        some_async_operation().add_done_callback(on_done)
+
+    If ``bar()`` fails, ``answer`` will never get filled with an exception or
+    a result. Now if we change ``on_done`` to,
+
+    .. code-block:: python
+
+        @fail_to(answer)
+        def on_done(future):
+            foo = bar()
+            answer.set_result(foo)
+
+    Uncaught exceptions in ``on_done`` will be caught and propagated to
+    ``answer``. Note that ``on_done`` will return None if an exception was
+    caught.
+
+    :param answer:
+        Future to which the result will be written.
+    """
+    assert is_future(future), 'you forgot to pass a future'
+
+    def decorator(f):
+
+        @wraps(f)
+        def new_f(*args, **kwargs):
+            try:
+                return f(*args, **kwargs)
+            except Exception:
+                future.set_exc_info(sys.exc_info())
+
+        return new_f
+
+    return decorator