p2pool/main.py

   1 from __future__ import division
   2
   3 import ConfigParser
   4 import StringIO
   5 import base64
   6 import json
   7 import os
   8 import random
   9 import sys
  10 import time
  11 import signal
  12 import traceback
  13 import urlparse
  14
  15 if '--iocp' in sys.argv:
  16     from twisted.internet import iocpreactor
  17     iocpreactor.install()
  18 from twisted.internet import defer, reactor, protocol, task
  19 from twisted.web import server
  20 from twisted.python import log
  21 from nattraverso import portmapper, ipdiscover
  22
  23 import bitcoin.p2p as bitcoin_p2p, bitcoin.data as bitcoin_data
  24 from bitcoin import worker_interface, height_tracker
  25 from util import expiring_dict, fixargparse, jsonrpc, variable, deferral, math, logging
  26 from . import p2p, networks, web, work
  27 import p2pool, p2pool.data as p2pool_data
  28
  29 @deferral.retry('Error getting work from bitcoind:', 3)
  30 @defer.inlineCallbacks
  31 def getwork(bitcoind, use_getblocktemplate=False):
  32     def go():
  33         if use_getblocktemplate:
  34             return bitcoind.rpc_getblocktemplate({})
  35         else:
  36             return bitcoind.rpc_getmemorypool()
  37     try:
  38         work = yield go()
  39     except jsonrpc.Error_for_code(-32601): # Method not found
  40         use_getblocktemplate = not use_getblocktemplate
  41         try:
  42             work = yield go()
  43         except jsonrpc.Error_for_code(-32601): # Method not found
  44             print >>sys.stderr, 'Error: Bitcoin version too old! Upgrade to v0.5 or newer!'
  45             raise deferral.RetrySilentlyException()
  46     packed_transactions = [(x['data'] if isinstance(x, dict) else x).decode('hex') for x in work['transactions']]
  47     if 'height' not in work:
  48         work['height'] = (yield bitcoind.rpc_getblock(work['previousblockhash']))['height'] + 1
  49     elif p2pool.DEBUG:
  50         assert work['height'] == (yield bitcoind.rpc_getblock(work['previousblockhash']))['height'] + 1
  51     defer.returnValue(dict(
  52         version=work['version'],
  53         previous_block=int(work['previousblockhash'], 16),
  54         transactions=map(bitcoin_data.tx_type.unpack, packed_transactions),
  55         merkle_link=bitcoin_data.calculate_merkle_link([None] + map(bitcoin_data.hash256, packed_transactions), 0),
  56         subsidy=work['coinbasevalue'],
  57         time=work['time'] if 'time' in work else work['curtime'],
  58         bits=bitcoin_data.FloatingIntegerType().unpack(work['bits'].decode('hex')[::-1]) if isinstance(work['bits'], (str, unicode)) else bitcoin_data.FloatingInteger(work['bits']),
  59         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 '',
  60         height=work['height'],
  61         clock_offset=time.time() - (work['curtime'] if 'curtime' in work else work['time']),
  62         last_update=time.time(),
  63         use_getblocktemplate=use_getblocktemplate,
  64     ))
  65
  66 @defer.inlineCallbacks
  67 def main(args, net, datadir_path, merged_urls, worker_endpoint):
  68     try:
  69         print 'p2pool (version %s)' % (p2pool.__version__,)
  70         print
  71
  72         # connect to bitcoind over JSON-RPC and do initial getmemorypool
  73         url = 'http://%s:%i/' % (args.bitcoind_address, args.bitcoind_rpc_port)
  74         print '''Testing bitcoind RPC connection to '%s' with username '%s'...''' % (url, args.bitcoind_rpc_username)
  75         bitcoind = jsonrpc.Proxy(url, dict(Authorization='Basic ' + base64.b64encode(args.bitcoind_rpc_username + ':' + args.bitcoind_rpc_password)), timeout=30)
  76         @deferral.retry('Error while checking Bitcoin connection:', 1)
  77         @defer.inlineCallbacks
  78         def check():
  79             if not (yield net.PARENT.RPC_CHECK(bitcoind)):
  80                 print >>sys.stderr, "    Check failed! Make sure that you're connected to the right bitcoind with --bitcoind-rpc-port!"
  81                 raise deferral.RetrySilentlyException()
  82             temp_work = yield getwork(bitcoind)
  83             if not net.VERSION_CHECK((yield bitcoind.rpc_getinfo())['version'], temp_work):
  84                 print >>sys.stderr, '    Bitcoin version too old! BIP16 support required! Upgrade to 0.6.0rc4 or greater!'
  85                 raise deferral.RetrySilentlyException()
  86             defer.returnValue(temp_work)
  87         temp_work = yield check()
  88
  89         block_height_var = variable.Variable(None)
  90         @defer.inlineCallbacks
  91         def poll_height():
  92             block_height_var.set((yield deferral.retry('Error while calling getblockcount:')(bitcoind.rpc_getblockcount)()))
  93         yield poll_height()
  94         task.LoopingCall(poll_height).start(60*60)
  95
  96         bitcoind_warning_var = variable.Variable(None)
  97         @defer.inlineCallbacks
  98         def poll_warnings():
  99             errors = (yield deferral.retry('Error while calling getmininginfo:')(bitcoind.rpc_getmininginfo)())['errors']
 100             bitcoind_warning_var.set(errors if errors != '' else None)
 101         yield poll_warnings()
 102         task.LoopingCall(poll_warnings).start(20*60)
 103
 104         print '    ...success!'
 105         print '    Current block hash: %x' % (temp_work['previous_block'],)
 106         print '    Current block height: %i' % (block_height_var.value,)
 107         print
 108
 109         # connect to bitcoind over bitcoin-p2p
 110         print '''Testing bitcoind P2P connection to '%s:%s'...''' % (args.bitcoind_address, args.bitcoind_p2p_port)
 111         factory = bitcoin_p2p.ClientFactory(net.PARENT)
 112         reactor.connectTCP(args.bitcoind_address, args.bitcoind_p2p_port, factory)
 113         yield factory.getProtocol() # waits until handshake is successful
 114         print '    ...success!'
 115         print
 116
 117         print 'Determining payout address...'
 118         if args.pubkey_hash is None:
 119             address_path = os.path.join(datadir_path, 'cached_payout_address')
 120
 121             if os.path.exists(address_path):
 122                 with open(address_path, 'rb') as f:
 123                     address = f.read().strip('\r\n')
 124                 print '    Loaded cached address: %s...' % (address,)
 125             else:
 126                 address = None
 127
 128             if address is not None:
 129                 res = yield deferral.retry('Error validating cached address:', 5)(lambda: bitcoind.rpc_validateaddress(address))()
 130                 if not res['isvalid'] or not res['ismine']:
 131                     print '    Cached address is either invalid or not controlled by local bitcoind!'
 132                     address = None
 133
 134             if address is None:
 135                 print '    Getting payout address from bitcoind...'
 136                 address = yield deferral.retry('Error getting payout address from bitcoind:', 5)(lambda: bitcoind.rpc_getaccountaddress('p2pool'))()
 137
 138             with open(address_path, 'wb') as f:
 139                 f.write(address)
 140
 141             my_pubkey_hash = bitcoin_data.address_to_pubkey_hash(address, net.PARENT)
 142         else:
 143             my_pubkey_hash = args.pubkey_hash
 144         print '    ...success! Payout address:', bitcoin_data.pubkey_hash_to_address(my_pubkey_hash, net.PARENT)
 145         print
 146
 147         my_share_hashes = set()
 148         my_doa_share_hashes = set()
 149
 150         tracker = p2pool_data.OkayTracker(net, my_share_hashes, my_doa_share_hashes)
 151         shared_share_hashes = set()
 152         ss = p2pool_data.ShareStore(os.path.join(datadir_path, 'shares.'), net)
 153         known_verified = set()
 154         print "Loading shares..."
 155         for i, (mode, contents) in enumerate(ss.get_shares()):
 156             if mode == 'share':
 157                 if contents.hash in tracker.items:
 158                     continue
 159                 shared_share_hashes.add(contents.hash)
 160                 contents.time_seen = 0
 161                 tracker.add(contents)
 162                 if len(tracker.items) % 1000 == 0 and tracker.items:
 163                     print "    %i" % (len(tracker.items),)
 164             elif mode == 'verified_hash':
 165                 known_verified.add(contents)
 166             else:
 167                 raise AssertionError()
 168         print "    ...inserting %i verified shares..." % (len(known_verified),)
 169         for h in known_verified:
 170             if h not in tracker.items:
 171                 ss.forget_verified_share(h)
 172                 continue
 173             tracker.verified.add(tracker.items[h])
 174         print "    ...done loading %i shares!" % (len(tracker.items),)
 175         print
 176         tracker.removed.watch(lambda share: ss.forget_share(share.hash))
 177         tracker.verified.removed.watch(lambda share: ss.forget_verified_share(share.hash))
 178         tracker.removed.watch(lambda share: shared_share_hashes.discard(share.hash))
 179
 180         print 'Initializing work...'
 181
 182
 183         # BITCOIND WORK
 184
 185         bitcoind_work = variable.Variable((yield getwork(bitcoind)))
 186         @defer.inlineCallbacks
 187         def work_poller():
 188             while True:
 189                 flag = factory.new_block.get_deferred()
 190                 try:
 191                     bitcoind_work.set((yield getwork(bitcoind, bitcoind_work.value['use_getblocktemplate'])))
 192                 except:
 193                     log.err()
 194                 yield defer.DeferredList([flag, deferral.sleep(15)], fireOnOneCallback=True)
 195         work_poller()
 196
 197         # PEER WORK
 198
 199         best_block_header = variable.Variable(None)
 200         def handle_header(new_header):
 201             # check that header matches current target
 202             if not (net.PARENT.POW_FUNC(bitcoin_data.block_header_type.pack(new_header)) <= bitcoind_work.value['bits'].target):
 203                 return
 204             bitcoind_best_block = bitcoind_work.value['previous_block']
 205             if (best_block_header.value is None
 206                 or (
 207                     new_header['previous_block'] == bitcoind_best_block and
 208                     bitcoin_data.hash256(bitcoin_data.block_header_type.pack(best_block_header.value)) == bitcoind_best_block
 209                 ) # new is child of current and previous is current
 210                 or (
 211                     bitcoin_data.hash256(bitcoin_data.block_header_type.pack(new_header)) == bitcoind_best_block and
 212                     best_block_header.value['previous_block'] != bitcoind_best_block
 213                 )): # new is current and previous is not a child of current
 214                 best_block_header.set(new_header)
 215         @defer.inlineCallbacks
 216         def poll_header():
 217             handle_header((yield factory.conn.value.get_block_header(bitcoind_work.value['previous_block'])))
 218         bitcoind_work.changed.watch(lambda _: poll_header())
 219         yield deferral.retry('Error while requesting best block header:')(poll_header)()
 220
 221         # BEST SHARE
 222
 223         get_height_rel_highest = yield height_tracker.get_height_rel_highest_func(bitcoind, factory, lambda: bitcoind_work.value['previous_block'], net)
 224         requested = expiring_dict.ExpiringDict(300)
 225         peer_heads = expiring_dict.ExpiringDict(300) # hash -> peers that know of it
 226
 227         best_share_var = variable.Variable(None)
 228         def set_best_share():
 229             best, desired = tracker.think(get_height_rel_highest, bitcoind_work.value['previous_block'], bitcoind_work.value['bits'])
 230
 231             best_share_var.set(best)
 232
 233             t = time.time()
 234             for peer2, share_hash in desired:
 235                 if share_hash not in tracker.tails: # was received in the time tracker.think was running
 236                     continue
 237                 last_request_time, count = requested.get(share_hash, (None, 0))
 238                 if last_request_time is not None and last_request_time - 5 < t < last_request_time + 10 * 1.5**count:
 239                     continue
 240                 potential_peers = set()
 241                 for head in tracker.tails[share_hash]:
 242                     potential_peers.update(peer_heads.get(head, set()))
 243                 potential_peers = [peer for peer in potential_peers if peer.connected2]
 244                 if count == 0 and peer2 is not None and peer2.connected2:
 245                     peer = peer2
 246                 else:
 247                     peer = random.choice(potential_peers) if potential_peers and random.random() > .2 else peer2
 248                     if peer is None:
 249                         continue
 250
 251                 print 'Requesting parent share %s from %s' % (p2pool_data.format_hash(share_hash), '%s:%i' % peer.addr)
 252                 peer.send_getshares(
 253                     hashes=[share_hash],
 254                     parents=2000,
 255                     stops=list(set(tracker.heads) | set(
 256                         tracker.get_nth_parent_hash(head, min(max(0, tracker.get_height_and_last(head)[0] - 1), 10)) for head in tracker.heads
 257                     ))[:100],
 258                 )
 259                 requested[share_hash] = t, count + 1
 260         bitcoind_work.changed.watch(lambda _: set_best_share())
 261         set_best_share()
 262
 263
 264         print '    ...success!'
 265         print
 266
 267         # setup p2p logic and join p2pool network
 268
 269         class Node(p2p.Node):
 270             def handle_shares(self, shares, peer):
 271                 if len(shares) > 5:
 272                     print 'Processing %i shares from %s...' % (len(shares), '%s:%i' % peer.addr if peer is not None else None)
 273
 274                 new_count = 0
 275                 for share in shares:
 276                     if share.hash in tracker.items:
 277                         #print 'Got duplicate share, ignoring. Hash: %s' % (p2pool_data.format_hash(share.hash),)
 278                         continue
 279
 280                     new_count += 1
 281
 282                     #print 'Received share %s from %r' % (p2pool_data.format_hash(share.hash), share.peer.addr if share.peer is not None else None)
 283
 284                     tracker.add(share)
 285
 286                 if shares and peer is not None:
 287                     peer_heads.setdefault(shares[0].hash, set()).add(peer)
 288
 289                 if new_count:
 290                     set_best_share()
 291
 292                 if len(shares) > 5:
 293                     print '... done processing %i shares. New: %i Have: %i/~%i' % (len(shares), new_count, len(tracker.items), 2*net.CHAIN_LENGTH)
 294
 295             def handle_share_hashes(self, hashes, peer):
 296                 t = time.time()
 297                 get_hashes = []
 298                 for share_hash in hashes:
 299                     if share_hash in tracker.items:
 300                         continue
 301                     last_request_time, count = requested.get(share_hash, (None, 0))
 302                     if last_request_time is not None and last_request_time - 5 < t < last_request_time + 10 * 1.5**count:
 303                         continue
 304                     print 'Got share hash, requesting! Hash: %s' % (p2pool_data.format_hash(share_hash),)
 305                     get_hashes.append(share_hash)
 306                     requested[share_hash] = t, count + 1
 307
 308                 if hashes and peer is not None:
 309                     peer_heads.setdefault(hashes[0], set()).add(peer)
 310                 if get_hashes:
 311                     peer.send_getshares(hashes=get_hashes, parents=0, stops=[])
 312
 313             def handle_get_shares(self, hashes, parents, stops, peer):
 314                 parents = min(parents, 1000//len(hashes))
 315                 stops = set(stops)
 316                 shares = []
 317                 for share_hash in hashes:
 318                     for share in tracker.get_chain(share_hash, min(parents + 1, tracker.get_height(share_hash))):
 319                         if share.hash in stops:
 320                             break
 321                         shares.append(share)
 322                 print 'Sending %i shares to %s:%i' % (len(shares), peer.addr[0], peer.addr[1])
 323                 return shares
 324
 325             def handle_bestblock(self, header, peer):
 326                 if net.PARENT.POW_FUNC(bitcoin_data.block_header_type.pack(header)) > header['bits'].target:
 327                     raise p2p.PeerMisbehavingError('received block header fails PoW test')
 328                 handle_header(header)
 329
 330         @deferral.retry('Error submitting primary block: (will retry)', 10, 10)
 331         def submit_block_p2p(block):
 332             if factory.conn.value is None:
 333                 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'])))
 334                 raise deferral.RetrySilentlyException()
 335             factory.conn.value.send_block(block=block)
 336
 337         @deferral.retry('Error submitting block: (will retry)', 10, 10)
 338         @defer.inlineCallbacks
 339         def submit_block_rpc(block, ignore_failure):
 340             if bitcoind_work.value['use_getblocktemplate']:
 341                 success = yield bitcoind.rpc_getblocktemplate(dict(data=bitcoin_data.block_type.pack(block).encode('hex')))
 342             else:
 343                 success = yield bitcoind.rpc_getmemorypool(bitcoin_data.block_type.pack(block).encode('hex'))
 344             success_expected = net.PARENT.POW_FUNC(bitcoin_data.block_header_type.pack(block['header'])) <= block['header']['bits'].target
 345             if (not success and success_expected and not ignore_failure) or (success and not success_expected):
 346                 print >>sys.stderr, 'Block submittal result: %s Expected: %s' % (success, success_expected)
 347
 348         def submit_block(block, ignore_failure):
 349             submit_block_p2p(block)
 350             submit_block_rpc(block, ignore_failure)
 351
 352         @tracker.verified.added.watch
 353         def _(share):
 354             if share.pow_hash <= share.header['bits'].target:
 355                 submit_block(share.as_block(tracker), ignore_failure=True)
 356                 print
 357                 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)
 358                 print
 359                 def spread():
 360                     if (get_height_rel_highest(share.header['previous_block']) > -5 or
 361                         bitcoind_work.value['previous_block'] in [share.header['previous_block'], share.header_hash]):
 362                         broadcast_share(share.hash)
 363                 spread()
 364                 reactor.callLater(5, spread) # so get_height_rel_highest can update
 365
 366         print 'Joining p2pool network using port %i...' % (args.p2pool_port,)
 367
 368         @defer.inlineCallbacks
 369         def parse(x):
 370             if ':' in x:
 371                 ip, port = x.split(':')
 372                 defer.returnValue(((yield reactor.resolve(ip)), int(port)))
 373             else:
 374                 defer.returnValue(((yield reactor.resolve(x)), net.P2P_PORT))
 375
 376         addrs = {}
 377         if os.path.exists(os.path.join(datadir_path, 'addrs')):
 378             try:
 379                 with open(os.path.join(datadir_path, 'addrs'), 'rb') as f:
 380                     addrs.update(dict((tuple(k), v) for k, v in json.loads(f.read())))
 381             except:
 382                 print >>sys.stderr, 'error parsing addrs'
 383         for addr_df in map(parse, net.BOOTSTRAP_ADDRS):
 384             try:
 385                 addr = yield addr_df
 386                 if addr not in addrs:
 387                     addrs[addr] = (0, time.time(), time.time())
 388             except:
 389                 log.err()
 390
 391         connect_addrs = set()
 392         for addr_df in map(parse, args.p2pool_nodes):
 393             try:
 394                 connect_addrs.add((yield addr_df))
 395             except:
 396                 log.err()
 397
 398         p2p_node = Node(
 399             best_share_hash_func=lambda: best_share_var.value,
 400             port=args.p2pool_port,
 401             net=net,
 402             addr_store=addrs,
 403             connect_addrs=connect_addrs,
 404             max_incoming_conns=args.p2pool_conns,
 405         )
 406         p2p_node.start()
 407
 408         def save_addrs():
 409             with open(os.path.join(datadir_path, 'addrs'), 'wb') as f:
 410                 f.write(json.dumps(p2p_node.addr_store.items()))
 411         task.LoopingCall(save_addrs).start(60)
 412
 413         @best_block_header.changed.watch
 414         def _(header):
 415             for peer in p2p_node.peers.itervalues():
 416                 peer.send_bestblock(header=header)
 417
 418         @defer.inlineCallbacks
 419         def broadcast_share(share_hash):
 420             shares = []
 421             for share in tracker.get_chain(share_hash, min(5, tracker.get_height(share_hash))):
 422                 if share.hash in shared_share_hashes:
 423                     break
 424                 shared_share_hashes.add(share.hash)
 425                 shares.append(share)
 426
 427             for peer in list(p2p_node.peers.itervalues()):
 428                 yield peer.sendShares([share for share in shares if share.peer is not peer])
 429
 430         # send share when the chain changes to their chain
 431         best_share_var.changed.watch(broadcast_share)
 432
 433         def save_shares():
 434             for share in tracker.get_chain(best_share_var.value, min(tracker.get_height(best_share_var.value), 2*net.CHAIN_LENGTH)):
 435                 ss.add_share(share)
 436                 if share.hash in tracker.verified.items:
 437                     ss.add_verified_hash(share.hash)
 438         task.LoopingCall(save_shares).start(60)
 439
 440         print '    ...success!'
 441         print
 442
 443         if args.upnp:
 444             @defer.inlineCallbacks
 445             def upnp_thread():
 446                 while True:
 447                     try:
 448                         is_lan, lan_ip = yield ipdiscover.get_local_ip()
 449                         if is_lan:
 450                             pm = yield portmapper.get_port_mapper()
 451                             yield pm._upnp.add_port_mapping(lan_ip, args.p2pool_port, args.p2pool_port, 'p2pool', 'TCP')
 452                     except defer.TimeoutError:
 453                         pass
 454                     except:
 455                         if p2pool.DEBUG:
 456                             log.err(None, 'UPnP error:')
 457                     yield deferral.sleep(random.expovariate(1/120))
 458             upnp_thread()
 459
 460         # start listening for workers with a JSON-RPC server
 461
 462         print 'Listening for workers on %r port %i...' % (worker_endpoint[0], worker_endpoint[1])
 463
 464         get_current_txouts = lambda: p2pool_data.get_expected_payouts(tracker, best_share_var.value, bitcoind_work.value['bits'].target, bitcoind_work.value['subsidy'], net)
 465
 466         wb = work.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, broadcast_share, block_height_var)
 467         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, bitcoind_warning_var)
 468         worker_interface.WorkerInterface(wb).attach_to(web_root, get_handler=lambda request: request.redirect('/static/'))
 469
 470         deferral.retry('Error binding to worker port:', traceback=False)(reactor.listenTCP)(worker_endpoint[1], server.Site(web_root), interface=worker_endpoint[0])
 471
 472         with open(os.path.join(os.path.join(datadir_path, 'ready_flag')), 'wb') as f:
 473             pass
 474
 475         print '    ...success!'
 476         print
 477
 478
 479         # done!
 480         print 'Started successfully!'
 481         print 'Go to http://127.0.0.1:%i/ to view graphs and statistics!' % (worker_endpoint[1],)
 482         if args.donation_percentage > 0.51:
 483             print '''Donating %.1f%% of work towards P2Pool's development. Thanks for the tip!''' % (args.donation_percentage,)
 484         elif args.donation_percentage < 0.49:
 485             print '''Donating %.1f%% of work towards P2Pool's development. Please donate to encourage further development of P2Pool!''' % (args.donation_percentage,)
 486         else:
 487             print '''Donating %.1f%% of work towards P2Pool's development. Thank you!''' % (args.donation_percentage,)
 488             print 'You can increase this amount with --give-author argument! (or decrease it, if you must)'
 489         print
 490
 491
 492         if hasattr(signal, 'SIGALRM'):
 493             signal.signal(signal.SIGALRM, lambda signum, frame: reactor.callFromThread(
 494                 sys.stderr.write, 'Watchdog timer went off at:\n' + ''.join(traceback.format_stack())
 495             ))
 496             signal.siginterrupt(signal.SIGALRM, False)
 497             task.LoopingCall(signal.alarm, 30).start(1)
 498
 499         if args.irc_announce:
 500             from twisted.words.protocols import irc
 501             class IRCClient(irc.IRCClient):
 502                 nickname = 'p2pool%02i' % (random.randrange(100),)
 503                 channel = net.ANNOUNCE_CHANNEL
 504                 def lineReceived(self, line):
 505                     if p2pool.DEBUG:
 506                         print repr(line)
 507                     irc.IRCClient.lineReceived(self, line)
 508                 def signedOn(self):
 509                     irc.IRCClient.signedOn(self)
 510                     self.factory.resetDelay()
 511                     self.join(self.channel)
 512                     @defer.inlineCallbacks
 513                     def new_share(share):
 514                         if share.pow_hash <= share.header['bits'].target and abs(share.timestamp - time.time()) < 10*60:
 515                             yield deferral.sleep(random.expovariate(1/60))
 516                             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)
 517                             if all('%x' % (share.header_hash,) not in old_message for old_message in self.recent_messages):
 518                                 self.say(self.channel, message)
 519                                 self._remember_message(message)
 520                     self.watch_id = tracker.verified.added.watch(new_share)
 521                     self.recent_messages = []
 522                 def _remember_message(self, message):
 523                     self.recent_messages.append(message)
 524                     while len(self.recent_messages) > 100:
 525                         self.recent_messages.pop(0)
 526                 def privmsg(self, user, channel, message):
 527                     if channel == self.channel:
 528                         self._remember_message(message)
 529                 def connectionLost(self, reason):
 530                     tracker.verified.added.unwatch(self.watch_id)
 531                     print 'IRC connection lost:', reason.getErrorMessage()
 532             class IRCClientFactory(protocol.ReconnectingClientFactory):
 533                 protocol = IRCClient
 534             reactor.connectTCP("irc.freenode.net", 6667, IRCClientFactory())
 535
 536         @defer.inlineCallbacks
 537         def status_thread():
 538             last_str = None
 539             last_time = 0
 540             while True:
 541                 yield deferral.sleep(3)
 542                 try:
 543                     height = tracker.get_height(best_share_var.value)
 544                     this_str = 'P2Pool: %i shares in chain (%i verified/%i total) Peers: %i (%i incoming)' % (
 545                         height,
 546                         len(tracker.verified.items),
 547                         len(tracker.items),
 548                         len(p2p_node.peers),
 549                         sum(1 for peer in p2p_node.peers.itervalues() if peer.incoming),
 550                     ) + (' FDs: %i R/%i W' % (len(reactor.getReaders()), len(reactor.getWriters())) if p2pool.DEBUG else '')
 551
 552                     datums, dt = wb.local_rate_monitor.get_datums_in_last()
 553                     my_att_s = sum(datum['work']/dt for datum in datums)
 554                     this_str += '\n Local: %sH/s in last %s Local dead on arrival: %s Expected time to share: %s' % (
 555                         math.format(int(my_att_s)),
 556                         math.format_dt(dt),
 557                         math.format_binomial_conf(sum(1 for datum in datums if datum['dead']), len(datums), 0.95),
 558                         math.format_dt(2**256 / tracker.items[best_share_var.value].max_target / my_att_s) if my_att_s and best_share_var.value else '???',
 559                     )
 560
 561                     if height > 2:
 562                         (stale_orphan_shares, stale_doa_shares), shares, _ = wb.get_stale_counts()
 563                         stale_prop = p2pool_data.get_average_stale_prop(tracker, best_share_var.value, min(60*60//net.SHARE_PERIOD, height))
 564                         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)
 565
 566                         this_str += '\n Shares: %i (%i orphan, %i dead) Stale rate: %s Efficiency: %s Current payout: %.4f %s' % (
 567                             shares, stale_orphan_shares, stale_doa_shares,
 568                             math.format_binomial_conf(stale_orphan_shares + stale_doa_shares, shares, 0.95),
 569                             math.format_binomial_conf(stale_orphan_shares + stale_doa_shares, shares, 0.95, lambda x: (1 - x)/(1 - stale_prop)),
 570                             get_current_txouts().get(bitcoin_data.pubkey_hash_to_script2(my_pubkey_hash), 0)*1e-8, net.PARENT.SYMBOL,
 571                         )
 572                         this_str += '\n Pool: %sH/s Stale rate: %.1f%% Expected time to block: %s' % (
 573                             math.format(int(real_att_s)),
 574                             100*stale_prop,
 575                             math.format_dt(2**256 / bitcoind_work.value['bits'].target / real_att_s),
 576                         )
 577
 578                         for warning in p2pool_data.get_warnings(tracker, best_share_var.value, net, bitcoind_warning_var.value, bitcoind_work.value):
 579                             print >>sys.stderr, '#'*40
 580                             print >>sys.stderr, '>>> Warning: ' + warning
 581                             print >>sys.stderr, '#'*40
 582
 583                     if this_str != last_str or time.time() > last_time + 15:
 584                         print this_str
 585                         last_str = this_str
 586                         last_time = time.time()
 587                 except:
 588                     log.err()
 589         status_thread()
 590     except:
 591         reactor.stop()
 592         log.err(None, 'Fatal error:')
 593
 594 def run():
 595     realnets = dict((name, net) for name, net in networks.nets.iteritems() if '_testnet' not in name)
 596
 597     parser = fixargparse.FixedArgumentParser(description='p2pool (version %s)' % (p2pool.__version__,), fromfile_prefix_chars='@')
 598     parser.add_argument('--version', action='version', version=p2pool.__version__)
 599     parser.add_argument('--net',
 600         help='use specified network (default: bitcoin)',
 601         action='store', choices=sorted(realnets), default='bitcoin', dest='net_name')
 602     parser.add_argument('--testnet',
 603         help='''use the network's testnet''',
 604         action='store_const', const=True, default=False, dest='testnet')
 605     parser.add_argument('--debug',
 606         help='enable debugging mode',
 607         action='store_const', const=True, default=False, dest='debug')
 608     parser.add_argument('-a', '--address',
 609         help='generate payouts to this address (default: <address requested from bitcoind>)',
 610         type=str, action='store', default=None, dest='address')
 611     parser.add_argument('--datadir',
 612         help='store data in this directory (default: <directory run_p2pool.py is in>/data)',
 613         type=str, action='store', default=None, dest='datadir')
 614     parser.add_argument('--logfile',
 615         help='''log to this file (default: data/<NET>/log)''',
 616         type=str, action='store', default=None, dest='logfile')
 617     parser.add_argument('--merged',
 618         help='call getauxblock on this url to get work for merged mining (example: http://ncuser:ncpass@127.0.0.1:10332/)',
 619         type=str, action='append', default=[], dest='merged_urls')
 620     parser.add_argument('--give-author', metavar='DONATION_PERCENTAGE',
 621         help='donate this percentage of work towards the development of p2pool (default: 0.5)',
 622         type=float, action='store', default=0.5, dest='donation_percentage')
 623     parser.add_argument('--iocp',
 624         help='use Windows IOCP API in order to avoid errors due to large number of sockets being open',
 625         action='store_true', default=False, dest='iocp')
 626     parser.add_argument('--irc-announce',
 627         help='announce any blocks found on irc://irc.freenode.net/#p2pool',
 628         action='store_true', default=False, dest='irc_announce')
 629     parser.add_argument('--no-bugreport',
 630         help='disable submitting caught exceptions to the author',
 631         action='store_true', default=False, dest='no_bugreport')
 632
 633     p2pool_group = parser.add_argument_group('p2pool interface')
 634     p2pool_group.add_argument('--p2pool-port', metavar='PORT',
 635         help='use port PORT to listen for connections (forward this port from your router!) (default: %s)' % ', '.join('%s:%i' % (name, net.P2P_PORT) for name, net in sorted(realnets.items())),
 636         type=int, action='store', default=None, dest='p2pool_port')
 637     p2pool_group.add_argument('-n', '--p2pool-node', metavar='ADDR[:PORT]',
 638         help='connect to existing p2pool node at ADDR listening on port PORT (defaults to default p2pool P2P port) in addition to builtin addresses',
 639         type=str, action='append', default=[], dest='p2pool_nodes')
 640     parser.add_argument('--disable-upnp',
 641         help='''don't attempt to use UPnP to forward p2pool's P2P port from the Internet to this computer''',
 642         action='store_false', default=True, dest='upnp')
 643     p2pool_group.add_argument('--max-conns', metavar='CONNS',
 644         help='maximum incoming connections (default: 40)',
 645         type=int, action='store', default=40, dest='p2pool_conns')
 646
 647     worker_group = parser.add_argument_group('worker interface')
 648     worker_group.add_argument('-w', '--worker-port', metavar='PORT or ADDR:PORT',
 649         help='listen on PORT on interface with ADDR for RPC connections from miners (default: all interfaces, %s)' % ', '.join('%s:%i' % (name, net.WORKER_PORT) for name, net in sorted(realnets.items())),
 650         type=str, action='store', default=None, dest='worker_endpoint')
 651     worker_group.add_argument('-f', '--fee', metavar='FEE_PERCENTAGE',
 652         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:WORKER_PORT/fee (default: 0)''',
 653         type=float, action='store', default=0, dest='worker_fee')
 654
 655     bitcoind_group = parser.add_argument_group('bitcoind interface')
 656     bitcoind_group.add_argument('--bitcoind-address', metavar='BITCOIND_ADDRESS',
 657         help='connect to this address (default: 127.0.0.1)',
 658         type=str, action='store', default='127.0.0.1', dest='bitcoind_address')
 659     bitcoind_group.add_argument('--bitcoind-rpc-port', metavar='BITCOIND_RPC_PORT',
 660         help='''connect to JSON-RPC interface at this port (default: %s <read from bitcoin.conf if password not provided>)''' % ', '.join('%s:%i' % (name, net.PARENT.RPC_PORT) for name, net in sorted(realnets.items())),
 661         type=int, action='store', default=None, dest='bitcoind_rpc_port')
 662     bitcoind_group.add_argument('--bitcoind-p2p-port', metavar='BITCOIND_P2P_PORT',
 663         help='''connect to P2P interface at this port (default: %s <read from bitcoin.conf if password not provided>)''' % ', '.join('%s:%i' % (name, net.PARENT.P2P_PORT) for name, net in sorted(realnets.items())),
 664         type=int, action='store', default=None, dest='bitcoind_p2p_port')
 665
 666     bitcoind_group.add_argument(metavar='BITCOIND_RPCUSERPASS',
 667         help='bitcoind RPC interface username, then password, space-separated (only one being provided will cause the username to default to being empty, and none will cause P2Pool to read them from bitcoin.conf)',
 668         type=str, action='store', default=[], nargs='*', dest='bitcoind_rpc_userpass')
 669
 670     args = parser.parse_args()
 671
 672     if args.debug:
 673         p2pool.DEBUG = True
 674         defer.setDebugging(True)
 675
 676     net_name = args.net_name + ('_testnet' if args.testnet else '')
 677     net = networks.nets[net_name]
 678
 679     datadir_path = os.path.join((os.path.join(os.path.dirname(sys.argv[0]), 'data') if args.datadir is None else args.datadir), net_name)
 680     if not os.path.exists(datadir_path):
 681         os.makedirs(datadir_path)
 682
 683     if len(args.bitcoind_rpc_userpass) > 2:
 684         parser.error('a maximum of two arguments are allowed')
 685     args.bitcoind_rpc_username, args.bitcoind_rpc_password = ([None, None] + args.bitcoind_rpc_userpass)[-2:]
 686
 687     if args.bitcoind_rpc_password is None:
 688         conf_path = net.PARENT.CONF_FILE_FUNC()
 689         if not os.path.exists(conf_path):
 690             parser.error('''Bitcoin configuration file not found. Manually enter your RPC password.\r\n'''
 691                 '''If you actually haven't created a configuration file, you should create one at %s with the text:\r\n'''
 692                 '''\r\n'''
 693                 '''server=1\r\n'''
 694                 '''rpcpassword=%x\r\n'''
 695                 '''\r\n'''
 696                 '''Keep that password secret! After creating the file, restart Bitcoin.''' % (conf_path, random.randrange(2**128)))
 697         with open(conf_path, 'rb') as f:
 698             cp = ConfigParser.RawConfigParser()
 699             cp.readfp(StringIO.StringIO('[x]\r\n' + f.read()))
 700             for conf_name, var_name, var_type in [
 701                 ('rpcuser', 'bitcoind_rpc_username', str),
 702                 ('rpcpassword', 'bitcoind_rpc_password', str),
 703                 ('rpcport', 'bitcoind_rpc_port', int),
 704                 ('port', 'bitcoind_p2p_port', int),
 705             ]:
 706                 if getattr(args, var_name) is None and cp.has_option('x', conf_name):
 707                     setattr(args, var_name, var_type(cp.get('x', conf_name)))
 708         if args.bitcoind_rpc_password is None:
 709             parser.error('''Bitcoin configuration file didn't contain an rpcpassword= line! Add one!''')
 710
 711     if args.bitcoind_rpc_username is None:
 712         args.bitcoind_rpc_username = ''
 713
 714     if args.bitcoind_rpc_port is None:
 715         args.bitcoind_rpc_port = net.PARENT.RPC_PORT
 716
 717     if args.bitcoind_p2p_port is None:
 718         args.bitcoind_p2p_port = net.PARENT.P2P_PORT
 719
 720     if args.p2pool_port is None:
 721         args.p2pool_port = net.P2P_PORT
 722
 723     if args.worker_endpoint is None:
 724         worker_endpoint = '', net.WORKER_PORT
 725     elif ':' not in args.worker_endpoint:
 726         worker_endpoint = '', int(args.worker_endpoint)
 727     else:
 728         addr, port = args.worker_endpoint.rsplit(':', 1)
 729         worker_endpoint = addr, int(port)
 730
 731     if args.address is not None:
 732         try:
 733             args.pubkey_hash = bitcoin_data.address_to_pubkey_hash(args.address, net.PARENT)
 734         except Exception, e:
 735             parser.error('error parsing address: ' + repr(e))
 736     else:
 737         args.pubkey_hash = None
 738
 739     def separate_url(url):
 740         s = urlparse.urlsplit(url)
 741         if '@' not in s.netloc:
 742             parser.error('merged url netloc must contain an "@"')
 743         userpass, new_netloc = s.netloc.rsplit('@', 1)
 744         return urlparse.urlunsplit(s._replace(netloc=new_netloc)), userpass
 745     merged_urls = map(separate_url, args.merged_urls)
 746
 747     if args.logfile is None:
 748         args.logfile = os.path.join(datadir_path, 'log')
 749
 750     logfile = logging.LogFile(args.logfile)
 751     pipe = logging.TimestampingPipe(logging.TeePipe([logging.EncodeReplacerPipe(sys.stderr), logfile]))
 752     sys.stdout = logging.AbortPipe(pipe)
 753     sys.stderr = log.DefaultObserver.stderr = logging.AbortPipe(logging.PrefixPipe(pipe, '> '))
 754     if hasattr(signal, "SIGUSR1"):
 755         def sigusr1(signum, frame):
 756             print 'Caught SIGUSR1, closing %r...' % (args.logfile,)
 757             logfile.reopen()
 758             print '...and reopened %r after catching SIGUSR1.' % (args.logfile,)
 759         signal.signal(signal.SIGUSR1, sigusr1)
 760     task.LoopingCall(logfile.reopen).start(5)
 761
 762     class ErrorReporter(object):
 763         def __init__(self):
 764             self.last_sent = None
 765
 766         def emit(self, eventDict):
 767             if not eventDict["isError"]:
 768                 return
 769
 770             if self.last_sent is not None and time.time() < self.last_sent + 5:
 771                 return
 772             self.last_sent = time.time()
 773
 774             if 'failure' in eventDict:
 775                 text = ((eventDict.get('why') or 'Unhandled Error')
 776                     + '\n' + eventDict['failure'].getTraceback())
 777             else:
 778                 text = " ".join([str(m) for m in eventDict["message"]]) + "\n"
 779
 780             from twisted.web import client
 781             client.getPage(
 782                 url='http://u.forre.st/p2pool_error.cgi',
 783                 method='POST',
 784                 postdata=p2pool.__version__ + ' ' + net.NAME + '\n' + text,
 785                 timeout=15,
 786             ).addBoth(lambda x: None)
 787     if not args.no_bugreport:
 788         log.addObserver(ErrorReporter().emit)
 789
 790     reactor.callWhenRunning(main, args, net, datadir_path, merged_urls, worker_endpoint)
 791     reactor.run()