You can not select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
761 lines
29 KiB
761 lines
29 KiB
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,
|
|
'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()
|
|
# 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, interface=None):
|
|
# If you want to queue a request on any interface it must go
|
|
# through this function so message ids are properly tracked
|
|
if interface is None:
|
|
interface = self.interface
|
|
message_id = self.message_id
|
|
self.message_id += 1
|
|
if self.debug:
|
|
self.print_error(interface.host, "-->", method, params, message_id)
|
|
interface.queue_request(method, params, message_id)
|
|
return message_id
|
|
|
|
def send_subscriptions(self):
|
|
self.print_error('sending subscriptions to', self.interface.server, len(self.unanswered_requests), len(self.subscribed_addresses))
|
|
# Resend unanswered requests
|
|
requests = self.unanswered_requests.values()
|
|
self.unanswered_requests = {}
|
|
for request in requests:
|
|
message_id = self.queue_request(request[0], request[1])
|
|
self.unanswered_requests[message_id] = request
|
|
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.interface = None
|
|
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.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:
|
|
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):
|
|
responses = interface.get_responses()
|
|
|
|
for request, response in responses:
|
|
callback = None
|
|
if request:
|
|
method, params, message_id = request
|
|
# client requests go through self.send() with a
|
|
# callback, are only sent to the current interface,
|
|
# and are placed in the unanswered_requests dictionary
|
|
client_req = self.unanswered_requests.pop(message_id, None)
|
|
if client_req:
|
|
assert interface == self.interface
|
|
callback = client_req[2]
|
|
# Copy the request method and params to the response
|
|
response['method'] = method
|
|
response['params'] = params
|
|
# Only once we've received a response to an addr subscription
|
|
# add it to the list; avoids double-sends on reconnection
|
|
if method == 'blockchain.address.subscribe':
|
|
self.subscribed_addresses.add(params[0])
|
|
else:
|
|
if not response: # Closed remotely / misbehaving
|
|
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]
|
|
|
|
# Response is now in canonical form
|
|
self.process_response(interface, response, callback)
|
|
|
|
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):
|
|
# Requests needs connectivity. If we don't have an interface,
|
|
# we cannot process them.
|
|
if not self.interface:
|
|
return
|
|
|
|
with self.lock:
|
|
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)
|
|
|
|
for method, params in messages:
|
|
message_id = self.queue_request(method, params)
|
|
self.unanswered_requests[message_id] = method, params, callback
|
|
|
|
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)
|
|
self.queue_request('blockchain.headers.subscribe', [], interface)
|
|
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():
|
|
params = [ELECTRUM_VERSION, PROTOCOL_VERSION]
|
|
self.queue_request('server.version', params, interface)
|
|
|
|
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)
|
|
self.queue_request('blockchain.block.get_chunk', [idx], interface)
|
|
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)
|
|
self.queue_request('blockchain.block.get_header', [height], interface)
|
|
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')
|
|
|