diff --git a/squeaknode/client/network_controller.py b/squeaknode/client/network_controller.py index df833db2..a0871b2f 100644 --- a/squeaknode/client/network_controller.py +++ b/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() diff --git a/squeaknode/client/peer_client.py b/squeaknode/client/peer_client.py index 8f01d9e1..8657bc87 100644 --- a/squeaknode/client/peer_client.py +++ b/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() diff --git a/squeaknode/client/peer_downloader.py b/squeaknode/client/peer_downloader.py index a84071a2..b976993a 100644 --- a/squeaknode/client/peer_downloader.py +++ b/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): diff --git a/squeaknode/config/config.py b/squeaknode/config/config.py index 50c3122b..2eb3190a 100644 --- a/squeaknode/config/config.py +++ b/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') diff --git a/squeaknode/node/squeak_controller.py b/squeaknode/node/squeak_controller.py index ec37f3cb..0e72775b 100644 --- a/squeaknode/node/squeak_controller.py +++ b/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, diff --git a/squeaknode/node/squeak_download_worker.py b/squeaknode/node/squeak_download_worker.py new file mode 100644 index 00000000..aa35b2a9 --- /dev/null +++ b/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" diff --git a/squeaknode/node/squeak_node.py b/squeaknode/node/squeak_node.py index cce86c45..25fb0268 100644 --- a/squeaknode/node/squeak_node.py +++ b/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,