From 7d6daa8e10d3c75cd88b80c2f9f8fec2f3dfe37e Mon Sep 17 00:00:00 2001 From: NIC619 Date: Tue, 17 Dec 2019 17:17:03 +0800 Subject: [PATCH 01/15] Minor cleanup: - remove outdated comment - add new peer at the end - turn peers to send from list to set --- libp2p/pubsub/floodsub.py | 10 ++++++---- libp2p/pubsub/gossipsub.py | 2 -- libp2p/pubsub/pubsub.py | 8 ++------ 3 files changed, 8 insertions(+), 12 deletions(-) diff --git a/libp2p/pubsub/floodsub.py b/libp2p/pubsub/floodsub.py index bac0bd7..678e509 100644 --- a/libp2p/pubsub/floodsub.py +++ b/libp2p/pubsub/floodsub.py @@ -77,10 +77,12 @@ class FloodSub(IPubsubRouter): :param pubsub_msg: pubsub message in protobuf. """ - peers_gen = self._get_peers_to_send( - pubsub_msg.topicIDs, - msg_forwarder=msg_forwarder, - origin=ID(pubsub_msg.from_id), + peers_gen = set( + self._get_peers_to_send( + pubsub_msg.topicIDs, + msg_forwarder=msg_forwarder, + origin=ID(pubsub_msg.from_id), + ) ) rpc_msg = rpc_pb2.RPC(publish=[pubsub_msg]) diff --git a/libp2p/pubsub/gossipsub.py b/libp2p/pubsub/gossipsub.py index 045ef39..1969d4f 100644 --- a/libp2p/pubsub/gossipsub.py +++ b/libp2p/pubsub/gossipsub.py @@ -148,11 +148,9 @@ class GossipSub(IPubsubRouter): for topic in self.mesh: if peer_id in self.mesh[topic]: - # Delete the entry if no other peers left self.mesh[topic].remove(peer_id) for topic in self.fanout: if peer_id in self.fanout[topic]: - # Delete the entry if no other peers left self.fanout[topic].remove(peer_id) self.peers_to_protocol.pop(peer_id, None) diff --git a/libp2p/pubsub/pubsub.py b/libp2p/pubsub/pubsub.py index a44aa05..6e9d948 100644 --- a/libp2p/pubsub/pubsub.py +++ b/libp2p/pubsub/pubsub.py @@ -289,24 +289,22 @@ class Pubsub: logger.debug("fail to add new peer %s, error %s", peer_id, error) return - self.peers[peer_id] = stream - # Send hello packet hello = self.get_hello_packet() try: await stream.write(encode_varint_prefixed(hello.SerializeToString())) except StreamClosed: logger.debug("Fail to add new peer %s: stream closed", peer_id) - del self.peers[peer_id] return # TODO: Check if the peer in black list. try: self.router.add_peer(peer_id, stream.get_protocol()) except Exception as error: logger.debug("fail to add new peer %s, error %s", peer_id, error) - del self.peers[peer_id] return + self.peers[peer_id] = stream + logger.debug("added new peer %s", peer_id) def _handle_dead_peer(self, peer_id: ID) -> None: @@ -316,7 +314,6 @@ class Pubsub: for topic in self.peer_topics: if peer_id in self.peer_topics[topic]: - # Delete the entry if no other peers left self.peer_topics[topic].remove(peer_id) self.router.remove_peer(peer_id) @@ -360,7 +357,6 @@ class Pubsub: else: if sub_message.topicid in self.peer_topics: if origin_id in self.peer_topics[sub_message.topicid]: - # Delete the entry if no other peers left self.peer_topics[sub_message.topicid].remove(origin_id) # FIXME(mhchia): Change the function name? From f1d58ef8ffe85bf3e6bb3140080f7709b9b2b152 Mon Sep 17 00:00:00 2001 From: NIC619 Date: Tue, 17 Dec 2019 17:30:24 +0800 Subject: [PATCH 02/15] Change type of peers from list to set: `peers_gossipsub`, `peers_floodsub` and mesh/fanout peers --- libp2p/pubsub/gossipsub.py | 42 +++++++++++++++++++------------------- 1 file changed, 21 insertions(+), 21 deletions(-) diff --git a/libp2p/pubsub/gossipsub.py b/libp2p/pubsub/gossipsub.py index 1969d4f..7ef1372 100644 --- a/libp2p/pubsub/gossipsub.py +++ b/libp2p/pubsub/gossipsub.py @@ -3,7 +3,7 @@ import asyncio from collections import defaultdict import logging import random -from typing import Any, DefaultDict, Dict, Iterable, List, Sequence, Tuple +from typing import Any, DefaultDict, Dict, Iterable, List, Sequence, Set, Tuple from libp2p.network.stream.exceptions import StreamClosed from libp2p.peer.id import ID @@ -32,15 +32,15 @@ class GossipSub(IPubsubRouter): time_to_live: int - mesh: Dict[str, List[ID]] - fanout: Dict[str, List[ID]] + mesh: Dict[str, Set[ID]] + fanout: Dict[str, Set[ID]] peers_to_protocol: Dict[ID, str] time_since_last_publish: Dict[str, int] - peers_gossipsub: List[ID] - peers_floodsub: List[ID] + peers_gossipsub: Set[ID] + peers_floodsub: Set[ID] mcache: MessageCache @@ -80,8 +80,8 @@ class GossipSub(IPubsubRouter): # Create topic --> time since last publish map self.time_since_last_publish = {} - self.peers_gossipsub = [] - self.peers_floodsub = [] + self.peers_gossipsub = set() + self.peers_floodsub = set() # Create message cache self.mcache = MessageCache(gossip_window, gossip_history) @@ -122,9 +122,9 @@ class GossipSub(IPubsubRouter): logger.debug("adding peer %s with protocol %s", peer_id, protocol_id) if protocol_id == PROTOCOL_ID: - self.peers_gossipsub.append(peer_id) + self.peers_gossipsub.add(peer_id) elif protocol_id == floodsub.PROTOCOL_ID: - self.peers_floodsub.append(peer_id) + self.peers_floodsub.add(peer_id) else: # We should never enter here. Becuase the `protocol_id` is registered by your pubsub # instance in multistream-select, but it is not the protocol that gossipsub supports. @@ -142,16 +142,16 @@ class GossipSub(IPubsubRouter): logger.debug("removing peer %s", peer_id) if peer_id in self.peers_gossipsub: - self.peers_gossipsub.remove(peer_id) + self.peers_gossipsub.discard(peer_id) elif peer_id in self.peers_floodsub: - self.peers_floodsub.remove(peer_id) + self.peers_floodsub.discard(peer_id) for topic in self.mesh: if peer_id in self.mesh[topic]: - self.mesh[topic].remove(peer_id) + self.mesh[topic].discard(peer_id) for topic in self.fanout: if peer_id in self.fanout[topic]: - self.fanout[topic].remove(peer_id) + self.fanout[topic].discard(peer_id) self.peers_to_protocol.pop(peer_id, None) @@ -246,7 +246,7 @@ class GossipSub(IPubsubRouter): fanout_peers += self._get_in_topic_gossipsub_peers_from_minus( topic, self.degree - fanout_size, fanout_peers ) - self.fanout[topic] = fanout_peers + self.fanout[topic] = set(fanout_peers) gossipsub_peers = fanout_peers send_to.extend(floodsub_peers + gossipsub_peers) # Excludes `msg_forwarder` and `origin` @@ -282,7 +282,7 @@ class GossipSub(IPubsubRouter): # Add fanout peers to mesh and notifies them with a GRAFT(topic) control message. for peer in fanout_peers: - self.mesh[topic].append(peer) + self.mesh[topic].add(peer) await self.emit_graft(topic, peer) self.fanout.pop(topic, None) @@ -419,7 +419,7 @@ class GossipSub(IPubsubRouter): for peer in selected_peers: # Add peer to mesh[topic] - self.mesh[topic].append(peer) + self.mesh[topic].add(peer) # Emit GRAFT(topic) control message to peer peers_to_graft[peer].append(topic) @@ -431,7 +431,7 @@ class GossipSub(IPubsubRouter): ) for peer in selected_peers: # Remove peer from mesh[topic] - self.mesh[topic].remove(peer) + self.mesh[topic].discard(peer) # Emit PRUNE(topic) control message to peer peers_to_prune[peer].append(topic) @@ -458,7 +458,7 @@ class GossipSub(IPubsubRouter): for peer in self.fanout[topic] if peer in self.pubsub.peer_topics[topic] ] - self.fanout[topic] = in_topic_fanout_peers + self.fanout[topic] = set(in_topic_fanout_peers) num_fanout_peers_in_topic = len(self.fanout[topic]) # If |fanout[topic]| < D @@ -470,7 +470,7 @@ class GossipSub(IPubsubRouter): self.fanout[topic], ) # Add the peers to fanout[topic] - self.fanout[topic].extend(selected_peers) + self.fanout[topic].Update(selected_peers) def gossip_heartbeat(self) -> DefaultDict[ID, Dict[str, List[str]]]: peers_to_gossip: DefaultDict[ID, Dict[str, List[str]]] = defaultdict(dict) @@ -621,7 +621,7 @@ class GossipSub(IPubsubRouter): # Add peer to mesh for topic if topic in self.mesh: if sender_peer_id not in self.mesh[topic]: - self.mesh[topic].append(sender_peer_id) + self.mesh[topic].add(sender_peer_id) else: # Respond with PRUNE if not subscribed to the topic await self.emit_prune(topic, sender_peer_id) @@ -633,7 +633,7 @@ class GossipSub(IPubsubRouter): # Remove peer from mesh for topic, if peer is in topic if topic in self.mesh and sender_peer_id in self.mesh[topic]: - self.mesh[topic].remove(sender_peer_id) + self.mesh[topic].discard(sender_peer_id) # RPC emitters From 65766ec9aca3a21eac7243ec834fd71f19521747 Mon Sep 17 00:00:00 2001 From: NIC619 Date: Tue, 17 Dec 2019 17:36:15 +0800 Subject: [PATCH 03/15] Change type of local peers var from list to set --- libp2p/pubsub/gossipsub.py | 29 ++++++++++++++++------------- 1 file changed, 16 insertions(+), 13 deletions(-) diff --git a/libp2p/pubsub/gossipsub.py b/libp2p/pubsub/gossipsub.py index 7ef1372..6128794 100644 --- a/libp2p/pubsub/gossipsub.py +++ b/libp2p/pubsub/gossipsub.py @@ -213,7 +213,7 @@ class GossipSub(IPubsubRouter): :param origin: peer id of the peer the message originate from. :return: a generator of the peer ids who we send data to. """ - send_to: List[ID] = [] + send_to: Set[ID] = set() for topic in topic_ids: if topic not in self.pubsub.peer_topics: continue @@ -221,14 +221,15 @@ class GossipSub(IPubsubRouter): # floodsub peers # FIXME: `gossipsub.peers_floodsub` can be changed to `gossipsub.peers` in go. # This will improve the efficiency when searching for a peer's protocol id. - floodsub_peers: List[ID] = [ + floodsub_peers: Set[ID] = set( peer_id for peer_id in self.pubsub.peer_topics[topic] if peer_id in self.peers_floodsub - ] + ) + send_to.update(floodsub_peers) # gossipsub peers - gossipsub_peers: List[ID] = [] + gossipsub_peers: Set[ID] = set() if topic in self.mesh: gossipsub_peers = self.mesh[topic] else: @@ -236,21 +237,23 @@ class GossipSub(IPubsubRouter): # `self.degree` number of peers who have subscribed to the topic and add them # as our `fanout` peers. topic_in_fanout: bool = topic in self.fanout - fanout_peers: List[ID] = self.fanout[topic] if topic_in_fanout else [] + fanout_peers: Set[ID] = self.fanout[topic] if topic_in_fanout else set() fanout_size = len(fanout_peers) if not topic_in_fanout or ( topic_in_fanout and fanout_size < self.degree ): if topic in self.pubsub.peer_topics: # Combine fanout peers with selected peers - fanout_peers += self._get_in_topic_gossipsub_peers_from_minus( - topic, self.degree - fanout_size, fanout_peers + fanout_peers.update( + self._get_in_topic_gossipsub_peers_from_minus( + topic, self.degree - fanout_size, fanout_peers + ) ) - self.fanout[topic] = set(fanout_peers) + self.fanout[topic] = fanout_peers gossipsub_peers = fanout_peers - send_to.extend(floodsub_peers + gossipsub_peers) + send_to.update(gossipsub_peers) # Excludes `msg_forwarder` and `origin` - yield from set(send_to).difference([msg_forwarder, origin]) + yield from send_to.difference([msg_forwarder, origin]) async def join(self, topic: str) -> None: """ @@ -264,10 +267,10 @@ class GossipSub(IPubsubRouter): if topic in self.mesh: return # Create mesh[topic] if it does not yet exist - self.mesh[topic] = [] + self.mesh[topic] = set() topic_in_fanout: bool = topic in self.fanout - fanout_peers: List[ID] = self.fanout[topic] if topic_in_fanout else [] + fanout_peers: Set[ID] = self.fanout[topic] if topic_in_fanout else set() fanout_size = len(fanout_peers) if not topic_in_fanout or (topic_in_fanout and fanout_size < self.degree): # There are less than D peers (let this number be x) @@ -278,7 +281,7 @@ class GossipSub(IPubsubRouter): topic, self.degree - fanout_size, fanout_peers ) # Combine fanout peers with selected peers - fanout_peers += selected_peers + fanout_peers.update(selected_peers) # Add fanout peers to mesh and notifies them with a GRAFT(topic) control message. for peer in fanout_peers: From b4bd997932ff5ee78a7f4f3c63e7e5cf1dddc397 Mon Sep 17 00:00:00 2001 From: NIC619 Date: Tue, 17 Dec 2019 17:49:49 +0800 Subject: [PATCH 04/15] Fix mypy --- libp2p/pubsub/gossipsub.py | 18 +++++++++--------- 1 file changed, 9 insertions(+), 9 deletions(-) diff --git a/libp2p/pubsub/gossipsub.py b/libp2p/pubsub/gossipsub.py index 6128794..bddafa0 100644 --- a/libp2p/pubsub/gossipsub.py +++ b/libp2p/pubsub/gossipsub.py @@ -246,7 +246,7 @@ class GossipSub(IPubsubRouter): # Combine fanout peers with selected peers fanout_peers.update( self._get_in_topic_gossipsub_peers_from_minus( - topic, self.degree - fanout_size, fanout_peers + topic, self.degree - fanout_size, list(fanout_peers) ) ) self.fanout[topic] = fanout_peers @@ -278,7 +278,7 @@ class GossipSub(IPubsubRouter): # Selects the remaining number of peers (D-x) from peers.gossipsub[topic]. if topic in self.pubsub.peer_topics: selected_peers = self._get_in_topic_gossipsub_peers_from_minus( - topic, self.degree - fanout_size, fanout_peers + topic, self.degree - fanout_size, list(fanout_peers) ) # Combine fanout peers with selected peers fanout_peers.update(selected_peers) @@ -417,7 +417,7 @@ class GossipSub(IPubsubRouter): if num_mesh_peers_in_topic < self.degree_low: # Select D - |mesh[topic]| peers from peers.gossipsub[topic] - mesh[topic] selected_peers = self._get_in_topic_gossipsub_peers_from_minus( - topic, self.degree - num_mesh_peers_in_topic, self.mesh[topic] + topic, self.degree - num_mesh_peers_in_topic, list(self.mesh[topic]) ) for peer in selected_peers: @@ -430,7 +430,7 @@ class GossipSub(IPubsubRouter): if num_mesh_peers_in_topic > self.degree_high: # Select |mesh[topic]| - D peers from mesh[topic] selected_peers = GossipSub.select_from_minus( - num_mesh_peers_in_topic - self.degree, self.mesh[topic], [] + num_mesh_peers_in_topic - self.degree, list(self.mesh[topic]), [] ) for peer in selected_peers: # Remove peer from mesh[topic] @@ -470,10 +470,10 @@ class GossipSub(IPubsubRouter): selected_peers = self._get_in_topic_gossipsub_peers_from_minus( topic, self.degree - num_fanout_peers_in_topic, - self.fanout[topic], + list(self.fanout[topic]), ) # Add the peers to fanout[topic] - self.fanout[topic].Update(selected_peers) + self.fanout[topic].update(selected_peers) def gossip_heartbeat(self) -> DefaultDict[ID, Dict[str, List[str]]]: peers_to_gossip: DefaultDict[ID, Dict[str, List[str]]] = defaultdict(dict) @@ -484,7 +484,7 @@ class GossipSub(IPubsubRouter): if topic in self.pubsub.peer_topics: # Select D peers from peers.gossipsub[topic] peers_to_emit_ihave_to = self._get_in_topic_gossipsub_peers_from_minus( - topic, self.degree, self.mesh[topic] + topic, self.degree, list(self.mesh[topic]) ) msg_id_strs = [str(msg_id) for msg_id in msg_ids] @@ -500,7 +500,7 @@ class GossipSub(IPubsubRouter): if topic in self.pubsub.peer_topics: # Select D peers from peers.gossipsub[topic] peers_to_emit_ihave_to = self._get_in_topic_gossipsub_peers_from_minus( - topic, self.degree, self.fanout[topic] + topic, self.degree, list(self.fanout[topic]) ) msg_id_strs = [str(msg) for msg in msg_ids] for peer in peers_to_emit_ihave_to: @@ -528,7 +528,7 @@ class GossipSub(IPubsubRouter): # If num_to_select > size(selection_pool), then return selection_pool (which has the most # possible elements s.t. the number of elements is less than num_to_select) - if num_to_select > len(selection_pool): + if num_to_select >= len(selection_pool): return selection_pool # Random selection From f10e3099cb04cd2b475759b283d0038b2fb02d7c Mon Sep 17 00:00:00 2001 From: NIC619 Date: Tue, 17 Dec 2019 17:55:13 +0800 Subject: [PATCH 05/15] Change type of peers in pubsub from list to set --- libp2p/pubsub/pubsub.py | 9 +++++---- 1 file changed, 5 insertions(+), 4 deletions(-) diff --git a/libp2p/pubsub/pubsub.py b/libp2p/pubsub/pubsub.py index 6e9d948..2d5625f 100644 --- a/libp2p/pubsub/pubsub.py +++ b/libp2p/pubsub/pubsub.py @@ -8,6 +8,7 @@ from typing import ( Dict, List, NamedTuple, + Set, Tuple, Union, cast, @@ -73,7 +74,7 @@ class Pubsub: my_topics: Dict[str, "asyncio.Queue[rpc_pb2.Message]"] - peer_topics: Dict[str, List[ID]] + peer_topics: Dict[str, Set[ID]] peers: Dict[ID, INetStream] topic_validators: Dict[str, TopicValidator] @@ -314,7 +315,7 @@ class Pubsub: for topic in self.peer_topics: if peer_id in self.peer_topics[topic]: - self.peer_topics[topic].remove(peer_id) + self.peer_topics[topic].discard(peer_id) self.router.remove_peer(peer_id) @@ -353,11 +354,11 @@ class Pubsub: self.peer_topics[sub_message.topicid] = [origin_id] elif origin_id not in self.peer_topics[sub_message.topicid]: # Add peer to topic - self.peer_topics[sub_message.topicid].append(origin_id) + self.peer_topics[sub_message.topicid].add(origin_id) else: if sub_message.topicid in self.peer_topics: if origin_id in self.peer_topics[sub_message.topicid]: - self.peer_topics[sub_message.topicid].remove(origin_id) + self.peer_topics[sub_message.topicid].discard(origin_id) # FIXME(mhchia): Change the function name? async def handle_talk(self, publish_message: rpc_pb2.Message) -> None: From 009df257bcd1b5bc15a86ea8f21cd27508db59ea Mon Sep 17 00:00:00 2001 From: NIC619 Date: Tue, 17 Dec 2019 18:47:58 +0800 Subject: [PATCH 06/15] Check peer id exist in dict before access --- libp2p/pubsub/floodsub.py | 2 ++ libp2p/pubsub/gossipsub.py | 9 +++++++++ 2 files changed, 11 insertions(+) diff --git a/libp2p/pubsub/floodsub.py b/libp2p/pubsub/floodsub.py index 678e509..b7b7910 100644 --- a/libp2p/pubsub/floodsub.py +++ b/libp2p/pubsub/floodsub.py @@ -89,6 +89,8 @@ class FloodSub(IPubsubRouter): logger.debug("publishing message %s", pubsub_msg) for peer_id in peers_gen: + if peer_id not in self.pubsub.peers: + continue stream = self.pubsub.peers[peer_id] # FIXME: We should add a `WriteMsg` similar to write delimited messages. # Ref: https://github.com/libp2p/go-libp2p-pubsub/blob/master/comm.go#L107 diff --git a/libp2p/pubsub/gossipsub.py b/libp2p/pubsub/gossipsub.py index bddafa0..a64b9ca 100644 --- a/libp2p/pubsub/gossipsub.py +++ b/libp2p/pubsub/gossipsub.py @@ -193,6 +193,8 @@ class GossipSub(IPubsubRouter): logger.debug("publishing message %s", pubsub_msg) for peer_id in peers_gen: + if peer_id not in self.pubsub.peers: + continue stream = self.pubsub.peers[peer_id] # FIXME: We should add a `WriteMsg` similar to write delimited messages. # Ref: https://github.com/libp2p/go-libp2p-pubsub/blob/master/comm.go#L107 @@ -604,6 +606,11 @@ class GossipSub(IPubsubRouter): rpc_msg: bytes = packet.SerializeToString() # 3) Get the stream to this peer + if sender_peer_id not in self.pubsub.peers: + logger.debug( + "Fail to responed to iwant request from %s: peer disconnected", + sender_peer_id, + ) peer_stream = self.pubsub.peers[sender_peer_id] # 4) And write the packet to the stream @@ -710,6 +717,8 @@ class GossipSub(IPubsubRouter): rpc_msg: bytes = packet.SerializeToString() # Get stream for peer from pubsub + if to_peer not in self.pubsub.peers: + logger.debug("Fail to emit control message to %s: peer disconnected", to_peer) peer_stream = self.pubsub.peers[to_peer] # Write rpc to stream From 474ed41652fc7a5f3439cec4d35ced98d14fa5b8 Mon Sep 17 00:00:00 2001 From: NIC619 Date: Tue, 17 Dec 2019 18:48:25 +0800 Subject: [PATCH 07/15] Remove dead peer if floodsub write stream fail --- libp2p/pubsub/floodsub.py | 1 + 1 file changed, 1 insertion(+) diff --git a/libp2p/pubsub/floodsub.py b/libp2p/pubsub/floodsub.py index b7b7910..bae2bf2 100644 --- a/libp2p/pubsub/floodsub.py +++ b/libp2p/pubsub/floodsub.py @@ -98,6 +98,7 @@ class FloodSub(IPubsubRouter): await stream.write(encode_varint_prefixed(rpc_msg.SerializeToString())) except StreamClosed: logger.debug("Fail to publish message to %s: stream closed", peer_id) + self.pubsub._handle_dead_peer(peer_id) async def join(self, topic: str) -> None: """ From 04b9d688f8ed55e3d3bb35d912dec9a9702fd399 Mon Sep 17 00:00:00 2001 From: NIC619 Date: Tue, 17 Dec 2019 19:09:15 +0800 Subject: [PATCH 08/15] Add newsfragment --- newsfragments/387.bugfix.rst | 1 + 1 file changed, 1 insertion(+) create mode 100644 newsfragments/387.bugfix.rst diff --git a/newsfragments/387.bugfix.rst b/newsfragments/387.bugfix.rst new file mode 100644 index 0000000..0ba3f07 --- /dev/null +++ b/newsfragments/387.bugfix.rst @@ -0,0 +1 @@ +Store peer ids in ``set`` instead of ``list`` and check if peer id exists in ``dict`` before accessing to prevent ``KeyError``. \ No newline at end of file From 19ce8a2140e0c00a1273c36f1bf6282f89f98e32 Mon Sep 17 00:00:00 2001 From: NIC619 Date: Tue, 17 Dec 2019 21:56:02 +0800 Subject: [PATCH 09/15] Fix mypy --- libp2p/pubsub/gossipsub.py | 2 ++ libp2p/pubsub/pubsub.py | 2 +- 2 files changed, 3 insertions(+), 1 deletion(-) diff --git a/libp2p/pubsub/gossipsub.py b/libp2p/pubsub/gossipsub.py index a64b9ca..55029cd 100644 --- a/libp2p/pubsub/gossipsub.py +++ b/libp2p/pubsub/gossipsub.py @@ -611,6 +611,7 @@ class GossipSub(IPubsubRouter): "Fail to responed to iwant request from %s: peer disconnected", sender_peer_id, ) + return peer_stream = self.pubsub.peers[sender_peer_id] # 4) And write the packet to the stream @@ -719,6 +720,7 @@ class GossipSub(IPubsubRouter): # Get stream for peer from pubsub if to_peer not in self.pubsub.peers: logger.debug("Fail to emit control message to %s: peer disconnected", to_peer) + return peer_stream = self.pubsub.peers[to_peer] # Write rpc to stream diff --git a/libp2p/pubsub/pubsub.py b/libp2p/pubsub/pubsub.py index 2d5625f..48ed7eb 100644 --- a/libp2p/pubsub/pubsub.py +++ b/libp2p/pubsub/pubsub.py @@ -351,7 +351,7 @@ class Pubsub: """ if sub_message.subscribe: if sub_message.topicid not in self.peer_topics: - self.peer_topics[sub_message.topicid] = [origin_id] + self.peer_topics[sub_message.topicid] = set([origin_id]) elif origin_id not in self.peer_topics[sub_message.topicid]: # Add peer to topic self.peer_topics[sub_message.topicid].add(origin_id) From f3732f94809e2ae7314641bcb066be6222370d9d Mon Sep 17 00:00:00 2001 From: NIC619 Date: Wed, 18 Dec 2019 12:37:04 +0800 Subject: [PATCH 10/15] Fix tests --- libp2p/pubsub/gossipsub.py | 4 +++- tests/pubsub/test_gossipsub.py | 16 ++++++++-------- tests_interop/test_pubsub.py | 2 +- 3 files changed, 12 insertions(+), 10 deletions(-) diff --git a/libp2p/pubsub/gossipsub.py b/libp2p/pubsub/gossipsub.py index 55029cd..51e1c92 100644 --- a/libp2p/pubsub/gossipsub.py +++ b/libp2p/pubsub/gossipsub.py @@ -719,7 +719,9 @@ class GossipSub(IPubsubRouter): # Get stream for peer from pubsub if to_peer not in self.pubsub.peers: - logger.debug("Fail to emit control message to %s: peer disconnected", to_peer) + logger.debug( + "Fail to emit control message to %s: peer disconnected", to_peer + ) return peer_stream = self.pubsub.peers[to_peer] diff --git a/tests/pubsub/test_gossipsub.py b/tests/pubsub/test_gossipsub.py index 19ec07c..fe6bf3b 100644 --- a/tests/pubsub/test_gossipsub.py +++ b/tests/pubsub/test_gossipsub.py @@ -390,15 +390,15 @@ async def test_mesh_heartbeat( fake_peer_ids = [ ID((i).to_bytes(2, byteorder="big")) for i in range(total_peer_count) ] - monkeypatch.setattr(pubsubs_gsub[0].router, "peers_gossipsub", fake_peer_ids) + monkeypatch.setattr(pubsubs_gsub[0].router, "peers_gossipsub", set(fake_peer_ids)) - peer_topics = {topic: fake_peer_ids} + peer_topics = {topic: set(fake_peer_ids)} # Monkeypatch the peer subscriptions monkeypatch.setattr(pubsubs_gsub[0], "peer_topics", peer_topics) mesh_peer_indices = random.sample(range(total_peer_count), initial_mesh_peer_count) mesh_peers = [fake_peer_ids[i] for i in mesh_peer_indices] - router_mesh = {topic: list(mesh_peers)} + router_mesh = {topic: set(mesh_peers)} # Monkeypatch our mesh peers monkeypatch.setattr(pubsubs_gsub[0].router, "mesh", router_mesh) @@ -437,27 +437,27 @@ async def test_gossip_heartbeat( fake_peer_ids = [ ID((i).to_bytes(2, byteorder="big")) for i in range(total_peer_count) ] - monkeypatch.setattr(pubsubs_gsub[0].router, "peers_gossipsub", fake_peer_ids) + monkeypatch.setattr(pubsubs_gsub[0].router, "peers_gossipsub", set(fake_peer_ids)) topic_mesh_peer_count = 14 # Split into mesh peers and fanout peers peer_topics = { - topic_mesh: fake_peer_ids[:topic_mesh_peer_count], - topic_fanout: fake_peer_ids[topic_mesh_peer_count:], + topic_mesh: set(fake_peer_ids[:topic_mesh_peer_count]), + topic_fanout: set(fake_peer_ids[topic_mesh_peer_count:]), } # Monkeypatch the peer subscriptions monkeypatch.setattr(pubsubs_gsub[0], "peer_topics", peer_topics) mesh_peer_indices = random.sample(range(topic_mesh_peer_count), initial_peer_count) mesh_peers = [fake_peer_ids[i] for i in mesh_peer_indices] - router_mesh = {topic_mesh: list(mesh_peers)} + router_mesh = {topic_mesh: set(mesh_peers)} # Monkeypatch our mesh peers monkeypatch.setattr(pubsubs_gsub[0].router, "mesh", router_mesh) fanout_peer_indices = random.sample( range(topic_mesh_peer_count, total_peer_count), initial_peer_count ) fanout_peers = [fake_peer_ids[i] for i in fanout_peer_indices] - router_fanout = {topic_fanout: list(fanout_peers)} + router_fanout = {topic_fanout: set(fanout_peers)} # Monkeypatch our fanout peers monkeypatch.setattr(pubsubs_gsub[0].router, "fanout", router_fanout) diff --git a/tests_interop/test_pubsub.py b/tests_interop/test_pubsub.py index f67b47a..db42c7c 100644 --- a/tests_interop/test_pubsub.py +++ b/tests_interop/test_pubsub.py @@ -99,7 +99,7 @@ async def test_pubsub(pubsubs, p2pds): go_0_topic_1_peers = await p2pds[0].control.pubsub_list_peers(TOPIC_1) assert len(go_0_topic_1_peers) == 1 and py_peer_id == go_0_topic_1_peers[0] # py - py_topic_0_peers = py_pubsub.peer_topics[TOPIC_0] + py_topic_0_peers = list(py_pubsub.peer_topics[TOPIC_0]) assert len(py_topic_0_peers) == 1 and p2pds[0].peer_id == py_topic_0_peers[0] # go_1 go_1_topic_1_peers = await p2pds[1].control.pubsub_list_peers(TOPIC_1) From 6cd3eb8faec2e60ca053356a37e58cf51399026d Mon Sep 17 00:00:00 2001 From: NIC619 Date: Thu, 19 Dec 2019 14:15:51 +0800 Subject: [PATCH 11/15] Apply PR feedback: change param type and remove check before `discard` --- libp2p/pubsub/gossipsub.py | 40 +++++++++++++++++--------------------- 1 file changed, 18 insertions(+), 22 deletions(-) diff --git a/libp2p/pubsub/gossipsub.py b/libp2p/pubsub/gossipsub.py index 51e1c92..7b0759c 100644 --- a/libp2p/pubsub/gossipsub.py +++ b/libp2p/pubsub/gossipsub.py @@ -141,17 +141,13 @@ class GossipSub(IPubsubRouter): """ logger.debug("removing peer %s", peer_id) - if peer_id in self.peers_gossipsub: - self.peers_gossipsub.discard(peer_id) - elif peer_id in self.peers_floodsub: - self.peers_floodsub.discard(peer_id) + self.peers_gossipsub.discard(peer_id) + self.peers_floodsub.discard(peer_id) for topic in self.mesh: - if peer_id in self.mesh[topic]: - self.mesh[topic].discard(peer_id) + self.mesh[topic].discard(peer_id) for topic in self.fanout: - if peer_id in self.fanout[topic]: - self.fanout[topic].discard(peer_id) + self.fanout[topic].discard(peer_id) self.peers_to_protocol.pop(peer_id, None) @@ -248,7 +244,7 @@ class GossipSub(IPubsubRouter): # Combine fanout peers with selected peers fanout_peers.update( self._get_in_topic_gossipsub_peers_from_minus( - topic, self.degree - fanout_size, list(fanout_peers) + topic, self.degree - fanout_size, fanout_peers ) ) self.fanout[topic] = fanout_peers @@ -280,7 +276,7 @@ class GossipSub(IPubsubRouter): # Selects the remaining number of peers (D-x) from peers.gossipsub[topic]. if topic in self.pubsub.peer_topics: selected_peers = self._get_in_topic_gossipsub_peers_from_minus( - topic, self.degree - fanout_size, list(fanout_peers) + topic, self.degree - fanout_size, fanout_peers ) # Combine fanout peers with selected peers fanout_peers.update(selected_peers) @@ -419,7 +415,7 @@ class GossipSub(IPubsubRouter): if num_mesh_peers_in_topic < self.degree_low: # Select D - |mesh[topic]| peers from peers.gossipsub[topic] - mesh[topic] selected_peers = self._get_in_topic_gossipsub_peers_from_minus( - topic, self.degree - num_mesh_peers_in_topic, list(self.mesh[topic]) + topic, self.degree - num_mesh_peers_in_topic, self.mesh[topic] ) for peer in selected_peers: @@ -432,7 +428,7 @@ class GossipSub(IPubsubRouter): if num_mesh_peers_in_topic > self.degree_high: # Select |mesh[topic]| - D peers from mesh[topic] selected_peers = GossipSub.select_from_minus( - num_mesh_peers_in_topic - self.degree, list(self.mesh[topic]), [] + num_mesh_peers_in_topic - self.degree, self.mesh[topic], set() ) for peer in selected_peers: # Remove peer from mesh[topic] @@ -472,7 +468,7 @@ class GossipSub(IPubsubRouter): selected_peers = self._get_in_topic_gossipsub_peers_from_minus( topic, self.degree - num_fanout_peers_in_topic, - list(self.fanout[topic]), + self.fanout[topic], ) # Add the peers to fanout[topic] self.fanout[topic].update(selected_peers) @@ -486,7 +482,7 @@ class GossipSub(IPubsubRouter): if topic in self.pubsub.peer_topics: # Select D peers from peers.gossipsub[topic] peers_to_emit_ihave_to = self._get_in_topic_gossipsub_peers_from_minus( - topic, self.degree, list(self.mesh[topic]) + topic, self.degree, self.mesh[topic] ) msg_id_strs = [str(msg_id) for msg_id in msg_ids] @@ -502,7 +498,7 @@ class GossipSub(IPubsubRouter): if topic in self.pubsub.peer_topics: # Select D peers from peers.gossipsub[topic] peers_to_emit_ihave_to = self._get_in_topic_gossipsub_peers_from_minus( - topic, self.degree, list(self.fanout[topic]) + topic, self.degree, self.fanout[topic] ) msg_id_strs = [str(msg) for msg in msg_ids] for peer in peers_to_emit_ihave_to: @@ -511,7 +507,7 @@ class GossipSub(IPubsubRouter): @staticmethod def select_from_minus( - num_to_select: int, pool: Sequence[Any], minus: Sequence[Any] + num_to_select: int, pool: Iterable[Any], minus: Iterable[Any] ) -> List[Any]: """ Select at most num_to_select subset of elements from the set (pool - minus) randomly. @@ -539,15 +535,15 @@ class GossipSub(IPubsubRouter): return selection def _get_in_topic_gossipsub_peers_from_minus( - self, topic: str, num_to_select: int, minus: Sequence[ID] + self, topic: str, num_to_select: int, minus: Iterable[ID] ) -> List[ID]: - gossipsub_peers_in_topic = [ + gossipsub_peers_in_topic = set( peer_id for peer_id in self.pubsub.peer_topics[topic] if peer_id in self.peers_gossipsub - ] + ) return self.select_from_minus( - num_to_select, gossipsub_peers_in_topic, list(minus) + num_to_select, gossipsub_peers_in_topic, minus ) # RPC handlers @@ -642,8 +638,8 @@ class GossipSub(IPubsubRouter): ) -> None: topic: str = prune_msg.topicID - # Remove peer from mesh for topic, if peer is in topic - if topic in self.mesh and sender_peer_id in self.mesh[topic]: + # Remove peer from mesh for topic + if topic in self.mesh: self.mesh[topic].discard(sender_peer_id) # RPC emitters From e51d376d5eb2568c4f7228e8a1e3c935bc64480b Mon Sep 17 00:00:00 2001 From: NIC619 Date: Thu, 19 Dec 2019 14:44:49 +0800 Subject: [PATCH 12/15] Combine `peers_gossipsub` and `peers_floodsub` --- libp2p/pubsub/gossipsub.py | 34 +++++++++------------------------- tests/pubsub/test_gossipsub.py | 11 +++++++---- 2 files changed, 16 insertions(+), 29 deletions(-) diff --git a/libp2p/pubsub/gossipsub.py b/libp2p/pubsub/gossipsub.py index 7b0759c..8097f70 100644 --- a/libp2p/pubsub/gossipsub.py +++ b/libp2p/pubsub/gossipsub.py @@ -35,13 +35,11 @@ class GossipSub(IPubsubRouter): mesh: Dict[str, Set[ID]] fanout: Dict[str, Set[ID]] - peers_to_protocol: Dict[ID, str] + # The protocol peer supports + peer_protocol: Dict[ID, TProtocol] time_since_last_publish: Dict[str, int] - peers_gossipsub: Set[ID] - peers_floodsub: Set[ID] - mcache: MessageCache heartbeat_initial_delay: float @@ -75,14 +73,11 @@ class GossipSub(IPubsubRouter): self.fanout = {} # Create peer --> protocol mapping - self.peers_to_protocol = {} + self.peer_protocol = {} # Create topic --> time since last publish map self.time_since_last_publish = {} - self.peers_gossipsub = set() - self.peers_floodsub = set() - # Create message cache self.mcache = MessageCache(gossip_window, gossip_history) @@ -121,17 +116,13 @@ class GossipSub(IPubsubRouter): """ logger.debug("adding peer %s with protocol %s", peer_id, protocol_id) - if protocol_id == PROTOCOL_ID: - self.peers_gossipsub.add(peer_id) - elif protocol_id == floodsub.PROTOCOL_ID: - self.peers_floodsub.add(peer_id) - else: + if protocol_id not in (PROTOCOL_ID, floodsub.PROTOCOL_ID): # We should never enter here. Becuase the `protocol_id` is registered by your pubsub # instance in multistream-select, but it is not the protocol that gossipsub supports. # In this case, probably we registered gossipsub to a wrong `protocol_id` # in multistream-select, or wrong versions. raise Exception(f"Unreachable: Protocol={protocol_id} is not supported.") - self.peers_to_protocol[peer_id] = protocol_id + self.peer_protocol[peer_id] = protocol_id def remove_peer(self, peer_id: ID) -> None: """ @@ -141,15 +132,12 @@ class GossipSub(IPubsubRouter): """ logger.debug("removing peer %s", peer_id) - self.peers_gossipsub.discard(peer_id) - self.peers_floodsub.discard(peer_id) - for topic in self.mesh: self.mesh[topic].discard(peer_id) for topic in self.fanout: self.fanout[topic].discard(peer_id) - self.peers_to_protocol.pop(peer_id, None) + self.peer_protocol.pop(peer_id, None) async def handle_rpc(self, rpc: rpc_pb2.RPC, sender_peer_id: ID) -> None: """ @@ -217,12 +205,10 @@ class GossipSub(IPubsubRouter): continue # floodsub peers - # FIXME: `gossipsub.peers_floodsub` can be changed to `gossipsub.peers` in go. - # This will improve the efficiency when searching for a peer's protocol id. floodsub_peers: Set[ID] = set( peer_id for peer_id in self.pubsub.peer_topics[topic] - if peer_id in self.peers_floodsub + if self.peer_protocol[peer_id] == floodsub.PROTOCOL_ID ) send_to.update(floodsub_peers) @@ -540,11 +526,9 @@ class GossipSub(IPubsubRouter): gossipsub_peers_in_topic = set( peer_id for peer_id in self.pubsub.peer_topics[topic] - if peer_id in self.peers_gossipsub - ) - return self.select_from_minus( - num_to_select, gossipsub_peers_in_topic, minus + if self.peer_protocol[peer_id] == PROTOCOL_ID ) + return self.select_from_minus(num_to_select, gossipsub_peers_in_topic, minus) # RPC handlers diff --git a/tests/pubsub/test_gossipsub.py b/tests/pubsub/test_gossipsub.py index fe6bf3b..1bc3426 100644 --- a/tests/pubsub/test_gossipsub.py +++ b/tests/pubsub/test_gossipsub.py @@ -4,6 +4,7 @@ import random import pytest from libp2p.peer.id import ID +from libp2p.pubsub.gossipsub import PROTOCOL_ID from libp2p.tools.constants import GOSSIPSUB_PARAMS, GossipsubParams from libp2p.tools.pubsub.utils import dense_connect, one_to_all_connect from libp2p.tools.utils import connect @@ -108,7 +109,7 @@ async def test_handle_graft(pubsubs_gsub, hosts, event_loop, monkeypatch): monkeypatch.setattr(gossipsubs[index_bob], "emit_prune", emit_prune) # Check that alice is bob's peer but not his mesh peer - assert id_alice in gossipsubs[index_bob].peers_gossipsub + assert gossipsubs[index_bob].peer_protocol[id_alice] == PROTOCOL_ID assert topic not in gossipsubs[index_bob].mesh await gossipsubs[index_alice].emit_graft(topic, id_bob) @@ -120,7 +121,7 @@ async def test_handle_graft(pubsubs_gsub, hosts, event_loop, monkeypatch): # Check that bob is alice's peer but not her mesh peer assert topic in gossipsubs[index_alice].mesh assert id_bob not in gossipsubs[index_alice].mesh[topic] - assert id_bob in gossipsubs[index_alice].peers_gossipsub + assert gossipsubs[index_alice].peer_protocol[id_bob] == PROTOCOL_ID await gossipsubs[index_bob].emit_graft(topic, id_alice) @@ -390,7 +391,8 @@ async def test_mesh_heartbeat( fake_peer_ids = [ ID((i).to_bytes(2, byteorder="big")) for i in range(total_peer_count) ] - monkeypatch.setattr(pubsubs_gsub[0].router, "peers_gossipsub", set(fake_peer_ids)) + peer_protocol = {peer_id: PROTOCOL_ID for peer_id in fake_peer_ids} + monkeypatch.setattr(pubsubs_gsub[0].router, "peer_protocol", peer_protocol) peer_topics = {topic: set(fake_peer_ids)} # Monkeypatch the peer subscriptions @@ -437,7 +439,8 @@ async def test_gossip_heartbeat( fake_peer_ids = [ ID((i).to_bytes(2, byteorder="big")) for i in range(total_peer_count) ] - monkeypatch.setattr(pubsubs_gsub[0].router, "peers_gossipsub", set(fake_peer_ids)) + peer_protocol = {peer_id: PROTOCOL_ID for peer_id in fake_peer_ids} + monkeypatch.setattr(pubsubs_gsub[0].router, "peer_protocol", peer_protocol) topic_mesh_peer_count = 14 # Split into mesh peers and fanout peers From 74092c13712e4d09d57d2431d5eebd53dbce0578 Mon Sep 17 00:00:00 2001 From: NIC619 Date: Thu, 19 Dec 2019 16:26:37 +0800 Subject: [PATCH 13/15] Apply PR feedback: update error msg --- libp2p/pubsub/gossipsub.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/libp2p/pubsub/gossipsub.py b/libp2p/pubsub/gossipsub.py index 8097f70..2c0f780 100644 --- a/libp2p/pubsub/gossipsub.py +++ b/libp2p/pubsub/gossipsub.py @@ -588,7 +588,7 @@ class GossipSub(IPubsubRouter): # 3) Get the stream to this peer if sender_peer_id not in self.pubsub.peers: logger.debug( - "Fail to responed to iwant request from %s: peer disconnected", + "Fail to responed to iwant request from %s: peer record not exist", sender_peer_id, ) return @@ -700,7 +700,7 @@ class GossipSub(IPubsubRouter): # Get stream for peer from pubsub if to_peer not in self.pubsub.peers: logger.debug( - "Fail to emit control message to %s: peer disconnected", to_peer + "Fail to emit control message to %s: peer record not exist", to_peer ) return peer_stream = self.pubsub.peers[to_peer] From cb80cfc50b886a614c3e02bbf477ae63e7589b1a Mon Sep 17 00:00:00 2001 From: NIC Lin Date: Thu, 19 Dec 2019 16:33:56 +0800 Subject: [PATCH 14/15] Update libp2p/pubsub/gossipsub.py Co-Authored-By: Chih Cheng Liang --- libp2p/pubsub/gossipsub.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/libp2p/pubsub/gossipsub.py b/libp2p/pubsub/gossipsub.py index 2c0f780..37a1e5d 100644 --- a/libp2p/pubsub/gossipsub.py +++ b/libp2p/pubsub/gossipsub.py @@ -121,7 +121,7 @@ class GossipSub(IPubsubRouter): # instance in multistream-select, but it is not the protocol that gossipsub supports. # In this case, probably we registered gossipsub to a wrong `protocol_id` # in multistream-select, or wrong versions. - raise Exception(f"Unreachable: Protocol={protocol_id} is not supported.") + raise ValueError(f"Protocol={protocol_id} is not supported.") self.peer_protocol[peer_id] = protocol_id def remove_peer(self, peer_id: ID) -> None: From 3c75c85d7f1ee993b8f2f37d5290e65cb80c40cc Mon Sep 17 00:00:00 2001 From: NIC619 Date: Thu, 19 Dec 2019 23:07:20 +0800 Subject: [PATCH 15/15] Fix extra white space --- libp2p/pubsub/gossipsub.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/libp2p/pubsub/gossipsub.py b/libp2p/pubsub/gossipsub.py index 37a1e5d..05efe84 100644 --- a/libp2p/pubsub/gossipsub.py +++ b/libp2p/pubsub/gossipsub.py @@ -121,7 +121,7 @@ class GossipSub(IPubsubRouter): # instance in multistream-select, but it is not the protocol that gossipsub supports. # In this case, probably we registered gossipsub to a wrong `protocol_id` # in multistream-select, or wrong versions. - raise ValueError(f"Protocol={protocol_id} is not supported.") + raise ValueError(f"Protocol={protocol_id} is not supported.") self.peer_protocol[peer_id] = protocol_id def remove_peer(self, peer_id: ID) -> None: