I am developing a program to simulate a P2P network using the multiprocessing package. Specifically, I have created a Node class that inherits from multiprocessing.Process and contains all the basic functionalities, while the SpecializedNode class implements specific features. SpecializedNode performs an operation for max_iter iterations and, at each iteration, sends the result to its peers (python broadcast_message(self, msg)) and collects all received messages before executing the task again (collect_messages(self)). To handle inter-process communication, each Node has a multiprocessing.Queue where it receives messages from other peers. Unfortunately, it seems that all processes get stuck when calling collect_messages(self), without raising any errors or exceptions. Since this is my first time using the multiprocessing package, I wonder if I am using it correctly. Moreover, does this implementation allow the node to correctly receive messages from peers while is performing the task in run()? from multiprocessing import Process, Queue class Node(Process): def __init__(self, node_id): super(Node, self).__init__() self.node_id = node_id # Incoming messages (e.g., models) queue self.message_queue = Queue() def set_peers(self, peers): self.peers = peers def collect_messages(self): messages = [] while not self.message_queue.empty(): msg = self.message_queue.get() messages.append(msg) return messages def broadcast_message(self, msg): for peer in self.peers: peer.message_queue.put(msg) Then I have my specialized Node class: class SpecializedNode(Node): def __init__(self, node_id, max_iter): super(node_id) self.max_iter = max_iter def run(self): for current_iter in range(self.max_iter): #do stuff msg = #build my message self.broadcast_msg(msg) incoming_messages = self.collect_messages() # do stuff based on the received messages if __name__ == "__main__": n_nodes = 30 # Create the nodes nodes = [SpecializedNode(i, 100) for i in range(n_nodes) # Simulate a fully-connected network for node in nodes: node.set_peers(peers=[n for n in nodes if n.node_id != node.node_id]) # Start the nodes (i.e., processes) for node in nodes: node.start() # Wait for completition for node in nodes: node.join() Continue reading...
