Refactor conditions to add a contact to the routing table

.appveyor.yml

@@ -23,7 +23,7 @@ clone_folder: c:\projects\lbry
 test_script:
 - cd C:\projects\lbry\
 - pip install cython
-- pip install mock pylint unqlite
+- pip install mock pylint unqlite Faker
 - pip install .
 - pylint lbrynet
 # disable tests for now so that appveyor can build the app

CHANGELOG.md

@@ -36,10 +36,10 @@ at anytime.
   * several internal dht functions to use inlineCallbacks
   * `DHTHashAnnouncer` and `Node` manage functions to use `LoopingCall`s instead of scheduling with `callLater`.
   * `store` kademlia rpc method to block on the call finishing and to return storing peer information
-  * refactored `DHTHashAnnouncer` to longer use locks, use a `DeferredSemaphore` to limit concurrent announcers
+  * refactored `DHTHashAnnouncer` to no longer use locks, use a `DeferredSemaphore` to limit concurrent announcers
   * decoupled `DiskBlobManager` from `DHTHashAnnouncer`
   * blob hashes to announce to be controlled by`SQLiteStorage`
-  * kademlia protocol to not delay writes to the UDP socket
+  * kademlia protocol to minimally delay writes to the UDP socket
   * `reactor` and `callLater`, `listenUDP`, and `resolve` functions to be configurable (to allow easier testing)
   * calls to get the current time to use `reactor.seconds` (to control callLater and LoopingCall timing in tests)
   * `blob_announce` to queue the blob announcement but not block on it
@@ -56,21 +56,34 @@ at anytime.
   * track successful reflector uploads in sqlite to minimize how many streams are attempted by auto re-reflect
   * increase the default `auto_re_reflect_interval` to a day
   * predictable result sorting for `claim_list` and `claim_list_mine`
+  * changed the bucket splitting condition in the dht routing table to be more aggressive
+  * ping dht nodes who have stored to us periodically to determine whether we should include them as an active peer for the hash when we are queried. Nodes that are known to be not reachable by the node storing the record are no longer returned as peers by the storing node.
+  * temporarily disabled data price negotiation, treat all data as free
+  * changed dht bootstrap join process to better populate the routing table initially
+  * cache dht node tokens used during announcement to minimize the number of requests that are needed
+  * implement BEP0005 dht rules to classify nodes as good, bad, or unknown and for when to add them to the routing table (http://www.bittorrent.org/beps/bep_0005.html)
+  * refactored internal dht contact class to track failure counts/times, the time the contact last replied to us, and the time the node last requested something fom us
+  * refactored dht iterativeFind
+  * sort dht contacts returned by `findCloseNodes` in the routing table
+  * disabled Cryptonator price feed
 
 ### Added
   * virtual kademlia network and mock udp transport for dht integration tests
-  * integration tests for bootstrapping the dht
+  * functional tests for bootstrapping the dht, announcing and expiring hashes, finding and pinging nodes, protocol version 0/1 backwards/forwards compatibility, and rejoining the network
   * configurable `concurrent_announcers` and `s3_headers_depth` settings
   * `peer_ping` command
   * `--sort` option in `file_list`
   * linux distro and desktop name added to analytics
   * certifi module for Twisted SSL verification on Windows
+  * protocol version to dht requests and to the response from `findValue`
+  * added `port` field to contacts returned by `routing_table_get`
 
 ### Removed
   * `announce_all` argument from `blob_announce`
   * old `blob_announce_all` command
   * `AuthJSONRPCServer.auth_required` decorator
   * unused `--wallet` argument to `lbrynet-daemon`, which used to be to support `PTCWallet`.
+  * `OptimizedTreeRoutingTable` class used by the dht node for the time being
 
 ## [0.19.3] - 2018-05-04
 ### Changed

lbrynet/conf.py

@@ -40,7 +40,7 @@
 KB = 2 ** 10
 MB = 2 ** 20
 
-DEFAULT_CONCURRENT_ANNOUNCERS = 100
+DEFAULT_CONCURRENT_ANNOUNCERS = 10
 
 DEFAULT_DHT_NODES = [
     ('lbrynet1.lbry.io', 4444),

lbrynet/core/BlobAvailability.py

@@ -23,18 +23,14 @@ def __init__(self, blob_manager, peer_finder, dht_node):
         self._blob_manager = blob_manager
         self._peer_finder = peer_finder
         self._dht_node = dht_node
-        self._check_popular = LoopingCall(self._update_most_popular)
         self._check_mine = LoopingCall(self._update_mine)
 
     def start(self):
         log.info("Starting blob availability tracker.")
-        self._check_popular.start(600)
         self._check_mine.start(600)
 
     def stop(self):
         log.info("Stopping blob availability tracker.")
-        if self._check_popular.running:
-            self._check_popular.stop()
         if self._check_mine.running:
             self._check_mine.stop()
 
@@ -68,17 +64,6 @@ def _save_peer_info(blob_hash, peers):
         d.addCallback(lambda peers: _save_peer_info(blob, peers))
         return d
 
-    def _get_most_popular(self):
-        dl = []
-        for (hash, _) in self._dht_node.get_most_popular_hashes(10):
-            encoded = hash.encode('hex')
-            dl.append(self._update_peers_for_blob(encoded))
-        return defer.DeferredList(dl)
-
-    def _update_most_popular(self):
-        d = self._get_most_popular()
-        d.addCallback(lambda _: self._set_mean_peers())
-
     def _update_mine(self):
         def _get_peers(blobs):
             dl = []

lbrynet/core/PaymentRateManager.py

@@ -93,6 +93,7 @@ def __init__(self, **kwargs):
 
         self.base = BasePaymentRateManager(0.0, 0.0)
         self.points_paid = 0.0
+        self.min_blob_data_payment_rate = 0.0
         self.generous = True
         self.strategy = OnlyFreeStrategy()
 

lbrynet/core/Session.py

@@ -6,8 +6,7 @@
 from lbrynet.database.storage import SQLiteStorage
 from lbrynet.core.RateLimiter import RateLimiter
 from lbrynet.core.utils import generate_id
-from lbrynet.core.PaymentRateManager import BasePaymentRateManager, NegotiatedPaymentRateManager
-from lbrynet.core.BlobAvailability import BlobAvailabilityTracker
+from lbrynet.core.PaymentRateManager import BasePaymentRateManager, OnlyFreePaymentsManager
 
 log = logging.getLogger(__name__)
 
@@ -33,14 +32,11 @@ class Session(object):
     peers can connect to this peer.
     """
 
-    def __init__(self, blob_data_payment_rate, db_dir=None,
-                 node_id=None, peer_manager=None, dht_node_port=None,
-                 known_dht_nodes=None, peer_finder=None,
-                 hash_announcer=None, blob_dir=None,
-                 blob_manager=None, peer_port=None, use_upnp=True,
-                 rate_limiter=None, wallet=None,
-                 dht_node_class=node.Node, blob_tracker_class=None,
-                 payment_rate_manager_class=None, is_generous=True, external_ip=None, storage=None):
+    def __init__(self, blob_data_payment_rate, db_dir=None, node_id=None, peer_manager=None, dht_node_port=None,
+                 known_dht_nodes=None, peer_finder=None, hash_announcer=None, blob_dir=None, blob_manager=None,
+                 peer_port=None, use_upnp=True, rate_limiter=None, wallet=None, dht_node_class=node.Node,
+                 blob_tracker_class=None, payment_rate_manager_class=None, is_generous=True, external_ip=None,
+                 storage=None):
         """@param blob_data_payment_rate: The default payment rate for blob data
 
         @param db_dir: The directory in which levelDB files should be stored
@@ -107,8 +103,8 @@ def __init__(self, blob_data_payment_rate, db_dir=None,
             self.known_dht_nodes = []
         self.blob_dir = blob_dir
         self.blob_manager = blob_manager
-        self.blob_tracker = None
-        self.blob_tracker_class = blob_tracker_class or BlobAvailabilityTracker
+        # self.blob_tracker = None
+        # self.blob_tracker_class = blob_tracker_class or BlobAvailabilityTracker
         self.peer_port = peer_port
         self.use_upnp = use_upnp
         self.rate_limiter = rate_limiter
@@ -118,11 +114,10 @@ def __init__(self, blob_data_payment_rate, db_dir=None,
         self.dht_node_class = dht_node_class
         self.dht_node = None
         self.base_payment_rate_manager = BasePaymentRateManager(blob_data_payment_rate)
-        self.payment_rate_manager = None
-        self.payment_rate_manager_class = payment_rate_manager_class or NegotiatedPaymentRateManager
-        self.is_generous = is_generous
+        self.payment_rate_manager = OnlyFreePaymentsManager()
+        # self.payment_rate_manager_class = payment_rate_manager_class or NegotiatedPaymentRateManager
+        # self.is_generous = is_generous
         self.storage = storage or SQLiteStorage(self.db_dir)
-        self._join_dht_deferred = None
 
     def setup(self):
         """Create the blob directory and database if necessary, start all desired services"""
@@ -147,8 +142,8 @@ def shut_down(self):
         ds = []
         if self.hash_announcer:
             self.hash_announcer.stop()
-        if self.blob_tracker is not None:
-            ds.append(defer.maybeDeferred(self.blob_tracker.stop))
+        # if self.blob_tracker is not None:
+        #     ds.append(defer.maybeDeferred(self.blob_tracker.stop))
         if self.dht_node is not None:
             ds.append(defer.maybeDeferred(self.dht_node.stop))
         if self.rate_limiter is not None:
@@ -171,24 +166,24 @@ def get_free_port(upnp, port, protocol):
             if not mapping:
                 return port
             if upnp.lanaddr == mapping[0]:
-                return mapping
+                return mapping[1]
             return get_free_port(upnp, port + 1, protocol)
 
-        def get_port_mapping(upnp, internal_port, protocol, description):
+        def get_port_mapping(upnp, port, protocol, description):
             # try to map to the requested port, if there is already a mapping use the next external
             # port available
             if protocol not in ['UDP', 'TCP']:
                 raise Exception("invalid protocol")
-            external_port = get_free_port(upnp, internal_port, protocol)
-            if isinstance(external_port, tuple):
+            port = get_free_port(upnp, port, protocol)
+            if isinstance(port, tuple):
                 log.info("Found existing UPnP redirect %s:%i (%s) to %s:%i, using it",
-                         self.external_ip, external_port[1], protocol, upnp.lanaddr, internal_port)
-                return external_port[1], protocol
-            upnp.addportmapping(external_port, protocol, upnp.lanaddr, internal_port,
+                         self.external_ip, port, protocol, upnp.lanaddr, port)
+                return port
+            upnp.addportmapping(port, protocol, upnp.lanaddr, port,
                                 description, '')
-            log.info("Set UPnP redirect %s:%i (%s) to %s:%i", self.external_ip, external_port,
-                     protocol, upnp.lanaddr, internal_port)
-            return external_port, protocol
+            log.info("Set UPnP redirect %s:%i (%s) to %s:%i", self.external_ip, port,
+                     protocol, upnp.lanaddr, port)
+            return port
 
         def threaded_try_upnp():
             if self.use_upnp is False:
@@ -203,13 +198,11 @@ def threaded_try_upnp():
                     # best not to rely on this external ip, the router can be behind layers of NATs
                     self.external_ip = external_ip
                 if self.peer_port:
-                    self.upnp_redirects.append(
-                        get_port_mapping(u, self.peer_port, 'TCP', 'LBRY peer port')
-                    )
+                    self.peer_port = get_port_mapping(u, self.peer_port, 'TCP', 'LBRY peer port')
+                    self.upnp_redirects.append((self.peer_port, 'TCP'))
                 if self.dht_node_port:
-                    self.upnp_redirects.append(
-                        get_port_mapping(u, self.dht_node_port, 'UDP', 'LBRY DHT port')
-                    )
+                    self.dht_node_port = get_port_mapping(u, self.dht_node_port, 'UDP', 'LBRY DHT port')
+                    self.upnp_redirects.append((self.dht_node_port, 'UDP'))
                 return True
             return False
 
@@ -234,9 +227,9 @@ def _setup_dht(self):  # does not block startup, the dht will re-attempt if nece
             self.hash_announcer = hashannouncer.DHTHashAnnouncer(self.dht_node, self.storage)
         self.peer_manager = self.dht_node.peer_manager
         self.peer_finder = self.dht_node.peer_finder
-        self._join_dht_deferred = self.dht_node.joinNetwork(self.known_dht_nodes)
-        self._join_dht_deferred.addCallback(lambda _: log.info("Joined the dht"))
-        self._join_dht_deferred.addCallback(lambda _: self.hash_announcer.start())
+        d = self.dht_node.start(self.known_dht_nodes)
+        d.addCallback(lambda _: log.info("Joined the dht"))
+        d.addCallback(lambda _: self.hash_announcer.start())
 
     def _setup_other_components(self):
         log.debug("Setting up the rest of the components")
@@ -251,19 +244,19 @@ def _setup_other_components(self):
             else:
                 self.blob_manager = DiskBlobManager(self.blob_dir, self.storage)
 
-        if self.blob_tracker is None:
-            self.blob_tracker = self.blob_tracker_class(
-                self.blob_manager, self.dht_node.peer_finder, self.dht_node
-            )
-        if self.payment_rate_manager is None:
-            self.payment_rate_manager = self.payment_rate_manager_class(
-                self.base_payment_rate_manager, self.blob_tracker, self.is_generous
-            )
+        # if self.blob_tracker is None:
+        #     self.blob_tracker = self.blob_tracker_class(
+        #         self.blob_manager, self.dht_node.peer_finder, self.dht_node
+        #     )
+        # if self.payment_rate_manager is None:
+        #     self.payment_rate_manager = self.payment_rate_manager_class(
+        #         self.base_payment_rate_manager, self.blob_tracker, self.is_generous
+        #     )
 
         self.rate_limiter.start()
         d = self.blob_manager.setup()
         d.addCallback(lambda _: self.wallet.start())
-        d.addCallback(lambda _: self.blob_tracker.start())
+        # d.addCallback(lambda _: self.blob_tracker.start())
         return d
 
     def _unset_upnp(self):

lbrynet/core/call_later_manager.py

@@ -9,22 +9,25 @@
 
 
 class CallLaterManager(object):
-    _callLater = None
-    _pendingCallLaters = []
-    _delay = MIN_DELAY
-
-    @classmethod
-    def get_min_delay(cls):
-        cls._pendingCallLaters = [cl for cl in cls._pendingCallLaters if cl.active()]
-        queue_size = len(cls._pendingCallLaters)
+    def __init__(self, callLater):
+        """
+        :param callLater: (IReactorTime.callLater)
+        """
+
+        self._callLater = callLater
+        self._pendingCallLaters = []
+        self._delay = MIN_DELAY
+
+    def get_min_delay(self):
+        self._pendingCallLaters = [cl for cl in self._pendingCallLaters if cl.active()]
+        queue_size = len(self._pendingCallLaters)
         if queue_size > QUEUE_SIZE_THRESHOLD:
-            cls._delay = min((cls._delay + DELAY_INCREMENT), MAX_DELAY)
+            self._delay = min((self._delay + DELAY_INCREMENT), MAX_DELAY)
         else:
-            cls._delay = max((cls._delay - 2.0 * DELAY_INCREMENT), MIN_DELAY)
-        return cls._delay
+            self._delay = max((self._delay - 2.0 * DELAY_INCREMENT), MIN_DELAY)
+        return self._delay
 
-    @classmethod
-    def _cancel(cls, call_later):
+    def _cancel(self, call_later):
         """
         :param call_later: DelayedCall
         :return: (callable) canceller function
@@ -38,26 +41,25 @@ def cancel(reason=None):
 
             if call_later.active():
                 call_later.cancel()
-            cls._pendingCallLaters.remove(call_later)
+            if call_later in self._pendingCallLaters:
+                self._pendingCallLaters.remove(call_later)
             return reason
         return cancel
 
-    @classmethod
-    def stop(cls):
+    def stop(self):
         """
         Cancel any callLaters that are still running
         """
 
         from twisted.internet import defer
-        while cls._pendingCallLaters:
-            canceller = cls._cancel(cls._pendingCallLaters[0])
+        while self._pendingCallLaters:
+            canceller = self._cancel(self._pendingCallLaters[0])
             try:
                 canceller()
-            except (defer.CancelledError, defer.AlreadyCalledError):
+            except (defer.CancelledError, defer.AlreadyCalledError, ValueError):
                 pass
 
-    @classmethod
-    def call_later(cls, when, what, *args, **kwargs):
+    def call_later(self, when, what, *args, **kwargs):
         """
         Schedule a call later and get a canceller callback function
 
@@ -69,21 +71,11 @@ def call_later(cls, when, what, *args, **kwargs):
         :return: (tuple) twisted.internet.base.DelayedCall object, canceller function
         """
 
-        call_later = cls._callLater(when, what, *args, **kwargs)
-        canceller = cls._cancel(call_later)
-        cls._pendingCallLaters.append(call_later)
+        call_later = self._callLater(when, what, *args, **kwargs)
+        canceller = self._cancel(call_later)
+        self._pendingCallLaters.append(call_later)
         return call_later, canceller
 
-    @classmethod
-    def call_soon(cls, what, *args, **kwargs):
-        delay = cls.get_min_delay()
-        return cls.call_later(delay, what, *args, **kwargs)
-
-    @classmethod
-    def setup(cls, callLater):
-        """
-        Setup the callLater function to use, supports the real reactor as well as task.Clock
-
-        :param callLater: (IReactorTime.callLater)
-        """
-        cls._callLater = callLater
+    def call_soon(self, what, *args, **kwargs):
+        delay = self.get_min_delay()
+        return self.call_later(delay, what, *args, **kwargs)