p2pool/main.py

   1 #!/usr/bin/python
   2
   3 from __future__ import division
   4
   5 import argparse
   6 import datetime
   7 import itertools
   8 import os
   9 import random
  10 import sqlite3
  11 import struct
  12 import sys
  13 import time
  14 import json
  15 import signal
  16 import traceback
  17
  18 from twisted.internet import defer, reactor, task
  19 from twisted.web import server, resource
  20 from twisted.python import log
  21 from nattraverso import portmapper, ipdiscover
  22
  23 import bitcoin.p2p, bitcoin.getwork, bitcoin.data
  24 from util import db, expiring_dict, jsonrpc, variable, deferral, math
  25 from . import p2p, worker_interface, skiplists
  26 import p2pool.data as p2pool
  27 import p2pool as p2pool_init
  28
  29 @deferral.retry('Error getting work from bitcoind:', 3)
  30 @defer.inlineCallbacks
  31 def getwork(bitcoind, ht, net):
  32     try:
  33         work = yield bitcoind.rpc_getmemorypool()
  34         defer.returnValue(dict(
  35             version=work['version'],
  36             previous_block_hash=int(work['previousblockhash'], 16),
  37             transactions=[bitcoin.data.tx_type.unpack(x.decode('hex')) for x in work['transactions']],
  38             subsidy=work['coinbasevalue'],
  39             time=work['time'],
  40             target=bitcoin.data.FloatingIntegerType().unpack(work['bits'].decode('hex')[::-1]) if isinstance(work['bits'], (str, unicode)) else bitcoin.data.FloatingInteger(work['bits']),
  41         ))
  42     except jsonrpc.Error, e:
  43         if e.code != -32601:
  44             raise
  45
  46         print "---> Update your bitcoind to support the 'getmemorypool' RPC call. Not including transactions in generated blocks! <---"
  47         work = bitcoin.getwork.BlockAttempt.from_getwork((yield bitcoind.rpc_getwork()))
  48         try:
  49             subsidy = net.BITCOIN_SUBSIDY_FUNC(ht.getHeight(work.previous_block))
  50         except ValueError:
  51             subsidy = net.BITCOIN_SUBSIDY_FUNC(1000)
  52
  53         defer.returnValue(dict(
  54             version=work.version,
  55             previous_block_hash=work.previous_block,
  56             transactions=[],
  57             subsidy=subsidy,
  58             time=work.timestamp,
  59             target=work.block_target,
  60         ))
  61
  62 @deferral.retry('Error getting payout script from bitcoind:', 1)
  63 @defer.inlineCallbacks
  64 def get_payout_script(factory):
  65     res = yield (yield factory.getProtocol()).check_order(order=bitcoin.p2p.Protocol.null_order)
  66     if res['reply'] == 'success':
  67         defer.returnValue(res['script'])
  68     elif res['reply'] == 'denied':
  69         defer.returnValue(None)
  70     else:
  71         raise ValueError('Unexpected reply: %r' % (res,))
  72
  73 @deferral.retry('Error creating payout script:', 10)
  74 @defer.inlineCallbacks
  75 def get_payout_script2(bitcoind, net):
  76     defer.returnValue(bitcoin.data.pubkey_hash_to_script2(bitcoin.data.address_to_pubkey_hash((yield bitcoind.rpc_getaccountaddress('p2pool')), net)))
  77
  78 @defer.inlineCallbacks
  79 def main(args):
  80     try:
  81         print 'p2pool (version %s)' % (p2pool_init.__version__,)
  82         print
  83         try:
  84             from . import draw
  85         except ImportError:
  86             draw = None
  87             print "Install Pygame and PIL to enable visualizations! Visualizations disabled."
  88             print
  89
  90         # connect to bitcoind over JSON-RPC and do initial getwork
  91         url = 'http://%s:%i/' % (args.bitcoind_address, args.bitcoind_rpc_port)
  92         print '''Testing bitcoind RPC connection to '%s' with username '%s'...''' % (url, args.bitcoind_rpc_username)
  93         bitcoind = jsonrpc.Proxy(url, (args.bitcoind_rpc_username, args.bitcoind_rpc_password))
  94         good = yield deferral.retry('Error while checking bitcoind identity:', 1)(args.net.BITCOIN_RPC_CHECK)(bitcoind)
  95         if not good:
  96             print "    Check failed! Make sure that you're connected to the right bitcoind with --bitcoind-rpc-port!"
  97             return
  98         temp_work = yield deferral.retry('Error while testing getwork:', 1)(defer.inlineCallbacks(lambda: defer.returnValue(bitcoin.getwork.BlockAttempt.from_getwork((yield bitcoind.rpc_getwork())))))()
  99         print '    ...success!'
 100         print '    Current block hash: %x' % (temp_work.previous_block,)
 101         print
 102
 103         # connect to bitcoind over bitcoin-p2p and do checkorder to get pubkey to send payouts to
 104         print '''Testing bitcoind P2P connection to '%s:%s'...''' % (args.bitcoind_address, args.bitcoind_p2p_port)
 105         factory = bitcoin.p2p.ClientFactory(args.net)
 106         reactor.connectTCP(args.bitcoind_address, args.bitcoind_p2p_port, factory)
 107         my_script = yield get_payout_script(factory)
 108         if args.pubkey_hash is None:
 109             if my_script is None:
 110                 print '    IP transaction denied ... falling back to sending to address.'
 111                 my_script = yield get_payout_script2(bitcoind, args.net)
 112         else:
 113             my_script = bitcoin.data.pubkey_hash_to_script2(args.pubkey_hash)
 114         print '    ...success!'
 115         print '    Payout script:', bitcoin.data.script2_to_human(my_script, args.net)
 116         print
 117
 118         print 'Loading cached block headers...'
 119         ht = bitcoin.p2p.HeightTracker(factory, args.net.NAME + '_headers.dat')
 120         print '   ...done loading %i cached block headers.' % (len(ht.tracker.shares),)
 121         print
 122
 123         tracker = p2pool.OkayTracker(args.net)
 124         ss = p2pool.ShareStore(os.path.join(os.path.dirname(sys.argv[0]), args.net.NAME + '_shares.'), args.net)
 125         known_verified = set()
 126         print "Loading shares..."
 127         for i, (mode, contents) in enumerate(ss.get_shares()):
 128             if mode == 'share':
 129                 if contents.hash in tracker.shares:
 130                     continue
 131                 contents.shared = True
 132                 contents.stored = True
 133                 contents.time_seen = 0
 134                 tracker.add(contents)
 135                 if len(tracker.shares) % 1000 == 0 and tracker.shares:
 136                     print "    %i" % (len(tracker.shares),)
 137             elif mode == 'verified_hash':
 138                 known_verified.add(contents)
 139             else:
 140                 raise AssertionError()
 141         print "    ...inserting %i verified shares..." % (len(known_verified),)
 142         for h in known_verified:
 143             if h not in tracker.shares:
 144                 ss.forget_verified_share(h)
 145                 continue
 146             tracker.verified.add(tracker.shares[h])
 147         print "    ...done loading %i shares!" % (len(tracker.shares),)
 148         print
 149         tracker.added.watch(lambda share: ss.add_share(share))
 150         tracker.verified.added.watch(lambda share: ss.add_verified_hash(share.hash))
 151         tracker.removed.watch(lambda share: ss.forget_share(share.hash))
 152         tracker.verified.removed.watch(lambda share: ss.forget_verified_share(share.hash))
 153
 154         peer_heads = expiring_dict.ExpiringDict(300) # hash -> peers that know of it
 155
 156         # information affecting work that should trigger a long-polling update
 157         current_work = variable.Variable(None)
 158         # information affecting work that should not trigger a long-polling update
 159         current_work2 = variable.Variable(None)
 160
 161         work_updated = variable.Event()
 162
 163         requested = expiring_dict.ExpiringDict(300)
 164
 165         @defer.inlineCallbacks
 166         def set_real_work1():
 167             work = yield getwork(bitcoind, ht, args.net)
 168             changed = work['previous_block_hash'] != current_work.value['previous_block'] if current_work.value is not None else True
 169             current_work.set(dict(
 170                 version=work['version'],
 171                 previous_block=work['previous_block_hash'],
 172                 target=work['target'],
 173                 best_share_hash=current_work.value['best_share_hash'] if current_work.value is not None else None,
 174                 aux_work=current_work.value['aux_work'] if current_work.value is not None else None,
 175             ))
 176             current_work2.set(dict(
 177                 transactions=work['transactions'],
 178                 subsidy=work['subsidy'],
 179                 clock_offset=time.time() - work['time'],
 180                 last_update=time.time(),
 181             ))
 182             if changed:
 183                 set_real_work2()
 184
 185         def set_real_work2():
 186             best, desired = tracker.think(ht, current_work.value['previous_block'], time.time() - current_work2.value['clock_offset'])
 187
 188             t = dict(current_work.value)
 189             t['best_share_hash'] = best
 190             current_work.set(t)
 191
 192             t = time.time()
 193             for peer2, share_hash in desired:
 194                 if share_hash not in tracker.tails: # was received in the time tracker.think was running
 195                     continue
 196                 last_request_time, count = requested.get(share_hash, (None, 0))
 197                 if last_request_time is not None and last_request_time - 5 < t < last_request_time + 10 * 1.5**count:
 198                     continue
 199                 potential_peers = set()
 200                 for head in tracker.tails[share_hash]:
 201                     potential_peers.update(peer_heads.get(head, set()))
 202                 potential_peers = [peer for peer in potential_peers if peer.connected2]
 203                 if count == 0 and peer2 is not None and peer2.connected2:
 204                     peer = peer2
 205                 else:
 206                     peer = random.choice(potential_peers) if potential_peers and random.random() > .2 else peer2
 207                     if peer is None:
 208                         continue
 209
 210                 print 'Requesting parent share %s from %s' % (p2pool.format_hash(share_hash), '%s:%i' % peer.addr)
 211                 peer.send_getshares(
 212                     hashes=[share_hash],
 213                     parents=2000,
 214                     stops=list(set(tracker.heads) | set(
 215                         tracker.get_nth_parent_hash(head, min(max(0, tracker.get_height_and_last(head)[0] - 1), 10)) for head in tracker.heads
 216                     ))[:100],
 217                 )
 218                 requested[share_hash] = t, count + 1
 219
 220         print 'Initializing work...'
 221         yield set_real_work1()
 222         set_real_work2()
 223         print '    ...success!'
 224         print
 225
 226         @defer.inlineCallbacks
 227         def set_merged_work():
 228             if not args.merged_url:
 229                 return
 230             while True:
 231                 merged = jsonrpc.Proxy(args.merged_url, (args.merged_userpass,))
 232                 auxblock = yield deferral.retry('Error while calling merged getauxblock:', 1)(merged.rpc_getauxblock)()
 233                 x = dict(current_work.value)
 234                 x['aux_work'] = dict(
 235                     hash=int(auxblock['hash'], 16),
 236                     target=bitcoin.data.HashType().unpack(auxblock['target'].decode('hex')),
 237                     chain_id=auxblock['chainid'],
 238                 )
 239                 #print x['aux_work']
 240                 current_work.set(x)
 241                 yield deferral.sleep(1)
 242         set_merged_work()
 243
 244         start_time = time.time() - current_work2.value['clock_offset']
 245
 246         # setup p2p logic and join p2pool network
 247
 248         def share_share(share, ignore_peer=None):
 249             for peer in p2p_node.peers.itervalues():
 250                 if peer is ignore_peer:
 251                     continue
 252                 #if p2pool_init.DEBUG:
 253                 #    print "Sending share %s to %r" % (p2pool.format_hash(share.hash), peer.addr)
 254                 peer.sendShares([share])
 255             share.flag_shared()
 256
 257         def p2p_shares(shares, peer=None):
 258             if len(shares) > 5:
 259                 print 'Processing %i shares...' % (len(shares),)
 260
 261             new_count = 0
 262             for share in shares:
 263                 if share.hash in tracker.shares:
 264                     #print 'Got duplicate share, ignoring. Hash: %s' % (p2pool.format_hash(share.hash),)
 265                     continue
 266
 267                 new_count += 1
 268
 269                 #print 'Received share %s from %r' % (p2pool.format_hash(share.hash), share.peer.addr if share.peer is not None else None)
 270
 271                 tracker.add(share)
 272
 273             if shares and peer is not None:
 274                 peer_heads.setdefault(shares[0].hash, set()).add(peer)
 275
 276             if new_count:
 277                 set_real_work2()
 278
 279             if len(shares) > 5:
 280                 print '... done processing %i shares. New: %i Have: %i/~%i' % (len(shares), new_count, len(tracker.shares), 2*args.net.CHAIN_LENGTH)
 281
 282         @tracker.verified.added.watch
 283         def _(share):
 284             if share.bitcoin_hash <= share.header['target']:
 285                 print
 286                 print 'GOT BLOCK! Passing to bitcoind! %s bitcoin: %x' % (p2pool.format_hash(share.hash), share.bitcoin_hash,)
 287                 print
 288                 if factory.conn.value is not None:
 289                     factory.conn.value.send_block(block=share.as_block(tracker, args.net))
 290                 else:
 291                     print 'No bitcoind connection! Erp!'
 292
 293         def p2p_share_hashes(share_hashes, peer):
 294             t = time.time()
 295             get_hashes = []
 296             for share_hash in share_hashes:
 297                 if share_hash in tracker.shares:
 298                     continue
 299                 last_request_time, count = requested.get(share_hash, (None, 0))
 300                 if last_request_time is not None and last_request_time - 5 < t < last_request_time + 10 * 1.5**count:
 301                     continue
 302                 print 'Got share hash, requesting! Hash: %s' % (p2pool.format_hash(share_hash),)
 303                 get_hashes.append(share_hash)
 304                 requested[share_hash] = t, count + 1
 305
 306             if share_hashes and peer is not None:
 307                 peer_heads.setdefault(share_hashes[0], set()).add(peer)
 308             if get_hashes:
 309                 peer.send_getshares(hashes=get_hashes, parents=0, stops=[])
 310
 311         def p2p_get_shares(share_hashes, parents, stops, peer):
 312             parents = min(parents, 1000//len(share_hashes))
 313             stops = set(stops)
 314             shares = []
 315             for share_hash in share_hashes:
 316                 for share in itertools.islice(tracker.get_chain_known(share_hash), parents + 1):
 317                     if share.hash in stops:
 318                         break
 319                     shares.append(share)
 320             print 'Sending %i shares to %s:%i' % (len(shares), peer.addr[0], peer.addr[1])
 321             peer.sendShares(shares, full=True)
 322
 323         print 'Joining p2pool network using TCP port %i...' % (args.p2pool_port,)
 324
 325         def parse(x):
 326             if ':' in x:
 327                 ip, port = x.split(':')
 328                 return ip, int(port)
 329             else:
 330                 return x, args.net.P2P_PORT
 331
 332         nodes = set([
 333             ('72.14.191.28', args.net.P2P_PORT),
 334             ('62.204.197.159', args.net.P2P_PORT),
 335             ('142.58.248.28', args.net.P2P_PORT),
 336             ('94.23.34.145', args.net.P2P_PORT),
 337         ])
 338         for host in [
 339             'p2pool.forre.st',
 340             'dabuttonfactory.com',
 341         ]:
 342             try:
 343                 nodes.add(((yield reactor.resolve(host)), args.net.P2P_PORT))
 344             except:
 345                 log.err(None, 'Error resolving bootstrap node IP:')
 346
 347         if args.net_name == 'litecoin':
 348             nodes.add(((yield reactor.resolve('liteco.in')), args.net.P2P_PORT))
 349
 350         p2p_node = p2p.Node(
 351             current_work=current_work,
 352             port=args.p2pool_port,
 353             net=args.net,
 354             addr_store=db.SQLiteDict(sqlite3.connect(os.path.join(os.path.dirname(sys.argv[0]), 'addrs.dat'), isolation_level=None), args.net.NAME),
 355             mode=0 if args.low_bandwidth else 1,
 356             preferred_addrs=set(map(parse, args.p2pool_nodes)) | nodes,
 357         )
 358         p2p_node.handle_shares = p2p_shares
 359         p2p_node.handle_share_hashes = p2p_share_hashes
 360         p2p_node.handle_get_shares = p2p_get_shares
 361
 362         p2p_node.start()
 363
 364         # send share when the chain changes to their chain
 365         def work_changed(new_work):
 366             #print 'Work changed:', new_work
 367             for share in tracker.get_chain_known(new_work['best_share_hash']):
 368                 if share.shared:
 369                     break
 370                 share_share(share, share.peer)
 371         current_work.changed.watch(work_changed)
 372
 373         print '    ...success!'
 374         print
 375
 376         @defer.inlineCallbacks
 377         def upnp_thread():
 378             while True:
 379                 try:
 380                     is_lan, lan_ip = yield ipdiscover.get_local_ip()
 381                     if is_lan:
 382                         pm = yield portmapper.get_port_mapper()
 383                         yield pm._upnp.add_port_mapping(lan_ip, args.p2pool_port, args.p2pool_port, 'p2pool', 'TCP') # XXX try to forward external correct port?
 384                 except defer.TimeoutError:
 385                     pass
 386                 except:
 387                     if p2pool_init.DEBUG:
 388                         log.err(None, "UPnP error:")
 389                 yield deferral.sleep(random.expovariate(1/120))
 390
 391         if args.upnp:
 392             upnp_thread()
 393
 394         # start listening for workers with a JSON-RPC server
 395
 396         print 'Listening for workers on port %i...' % (args.worker_port,)
 397
 398         # setup worker logic
 399
 400         merkle_root_to_transactions = expiring_dict.ExpiringDict(300)
 401         run_identifier = struct.pack('<I', random.randrange(2**32))
 402
 403         share_counter = skiplists.CountsSkipList(tracker, run_identifier)
 404         removed_unstales = set()
 405         def get_share_counts(doa=False):
 406             height, last = tracker.get_height_and_last(current_work.value['best_share_hash'])
 407             matching_in_chain = share_counter(current_work.value['best_share_hash'], max(0, height - 1)) | removed_unstales
 408             shares_in_chain = my_shares & matching_in_chain
 409             stale_shares = my_shares - matching_in_chain
 410             if doa:
 411                 stale_doa_shares = stale_shares & doa_shares
 412                 stale_not_doa_shares = stale_shares - stale_doa_shares
 413                 return len(shares_in_chain) + len(stale_shares), len(stale_doa_shares), len(stale_not_doa_shares)
 414             return len(shares_in_chain) + len(stale_shares), len(stale_shares)
 415         @tracker.verified.removed.watch
 416         def _(share):
 417             if share.hash in my_shares and tracker.is_child_of(share.hash, current_work.value['best_share_hash']):
 418                 removed_unstales.add(share.hash)
 419
 420         def compute(state, payout_script):
 421             if payout_script is None or random.uniform(0, 100) < args.worker_fee:
 422                 payout_script = my_script
 423             if state['best_share_hash'] is None and args.net.PERSIST:
 424                 raise jsonrpc.Error(-12345, u'p2pool is downloading shares')
 425             if len(p2p_node.peers) == 0 and args.net.PERSIST:
 426                 raise jsonrpc.Error(-12345, u'p2pool is not connected to any peers')
 427             if time.time() > current_work2.value['last_update'] + 60:
 428                 raise jsonrpc.Error(-12345, u'lost contact with bitcoind')
 429
 430             if state['aux_work'] is not None:
 431                 aux_str = '\xfa\xbemm' + bitcoin.data.HashType().pack(state['aux_work']['hash'])[::-1] + struct.pack('<ii', 1, 0)
 432             else:
 433                 aux_str = ''
 434
 435             # XXX assuming generate_tx is smallish here..
 436             def get_stale_frac():
 437                 shares, stale_shares = get_share_counts()
 438                 if shares == 0:
 439                     return ""
 440                 frac = stale_shares/shares
 441                 return 2*struct.pack('<H', int(65535*frac + .5))
 442             subsidy = current_work2.value['subsidy']
 443
 444             timestamp = int(time.time() - current_work2.value['clock_offset'])
 445             if state['best_share_hash'] is not None:
 446                 timestamp2 = math.median((s.timestamp for s in itertools.islice(tracker.get_chain_to_root(state['best_share_hash']), 11)), use_float=False) + 1
 447                 if timestamp2 > timestamp:
 448                     print 'Toff', timestamp2 - timestamp
 449                     timestamp = timestamp2
 450
 451             if timestamp > 42e2:
 452                 is_new = True
 453                 new_share_info, generate_tx = p2pool.new_generate_transaction(
 454                     tracker=tracker,
 455                     new_share_data=dict(
 456                         previous_share_hash=state['best_share_hash'],
 457                         coinbase=aux_str,
 458                         nonce=run_identifier + struct.pack('<Q', random.randrange(2**64)) + get_stale_frac(),
 459                         new_script=payout_script,
 460                         subsidy=subsidy,
 461                         donation=math.perfect_round(65535*args.donation_percentage/100),
 462                     ),
 463                     block_target=state['target'],
 464                     net=args.net,
 465                 )
 466             else:
 467                 is_new = False
 468                 share_info, generate_tx = p2pool.generate_transaction(
 469                     tracker=tracker,
 470                     previous_share_hash=state['best_share_hash'],
 471                     new_script=payout_script,
 472                     subsidy=subsidy,
 473                     nonce=run_identifier + struct.pack('<H', random.randrange(2**16)) + aux_str + get_stale_frac(),
 474                     block_target=state['target'],
 475                     net=args.net,
 476                 )
 477
 478             print 'New work for worker! Difficulty: %.06f Payout if block: %.6f %s Total block value: %.6f %s including %i transactions' % (bitcoin.data.target_to_difficulty((new_share_info if is_new else share_info)['target']), (generate_tx['tx_outs'][-1]['value']-subsidy//200)*1e-8, args.net.BITCOIN_SYMBOL, subsidy*1e-8, args.net.BITCOIN_SYMBOL, len(current_work2.value['transactions']))
 479             #print 'Target: %x' % (p2pool.coinbase_type.unpack(generate_tx['tx_ins'][0]['script'])['share_data']['target'],)
 480             #, have', shares.count(my_script) - 2, 'share(s) in the current chain. Fee:', sum(tx.value_in - tx.value_out for tx in extra_txs)/100000000
 481             transactions = [generate_tx] + list(current_work2.value['transactions'])
 482             merkle_root = bitcoin.data.merkle_hash(transactions)
 483             merkle_root_to_transactions[merkle_root] = is_new, new_share_info if is_new else share_info, transactions
 484
 485             target2 = (new_share_info if is_new else share_info)['target']
 486             times[merkle_root] = time.time()
 487             #print 'SENT', 2**256//p2pool.coinbase_type.unpack(generate_tx['tx_ins'][0]['script'])['share_data']['target']
 488             return bitcoin.getwork.BlockAttempt(state['version'], state['previous_block'], merkle_root, timestamp, state['target'], target2)
 489
 490         my_shares = set()
 491         doa_shares = set()
 492         times = {}
 493
 494         def got_response(data, user):
 495             try:
 496                 # match up with transactions
 497                 header = bitcoin.getwork.decode_data(data)
 498                 xxx = merkle_root_to_transactions.get(header['merkle_root'], None)
 499                 if xxx is None:
 500                     print '''Couldn't link returned work's merkle root with its transactions - should only happen if you recently restarted p2pool'''
 501                     return False
 502                 is_new, share_info, transactions = xxx
 503                 new_share_info = share_info
 504
 505                 hash_ = bitcoin.data.block_header_type.hash256(header)
 506
 507                 pow = bitcoin.data.block_header_type.scrypt(header) if getattr(args.net, 'BITCOIN_POW_SCRYPT', False) else hash_
 508
 509                 if pow <= header['target'] or p2pool_init.DEBUG:
 510                     if factory.conn.value is not None:
 511                         factory.conn.value.send_block(block=dict(header=header, txs=transactions))
 512                     else:
 513                         print 'No bitcoind connection! Erp!'
 514                     if pow <= header['target']:
 515                         print
 516                         print 'GOT BLOCK! Passing to bitcoind! bitcoin: %x' % (hash_,)
 517                         print
 518
 519                 if current_work.value['aux_work'] is not None and pow <= current_work.value['aux_work']['target']:
 520                     try:
 521                         aux_pow = dict(
 522                             merkle_tx=dict(
 523                                 tx=transactions[0],
 524                                 block_hash=hash_,
 525                                 merkle_branch=[x['hash'] for x in p2pool.calculate_merkle_branch(transactions, 0)],
 526                                 index=0,
 527                             ),
 528                             merkle_branch=[],
 529                             index=0,
 530                             parent_block_header=header,
 531                         )
 532
 533                         a, b = transactions[0]['tx_ins'][0]['script'][-32-8:-8].encode('hex'), bitcoin.data.aux_pow_type.pack(aux_pow).encode('hex')
 534                         #print a, b
 535                         merged = jsonrpc.Proxy(args.merged_url, (args.merged_userpass,))
 536                         def _(res):
 537                             print "MERGED RESULT:", res
 538                         merged.rpc_getauxblock(a, b).addBoth(_)
 539                     except:
 540                         log.err(None, 'Error while processing merged mining POW:')
 541
 542                 target = new_share_info['target']
 543                 if pow > target:
 544                     print 'Worker submitted share with hash > target:\nhash  : %x\ntarget: %x' % (pow, target)
 545                     return False
 546                 if is_new:
 547                     share = p2pool.NewShare(args.net, header, new_share_info, other_txs=transactions[1:])
 548                 else:
 549                     share = p2pool.Share(args.net, header, share_info, other_txs=transactions[1:])
 550                 my_shares.add(share.hash)
 551                 if share.previous_hash != current_work.value['best_share_hash']:
 552                     doa_shares.add(share.hash)
 553                 print 'GOT SHARE! %s %s prev %s age %.2fs' % (user, p2pool.format_hash(share.hash), p2pool.format_hash(share.previous_hash), time.time() - times[header['merkle_root']]) + (' DEAD ON ARRIVAL' if share.previous_hash != current_work.value['best_share_hash'] else '')
 554                 good = share.previous_hash == current_work.value['best_share_hash']
 555                 # maybe revert back to tracker being non-blocking so 'good' can be more accurate?
 556                 p2p_shares([share])
 557                 # eg. good = share.hash == current_work.value['best_share_hash'] here
 558                 return good
 559             except:
 560                 log.err(None, 'Error processing data received from worker:')
 561                 return False
 562
 563         web_root = worker_interface.WorkerInterface(current_work, compute, got_response, args.net)
 564
 565         def get_rate():
 566             if current_work.value['best_share_hash'] is not None:
 567                 height, last = tracker.get_height_and_last(current_work.value['best_share_hash'])
 568                 att_s = p2pool.get_pool_attempts_per_second(tracker, current_work.value['best_share_hash'], args.net, min(height - 1, 720))
 569                 fracs = [read_stale_frac(share) for share in itertools.islice(tracker.get_chain_known(current_work.value['best_share_hash']), 120) if read_stale_frac(share) is not None]
 570                 return json.dumps(int(att_s / (1. - (math.median(fracs) if fracs else 0))))
 571             return json.dumps(None)
 572
 573         def get_users():
 574             height, last = tracker.get_height_and_last(current_work.value['best_share_hash'])
 575             weights, total_weight, donation_weight = tracker.get_cumulative_weights(current_work.value['best_share_hash'], min(height, 720), 65535*2**256)
 576             res = {}
 577             for script in sorted(weights, key=lambda s: weights[s]):
 578                 res[bitcoin.data.script2_to_human(script, args.net)] = weights[script]/total_weight
 579             return json.dumps(res)
 580
 581         class WebInterface(resource.Resource):
 582             def __init__(self, func, mime_type):
 583                 self.func, self.mime_type = func, mime_type
 584
 585             def render_GET(self, request):
 586                 request.setHeader('Content-Type', self.mime_type)
 587                 return self.func()
 588
 589         web_root.putChild('rate', WebInterface(get_rate, 'application/json'))
 590         web_root.putChild('users', WebInterface(get_users, 'application/json'))
 591         web_root.putChild('fee', WebInterface(lambda: json.dumps(args.worker_fee), 'application/json'))
 592         if draw is not None:
 593             web_root.putChild('chain_img', WebInterface(lambda: draw.get(tracker, current_work.value['best_share_hash']), 'image/png'))
 594
 595         reactor.listenTCP(args.worker_port, server.Site(web_root))
 596
 597         print '    ...success!'
 598         print
 599
 600         # done!
 601
 602         # do new getwork when a block is heard on the p2p interface
 603
 604         def new_block(block_hash):
 605             work_updated.happened()
 606         factory.new_block.watch(new_block)
 607
 608         print 'Started successfully!'
 609         print
 610
 611         ht.updated.watch(set_real_work2)
 612
 613         @defer.inlineCallbacks
 614         def work1_thread():
 615             while True:
 616                 flag = work_updated.get_deferred()
 617                 try:
 618                     yield set_real_work1()
 619                 except:
 620                     log.err()
 621                 yield defer.DeferredList([flag, deferral.sleep(random.uniform(1, 10))], fireOnOneCallback=True)
 622
 623         @defer.inlineCallbacks
 624         def work2_thread():
 625             while True:
 626                 try:
 627                     set_real_work2()
 628                 except:
 629                     log.err()
 630                 yield deferral.sleep(random.expovariate(1/20))
 631
 632         work1_thread()
 633         work2_thread()
 634
 635
 636         if hasattr(signal, 'SIGALRM'):
 637             def watchdog_handler(signum, frame):
 638                 print 'Watchdog timer went off at:'
 639                 traceback.print_stack()
 640
 641             signal.signal(signal.SIGALRM, watchdog_handler)
 642             task.LoopingCall(signal.alarm, 30).start(1)
 643
 644
 645         def read_stale_frac(share):
 646             if len(share.nonce) < 4:
 647                 return None
 648             a, b = struct.unpack("<HH", share.nonce[-4:])
 649             if a == 0 or a != b:
 650                 return None
 651             return a/65535
 652
 653         pool_str = None;
 654         while True:
 655             yield deferral.sleep(3)
 656             try:
 657                 if time.time() > current_work2.value['last_update'] + 60:
 658                     print '''---> LOST CONTACT WITH BITCOIND for 60 seconds, check that it isn't frozen or dead <---'''
 659                 if current_work.value['best_share_hash'] is not None:
 660                     height, last = tracker.get_height_and_last(current_work.value['best_share_hash'])
 661                     if height > 2:
 662                         att_s = p2pool.get_pool_attempts_per_second(tracker, current_work.value['best_share_hash'], args.net, min(height - 1, 720))
 663                         weights, total_weight, donation_weight = tracker.get_cumulative_weights(current_work.value['best_share_hash'], min(height, 720), 65535*2**256)
 664                         shares, stale_doa_shares, stale_not_doa_shares = get_share_counts(True)
 665                         stale_shares = stale_doa_shares + stale_not_doa_shares
 666                         fracs = [read_stale_frac(share) for share in itertools.islice(tracker.get_chain_known(current_work.value['best_share_hash']), 120) if read_stale_frac(share) is not None]
 667                         str = 'Pool: %sH/s in %i shares (%i/%i verified) Recent: %.02f%% >%sH/s Shares: %i (%i orphan, %i dead) Peers: %i' % (
 668                             math.format(int(att_s / (1. - (math.median(fracs) if fracs else 0)))),
 669                             height,
 670                             len(tracker.verified.shares),
 671                             len(tracker.shares),
 672                             weights.get(my_script, 0)/total_weight*100,
 673                             math.format(int(weights.get(my_script, 0)*att_s//total_weight / (1. - (math.median(fracs) if fracs else 0)))),
 674                             shares,
 675                             stale_not_doa_shares,
 676                             stale_doa_shares,
 677                             len(p2p_node.peers),
 678                         ) + (' FDs: %i R/%i W' % (len(reactor.getReaders()), len(reactor.getWriters())) if p2pool_init.DEBUG else '')
 679                         if (str != pool_str):
 680                             print str;
 681                             pool_str = str;
 682                         if fracs:
 683                             med = math.median(fracs)
 684                             print 'Median stale proportion:', med
 685                             if shares:
 686                                 print '    Own:', stale_shares/shares
 687                                 if med < .99:
 688                                     print '    Own efficiency: %.02f%%' % (100*(1 - stale_shares/shares)/(1 - med),)
 689
 690
 691             except:
 692                 log.err()
 693     except:
 694         log.err(None, 'Fatal error:')
 695     finally:
 696         reactor.stop()
 697
 698 def run():
 699     class FixedArgumentParser(argparse.ArgumentParser):
 700         def _read_args_from_files(self, arg_strings):
 701             # expand arguments referencing files
 702             new_arg_strings = []
 703             for arg_string in arg_strings:
 704
 705                 # for regular arguments, just add them back into the list
 706                 if not arg_string or arg_string[0] not in self.fromfile_prefix_chars:
 707                     new_arg_strings.append(arg_string)
 708
 709                 # replace arguments referencing files with the file content
 710                 else:
 711                     try:
 712                         args_file = open(arg_string[1:])
 713                         try:
 714                             arg_strings = []
 715                             for arg_line in args_file.read().splitlines():
 716                                 for arg in self.convert_arg_line_to_args(arg_line):
 717                                     arg_strings.append(arg)
 718                             arg_strings = self._read_args_from_files(arg_strings)
 719                             new_arg_strings.extend(arg_strings)
 720                         finally:
 721                             args_file.close()
 722                     except IOError:
 723                         err = sys.exc_info()[1]
 724                         self.error(str(err))
 725
 726             # return the modified argument list
 727             return new_arg_strings
 728
 729         def convert_arg_line_to_args(self, arg_line):
 730             return [arg for arg in arg_line.split() if arg.strip()]
 731
 732     parser = FixedArgumentParser(description='p2pool (version %s)' % (p2pool_init.__version__,), fromfile_prefix_chars='@')
 733     parser.add_argument('--version', action='version', version=p2pool_init.__version__)
 734     parser.add_argument('--net',
 735         help='use specified network (default: bitcoin)',
 736         action='store', choices=sorted(x for x in p2pool.nets if 'testnet' not in x), default='bitcoin', dest='net_name')
 737     parser.add_argument('--testnet',
 738         help='''use the network's testnet''',
 739         action='store_const', const=True, default=False, dest='testnet')
 740     parser.add_argument('--debug',
 741         help='debugging mode',
 742         action='store_const', const=True, default=False, dest='debug')
 743     parser.add_argument('-a', '--address',
 744         help='generate to this address (defaults to requesting one from bitcoind)',
 745         type=str, action='store', default=None, dest='address')
 746     parser.add_argument('--logfile',
 747         help='''log to specific file (defaults to <network_name>.log in run_p2pool.py's directory)''',
 748         type=str, action='store', default=None, dest='logfile')
 749     parser.add_argument('--merged-url',
 750         help='call getauxblock on this url to get work for merged mining',
 751         type=str, action='store', default=None, dest='merged_url')
 752     parser.add_argument('--merged-userpass',
 753         help='merge daemon user and password, separated by a colon. Example: ncuser:ncpass',
 754         type=str, action='store', default=None, dest='merged_userpass')
 755     parser.add_argument('--give-author', metavar='DONATION_PERCENTAGE',
 756         help='percentage amount to donate to author of p2pool. Default: 0.5',
 757         type=float, action='store', default=0.5, dest='donation_percentage')
 758
 759     p2pool_group = parser.add_argument_group('p2pool interface')
 760     p2pool_group.add_argument('--p2pool-port', metavar='PORT',
 761         help='use TCP port PORT to listen for connections (default: 9333 normally, 19333 for testnet) (forward this port from your router!)',
 762         type=int, action='store', default=None, dest='p2pool_port')
 763     p2pool_group.add_argument('-n', '--p2pool-node', metavar='ADDR[:PORT]',
 764         help='connect to existing p2pool node at ADDR listening on TCP port PORT (defaults to 9333 normally, 19333 for testnet), in addition to builtin addresses',
 765         type=str, action='append', default=[], dest='p2pool_nodes')
 766     parser.add_argument('-l', '--low-bandwidth',
 767         help='trade lower bandwidth usage for higher latency (reduced efficiency)',
 768         action='store_true', default=False, dest='low_bandwidth')
 769     parser.add_argument('--disable-upnp',
 770         help='''don't attempt to forward port 9333 (19333 for testnet) from the WAN to this computer using UPnP''',
 771         action='store_false', default=True, dest='upnp')
 772
 773     worker_group = parser.add_argument_group('worker interface')
 774     worker_group.add_argument('-w', '--worker-port', metavar='PORT',
 775         help='listen on PORT for RPC connections from miners asking for work and providing responses (default: bitcoin: 9332 namecoin: 9331 ixcoin: 9330 i0coin: 9329 solidcoin: 9328 litecoin: 9327, +10000 for testnets)',
 776         type=int, action='store', default=None, dest='worker_port')
 777     worker_group.add_argument('-f', '--fee', metavar='FEE_PERCENTAGE',
 778         help='''charge workers mining to their own bitcoin address (by setting their miner's username to a bitcoin address) this percentage fee to mine on your p2pool instance. Amount displayed at http://127.0.0.1:9332/fee . default: 0''',
 779         type=float, action='store', default=0, dest='worker_fee')
 780
 781     bitcoind_group = parser.add_argument_group('bitcoind interface')
 782     bitcoind_group.add_argument('--bitcoind-address', metavar='BITCOIND_ADDRESS',
 783         help='connect to a bitcoind at this address (default: 127.0.0.1)',
 784         type=str, action='store', default='127.0.0.1', dest='bitcoind_address')
 785     bitcoind_group.add_argument('--bitcoind-rpc-port', metavar='BITCOIND_RPC_PORT',
 786         help='connect to a bitcoind at this port over the RPC interface - used to get the current highest block via getwork (default: 8332 ixcoin: 8338 i0coin: 7332 litecoin: 9332)',
 787         type=int, action='store', default=None, dest='bitcoind_rpc_port')
 788     bitcoind_group.add_argument('--bitcoind-p2p-port', metavar='BITCOIND_P2P_PORT',
 789         help='connect to a bitcoind at this port over the p2p interface - used to submit blocks and get the pubkey to generate to via an IP transaction (default: 8333 namecoin: 8334 ixcoin: 8337 i0coin: 7333 solidcoin: 7555 litecoin: 9333, +10000 for testnets)',
 790         type=int, action='store', default=None, dest='bitcoind_p2p_port')
 791
 792     bitcoind_group.add_argument(metavar='BITCOIND_RPCUSER',
 793         help='bitcoind RPC interface username (default: empty)',
 794         type=str, action='store', default='', nargs='?', dest='bitcoind_rpc_username')
 795     bitcoind_group.add_argument(metavar='BITCOIND_RPCPASSWORD',
 796         help='bitcoind RPC interface password',
 797         type=str, action='store', dest='bitcoind_rpc_password')
 798
 799     args = parser.parse_args()
 800
 801     if args.debug:
 802         p2pool_init.DEBUG = True
 803
 804     if args.logfile is None:
 805         args.logfile = os.path.join(os.path.dirname(sys.argv[0]), args.net_name + ('_testnet' if args.testnet else '') + '.log')
 806
 807     class LogFile(object):
 808         def __init__(self, filename):
 809             self.filename = filename
 810             self.inner_file = None
 811             self.reopen()
 812         def reopen(self):
 813             if self.inner_file is not None:
 814                 self.inner_file.close()
 815             open(self.filename, 'a').close()
 816             f = open(self.filename, 'rb')
 817             f.seek(0, os.SEEK_END)
 818             length = f.tell()
 819             if length > 100*1000*1000:
 820                 f.seek(-1000*1000, os.SEEK_END)
 821                 while True:
 822                     if f.read(1) in ('', '\n'):
 823                         break
 824                 data = f.read()
 825                 f.close()
 826                 f = open(self.filename, 'wb')
 827                 f.write(data)
 828             f.close()
 829             self.inner_file = open(self.filename, 'a')
 830         def write(self, data):
 831             self.inner_file.write(data)
 832         def flush(self):
 833             self.inner_file.flush()
 834     class TeePipe(object):
 835         def __init__(self, outputs):
 836             self.outputs = outputs
 837         def write(self, data):
 838             for output in self.outputs:
 839                 output.write(data)
 840         def flush(self):
 841             for output in self.outputs:
 842                 output.flush()
 843     class TimestampingPipe(object):
 844         def __init__(self, inner_file):
 845             self.inner_file = inner_file
 846             self.buf = ''
 847             self.softspace = 0
 848         def write(self, data):
 849             buf = self.buf + data
 850             lines = buf.split('\n')
 851             for line in lines[:-1]:
 852                 self.inner_file.write('%s %s\n' % (datetime.datetime.now().strftime("%H:%M:%S.%f"), line))
 853                 self.inner_file.flush()
 854             self.buf = lines[-1]
 855         def flush(self):
 856             pass
 857     logfile = LogFile(args.logfile)
 858     sys.stdout = sys.stderr = log.DefaultObserver.stderr = TimestampingPipe(TeePipe([sys.stderr, logfile]))
 859     if hasattr(signal, "SIGUSR1"):
 860         def sigusr1(signum, frame):
 861             print 'Caught SIGUSR1, closing %r...' % (args.logfile,)
 862             logfile.reopen()
 863             print '...and reopened %r after catching SIGUSR1.' % (args.logfile,)
 864         signal.signal(signal.SIGUSR1, sigusr1)
 865     task.LoopingCall(logfile.reopen).start(5)
 866
 867     args.net = p2pool.nets[args.net_name + ('_testnet' if args.testnet else '')]
 868
 869     if args.bitcoind_rpc_port is None:
 870         args.bitcoind_rpc_port = args.net.BITCOIN_RPC_PORT
 871
 872     if args.bitcoind_p2p_port is None:
 873         args.bitcoind_p2p_port = args.net.BITCOIN_P2P_PORT
 874
 875     if args.p2pool_port is None:
 876         args.p2pool_port = args.net.P2P_PORT
 877
 878     if args.worker_port is None:
 879         args.worker_port = args.net.WORKER_PORT
 880
 881     if args.address is not None:
 882         try:
 883             args.pubkey_hash = bitcoin.data.address_to_pubkey_hash(args.address, args.net)
 884         except Exception, e:
 885             parser.error('error parsing address: ' + repr(e))
 886     else:
 887         args.pubkey_hash = None
 888
 889     if (args.merged_url is None) ^ (args.merged_userpass is None):
 890         parser.error('must specify --merged-url and --merged-userpass')
 891
 892     reactor.callWhenRunning(main, args)
 893     reactor.run()