Skip to content

fix: add atexit and signal handlers for ZMQ cleanup #206

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
pradeeban merged 3 commits into from
Feb 9, 2026
Merged

fix: add atexit and signal handlers for ZMQ cleanup #206

Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
ff3206a
fix: add atexit and signal handlers for ZMQ cleanup
Sahil-u07 Feb 4, 2026
b3268f3
fix: address Copilot review, prevent reentrant cleanup and platform-s…
Sahil-u07 Feb 4, 2026
6355f7b
Merge upstream/dev and resolve conflicts in concore.py
Sahil-u07 Feb 9, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
36 changes: 35 additions & 1 deletion concore.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -7,6 +7,8 @@
import re
import zmq
import numpy as np
import signal

logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
Expand Down Expand Up @@ -81,6 +83,7 @@ def recv_json_with_retry(self):

# Global ZeroMQ ports registry
zmq_ports = {}
_cleanup_in_progress = False

def init_zmq_port(port_name, port_type, address, socket_type_str):
"""
Expand All @@ -107,14 +110,45 @@ def init_zmq_port(port_name, port_type, address, socket_type_str):
logging.error(f"An unexpected error occurred during ZMQ port initialization for {port_name}: {e}")

def terminate_zmq():
for port in zmq_ports.values():
"""Clean up all ZMQ sockets and contexts before exit."""
global _cleanup_in_progress

if _cleanup_in_progress:
return # Already cleaning up, prevent reentrant calls

if not zmq_ports:
return # No ports to clean up

_cleanup_in_progress = True
print("\nCleaning up ZMQ resources...")
for port_name, port in zmq_ports.items():
try:
port.socket.close()
port.context.term()
print(f"Closed ZMQ port: {port_name}")
except Exception as e:
logging.error(f"Error while terminating ZMQ port {port.address}: {e}")
zmq_ports.clear()
Comment on lines 112 to +131
Copy link

Copilot AI Feb 4, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

If a signal arrives while terminate_zmq() is already executing (from normal program flow or atexit), the signal_handler will call terminate_zmq() again recursively. While the 'if not zmq_ports' guard at line 98 provides some protection, it won't help if the signal arrives after the check but during the iteration. Consider adding a boolean flag (e.g., '_cleanup_in_progress') to prevent reentrant calls to terminate_zmq().

Copilot uses AI. Check for mistakes.
_cleanup_in_progress = False

def signal_handler(sig, frame):
"""Handle interrupt signals gracefully."""
print(f"\nReceived signal {sig}, shutting down gracefully...")
Copy link

Copilot AI Feb 4, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The signal_handler calls sys.exit(0) which will trigger the atexit handler, resulting in terminate_zmq() being called twice. While the code guards against double cleanup with the 'if not zmq_ports' check, this creates unnecessary overhead and potential confusion. Consider either: (1) not calling sys.exit(0) in the signal handler and instead just returning, allowing the program to exit naturally and trigger atexit, or (2) unregistering the atexit handler before calling sys.exit in the signal handler using atexit.unregister(terminate_zmq).

Suggested change
print(f"\nReceived signal {sig}, shutting down gracefully...")
print(f"\nReceived signal {sig}, shutting down gracefully...")
# Prevent terminate_zmq from being called twice: once here and once via atexit
try:
atexit.unregister(terminate_zmq)
except Exception:
# If unregister fails for any reason, proceed with explicit cleanup anyway
pass

Copilot uses AI. Check for mistakes.
Copy link
Copy Markdown
Member

@pradeeban pradeeban Feb 8, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

@Sahil-u07 pls check this comment and see if it makes sense?

Copy link
Copy Markdown
Contributor Author

@Sahil-u07 Sahil-u07 Feb 9, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

@pradeeban
Copilot suggests two approaches:

  1. Just return from the signal handler - let the program exit naturally and trigger atexit
  2. Unregister atexit before calling sys.exit - prevents double cleanup

My current implementation uses approach #2, but with explicit cleanup:

try:
    atexit.unregister(terminate_zmq)
except Exception:
    pass
terminate_zmq()  # Explicit cleanup
sys.exit(0)

# Prevent terminate_zmq from being called twice: once here and once via atexit
try:
atexit.unregister(terminate_zmq)
except Exception:
# If unregister fails for any reason, proceed with explicit cleanup anyway
pass
terminate_zmq()
sys.exit(0)

# Register cleanup handlers
atexit.register(terminate_zmq)
signal.signal(signal.SIGINT, signal_handler) # Handle Ctrl+C
if not hasattr(sys, 'getwindowsversion'):
signal.signal(signal.SIGTERM, signal_handler) # Handle termination (Unix only)

# --- ZeroMQ Integration End ---


Expand Down
Loading