moved dandelion_enabled from state to Dandelion class as enabled attr

Author: anand-skss

src/bitmessagemain.py

@@ -12,6 +12,7 @@
 import os
 import sys
 
+
 try:
     import pathmagic
 except ImportError:
@@ -156,13 +157,6 @@ def start(self):
 
         set_thread_name("PyBitmessage")
 
-        state.dandelion_enabled = config.safeGetInt('network', 'dandelion')
-        # dandelion requires outbound connections, without them,
-        # stem objects will get stuck forever
-        if state.dandelion_enabled and not config.safeGetBoolean(
-                'bitmessagesettings', 'sendoutgoingconnections'):
-            state.dandelion_enabled = 0
-
         if state.testmode or config.safeGetBoolean(
                 'bitmessagesettings', 'extralowdifficulty'):
             defaults.networkDefaultProofOfWorkNonceTrialsPerByte = int(

src/network/__init__.py

@@ -1,9 +1,10 @@
 """
 Network subsystem package
 """
-
+from .dandelion import Dandelion
 from .threads import StoppableThread
 
+dandelion_ins = Dandelion()
 
 __all__ = ["StoppableThread"]
 
@@ -21,6 +22,11 @@ def start(config, state):
     from .receivequeuethread import ReceiveQueueThread
     from .uploadthread import UploadThread
 
+    # check and set dandelion enabled value at network startup
+    dandelion_ins.init_dandelion_enabled(config)
+    # pass pool instance into dandelion class instance
+    dandelion_ins.init_pool(connectionpool.pool)
+
     readKnownNodes()
     connectionpool.pool.connectToStream(1)
     for thread in (

src/network/bmobject.py

@@ -6,8 +6,8 @@
 
 import protocol
 import state
-import dandelion
 import connectionpool
+from network import dandelion_ins
 from highlevelcrypto import calculateInventoryHash
 
 logger = logging.getLogger('default')
@@ -113,7 +113,7 @@ def checkAlreadyHave(self):
         or advertise it unnecessarily)
         """
         # if it's a stem duplicate, pretend we don't have it
-        if dandelion.instance.hasHash(self.inventoryHash):
+        if dandelion_ins.hasHash(self.inventoryHash):
             return
         if self.inventoryHash in state.Inventory:
             raise BMObjectAlreadyHaveError()

src/network/bmproto.py

@@ -15,7 +15,6 @@
 import knownnodes
 import protocol
 import state
-import dandelion
 import connectionpool
 from bmconfigparser import config
 from queues import invQueue, objectProcessorQueue, portCheckerQueue
@@ -27,7 +26,7 @@
     BMObjectUnwantedStreamError
 )
 from network.proxy import ProxyError
-
+from network import dandelion_ins
 from node import Node, Peer
 from objectracker import ObjectTracker, missingObjects
 
@@ -351,14 +350,14 @@ def _command_inv(self, extend_dandelion_stem=False):
             raise BMProtoExcessiveDataError()
 
         # ignore dinv if dandelion turned off
-        if extend_dandelion_stem and not state.dandelion_enabled:
+        if extend_dandelion_stem and not dandelion_ins.enabled:
             return True
 
         for i in map(str, items):
-            if i in state.Inventory and not dandelion.instance.hasHash(i):
+            if i in state.Inventory and not dandelion_ins.hasHash(i):
                 continue
-            if extend_dandelion_stem and not dandelion.instance.hasHash(i):
-                dandelion.instance.addHash(i, self)
+            if extend_dandelion_stem and not dandelion_ins.hasHash(i):
+                dandelion_ins.addHash(i, self)
             self.handleReceivedInventory(i)
 
         return True
@@ -420,9 +419,9 @@ def bm_command_object(self):
             except KeyError:
                 pass
 
-        if self.object.inventoryHash in state.Inventory and dandelion.instance.hasHash(
+        if self.object.inventoryHash in state.Inventory and dandelion_ins.hasHash(
                 self.object.inventoryHash):
-            dandelion.instance.removeHash(
+            dandelion_ins.removeHash(
                 self.object.inventoryHash, "cycle detection")
 
         state.Inventory[self.object.inventoryHash] = (
@@ -541,7 +540,7 @@ def bm_command_version(self):
         if not self.isOutbound:
             self.append_write_buf(protocol.assembleVersionMessage(
                 self.destination.host, self.destination.port,
-                connectionpool.pool.streams, True,
+                connectionpool.pool.streams, dandelion_ins.enabled, True,
                 nodeid=self.nodeid))
             logger.debug(
                 '%(host)s:%(port)i sending version',

src/network/dandelion.py

@@ -7,9 +7,6 @@
 from threading import RLock
 from time import time
 
-import connectionpool
-import state
-from queues import invQueue
 
 # randomise routes after 600 seconds
 REASSIGN_INTERVAL = 600
@@ -37,6 +34,8 @@ def __init__(self):
         # when to rerandomise routes
         self.refresh = time() + REASSIGN_INTERVAL
         self.lock = RLock()
+        self.enabled = None
+        self.pool = None
 
     @staticmethod
     def poissonTimeout(start=None, average=0):
@@ -47,10 +46,23 @@ def poissonTimeout(start=None, average=0):
             average = FLUFF_TRIGGER_MEAN_DELAY
         return start + expovariate(1.0 / average) + FLUFF_TRIGGER_FIXED_DELAY
 
+    def init_pool(self, pool):
+        """pass pool instance"""
+        self.pool = pool
+
+    def init_dandelion_enabled(self, config):
+        """Check if Dandelion is enabled and set value in enabled attribute"""
+        dandelion_enabled = config.safeGetInt('network', 'dandelion')
+        # dandelion requires outbound connections, without them,
+        # stem objects will get stuck forever
+        if not config.safeGetBoolean(
+                'bitmessagesettings', 'sendoutgoingconnections'):
+            dandelion_enabled = 0
+        self.enabled = dandelion_enabled
+
     def addHash(self, hashId, source=None, stream=1):
-        """Add inventory vector to dandelion stem"""
-        if not state.dandelion_enabled:
-            return
+        """Add inventory vector to dandelion stem return status of dandelion enabled"""
+        assert self.enabled is not None
         with self.lock:
             self.hashMap[hashId] = Stem(
                 self.getNodeStem(source),
@@ -89,7 +101,7 @@ def objectChildStem(self, hashId):
         """Child (i.e. next) node for an inventory vector during stem mode"""
         return self.hashMap[hashId].child
 
-    def maybeAddStem(self, connection):
+    def maybeAddStem(self, connection, invQueue):
         """
         If we had too few outbound connections, add the current one to the
         current stem list. Dandelion as designed by the authors should
@@ -163,7 +175,7 @@ def getNodeStem(self, node=None):
                 self.nodeMap[node] = self.pickStem(node)
                 return self.nodeMap[node]
 
-    def expire(self):
+    def expire(self, invQueue):
         """Switch expired objects from stem to fluff mode"""
         with self.lock:
             deadline = time()
@@ -179,19 +191,18 @@ def expire(self):
 
     def reRandomiseStems(self):
         """Re-shuffle stem mapping (parent <-> child pairs)"""
+        assert self.pool is not None
+        if self.refresh > time():
+            return
+
         with self.lock:
             try:
                 # random two connections
                 self.stem = sample(
-                    connectionpool.BMConnectionPool(
-                    ).outboundConnections.values(), MAX_STEMS)
+                    self.pool.outboundConnections.values(), MAX_STEMS)
             # not enough stems available
             except ValueError:
-                self.stem = connectionpool.BMConnectionPool(
-                ).outboundConnections.values()
+                self.stem = self.pool.outboundConnections.values()
             self.nodeMap = {}
             # hashMap stays to cater for pending stems
         self.refresh = time() + REASSIGN_INTERVAL
-
-
-instance = Dandelion()

src/network/downloadthread.py

@@ -6,8 +6,8 @@
 import addresses
 import helper_random
 import protocol
-import dandelion
 import connectionpool
+from network import dandelion_ins
 from objectracker import missingObjects
 from threads import StoppableThread
 
@@ -60,7 +60,7 @@ def run(self):
                 payload = bytearray()
                 chunkCount = 0
                 for chunk in request:
-                    if chunk in state.Inventory and not dandelion.instance.hasHash(chunk):
+                    if chunk in state.Inventory and not dandelion_ins.hasHash(chunk):
                         try:
                             del i.objectsNewToMe[chunk]
                         except KeyError:

src/network/invthread.py

@@ -8,8 +8,8 @@
 import addresses
 import protocol
 import state
-import dandelion
 import connectionpool
+from network import dandelion_ins
 from queues import invQueue
 from threads import StoppableThread
 
@@ -40,10 +40,10 @@ class InvThread(StoppableThread):
     @staticmethod
     def handleLocallyGenerated(stream, hashId):
         """Locally generated inventory items require special handling"""
-        dandelion.instance.addHash(hashId, stream=stream)
+        dandelion_ins.addHash(hashId, stream=stream)
         for connection in connectionpool.pool.connections():
-            if state.dandelion_enabled and connection != \
-                    dandelion.instance.objectChildStem(hashId):
+            if dandelion_ins.enabled and connection != \
+                    dandelion_ins.objectChildStem(hashId):
                 continue
             connection.objectsNewToThem[hashId] = time()
 
@@ -52,7 +52,7 @@ def run(self):  # pylint: disable=too-many-branches
             chunk = []
             while True:
                 # Dandelion fluff trigger by expiration
-                handleExpiredDandelion(dandelion.instance.expire())
+                handleExpiredDandelion(dandelion_ins.expire(invQueue))
                 try:
                     data = invQueue.get(False)
                     chunk.append((data[0], data[1]))
@@ -75,10 +75,10 @@ def run(self):  # pylint: disable=too-many-branches
                         except KeyError:
                             continue
                         try:
-                            if connection == dandelion.instance.objectChildStem(inv[1]):
+                            if connection == dandelion_ins.objectChildStem(inv[1]):
                                 # Fluff trigger by RNG
                                 # auto-ignore if config set to 0, i.e. dandelion is off
-                                if random.randint(1, 100) >= state.dandelion_enabled:  # nosec B311
+                                if random.randint(1, 100) >= dandelion_ins.enabled:  # nosec B311
                                     fluffs.append(inv[1])
                                 # send a dinv only if the stem node supports dandelion
                                 elif connection.services & protocol.NODE_DANDELION > 0:
@@ -105,7 +105,6 @@ def run(self):  # pylint: disable=too-many-branches
             for _ in range(len(chunk)):
                 invQueue.task_done()
 
-            if dandelion.instance.refresh < time():
-                dandelion.instance.reRandomiseStems()
+            dandelion_ins.reRandomiseStems()
 
             self.stop.wait(1)

src/network/objectracker.py

@@ -4,8 +4,8 @@
 import time
 from threading import RLock
 
-import dandelion
 import connectionpool
+from network import dandelion_ins
 from randomtrackingdict import RandomTrackingDict
 
 haveBloom = False
@@ -107,14 +107,14 @@ def handleReceivedObject(self, streamNumber, hashid):
                 del i.objectsNewToMe[hashid]
             except KeyError:
                 if streamNumber in i.streams and (
-                        not dandelion.instance.hasHash(hashid)
-                        or dandelion.instance.objectChildStem(hashid) == i):
+                        not dandelion_ins.hasHash(hashid)
+                        or dandelion_ins.objectChildStem(hashid) == i):
                     with i.objectsNewToThemLock:
                         i.objectsNewToThem[hashid] = time.time()
                     # update stream number,
                     # which we didn't have when we just received the dinv
                     # also resets expiration of the stem mode
-                    dandelion.instance.setHashStream(hashid, streamNumber)
+                    dandelion_ins.setHashStream(hashid, streamNumber)
 
             if i == self:
                 try:

src/network/tcp.py

@@ -15,10 +15,10 @@
 import l10n
 import protocol
 import state
-import dandelion
 import connectionpool
 from bmconfigparser import config
 from highlevelcrypto import randomBytes
+from network import dandelion_ins
 from queues import invQueue, receiveDataQueue, UISignalQueue
 from tr import _translate
 
@@ -169,7 +169,7 @@ def set_connection_fully_established(self):
             knownnodes.increaseRating(self.destination)
             knownnodes.addKnownNode(
                 self.streams, self.destination, time.time())
-            dandelion.instance.maybeAddStem(self)
+            dandelion_ins.maybeAddStem(self, invQueue)
         self.sendAddr()
         self.sendBigInv()
 
@@ -231,7 +231,7 @@ def sendChunk():
             with self.objectsNewToThemLock:
                 for objHash in state.Inventory.unexpired_hashes_by_stream(stream):
                     # don't advertise stem objects on bigInv
-                    if dandelion.instance.hasHash(objHash):
+                    if dandelion_ins.hasHash(objHash):
                         continue
                     bigInvList[objHash] = 0
         objectCount = 0
@@ -268,7 +268,7 @@ def handle_connect(self):
         self.append_write_buf(
             protocol.assembleVersionMessage(
                 self.destination.host, self.destination.port,
-                connectionpool.pool.streams,
+                connectionpool.pool.streams, dandelion_ins.enabled,
                 False, nodeid=self.nodeid))
         self.connectedAt = time.time()
         receiveDataQueue.put(self.destination)
@@ -293,7 +293,7 @@ def handle_close(self):
             if host_is_global:
                 knownnodes.addKnownNode(
                     self.streams, self.destination, time.time())
-                dandelion.instance.maybeRemoveStem(self)
+                dandelion_ins.maybeRemoveStem(self)
         else:
             self.checkTimeOffsetNotification()
             if host_is_global:
@@ -319,7 +319,7 @@ def state_proxy_handshake_done(self):
         self.append_write_buf(
             protocol.assembleVersionMessage(
                 self.destination.host, self.destination.port,
-                connectionpool.pool.streams,
+                connectionpool.pool.streams, dandelion_ins.enabled,
                 False, nodeid=self.nodeid))
         self.set_state("bm_header", expectBytes=protocol.Header.size)
         return True
@@ -343,7 +343,7 @@ def state_proxy_handshake_done(self):
         self.append_write_buf(
             protocol.assembleVersionMessage(
                 self.destination.host, self.destination.port,
-                connectionpool.pool.streams,
+                connectionpool.pool.streams, dandelion_ins.enabled,
                 False, nodeid=self.nodeid))
         self.set_state("bm_header", expectBytes=protocol.Header.size)
         return True