Download timeline in background thread. (#2016)

* Successfully sync download timeline in background thread.

* Got background thread download timeline working

* Simplify params for network controller

* Change peer download interval to 30 seconds

squeaknode/client/network_controller.py

@@ -35,11 +35,13 @@ DOWNLOAD_TIMEOUT_S = 10
 
 class NetworkController:
 
-    def __init__(self, squeak_store: SqueakStore, config):
+    def __init__(self, squeak_store: SqueakStore):
         self.squeak_store = squeak_store
-        self.config = config
 
-    def download_timeline(self) -> None:
+    def download_timeline_async(
+            self,
+            interest_block_interval: int,
+    ) -> None:
         min_block = 0  # TODO
         max_block = 999999999999  # TODO
         followed_public_keys = self.squeak_store.get_followed_public_keys()
@@ -52,9 +54,9 @@ class NetworkController:
                 max_block,
                 followed_public_keys,
             )
-            downloader.download()
+            downloader.download_async()
 
-    def download_pubkey_squeaks(self, pubkey: SqueakPublicKey) -> None:
+    def download_pubkey_squeaks_async(self, pubkey: SqueakPublicKey) -> None:
         min_block = 0  # TODO
         max_block = 999999999999  # TODO
         peers = self.squeak_store.get_autoconnect_peers()
@@ -66,7 +68,7 @@ class NetworkController:
                 max_block,
                 [pubkey],
             )
-            downloader.download()
+            downloader.download_async()
 
     def download_single_squeak(self, squeak_hash: bytes) -> None:
         peers = self.squeak_store.get_autoconnect_peers()
@@ -76,4 +78,4 @@ class NetworkController:
                 self.squeak_store,
                 squeak_hash,
             )
-            downloader.download()
+            downloader.download_async()

squeaknode/client/peer_client.py

@@ -33,7 +33,7 @@ from squeaknode.core.squeak_peer import SqueakPeer
 logger = logging.getLogger(__name__)
 
 
-DOWNLOAD_TIMEOUT_S = 10
+REQUEST_TIMEOUT_S = 10
 
 
 class PeerClient:
@@ -58,7 +58,11 @@ class PeerClient:
             'pubkeys': pubkeys_str,
         }
         url = f"{self.base_url}/lookup"
-        r = requests.get(url, params=payload)  # type: ignore
+        r = requests.get(  # type: ignore
+            url,
+            params=payload,  # type: ignore
+            timeout=REQUEST_TIMEOUT_S,
+        )
         squeak_hashes_str = r.json()
         return [
             bytes.fromhex(squeak_hash_str)
@@ -68,7 +72,7 @@ class PeerClient:
     def get_squeak(self, squeak_hash: bytes) -> Optional[CSqueak]:
         squeak_hash_str = squeak_hash.hex()
         url = f"{self.base_url}/squeak/{squeak_hash_str}"
-        r = requests.get(url)
+        r = requests.get(url, timeout=REQUEST_TIMEOUT_S)
         if r.status_code != requests.codes.ok:
             return None
         squeak_bytes = r.content
@@ -77,7 +81,7 @@ class PeerClient:
     def get_secret_key(self, squeak_hash: bytes) -> Optional[bytes]:
         squeak_hash_str = squeak_hash.hex()
         url = f"{self.base_url}/secretkey/{squeak_hash_str}"
-        r = requests.get(url)
+        r = requests.get(url, timeout=REQUEST_TIMEOUT_S)
         if r.status_code != requests.codes.ok:
             return None
         secret_key = r.content
@@ -86,7 +90,7 @@ class PeerClient:
     def get_offer(self, squeak_hash: bytes) -> Optional[Offer]:
         squeak_hash_str = squeak_hash.hex()
         url = f"{self.base_url}/offer/{squeak_hash_str}"
-        r = requests.get(url)
+        r = requests.get(url, timeout=REQUEST_TIMEOUT_S)
         if r.status_code != requests.codes.ok:
             return None
         offer_json = r.json()

squeaknode/client/peer_downloader.py

@@ -20,6 +20,7 @@
 # OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
 # SOFTWARE.
 import logging
+import threading
 from abc import ABC
 from abc import abstractmethod
 from typing import List
@@ -55,17 +56,24 @@ class PeerDownloader(ABC):
         """Return true if squeak is supposed to be downloaded.
         """
 
+    def download_async(self) -> None:
+        thread = threading.Thread(
+            target=self.download,
+            args=(),
+        )
+        thread.start()
+
     def download(self) -> None:
         squeak_hashes = self.get_hashes()
         for squeak_hash in squeak_hashes:
             # Download the squeak if not already owned.
-            self.download_squeak(squeak_hash)
+            self.get_squeak(squeak_hash)
             # Download the secret key if not already unlocked.
-            self.download_secret_key(squeak_hash)
+            self.get_secret_key(squeak_hash)
             # Download the offer if not already unlocked.
-            self.download_offer(squeak_hash)
+            self.get_offer(squeak_hash)
 
-    def download_squeak(self, squeak_hash: bytes) -> None:
+    def get_squeak(self, squeak_hash: bytes) -> None:
         # Download the squeak if not already owned.
         if self.squeak_store.get_squeak(squeak_hash):
             return
@@ -73,7 +81,7 @@ class PeerDownloader(ABC):
         if squeak and self.is_squeak_wanted(squeak):
             self.squeak_store.save_squeak(squeak)
 
-    def download_secret_key(self, squeak_hash: bytes) -> None:
+    def get_secret_key(self, squeak_hash: bytes) -> None:
         # Get the squeak from the database.
         squeak = self.squeak_store.get_squeak(squeak_hash)
         if squeak and self.is_squeak_wanted(squeak):
@@ -84,7 +92,7 @@ class PeerDownloader(ABC):
             if secret_key:
                 self.squeak_store.save_secret_key(squeak_hash, secret_key)
 
-    def download_offer(self, squeak_hash: bytes) -> None:
+    def get_offer(self, squeak_hash: bytes) -> None:
         # Get the squeak from the database.
         squeak = self.squeak_store.get_squeak(squeak_hash)
         if squeak and self.is_squeak_wanted(squeak):

squeaknode/config/config.py

@@ -62,7 +62,7 @@ DEFAULT_INTEREST_BLOCK_INTERVAL = 2016
 DEFAULT_SENT_OFFER_RETENTION_S = 86400
 DEFAULT_RECEIVED_OFFER_RETENTION_S = 86400
 DEFAULT_OFFER_DELETION_INTERVAL_S = 10
-DEFAULT_PEER_AUTOCONNECT_INTERVAL_S = 10
+DEFAULT_PEER_DOWNLOAD_INTERVAL_S = 30
 DEFAULT_SUBSCRIBE_INVOICES_RETRY_S = 10
 DEFAULT_SQUEAK_RETENTION_S = 604800
 DEFAULT_SQUEAK_DELETION_INTERVAL_S = 10
@@ -152,8 +152,8 @@ class NodeConfig(Config):
         cast=int, required=False, default=DEFAULT_OFFER_DELETION_INTERVAL_S)
     interest_block_interval = key(
         cast=int, required=False, default=DEFAULT_INTEREST_BLOCK_INTERVAL)
-    peer_autoconnect_interval_s = key(
-        cast=int, required=False, default=DEFAULT_PEER_AUTOCONNECT_INTERVAL_S)
+    peer_download_interval_s = key(
+        cast=int, required=False, default=DEFAULT_PEER_DOWNLOAD_INTERVAL_S)
 
 
 @section('db')

squeaknode/node/squeak_controller.py

@@ -296,7 +296,7 @@ class SqueakController:
         return self.squeak_store.get_squeak_entry(squeak_hash)
 
     def download_single_squeak(self, squeak_hash: bytes) -> DownloadResult:
-        network_controller = NetworkController(self.squeak_store, self.config)
+        network_controller = NetworkController(self.squeak_store)
         network_controller.download_single_squeak(squeak_hash)
         return DownloadResult(1, 1, 0, 9999)
 
@@ -306,10 +306,10 @@ class SqueakController:
             last_entry: Optional[SqueakEntry],
     ) -> List[SqueakEntry]:
         # TODO: remove this temporary hack, after converting this to websockets.
-        logger.info('Start downloading timeline...')
-        network_controller = NetworkController(self.squeak_store, self.config)
-        network_controller.download_timeline()
-        logger.info('Finished downloading timeline.')
+        # logger.info('Start downloading timeline...')
+        # network_controller = NetworkController(self.squeak_store)
+        # network_controller.download_timeline()
+        # logger.info('Finished downloading timeline.')
         return self.squeak_store.get_timeline_squeak_entries(limit, last_entry)
 
     def get_liked_squeak_entries(
@@ -341,8 +341,8 @@ class SqueakController:
     ) -> List[SqueakEntry]:
         # TODO: remove this temporary hack, after converting this to websockets.
         logger.info('Start downloading pubkey squeaks...')
-        network_controller = NetworkController(self.squeak_store, self.config)
-        network_controller.download_pubkey_squeaks(public_key)
+        network_controller = NetworkController(self.squeak_store)
+        network_controller.download_pubkey_squeaks_async(public_key)
         logger.info('Finished downloading pubkey squeaks.')
         return self.squeak_store.get_squeak_entries_for_public_key(
             public_key,

squeaknode/node/squeak_download_worker.py

@@ -0,0 +1,54 @@
+# MIT License
+#
+# Copyright (c) 2020 Jonathan Zernik
+#
+# Permission is hereby granted, free of charge, to any person obtaining a copy
+# of this software and associated documentation files (the "Software"), to deal
+# in the Software without restriction, including without limitation the rights
+# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+# copies of the Software, and to permit persons to whom the Software is
+# furnished to do so, subject to the following conditions:
+#
+# The above copyright notice and this permission notice shall be included in all
+# copies or substantial portions of the Software.
+#
+# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+# SOFTWARE.
+import logging
+
+from squeaknode.client.network_controller import NetworkController
+from squeaknode.node.periodic_worker import PeriodicWorker
+from squeaknode.node.squeak_store import SqueakStore
+
+logger = logging.getLogger(__name__)
+
+
+class SqueakDownloadWorker(PeriodicWorker):
+    def __init__(
+        self,
+        squeak_store: SqueakStore,
+        download_timeline_interval_s: int,
+        interest_block_interval: int,
+    ):
+        self.squeak_store = squeak_store
+        self.download_timeline_interval_s = download_timeline_interval_s
+        self.interest_block_interval = interest_block_interval
+        self.network_controller = NetworkController(
+            self.squeak_store,
+        )
+
+    def work_fn(self):
+        self.network_controller.download_timeline_async(
+            self.interest_block_interval,
+        )
+
+    def get_interval_s(self):
+        return self.download_timeline_interval_s
+
+    def get_name(self):
+        return "squeak_download_worker"

squeaknode/node/squeak_node.py

@@ -40,6 +40,7 @@ from squeaknode.node.process_forward_tweets_worker import ProcessForwardTweetsWo
 from squeaknode.node.process_received_payments_worker import ProcessReceivedPaymentsWorker
 from squeaknode.node.squeak_controller import SqueakController
 from squeaknode.node.squeak_deletion_worker import SqueakDeletionWorker
+from squeaknode.node.squeak_download_worker import SqueakDownloadWorker
 from squeaknode.node.squeak_offer_expiry_worker import SqueakOfferExpiryWorker
 from squeaknode.node.squeak_store import SqueakStore
 from squeaknode.server.app import SqueakPeerWebServer
@@ -72,6 +73,7 @@ class SqueakNode:
         self.create_admin_web_server()
         self.create_received_payment_processor_worker()
         self.create_squeak_deletion_worker()
+        self.create_squeak_download_worker()
         self.create_offer_expiry_worker()
         self.create_forward_tweets_processor_worker()
 
@@ -86,6 +88,7 @@ class SqueakNode:
         self.peer_web_server.start()
         self.received_payment_processor_worker.start_running()
         self.squeak_deletion_worker.start()
+        self.squeak_download_worker.start()
         self.offer_expiry_worker.start()
         self.forward_tweets_processor_worker.start_running()
 
@@ -222,6 +225,13 @@ class SqueakNode:
             self.config.node.squeak_deletion_interval_s,
         )
 
+    def create_squeak_download_worker(self):
+        self.squeak_download_worker = SqueakDownloadWorker(
+            self.squeak_store,
+            self.config.node.peer_download_interval_s,
+            self.config.node.interest_block_interval,
+        )
+
     def create_offer_expiry_worker(self):
         self.offer_expiry_worker = SqueakOfferExpiryWorker(
             self.squeak_store,