p2pfl.communication.protocols.protobuff.gossiper module

Protocol agnostic gossiper.

class p2pfl.communication.protocols.protobuff.gossiper.Gossiper(neighbors, build_msg, period=None, messages_per_period=None)

Bases: Thread, NodeComponent

Gossiper for agnostic communication protocol.

Parameters:

• neighbors (Neighbors)

• build_msg (Callable[..., RootMessage])

• period (float | None)

• messages_per_period (int | None)

add_message(msg)

Add message to pending.

Parameters:

• msg (RootMessage) – Message to send.

• pending_neis – Neighbors to send the message.

Return type:

None

check_and_set_processed(msg)

Check if message was already processed and set it as processed.

Parameters:

msg (RootMessage) – Message to check.

Return type:

bool

gossip_weights(early_stopping_fn, get_candidates_fn, status_fn, model_fn, period, temporal_connection)

Gossip model weights. This is a synchronous gossip. End when there are no more neighbors to gossip.

Parameters:

• early_stopping_fn (Callable[[], bool]) – Function to check if the gossip should stop.

• get_candidates_fn (Callable[[], list[str]]) – Function to get the neighbors to gossip.

• status_fn (Callable[[], Any]) – Function to get the status of the node.

• model_fn (Callable[[str], tuple[Any, str, int, list[str]]]) – Function to get the model of a neighbor.

• period (float) – Period of gossip.

• temporal_connection (bool) – Flag to create a connection if neis not connected directly.

Return type:

None

run()

Run the gossiper thread.

Return type:

None

set_addr(addr)

Set the address.

Parameters:

addr (str)

Return type:

str

start()

Start the gossiper thread.

Return type:

None

stop()

Stop the gossiper thread.

Return type:

None