aio-libs / aiozmq Goto Github PK

View Code? Open in Web Editor NEW

417.0 18.0 56.0 1.01 MB

Asyncio (pep 3156) integration with ZeroMQ

Home Page: aiozmq.readthedocs.org

License: BSD 2-Clause "Simplified" License

Makefile 0.31% Python 99.69%

asyncio zeromq zmq

aiozmq's Introduction

asyncio integration with ZeroMQ

asyncio (PEP 3156) support for ZeroMQ.

https://travis-ci.com/aio-libs/aiozmq.svg?branch=master

The difference between aiozmq and vanilla pyzmq (zmq.asyncio) is:

zmq.asyncio works only by replacing the base event loop with a custom one. This approach works but has two disadvantages:

zmq.asyncio.ZMQEventLoop cannot be combined with other loop implementations (most notable is the ultra fast uvloop).
It uses the internal ZMQ Poller which has fast ZMQ Sockets support but isn't intended to work fast with many (thousands) regular TCP sockets.

In practice it means that zmq.asyncio is not recommended to be used with web servers like aiohttp.

See also zeromq/pyzmq#894

Documentation

See http://aiozmq.readthedocs.org

Simple high-level client-server RPC example:

import asyncio
import aiozmq.rpc


class ServerHandler(aiozmq.rpc.AttrHandler):

    @aiozmq.rpc.method
    def remote_func(self, a:int, b:int) -> int:
        return a + b


async def go():
    server = await aiozmq.rpc.serve_rpc(
        ServerHandler(), bind='tcp://127.0.0.1:5555')
    client = await aiozmq.rpc.connect_rpc(
        connect='tcp://127.0.0.1:5555')

    ret = await client.call.remote_func(1, 2)
    assert 3 == ret

    server.close()
    client.close()

asyncio.run(go())

Low-level request-reply example:

import asyncio
import aiozmq
import zmq

async def go():
    router = await aiozmq.create_zmq_stream(
        zmq.ROUTER,
        bind='tcp://127.0.0.1:*')

    addr = list(router.transport.bindings())[0]
    dealer = await aiozmq.create_zmq_stream(
        zmq.DEALER,
        connect=addr)

    for i in range(10):
        msg = (b'data', b'ask', str(i).encode('utf-8'))
        dealer.write(msg)
        data = await router.read()
        router.write(data)
        answer = await dealer.read()
        print(answer)
    dealer.close()
    router.close()

asyncio.run(go())

Comparison to pyzmq

zmq.asyncio provides an asyncio compatible loop implementation.

But it's based on zmq.Poller which doesn't work well with massive non-zmq socket usage.

E.g. if you build a web server for handling at least thousands of parallel web requests (1000-5000) pyzmq's internal poller will be slow.

aiozmq works with epoll natively, it doesn't need a custom loop implementation and cooperates pretty well with uvloop for example.

For details see zeromq/pyzmq#894

Requirements

Python 3.6+
pyzmq 13.1+
optional submodule aiozmq.rpc requires msgpack 0.5+

License

aiozmq is offered under the BSD license.

aiozmq's People

Contributors

Stargazers

Watchers

aiozmq's Issues

pyzmqtulip replace tulip eventloop it doesn't integrate with other eventloops

It should be possible to use several eventloop with tulip, with this integration library one can only zmq eventloop making not possible to use gobject eventloop for instance.

If we have PubSubService.subscribe then we need for .unsubscribe also

.subscribe should accept only str or bytes, not iterable
I'm not 100% sure through

Appropriate msgpack-python not installed

I just did a pip install aiozmq and the requirement msgpack-python>=0.4 was not enforced:

pip install aiozmq

Collecting aiozmq
  Downloading aiozmq-0.7.1-py3-none-any.whl
Requirement already satisfied (use --upgrade to upgrade): pyzmq>=13.1 in /usr/lib/python3/dist-packages (from aiozmq)
Installing collected packages: aiozmq
Successfully installed aiozmq-0.7.1

python test1.py

...

ImportError: cannot import name 'ExtType'

Yet I see the appropriate lines in setup.py. I'm not sure what could have gone wrong.

Rename PipelineClient.pipeline to .notify

Zero Copy Strategy

As I only send serialised data bytes I prefer zero copy sending.
aiozmq may employ zero copy safely in case of immutable types like bytes.

Here is a patch which does not not distinguish between mutable and immutable types.
https://github.com/peteut/aiozmq/commit/580967db357379e6897e6a08dd5f5293d0dde9e8

wait for transport buffer flush

Hey again, is it possible to wait till all transport buffers are flush or have a callback called when all buffers are flushed?
Getting a callback to the protocol whenever remove_writer and add_write are called inside the transport would be enough, but I don't know if this is conceptually ok or if there is any better solution.

I run loop.run_until_completed(...) and it exits after the last write, but the transports buffers still are full. The loop is resumed later, so I don't want to close the connection. I could set high and low watermark to 1 and 0, and use pause/resume_writing, but that makes the normal case more inefficient. Any ideas?

New release

Hello,

I hit the bug fixed in commit dcdf4c8 while installing another software that makes use
of aiozmq. Steps to reproduce:

$ virtualenv -p /usr/bin/python3.4 xbus
$ ./xbus/bin/pip install xbus.broker
$ ./xbus/bin/start_xbusbroker
Traceback (most recent call last):
[...]
File "[...]/xbus/lib/python3.4/site-packages/aiozmq/core.py", line 211, in _BaseTransport
  zmq.STREAM: 'STREAM'}
AttributeError: 'module' object has no attribute 'STREAM'

Any plans for a new release?

Thanks.

Failing tests Python 3.5 MessagePack (question/bug)

I just came across this package and it looks very promising, but I can see that the last push was five months ago and some of the Python 3.5 tests using MessagePack failed. Is this package still being maintained?

decimal.Decimal type support in RPC

Is Decimal not supported by design? I have fixed this with value translator:

import msgpack
from decimal import Decimal

translation_table = {
    0: (Decimal,
        lambda value: msgpack.packb(str(value)),
        lambda binary: Decimal(msgpack.unpackb(binary).decode())),
}

And it does not throw any error if conversion failed. Look like it just hags on the call.

Implement transport's monitor

To get notifications about connection/disconnection.

RPC calls to multiple services with round robin DNS lookup

When I do something like await rpc.connect_rpc(connect='tcp://rpc-service:5555'), where "rpc-service" is a round robin A record with multiple addresses, can I expect the client to choose a different address every time it is initialized?

From experimentation it seems that this is not the case, but I might be doing something wrong.

Add limited support for Windows

Just inherit ZmqEventLoop from _WindowsSelectorEventLoop.
subprocess API is not supported in this case and total performance may be poor.

Make 100% code coverage

Weird ZmqEventLoop errors

python3 runtests.py --forever sometimes causes:

======================================================================
ERROR: test_zmq_socket (zmq_events_test.ZmqEventLoopTests)
----------------------------------------------------------------------
Traceback (most recent call last):
  File "/home/alexey/dev/aiozmq/aiozmq/selector.py", line 176, in select
    z_events = self._poller.poll(timeout)
  File "/usr/lib/python3/dist-packages/zmq/sugar/poll.py", line 97, in poll
    return zmq_poll(list(self.sockets.items()), timeout=timeout)
  File "_poll.pyx", line 116, in zmq.core._poll.zmq_poll (zmq/core/_poll.c:1484)
  File "checkrc.pxd", line 21, in zmq.core.checkrc._check_rc (zmq/core/_poll.c:1866)
zmq.error.ZMQError: Socket operation on non-socket

The above exception was the direct cause of the following exception:

Traceback (most recent call last):
  File "tests/zmq_events_test.py", line 286, in test_zmq_socket
    tr, pr = self.loop.run_until_complete(connect())
  File "/home/alexey/.local/lib/python3.3/site-packages/asyncio-0.4.1-py3.3.egg/asyncio/base_events.py", line 203, in run_until_complete
    self.run_forever()
  File "/home/alexey/.local/lib/python3.3/site-packages/asyncio-0.4.1-py3.3.egg/asyncio/base_events.py", line 184, in run_forever
    self._run_once()
  File "/home/alexey/.local/lib/python3.3/site-packages/asyncio-0.4.1-py3.3.egg/asyncio/base_events.py", line 799, in _run_once
    event_list = self._selector.select(timeout)
  File "/home/alexey/dev/aiozmq/aiozmq/selector.py", line 181, in select
    raise OSError(exc.errno, exc.strerror) from exc
OSError: [Errno 88] Socket operation on non-socket

----------------------------------------------------------------------

======================================================================
ERROR: test_zmq_socket (zmq_events_test.ZmqEventLoopTests)
----------------------------------------------------------------------
Traceback (most recent call last):
  File "/home/alexey/dev/aiozmq/aiozmq/events.py", line 87, in create_zmq_connection
    elif zmq_sock.getsockopt(zmq.TYPE) != zmq_type:
  File "socket.pyx", line 359, in zmq.core.socket.Socket.get (zmq/core/socket.c:3249)
  File "socket.pyx", line 105, in zmq.core.socket._check_closed (zmq/core/socket.c:1325)
zmq.error.ZMQError: Operation not supported

The above exception was the direct cause of the following exception:

Traceback (most recent call last):
  File "tests/zmq_events_test.py", line 286, in test_zmq_socket
    tr, pr = self.loop.run_until_complete(connect())
  File "/home/alexey/.local/lib/python3.3/site-packages/asyncio-0.4.1-py3.3.egg/asyncio/base_events.py", line 208, in run_until_complete
    return future.result()
  File "/home/alexey/.local/lib/python3.3/site-packages/asyncio-0.4.1-py3.3.egg/asyncio/futures.py", line 243, in result
    raise self._exception
  File "/home/alexey/.local/lib/python3.3/site-packages/asyncio-0.4.1-py3.3.egg/asyncio/tasks.py", line 283, in _step
    result = next(coro)
  File "tests/zmq_events_test.py", line 282, in connect
    zmq_sock=zmq_sock)
  File "/home/alexey/dev/aiozmq/aiozmq/events.py", line 90, in create_zmq_connection
    raise OSError(exc.errno, exc.strerror) from exc
OSError: [Errno 95] Operation not supported

----------------------------------------------------------------------

transport.connect with a DNS name fails -- works with a pre-made zmq_sock

Connecting a (Dealer) socket with a DNS name raises ValueError: 'example.com' does not appear to be an IPv4 or IPv6 address:

import zmq, aiozmq, asyncio
def test():
  socket = yield from aiozmq.create_zmq_stream(zmq.DEALER)
  socket.transport.setsockopt(zmq.IDENTITY, b'myidentity')
  yield from socket.transport.connect('tcp://example.com:1234')
asyncio.get_event_loop().run_until_complete(test())

The following works, however:

context = zmq.Context()
zmq_socket = context.socket(zmq.DEALER)
zmq_socket.setsockopt(zmq.IDENTITY, b'myidentity')
zmq_socket.connect('tcp://example.com:1234')
socket = yield from aiozmq.create_zmq_stream(zmq.DEALER, zmq_sock=zmq_socket)

Versions:
aiozmq==0.6.1
pyzmq==14.7.0

Change aiozmq.rpc rpc protocol to support REQ/REP pair

await branch like aiopg

Could aiozmq have an await branch to demonstrate examples using the new syntax?

Using async def and await everywhere?

Add check for libzmq version

setup.py should fail if libzmq<3.0

Fix packer's translation table lookup

Don't use lazy cache calculation, it should be explicit procedure to avoid false positives.
Say, asking for datetime.date before datetime.datetime makes date as best-match for datetime also.

Add more examples for using aiozmq in conventional way

Although the documentation says that aiozmq.rpc is optional, there are no examples that do not use this module. I want to use zeromq similar to how we use zmq, for example: http://zguide.zeromq.org/py:rrclient . I am at a loss as to how to do so using aiozmq - namely how to obtain an object analogous to zmq.Context.

Please provide a way to cach route error

Greetings.

in https://github.com/aio-libs/aiozmq/blob/master/aiozmq/core.py#L266 transoport's write method returns None in both cases when socket.write raises error or when message sended successfully. So I can't figure out is message was sent. Please make transport's write method raise exception if it can't send message.

Add docstrings to aiozmq.rpc public functions

Collect statistics about bad rpc calls

In some form like {'ValueError': 5, 'my_module.MyClass': 33}

rename client.rpc to client.call

Add zmq streams

should be mimic to asyncio.streams but operates with multipart messages instead of bytes.

Add more examples

Add strict flag to msgpack unpacker

Use strict search for class (don't use isinstance check) if flag is True

stream.write() prevents Python process from terminating after loop clean up using task cancellation

write() calls to an aiozmq stream prevent "clean termination" of the Python process.
Here is a minimal working example:

#! /usr/bin/env python3
import asyncio, aiozmq, zmq

@asyncio.coroutine
def test(loop):
    # connect to a non-existent endpoint (this returns immediately)
    sock = yield from aiozmq.create_zmq_stream(zmq.REQ, connect='tcp://localhost:9999', loop=loop)
    sock.write([b'test'])
    # just make some time to press Ctrl+C
    yield from asyncio.sleep(5)
    # below is not executed as sleep is cancelled.
    _ = yield from sock.read()
    sock.close()

if __name__ == '__main__':
    loop = asyncio.get_event_loop()
    asyncio.async(test(loop))
    try: 
        loop.run_forever()
    except KeyboardInterrupt:
        for t in asyncio.Task.all_tasks():
            if not t.done():
                t.cancel()  # cancels the sleep call
        loop.run_until_complete(asyncio.sleep(0))
    finally:
        loop.close()
        print('exit')

The result when you press Ctrl+C during sleep is:

$ python3 test-aiozmq-write-cancel.py
^Cexit   # Python hangs here; If I comment the write() call, it goes normal.
^C       # I need to press Ctrl+C again to terminate the Python process completely.

Is there any way to correctly interrupt/cancel the background behavior of write() calls?

Allow to create zmq transport without bindings/connections

That can be done later via transport methods.

Return traceback info if rpc call produces exception on remote side.

should work only if server has been created in debug mode (parameteer debug=True for serve_rpc, should be False by debault).

Finish the doc

Document all public API

Add new transport implementation that uses 0MQ sockets with standard event loop.

That also gets rid of problems with ssl connections.

RPC example doesn't work

The RPC example in the docs doesn't work. For one, it misses import asyncio, but even after rectifying this the example fails like so:

✗ ./docker/templates/sync-with-github.py
Traceback (most recent call last):
  File "./docker/templates/sync-with-github.py", line 24, in <module>
    asyncio.get_event_loop().run_until_complete(go())
  File "/usr/local/Cellar/python3/3.5.1/Frameworks/Python.framework/Versions/3.5/lib/python3.5/asyncio/base_events.py", line 337, in run_until_complete
    return future.result()
  File "/usr/local/Cellar/python3/3.5.1/Frameworks/Python.framework/Versions/3.5/lib/python3.5/asyncio/futures.py", line 274, in result
    raise self._exception
  File "/usr/local/Cellar/python3/3.5.1/Frameworks/Python.framework/Versions/3.5/lib/python3.5/asyncio/tasks.py", line 239, in _step
    result = coro.send(None)
  File "./docker/templates/sync-with-github.py", line 18, in go
    ret = yield from client.rpc.remote_func(1, 2)
AttributeError: 'RPCClient' object has no attribute 'rpc'

I have a PR for this issue, even though it's incomplete.

Make a check for libzmq version

We doesn't support libzmq2 etc.
The check should be done in init.py

Add optional list of exceptions which passed silently via log_exception machinery

Add flow control for monitor events

Looking on the code I've figured out that monitoring events are need flow control for upcoming events like we have it for regular messages.

I mean transport.pause_reading() and transport.resume_reading() methods.
The problem is: if protocol implementation cannot consume monitoring events as fast as they are appears there is useful to have a way to inform pair for stopping sending new events.

I guess it can be done by unsubscribing from socket events in the poller like regular pause_reading() does.
BTW look like I don't support the feature for loop-less transport, it should be fixed.

Maybe we need pause_monitor() and resume_monitor() methods in transport.

@claws what do you think about?

Add timeouts to rpc servers

We should cancel long-running server handlers by wrapping handled task into wait_for with timeout.

Can no longer close server via RPC with loopless design

Before the new loopless design, a client could close a server through an RPC call and receive a response to the call. After switching to the loopless design in 0.5, the server exits, but the client doesn't receive a response.

With the old design, I could ensure that the server responded before closing by scheduling the server.close() call with loop.call_soon. This is no longer sufficient unfortunately.

Server

import asyncio
import aiozmq.rpc

class ServerHandler(aiozmq.rpc.AttrHandler):
    @aiozmq.rpc.method
    def exit(self):
        loop = asyncio.get_event_loop()
        loop.call_soon(self.server.close)
        print('Service shutting down...')


@asyncio.coroutine
def go():
    handler = ServerHandler()
    server = yield from aiozmq.rpc.serve_rpc(
        handler, bind='tcp://127.0.0.1:5555')
    handler.server = server
    yield from server.wait_closed()

asyncio.get_event_loop().run_until_complete(go())

Client

import asyncio
import zmq
import aiozmq.rpc

@asyncio.coroutine
def go():
    client = yield from aiozmq.rpc.connect_rpc(
        connect='tcp://127.0.0.1:5555')
    # Tell zmq not to wait until buffered data is sent after socket is closed
    client.transport.setsockopt(zmq.LINGER, 0)

    ret = yield from client.call.exit()
    client.close()

asyncio.get_event_loop().run_until_complete(go())

Implement timeouts for rpc clients

Make a check at aiozmq.rpc import for msgpack version

Should raise ImportError with good message like 'Please use msgpack >= 0.4"

The reason is the aiozmq.rpc uses ext type for custom value translators.

Fix travis build

Get rid of annoying:

test_subscribe_to_bytes (rpc_pubsub_test.PubSubTests) ... Bad file descriptor (epoll.cpp:81)

======================================================================
ERROR: test_unknown_req_id_at_client (rpc_test.RpcTests)
----------------------------------------------------------------------
Traceback (most recent call last):
  File "/home/travis/build/aio-libs/aiozmq/aiozmq/selector.py", line 186, in select
    z_events = self._poller.poll(timeout)
  File "/home/travis/virtualenv/python3.3/lib/python3.3/site-packages/pyzmq-14.1.1-py3.3-linux-x86_64.egg/zmq/sugar/poll.py", line 110, in poll
    return zmq_poll(self.sockets, timeout=timeout)
  File "_poll.pyx", line 115, in zmq.backend.cython._poll.zmq_poll (zmq/backend/cython/_poll.c:1678)
  File "checkrc.pxd", line 21, in zmq.backend.cython.checkrc._check_rc (zmq/backend/cython/_poll.c:2048)
zmq.error.ZMQError: Socket operation on non-socket
The above exception was the direct cause of the following exception:
Traceback (most recent call last):
  File "/usr/lib/python3.3/unittest/mock.py", line 1087, in patched
    return func(*args, **keywargs)
  File "tests/rpc_test.py", line 502, in test_unknown_req_id_at_client
    self.loop.run_until_complete(go())
  File "/home/travis/virtualenv/python3.3/lib/python3.3/site-packages/asyncio-0.4.1-py3.3.egg/asyncio/base_events.py", line 203, in run_until_complete
    self.run_forever()
  File "/home/travis/virtualenv/python3.3/lib/python3.3/site-packages/asyncio-0.4.1-py3.3.egg/asyncio/base_events.py", line 184, in run_forever
    self._run_once()
  File "/home/travis/virtualenv/python3.3/lib/python3.3/site-packages/asyncio-0.4.1-py3.3.egg/asyncio/base_events.py", line 778, in _run_once
    event_list = self._selector.select(timeout)
  File "/home/travis/build/aio-libs/aiozmq/aiozmq/selector.py", line 191, in select
    raise OSError(exc.errno, exc.strerror) from exc
OSError: [Errno 88] Socket operation on non-socket

etc.

The problem will be fixed #18 (I hope) but we need to work on ZeroMQ 3 also.

The solution should add sleeps (or maybe asyncio.sleep) to failed tests.

implement transport's pause_reading()/resume_reading() methods

I tried to use aiozmq but have problems when processing data from recv_message is to slow. It looks like the following, with process_data being a coroutine:

def msg_received(self, data):
     asyncio.async(self.process_data(data))

As process_data is waits too long for other resources, asyncio will fill up, so I would like to stop reading after some requests pile up.

Revert exceptions raised by ZmqEventLoopPolicy methods back to assertions

To follow asyncio EventLoopPolicy behavior

ZMQ proxy dies without log

Still diagnosting, but forwarder stops after some time. No logs, just stops forwarding. Restarting forwarder resends pending pub-sub calls. Most likely the ZMQ_XSUB side dies.

4 tests are failing

4 tests are failing for me with aiozmq revision 7ab57b7, Python 3.4.1 on Windows 7. Is this expected?

I am investigating RPC calls timing out (although my RPC methods finish on time), so I'm wondering if the failing tests may be relevant.

Q:\Users\arvek\Repositories\accc-gui\aiozmq [master]> python .\runtests.py                                                                                                                    
Skipping 'events_test': don't check                                                                                                                                                           
..ssssss......................................................................................................................F............................F.......s..........s......s........
........................................................F..ss................................F..ss............................                                                                
======================================================================                                                                                                                        
FAIL: test_server_close (rpc_test.LoopLessRpcTests)                                                                                                                                           
----------------------------------------------------------------------                                                                                                                        
Traceback (most recent call last):                                                                                                                                                            
  File "tests\rpc_test.py", line 615, in test_server_close                                                                                                                                    
    self.loop.run_until_complete(communicate())                                                                                                                                               
  File "C:\Python34\lib\asyncio\base_events.py", line 208, in run_until_complete                                                                                                              
    return future.result()                                                                                                                                                                    
  File "C:\Python34\lib\asyncio\futures.py", line 243, in result                                                                                                                              
    raise self._exception                                                                                                                                                                     
  File "C:\Python34\lib\asyncio\tasks.py", line 321, in _step                                                                                                                                 
    result = next(coro)                                                                                                                                                                       
  File "tests\rpc_test.py", line 606, in communicate                                                                                                                                          
    self.assertEqual(1, len(server._proto.pending_waiters))                                                                                                                                   
AssertionError: 1 != 0                                                                                                                                                                        

======================================================================                                                                                                                        
FAIL: test_server_close (rpc_test.LoopRpcTests)                                                                                                                                               
----------------------------------------------------------------------                                                                                                                        
Traceback (most recent call last):                                                                                                                                                            
  File "tests\rpc_test.py", line 615, in test_server_close                                                                                                                                    
    self.loop.run_until_complete(communicate())                                                                                                                                               
  File "C:\Python34\lib\asyncio\base_events.py", line 208, in run_until_complete                                                                                                              
    return future.result()                                                                                                                                                                    
  File "C:\Python34\lib\asyncio\futures.py", line 243, in result                                                                                                                              
    raise self._exception                                                                                                                                                                     
  File "C:\Python34\lib\asyncio\tasks.py", line 321, in _step                                                                                                                                 
    result = next(coro)                                                                                                                                                                       
  File "tests\rpc_test.py", line 606, in communicate                                                                                                                                          
    self.assertEqual(1, len(server._proto.pending_waiters))                                                                                                                                   
AssertionError: 1 != 0                                                                                                                                                                        

======================================================================                                                                                                                        
FAIL: test_close_on_error (zmq_events_test.ZmqLooplessTests)                                                                                                                                  
----------------------------------------------------------------------                                                                                                                        
Traceback (most recent call last):                                                                                                                                                            
  File "tests\zmq_events_test.py", line 745, in test_close_on_error                                                                                                                           
    check_errno(88, handler.call_args[0][1]['exception'])                                                                                                                                     
  File "Q:\Users\arvek\Repositories\accc-gui\aiozmq\aiozmq\_test_util.py", line 226, in check_errno                                                                                           
    assert exc.errno == errno, (exc, errno)                                                                                                                                                   
AssertionError: (OSError(128, 'Unknown error'), 88)                                                                                                                                           

======================================================================                                                                                                                        
FAIL: test_close_on_error (zmq_events_test.ZmqEventLoopTests)                                                                                                                                 
----------------------------------------------------------------------                                                                                                                        
Traceback (most recent call last):                                                                                                                                                            
  File "tests\zmq_events_test.py", line 745, in test_close_on_error                                                                                                                           
    check_errno(88, handler.call_args[0][1]['exception'])                                                                                                                                     
  File "Q:\Users\arvek\Repositories\accc-gui\aiozmq\aiozmq\_test_util.py", line 226, in check_errno                                                                                           
    assert exc.errno == errno, (exc, errno)                                                                                                                                                   
AssertionError: (OSError(128, 'Unknown error'), 88)                                                                                                                                           

----------------------------------------------------------------------                                                                                                                        
Ran 316 tests in 15.244s                                                                                                                                                                      

FAILED (failures=4, skipped=13)

Implement auth from ZeroMQ 4

RPC calls time out

We are suffering from this issue where RPC calls time out without answering the client, our current timeout being 10 seconds. We are on Python 3.4.1 and Windows 7 SP1 x64.

I have now added a bunch of logging to aiozmq, and have pinpointed that, while the server does write the response to the socket on time, the response is not received by the client before the deadline. There is actually a 10 seconds interval between the time that the response is sent and that the client times out. Is there a possibility that self._zmq_sock.send_multipart(data, zmq.DONTWAIT) is failing? How can you tell?

Reimplement rpc layer with zmq streams

That should simplify the code

Convert transport's bind, unbind, connect, disconnect calls to coroutines

Literally that methods should return Future.
For now it is the ready future, but behavior will be changed after implementing support for zmq monitor for the transport.