Merge branch 'tx_preforwarding'

diff --combined p2pool/main.py
index 82b384d,8ff9e0f..f1de657
--- 1/p2pool/main.py
--- 2/p2pool/main.py
+++ b/p2pool/main.py
@@@ -50,6 -50,7 +50,7 @@@ def getwork(bitcoind, use_getblocktempl
          version=work['version'],
          previous_block=int(work['previousblockhash'], 16),
          transactions=map(bitcoin_data.tx_type.unpack, packed_transactions),
+         transaction_hashes=map(bitcoin_data.hash256, packed_transactions),
          subsidy=work['coinbasevalue'],
          time=work['time'] if 'time' in work else work['curtime'],
          bits=bitcoin_data.FloatingIntegerType().unpack(work['bits'].decode('hex')[::-1]) if isinstance(work['bits'], (str, unicode)) else bitcoin_data.FloatingInteger(work['bits']),
@@@ -226,12 -227,14 +227,14 @@@ def main(args, net, datadir_path, merge
          
          # BEST SHARE
          
+         known_txs_var = variable.Variable({}) # hash -> tx
+         mining_txs_var = variable.Variable({}) # hash -> tx
          get_height_rel_highest = yield height_tracker.get_height_rel_highest_func(bitcoind, factory, lambda: bitcoind_work.value['previous_block'], net)
          
          best_share_var = variable.Variable(None)
          desired_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, desired = tracker.think(get_height_rel_highest, bitcoind_work.value['previous_block'], bitcoind_work.value['bits'], known_txs_var.value)
              
              best_share_var.set(best)
              desired_var.set(desired)
@@@ -243,6 -246,30 +246,30 @@@
          
          # setup p2p logic and join p2pool network
          
+         # update mining_txs according to getwork results
+         @bitcoind_work.changed.run_and_watch
+         def _(_=None):
+             new_mining_txs = {}
+             new_known_txs = dict(known_txs_var.value)
+             for tx_hash, tx in zip(bitcoind_work.value['transaction_hashes'], bitcoind_work.value['transactions']):
+                 new_mining_txs[tx_hash] = tx
+                 new_known_txs[tx_hash] = tx
+             mining_txs_var.set(new_mining_txs)
+             known_txs_var.set(new_known_txs)
+         # add p2p transactions from bitcoind to known_txs
+         @factory.new_tx.watch
+         def _(tx):
+             new_known_txs = dict(known_txs_var.value)
+             new_known_txs[bitcoin_data.hash256(bitcoin_data.tx_type.pack(tx))] = tx
+             known_txs_var.set(new_known_txs)
+         # forward transactions seen to bitcoind
+         @known_txs_var.transitioned.watch
+         @defer.inlineCallbacks
+         def _(before, after):
+             yield deferral.sleep(random.expovariate(1/1))
+             for tx_hash in set(after) - set(before):
+                 factory.conn.value.send_tx(tx=after[tx_hash])
+         
          class Node(p2p.Node):
              def handle_shares(self, shares, peer):
                  if len(shares) > 5:
@@@ -326,7 -353,11 +353,11 @@@
          @tracker.verified.added.watch
          def _(share):
              if share.pow_hash <= share.header['bits'].target:
-                 submit_block(share.as_block(tracker), ignore_failure=True)
+                 block = share.as_block(tracker, known_txs_var.value)
+                 if block is None:
+                     print >>sys.stderr, 'GOT INCOMPLETE BLOCK FROM PEER! %s bitcoin: %s%064x' % (p2pool_data.format_hash(share.hash), net.PARENT.BLOCK_EXPLORER_URL_PREFIX, share.header_hash)
+                     return
+                 submit_block(block, 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
@@@ -377,9 -408,23 +408,23 @@@
              connect_addrs=connect_addrs,
              max_incoming_conns=args.p2pool_conns,
              traffic_happened=traffic_happened,
+             known_txs_var=known_txs_var,
+             mining_txs_var=mining_txs_var,
          )
          p2p_node.start()
          
+         def forget_old_txs():
+             new_known_txs = {}
+             for peer in p2p_node.peers.itervalues():
+                 new_known_txs.update(peer.remembered_txs)
+             new_known_txs.update(mining_txs_var.value)
+             for share in tracker.get_chain(best_share_var.value, min(120, tracker.get_height(best_share_var.value))):
+                 for tx_hash in share.new_transaction_hashes:
+                     if tx_hash in known_txs_var.value:
+                         new_known_txs[tx_hash] = known_txs_var.value[tx_hash]
+             known_txs_var.set(new_known_txs)
+         task.LoopingCall(forget_old_txs).start(10)
+         
          def save_addrs():
              with open(os.path.join(datadir_path, 'addrs'), 'wb') as f:
                  f.write(json.dumps(p2p_node.addr_store.items()))
@@@ -400,7 -445,7 +445,7 @@@
                  shares.append(share)
              
              for peer in list(p2p_node.peers.itervalues()):
-                 yield peer.sendShares([share for share in shares if share.peer is not peer])
+                 yield peer.sendShares([share for share in shares if share.peer is not peer], tracker, known_txs_var.value, include_txs_with=[share_hash])
          
          # send share when the chain changes to their chain
          best_share_var.changed.watch(broadcast_share)
@@@ -467,7 -512,7 +512,7 @@@
          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 = 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)
 -        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, traffic_happened)
 +        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, traffic_happened, args.donation_percentage)
          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])
@@@ -509,14 -554,11 +554,14 @@@
                          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)
@@@ -525,10 -567,6 +570,10 @@@
                                  self._remember_message(message)
                      self.watch_id = 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:

diff --combined p2pool/web.py
index c843671,a0060c8..3eafa1b
--- 1/p2pool/web.py
--- 2/p2pool/web.py
+++ b/p2pool/web.py
@@@ -13,7 -13,7 +13,7 @@@ from twisted.web import resource, stati
  import p2pool
  from bitcoin import data as bitcoin_data
  from . import data as p2pool_data
 -from util import deferred_resource, graph, math
 +from util import deferred_resource, graph, math, pack
  
  def _atomic_read(filename):
      try:
@@@ -44,7 -44,7 +44,7 @@@ def _atomic_write(filename, data)
          os.remove(filename)
          os.rename(filename + '.new', filename)
  
 -def get_web_root(tracker, bitcoind_work, get_current_txouts, datadir_path, net, get_stale_counts, my_pubkey_hash, local_rate_monitor, worker_fee, p2p_node, my_share_hashes, pseudoshare_received, share_received, best_share_var, bitcoin_warning_var, traffic_happened):
 +def get_web_root(tracker, bitcoind_work, get_current_txouts, datadir_path, net, get_stale_counts, my_pubkey_hash, local_rate_monitor, worker_fee, p2p_node, my_share_hashes, pseudoshare_received, share_received, best_share_var, bitcoin_warning_var, traffic_happened, donation_percentage):
      start_time = time.time()
      
      web_root = resource.Resource()
@@@ -176,7 -176,6 +176,7 @@@
              attempts_to_block=bitcoin_data.target_to_average_attempts(bitcoind_work.value['bits'].target),
              block_value=bitcoind_work.value['subsidy']*1e-8,
              warnings=p2pool_data.get_warnings(tracker, best_share_var.value, net, bitcoin_warning_var.value, bitcoind_work.value),
 +            donation_proportion=donation_percentage/100,
          )
      
      class WebInterface(deferred_resource.DeferredResource):
@@@ -205,6 -204,7 +205,7 @@@
      web_root.putChild('global_stats', WebInterface(get_global_stats))
      web_root.putChild('local_stats', WebInterface(get_local_stats))
      web_root.putChild('peer_addresses', WebInterface(lambda: ['%s:%i' % (peer.transport.getPeer().host, peer.transport.getPeer().port) for peer in p2p_node.peers.itervalues()]))
+     web_root.putChild('peer_txpool_sizes', WebInterface(lambda: dict(('%s:%i' % (peer.transport.getPeer().host, peer.transport.getPeer().port), peer.remembered_txs_size) for peer in p2p_node.peers.itervalues())))
      web_root.putChild('pings', WebInterface(defer.inlineCallbacks(lambda: defer.returnValue(
          dict([(a, (yield b)) for a, b in
              [(
@@@ -217,12 -217,7 +218,12 @@@
      ))))
      web_root.putChild('peer_versions', WebInterface(lambda: dict(('%s:%i' % peer.addr, peer.other_sub_version) for peer in p2p_node.peers.itervalues())))
      web_root.putChild('payout_addr', WebInterface(lambda: bitcoin_data.pubkey_hash_to_address(my_pubkey_hash, net.PARENT)))
 -    web_root.putChild('recent_blocks', WebInterface(lambda: [dict(ts=s.timestamp, hash='%064x' % s.header_hash) for s in tracker.get_chain(best_share_var.value, min(tracker.get_height(best_share_var.value), 24*60*60//net.SHARE_PERIOD)) if s.pow_hash <= s.header['bits'].target]))
 +    web_root.putChild('recent_blocks', WebInterface(lambda: [dict(
 +        ts=s.timestamp,
 +        hash='%064x' % s.header_hash,
 +        number=pack.IntType(24).unpack(s.share_data['coinbase'][1:4]),
 +        share='%064x' % s.hash,
 +    ) for s in tracker.get_chain(best_share_var.value, min(tracker.get_height(best_share_var.value), 24*60*60//net.SHARE_PERIOD)) if s.pow_hash <= s.header['bits'].target]))
      web_root.putChild('uptime', WebInterface(lambda: time.time() - start_time))
      web_root.putChild('stale_rates', WebInterface(lambda: p2pool_data.get_stale_counts(tracker, best_share_var.value, 720, rates=True)))