From 7d6daa8e10d3c75cd88b80c2f9f8fec2f3dfe37e Mon Sep 17 00:00:00 2001
From: NIC619 <twedusuck@gmail.com>
Date: Tue, 17 Dec 2019 17:17:03 +0800
Subject: [PATCH 01/23] 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 <twedusuck@gmail.com>
Date: Tue, 17 Dec 2019 17:30:24 +0800
Subject: [PATCH 02/23] 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 <twedusuck@gmail.com>
Date: Tue, 17 Dec 2019 17:36:15 +0800
Subject: [PATCH 03/23] 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 <twedusuck@gmail.com>
Date: Tue, 17 Dec 2019 17:49:49 +0800
Subject: [PATCH 04/23] 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 <twedusuck@gmail.com>
Date: Tue, 17 Dec 2019 17:55:13 +0800
Subject: [PATCH 05/23] 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 <twedusuck@gmail.com>
Date: Tue, 17 Dec 2019 18:47:58 +0800
Subject: [PATCH 06/23] 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 <twedusuck@gmail.com>
Date: Tue, 17 Dec 2019 18:48:25 +0800
Subject: [PATCH 07/23] 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 3cbe24caab851eb1f1ebdf2691445a34425a53a1 Mon Sep 17 00:00:00 2001
From: Jonathan de Jong <jonathandejong02@gmail.com>
Date: Tue, 17 Dec 2019 12:00:11 +0100
Subject: [PATCH 08/23] fixes #384

also adds MultiError to libp2p/exceptions.py

and an additional fixme I have noticed
---
 libp2p/exceptions.py    |  6 ++++++
 libp2p/network/swarm.py | 44 +++++++++++++++++++++++++++++++++++++----
 2 files changed, 46 insertions(+), 4 deletions(-)

diff --git a/libp2p/exceptions.py b/libp2p/exceptions.py
index bdecfad..5b78114 100644
--- a/libp2p/exceptions.py
+++ b/libp2p/exceptions.py
@@ -8,3 +8,9 @@ class ValidationError(BaseLibp2pError):
 
 class ParseError(BaseLibp2pError):
     pass
+
+
+class MultiError(BaseLibp2pError):
+    """Raised with multiple exceptions."""
+
+    # todo: find some way for this to fancy-print all encapsulated errors
diff --git a/libp2p/network/swarm.py b/libp2p/network/swarm.py
index 0c40510..5099a5d 100644
--- a/libp2p/network/swarm.py
+++ b/libp2p/network/swarm.py
@@ -19,6 +19,7 @@ from libp2p.transport.transport_interface import ITransport
 from libp2p.transport.upgrader import TransportUpgrader
 from libp2p.typing import StreamHandlerFn
 
+from ..exceptions import MultiError
 from .connection.raw_connection import RawConnection
 from .connection.swarm_connection import SwarmConn
 from .exceptions import SwarmException
@@ -90,14 +91,49 @@ class Swarm(INetwork):
         except PeerStoreError:
             raise SwarmException(f"No known addresses to peer {peer_id}")
 
-        if not addrs:
+        if len(addrs) == 0:
             raise SwarmException(f"No known addresses to peer {peer_id}")
 
-        multiaddr = addrs[0]
+        exceptions: List[SwarmException] = []
+
+        # Try all known addresses
+        for multiaddr in addrs:
+            try:
+                return await self.dial_addr(multiaddr, peer_id)
+            except SwarmException as e:
+                exceptions.append(e)
+                logger.debug(
+                    "encountered swarm exception when trying to connect to %s, "
+                    "trying next address...",
+                    multiaddr,
+                    exc_info=e,
+                )
+
+        # Tried all addresses, raising exception.
+        if len(exceptions) > 0:
+            raise SwarmException(
+                "unable to connect to %s, all addresses failed to dial (with exceptions)",
+                peer_id,
+            ) from MultiError(exceptions)
+        else:
+            raise SwarmException(
+                "unable to connect to %s, all addresses failed to dial", peer_id
+            )
+
+    async def dial_addr(self, addr: Multiaddr, peer_id: ID) -> INetConn:
+        """
+        dial_addr try to create a connection to peer_id with addr.
+
+        :param addr: the address we want to connect with
+        :param peer_id: the peer we want to connect to
+        :raises SwarmException: raised when an error occurs
+        :return: muxed connection
+        """
+
         # Dial peer (connection to peer does not yet exist)
         # Transport dials peer (gets back a raw conn)
         try:
-            raw_conn = await self.transport.dial(multiaddr)
+            raw_conn = await self.transport.dial(addr)
         except OpenConnectionError as error:
             logger.debug("fail to dial peer %s over base transport", peer_id)
             raise SwarmException(
@@ -137,7 +173,7 @@ class Swarm(INetwork):
     async def new_stream(self, peer_id: ID) -> INetStream:
         """
         :param peer_id: peer_id of destination
-        :param protocol_id: protocol id
+        :param protocol_id: protocol id fixme: protocol_id not in parameters
         :raises SwarmException: raised when an error occurs
         :return: net stream instance
         """

From 04b9d688f8ed55e3d3bb35d912dec9a9702fd399 Mon Sep 17 00:00:00 2001
From: NIC619 <twedusuck@gmail.com>
Date: Tue, 17 Dec 2019 19:09:15 +0800
Subject: [PATCH 09/23] 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 <twedusuck@gmail.com>
Date: Tue, 17 Dec 2019 21:56:02 +0800
Subject: [PATCH 10/23] 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 6b759012438773067e1642a4a7a7e158d54eeec5 Mon Sep 17 00:00:00 2001
From: Jonathan de Jong <jonathandejong02@gmail.com>
Date: Tue, 17 Dec 2019 20:20:09 +0100
Subject: [PATCH 11/23] apply PR feedback (remote len == 0 block, remove
 redundant fixme comment + docstring line)

change wording of final SwarmException to include possible case of no addresses in returned address set

add `from error` in except clause
---
 libp2p/network/swarm.py | 16 ++++++++--------
 1 file changed, 8 insertions(+), 8 deletions(-)

diff --git a/libp2p/network/swarm.py b/libp2p/network/swarm.py
index 5099a5d..2e36ddd 100644
--- a/libp2p/network/swarm.py
+++ b/libp2p/network/swarm.py
@@ -88,11 +88,8 @@ class Swarm(INetwork):
         try:
             # Get peer info from peer store
             addrs = self.peerstore.addrs(peer_id)
-        except PeerStoreError:
-            raise SwarmException(f"No known addresses to peer {peer_id}")
-
-        if len(addrs) == 0:
-            raise SwarmException(f"No known addresses to peer {peer_id}")
+        except PeerStoreError as error:
+            raise SwarmException(f"No known addresses to peer {peer_id}") from error
 
         exceptions: List[SwarmException] = []
 
@@ -112,12 +109,16 @@ class Swarm(INetwork):
         # Tried all addresses, raising exception.
         if len(exceptions) > 0:
             raise SwarmException(
-                "unable to connect to %s, all addresses failed to dial (with exceptions)",
+                "unable to connect to %s, no addresses established a successful connection "
+                "(with exceptions)",
                 peer_id,
             ) from MultiError(exceptions)
         else:
             raise SwarmException(
-                "unable to connect to %s, all addresses failed to dial", peer_id
+                "unable to connect to %s, no addresses established a successful connection "
+                "(tried %d addresses)",
+                peer_id,
+                len(addrs),
             )
 
     async def dial_addr(self, addr: Multiaddr, peer_id: ID) -> INetConn:
@@ -173,7 +174,6 @@ class Swarm(INetwork):
     async def new_stream(self, peer_id: ID) -> INetStream:
         """
         :param peer_id: peer_id of destination
-        :param protocol_id: protocol id fixme: protocol_id not in parameters
         :raises SwarmException: raised when an error occurs
         :return: net stream instance
         """

From f3732f94809e2ae7314641bcb066be6222370d9d Mon Sep 17 00:00:00 2001
From: NIC619 <twedusuck@gmail.com>
Date: Wed, 18 Dec 2019 12:37:04 +0800
Subject: [PATCH 12/23] 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 4e4d91b2e20a76f90cd23245a69d3c1e7bf73666 Mon Sep 17 00:00:00 2001
From: Jonathan de Jong <jonathandejong02@gmail.com>
Date: Wed, 18 Dec 2019 10:54:52 +0100
Subject: [PATCH 13/23] Apply PR review suggestion (change "muxed" to "network"
 in docstrings)

Co-Authored-By: Kevin Mai-Husan Chia <mhchia@users.noreply.github.com>
---
 libp2p/network/swarm.py | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/libp2p/network/swarm.py b/libp2p/network/swarm.py
index 2e36ddd..4c36ab4 100644
--- a/libp2p/network/swarm.py
+++ b/libp2p/network/swarm.py
@@ -128,7 +128,7 @@ class Swarm(INetwork):
         :param addr: the address we want to connect with
         :param peer_id: the peer we want to connect to
         :raises SwarmException: raised when an error occurs
-        :return: muxed connection
+        :return: network connection
         """
 
         # Dial peer (connection to peer does not yet exist)

From 81fe4049cf038d098f8b3f714e65471a4a998dfb Mon Sep 17 00:00:00 2001
From: Jonathan de Jong <jonathandejong02@gmail.com>
Date: Wed, 18 Dec 2019 18:47:03 +0100
Subject: [PATCH 14/23] Apply PR review feedback > add `if not addr` clause
 back > use f-strings for exceptions instead of %s

---
 libp2p/network/swarm.py | 22 ++++++++--------------
 1 file changed, 8 insertions(+), 14 deletions(-)

diff --git a/libp2p/network/swarm.py b/libp2p/network/swarm.py
index 4c36ab4..eb7e306 100644
--- a/libp2p/network/swarm.py
+++ b/libp2p/network/swarm.py
@@ -91,6 +91,9 @@ class Swarm(INetwork):
         except PeerStoreError as error:
             raise SwarmException(f"No known addresses to peer {peer_id}") from error
 
+        if not addrs:
+            raise SwarmException(f"No known addresses to peer {peer_id}")
+
         exceptions: List[SwarmException] = []
 
         # Try all known addresses
@@ -107,19 +110,10 @@ class Swarm(INetwork):
                 )
 
         # Tried all addresses, raising exception.
-        if len(exceptions) > 0:
-            raise SwarmException(
-                "unable to connect to %s, no addresses established a successful connection "
-                "(with exceptions)",
-                peer_id,
-            ) from MultiError(exceptions)
-        else:
-            raise SwarmException(
-                "unable to connect to %s, no addresses established a successful connection "
-                "(tried %d addresses)",
-                peer_id,
-                len(addrs),
-            )
+        raise SwarmException(
+            f"unable to connect to {peer_id}, no addresses established a successful connection "
+            "(with exceptions)",
+        ) from MultiError(exceptions)
 
     async def dial_addr(self, addr: Multiaddr, peer_id: ID) -> INetConn:
         """
@@ -138,7 +132,7 @@ class Swarm(INetwork):
         except OpenConnectionError as error:
             logger.debug("fail to dial peer %s over base transport", peer_id)
             raise SwarmException(
-                "fail to open connection to peer %s", peer_id
+                f"fail to open connection to peer {peer_id}"
             ) from error
 
         logger.debug("dialed peer %s over base transport", peer_id)

From f54bc9d1afcad7b0ac883f597277ab1db1e3ea78 Mon Sep 17 00:00:00 2001
From: Jonathan de Jong <jonathandejong02@gmail.com>
Date: Wed, 18 Dec 2019 19:05:22 +0100
Subject: [PATCH 15/23] Make linter happy

---
 libp2p/network/swarm.py | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/libp2p/network/swarm.py b/libp2p/network/swarm.py
index eb7e306..c861877 100644
--- a/libp2p/network/swarm.py
+++ b/libp2p/network/swarm.py
@@ -112,7 +112,7 @@ class Swarm(INetwork):
         # Tried all addresses, raising exception.
         raise SwarmException(
             f"unable to connect to {peer_id}, no addresses established a successful connection "
-            "(with exceptions)",
+            "(with exceptions)"
         ) from MultiError(exceptions)
 
     async def dial_addr(self, addr: Multiaddr, peer_id: ID) -> INetConn:

From 1124fc8211565b060a9169647d93af199ee49c9c Mon Sep 17 00:00:00 2001
From: Jonathan de Jong <jonathandejong02@gmail.com>
Date: Thu, 19 Dec 2019 01:25:57 +0100
Subject: [PATCH 16/23] add eth_utils add fastecdsa-any requirements

---
 setup.py | 17 ++++++++++++++++-
 1 file changed, 16 insertions(+), 1 deletion(-)

diff --git a/setup.py b/setup.py
index 28d29fa..0a9f8a0 100644
--- a/setup.py
+++ b/setup.py
@@ -33,6 +33,7 @@ extras_require = {
         "ipython",
         "setuptools>=36.2.0",
         "tox>=3.13.2,<4.0.0",
+        "eth_utils",
     ],
 }
 
@@ -43,6 +44,20 @@ extras_require["dev"] = (
     + extras_require["doc"]
 )
 
+fastecdsa = [
+    # No official fastecdsa==1.7.4,1.7.5 wheels for Windows, using a pypi package that includes
+    # the original library, but also windows-built wheels (32+64-bit) on those versions.
+    # Fixme: Remove section when fastecdsa has released a windows-compatible wheel
+    #  (specifically: both win32 and win_amd64 targets)
+    #  See the following issues for more information;
+    #  https://github.com/libp2p/py-libp2p/issues/363
+    #  https://github.com/AntonKueltz/fastecdsa/issues/11
+    "fastecdsa-any==1.7.5;sys_platform=='win32'",
+    # Wheels are provided for these platforms, or compiling one is minimally frustrating in a
+    # default python installation.
+    "fastecdsa==1.7.5;sys_platform!='win32'",
+]
+
 
 with open("./README.md") as readme:
     long_description = readme.read()
@@ -67,7 +82,7 @@ install_requires = [
 # RTD system so we have to exclude these dependencies when we are in an RTD environment.
 readthedocs_is_building = os.environ.get("READTHEDOCS", False)
 if not readthedocs_is_building:
-    install_requires.append("fastecdsa==1.7.4")
+    install_requires.extend(fastecdsa)
 
 
 setup(

From 6cd3eb8faec2e60ca053356a37e58cf51399026d Mon Sep 17 00:00:00 2001
From: NIC619 <twedusuck@gmail.com>
Date: Thu, 19 Dec 2019 14:15:51 +0800
Subject: [PATCH 17/23] 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 <twedusuck@gmail.com>
Date: Thu, 19 Dec 2019 14:44:49 +0800
Subject: [PATCH 18/23] 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 <twedusuck@gmail.com>
Date: Thu, 19 Dec 2019 16:26:37 +0800
Subject: [PATCH 19/23] 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 <twedusuck@gmail.com>
Date: Thu, 19 Dec 2019 16:33:56 +0800
Subject: [PATCH 20/23] Update libp2p/pubsub/gossipsub.py

Co-Authored-By: Chih Cheng Liang <chihchengliang@gmail.com>
---
 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 <twedusuck@gmail.com>
Date: Thu, 19 Dec 2019 23:07:20 +0800
Subject: [PATCH 21/23] 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:

From 0b97f4d57cc988976a7909d75df8e4c7c9228ee0 Mon Sep 17 00:00:00 2001
From: Jonathan de Jong <jonathandejong02@gmail.com>
Date: Sat, 21 Dec 2019 08:59:07 +0100
Subject: [PATCH 22/23] remove eth_utils per PR request

---
 setup.py | 1 -
 1 file changed, 1 deletion(-)

diff --git a/setup.py b/setup.py
index 0a9f8a0..356487a 100644
--- a/setup.py
+++ b/setup.py
@@ -33,7 +33,6 @@ extras_require = {
         "ipython",
         "setuptools>=36.2.0",
         "tox>=3.13.2,<4.0.0",
-        "eth_utils",
     ],
 }
 

From 17074dded0a48b2fb942e2da55ec02c3f4a56292 Mon Sep 17 00:00:00 2001
From: Jonathan de Jong <jonathandejong02@gmail.com>
Date: Sat, 21 Dec 2019 10:35:34 +0100
Subject: [PATCH 23/23] add tests to new multiple multiaddr change

---
 tests/network/test_swarm.py | 57 +++++++++++++++++++++++++++++++++++++
 1 file changed, 57 insertions(+)

diff --git a/tests/network/test_swarm.py b/tests/network/test_swarm.py
index 6fe2543..3cc9375 100644
--- a/tests/network/test_swarm.py
+++ b/tests/network/test_swarm.py
@@ -1,5 +1,6 @@
 import asyncio
 
+from multiaddr import Multiaddr
 import pytest
 
 from libp2p.network.exceptions import SwarmException
@@ -91,3 +92,59 @@ async def test_swarm_remove_conn(swarm_pair):
     # Test: Remove twice. There should not be errors.
     swarm_0.remove_conn(conn_0)
     assert swarm_1.get_peer_id() not in swarm_0.connections
+
+
+@pytest.mark.asyncio
+async def test_swarm_multiaddr(is_host_secure):
+    swarms = await SwarmFactory.create_batch_and_listen(is_host_secure, 3)
+
+    def clear():
+        swarms[0].peerstore.clear_addrs(swarms[1].get_peer_id())
+
+    clear()
+    # No addresses
+    with pytest.raises(SwarmException):
+        await swarms[0].dial_peer(swarms[1].get_peer_id())
+
+    clear()
+    # Wrong addresses
+    swarms[0].peerstore.add_addrs(
+        swarms[1].get_peer_id(), [Multiaddr("/ip4/0.0.0.0/tcp/9999")], 10000
+    )
+
+    with pytest.raises(SwarmException):
+        await swarms[0].dial_peer(swarms[1].get_peer_id())
+
+    clear()
+    # Multiple wrong addresses
+    swarms[0].peerstore.add_addrs(
+        swarms[1].get_peer_id(),
+        [Multiaddr("/ip4/0.0.0.0/tcp/9999"), Multiaddr("/ip4/0.0.0.0/tcp/9998")],
+        10000,
+    )
+
+    with pytest.raises(SwarmException):
+        await swarms[0].dial_peer(swarms[1].get_peer_id())
+
+    # Test one address
+    addrs = tuple(
+        addr
+        for transport in swarms[1].listeners.values()
+        for addr in transport.get_addrs()
+    )
+
+    swarms[0].peerstore.add_addrs(swarms[1].get_peer_id(), addrs[:1], 10000)
+    await swarms[0].dial_peer(swarms[1].get_peer_id())
+
+    # Test multiple addresses
+    addrs = tuple(
+        addr
+        for transport in swarms[1].listeners.values()
+        for addr in transport.get_addrs()
+    )
+
+    swarms[0].peerstore.add_addrs(swarms[1].get_peer_id(), addrs + addrs, 10000)
+    await swarms[0].dial_peer(swarms[1].get_peer_id())
+
+    for swarm in swarms:
+        await swarm.close()