azaion/annotations
1
0
Fork 0
mirror of https://github.com/azaion/annotations.git synced 2026-04-22 11:46:30 +00:00
Code Issues Packages Projects Releases Wiki Activity

make cython app exit correctly

Browse Source
This commit is contained in:
Alex Bezdieniezhnykh
2025-02-11 20:40:49 +02:00
parent 9973a16ada
commit 43cae0d03c
9 changed files with 87 additions and 65 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Azaion.Inference/remote_command_handler.pyx
+42 -20
View File
@@ -9,7 +9,6 @@ cdef class RemoteCommandHandler:
def __init__(self, object on_command):
self._on_command = on_command
self._context = zmq.Context.instance()
self._shutdown_event = Event()
self._router = self._context.socket(zmq.ROUTER)
self._router.setsockopt(zmq.LINGER, 0)
@@ -19,6 +18,10 @@ cdef class RemoteCommandHandler:
self._dealer.setsockopt(zmq.LINGER, 0)
self._dealer.bind("inproc://backend")
self._control = self._context.socket(zmq.PAIR)
self._control.bind("inproc://control")
self._shutdown_event = Event()
self._proxy_thread = Thread(target=self._proxy_loop, daemon=True)
self._workers = []
@@ -32,7 +35,13 @@ cdef class RemoteCommandHandler:
worker.start()
cdef _proxy_loop(self):
zmq.proxy(self._router, self._dealer)
try:
zmq.proxy_steerable(self._router, self._dealer, control=self._control)
except zmq.error.ZMQError as e:
if self._shutdown_event.is_set():
print("Shutdown, exit proxy loop.")
else:
raise
cdef _worker_loop(self):
worker_socket = self._context.socket(zmq.DEALER)
@@ -40,20 +49,24 @@ cdef class RemoteCommandHandler:
worker_socket.connect("inproc://backend")
poller = zmq.Poller()
poller.register(worker_socket, zmq.POLLIN)
print('started receiver loop...')
while not self._shutdown_event.is_set():
try:
socks = dict(poller.poll(500))
if worker_socket in socks:
client_id, message = worker_socket.recv_multipart()
cmd = RemoteCommand.from_msgpack(<bytes> message)
cmd.client_id = client_id
print(f'Received [{cmd}] from the client {client_id}')
self._on_command(cmd)
except Exception as e:
print(f"Worker error: {e}")
import traceback
traceback.print_exc()
try:
while not self._shutdown_event.is_set():
try:
socks = dict(poller.poll(500))
if worker_socket in socks:
client_id, message = worker_socket.recv_multipart()
cmd = RemoteCommand.from_msgpack(<bytes> message)
cmd.client_id = client_id
print(f'Received [{cmd}] from the client {client_id}')
self._on_command(cmd)
except Exception as e:
if not self._shutdown_event.is_set():
print(f"Worker error: {e}")
import traceback
traceback.print_exc()
finally:
worker_socket.close()
cdef send(self, bytes client_id, bytes data):
with self._context.socket(zmq.DEALER) as socket:
@@ -63,7 +76,16 @@ cdef class RemoteCommandHandler:
cdef stop(self):
self._shutdown_event.set()
time.sleep(0.5)
self._router.close()
self._dealer.close()
self._context.term()
try:
self._control.send(b"TERMINATE", flags=zmq.DONTWAIT)
except zmq.error.ZMQError:
pass
self._router.close(linger=0)
self._dealer.close(linger=0)
self._control.close(linger=0)
self._proxy_thread.join(timeout=2)
while any(w.is_alive() for w in self._workers):
time.sleep(0.1)
self._context.term()