traviscline/gevent-zeromq — Issue #24: strange deadlock issue with queues
Author: hhuuggoo
Created: 2012-01-25 15:23:20+00:00
State: open

Here are 2 files, one is a server that starts up N request reply echo green threads and write the port numbers to a file.
the second is a simple test case, which creates a test object, with an XREQ socket which reads data off of a gevent queue, forwards it through the zeromq socket, reads the resulting message, and puts the msg on a number of outgoing gevent queues.

this simple test case deadlocks quite quickly, while waiting to recv a message on the zeromq socket.

simple_test.py:

from gevent_zeromq import zmq
import gevent
import gevent.monkey
gevent.monkey.patch_all()
import numpy as np
from gevent import spawn
import time
import logging
import uuid
import gevent.queue
log = logging.getLogger(__name__)
logging.basicConfig(level=logging.DEBUG)

with open("rrports.txt","r") as f:
ports = [int(x) for x in f.read().split(",")]
results = {}
ctx = zmq.Context()
status = {}
testers = {}

class Test(object):
def __init__(self, zmq_conn_string, ctx):
socket = ctx.socket(zmq.XREQ)
socket.connect(zmq_conn_string)
socket.setsockopt(zmq.HWM, 100)
self.s = socket
self.inqueue = gevent.queue.Queue()
self.outqueues = {}
def run_send(self):
while(True):
msg = self.inqueue.get()
multipart_msg = ['', msg]
log.debug('sending %s', msg)
self.s.send_multipart(multipart_msg)
log.debug('sent %s', msg)
def run_recv(self):
while(True):
log.debug('waiting for recv', )
sep, msg = self.s.recv_multipart()
log.debug('received %s', msg)
self.outqueues[msg].put(msg)

def start(self):
self.t = spawn(self.run_send)
self.t2 = spawn(self.run_recv)

def test(port, num_reqs):
identity = str(uuid.uuid4())
results[port, identity] = 0
zmq_conn_string = "tcp://127.0.0.1:" + str(port)
if zmq_conn_string in testers:
tester = testers[zmq_conn_string]
else:
tester = Test(zmq_conn_string, ctx)
tester.start()
testers[zmq_conn_string] = tester
tester.outqueues[identity] = gevent.queue.Queue()
for c in range(num_reqs):
tester.inqueue.put(identity)
msg = tester.outqueues[identity].get()
assert msg == identity
results[port, identity] += 1
num_reqs = 30
num_per_port = 2
num_ports = len(ports)
threads = []
for p in ports:
for c in range(num_per_port):
threads.append(spawn(test, p, num_reqs))

def report():
while(True):
log.debug("%s, %s, %s", len(results),
np.sum(np.array(results.values())),
num_reqs * num_per_port * num_ports)
gevent.sleep(1)
threads.append(spawn(report))
gevent.joinall(threads)

server.py:

from gevent_zeromq import zmq
import gevent
import gevent.monkey
gevent.monkey.patch_all()
from gevent import spawn
import time
import logging
log = logging.getLogger(__name__)
logging.basicConfig(level=logging.DEBUG)

num_reqrep = 1
rr = []
def rr_test(s):
while True:
msg = s.recv()
log.debug(msg)
s.send(msg)

for c in range(num_reqrep):
ctx = zmq.Context()
s = ctx.socket(zmq.REP)
s.setsockopt(zmq.HWM, 100)
port = s.bind_to_random_port("tcp://127.0.0.1")
t = spawn(rr_test, s)
rr.append((s,t, port))

with open("rrports.txt","w+") as f:
f.write(",".join([str(x[-1]) for x in rr]))

threads = [x[1] for x in rr]
gevent.joinall(threads)