Merge branch 'main' into feature/bootstrap

Author: seetadev

libp2p/abc.py

@@ -18,6 +18,13 @@
     PublicKey,
 )
 
+"""
+Latency EWMA Smoothing governs the deacy of the EWMA (the speed at which
+is changes). This must be a normalized (0-1) value.
+1 is 100% change, 0 is no change.
+"""
+LATENCY_EWMA_SMOOTHING = 0.1
+
 
 class PeerData(IPeerData):
     pubkey: PublicKey | None
@@ -27,6 +34,7 @@ class PeerData(IPeerData):
     addrs: list[Multiaddr]
     last_identified: int
     ttl: int  # Keep ttl=0 by default for always valid
+    latmap: float
 
     def __init__(self) -> None:
         self.pubkey = None
@@ -36,6 +44,9 @@ def __init__(self) -> None:
         self.addrs = []
         self.last_identified = int(time.time())
         self.ttl = 0
+        self.latmap = 0
+
+    # --------PROTO-BOOK--------
 
     def get_protocols(self) -> list[str]:
         """
@@ -55,6 +66,37 @@ def set_protocols(self, protocols: Sequence[str]) -> None:
         """
         self.protocols = list(protocols)
 
+    def remove_protocols(self, protocols: Sequence[str]) -> None:
+        """
+        :param protocols: protocols to remove
+        """
+        for protocol in protocols:
+            if protocol in self.protocols:
+                self.protocols.remove(protocol)
+
+    def supports_protocols(self, protocols: Sequence[str]) -> list[str]:
+        """
+        :param protocols: protocols to check from
+        :return: all supported protocols in the given list
+        """
+        return [proto for proto in protocols if proto in self.protocols]
+
+    def first_supported_protocol(self, protocols: Sequence[str]) -> str:
+        """
+        :param protocols: protocols to check from
+        :return: first supported protocol in the given list
+        """
+        for protocol in protocols:
+            if protocol in self.protocols:
+                return protocol
+
+        return "None supported"
+
+    def clear_protocol_data(self) -> None:
+        """Clear all protocols"""
+        self.protocols = []
+
+    # -------ADDR-BOOK---------
     def add_addrs(self, addrs: Sequence[Multiaddr]) -> None:
         """
         :param addrs: multiaddresses to add
@@ -73,6 +115,7 @@ def clear_addrs(self) -> None:
         """Clear all addresses."""
         self.addrs = []
 
+    # -------METADATA-----------
     def put_metadata(self, key: str, val: Any) -> None:
         """
         :param key: key in KV pair
@@ -90,6 +133,11 @@ def get_metadata(self, key: str) -> Any:
             return self.metadata[key]
         raise PeerDataError("key not found")
 
+    def clear_metadata(self) -> None:
+        """Clears metadata."""
+        self.metadata = {}
+
+    # -------KEY-BOOK---------------
     def add_pubkey(self, pubkey: PublicKey) -> None:
         """
         :param pubkey:
@@ -120,9 +168,41 @@ def get_privkey(self) -> PrivateKey:
             raise PeerDataError("private key not found")
         return self.privkey
 
+    def clear_keydata(self) -> None:
+        """Clears keydata"""
+        self.pubkey = None
+        self.privkey = None
+
+    # ----------METRICS--------------
+    def record_latency(self, new_latency: float) -> None:
+        """
+        Records a new latency measurement for the given peer
+        using Exponentially Weighted Moving Average (EWMA)
+        :param new_latency: the new latency value
+        """
+        s = LATENCY_EWMA_SMOOTHING
+        if s > 1 or s < 0:
+            s = 0.1
+
+        if self.latmap == 0:
+            self.latmap = new_latency
+        else:
+            prev = self.latmap
+            updated = ((1.0 - s) * prev) + (s * new_latency)
+            self.latmap = updated
+
+    def latency_EWMA(self) -> float:
+        """Returns the latency EWMA value"""
+        return self.latmap
+
+    def clear_metrics(self) -> None:
+        """Clear the latency metrics"""
+        self.latmap = 0
+
     def update_last_identified(self) -> None:
         self.last_identified = int(time.time())
 
+    # ----------TTL------------------
     def get_last_identified(self) -> int:
         """
         :return: last identified timestamp

libp2p/peer/peerdata.py

@@ -2,6 +2,7 @@
     defaultdict,
 )
 from collections.abc import (
+    AsyncIterable,
     Sequence,
 )
 from typing import (
@@ -11,6 +12,8 @@
 from multiaddr import (
     Multiaddr,
 )
+import trio
+from trio import MemoryReceiveChannel, MemorySendChannel
 
 from libp2p.abc import (
     IPeerStore,
@@ -40,6 +43,7 @@ class PeerStore(IPeerStore):
 
     def __init__(self) -> None:
         self.peer_data_map = defaultdict(PeerData)
+        self.addr_update_channels: dict[ID, MemorySendChannel[Multiaddr]] = {}
 
     def peer_info(self, peer_id: ID) -> PeerInfo:
         """
@@ -53,6 +57,29 @@ def peer_info(self, peer_id: ID) -> PeerInfo:
             return PeerInfo(peer_id, peer_data.get_addrs())
         raise PeerStoreError("peer ID not found")
 
+    def peer_ids(self) -> list[ID]:
+        """
+        :return: all of the peer IDs stored in peer store
+        """
+        return list(self.peer_data_map.keys())
+
+    def clear_peerdata(self, peer_id: ID) -> None:
+        """Clears the peer data of the peer"""
+
+    def valid_peer_ids(self) -> list[ID]:
+        """
+        :return: all of the valid peer IDs stored in peer store
+        """
+        valid_peer_ids: list[ID] = []
+        for peer_id, peer_data in self.peer_data_map.items():
+            if not peer_data.is_expired():
+                valid_peer_ids.append(peer_id)
+            else:
+                peer_data.clear_addrs()
+        return valid_peer_ids
+
+    # --------PROTO-BOOK--------
+
     def get_protocols(self, peer_id: ID) -> list[str]:
         """
         :param peer_id: peer ID to get protocols for
@@ -79,23 +106,31 @@ def set_protocols(self, peer_id: ID, protocols: Sequence[str]) -> None:
         peer_data = self.peer_data_map[peer_id]
         peer_data.set_protocols(list(protocols))
 
-    def peer_ids(self) -> list[ID]:
+    def remove_protocols(self, peer_id: ID, protocols: Sequence[str]) -> None:
         """
-        :return: all of the peer IDs stored in peer store
+        :param peer_id: peer ID to get info for
+        :param protocols: unsupported protocols to remove
         """
-        return list(self.peer_data_map.keys())
+        peer_data = self.peer_data_map[peer_id]
+        peer_data.remove_protocols(protocols)
 
-    def valid_peer_ids(self) -> list[ID]:
+    def supports_protocols(self, peer_id: ID, protocols: Sequence[str]) -> list[str]:
         """
-        :return: all of the valid peer IDs stored in peer store
+        :return: all of the peer IDs stored in peer store
         """
-        valid_peer_ids: list[ID] = []
-        for peer_id, peer_data in self.peer_data_map.items():
-            if not peer_data.is_expired():
-                valid_peer_ids.append(peer_id)
-            else:
-                peer_data.clear_addrs()
-        return valid_peer_ids
+        peer_data = self.peer_data_map[peer_id]
+        return peer_data.supports_protocols(protocols)
+
+    def first_supported_protocol(self, peer_id: ID, protocols: Sequence[str]) -> str:
+        peer_data = self.peer_data_map[peer_id]
+        return peer_data.first_supported_protocol(protocols)
+
+    def clear_protocol_data(self, peer_id: ID) -> None:
+        """Clears prtocoldata"""
+        peer_data = self.peer_data_map[peer_id]
+        peer_data.clear_protocol_data()
+
+    # ------METADATA---------
 
     def get(self, peer_id: ID, key: str) -> Any:
         """
@@ -121,6 +156,13 @@ def put(self, peer_id: ID, key: str, val: Any) -> None:
         peer_data = self.peer_data_map[peer_id]
         peer_data.put_metadata(key, val)
 
+    def clear_metadata(self, peer_id: ID) -> None:
+        """Clears metadata"""
+        peer_data = self.peer_data_map[peer_id]
+        peer_data.clear_metadata()
+
+    # -------ADDR-BOOK--------
+
     def add_addr(self, peer_id: ID, addr: Multiaddr, ttl: int = 0) -> None:
         """
         :param peer_id: peer ID to add address for
@@ -140,6 +182,13 @@ def add_addrs(self, peer_id: ID, addrs: Sequence[Multiaddr], ttl: int = 0) -> No
         peer_data.set_ttl(ttl)
         peer_data.update_last_identified()
 
+        if peer_id in self.addr_update_channels:
+            for addr in addrs:
+                try:
+                    self.addr_update_channels[peer_id].send_nowait(addr)
+                except trio.WouldBlock:
+                    pass  # Or consider logging / dropping / replacing stream
+
     def addrs(self, peer_id: ID) -> list[Multiaddr]:
         """
         :param peer_id: peer ID to get addrs for
@@ -165,7 +214,7 @@ def clear_addrs(self, peer_id: ID) -> None:
 
     def peers_with_addrs(self) -> list[ID]:
         """
-        :return: all of the peer IDs which has addrs stored in peer store
+        :return: all of the peer IDs which has addrsfloat stored in peer store
         """
         # Add all peers with addrs at least 1 to output
         output: list[ID] = []
@@ -179,6 +228,27 @@ def peers_with_addrs(self) -> list[ID]:
                     peer_data.clear_addrs()
         return output
 
+    async def addr_stream(self, peer_id: ID) -> AsyncIterable[Multiaddr]:
+        """
+        Returns an async stream of newly added addresses for the given peer.
+
+        This function allows consumers to subscribe to address updates for a peer
+        and receive each new address as it is added via `add_addr` or `add_addrs`.
+
+        :param peer_id: The ID of the peer to monitor address updates for.
+        :return: An async iterator yielding Multiaddr instances as they are added.
+        """
+        send: MemorySendChannel[Multiaddr]
+        receive: MemoryReceiveChannel[Multiaddr]
+
+        send, receive = trio.open_memory_channel(0)
+        self.addr_update_channels[peer_id] = send
+
+        async for addr in receive:
+            yield addr
+
+    # -------KEY-BOOK---------
+
     def add_pubkey(self, peer_id: ID, pubkey: PublicKey) -> None:
         """
         :param peer_id: peer ID to add public key for
@@ -239,6 +309,45 @@ def add_key_pair(self, peer_id: ID, key_pair: KeyPair) -> None:
         self.add_pubkey(peer_id, key_pair.public_key)
         self.add_privkey(peer_id, key_pair.private_key)
 
+    def peer_with_keys(self) -> list[ID]:
+        """Returns the peer_ids for which keys are stored"""
+        return [
+            peer_id
+            for peer_id, pdata in self.peer_data_map.items()
+            if pdata.pubkey is not None
+        ]
+
+    def clear_keydata(self, peer_id: ID) -> None:
+        """Clears the keys of the peer"""
+        peer_data = self.peer_data_map[peer_id]
+        peer_data.clear_keydata()
+
+    # --------METRICS--------
+
+    def record_latency(self, peer_id: ID, RTT: float) -> None:
+        """
+        Records a new latency measurement for the given peer
+        using Exponentially Weighted Moving Average (EWMA)
+
+        :param peer_id: peer ID to get private key for
+        :param RTT: the new latency value (round trip time)
+        """
+        peer_data = self.peer_data_map[peer_id]
+        peer_data.record_latency(RTT)
+
+    def latency_EWMA(self, peer_id: ID) -> float:
+        """
+        :param peer_id: peer ID to get private key for
+        :return: The latency EWMA value for that peer
+        """
+        peer_data = self.peer_data_map[peer_id]
+        return peer_data.latency_EWMA()
+
+    def clear_metrics(self, peer_id: ID) -> None:
+        """Clear the latency metrics"""
+        peer_data = self.peer_data_map[peer_id]
+        peer_data.clear_metrics()
+
 
 class PeerStoreError(KeyError):
     """Raised when peer ID is not found in peer store."""