X-Git-Url: https://git.novaco.in/?a=blobdiff_plain;f=p2pool%2Fmain.py;h=139c75dca82e1f6934b4b64c35e375b6287f5978;hb=2cb4d8381e179f71ea2075cdce948ea83cf0dc55;hp=a168f92ba5755c361f4eefe60df9470552ecc403;hpb=70d337b9024ff6564fcbebae114c95b91422aed3;p=p2pool.git diff --git a/p2pool/main.py b/p2pool/main.py index a168f92..139c75d 100644 --- a/p2pool/main.py +++ b/p2pool/main.py @@ -1,7 +1,5 @@ from __future__ import division -import ConfigParser -import StringIO import base64 import json import os @@ -20,365 +18,11 @@ from twisted.web import server from twisted.python import log from nattraverso import portmapper, ipdiscover -import bitcoin.p2p as bitcoin_p2p, bitcoin.getwork as bitcoin_getwork, bitcoin.data as bitcoin_data -from bitcoin import worker_interface, height_tracker -from util import expiring_dict, fixargparse, jsonrpc, variable, deferral, math, logging, pack -from . import p2p, networks, web -import p2pool, p2pool.data as p2pool_data - -@deferral.retry('Error getting work from bitcoind:', 3) -@defer.inlineCallbacks -def getwork(bitcoind): - try: - work = yield bitcoind.rpc_getmemorypool() - except jsonrpc.Error, e: - if e.code == -32601: # Method not found - print >>sys.stderr, 'Error: Bitcoin version too old! Upgrade to v0.5 or newer!' - raise deferral.RetrySilentlyException() - raise - packed_transactions = [x.decode('hex') for x in work['transactions']] - defer.returnValue(dict( - version=work['version'], - previous_block_hash=int(work['previousblockhash'], 16), - transactions=map(bitcoin_data.tx_type.unpack, packed_transactions), - merkle_link=bitcoin_data.calculate_merkle_link([None] + map(bitcoin_data.hash256, packed_transactions), 0), - subsidy=work['coinbasevalue'], - time=work['time'], - bits=bitcoin_data.FloatingIntegerType().unpack(work['bits'].decode('hex')[::-1]) if isinstance(work['bits'], (str, unicode)) else bitcoin_data.FloatingInteger(work['bits']), - coinbaseflags=work['coinbaseflags'].decode('hex') if 'coinbaseflags' in work else ''.join(x.decode('hex') for x in work['coinbaseaux'].itervalues()) if 'coinbaseaux' in work else '', - )) - -class WorkerBridge(worker_interface.WorkerBridge): - def __init__(self, my_pubkey_hash, net, donation_percentage, bitcoind_work, best_block_header, merged_urls, best_share_var, tracker, my_share_hashes, my_doa_share_hashes, worker_fee, p2p_node, submit_block, set_best_share, shared_share_hashes, block_height_var): - worker_interface.WorkerBridge.__init__(self) - self.recent_shares_ts_work = [] - - self.my_pubkey_hash = my_pubkey_hash - self.net = net - self.donation_percentage = donation_percentage - self.bitcoind_work = bitcoind_work - self.best_block_header = best_block_header - self.best_share_var = best_share_var - self.tracker = tracker - self.my_share_hashes = my_share_hashes - self.my_doa_share_hashes = my_doa_share_hashes - self.worker_fee = worker_fee - self.p2p_node = p2p_node - self.submit_block = submit_block - self.set_best_share = set_best_share - self.shared_share_hashes = shared_share_hashes - self.block_height_var = block_height_var - - self.pseudoshare_received = variable.Event() - self.share_received = variable.Event() - self.local_rate_monitor = math.RateMonitor(10*60) - - self.removed_unstales_var = variable.Variable((0, 0, 0)) - self.removed_doa_unstales_var = variable.Variable(0) - - @tracker.verified.removed.watch - def _(share): - if share.hash in self.my_share_hashes and tracker.is_child_of(share.hash, self.best_share_var.value): - assert share.share_data['stale_info'] in [None, 'orphan', 'doa'] # we made these shares in this instance - self.removed_unstales_var.set(( - self.removed_unstales_var.value[0] + 1, - self.removed_unstales_var.value[1] + (1 if share.share_data['stale_info'] == 'orphan' else 0), - self.removed_unstales_var.value[2] + (1 if share.share_data['stale_info'] == 'doa' else 0), - )) - if share.hash in self.my_doa_share_hashes and self.tracker.is_child_of(share.hash, self.best_share_var.value): - self.removed_doa_unstales_var.set(self.removed_doa_unstales_var.value + 1) - - # MERGED WORK - - self.merged_work = variable.Variable({}) - - @defer.inlineCallbacks - def set_merged_work(merged_url, merged_userpass): - merged_proxy = jsonrpc.Proxy(merged_url, dict(Authorization='Basic ' + base64.b64encode(merged_userpass))) - while True: - auxblock = yield deferral.retry('Error while calling merged getauxblock:', 1)(merged_proxy.rpc_getauxblock)() - self.merged_work.set(dict(self.merged_work.value, **{auxblock['chainid']: dict( - hash=int(auxblock['hash'], 16), - target=pack.IntType(256).unpack(auxblock['target'].decode('hex')), - merged_proxy=merged_proxy, - )})) - yield deferral.sleep(1) - for merged_url, merged_userpass in merged_urls: - set_merged_work(merged_url, merged_userpass) - - @self.merged_work.changed.watch - def _(new_merged_work): - print 'Got new merged mining work!' - - # COMBINE WORK - - self.current_work = variable.Variable(None) - def compute_work(): - t = dict(self.bitcoind_work.value) - - bb = self.best_block_header.value - if bb is not None and bb['previous_block'] == t['previous_block'] and net.PARENT.POW_FUNC(bitcoin_data.block_header_type.pack(bb)) <= t['bits'].target: - print 'Skipping from block %x to block %x!' % (bb['previous_block'], - bitcoin_data.hash256(bitcoin_data.block_header_type.pack(bb))) - t = dict( - version=bb['version'], - previous_block=bitcoin_data.hash256(bitcoin_data.block_header_type.pack(bb)), - bits=bb['bits'], # not always true - coinbaseflags='', - time=bb['timestamp'] + 600, # better way? - transactions=[], - merkle_link=bitcoin_data.calculate_merkle_link([None], 0), - subsidy=net.PARENT.SUBSIDY_FUNC(self.block_height_var.value), - clock_offset=self.current_work.value['clock_offset'], - last_update=self.current_work.value['last_update'], - ) - - self.current_work.set(t) - self.bitcoind_work.changed.watch(lambda _: compute_work()) - self.best_block_header.changed.watch(lambda _: compute_work()) - compute_work() - - self.new_work_event = variable.Event() - @self.current_work.transitioned.watch - def _(before, after): - # trigger LP if version/previous_block/bits changed or transactions changed from nothing - if any(before[x] != after[x] for x in ['version', 'previous_block', 'bits']) or (not before['transactions'] and after['transactions']): - self.new_work_event.happened() - self.merged_work.changed.watch(lambda _: self.new_work_event.happened()) - self.best_share_var.changed.watch(lambda _: self.new_work_event.happened()) - - def get_stale_counts(self): - '''Returns (orphans, doas), total, (orphans_recorded_in_chain, doas_recorded_in_chain)''' - my_shares = len(self.my_share_hashes) - my_doa_shares = len(self.my_doa_share_hashes) - delta = self.tracker.verified.get_delta_to_last(self.best_share_var.value) - my_shares_in_chain = delta.my_count + self.removed_unstales_var.value[0] - my_doa_shares_in_chain = delta.my_doa_count + self.removed_doa_unstales_var.value - orphans_recorded_in_chain = delta.my_orphan_announce_count + self.removed_unstales_var.value[1] - doas_recorded_in_chain = delta.my_dead_announce_count + self.removed_unstales_var.value[2] - - my_shares_not_in_chain = my_shares - my_shares_in_chain - my_doa_shares_not_in_chain = my_doa_shares - my_doa_shares_in_chain - - return (my_shares_not_in_chain - my_doa_shares_not_in_chain, my_doa_shares_not_in_chain), my_shares, (orphans_recorded_in_chain, doas_recorded_in_chain) - - def get_user_details(self, request): - user = request.getUser() if request.getUser() is not None else '' - - desired_pseudoshare_target = None - if '+' in user: - user, desired_pseudoshare_difficulty_str = user.rsplit('+', 1) - try: - desired_pseudoshare_target = bitcoin_data.difficulty_to_target(float(desired_pseudoshare_difficulty_str)) - except: - pass - - desired_share_target = 2**256 - 1 - if '/' in user: - user, min_diff_str = user.rsplit('/', 1) - try: - desired_share_target = bitcoin_data.difficulty_to_target(float(min_diff_str)) - except: - pass - - if random.uniform(0, 100) < self.worker_fee: - pubkey_hash = self.my_pubkey_hash - else: - try: - pubkey_hash = bitcoin_data.address_to_pubkey_hash(user, self.net.PARENT) - except: # XXX blah - pubkey_hash = self.my_pubkey_hash - - return user, pubkey_hash, desired_share_target, desired_pseudoshare_target - - def preprocess_request(self, request): - user, pubkey_hash, desired_share_target, desired_pseudoshare_target = self.get_user_details(request) - return pubkey_hash, desired_share_target, desired_pseudoshare_target - - def get_work(self, pubkey_hash, desired_share_target, desired_pseudoshare_target): - if len(self.p2p_node.peers) == 0 and self.net.PERSIST: - raise jsonrpc.Error(-12345, u'p2pool is not connected to any peers') - if self.best_share_var.value is None and self.net.PERSIST: - raise jsonrpc.Error(-12345, u'p2pool is downloading shares') - if time.time() > self.current_work.value['last_update'] + 60: - raise jsonrpc.Error(-12345, u'lost contact with bitcoind') - - if self.merged_work.value: - tree, size = bitcoin_data.make_auxpow_tree(self.merged_work.value) - mm_hashes = [self.merged_work.value.get(tree.get(i), dict(hash=0))['hash'] for i in xrange(size)] - mm_data = '\xfa\xbemm' + bitcoin_data.aux_pow_coinbase_type.pack(dict( - merkle_root=bitcoin_data.merkle_hash(mm_hashes), - size=size, - nonce=0, - )) - mm_later = [(aux_work, mm_hashes.index(aux_work['hash']), mm_hashes) for chain_id, aux_work in self.merged_work.value.iteritems()] - else: - mm_data = '' - mm_later = [] - - if True: - share_info, generate_tx = p2pool_data.Share.generate_transaction( - tracker=self.tracker, - share_data=dict( - previous_share_hash=self.best_share_var.value, - coinbase=(mm_data + self.current_work.value['coinbaseflags'])[:100], - nonce=random.randrange(2**32), - pubkey_hash=pubkey_hash, - subsidy=self.current_work.value['subsidy'], - donation=math.perfect_round(65535*self.donation_percentage/100), - stale_info=(lambda (orphans, doas), total, (orphans_recorded_in_chain, doas_recorded_in_chain): - 'orphan' if orphans > orphans_recorded_in_chain else - 'doa' if doas > doas_recorded_in_chain else - None - )(*self.get_stale_counts()), - desired_version=3, - ), - block_target=self.current_work.value['bits'].target, - desired_timestamp=int(time.time() - self.current_work.value['clock_offset']), - desired_target=desired_share_target, - ref_merkle_link=dict(branch=[], index=0), - net=self.net, - ) - - if desired_pseudoshare_target is None: - target = 2**256-1 - if len(self.recent_shares_ts_work) == 50: - hash_rate = sum(work for ts, work in self.recent_shares_ts_work[1:])//(self.recent_shares_ts_work[-1][0] - self.recent_shares_ts_work[0][0]) - if hash_rate: - target = min(target, int(2**256/hash_rate)) - else: - target = desired_pseudoshare_target - target = max(target, share_info['bits'].target) - for aux_work in self.merged_work.value.itervalues(): - target = max(target, aux_work['target']) - target = math.clip(target, self.net.PARENT.SANE_TARGET_RANGE) - - transactions = [generate_tx] + list(self.current_work.value['transactions']) - packed_generate_tx = bitcoin_data.tx_type.pack(generate_tx) - merkle_root = bitcoin_data.check_merkle_link(bitcoin_data.hash256(packed_generate_tx), self.current_work.value['merkle_link']) - - getwork_time = time.time() - merkle_link = self.current_work.value['merkle_link'] - - print 'New work for worker! Difficulty: %.06f Share difficulty: %.06f Total block value: %.6f %s including %i transactions' % ( - bitcoin_data.target_to_difficulty(target), - bitcoin_data.target_to_difficulty(share_info['bits'].target), - self.current_work.value['subsidy']*1e-8, self.net.PARENT.SYMBOL, - len(self.current_work.value['transactions']), - ) - - bits = self.current_work.value['bits'] - previous_block = self.current_work.value['previous_block'] - ba = bitcoin_getwork.BlockAttempt( - version=self.current_work.value['version'], - previous_block=self.current_work.value['previous_block'], - merkle_root=merkle_root, - timestamp=self.current_work.value['time'], - bits=self.current_work.value['bits'], - share_target=target, - ) - - received_header_hashes = set() - - def got_response(header, request): - header_hash = bitcoin_data.hash256(bitcoin_data.block_header_type.pack(header)) - pow_hash = self.net.PARENT.POW_FUNC(bitcoin_data.block_header_type.pack(header)) - try: - if pow_hash <= header['bits'].target or p2pool.DEBUG: - self.submit_block(dict(header=header, txs=transactions), ignore_failure=False) - if pow_hash <= header['bits'].target: - print - print 'GOT BLOCK FROM MINER! Passing to bitcoind! %s%064x' % (self.net.PARENT.BLOCK_EXPLORER_URL_PREFIX, header_hash) - print - except: - log.err(None, 'Error while processing potential block:') - - user, _, _, _ = self.get_user_details(request) - assert header['merkle_root'] == merkle_root - assert header['previous_block'] == previous_block - assert header['bits'] == bits - - on_time = self.best_share_var.value == share_info['share_data']['previous_share_hash'] - - for aux_work, index, hashes in mm_later: - try: - if pow_hash <= aux_work['target'] or p2pool.DEBUG: - df = deferral.retry('Error submitting merged block: (will retry)', 10, 10)(aux_work['merged_proxy'].rpc_getauxblock)( - pack.IntType(256, 'big').pack(aux_work['hash']).encode('hex'), - bitcoin_data.aux_pow_type.pack(dict( - merkle_tx=dict( - tx=transactions[0], - block_hash=header_hash, - merkle_link=merkle_link, - ), - merkle_link=bitcoin_data.calculate_merkle_link(hashes, index), - parent_block_header=header, - )).encode('hex'), - ) - @df.addCallback - def _(result): - if result != (pow_hash <= aux_work['target']): - print >>sys.stderr, 'Merged block submittal result: %s Expected: %s' % (result, pow_hash <= aux_work['target']) - else: - print 'Merged block submittal result: %s' % (result,) - @df.addErrback - def _(err): - log.err(err, 'Error submitting merged block:') - except: - log.err(None, 'Error while processing merged mining POW:') - - if pow_hash <= share_info['bits'].target and header_hash not in received_header_hashes: - min_header = dict(header);del min_header['merkle_root'] - hash_link = p2pool_data.prefix_to_hash_link(packed_generate_tx[:-32-4], p2pool_data.Share.gentx_before_refhash) - share = p2pool_data.Share(self.net, None, dict( - min_header=min_header, share_info=share_info, hash_link=hash_link, - ref_merkle_link=dict(branch=[], index=0), - ), merkle_link=merkle_link, other_txs=transactions[1:] if pow_hash <= header['bits'].target else None) - - print 'GOT SHARE! %s %s prev %s age %.2fs%s' % ( - request.getUser(), - p2pool_data.format_hash(share.hash), - p2pool_data.format_hash(share.previous_hash), - time.time() - getwork_time, - ' DEAD ON ARRIVAL' if not on_time else '', - ) - self.my_share_hashes.add(share.hash) - if not on_time: - self.my_doa_share_hashes.add(share.hash) - - self.tracker.add(share) - if not p2pool.DEBUG: - self.tracker.verified.add(share) - self.set_best_share() - - try: - if pow_hash <= header['bits'].target or p2pool.DEBUG: - for peer in self.p2p_node.peers.itervalues(): - peer.sendShares([share]) - self.shared_share_hashes.add(share.hash) - except: - log.err(None, 'Error forwarding block solution:') - - self.share_received.happened(bitcoin_data.target_to_average_attempts(share.target), not on_time) - - if pow_hash > target: - print 'Worker %s submitted share with hash > target:' % (request.getUser(),) - print ' Hash: %56x' % (pow_hash,) - print ' Target: %56x' % (target,) - elif header_hash in received_header_hashes: - print >>sys.stderr, 'Worker %s @ %s submitted share more than once!' % (request.getUser(), request.getClientIP()) - else: - received_header_hashes.add(header_hash) - - self.pseudoshare_received.happened(bitcoin_data.target_to_average_attempts(target), not on_time, user) - self.recent_shares_ts_work.append((time.time(), bitcoin_data.target_to_average_attempts(target))) - while len(self.recent_shares_ts_work) > 50: - self.recent_shares_ts_work.pop(0) - self.local_rate_monitor.add_datum(dict(work=bitcoin_data.target_to_average_attempts(target), dead=not on_time, user=user)) - - return on_time - - return ba, got_response +import bitcoin.p2p as bitcoin_p2p, bitcoin.data as bitcoin_data +from bitcoin import worker_interface, helper +from util import fixargparse, jsonrpc, variable, deferral, math, logging +from . import networks, web, work +import p2pool, p2pool.data as p2pool_data, p2pool.node as p2pool_node @defer.inlineCallbacks def main(args, net, datadir_path, merged_urls, worker_endpoint): @@ -386,42 +30,42 @@ def main(args, net, datadir_path, merged_urls, worker_endpoint): print 'p2pool (version %s)' % (p2pool.__version__,) print + @defer.inlineCallbacks + def connect_p2p(): + # connect to bitcoind over bitcoin-p2p + print '''Testing bitcoind P2P connection to '%s:%s'...''' % (args.bitcoind_address, args.bitcoind_p2p_port) + factory = bitcoin_p2p.ClientFactory(net.PARENT) + reactor.connectTCP(args.bitcoind_address, args.bitcoind_p2p_port, factory) + yield factory.getProtocol() # waits until handshake is successful + print ' ...success!' + print + defer.returnValue(factory) + + if args.testnet: # establish p2p connection first if testnet so bitcoind can work without connections + factory = yield connect_p2p() + # connect to bitcoind over JSON-RPC and do initial getmemorypool - url = 'http://%s:%i/' % (args.bitcoind_address, args.bitcoind_rpc_port) + url = '%s://%s:%i/' % ('https' if args.bitcoind_rpc_ssl else 'http', args.bitcoind_address, args.bitcoind_rpc_port) print '''Testing bitcoind RPC connection to '%s' with username '%s'...''' % (url, args.bitcoind_rpc_username) bitcoind = jsonrpc.Proxy(url, dict(Authorization='Basic ' + base64.b64encode(args.bitcoind_rpc_username + ':' + args.bitcoind_rpc_password)), timeout=30) - @deferral.retry('Error while checking Bitcoin connection:', 1) - @defer.inlineCallbacks - def check(): - if not (yield net.PARENT.RPC_CHECK)(bitcoind): - print >>sys.stderr, " Check failed! Make sure that you're connected to the right bitcoind with --bitcoind-rpc-port!" - raise deferral.RetrySilentlyException() - temp_work = yield getwork(bitcoind) - if not net.VERSION_CHECK((yield bitcoind.rpc_getinfo())['version'], temp_work): - print >>sys.stderr, ' Bitcoin version too old! BIP16 support required! Upgrade to 0.6.0rc4 or greater!' - raise deferral.RetrySilentlyException() - defer.returnValue(temp_work) - temp_work = yield check() + yield helper.check(bitcoind, net) + temp_work = yield helper.getwork(bitcoind) - block_height_var = variable.Variable(None) + bitcoind_warning_var = variable.Variable(None) @defer.inlineCallbacks - def poll_height(): - block_height_var.set((yield deferral.retry('Error while calling getblockcount:')(bitcoind.rpc_getblockcount)())) - yield poll_height() - task.LoopingCall(poll_height).start(60*60) + def poll_warnings(): + errors = (yield deferral.retry('Error while calling getmininginfo:')(bitcoind.rpc_getmininginfo)())['errors'] + bitcoind_warning_var.set(errors if errors != '' else None) + yield poll_warnings() + task.LoopingCall(poll_warnings).start(20*60) print ' ...success!' - print ' Current block hash: %x' % (temp_work['previous_block_hash'],) - print ' Current block height: %i' % (block_height_var.value,) + print ' Current block hash: %x' % (temp_work['previous_block'],) + print ' Current block height: %i' % (temp_work['height'] - 1,) print - # connect to bitcoind over bitcoin-p2p - print '''Testing bitcoind P2P connection to '%s:%s'...''' % (args.bitcoind_address, args.bitcoind_p2p_port) - factory = bitcoin_p2p.ClientFactory(net.PARENT) - reactor.connectTCP(args.bitcoind_address, args.bitcoind_p2p_port, factory) - yield factory.getProtocol() # waits until handshake is successful - print ' ...success!' - print + if not args.testnet: + factory = yield connect_p2p() print 'Determining payout address...' if args.pubkey_hash is None: @@ -453,239 +97,49 @@ def main(args, net, datadir_path, merged_urls, worker_endpoint): print ' ...success! Payout address:', bitcoin_data.pubkey_hash_to_address(my_pubkey_hash, net.PARENT) print - my_share_hashes = set() - my_doa_share_hashes = set() - - tracker = p2pool_data.OkayTracker(net, my_share_hashes, my_doa_share_hashes) - shared_share_hashes = set() ss = p2pool_data.ShareStore(os.path.join(datadir_path, 'shares.'), net) + shares = {} known_verified = set() print "Loading shares..." for i, (mode, contents) in enumerate(ss.get_shares()): if mode == 'share': - if contents.hash in tracker.shares: - continue - shared_share_hashes.add(contents.hash) contents.time_seen = 0 - tracker.add(contents) - if len(tracker.shares) % 1000 == 0 and tracker.shares: - print " %i" % (len(tracker.shares),) + shares[contents.hash] = contents + if len(shares) % 1000 == 0 and shares: + print " %i" % (len(shares),) elif mode == 'verified_hash': known_verified.add(contents) else: raise AssertionError() - print " ...inserting %i verified shares..." % (len(known_verified),) - for h in known_verified: - if h not in tracker.shares: - ss.forget_verified_share(h) - continue - tracker.verified.add(tracker.shares[h]) - print " ...done loading %i shares!" % (len(tracker.shares),) + print " ...done loading %i shares (%i verified)!" % (len(shares), len(known_verified)) print - tracker.removed.watch(lambda share: ss.forget_share(share.hash)) - tracker.verified.removed.watch(lambda share: ss.forget_verified_share(share.hash)) - tracker.removed.watch(lambda share: shared_share_hashes.discard(share.hash)) - - print 'Initializing work...' - - # BITCOIND WORK - - bitcoind_work = variable.Variable(None) - - @defer.inlineCallbacks - def poll_bitcoind(): - work = yield getwork(bitcoind) - bitcoind_work.set(dict( - version=work['version'], - previous_block=work['previous_block_hash'], - bits=work['bits'], - coinbaseflags=work['coinbaseflags'], - time=work['time'], - transactions=work['transactions'], - merkle_link=work['merkle_link'], - subsidy=work['subsidy'], - clock_offset=time.time() - work['time'], - last_update=time.time(), - )) - yield poll_bitcoind() - @defer.inlineCallbacks - def work_poller(): - while True: - flag = factory.new_block.get_deferred() - try: - yield poll_bitcoind() - except: - log.err() - yield defer.DeferredList([flag, deferral.sleep(15)], fireOnOneCallback=True) - work_poller() - - # PEER WORK - - best_block_header = variable.Variable(None) - def handle_header(new_header): - # check that header matches current target - if not (net.PARENT.POW_FUNC(bitcoin_data.block_header_type.pack(new_header)) <= bitcoind_work.value['bits'].target): - return - bitcoind_best_block = bitcoind_work.value['previous_block'] - if (best_block_header.value is None - or ( - new_header['previous_block'] == bitcoind_best_block and - bitcoin_data.hash256(bitcoin_data.block_header_type.pack(best_block_header.value)) == bitcoind_best_block - ) # new is child of current and previous is current - or ( - bitcoin_data.hash256(bitcoin_data.block_header_type.pack(new_header)) == bitcoind_best_block and - best_block_header.value['previous_block'] != bitcoind_best_block - )): # new is current and previous is not a child of current - best_block_header.set(new_header) - @defer.inlineCallbacks - def poll_header(): - handle_header((yield factory.conn.value.get_block_header(bitcoind_work.value['previous_block']))) - bitcoind_work.changed.watch(lambda _: poll_header()) - yield poll_header() + print 'Initializing work...' - # BEST SHARE + node = p2pool_node.Node(factory, bitcoind, shares.values(), known_verified, net) + yield node.start() - get_height_rel_highest = yield height_tracker.get_height_rel_highest_func(bitcoind, factory, lambda: bitcoind_work.value['previous_block'], net) - requested = expiring_dict.ExpiringDict(300) - peer_heads = expiring_dict.ExpiringDict(300) # hash -> peers that know of it - - best_share_var = variable.Variable(None) - def set_best_share(): - best, desired = tracker.think(get_height_rel_highest, bitcoind_work.value['previous_block'], bitcoind_work.value['bits']) - - best_share_var.set(best) - - t = time.time() - for peer2, share_hash in desired: - if share_hash not in tracker.tails: # was received in the time tracker.think was running - continue - last_request_time, count = requested.get(share_hash, (None, 0)) - if last_request_time is not None and last_request_time - 5 < t < last_request_time + 10 * 1.5**count: - continue - potential_peers = set() - for head in tracker.tails[share_hash]: - potential_peers.update(peer_heads.get(head, set())) - potential_peers = [peer for peer in potential_peers if peer.connected2] - if count == 0 and peer2 is not None and peer2.connected2: - peer = peer2 - else: - peer = random.choice(potential_peers) if potential_peers and random.random() > .2 else peer2 - if peer is None: - continue - - print 'Requesting parent share %s from %s' % (p2pool_data.format_hash(share_hash), '%s:%i' % peer.addr) - peer.send_getshares( - hashes=[share_hash], - parents=2000, - stops=list(set(tracker.heads) | set( - tracker.get_nth_parent_hash(head, min(max(0, tracker.get_height_and_last(head)[0] - 1), 10)) for head in tracker.heads - ))[:100], - ) - requested[share_hash] = t, count + 1 - bitcoind_work.changed.watch(lambda _: set_best_share()) - set_best_share() + for share_hash in shares: + if share_hash not in node.tracker.items: + ss.forget_share(share_hash) + for share_hash in known_verified: + if share_hash not in node.tracker.verified.items: + ss.forget_verified_share(share_hash) + del shares, known_verified + node.tracker.removed.watch(lambda share: ss.forget_share(share.hash)) + node.tracker.verified.removed.watch(lambda share: ss.forget_verified_share(share.hash)) + def save_shares(): + for share in node.tracker.get_chain(node.best_share_var.value, min(node.tracker.get_height(node.best_share_var.value), 2*net.CHAIN_LENGTH)): + ss.add_share(share) + if share.hash in node.tracker.verified.items: + ss.add_verified_hash(share.hash) + task.LoopingCall(save_shares).start(60) print ' ...success!' print - # setup p2p logic and join p2pool network - - class Node(p2p.Node): - def handle_shares(self, shares, peer): - if len(shares) > 5: - print 'Processing %i shares from %s...' % (len(shares), '%s:%i' % peer.addr if peer is not None else None) - - new_count = 0 - for share in shares: - if share.hash in tracker.shares: - #print 'Got duplicate share, ignoring. Hash: %s' % (p2pool_data.format_hash(share.hash),) - continue - - new_count += 1 - - #print 'Received share %s from %r' % (p2pool_data.format_hash(share.hash), share.peer.addr if share.peer is not None else None) - - tracker.add(share) - - if shares and peer is not None: - peer_heads.setdefault(shares[0].hash, set()).add(peer) - - if new_count: - set_best_share() - - if len(shares) > 5: - print '... done processing %i shares. New: %i Have: %i/~%i' % (len(shares), new_count, len(tracker.shares), 2*net.CHAIN_LENGTH) - - def handle_share_hashes(self, hashes, peer): - t = time.time() - get_hashes = [] - for share_hash in hashes: - if share_hash in tracker.shares: - continue - last_request_time, count = requested.get(share_hash, (None, 0)) - if last_request_time is not None and last_request_time - 5 < t < last_request_time + 10 * 1.5**count: - continue - print 'Got share hash, requesting! Hash: %s' % (p2pool_data.format_hash(share_hash),) - get_hashes.append(share_hash) - requested[share_hash] = t, count + 1 - - if hashes and peer is not None: - peer_heads.setdefault(hashes[0], set()).add(peer) - if get_hashes: - peer.send_getshares(hashes=get_hashes, parents=0, stops=[]) - - def handle_get_shares(self, hashes, parents, stops, peer): - parents = min(parents, 1000//len(hashes)) - stops = set(stops) - shares = [] - for share_hash in hashes: - for share in tracker.get_chain(share_hash, min(parents + 1, tracker.get_height(share_hash))): - if share.hash in stops: - break - shares.append(share) - print 'Sending %i shares to %s:%i' % (len(shares), peer.addr[0], peer.addr[1]) - return shares - - def handle_bestblock(self, header, peer): - if net.PARENT.POW_FUNC(bitcoin_data.block_header_type.pack(header)) > header['bits'].target: - raise p2p.PeerMisbehavingError('received block header fails PoW test') - handle_header(header) - - @deferral.retry('Error submitting primary block: (will retry)', 10, 10) - def submit_block_p2p(block): - if factory.conn.value is None: - print >>sys.stderr, 'No bitcoind connection when block submittal attempted! %s%32x' % (net.PARENT.BLOCK_EXPLORER_URL_PREFIX, bitcoin_data.hash256(bitcoin_data.block_header_type.pack(block['header']))) - raise deferral.RetrySilentlyException() - factory.conn.value.send_block(block=block) - - @deferral.retry('Error submitting block: (will retry)', 10, 10) - @defer.inlineCallbacks - def submit_block_rpc(block, ignore_failure): - success = yield bitcoind.rpc_getmemorypool(bitcoin_data.block_type.pack(block).encode('hex')) - success_expected = net.PARENT.POW_FUNC(bitcoin_data.block_header_type.pack(block['header'])) <= block['header']['bits'].target - if (not success and success_expected and not ignore_failure) or (success and not success_expected): - print >>sys.stderr, 'Block submittal result: %s Expected: %s' % (success, success_expected) - - def submit_block(block, ignore_failure): - submit_block_p2p(block) - submit_block_rpc(block, ignore_failure) - - @tracker.verified.added.watch - def _(share): - if share.pow_hash <= share.header['bits'].target: - submit_block(share.as_block(tracker), ignore_failure=True) - print - print 'GOT BLOCK FROM PEER! Passing to bitcoind! %s bitcoin: %s%064x' % (p2pool_data.format_hash(share.hash), net.PARENT.BLOCK_EXPLORER_URL_PREFIX, share.header_hash) - print - def spread(): - if (get_height_rel_highest(share.header['previous_block']) > -5 or - bitcoind_work.value['previous_block'] in [share.header['previous_block'], share.header_hash]): - broadcast_share(share.hash) - spread() - reactor.callLater(5, spread) # so get_height_rel_highest can update print 'Joining p2pool network using port %i...' % (args.p2pool_port,) @@ -719,47 +173,20 @@ def main(args, net, datadir_path, merged_urls, worker_endpoint): except: log.err() - p2p_node = Node( - best_share_hash_func=lambda: best_share_var.value, + node.p2p_node = p2pool_node.P2PNode(node, port=args.p2pool_port, - net=net, + max_incoming_conns=args.p2pool_conns, addr_store=addrs, connect_addrs=connect_addrs, - max_incoming_conns=args.p2pool_conns, + desired_outgoing_conns=args.p2pool_outgoing_conns, ) - p2p_node.start() + node.p2p_node.start() def save_addrs(): with open(os.path.join(datadir_path, 'addrs'), 'wb') as f: - f.write(json.dumps(p2p_node.addr_store.items())) + f.write(json.dumps(node.p2p_node.addr_store.items())) task.LoopingCall(save_addrs).start(60) - @best_block_header.changed.watch - def _(header): - for peer in p2p_node.peers.itervalues(): - peer.send_bestblock(header=header) - - def broadcast_share(share_hash): - shares = [] - for share in tracker.get_chain(share_hash, min(5, tracker.get_height(share_hash))): - if share.hash in shared_share_hashes: - break - shared_share_hashes.add(share.hash) - shares.append(share) - - for peer in p2p_node.peers.itervalues(): - peer.sendShares([share for share in shares if share.peer is not peer]) - - # send share when the chain changes to their chain - best_share_var.changed.watch(broadcast_share) - - def save_shares(): - for share in tracker.get_chain(best_share_var.value, min(tracker.get_height(best_share_var.value), 2*net.CHAIN_LENGTH)): - ss.add_share(share) - if share.hash in tracker.verified.shares: - ss.add_verified_hash(share.hash) - task.LoopingCall(save_shares).start(60) - print ' ...success!' print @@ -784,10 +211,8 @@ def main(args, net, datadir_path, merged_urls, worker_endpoint): print 'Listening for workers on %r port %i...' % (worker_endpoint[0], worker_endpoint[1]) - get_current_txouts = lambda: p2pool_data.get_expected_payouts(tracker, best_share_var.value, bitcoind_work.value['bits'].target, bitcoind_work.value['subsidy'], net) - - wb = WorkerBridge(my_pubkey_hash, net, args.donation_percentage, bitcoind_work, best_block_header, merged_urls, best_share_var, tracker, my_share_hashes, my_doa_share_hashes, args.worker_fee, p2p_node, submit_block, set_best_share, shared_share_hashes, block_height_var) - web_root = web.get_web_root(tracker, bitcoind_work, get_current_txouts, datadir_path, net, wb.get_stale_counts, my_pubkey_hash, wb.local_rate_monitor, args.worker_fee, p2p_node, wb.my_share_hashes, wb.pseudoshare_received, wb.share_received, best_share_var) + wb = work.WorkerBridge(node, my_pubkey_hash, args.donation_percentage, merged_urls, args.worker_fee) + web_root = web.get_web_root(wb, datadir_path, bitcoind_warning_var) worker_interface.WorkerInterface(wb).attach_to(web_root, get_handler=lambda request: request.redirect('/static/')) deferral.retry('Error binding to worker port:', traceback=False)(reactor.listenTCP)(worker_endpoint[1], server.Site(web_root), interface=worker_endpoint[0]) @@ -829,19 +254,26 @@ def main(args, net, datadir_path, merged_urls, worker_endpoint): print repr(line) irc.IRCClient.lineReceived(self, line) def signedOn(self): + self.in_channel = False irc.IRCClient.signedOn(self) self.factory.resetDelay() self.join(self.channel) @defer.inlineCallbacks def new_share(share): + if not self.in_channel: + return if share.pow_hash <= share.header['bits'].target and abs(share.timestamp - time.time()) < 10*60: yield deferral.sleep(random.expovariate(1/60)) message = '\x02%s BLOCK FOUND by %s! %s%064x' % (net.NAME.upper(), bitcoin_data.script2_to_address(share.new_script, net.PARENT), net.PARENT.BLOCK_EXPLORER_URL_PREFIX, share.header_hash) - if message not in self.recent_messages: + if all('%x' % (share.header_hash,) not in old_message for old_message in self.recent_messages): self.say(self.channel, message) self._remember_message(message) - self.watch_id = tracker.verified.added.watch(new_share) + self.watch_id = node.tracker.verified.added.watch(new_share) self.recent_messages = [] + def joined(self, channel): + self.in_channel = True + def left(self, channel): + self.in_channel = False def _remember_message(self, message): self.recent_messages.append(message) while len(self.recent_messages) > 100: @@ -850,7 +282,7 @@ def main(args, net, datadir_path, merged_urls, worker_endpoint): if channel == self.channel: self._remember_message(message) def connectionLost(self, reason): - tracker.verified.added.unwatch(self.watch_id) + node.tracker.verified.added.unwatch(self.watch_id) print 'IRC connection lost:', reason.getErrorMessage() class IRCClientFactory(protocol.ReconnectingClientFactory): protocol = IRCClient @@ -863,16 +295,13 @@ def main(args, net, datadir_path, merged_urls, worker_endpoint): while True: yield deferral.sleep(3) try: - if time.time() > bitcoind_work.value['last_update'] + 60: - print >>sys.stderr, '''---> LOST CONTACT WITH BITCOIND for %s! Check that it isn't frozen or dead! <---''' % (math.format_dt(time.time() - bitcoind_work.value['last_update']),) - - height = tracker.get_height(best_share_var.value) + height = node.tracker.get_height(node.best_share_var.value) this_str = 'P2Pool: %i shares in chain (%i verified/%i total) Peers: %i (%i incoming)' % ( height, - len(tracker.verified.shares), - len(tracker.shares), - len(p2p_node.peers), - sum(1 for peer in p2p_node.peers.itervalues() if peer.incoming), + len(node.tracker.verified.items), + len(node.tracker.items), + len(node.p2p_node.peers), + sum(1 for peer in node.p2p_node.peers.itervalues() if peer.incoming), ) + (' FDs: %i R/%i W' % (len(reactor.getReaders()), len(reactor.getWriters())) if p2pool.DEBUG else '') datums, dt = wb.local_rate_monitor.get_datums_in_last() @@ -881,27 +310,27 @@ def main(args, net, datadir_path, merged_urls, worker_endpoint): math.format(int(my_att_s)), math.format_dt(dt), math.format_binomial_conf(sum(1 for datum in datums if datum['dead']), len(datums), 0.95), - math.format_dt(2**256 / tracker.shares[best_share_var.value].max_target / my_att_s) if my_att_s and best_share_var.value else '???', + math.format_dt(2**256 / node.tracker.items[node.best_share_var.value].max_target / my_att_s) if my_att_s and node.best_share_var.value else '???', ) if height > 2: (stale_orphan_shares, stale_doa_shares), shares, _ = wb.get_stale_counts() - stale_prop = p2pool_data.get_average_stale_prop(tracker, best_share_var.value, min(60*60//net.SHARE_PERIOD, height)) - real_att_s = p2pool_data.get_pool_attempts_per_second(tracker, best_share_var.value, min(height - 1, 60*60//net.SHARE_PERIOD)) / (1 - stale_prop) + stale_prop = p2pool_data.get_average_stale_prop(node.tracker, node.best_share_var.value, min(60*60//net.SHARE_PERIOD, height)) + real_att_s = p2pool_data.get_pool_attempts_per_second(node.tracker, node.best_share_var.value, min(height - 1, 60*60//net.SHARE_PERIOD)) / (1 - stale_prop) this_str += '\n Shares: %i (%i orphan, %i dead) Stale rate: %s Efficiency: %s Current payout: %.4f %s' % ( shares, stale_orphan_shares, stale_doa_shares, math.format_binomial_conf(stale_orphan_shares + stale_doa_shares, shares, 0.95), math.format_binomial_conf(stale_orphan_shares + stale_doa_shares, shares, 0.95, lambda x: (1 - x)/(1 - stale_prop)), - get_current_txouts().get(bitcoin_data.pubkey_hash_to_script2(my_pubkey_hash), 0)*1e-8, net.PARENT.SYMBOL, + node.get_current_txouts().get(bitcoin_data.pubkey_hash_to_script2(my_pubkey_hash), 0)*1e-8, net.PARENT.SYMBOL, ) this_str += '\n Pool: %sH/s Stale rate: %.1f%% Expected time to block: %s' % ( math.format(int(real_att_s)), 100*stale_prop, - math.format_dt(2**256 / bitcoind_work.value['bits'].target / real_att_s), + math.format_dt(2**256 / node.bitcoind_work.value['bits'].target / real_att_s), ) - for warning in p2pool_data.get_warnings(tracker, best_share_var.value, net): + for warning in p2pool_data.get_warnings(node.tracker, node.best_share_var.value, net, bitcoind_warning_var.value, node.bitcoind_work.value): print >>sys.stderr, '#'*40 print >>sys.stderr, '>>> Warning: ' + warning print >>sys.stderr, '#'*40 @@ -918,7 +347,7 @@ def main(args, net, datadir_path, merged_urls, worker_endpoint): log.err(None, 'Fatal error:') def run(): - realnets=dict((name, net) for name, net in networks.nets.iteritems() if '_testnet' not in name) + realnets = dict((name, net) for name, net in networks.nets.iteritems() if '_testnet' not in name) parser = fixargparse.FixedArgumentParser(description='p2pool (version %s)' % (p2pool.__version__,), fromfile_prefix_chars='@') parser.add_argument('--version', action='version', version=p2pool.__version__) @@ -969,6 +398,9 @@ def run(): p2pool_group.add_argument('--max-conns', metavar='CONNS', help='maximum incoming connections (default: 40)', type=int, action='store', default=40, dest='p2pool_conns') + p2pool_group.add_argument('--outgoing-conns', metavar='CONNS', + help='outgoing connections (default: 10)', + type=int, action='store', default=10, dest='p2pool_outgoing_conns') worker_group = parser.add_argument_group('worker interface') worker_group.add_argument('-w', '--worker-port', metavar='PORT or ADDR:PORT', @@ -985,6 +417,9 @@ def run(): bitcoind_group.add_argument('--bitcoind-rpc-port', metavar='BITCOIND_RPC_PORT', help='''connect to JSON-RPC interface at this port (default: %s )''' % ', '.join('%s:%i' % (name, net.PARENT.RPC_PORT) for name, net in sorted(realnets.items())), type=int, action='store', default=None, dest='bitcoind_rpc_port') + bitcoind_group.add_argument('--bitcoind-rpc-ssl', + help='connect to JSON-RPC interface using SSL', + action='store_true', default=False, dest='bitcoind_rpc_ssl') bitcoind_group.add_argument('--bitcoind-p2p-port', metavar='BITCOIND_P2P_PORT', help='''connect to P2P interface at this port (default: %s )''' % ', '.join('%s:%i' % (name, net.PARENT.P2P_PORT) for name, net in sorted(realnets.items())), type=int, action='store', default=None, dest='bitcoind_p2p_port') @@ -997,6 +432,7 @@ def run(): if args.debug: p2pool.DEBUG = True + defer.setDebugging(True) net_name = args.net_name + ('_testnet' if args.testnet else '') net = networks.nets[net_name] @@ -1010,26 +446,32 @@ def run(): args.bitcoind_rpc_username, args.bitcoind_rpc_password = ([None, None] + args.bitcoind_rpc_userpass)[-2:] if args.bitcoind_rpc_password is None: - if not hasattr(net.PARENT, 'CONF_FILE_FUNC'): - parser.error('This network has no configuration file function. Manually enter your RPC password.') conf_path = net.PARENT.CONF_FILE_FUNC() if not os.path.exists(conf_path): parser.error('''Bitcoin configuration file not found. Manually enter your RPC password.\r\n''' '''If you actually haven't created a configuration file, you should create one at %s with the text:\r\n''' '''\r\n''' '''server=1\r\n''' - '''rpcpassword=%x''' % (conf_path, random.randrange(2**128))) - with open(conf_path, 'rb') as f: - cp = ConfigParser.RawConfigParser() - cp.readfp(StringIO.StringIO('[x]\r\n' + f.read())) - for conf_name, var_name, var_type in [ - ('rpcuser', 'bitcoind_rpc_username', str), - ('rpcpassword', 'bitcoind_rpc_password', str), - ('rpcport', 'bitcoind_rpc_port', int), - ('port', 'bitcoind_p2p_port', int), - ]: - if getattr(args, var_name) is None and cp.has_option('x', conf_name): - setattr(args, var_name, var_type(cp.get('x', conf_name))) + '''rpcpassword=%x\r\n''' + '''\r\n''' + '''Keep that password secret! After creating the file, restart Bitcoin.''' % (conf_path, random.randrange(2**128))) + conf = open(conf_path, 'rb').read() + contents = {} + for line in conf.splitlines(True): + if '#' in line: + line = line[:line.index('#')] + if '=' not in line: + continue + k, v = line.split('=', 1) + contents[k.strip()] = v.strip() + for conf_name, var_name, var_type in [ + ('rpcuser', 'bitcoind_rpc_username', str), + ('rpcpassword', 'bitcoind_rpc_password', str), + ('rpcport', 'bitcoind_rpc_port', int), + ('port', 'bitcoind_p2p_port', int), + ]: + if getattr(args, var_name) is None and conf_name in contents: + setattr(args, var_name, var_type(contents[conf_name])) if args.bitcoind_rpc_password is None: parser.error('''Bitcoin configuration file didn't contain an rpcpassword= line! Add one!''') @@ -1045,6 +487,9 @@ def run(): if args.p2pool_port is None: args.p2pool_port = net.P2P_PORT + if args.p2pool_outgoing_conns > 10: + parser.error('''--outgoing-conns can't be more than 10''') + if args.worker_endpoint is None: worker_endpoint = '', net.WORKER_PORT elif ':' not in args.worker_endpoint: