Skip to content
Permalink
Browse files

feat(frontend): add max pending request to frontend

  • Loading branch information...
hanxiao committed Sep 29, 2019
1 parent e750cf9 commit bbf1ed8e392be2a68f697ce8b8d558acf71d68ad
Showing with 18 additions and 14 deletions.
  1. +18 −14 gnes/service/frontend.py
@@ -58,6 +58,7 @@ def __init__(self, args):
check_version=self.args.check_version,
timeout=self.args.timeout,
squeeze_pb=self.args.squeeze_pb)
self.pending_request = 0

def add_envelope(self, body: 'gnes_pb2.Request', zmq_client: 'ZmqClient'):
msg = gnes_pb2.Message()
@@ -103,24 +104,27 @@ def Search(self, request, context):
return self.Call(request, context)

def StreamCall(self, request_iterator, context):
self.pending_request = 0

def get_response(num_recv, blocked=False):
for _ in range(num_recv):
if blocked or zmq_client.receiver.poll(1):
msg = zmq_client.recv_message(**self.send_recv_kwargs)
self.pending_request -= 1
yield self.remove_envelope(msg)

with self.zmq_context as zmq_client:
pending_request = 0

for request in request_iterator:
zmq_client.send_message(self.add_envelope(request, zmq_client), **self.send_recv_kwargs)
pending_request += 1

while pending_request > self.args.max_pending_request:
# too many pending requests, the whole network is pretty busy
# slow down the sending rate by waiting responses
if zmq_client.receiver.poll(1):
msg = zmq_client.recv_message(**self.send_recv_kwargs)
pending_request -= 1
yield self.remove_envelope(msg)

for _ in range(pending_request):
msg = zmq_client.recv_message(**self.send_recv_kwargs)
yield self.remove_envelope(msg)
self.pending_request += 1

num_recv = max(self.pending_request - self.args.max_pending_request, 1)

# switch to blocked recv when too many pending requests
yield from get_response(num_recv, num_recv > 1)

yield from get_response(self.pending_request, blocked=True)

class ZmqContext:
"""The zmq context class."""

0 comments on commit bbf1ed8

Please sign in to comment.
You can’t perform that action at this time.