1111# ===================================================================
1212# ZeroMQ Communication Wrapper
1313# ===================================================================
14+ # single shared ZMQ context for the entire process.
15+ _zmq_context = zmq .Context ()
16+
1417class ZeroMQPort :
15- def __init__ (self , port_type , address , zmq_socket_type ):
18+ def __init__ (self , port_type , address , zmq_socket_type , context ):
1619 """
1720 port_type: "bind" or "connect"
1821 address: ZeroMQ address (e.g., "tcp://*:5555")
1922 zmq_socket_type: zmq.REQ, zmq.REP, zmq.PUB, zmq.SUB etc.
23+ context: shared zmq.Context() for the process (do not create one per port)
2024 """
21- self .context = zmq .Context ()
22- self .socket = self .context .socket (zmq_socket_type )
25+ self .socket = context .socket (zmq_socket_type )
2326 self .port_type = port_type # "bind" or "connect"
2427 self .address = address
2528
@@ -76,7 +79,7 @@ def init_zmq_port(mod, port_name, port_type, address, socket_type_str):
7679 try :
7780 # Map socket type string to actual ZMQ constant (e.g., zmq.REQ, zmq.REP)
7881 zmq_socket_type = getattr (zmq , socket_type_str .upper ())
79- mod .zmq_ports [port_name ] = ZeroMQPort (port_type , address , zmq_socket_type )
82+ mod .zmq_ports [port_name ] = ZeroMQPort (port_type , address , zmq_socket_type , _zmq_context )
8083 logger .info (f"Initialized ZMQ port: { port_name } ({ socket_type_str } ) on { address } " )
8184 except AttributeError :
8285 logger .error (f"Error: Invalid ZMQ socket type string '{ socket_type_str } '." )
@@ -86,23 +89,31 @@ def init_zmq_port(mod, port_name, port_type, address, socket_type_str):
8689 logger .error (f"An unexpected error occurred during ZMQ port initialization for { port_name } : { e } " )
8790
8891def terminate_zmq (mod ):
89- """Clean up all ZMQ sockets and contexts before exit ."""
92+ """Clean up all ZMQ sockets, then terminate the shared context once ."""
9093 if mod ._cleanup_in_progress :
9194 return # Already cleaning up, prevent reentrant calls
92-
95+
9396 if not mod .zmq_ports :
9497 return # No ports to clean up
95-
98+
9699 mod ._cleanup_in_progress = True
97100 print ("\n Cleaning up ZMQ resources..." )
101+
102+ # all sockets must be closed before context.term() is called.
98103 for port_name , port in mod .zmq_ports .items ():
99104 try :
100105 port .socket .close ()
101- port .context .term ()
102106 print (f"Closed ZMQ port: { port_name } " )
103107 except Exception as e :
104108 logger .error (f"Error while terminating ZMQ port { port .address } : { e } " )
105109 mod .zmq_ports .clear ()
110+
111+ # terminate the single shared context exactly once.
112+ try :
113+ _zmq_context .term ()
114+ except Exception as e :
115+ logger .error (f"Error while terminating shared ZMQ context: { e } " )
116+
106117 mod ._cleanup_in_progress = False
107118
108119# --- ZeroMQ Integration End ---
0 commit comments