4 from twisted.internet import defer, reactor, task
5 from twisted.python import log
7 from p2pool import data as p2pool_data, p2p
8 from p2pool.bitcoin import data as bitcoin_data, helper, height_tracker
9 from p2pool.util import deferral, variable
12 class P2PNode(p2p.Node):
13 def __init__(self, node, p2pool_port, p2pool_conns, addrs, connect_addrs):
15 p2p.Node.__init__(self,
16 best_share_hash_func=lambda: node.best_share_var.value,
20 connect_addrs=connect_addrs,
21 max_incoming_conns=p2pool_conns,
22 known_txs_var=node.known_txs_var,
23 mining_txs_var=node.mining_txs_var,
26 def handle_shares(self, shares, peer):
28 print 'Processing %i shares from %s...' % (len(shares), '%s:%i' % peer.addr if peer is not None else None)
32 if share.hash in self.node.tracker.items:
33 #print 'Got duplicate share, ignoring. Hash: %s' % (p2pool_data.format_hash(share.hash),)
38 #print 'Received share %s from %r' % (p2pool_data.format_hash(share.hash), share.peer.addr if share.peer is not None else None)
40 self.node.tracker.add(share)
43 self.node.set_best_share()
46 print '... done processing %i shares. New: %i Have: %i/~%i' % (len(shares), new_count, len(self.node.tracker.items), 2*self.node.net.CHAIN_LENGTH)
48 @defer.inlineCallbacks
49 def handle_share_hashes(self, hashes, peer):
50 new_hashes = [x for x in hashes if x not in self.node.tracker.items]
54 shares = yield peer.get_shares(
60 log.err(None, 'in handle_share_hashes:')
62 self.handle_shares(shares, peer)
64 def handle_get_shares(self, hashes, parents, stops, peer):
65 parents = min(parents, 1000//len(hashes))
68 for share_hash in hashes:
69 for share in self.node.tracker.get_chain(share_hash, min(parents + 1, self.node.tracker.get_height(share_hash))):
70 if share.hash in stops:
73 print 'Sending %i shares to %s:%i' % (len(shares), peer.addr[0], peer.addr[1])
76 def handle_bestblock(self, header, peer):
77 if self.node.net.PARENT.POW_FUNC(bitcoin_data.block_header_type.pack(header)) > header['bits'].target:
78 raise p2p.PeerMisbehavingError('received block header fails PoW test')
79 self.node.handle_header(header)
81 @defer.inlineCallbacks
82 def broadcast_share(self, share_hash):
84 for share in self.node.tracker.get_chain(share_hash, min(5, self.node.tracker.get_height(share_hash))):
85 if share.hash in self.shared_share_hashes:
87 self.shared_share_hashes.add(share.hash)
90 for peer in list(self.peers.itervalues()):
91 yield peer.sendShares([share for share in shares if share.peer is not peer], self.node.tracker, self.node.known_txs_var.value, include_txs_with=[share_hash])
96 self.shared_share_hashes = set(self.node.tracker.items)
97 self.node.tracker.removed.watch_weakref(self, lambda self, share: self.shared_share_hashes.discard(share.hash))
100 @defer.inlineCallbacks
101 def download_shares():
103 desired = yield self.node.desired_var.get_when_satisfies(lambda val: len(val) != 0)
104 peer2, share_hash = random.choice(desired)
106 if len(self.peers) == 0:
107 yield deferral.sleep(1)
109 peer = random.choice(self.peers.values())
111 print 'Requesting parent share %s from %s' % (p2pool_data.format_hash(share_hash), '%s:%i' % peer.addr)
113 shares = yield peer.get_shares(
119 log.err(None, 'in download_shares:')
123 yield deferral.sleep(1) # sleep so we don't keep rerequesting the same share nobody has
125 self.handle_shares(shares, peer)
128 @self.node.best_block_header.changed.watch
130 for peer in self.peers.itervalues():
131 peer.send_bestblock(header=header)
133 # send share when the chain changes to their chain
134 self.node.best_share_var.changed.watch(self.broadcast_share)
136 @self.node.tracker.verified.added.watch
138 if not (share.pow_hash <= share.header['bits'].target):
142 if (self.node.get_height_rel_highest(share.header['previous_block']) > -5 or
143 self.node.bitcoind_work.value['previous_block'] in [share.header['previous_block'], share.header_hash]):
144 self.broadcast_share(share.hash)
146 reactor.callLater(5, spread) # so get_height_rel_highest can update
150 def __init__(self, factory, bitcoind, shares, known_verified_share_hashes, net):
151 self.factory = factory
152 self.bitcoind = bitcoind
155 self.tracker = p2pool_data.OkayTracker(self.net)
158 self.tracker.add(share)
160 for share_hash in known_verified_share_hashes:
161 if share_hash in self.tracker.items:
162 self.tracker.verified.add(self.tracker.items[share_hash])
164 self.p2p_node = None # overwritten externally
166 @defer.inlineCallbacks
168 stop_signal = variable.Event()
169 self.stop = stop_signal.happened
173 self.bitcoind_work = variable.Variable((yield helper.getwork(self.bitcoind)))
174 @defer.inlineCallbacks
177 flag = self.factory.new_block.get_deferred()
179 self.bitcoind_work.set((yield helper.getwork(self.bitcoind, self.bitcoind_work.value['use_getblocktemplate'])))
182 yield defer.DeferredList([flag, deferral.sleep(15)], fireOnOneCallback=True)
187 self.best_block_header = variable.Variable(None)
188 def handle_header(new_header):
189 # check that header matches current target
190 if not (self.net.PARENT.POW_FUNC(bitcoin_data.block_header_type.pack(new_header)) <= self.bitcoind_work.value['bits'].target):
192 bitcoind_best_block = self.bitcoind_work.value['previous_block']
193 if (self.best_block_header.value is None
195 new_header['previous_block'] == bitcoind_best_block and
196 bitcoin_data.hash256(bitcoin_data.block_header_type.pack(self.best_block_header.value)) == bitcoind_best_block
197 ) # new is child of current and previous is current
199 bitcoin_data.hash256(bitcoin_data.block_header_type.pack(new_header)) == bitcoind_best_block and
200 self.best_block_header.value['previous_block'] != bitcoind_best_block
201 )): # new is current and previous is not a child of current
202 self.best_block_header.set(new_header)
203 self.handle_header = handle_header
204 @defer.inlineCallbacks
206 handle_header((yield self.factory.conn.value.get_block_header(self.bitcoind_work.value['previous_block'])))
207 self.bitcoind_work.changed.watch(lambda _: poll_header())
208 yield deferral.retry('Error while requesting best block header:')(poll_header)()
212 self.known_txs_var = variable.Variable({}) # hash -> tx
213 self.mining_txs_var = variable.Variable({}) # hash -> tx
214 self.get_height_rel_highest = yield height_tracker.get_height_rel_highest_func(self.bitcoind, self.factory, lambda: self.bitcoind_work.value['previous_block'], self.net)
216 self.best_share_var = variable.Variable(None)
217 self.desired_var = variable.Variable(None)
218 self.bitcoind_work.changed.watch(lambda _: self.set_best_share())
219 self.set_best_share()
221 # setup p2p logic and join p2pool network
223 # update mining_txs according to getwork results
224 @self.bitcoind_work.changed.run_and_watch
227 new_known_txs = dict(self.known_txs_var.value)
228 for tx_hash, tx in zip(self.bitcoind_work.value['transaction_hashes'], self.bitcoind_work.value['transactions']):
229 new_mining_txs[tx_hash] = tx
230 new_known_txs[tx_hash] = tx
231 self.mining_txs_var.set(new_mining_txs)
232 self.known_txs_var.set(new_known_txs)
233 # add p2p transactions from bitcoind to known_txs
234 @self.factory.new_tx.watch
236 new_known_txs = dict(self.known_txs_var.value)
237 new_known_txs[bitcoin_data.hash256(bitcoin_data.tx_type.pack(tx))] = tx
238 self.known_txs_var.set(new_known_txs)
239 # forward transactions seen to bitcoind
240 @self.known_txs_var.transitioned.watch
241 @defer.inlineCallbacks
242 def _(before, after):
243 yield deferral.sleep(random.expovariate(1/1))
244 for tx_hash in set(after) - set(before):
245 self.factory.conn.value.send_tx(tx=after[tx_hash])
247 @self.tracker.verified.added.watch
249 if not (share.pow_hash <= share.header['bits'].target):
252 block = share.as_block(self.tracker, self.known_txs_var.value)
254 print >>sys.stderr, 'GOT INCOMPLETE BLOCK FROM PEER! %s bitcoin: %s%064x' % (p2pool_data.format_hash(share.hash), self.net.PARENT.BLOCK_EXPLORER_URL_PREFIX, share.header_hash)
256 helper.submit_block(block, True, self.factory, self.bitcoind, self.bitcoind_work, self.net)
258 print 'GOT BLOCK FROM PEER! Passing to bitcoind! %s bitcoin: %s%064x' % (p2pool_data.format_hash(share.hash), self.net.PARENT.BLOCK_EXPLORER_URL_PREFIX, share.header_hash)
261 def forget_old_txs():
263 if self.p2p_node is not None:
264 for peer in self.p2p_node.peers.itervalues():
265 new_known_txs.update(peer.remembered_txs)
266 new_known_txs.update(self.mining_txs_var.value)
267 for share in self.tracker.get_chain(self.best_share_var.value, min(120, self.tracker.get_height(self.best_share_var.value))):
268 for tx_hash in share.new_transaction_hashes:
269 if tx_hash in self.known_txs_var.value:
270 new_known_txs[tx_hash] = self.known_txs_var.value[tx_hash]
271 self.known_txs_var.set(new_known_txs)
272 task.LoopingCall(forget_old_txs).start(10)
274 def set_best_share(self):
275 best, desired = self.tracker.think(self.get_height_rel_highest, self.bitcoind_work.value['previous_block'], self.bitcoind_work.value['bits'], self.known_txs_var.value)
277 self.best_share_var.set(best)
278 self.desired_var.set(desired)
280 def get_current_txouts(self):
281 return p2pool_data.get_expected_payouts(self.tracker, self.best_share_var.value, self.bitcoind_work.value['bits'].target, self.bitcoind_work.value['subsidy'], self.net)