X-Git-Url: https://git.novaco.in/?a=blobdiff_plain;f=p2pool%2Fbitcoin%2Fp2p.py;h=83a3403f0f94695db3153d5e8419594e26f21aaf;hb=0a3493d6873cfef4fb189d39e64dfbc6e162e2a7;hp=9905f69a2d3a365dc4e462a015776c4dead3b076;hpb=f1e5ef717505bd0c907cbe86f683c696319f0573;p=p2pool.git diff --git a/p2pool/bitcoin/p2p.py b/p2pool/bitcoin/p2p.py index 9905f69..83a3403 100644 --- a/p2pool/bitcoin/p2p.py +++ b/p2pool/bitcoin/p2p.py @@ -2,120 +2,21 @@ Implementation of Bitcoin's p2p protocol ''' -from __future__ import division - -import hashlib import random -import struct +import sys import time -from twisted.internet import defer, protocol, reactor, task -from twisted.python import log - -from . import data as bitcoin_data, getwork -from p2pool.util import variable, datachunker, deferral, forest +from twisted.internet import protocol -class TooLong(Exception): - pass +import p2pool +from . import data as bitcoin_data +from p2pool.util import deferral, p2protocol, pack, variable -class BaseProtocol(protocol.Protocol): - def connectionMade(self): - self.dataReceived = datachunker.DataChunker(self.dataReceiver()) - - def dataReceiver(self): - while True: - start = '' - while start != self._prefix: - start = (start + (yield 1))[-len(self._prefix):] - - command = (yield 12).rstrip('\0') - length, = struct.unpack(' self.max_payload_length: - print 'length too large' - continue - - if self.use_checksum: - checksum = yield 4 - else: - checksum = None - - payload = yield length - - if checksum is not None: - if hashlib.sha256(hashlib.sha256(payload).digest()).digest()[:4] != checksum: - print 'invalid hash for', repr(command), checksum.encode('hex') if checksum is not None else None, repr(payload[:100].encode('hex')), len(payload) - continue - - type_ = getattr(self, 'message_' + command, None) - if type_ is None: - print 'no type for', repr(command) - continue - - try: - payload2 = type_.unpack(payload) - except: - print 'RECV', command, checksum.encode('hex') if checksum is not None else None, repr(payload.encode('hex')), len(payload) - log.err(None, 'Error parsing message: (see RECV line)') - continue - - handler = getattr(self, 'handle_' + command, None) - if handler is None: - print 'no handler for', repr(command) - continue - - try: - handler(**payload2) - except: - print 'RECV', command, repr(payload2)[:100] - log.err(None, 'Error handling message: (see RECV line)') - continue - - self.gotPacket() - - def gotPacket(self): - pass - - def sendPacket(self, command, payload2): - if len(command) >= 12: - raise ValueError('command too long') - type_ = getattr(self, 'message_' + command, None) - if type_ is None: - raise ValueError('invalid command') - #print 'SEND', command, repr(payload2)[:500] - payload = type_.pack(payload2) - if len(payload) > self.max_payload_length: - raise TooLong('payload too long') - if self.use_checksum: - checksum = hashlib.sha256(hashlib.sha256(payload).digest()).digest()[:4] - else: - checksum = '' - data = self._prefix + struct.pack('<12sI', command, len(payload)) + checksum + payload - self.transport.write(data) - - def __getattr__(self, attr): - prefix = 'send_' - if attr.startswith(prefix): - command = attr[len(prefix):] - return lambda **payload2: self.sendPacket(command, payload2) - #return protocol.Protocol.__getattr__(self, attr) - raise AttributeError(attr) - -class Protocol(BaseProtocol): +class Protocol(p2protocol.Protocol): def __init__(self, net): - self._prefix = net.BITCOIN_P2P_PREFIX - - version = 0 - - max_payload_length = 1000000 - - @property - def use_checksum(self): - return self.version >= 209 + p2protocol.Protocol.__init__(self, net.P2P_PREFIX, 1000000, ignore_trailing_payload=True) def connectionMade(self): - BaseProtocol.connectionMade(self) - self.send_version( version=32200, services=1, @@ -131,85 +32,72 @@ class Protocol(BaseProtocol): port=self.transport.getHost().port, ), nonce=random.randrange(2**64), - sub_version_num='', + sub_version_num='/P2Pool:%s/' % (p2pool.__version__,), start_height=0, ) - message_version = bitcoin_data.ComposedType([ - ('version', bitcoin_data.StructType('>sys.stderr, 'Bitcoin connection lost. Reason:', reason.getErrorMessage() class ClientFactory(protocol.ReconnectingClientFactory): protocol = Protocol @@ -284,116 +170,3 @@ class ClientFactory(protocol.ReconnectingClientFactory): def getProtocol(self): return self.conn.get_not_none() - -class HeaderWrapper(object): - target = 2**256 - 1 - __slots__ = 'hash previous_hash'.split(' ') - - @classmethod - def from_header(cls, header): - return cls(bitcoin_data.block_header_type.hash256(header), header['previous_block']) - - def __init__(self, hash, previous_hash): - self.hash, self.previous_hash = hash, previous_hash - -class HeightTracker(object): - '''Point this at a factory and let it take care of getting block heights''' - - def __init__(self, rpc_proxy, factory, backlog_needed=1000): - self._rpc_proxy = rpc_proxy - self._factory = factory - self._backlog_needed = backlog_needed - - self._tracker = forest.Tracker() - - self._watch1 = self._factory.new_headers.watch(self._heard_headers) - self._watch2 = self._factory.new_block.watch(self._heard_block) - - self._requested = set() - self._clear_task = task.LoopingCall(self._requested.clear) - self._clear_task.start(60) - - self._last_notified_size = 0 - - self.updated = variable.Event() - - self._think_task = task.LoopingCall(self._think) - self._think_task.start(15) - self._think2_task = task.LoopingCall(self._think2) - self._think2_task.start(15) - - def _think(self): - try: - highest_head = max(self._tracker.heads, key=lambda h: self._tracker.get_height_and_last(h)[0]) if self._tracker.heads else None - if highest_head is None: - return # wait for think2 - height, last = self._tracker.get_height_and_last(highest_head) - if height < self._backlog_needed: - self._request(last) - except: - log.err(None, 'Error in HeightTracker._think:') - - @defer.inlineCallbacks - def _think2(self): - try: - ba = getwork.BlockAttempt.from_getwork((yield self._rpc_proxy.rpc_getwork())) - self._request(ba.previous_block) - except: - log.err(None, 'Error in HeightTracker._think2:') - - def _heard_headers(self, headers): - changed = False - for header in headers: - hw = HeaderWrapper.from_header(header) - if hw.hash in self._tracker.shares: - continue - changed = True - self._tracker.add(hw) - if changed: - self.updated.happened() - self._think() - - if len(self._tracker.shares) >= self._last_notified_size + 100: - print 'Have %i/%i block headers' % (len(self._tracker.shares), self._backlog_needed) - self._last_notified_size = len(self._tracker.shares) - - def _heard_block(self, block_hash): - self._request(block_hash) - - @defer.inlineCallbacks - def _request(self, last): - if last in self._tracker.shares: - return - if last in self._requested: - return - self._requested.add(last) - (yield self._factory.getProtocol()).send_getheaders(version=1, have=[], last=last) - - def get_height_rel_highest(self, block_hash): - # callers: highest height can change during yields! - height, last = self._tracker.get_height_and_last(block_hash) - if last not in self._tracker.tails: - return -1e300 - return height - max(self._tracker.get_height(head_hash) for head_hash in self._tracker.tails[last]) - - def stop(self): - self._factory.new_headers.unwatch(self._watch1) - self._factory.new_block.unwatch(self._watch2) - self._clear_task.stop() - self._think_task.stop() - self._think2_task.stop() - -if __name__ == '__main__': - factory = ClientFactory(bitcoin_data.BitcoinMainnet) - reactor.connectTCP('127.0.0.1', 8333, factory) - h = HeightTracker(factory) - - @repr - @apply - @defer.inlineCallbacks - def think(): - while True: - yield deferral.sleep(1) - print h.get_min_height(0xa285c3cb2a90ac7194cca034512748289e2526d9d7ae6ee7523) - - reactor.run()