merklemaker.py

   1 from binascii import a2b_hex
   2 from bitcoin.script import countSigOps
   3 from bitcoin.txn import Txn
   4 from collections import deque
   5 from queue import Queue
   6 import jsonrpc
   7 import logging
   8 from merkletree import MerkleTree
   9 import threading
  10 from time import sleep, time
  11 import traceback
  12
  13 clearMerkleTree = MerkleTree([None])
  14 clearMerkleTree.coinbaseValue = 5000000000  # FIXME
  15
  16 class merkleMaker(threading.Thread):
  17         def __init__(self, *a, **k):
  18                 super().__init__(*a, **k)
  19                 self.daemon = True
  20                 self.logger = logging.getLogger('merkleMaker')
  21
  22         def _prepare(self):
  23                 self.access = jsonrpc.ServiceProxy(self.UpstreamURI)
  24
  25                 self.currentBlock = (None, None)
  26                 self.currentMerkleTree = None
  27                 self.merkleRoots = deque(maxlen=self.WorkQueueSizeRegular[1])
  28                 self.clearMerkleRoots = Queue(self.WorkQueueSizeLongpoll[1])
  29
  30                 self.nextMerkleUpdate = 0
  31                 global now
  32                 now = time()
  33                 self.updateMerkleTree()
  34
  35         def updateMerkleTree(self):
  36                 global now
  37                 self.logger.debug('Polling bitcoind for memorypool')
  38                 self.nextMerkleUpdate = now + self.TxnUpdateRetryWait
  39                 MP = self.access.getmemorypool()
  40                 prevBlock = a2b_hex(MP['previousblockhash'])[::-1]
  41                 if prevBlock != self.currentBlock[0]:
  42                         self.logger.debug('New block: %s' % (MP['previousblockhash'],))
  43                         self.merkleRoots.clear()
  44                         tmpMT = MerkleTree([None])
  45                         tmpMT.coinbaseValue = 5000000000  # FIXME
  46                         self.currentMerkleTree = tmpMT
  47                         bits = a2b_hex(MP['bits'])[::-1]
  48                         self.lastBlock = self.currentBlock
  49                         self.currentBlock = (prevBlock, bits)
  50                         self.onBlockChange()
  51                 # TODO: cache Txn or at least txid from previous merkle roots?
  52                 txnlist = map(a2b_hex, MP['transactions'])
  53
  54                 txnlistsz = sum(map(len, txnlist))
  55                 while txnlistsz > 934464:  # TODO: 1 "MB" limit - 64 KB breathing room
  56                         self.logger.debug('Trimming transaction for size limit')
  57                         txnlistsz -= len(txnlist.pop())
  58
  59                 txnlistsz = sum(map(countSigOps, txnlist))
  60                 while txnlistsz > 19488:  # TODO: 20k limit - 0x200 breathing room
  61                         self.logger.debug('Trimming transaction for SigOp limit')
  62                         txnlistsz -= countSigOps(txnlist.pop())
  63
  64                 txnlist = map(Txn, txnlist)
  65                 txnlist = [None] + list(txnlist)
  66                 newMerkleTree = MerkleTree(txnlist)
  67                 if newMerkleTree.withFirst(b'') != self.currentMerkleTree.withFirst(b''):
  68                         self.logger.debug('Updating merkle tree')
  69                         newMerkleTree.coinbaseValue = MP['coinbasevalue']
  70                         self.currentMerkleTree = newMerkleTree
  71                 self.nextMerkleUpdate = now + self.MinimumTxnUpdateWait
  72
  73         def makeMerkleRoot(self, merkleTree):
  74                 coinbaseTxn = self.makeCoinbaseTxn(merkleTree.coinbaseValue)
  75                 merkleRoot = merkleTree.withFirst(coinbaseTxn)
  76                 return (merkleRoot, merkleTree, coinbaseTxn)
  77
  78         _doing_last = None
  79         def _doing(self, what):
  80                 if self._doing_last == what:
  81                         self._doing_i += 1
  82                         return
  83                 global now
  84                 if self._doing_last:
  85                         self.logger.debug("Switching from (%4dx in %5.3f seconds) %s => %s" % (self._doing_i, now - self._doing_s, self._doing_last, what))
  86                 self._doing_last = what
  87                 self._doing_i = 1
  88                 self._doing_s = now
  89
  90         def merkleMaker_I(self):
  91                 global now
  92
  93                 # First, update merkle tree if we haven't for a while and aren't crunched for time
  94                 now = time()
  95                 if self.nextMerkleUpdate <= now and self.clearMerkleRoots.qsize() > self.WorkQueueSizeLongpoll[0] and len(self.merkleRoots) > self.WorkQueueSizeRegular[0]:
  96                         self.updateMerkleTree()
  97                 # Next, fill up the longpoll queue first, since it can be used as a failover for the main queue
  98                 elif not self.clearMerkleRoots.full():
  99                         self._doing('blank merkle roots')
 100                         self.clearMerkleRoots.put(self.makeMerkleRoot(clearMerkleTree))
 101                 # Next, fill up the main queue (until they're all current)
 102                 elif len(self.merkleRoots) < self.WorkQueueSizeRegular[1] or self.merkleRoots[0][1] != self.currentMerkleTree:
 103                         self._doing('regular merkle roots')
 104                         self.merkleRoots.append(self.makeMerkleRoot(self.currentMerkleTree))
 105                 else:
 106                         self._doing('idle')
 107                         # TODO: rather than sleepspin, block until MinimumTxnUpdateWait expires or threading.Condition(?)
 108                         sleep(self.IdleSleepTime)
 109
 110         def run(self):
 111                 while True:
 112                         try:
 113                                 self.merkleMaker_I()
 114                                 self._THISISUGLY._flushrecv()
 115                         except:
 116                                 self.logger.critical(traceback.format_exc())
 117
 118         def start(self, *a, **k):
 119                 self._prepare()
 120                 super().start(*a, **k)
 121
 122         def getMRD(self):
 123                 (prevBlock, bits) = self.currentBlock
 124                 try:
 125                         MRD = self.merkleRoots.pop()
 126                         rollPrevBlk = False
 127                 except IndexError:
 128                         MRD = self.clearMerkleRoots.get()
 129                         rollPrevBlk = True
 130                 (merkleRoot, merkleTree, coinbaseTxn) = MRD
 131                 return (merkleRoot, merkleTree, coinbaseTxn, prevBlock, bits, rollPrevBlk)