maxwelllink.sockets.sockets module

Base socket hub for MaxwellLink drivers and servers.

DummySocketHub implements the full multi-client server machinery (accept loop, INIT binding, the step_barrier engine) and, following the package-wide Dummy* base-class pattern (DummyEMSimulation in em_solvers/, DummyModel in mxl_drivers/), every concrete hub inherits from it:

DummySocketHub                             this module (all machinery)
└── SocketHub                              this module (concrete default hub)
    ├── AggregatedSocketHub                aggregated.py (bridge transport)
    │   └── _AggregatedSusceptibilitySocketHubServer
    └── _SusceptibilitySocketHubServer     susceptibility hubs for Meep
                                           (see _meep_hub_base.py)

The wire formats (headers, framed arrays, the step_barrier fast path) live in protocol.py and are re-exported here for backward compatibility. The MPI helpers (am_master, mpi_bcast_from_master) and the host/port discovery used by Slurm workflows (get_available_host_port) also live here.

class maxwelllink.sockets.sockets.DummySocketHub[source]

Bases: object

Base socket server coordinating driver connections with an FDTD engine.

Following the package-wide Dummy* pattern (DummyEMSimulation in em_solvers/, DummyModel in mxl_drivers/), this base class implements the complete hub machinery; concrete hubs (SocketHub and its subclasses) inherit it and override only the hooks below. This server:

  • Accepts and tracks many driver connections.

  • Handles initialization handshakes, field dispatch, and result collection.

  • Provides a barrier-style step to send fields and receive source amplitudes from all registered molecules.

The accept thread starts during __init__; no separate start() call is needed.

Subclassing contract (followed by every hub in this package):

may be overridden

_handle_accepted (custom client classification; the accept loop itself should normally stay untouched), wait_until_bound / step_barrier (custom transport), stop (extra teardown; call super().stop())

do not override

the binding internals (_progress_binds_locked, _bind_client_locked) and the step_barrier fast path (_dispatch_field, _read_source_ready), which share scratch buffers and locking assumptions with step_barrier itself

Pinned invariants (relied upon by subclasses and tests — do not “clean up”):

  • _scratch_send/_scratch_recv/_scratch_recv_mv are plain instance attributes and _read_source_ready is a self-contained method (tests construct a bare hub and call it directly);

  • _lock is a reentrant lock: helper methods that acquire it are called from sections that already hold it;

  • on non-master MPI ranks __init__ deliberately skips the socket, selector, and lock attributes — only the master rank may call methods that touch them;

  • step_barrier returns {} (never raises) on pause, disconnect, or timeout, and keeps the frozen barrier for retry — callers’ retry loops depend on this.

__init__(host=None, port=31415, unixsocket=None, timeout=60000.0, latency=0.01)[source]

Initialize the socket hub.

Parameters:

  • host (str or None, default: None) – Host address for AF_INET sockets. Ignored when using a UNIX socket.

  • port (int or None, default: 31415) – TCP port for AF_INET sockets. Ignored for UNIX sockets.

  • unixsocket (str or None, default: None) – Path (or name under /tmp/socketmxl_*) for a UNIX domain socket. When provided, host and port are ignored.

  • timeout (float, default: 60000.0) – Socket timeout (seconds) for client operations.

  • latency (float, default: 0.01) – Polling sleep (seconds) between hub sweeps; can be very small for local runs.

addrmap: Dict[str, int]
all_bound(molecule_ids, require_init=True)[source]

Check if all given molecule IDs are bound (and optionally initialized).

Parameters:

  • molecule_ids (iterable of int) – Molecule IDs to check.

  • require_init (bool, default: True) – Also require that clients completed INIT.

Returns:

True if all are bound (and initialized if requested), else False.

Return type:

bool

bound: Dict[int, _ClientState]
clients: Dict[int, _ClientState]
expected: set[int]
graceful_shutdown(reason=None, wait=2.0)[source]

Politely ask all connected drivers to exit and wait briefly for BYE.

Parameters:

  • reason (str or None, optional) – Optional reason to log for shutdown.

  • wait (float, default: 2.0) – Seconds to wait for clean replies.

register_molecule(molecule_id)[source]

Reserve a slot for a given molecule ID (client may connect later).

Parameters:

molecule_id (int) – Molecule ID to register.

Raises:

ValueError – If the molecule ID is already registered.

Return type:

None

register_molecule_return_id()[source]

Reserve a slot for a molecule and return an auto-assigned ID.

Returns:

The assigned unique molecule ID.

Return type:

int

step_barrier(requests, timeout=None)[source]

Barrier step: dispatch fields and collect source amplitudes from all clients.

Coordinates sending fields, waiting for results, and jointly committing the results once every requested molecule is ready. A frozen barrier is reused if a disconnect occurs mid-step.

Parameters:

  • requests (dict[int, dict]) – Mapping from molecule ID to request dict with keys: - "efield_au" : array-like (3,) field vector in a.u. - "meta" : dict, optional metadata per send. - "init" : dict, optional INIT payload for first bind.

  • timeout (float, optional) – Maximum time (seconds) to wait for the barrier to complete. Defaults to the hub’s timeout setting.

Returns:

Mapping molid -> {"amp": ndarray(3,), "extra": bytes}. Returns {} when paused, on abort, or if the barrier is incomplete.

Return type:

dict[int, dict]

stop()[source]

Stop accepting new connections, request clients to exit, and close sockets.

Also removes the UNIX socket path if one was created.

wait_until_bound(init_payloads, require_init=True, timeout=None)[source]

Block until all requested molecule IDs are bound (and optionally initialized).

Parameters:

  • init_payloads (dict[int, dict]) – Mapping from molecule ID to INIT payload to use on bind.

  • require_init (bool, default: True) – Also require that clients completed INIT.

  • timeout (float or None, optional) – Maximum time to wait (seconds). When None (the default) this method waits indefinitely — the hub-wide self.timeout is not applied here.

Returns:

True if all requested IDs became bound within the time limit, else False.

Return type:

bool

class maxwelllink.sockets.sockets.SocketHub[source]

Bases: DummySocketHub

The concrete default socket hub used by MaxwellLink simulations.

All machinery is inherited unchanged from DummySocketHub; this subclass exists for naming symmetry with the rest of the package (every Dummy* base has a concrete counterpart) and is the class users instantiate:

from maxwelllink import SocketHub

hub = SocketHub(host="127.0.0.1", port=31415)
hub.wait_until_bound({0: {"molecule_id": 0}})
results = hub.step_barrier({0: {"efield_au": [0.0, 0.0, 1e-6]}})
hub.stop()

See DummySocketHub for the full method documentation and the subclassing contract.

__init__(host=None, port=31415, unixsocket=None, timeout=60000.0, latency=0.01)

Initialize the socket hub.

Parameters:

  • host (str or None, default: None) – Host address for AF_INET sockets. Ignored when using a UNIX socket.

  • port (int or None, default: 31415) – TCP port for AF_INET sockets. Ignored for UNIX sockets.

  • unixsocket (str or None, default: None) – Path (or name under /tmp/socketmxl_*) for a UNIX domain socket. When provided, host and port are ignored.

  • timeout (float, default: 60000.0) – Socket timeout (seconds) for client operations.

  • latency (float, default: 0.01) – Polling sleep (seconds) between hub sweeps; can be very small for local runs.

addrmap: Dict[str, int]
all_bound(molecule_ids, require_init=True)

Check if all given molecule IDs are bound (and optionally initialized).

Parameters:

  • molecule_ids (iterable of int) – Molecule IDs to check.

  • require_init (bool, default: True) – Also require that clients completed INIT.

Returns:

True if all are bound (and initialized if requested), else False.

Return type:

bool

bound: Dict[int, _ClientState]
clients: Dict[int, _ClientState]
expected: set[int]
graceful_shutdown(reason=None, wait=2.0)

Politely ask all connected drivers to exit and wait briefly for BYE.

Parameters:

  • reason (str or None, optional) – Optional reason to log for shutdown.

  • wait (float, default: 2.0) – Seconds to wait for clean replies.

register_molecule(molecule_id)

Reserve a slot for a given molecule ID (client may connect later).

Parameters:

molecule_id (int) – Molecule ID to register.

Raises:

ValueError – If the molecule ID is already registered.

Return type:

None

register_molecule_return_id()

Reserve a slot for a molecule and return an auto-assigned ID.

Returns:

The assigned unique molecule ID.

Return type:

int

step_barrier(requests, timeout=None)

Barrier step: dispatch fields and collect source amplitudes from all clients.

Coordinates sending fields, waiting for results, and jointly committing the results once every requested molecule is ready. A frozen barrier is reused if a disconnect occurs mid-step.

Parameters:

  • requests (dict[int, dict]) – Mapping from molecule ID to request dict with keys: - "efield_au" : array-like (3,) field vector in a.u. - "meta" : dict, optional metadata per send. - "init" : dict, optional INIT payload for first bind.

  • timeout (float, optional) – Maximum time (seconds) to wait for the barrier to complete. Defaults to the hub’s timeout setting.

Returns:

Mapping molid -> {"amp": ndarray(3,), "extra": bytes}. Returns {} when paused, on abort, or if the barrier is incomplete.

Return type:

dict[int, dict]

stop()

Stop accepting new connections, request clients to exit, and close sockets.

Also removes the UNIX socket path if one was created.

wait_until_bound(init_payloads, require_init=True, timeout=None)

Block until all requested molecule IDs are bound (and optionally initialized).

Parameters:

  • init_payloads (dict[int, dict]) – Mapping from molecule ID to INIT payload to use on bind.

  • require_init (bool, default: True) – Also require that clients completed INIT.

  • timeout (float or None, optional) – Maximum time to wait (seconds). When None (the default) this method waits indefinitely — the hub-wide self.timeout is not applied here.

Returns:

True if all requested IDs became bound within the time limit, else False.

Return type:

bool

maxwelllink.sockets.sockets.am_master()[source]

Return True if this process is the MPI master rank (rank 0), otherwise False.

Notes

Attempts to import mpi4py and query COMM_WORLD. If unavailable, returns True by treating the single process as rank 0.

maxwelllink.sockets.sockets.get_available_host_port(localhost=True, save_to_file=None)[source]

Ask the OS for an available localhost TCP port.

Parameters:

  • localhost (bool, default: True) – If True, bind to the localhost interface (“127.0.0.1”). If False, bind to all interfaces (“0.0.0.0”).

  • save_to_file (str or None, default: None) – If provided, save the selected host and port to the given file with filename provided by save_to_file. The first line contains the host, and the second line contains the port.

Returns:

(host, port) pair, e.g., ("127.0.0.1", 34567).

Return type:

tuple

maxwelllink.sockets.sockets.mpi_bcast_from_master(value)[source]

Broadcast a Python value from the master rank to all ranks via MPI.

Parameters:

value (any) – The value to broadcast.

Returns:

The broadcast value (unchanged when MPI is unavailable).

Return type:

any