%% @doc Internal pub/sub for neural network component communication using pg. %% %% This module provides a thin wrapper around OTP's pg (process groups) %% for communication between network components (cortex, sensors, neurons, %% actuators). It uses pg's built-in group management with network-specific %% naming conventions. %% %% == Design Philosophy == %% %% Network components communicate through events rather than direct calls: %% - Publishers don't need to know subscriber PIDs %% - Subscribers don't need to know publisher PIDs %% - New observers can be added without modifying existing code %% - Events form a clear contract between components %% %% == Event Types == %% %% | Event | Publisher | Description | %% |-------|-----------|-------------| %% | evaluation_cycle_started | cortex | When sync is triggered | %% | sensor_output_ready | sensor | When sensor produces output | %% | neuron_output_ready | neuron | When neuron fires | %% | actuator_output_ready | actuator | When actuator produces output | %% | backup_requested | cortex | When weight backup is needed | %% | network_terminating | cortex | When network is shutting down | %% %% == Usage == %% %% Initialize pubsub for a network (typically in cortex): %% network_pubsub:init(NetworkId) %% %% Subscribe to events: %% network_pubsub:subscribe(NetworkId, evaluation_cycle_started) %% network_pubsub:subscribe(NetworkId, [sensor_output_ready, neuron_output_ready]) %% %% Publish events: %% network_pubsub:publish(NetworkId, evaluation_cycle_started, #{cycle => 1}) %% network_pubsub:publish(NetworkId, sensor_output_ready, #{from => SensorPid, signal => Signal}) %% %% Receive events in subscriber: %% receive %% {network_event, evaluation_cycle_started, Data} -> %% %% Handle evaluation start %% ... %% end %% %% == Implementation == %% %% Uses OTP pg (process groups) with group names of the form: %% {network_pubsub, NetworkId, Topic} %% %% This allows multiple network instances to have independent pubsub. %% %% @copyright 2025 Macula.io -module(network_pubsub). %% API -export([ %% Lifecycle init/1, cleanup/1, %% Subscription management subscribe/2, subscribe/3, unsubscribe/2, unsubscribe/3, get_subscribers/2, %% Publishing publish/3, %% Introspection list_topics/1 ]). %% Event types (for documentation) -type event_type() :: evaluation_cycle_started | sensor_output_ready | neuron_output_ready | actuator_output_ready | backup_requested | weights_backed_up | network_terminating | atom(). %% Allow custom events -type event_data() :: map(). -type network_id() :: term(). -export_type([event_type/0, event_data/0, network_id/0]). %% pg scope for network pubsub -define(SCOPE, network_pubsub_scope). %%============================================================================== %% API Functions %%============================================================================== %% @doc Initialize pubsub for a network instance. %% %% This ensures the pg scope is started. Safe to call multiple times. %% Typically called from cortex during initialization. -spec init(network_id()) -> ok. init(_NetworkId) -> %% Ensure pg scope is started case pg:start(?SCOPE) of {ok, _Pid} -> ok; {error, {already_started, _Pid}} -> ok end. %% @doc Cleanup pubsub for a network instance. %% %% Removes all subscriptions for the current process related to this network. %% Typically called during network shutdown. -spec cleanup(network_id()) -> ok. cleanup(NetworkId) -> %% Get all groups this process belongs to Groups = pg:which_groups(?SCOPE), %% Leave any groups that belong to this network lists:foreach( fun(Group) -> case Group of {network_pubsub, NetworkId, _Topic} -> pg:leave(?SCOPE, Group, self()); _ -> ok end end, Groups ), ok. %% @doc Subscribe calling process to event type(s). -spec subscribe(network_id(), event_type() | [event_type()]) -> ok. subscribe(NetworkId, Topics) when is_list(Topics) -> lists:foreach(fun(Topic) -> subscribe(NetworkId, Topic, self()) end, Topics); subscribe(NetworkId, Topic) -> subscribe(NetworkId, Topic, self()). %% @doc Subscribe a specific process to an event type. -spec subscribe(network_id(), event_type(), pid()) -> ok. subscribe(NetworkId, Topic, Pid) -> Group = make_group(NetworkId, Topic), pg:join(?SCOPE, Group, Pid). %% @doc Unsubscribe calling process from event type(s). -spec unsubscribe(network_id(), event_type() | [event_type()]) -> ok. unsubscribe(NetworkId, Topics) when is_list(Topics) -> lists:foreach(fun(Topic) -> unsubscribe(NetworkId, Topic, self()) end, Topics); unsubscribe(NetworkId, Topic) -> unsubscribe(NetworkId, Topic, self()). %% @doc Unsubscribe a specific process from an event type. -spec unsubscribe(network_id(), event_type(), pid()) -> ok. unsubscribe(NetworkId, Topic, Pid) -> Group = make_group(NetworkId, Topic), pg:leave(?SCOPE, Group, Pid). %% @doc Get list of subscribers for a topic. -spec get_subscribers(network_id(), event_type()) -> [pid()]. get_subscribers(NetworkId, Topic) -> Group = make_group(NetworkId, Topic), pg:get_members(?SCOPE, Group). %% @doc Publish an event to all subscribers. %% %% Sends {network_event, Topic, Data} to all processes subscribed to this topic. %% This is asynchronous - returns immediately after sending. -spec publish(network_id(), event_type(), event_data()) -> ok. publish(NetworkId, Topic, Data) -> Group = make_group(NetworkId, Topic), Subscribers = pg:get_members(?SCOPE, Group), Message = {network_event, Topic, Data}, lists:foreach( fun(Pid) -> Pid ! Message end, Subscribers ), ok. %% @doc List all topics with active subscriptions for a network. -spec list_topics(network_id()) -> [event_type()]. list_topics(NetworkId) -> Groups = pg:which_groups(?SCOPE), %% Filter to this network's groups and extract topics lists:filtermap( fun(Group) -> case Group of {network_pubsub, NetworkId, Topic} -> {true, Topic}; _ -> false end end, Groups ). %%============================================================================== %% Internal Functions %%============================================================================== %% @private Create a pg group name for a network topic -spec make_group(network_id(), event_type()) -> term(). make_group(NetworkId, Topic) -> {network_pubsub, NetworkId, Topic}.