lbry-sdk/lbrynet/dht/hashannouncer.py

80 lines
3.1 KiB
Python
Raw Normal View History

2015-08-20 17:27:15 +02:00
import binascii
2016-11-03 20:42:45 +01:00
import logging
from twisted.internet import defer, task
from lbrynet.core import utils
2018-03-27 22:07:55 +02:00
from lbrynet import conf
2016-11-03 20:42:45 +01:00
log = logging.getLogger(__name__)
2015-08-20 17:27:15 +02:00
class DHTHashAnnouncer(object):
2018-03-27 22:07:55 +02:00
def __init__(self, dht_node, storage, concurrent_announcers=None):
2015-08-20 17:27:15 +02:00
self.dht_node = dht_node
self.storage = storage
self.clock = dht_node.clock
self.peer_port = dht_node.peerPort
self.hash_queue = []
2018-03-27 22:07:55 +02:00
self.concurrent_announcers = concurrent_announcers or conf.settings['concurrent_announcers']
self._manage_lc = task.LoopingCall(self.manage)
self._manage_lc.clock = self.clock
2015-08-20 17:27:15 +02:00
def start(self):
self._manage_lc.start(30)
2015-08-20 17:27:15 +02:00
def stop(self):
if self._manage_lc.running:
self._manage_lc.stop()
2015-08-20 17:27:15 +02:00
@defer.inlineCallbacks
def do_store(self, blob_hash):
storing_node_ids = yield self.dht_node.announceHaveBlob(binascii.unhexlify(blob_hash))
now = self.clock.seconds()
if storing_node_ids:
result = (now, storing_node_ids)
yield self.storage.update_last_announced_blob(blob_hash, now)
log.debug("Stored %s to %i peers", blob_hash[:16], len(storing_node_ids))
else:
result = (None, [])
self.hash_queue.remove(blob_hash)
defer.returnValue(result)
2018-03-29 21:06:08 +02:00
def hash_queue_size(self):
return len(self.hash_queue)
def _show_announce_progress(self, size, start):
queue_size = len(self.hash_queue)
average_blobs_per_second = float(size - queue_size) / (self.clock.seconds() - start)
log.info("Announced %i/%i blobs, %f blobs per second", size - queue_size, size, average_blobs_per_second)
2015-08-20 17:27:15 +02:00
@defer.inlineCallbacks
def immediate_announce(self, blob_hashes):
self.hash_queue.extend(b for b in blob_hashes if b not in self.hash_queue)
log.info("Announcing %i blobs", len(self.hash_queue))
start = self.clock.seconds()
progress_lc = task.LoopingCall(self._show_announce_progress, len(self.hash_queue), start)
2018-05-24 18:56:39 +02:00
progress_lc.clock = self.clock
progress_lc.start(60, now=False)
s = defer.DeferredSemaphore(self.concurrent_announcers)
results = yield utils.DeferredDict({blob_hash: s.run(self.do_store, blob_hash) for blob_hash in blob_hashes})
now = self.clock.seconds()
progress_lc.stop()
announced_to = [blob_hash for blob_hash in results if results[blob_hash][0]]
if len(announced_to) != len(results):
log.debug("Failed to announce %i blobs", len(results) - len(announced_to))
if announced_to:
log.info('Took %s seconds to announce %i of %i attempted hashes (%f hashes per second)',
now - start, len(announced_to), len(blob_hashes),
int(float(len(blob_hashes)) / float(now - start)))
defer.returnValue(results)
2015-08-20 17:27:15 +02:00
@defer.inlineCallbacks
def manage(self):
need_reannouncement = yield self.storage.get_blobs_to_announce()
if need_reannouncement:
yield self.immediate_announce(need_reannouncement)
else:
log.debug("Nothing to announce")