mirror of
https://github.com/LBRYFoundation/LBRY-Vault.git
synced 2025-08-23 17:47:31 +00:00
da5c18b2f9
On startup we make several connections simultaneously. Socket maintenance code checks if we're not connected, and if not switches to a connected interface if auto_connect. Unfortunately this meant that we frequently didn't reconnect to the prior good server on startup, because some other connection would happen first and this code would decide to switch to it. Instead, only switch if a connection attempt is not in progress. If that times out at the O/S level the switch will happen.
799 lines
30 KiB
Python
799 lines
30 KiB
Python
import time
|
|
import Queue
|
|
import os
|
|
import sys
|
|
import random
|
|
import select
|
|
import traceback
|
|
from collections import defaultdict, deque
|
|
from threading import Lock
|
|
|
|
import socks
|
|
import socket
|
|
import json
|
|
|
|
import util
|
|
from bitcoin import *
|
|
from interface import Connection, Interface
|
|
from blockchain import Blockchain
|
|
from version import ELECTRUM_VERSION, PROTOCOL_VERSION
|
|
|
|
DEFAULT_PORTS = {'t':'50001', 's':'50002', 'h':'8081', 'g':'8082'}
|
|
|
|
DEFAULT_SERVERS = {
|
|
'electrum.be':{'t':'50001', 's':'50002'},
|
|
'electrum.drollette.com':{'t':'50001', 's':'50002'},
|
|
'erbium1.sytes.net':{'t':'50001', 's':'50002'},
|
|
'ecdsa.net':{'t':'50001', 's':'110'},
|
|
'electrum0.electricnewyear.net':{'t':'50001', 's':'50002'},
|
|
'kirsche.emzy.de':DEFAULT_PORTS,
|
|
'jwu42.hopto.org':{'t':'50001', 's':'50002'},
|
|
'VPS.hsmiths.com':{'t':'50001', 's':'50002'},
|
|
'ELECTRUM.jdubya.info':{'t':'50001', 's':'50002'},
|
|
'electrum.no-ip.org':{'t':'50001', 's':'50002', 'g':'443'},
|
|
'electrum.thwg.org':DEFAULT_PORTS,
|
|
'us.electrum.be':{'t':'50001', 's':'50002'},
|
|
}
|
|
|
|
NODES_RETRY_INTERVAL = 60
|
|
SERVER_RETRY_INTERVAL = 10
|
|
|
|
|
|
def parse_servers(result):
|
|
""" parse servers list into dict format"""
|
|
from version import PROTOCOL_VERSION
|
|
servers = {}
|
|
for item in result:
|
|
host = item[1]
|
|
out = {}
|
|
version = None
|
|
pruning_level = '-'
|
|
if len(item) > 2:
|
|
for v in item[2]:
|
|
if re.match("[stgh]\d*", v):
|
|
protocol, port = v[0], v[1:]
|
|
if port == '': port = DEFAULT_PORTS[protocol]
|
|
out[protocol] = port
|
|
elif re.match("v(.?)+", v):
|
|
version = v[1:]
|
|
elif re.match("p\d*", v):
|
|
pruning_level = v[1:]
|
|
if pruning_level == '': pruning_level = '0'
|
|
try:
|
|
is_recent = cmp(util.normalize_version(version), util.normalize_version(PROTOCOL_VERSION)) >= 0
|
|
except Exception:
|
|
is_recent = False
|
|
|
|
if out and is_recent:
|
|
out['pruning'] = pruning_level
|
|
servers[host] = out
|
|
|
|
return servers
|
|
|
|
def filter_protocol(hostmap = DEFAULT_SERVERS, protocol = 's'):
|
|
'''Filters the hostmap for those implementing protocol.
|
|
The result is a list in serialized form.'''
|
|
eligible = []
|
|
for host, portmap in hostmap.items():
|
|
port = portmap.get(protocol)
|
|
if port:
|
|
eligible.append(serialize_server(host, port, protocol))
|
|
return eligible
|
|
|
|
def pick_random_server(hostmap = DEFAULT_SERVERS, protocol = 's', exclude_set = set()):
|
|
eligible = list(set(filter_protocol(hostmap, protocol)) - exclude_set)
|
|
return random.choice(eligible) if eligible else None
|
|
|
|
from simple_config import SimpleConfig
|
|
|
|
proxy_modes = ['socks4', 'socks5', 'http']
|
|
|
|
def serialize_proxy(p):
|
|
if type(p) != dict:
|
|
return None
|
|
return ':'.join([p.get('mode'),p.get('host'), p.get('port')])
|
|
|
|
def deserialize_proxy(s):
|
|
if type(s) not in [str, unicode]:
|
|
return None
|
|
if s.lower() == 'none':
|
|
return None
|
|
proxy = { "mode":"socks5", "host":"localhost" }
|
|
args = s.split(':')
|
|
n = 0
|
|
if proxy_modes.count(args[n]) == 1:
|
|
proxy["mode"] = args[n]
|
|
n += 1
|
|
if len(args) > n:
|
|
proxy["host"] = args[n]
|
|
n += 1
|
|
if len(args) > n:
|
|
proxy["port"] = args[n]
|
|
else:
|
|
proxy["port"] = "8080" if proxy["mode"] == "http" else "1080"
|
|
return proxy
|
|
|
|
def deserialize_server(server_str):
|
|
host, port, protocol = str(server_str).split(':')
|
|
assert protocol in 'st'
|
|
int(port) # Throw if cannot be converted to int
|
|
return host, port, protocol
|
|
|
|
def serialize_server(host, port, protocol):
|
|
return str(':'.join([host, port, protocol]))
|
|
|
|
class Network(util.DaemonThread):
|
|
"""The Network class manages a set of connections to remote electrum
|
|
servers, each connected socket is handled by an Interface() object.
|
|
Connections are initiated by a Connection() thread which stops once
|
|
the connection succeeds or fails.
|
|
|
|
Our external API:
|
|
|
|
- Member functions get_header(), get_interfaces(), get_local_height(),
|
|
get_parameters(), get_server_height(), get_status_value(),
|
|
is_connected(), new_blockchain_height(), set_parameters(),
|
|
stop()
|
|
"""
|
|
|
|
def __init__(self, config=None, plugins=None):
|
|
if config is None:
|
|
config = {} # Do not use mutables as default values!
|
|
util.DaemonThread.__init__(self)
|
|
self.config = SimpleConfig(config) if type(config) == type({}) else config
|
|
self.num_server = 8 if not self.config.get('oneserver') else 0
|
|
self.blockchain = Blockchain(self.config, self)
|
|
# A deque of interface header requests, processed left-to-right
|
|
self.bc_requests = deque()
|
|
# Server for addresses and transactions
|
|
self.default_server = self.config.get('server')
|
|
# Sanitize default server
|
|
try:
|
|
deserialize_server(self.default_server)
|
|
except:
|
|
self.default_server = None
|
|
if not self.default_server:
|
|
self.default_server = pick_random_server()
|
|
|
|
self.lock = Lock()
|
|
self.pending_sends = []
|
|
self.message_id = 0
|
|
self.debug = False
|
|
self.irc_servers = {} # returned by interface (list from irc)
|
|
self.recent_servers = self.read_recent_servers()
|
|
|
|
self.banner = ''
|
|
self.fee = None
|
|
self.heights = {}
|
|
self.merkle_roots = {}
|
|
self.utxo_roots = {}
|
|
self.subscriptions = defaultdict(list)
|
|
self.callbacks = defaultdict(list)
|
|
|
|
dir_path = os.path.join( self.config.path, 'certs')
|
|
if not os.path.exists(dir_path):
|
|
os.mkdir(dir_path)
|
|
|
|
# subscriptions and requests
|
|
self.subscribed_addresses = set()
|
|
# cached address status
|
|
self.addr_responses = {}
|
|
# Requests from client we've not seen a response to
|
|
self.unanswered_requests = {}
|
|
# retry times
|
|
self.server_retry_time = time.time()
|
|
self.nodes_retry_time = time.time()
|
|
# kick off the network. interface is the main server we are currently
|
|
# communicating with. interfaces is the set of servers we are connecting
|
|
# to or have an ongoing connection with
|
|
self.interface = None
|
|
self.interfaces = {}
|
|
self.auto_connect = self.config.get('auto_connect', False)
|
|
self.connecting = {}
|
|
self.socket_queue = Queue.Queue()
|
|
self.start_network(deserialize_server(self.default_server)[2],
|
|
deserialize_proxy(self.config.get('proxy')))
|
|
self.plugins = plugins
|
|
if self.plugins:
|
|
self.plugins.set_network(self)
|
|
|
|
def register_callback(self, event, callback):
|
|
with self.lock:
|
|
self.callbacks[event].append(callback)
|
|
|
|
def trigger_callback(self, event, params=()):
|
|
with self.lock:
|
|
callbacks = self.callbacks[event][:]
|
|
[callback(*params) for callback in callbacks]
|
|
|
|
def read_recent_servers(self):
|
|
if not self.config.path:
|
|
return []
|
|
path = os.path.join(self.config.path, "recent_servers")
|
|
try:
|
|
with open(path, "r") as f:
|
|
data = f.read()
|
|
return json.loads(data)
|
|
except:
|
|
return []
|
|
|
|
def save_recent_servers(self):
|
|
if not self.config.path:
|
|
return
|
|
path = os.path.join(self.config.path, "recent_servers")
|
|
s = json.dumps(self.recent_servers, indent=4, sort_keys=True)
|
|
try:
|
|
with open(path, "w") as f:
|
|
f.write(s)
|
|
except:
|
|
pass
|
|
|
|
def get_server_height(self):
|
|
return self.heights.get(self.default_server, 0)
|
|
|
|
def server_is_lagging(self):
|
|
sh = self.get_server_height()
|
|
if not sh:
|
|
self.print_error('no height for main interface')
|
|
return False
|
|
lh = self.get_local_height()
|
|
result = (lh - sh) > 1
|
|
if result:
|
|
self.print_error('%s is lagging (%d vs %d)' % (self.default_server, sh, lh))
|
|
return result
|
|
|
|
def set_status(self, status):
|
|
self.connection_status = status
|
|
self.notify('status')
|
|
|
|
def is_connected(self):
|
|
return self.interface is not None
|
|
|
|
def is_connecting(self):
|
|
return self.connection_status == 'connecting'
|
|
|
|
def is_up_to_date(self):
|
|
return self.unanswered_requests == {}
|
|
|
|
def queue_request(self, method, params):
|
|
self.interface.queue_request({'method': method, 'params': params})
|
|
|
|
def send_subscriptions(self):
|
|
# clear cache
|
|
self.cached_responses = {}
|
|
self.print_error('sending subscriptions to', self.interface.server, len(self.unanswered_requests), len(self.subscribed_addresses))
|
|
for r in self.unanswered_requests.values():
|
|
self.interface.queue_request(r[0])
|
|
for addr in self.subscribed_addresses:
|
|
self.queue_request('blockchain.address.subscribe', [addr])
|
|
self.queue_request('server.banner', [])
|
|
self.queue_request('server.peers.subscribe', [])
|
|
self.queue_request('blockchain.estimatefee', [2])
|
|
|
|
def get_status_value(self, key):
|
|
if key == 'status':
|
|
value = self.connection_status
|
|
elif key == 'banner':
|
|
value = self.banner
|
|
elif key == 'fee':
|
|
value = self.fee
|
|
elif key == 'updated':
|
|
value = (self.get_local_height(), self.get_server_height())
|
|
elif key == 'servers':
|
|
value = self.get_servers()
|
|
elif key == 'interfaces':
|
|
value = self.get_interfaces()
|
|
return value
|
|
|
|
def notify(self, key):
|
|
value = self.get_status_value(key)
|
|
if key in ['status', 'updated']:
|
|
self.trigger_callback(key)
|
|
else:
|
|
self.trigger_callback(key, (value,))
|
|
|
|
def get_parameters(self):
|
|
host, port, protocol = deserialize_server(self.default_server)
|
|
return host, port, protocol, self.proxy, self.auto_connect
|
|
|
|
def get_interfaces(self):
|
|
'''The interfaces that are in connected state'''
|
|
return self.interfaces.keys()
|
|
|
|
def get_servers(self):
|
|
if self.irc_servers:
|
|
out = self.irc_servers
|
|
else:
|
|
out = DEFAULT_SERVERS
|
|
for s in self.recent_servers:
|
|
try:
|
|
host, port, protocol = deserialize_server(s)
|
|
except:
|
|
continue
|
|
if host not in out:
|
|
out[host] = { protocol:port }
|
|
return out
|
|
|
|
def start_interface(self, server):
|
|
if (not server in self.interfaces and not server in self.connecting):
|
|
if server == self.default_server:
|
|
self.print_error("connecting to %s as new interface" % server)
|
|
self.set_status('connecting')
|
|
c = Connection(server, self.socket_queue, self.config.path)
|
|
self.connecting[server] = c
|
|
|
|
def start_random_interface(self):
|
|
exclude_set = self.disconnected_servers.union(set(self.interfaces))
|
|
server = pick_random_server(self.get_servers(), self.protocol, exclude_set)
|
|
if server:
|
|
self.start_interface(server)
|
|
|
|
def start_interfaces(self):
|
|
self.start_interface(self.default_server)
|
|
for i in range(self.num_server - 1):
|
|
self.start_random_interface()
|
|
|
|
def set_proxy(self, proxy):
|
|
self.proxy = proxy
|
|
if proxy:
|
|
proxy_mode = proxy_modes.index(proxy["mode"]) + 1
|
|
socks.setdefaultproxy(proxy_mode, proxy["host"], int(proxy["port"]))
|
|
socket.socket = socks.socksocket
|
|
# prevent dns leaks, see http://stackoverflow.com/questions/13184205/dns-over-proxy
|
|
socket.getaddrinfo = lambda *args: [(socket.AF_INET, socket.SOCK_STREAM, 6, '', (args[0], args[1]))]
|
|
else:
|
|
socket.socket = socket._socketobject
|
|
socket.getaddrinfo = socket._socket.getaddrinfo
|
|
|
|
def start_network(self, protocol, proxy):
|
|
assert not self.interface and not self.interfaces
|
|
assert not self.connecting and self.socket_queue.empty()
|
|
self.print_error('starting network')
|
|
self.disconnected_servers = set([])
|
|
self.protocol = protocol
|
|
self.set_proxy(proxy)
|
|
self.start_interfaces()
|
|
|
|
def stop_network(self):
|
|
self.print_error("stopping network")
|
|
for interface in self.interfaces.values():
|
|
self.close_interface(interface)
|
|
assert self.interface is None
|
|
assert not self.interfaces
|
|
self.connecting = {}
|
|
# Get a new queue - no old pending connections thanks!
|
|
self.socket_queue = Queue.Queue()
|
|
|
|
def set_parameters(self, host, port, protocol, proxy, auto_connect):
|
|
proxy_str = serialize_proxy(proxy)
|
|
server = serialize_server(host, port, protocol)
|
|
self.config.set_key('auto_connect', auto_connect, False)
|
|
self.config.set_key("proxy", proxy_str, False)
|
|
self.config.set_key("server", server, True)
|
|
# abort if changes were not allowed by config
|
|
if self.config.get('server') != server or self.config.get('proxy') != proxy_str:
|
|
return
|
|
|
|
self.auto_connect = auto_connect
|
|
if self.proxy != proxy or self.protocol != protocol:
|
|
# Restart the network defaulting to the given server
|
|
self.stop_network()
|
|
self.default_server = server
|
|
self.start_network(protocol, proxy)
|
|
elif self.default_server != server:
|
|
self.switch_to_interface(server)
|
|
else:
|
|
self.switch_lagging_interface()
|
|
|
|
def switch_to_random_interface(self):
|
|
'''Switch to a random connected server other than the current one'''
|
|
servers = self.get_interfaces() # Those in connected state
|
|
if self.default_server in servers:
|
|
servers.remove(self.default_server)
|
|
if servers:
|
|
self.switch_to_interface(random.choice(servers))
|
|
|
|
def switch_lagging_interface(self, suggestion = None):
|
|
'''If auto_connect and lagging, switch interface'''
|
|
if self.server_is_lagging() and self.auto_connect:
|
|
if suggestion and self.protocol == deserialize_server(suggestion)[2]:
|
|
self.switch_to_interface(suggestion)
|
|
else:
|
|
self.switch_to_random_interface()
|
|
|
|
def switch_to_interface(self, server):
|
|
'''Switch to server as our interface. If no connection exists nor
|
|
being opened, start a thread to connect. The actual switch will
|
|
happen on receipt of the connection notification. Do nothing
|
|
if server already is our interface.'''
|
|
self.default_server = server
|
|
if server not in self.interfaces:
|
|
self.start_interface(server)
|
|
return
|
|
i = self.interfaces[server]
|
|
if self.interface != i:
|
|
self.print_error("switching to", server)
|
|
# stop any current interface in order to terminate subscriptions
|
|
self.close_interface(self.interface)
|
|
self.interface = i
|
|
self.addr_responses = {}
|
|
self.send_subscriptions()
|
|
self.set_status('connected')
|
|
self.notify('updated')
|
|
|
|
def close_interface(self, interface):
|
|
if interface:
|
|
self.interfaces.pop(interface.server)
|
|
if interface.server == self.default_server:
|
|
self.interface = None
|
|
interface.close()
|
|
|
|
def add_recent_server(self, server):
|
|
# list is ordered
|
|
if server in self.recent_servers:
|
|
self.recent_servers.remove(server)
|
|
self.recent_servers.insert(0, server)
|
|
self.recent_servers = self.recent_servers[0:20]
|
|
self.save_recent_servers()
|
|
|
|
def new_blockchain_height(self, blockchain_height, i):
|
|
self.switch_lagging_interface(i.server)
|
|
self.notify('updated')
|
|
|
|
def process_response(self, interface, response, callback):
|
|
if self.debug:
|
|
self.print_error("<--", response)
|
|
error = response.get('error')
|
|
result = response.get('result')
|
|
method = response.get('method')
|
|
|
|
# We handle some responses; return the rest to the client.
|
|
if method == 'server.version':
|
|
interface.server_version = result
|
|
elif method == 'blockchain.headers.subscribe':
|
|
if error is None:
|
|
self.on_header(interface, result)
|
|
elif method == 'server.peers.subscribe':
|
|
if error is None:
|
|
self.irc_servers = parse_servers(result)
|
|
self.notify('servers')
|
|
elif method == 'server.banner':
|
|
if error is None:
|
|
self.banner = result
|
|
self.notify('banner')
|
|
elif method == 'blockchain.estimatefee':
|
|
if error is None:
|
|
self.fee = int(result * COIN)
|
|
self.print_error("recommended fee", self.fee)
|
|
self.notify('fee')
|
|
elif method == 'blockchain.block.get_chunk':
|
|
self.on_get_chunk(interface, response)
|
|
elif method == 'blockchain.block.get_header':
|
|
self.on_get_header(interface, response)
|
|
else:
|
|
# Cache address subscription results
|
|
if method == 'blockchain.address.subscribe' and error is None:
|
|
addr = response['params'][0]
|
|
self.addr_responses[addr] = response
|
|
if callback is None:
|
|
params = response['params']
|
|
with self.lock:
|
|
for k,v in self.subscriptions.items():
|
|
if (method, params) in v:
|
|
callback = k
|
|
break
|
|
if callback is None:
|
|
self.print_error("received unexpected notification",
|
|
method, params)
|
|
else:
|
|
callback(response)
|
|
|
|
def process_responses(self, interface):
|
|
notifications, responses = interface.get_responses()
|
|
|
|
for request, response in responses:
|
|
# Client ID was given by the daemon or proxy
|
|
client_id = request.get('id')
|
|
if client_id is not None:
|
|
if interface != self.interface:
|
|
continue
|
|
_req, callback = self.unanswered_requests.pop(client_id)
|
|
else:
|
|
callback = None
|
|
# Copy the request method and params to the response
|
|
response['method'] = request.get('method')
|
|
response['params'] = request.get('params')
|
|
response['id'] = client_id
|
|
self.process_response(interface, response, callback)
|
|
|
|
for response in notifications:
|
|
if not response: # Closed remotely
|
|
self.connection_down(interface.server)
|
|
break
|
|
# Rewrite response shape to match subscription request response
|
|
method = response.get('method')
|
|
params = response.get('params')
|
|
if method == 'blockchain.headers.subscribe':
|
|
response['result'] = params[0]
|
|
response['params'] = []
|
|
elif method == 'blockchain.address.subscribe':
|
|
response['params'] = [params[0]] # addr
|
|
response['result'] = params[1]
|
|
self.process_response(interface, response, None)
|
|
|
|
def send(self, messages, callback):
|
|
'''Messages is a list of (method, value) tuples'''
|
|
with self.lock:
|
|
self.pending_sends.append((messages, callback))
|
|
|
|
def process_pending_sends(self):
|
|
sends = self.pending_sends
|
|
self.pending_sends = []
|
|
|
|
for messages, callback in sends:
|
|
subs = filter(lambda (m,v): m.endswith('.subscribe'), messages)
|
|
with self.lock:
|
|
for sub in subs:
|
|
if sub not in self.subscriptions[callback]:
|
|
self.subscriptions[callback].append(sub)
|
|
_id = self.message_id
|
|
self.message_id += len(messages)
|
|
|
|
unsent = []
|
|
for message in messages:
|
|
method, params = message
|
|
request = {'id': _id, 'method': method, 'params': params}
|
|
if not self.process_request(request, callback):
|
|
unsent.append(message)
|
|
_id += 1
|
|
|
|
if unsent:
|
|
with self.lock:
|
|
self.pending_sends.append((unsent, callback))
|
|
|
|
# FIXME: inline this function
|
|
def process_request(self, request, callback):
|
|
'''Returns true if the request was processed.'''
|
|
method = request['method']
|
|
params = request['params']
|
|
_id = request['id']
|
|
|
|
if method.startswith('network.'):
|
|
out = {'id':_id}
|
|
try:
|
|
f = getattr(self, method[8:])
|
|
out['result'] = f(*params)
|
|
except AttributeError:
|
|
out['error'] = "unknown method"
|
|
except BaseException as e:
|
|
out['error'] = str(e)
|
|
traceback.print_exc(file=sys.stdout)
|
|
self.print_error("network error", str(e))
|
|
callback(out)
|
|
return True
|
|
|
|
if method == 'blockchain.address.subscribe':
|
|
addr = params[0]
|
|
self.subscribed_addresses.add(addr)
|
|
if addr in self.addr_responses:
|
|
callback(self.addr_responses[addr])
|
|
return True
|
|
|
|
# This request needs connectivity. If we don't have an
|
|
# interface, we cannot process it.
|
|
if not self.interface:
|
|
return False
|
|
|
|
if self.debug:
|
|
self.print_error("-->", request)
|
|
self.unanswered_requests[_id] = request, callback
|
|
self.interface.queue_request(request)
|
|
return True
|
|
|
|
def connection_down(self, server):
|
|
'''A connection to server either went down, or was never made.
|
|
We distinguish by whether it is in self.interfaces.'''
|
|
self.disconnected_servers.add(server)
|
|
if server == self.default_server:
|
|
self.set_status('disconnected')
|
|
if server in self.interfaces:
|
|
self.close_interface(self.interfaces[server])
|
|
self.heights.pop(server, None)
|
|
self.notify('interfaces')
|
|
|
|
def new_interface(self, server, socket):
|
|
self.add_recent_server(server)
|
|
self.interfaces[server] = interface = Interface(server, socket)
|
|
interface.queue_request({'method': 'blockchain.headers.subscribe',
|
|
'params': []})
|
|
if server == self.default_server:
|
|
self.switch_to_interface(server)
|
|
self.notify('interfaces')
|
|
|
|
def maintain_sockets(self):
|
|
'''Socket maintenance.'''
|
|
# Responses to connection attempts?
|
|
while not self.socket_queue.empty():
|
|
server, socket = self.socket_queue.get()
|
|
self.connecting.pop(server)
|
|
if socket:
|
|
self.new_interface(server, socket)
|
|
else:
|
|
self.connection_down(server)
|
|
|
|
# Send pings and shut down stale interfaces
|
|
for interface in self.interfaces.values():
|
|
if interface.has_timed_out():
|
|
self.connection_down(interface.server)
|
|
elif interface.ping_required():
|
|
version_req = {'method': 'server.version',
|
|
'params': [ELECTRUM_VERSION, PROTOCOL_VERSION]}
|
|
interface.queue_request(version_req)
|
|
|
|
now = time.time()
|
|
# nodes
|
|
if len(self.interfaces) + len(self.connecting) < self.num_server:
|
|
self.start_random_interface()
|
|
if now - self.nodes_retry_time > NODES_RETRY_INTERVAL:
|
|
self.print_error('network: retrying connections')
|
|
self.disconnected_servers = set([])
|
|
self.nodes_retry_time = now
|
|
|
|
# main interface
|
|
if not self.is_connected():
|
|
if self.auto_connect:
|
|
if not self.is_connecting():
|
|
self.switch_to_random_interface()
|
|
else:
|
|
if self.default_server in self.disconnected_servers:
|
|
if now - self.server_retry_time > SERVER_RETRY_INTERVAL:
|
|
self.disconnected_servers.remove(self.default_server)
|
|
self.server_retry_time = now
|
|
else:
|
|
self.switch_to_interface(self.default_server)
|
|
|
|
def request_chunk(self, interface, data, idx):
|
|
interface.print_error("requesting chunk %d" % idx)
|
|
interface.queue_request({'method':'blockchain.block.get_chunk',
|
|
'params':[idx]})
|
|
data['chunk_idx'] = idx
|
|
data['req_time'] = time.time()
|
|
|
|
def on_get_chunk(self, interface, response):
|
|
'''Handle receiving a chunk of block headers'''
|
|
if self.bc_requests:
|
|
req_if, data = self.bc_requests[0]
|
|
req_idx = data.get('chunk_idx')
|
|
# Ignore unsolicited chunks
|
|
if req_if == interface and req_idx == response['params'][0]:
|
|
idx = self.blockchain.connect_chunk(req_idx, response['result'])
|
|
# If not finished, get the next chunk
|
|
if idx < 0 or self.get_local_height() >= data['if_height']:
|
|
self.bc_requests.popleft()
|
|
self.notify('updated')
|
|
else:
|
|
self.request_chunk(interface, data, idx)
|
|
|
|
def request_header(self, interface, data, height):
|
|
interface.print_error("requesting header %d" % height)
|
|
interface.queue_request({'method':'blockchain.block.get_header',
|
|
'params':[height]})
|
|
data['header_height'] = height
|
|
data['req_time'] = time.time()
|
|
if not 'chain' in data:
|
|
data['chain'] = []
|
|
|
|
def on_get_header(self, interface, response):
|
|
'''Handle receiving a single block header'''
|
|
if self.bc_requests:
|
|
req_if, data = self.bc_requests[0]
|
|
req_height = data.get('header_height', -1)
|
|
# Ignore unsolicited headers
|
|
if req_if == interface and req_height == response['params'][0]:
|
|
next_height = self.blockchain.connect_header(data['chain'], response['result'])
|
|
# If not finished, get the next header
|
|
if next_height in [True, False]:
|
|
self.bc_requests.popleft()
|
|
if next_height:
|
|
self.notify('updated')
|
|
else:
|
|
interface.print_error("header didn't connect, dismissing interface")
|
|
interface.stop()
|
|
else:
|
|
self.request_header(interface, data, next_height)
|
|
|
|
def bc_request_headers(self, interface, data):
|
|
'''Send a request for the next header, or a chunk of them,
|
|
if necessary.
|
|
'''
|
|
local_height, if_height = self.get_local_height(), data['if_height']
|
|
if if_height <= local_height:
|
|
return False
|
|
elif if_height > local_height + 50:
|
|
self.request_chunk(interface, data, (local_height + 1) / 2016)
|
|
else:
|
|
self.request_header(interface, data, if_height)
|
|
return True
|
|
|
|
def handle_bc_requests(self):
|
|
'''Work through each interface that has notified us of a new header.
|
|
Send it requests if it is ahead of our blockchain object.
|
|
'''
|
|
while self.bc_requests:
|
|
interface, data = self.bc_requests.popleft()
|
|
# If the connection was lost move on
|
|
if not interface in self.interfaces.values():
|
|
continue
|
|
|
|
req_time = data.get('req_time')
|
|
if not req_time:
|
|
# No requests sent yet. This interface has a new height.
|
|
# Request headers if it is ahead of our blockchain
|
|
if not self.bc_request_headers(interface, data):
|
|
continue
|
|
elif time.time() - req_time > 10:
|
|
interface.print_error("blockchain request timed out")
|
|
self.connection_down(interface.server)
|
|
continue
|
|
# Put updated request state back at head of deque
|
|
self.bc_requests.appendleft((interface, data))
|
|
break
|
|
|
|
def wait_on_sockets(self):
|
|
# Python docs say Windows doesn't like empty selects.
|
|
# Sleep to prevent busy looping
|
|
if not self.interfaces:
|
|
time.sleep(0.1)
|
|
return
|
|
rin = [i for i in self.interfaces.values()]
|
|
win = [i for i in self.interfaces.values() if i.unsent_requests]
|
|
rout, wout, xout = select.select(rin, win, [], 0.1)
|
|
assert not xout
|
|
for interface in wout:
|
|
interface.send_requests()
|
|
for interface in rout:
|
|
self.process_responses(interface)
|
|
|
|
def run(self):
|
|
self.blockchain.init()
|
|
while self.is_running():
|
|
self.maintain_sockets()
|
|
self.wait_on_sockets()
|
|
self.handle_bc_requests()
|
|
self.run_jobs() # Synchronizer and Verifier
|
|
self.process_pending_sends()
|
|
|
|
self.stop_network()
|
|
if self.plugins:
|
|
self.plugins.set_network(None)
|
|
self.print_error("stopped")
|
|
|
|
def on_header(self, i, header):
|
|
height = header.get('block_height')
|
|
if not height:
|
|
return
|
|
self.heights[i.server] = height
|
|
self.merkle_roots[i.server] = header.get('merkle_root')
|
|
self.utxo_roots[i.server] = header.get('utxo_root')
|
|
|
|
# Queue this interface's height for asynchronous catch-up
|
|
self.bc_requests.append((i, {'if_height': height}))
|
|
|
|
if i == self.interface:
|
|
self.switch_lagging_interface()
|
|
self.notify('updated')
|
|
|
|
|
|
def get_header(self, tx_height):
|
|
return self.blockchain.read_header(tx_height)
|
|
|
|
def get_local_height(self):
|
|
return self.blockchain.height()
|
|
|
|
def synchronous_get(self, request, timeout=100000000):
|
|
queue = Queue.Queue()
|
|
self.send([request], queue.put)
|
|
r = queue.get(True, timeout)
|
|
if r.get('error'):
|
|
raise BaseException(r.get('error'))
|
|
return r.get('result')
|