why asyncio.Queue could not work as expected?

Question

I am writing simple producer/consumer program.

import zmq

@asyncio.coroutine
def receive_data(future,s):
        print("begin to recv sth from.....socket"
        my_data = s.recv()
        future.set_result(my_data)

@asyncio.coroutine
def producer(loop,q,s):
        while True:
                future = asyncio.Future()
                yield from receive_data(future,s)
                data = str(future.result())
                yield from q.put(data)
@asyncio.coroutine
def consumer(loop,q):
       while True:
          a = yield from q.get()
          print("i am get..."+str(a)+"..."+str(type(a)))  
loop = asyncio.get_event_loop()

c = zmq.Context()
s = c.socket(zmq.REP)
s.bind('tcp://127.0.0.1:5515')

q = asyncio.Queue()
tasks=[asyncio.Task(producer(loop,q,s)),asyncio.Task(comsumer(loop,q))]
loop.run_until_complete(asyncio.gather(*tasks))
loop.close()
s.close()

It appears the consumer has no chance to execute.

The sockets receive data every 500ms, so when yield from in receive_data function suspends the producer coroutine, the consumer coroutine will print info.

What could explain this?

Answer 1

s.recv() is blocking call, so receive_data hungs until new ZMQ message arrives.

That blocks event loop and consumer has no chance to execute itself.

You can pass zmq.NOBLOCK flag to .recv and call asyncio.sleep(0) if no data available to give eventloop a chance to iterate over other ready tasks.

Or just use aiozmq library :)