Merge branch 'tx_preforwarding'

[p2pool.git] / p2pool / main.py

diff --git a/p2pool/main.py b/p2pool/main.py
index 7b3ed2f..f1de657 100644 (file)
--- a/p2pool/main.py
+++ b/p2pool/main.py
@@ -50,6 +50,7 @@ def getwork(bitcoind, use_getblocktemplate=False):
         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']),
@@ -250,8 +251,7 @@ def main(args, net, datadir_path, merged_urls, worker_endpoint):
         def _(_=None):
             new_mining_txs = {}
             new_known_txs = dict(known_txs_var.value)
-            for tx in bitcoind_work.value['transactions']:
-                tx_hash = bitcoin_data.hash256(bitcoin_data.tx_type.pack(tx))
+            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)
@@ -264,7 +264,9 @@ def main(args, net, datadir_path, merged_urls, worker_endpoint):
             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])
         
@@ -411,6 +413,18 @@ def main(args, net, datadir_path, merged_urls, worker_endpoint):
         )
         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()))
@@ -498,7 +512,7 @@ def main(args, net, datadir_path, merged_urls, worker_endpoint):
         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])
@@ -540,11 +554,14 @@ def main(args, net, datadir_path, merged_urls, worker_endpoint):
                         print repr(line)
                     irc.IRCClient.lineReceived(self, line)
                 def signedOn(self):
+                    self.in_channel = False
                     irc.IRCClient.signedOn(self)
                     self.factory.resetDelay()
                     self.join(self.channel)
                     @defer.inlineCallbacks
                     def new_share(share):
+                        if not self.in_channel:
+                            return
                         if share.pow_hash <= share.header['bits'].target and abs(share.timestamp - time.time()) < 10*60:
                             yield deferral.sleep(random.expovariate(1/60))
                             message = '\x02%s BLOCK FOUND by %s! %s%064x' % (net.NAME.upper(), bitcoin_data.script2_to_address(share.new_script, net.PARENT), net.PARENT.BLOCK_EXPLORER_URL_PREFIX, share.header_hash)
@@ -553,6 +570,10 @@ def main(args, net, datadir_path, merged_urls, worker_endpoint):
                                 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: