diff --git a/itests/tests/test_squeak_node.py b/itests/tests/test_squeak_node.py index 8acaf187..c557a03b 100644 --- a/itests/tests/test_squeak_node.py +++ b/itests/tests/test_squeak_node.py @@ -873,12 +873,12 @@ def test_connect_other_node( ) # Sync squeaks - sync_squeaks_response = other_admin_stub.SyncSqueaks( + other_admin_stub.SyncSqueaks( squeak_admin_pb2.SyncSqueaksRequest(), ) - # time.sleep(10) - print(sync_squeaks_response) - assert peer_id in sync_squeaks_response.sync_result.completed_peer_ids + time.sleep(5) + # print(sync_squeaks_response) + # assert peer_id in sync_squeaks_response.sync_result.completed_peer_ids # Get the sent offers from the seller node get_sent_offers_response = admin_stub.GetSentOffers( @@ -1072,9 +1072,9 @@ def test_download_single_squeak( squeak_hash=saved_squeak_hash, ), ) - # time.sleep(10) + time.sleep(10) print(sync_squeak_response) - assert peer_id in sync_squeak_response.sync_result.completed_peer_ids + # assert peer_id in sync_squeak_response.sync_result.completed_peer_ids # Get the squeak display item get_squeak_display_response = other_admin_stub.GetSqueakDisplay( diff --git a/squeaknode/admin/squeak_admin_server_handler.py b/squeaknode/admin/squeak_admin_server_handler.py index e32e82ce..2d7e55e8 100644 --- a/squeaknode/admin/squeak_admin_server_handler.py +++ b/squeaknode/admin/squeak_admin_server_handler.py @@ -10,7 +10,6 @@ from squeaknode.admin.util import squeak_entry_to_detail_message from squeaknode.admin.util import squeak_entry_to_message from squeaknode.admin.util import squeak_peer_to_message from squeaknode.admin.util import squeak_profile_to_message -from squeaknode.admin.util import sync_result_to_message from squeaknode.core.squeak_controller import SqueakController from squeaknode.lightning.lnd_lightning_client import LNDLightningClient from squeaknode.sync.squeak_sync_status import SqueakSyncController @@ -390,10 +389,10 @@ class SqueakAdminServerHandler(object): def handle_sync_squeaks(self, request): logger.info("Handle sync squeaks") # sync_result = self.squeak_controller.sync_squeaks() - sync_result = self.sync_controller.sync_timeline() - sync_result_msg = sync_result_to_message(sync_result) + self.sync_controller.download_timeline() + self.sync_controller.upload_timeline() return squeak_admin_pb2.SyncSqueaksReply( - sync_result=sync_result_msg, + sync_result=None, ) def handle_sync_squeak(self, request): @@ -401,11 +400,11 @@ class SqueakAdminServerHandler(object): squeak_hash = bytes.fromhex(squeak_hash_str) logger.info( "Handle download squeak with hash: {}".format(squeak_hash_str)) - # sync_result = self.squeak_controller.sync_squeak(squeak_hash) - sync_result = self.sync_controller.sync_single_squeak(squeak_hash) - sync_result_msg = sync_result_to_message(sync_result) + # TODO: Add a separate method for download squeak and upload squeak. + self.sync_controller.download_single_squeak(squeak_hash) + self.sync_controller.upload_single_squeak(squeak_hash) return squeak_admin_pb2.SyncSqueakReply( - sync_result=sync_result_msg, + sync_result=None, ) def handle_pay_offer(self, request): diff --git a/squeaknode/core/squeak_controller.py b/squeaknode/core/squeak_controller.py index b9c2837f..22a88a11 100644 --- a/squeaknode/core/squeak_controller.py +++ b/squeaknode/core/squeak_controller.py @@ -187,6 +187,12 @@ class SqueakController: def get_peers(self): return self.squeak_db.get_peers() + def get_downloading_peers(self): + return self.squeak_db.get_downloading_peers() + + def get_uploading_peers(self): + return self.squeak_db.get_uploading_peers() + def set_peer_downloading(self, peer_id: int, downloading: bool): self.squeak_db.set_peer_downloading(peer_id, downloading) diff --git a/squeaknode/db/squeak_db.py b/squeaknode/db/squeak_db.py index 267fa198..7ce7d713 100644 --- a/squeaknode/db/squeak_db.py +++ b/squeaknode/db/squeak_db.py @@ -702,6 +702,24 @@ class SqueakDb: peers = [self._parse_squeak_peer(row) for row in rows] return peers + def get_downloading_peers(self): + """ Get peers that are set to be downloading. """ + s = select([self.peers]).where(self.peers.c.downloading) + with self.get_connection() as connection: + result = connection.execute(s) + rows = result.fetchall() + peers = [self._parse_squeak_peer(row) for row in rows] + return peers + + def get_uploading_peers(self): + """ Get peers that are set to be uploading. """ + s = select([self.peers]).where(self.peers.c.uploading) + with self.get_connection() as connection: + result = connection.execute(s) + rows = result.fetchall() + peers = [self._parse_squeak_peer(row) for row in rows] + return peers + def set_peer_downloading(self, peer_id, downloading): """ Set a peer is downloading. """ stmt = ( diff --git a/squeaknode/sync/network_sync.py b/squeaknode/sync/network_sync.py index 1f69df53..a5d931da 100644 --- a/squeaknode/sync/network_sync.py +++ b/squeaknode/sync/network_sync.py @@ -1,43 +1,42 @@ import logging -from squeaknode.sync.peer_task import PeerSyncTask logger = logging.getLogger(__name__) -class NetworkSync: - def __init__( - self, - squeak_controller, - ): - self.squeak_controller = squeak_controller +# class NetworkSync: +# def __init__( +# self, +# squeak_controller, +# ): +# self.squeak_controller = squeak_controller - def sync_timeline(self, peer, min_block, max_block): - if not peer.downloading: - return - # peer_connection = PeerConnection(peer) - # with PeerConnection(peer).open_connection() as peer_connection: - with PeerSyncTask( - self.squeak_controller, - peer, - None, - ).open_peer_sync_task() as peer_sync_task: - if peer.uploading: - peer_sync_task.upload(min_block, max_block) - if peer.downloading: - peer_sync_task.download(min_block, max_block) +# def sync_timeline(self, peer, min_block, max_block): +# if not peer.downloading: +# return +# # peer_connection = PeerConnection(peer) +# # with PeerConnection(peer).open_connection() as peer_connection: +# with PeerSyncTask( +# self.squeak_controller, +# peer, +# None, +# ).open_peer_sync_task() as peer_sync_task: +# if peer.uploading: +# peer_sync_task.upload(min_block, max_block) +# if peer.downloading: +# peer_sync_task.download(min_block, max_block) - def sync_single_squeak(self, peer, squeak_hash: bytes): - if not peer.downloading: - return - # peer_connection = PeerConnection(peer) - # with PeerConnection(peer).open_connection() as peer_connection: - with PeerSyncTask( - self.squeak_controller, - peer, - None, - ).open_peer_sync_task() as peer_sync_task: - if peer.uploading: - peer_sync_task.upload_single_squeak(squeak_hash) - if peer.downloading: - peer_sync_task.download_single_squeak(squeak_hash) +# def sync_single_squeak(self, peer, squeak_hash: bytes): +# if not peer.downloading: +# return +# # peer_connection = PeerConnection(peer) +# # with PeerConnection(peer).open_connection() as peer_connection: +# with PeerSyncTask( +# self.squeak_controller, +# peer, +# None, +# ).open_peer_sync_task() as peer_sync_task: +# if peer.uploading: +# peer_sync_task.upload_single_squeak(squeak_hash) +# if peer.downloading: +# peer_sync_task.download_single_squeak(squeak_hash) diff --git a/squeaknode/sync/network_task.py b/squeaknode/sync/network_task.py index ee370af2..1ad2c6b6 100644 --- a/squeaknode/sync/network_task.py +++ b/squeaknode/sync/network_task.py @@ -1,95 +1,194 @@ import logging -import queue import threading -from typing import Any -from typing import List -from typing import NamedTuple +from abc import ABC +from abc import abstractmethod -from squeaknode.sync.network_sync import NetworkSync +from squeaknode.sync.peer_task import PeerSyncTask logger = logging.getLogger(__name__) -class PeerSyncResult(NamedTuple): - completed_peer_id: Any = None - failed_peer_id: Any = None - timeout: Any = None +# class PeerSyncResult(NamedTuple): +# completed_peer_id: Any = None +# failed_peer_id: Any = None +# timeout: Any = None -class NetworkSyncResult(NamedTuple): - completed_peer_ids: List[int] - failed_peer_ids: List[int] - timeout_peer_ids: List[int] +# class NetworkSyncResult(NamedTuple): +# completed_peer_ids: List[int] +# failed_peer_ids: List[int] +# timeout_peer_ids: List[int] -class NetworkSyncTask: +# class NetworkSyncTask: +# def __init__( +# self, +# squeak_controller, +# ): +# self.squeak_controller = squeak_controller +# self.queue = queue.Queue() + +# def sync(self): +# peers = self.squeak_controller.get_peers() +# logger.debug( +# "Network sync for class {} with peers: {}".format( +# self.__class__, +# peers, +# ) +# ) +# run_sync_thread = threading.Thread( +# target=self._run_sync, +# args=(peers,), +# ) +# run_sync_thread.start() +# remaining_peer_ids = set(peer.peer_id for peer in peers) +# completed_peer_ids = set() +# failed_peer_ids = set() + +# while len(remaining_peer_ids) > 0: +# item = self.queue.get() +# logger.debug(f"Working on {item}") +# if item.completed_peer_id: +# completed_peer_ids.add(item.completed_peer_id) +# remaining_peer_ids.remove(item.completed_peer_id) +# if item.failed_peer_id: +# failed_peer_ids.add(item.failed_peer_id) +# remaining_peer_ids.remove(item.failed_peer_id) +# logger.debug(f"Finished {item}") +# self.queue.task_done() + +# logger.info("Finished sync with peers.") +# return NetworkSyncResult( +# completed_peer_ids=list(completed_peer_ids), +# failed_peer_ids=list(failed_peer_ids), +# timeout_peer_ids=list(remaining_peer_ids), +# ) + +# def _run_sync(self, peers): +# for peer in peers: +# sync_peer_thread = threading.Thread( +# target=self._sync_peer, +# args=(peer,), +# ) +# sync_peer_thread.start() + +# def sync_peer(self, peer): +# pass + +# def _sync_peer(self, peer): +# try: +# logger.debug("Trying to sync with peer: {}".format(peer.peer_id)) +# self.sync_peer(peer) +# self.queue.put(PeerSyncResult(completed_peer_id=peer.peer_id)) +# logger.info("Finished sync with peer: {}".format(peer)) +# except Exception: +# logger.error("Failed Sync with peer: {}.".format( +# peer), exc_info=True) +# self.queue.put(PeerSyncResult(failed_peer_id=peer.peer_id)) + + +# class TimelineNetworkSyncTask(NetworkSyncTask): +# def __init__( +# self, +# squeak_controller, +# min_block, +# max_block, +# ): +# super().__init__(squeak_controller) +# self.min_block = min_block +# self.max_block = max_block + +# def sync_peer(self, peer): +# network_sync = NetworkSync( +# self.squeak_controller, +# ) +# network_sync.sync_timeline(peer, self.min_block, self.max_block) + + +# class SingleSqueakNetworkSyncTask(NetworkSyncTask): +# def __init__( +# self, +# squeak_controller, +# squeak_hash: bytes, +# ): +# super().__init__(squeak_controller) +# self.squeak_hash = squeak_hash + +# def sync_peer(self, peer): +# network_sync = NetworkSync( +# self.squeak_controller, +# ) +# network_sync.sync_single_squeak(peer, self.squeak_hash) + + +class NetworkSync(ABC): + def __init__( self, squeak_controller, ): self.squeak_controller = squeak_controller - self.queue = queue.Queue() + self.stopped = threading.Event() + + @abstractmethod + def get_peers_to_sync(self): + pass + + # TODO: Rename peer_sync_task to peer_connection or something. + @abstractmethod + def sync_peer(self, peer_sync_task): + pass + + def stop(self): + self.stopped.set() def sync(self): peers = self.squeak_controller.get_peers() - logger.debug( - "Network sync for class {} with peers: {}".format( - self.__class__, - peers, - ) - ) - run_sync_thread = threading.Thread( - target=self._run_sync, - args=(peers,), - ) - run_sync_thread.start() - remaining_peer_ids = set(peer.peer_id for peer in peers) - completed_peer_ids = set() - failed_peer_ids = set() - - while len(remaining_peer_ids) > 0: - item = self.queue.get() - logger.debug(f"Working on {item}") - if item.completed_peer_id: - completed_peer_ids.add(item.completed_peer_id) - remaining_peer_ids.remove(item.completed_peer_id) - if item.failed_peer_id: - failed_peer_ids.add(item.failed_peer_id) - remaining_peer_ids.remove(item.failed_peer_id) - logger.debug(f"Finished {item}") - self.queue.task_done() - - logger.info("Finished sync with peers.") - return NetworkSyncResult( - completed_peer_ids=list(completed_peer_ids), - failed_peer_ids=list(failed_peer_ids), - timeout_peer_ids=list(remaining_peer_ids), - ) - - def _run_sync(self, peers): for peer in peers: sync_peer_thread = threading.Thread( target=self._sync_peer, args=(peer,), ) sync_peer_thread.start() - - def sync_peer(self, peer): - pass + # TODO: sleep for timeout and then call self._stop() def _sync_peer(self, peer): - try: - logger.debug("Trying to sync with peer: {}".format(peer.peer_id)) - self.sync_peer(peer) - self.queue.put(PeerSyncResult(completed_peer_id=peer.peer_id)) - logger.info("Finished sync with peer: {}".format(peer)) - except Exception: - logger.error("Failed Sync with peer: {}.".format( - peer), exc_info=True) - self.queue.put(PeerSyncResult(failed_peer_id=peer.peer_id)) + # with PeerConnection(peer) as peer_connection: + # self.sync_peer(peer_connection) + with PeerSyncTask( + self.squeak_controller, + peer, + self.stopped, + ).open_peer_sync_task() as peer_sync_task: + # if peer.uploading: + # peer_sync_task.upload(min_block, max_block) + # if peer.downloading: + # peer_sync_task.download(min_block, max_block) + self.sync_peer(peer_sync_task) -class TimelineNetworkSyncTask(NetworkSyncTask): +class DownloadSync(NetworkSync): + + def get_peers_to_sync(self): + return self.squeak_controller.get_uploading_peers() + + @abstractmethod + def sync_peer(self, peer_sync_task): + pass + + +class UploadSync(NetworkSync): + + def get_peers_to_sync(self): + return self.squeak_controller.get_uploading_peers() + + @abstractmethod + def sync_peer(self, peer_sync_task): + pass + + +class TimelineDownloadSync(DownloadSync): + def __init__( self, squeak_controller, @@ -100,14 +199,12 @@ class TimelineNetworkSyncTask(NetworkSyncTask): self.min_block = min_block self.max_block = max_block - def sync_peer(self, peer): - network_sync = NetworkSync( - self.squeak_controller, - ) - network_sync.sync_timeline(peer, self.min_block, self.max_block) + def sync_peer(self, peer_sync_task): + peer_sync_task.download(self.min_block, self.max_block) -class SingleSqueakNetworkSyncTask(NetworkSyncTask): +class SingleSqueakDownloadSync(DownloadSync): + def __init__( self, squeak_controller, @@ -116,8 +213,35 @@ class SingleSqueakNetworkSyncTask(NetworkSyncTask): super().__init__(squeak_controller) self.squeak_hash = squeak_hash - def sync_peer(self, peer): - network_sync = NetworkSync( - self.squeak_controller, - ) - network_sync.sync_single_squeak(peer, self.squeak_hash) + def sync_peer(self, peer_sync_task): + peer_sync_task.download_single_squeak(self.squeak_hash) + + +class TimelineUploadSync(UploadSync): + + def __init__( + self, + squeak_controller, + min_block, + max_block, + ): + super().__init__(squeak_controller) + self.min_block = min_block + self.max_block = max_block + + def sync_peer(self, peer_sync_task): + peer_sync_task.upload(self.min_block, self.max_block) + + +class SingleSqueakUploadSync(UploadSync): + + def __init__( + self, + squeak_controller, + squeak_hash: bytes, + ): + super().__init__(squeak_controller) + self.squeak_hash = squeak_hash + + def sync_peer(self, peer_sync_task): + peer_sync_task.upload_single_squeak(self.squeak_hash) diff --git a/squeaknode/sync/squeak_peer_sync_worker.py b/squeaknode/sync/squeak_peer_sync_worker.py index 39a230f8..650879e8 100644 --- a/squeaknode/sync/squeak_peer_sync_worker.py +++ b/squeaknode/sync/squeak_peer_sync_worker.py @@ -1,13 +1,15 @@ import logging import threading +from squeaknode.sync.squeak_sync_status import SqueakSyncController + logger = logging.getLogger(__name__) class SqueakPeerSyncWorker: def __init__( self, - squeak_sync_controller, + squeak_sync_controller: SqueakSyncController, sync_interval_s, ): self.squeak_sync_controller = squeak_sync_controller @@ -15,8 +17,8 @@ class SqueakPeerSyncWorker: def sync_timeline(self): logger.info("Syncing timeline with peers...") - # self.squeak_sync_controller.sync_squeaks() - self.squeak_sync_controller.sync_timeline() + self.squeak_sync_controller.download_timeline() + self.squeak_sync_controller.upload_timeline() def start_running(self): if self.sync_interval_s: diff --git a/squeaknode/sync/squeak_sync_status.py b/squeaknode/sync/squeak_sync_status.py index 31f43182..75892865 100644 --- a/squeaknode/sync/squeak_sync_status.py +++ b/squeaknode/sync/squeak_sync_status.py @@ -1,7 +1,9 @@ import logging -from squeaknode.sync.network_task import SingleSqueakNetworkSyncTask -from squeaknode.sync.network_task import TimelineNetworkSyncTask +from squeaknode.sync.network_task import SingleSqueakDownloadSync +from squeaknode.sync.network_task import SingleSqueakUploadSync +from squeaknode.sync.network_task import TimelineDownloadSync +from squeaknode.sync.network_task import TimelineUploadSync logger = logging.getLogger(__name__) @@ -11,35 +13,60 @@ class SqueakSyncController: self.squeak_controller = squeak_controller self.sync_block_range = sync_block_range - def sync_timeline(self, block_range=None): + def download_timeline(self, block_range=None): block_range = block_range or self.sync_block_range try: block_height = self.squeak_controller.get_best_block_height() except Exception: logger.error( - "Failed to sync timeline because unable to get best block height.", exc_info=False + "Failed to download timeline because unable to get best block height.", exc_info=False ) return min_block = block_height - block_range max_block = block_height - dowload_timeline_task = TimelineNetworkSyncTask( + # dowload_timeline_task = DownloadTimelineSync( + # self.squeak_controller, + # min_block, + # max_block, + # ) + # dowload_timeline_task.sync() + TimelineDownloadSync( self.squeak_controller, min_block, max_block, - ) - network_sync_result = dowload_timeline_task.sync() - logger.info("Upload network_sync_result: {}".format( - network_sync_result)) - return network_sync_result + ).sync() - def sync_single_squeak(self, squeak_hash): - timeline_sync_task = SingleSqueakNetworkSyncTask( + def upload_timeline(self, block_range=None): + block_range = block_range or self.sync_block_range + try: + block_height = self.squeak_controller.get_best_block_height() + except Exception: + logger.error( + "Failed to upload timeline because unable to get best block height.", exc_info=False + ) + return + min_block = block_height - block_range + max_block = block_height + # dowload_timeline_task = DownloadTimelineSync( + # self.squeak_controller, + # min_block, + # max_block, + # ) + # dowload_timeline_task.sync() + TimelineUploadSync( + self.squeak_controller, + min_block, + max_block, + ).sync() + + def download_single_squeak(self, squeak_hash): + SingleSqueakDownloadSync( self.squeak_controller, squeak_hash, - ) - network_sync_result = timeline_sync_task.sync() - logger.info( - "Download single squeak network_sync_result: {}".format( - network_sync_result) - ) - return network_sync_result + ).sync() + + def upload_single_squeak(self, squeak_hash): + SingleSqueakUploadSync( + self.squeak_controller, + squeak_hash, + ).sync()