always split option in kademlia

deccom/protocols/defaultprotocol.py

@@ -86,14 +86,14 @@ async def send_ping(self, addr, success, error, dt = 10):
         self.pings[msg_id] = (success, timeout)
         trmp = bytearray([DefaultProtocol.PING])
         trmp = trmp + bts
-        self.send_datagram(trmp, addr=addr)
+        await self.send_datagram(trmp, addr=addr)
 
         return
 
     async def handle_ping(self, addr, data):
         trmp = bytearray([DefaultProtocol.PONG])
         trmp = trmp + data
-        self.send_datagram(trmp, addr=addr)
+        await self.send_datagram(trmp, addr=addr)
         # print("sent pong",addr)
         return
 

deccom/protocols/peerdiscovery/_kademlia_routing.py

@@ -72,14 +72,14 @@ def remove_peer(self, dist):
     def get_peer(self, dist):
 
         return self.peers.get(dist)
-    def add_peer(self, dist, node):
+    def add_peer(self, dist, node, always_split = False):
         self.touch()
         if self.peers.get(dist) != None:
             del self.peers[dist]
             self.peers[dist] = node
             return 0
         if len(self.peers) >= self.k:
-            if self.originator:
+            if self.originator or (always_split and self.max_dist - self.min_dist > self.k):
                 self.toadd.append((dist,node))
                 return 1
 
@@ -103,13 +103,14 @@ def __len__(self):
 
 
 class BucketManager(object):
-    def __init__(self, id, k, success_call, max_l = 256) -> None:
+    def __init__(self, id, k, success_call, max_l = 256, always_split = False) -> None:
         if isinstance(id, bytearray):
             id = bytes(id)
         if isinstance(id, bytes):
             id = int.from_bytes(id, byteorder="big")
         self.id = id
         self.k = k
+        self.always_split = always_split
         self.success_call = success_call
         self.buckets = [KBucket(0, 2**max_l, k, originator=True, success_call=self.success_call)]
     def bytexor(self, b1,b2):
@@ -164,7 +165,7 @@ def add_peer(self,id,node, lv=0):
             return
         indx = self._get_index(dist)
 
-        ret = self.buckets[indx].add_peer(dist,node)
+        ret = self.buckets[indx].add_peer(dist,node,always_split=self.always_split)
 
         if ret == 0:
 

deccom/protocols/peerdiscovery/biggossip.py

@@ -139,7 +139,7 @@ def process_datagram(self, addr: tuple[str, int], data: bytes):
 
                 self.connection_approval(addr,seeker,self.add_peer,self.ban_peer)
             elif self.peers.get(id) == None:
-                l = list(self.get_peers())[:10]
+                l = list(self.get_peers())
                 for p in l:
                     if self.get_peers().get(p) is None:
                         continue
@@ -176,7 +176,8 @@ def add_peer(self, addr: tuple[str,int], p: Peer):
         # print(p)
 
         self.peers[p.id_node] = p
-
+        return super().add_peer(addr, p)
+
 
 
 

deccom/protocols/peerdiscovery/kademliadiscovery.py

@@ -21,18 +21,19 @@ class KademliaDiscovery(AbstractPeerDiscovery):
                     "_lower_ping": "send_ping"
                     })
     required_lower = AbstractPeerDiscovery.required_lower + ["send_ping"]
-    def __init__(self, bootstrap_peers: list[Peer] = [], interval: int = 60, k: int = 20, submodule=None, callback: Callable[[tuple[str, int], bytes], None] = None, disconnected_callback=lambda *args:..., connected_callback: Callable[[Peer], None] =lambda *args:...):
+    def __init__(self, bootstrap_peers: list[Peer] = [], interval: int = 60, k: int = 20, always_split = False, submodule=None, callback: Callable[[tuple[str, int], bytes], None] = None, disconnected_callback=lambda *args:..., connected_callback: Callable[[Peer], None] =lambda *args:...):
         super().__init__(bootstrap_peers, interval, submodule, callback, disconnected_callback, connected_callback)
         self.k = k
         self.peer_crawls = dict()
         self.sent_finds = dict()
         self.warmup = 0
         self.max_warmup = 30 * (k//10)
+        self.always_split = always_split
         self.searches: dict[bytes,bytes] = dict()
         self.finders: dict[bytes, Finder] = dict()
     async def start(self, p: Peer):
         await super().start(p)
-        self.bucket_manager = BucketManager(self.peer.id_node,self.k,self._add)
+        self.bucket_manager = BucketManager(self.peer.id_node,self.k,self._add,always_split = self.always_split)
         for p in self.bootstrap_peers:
             await self.introduce_to_peer(p)
             msg = bytearray([KademliaDiscovery.ASK_FOR_ID])