backends/bitcoind/blockchain_processor.py

   1 import ast
   2 import hashlib
   3 from json import dumps, loads
   4 import leveldb
   5 import os
   6 from Queue import Queue
   7 import random
   8 import sys
   9 import time
  10 import threading
  11 import traceback
  12 import urllib
  13
  14 from backends.bitcoind import deserialize
  15 from processor import Processor, print_log
  16 from utils import *
  17
  18
  19 class BlockchainProcessor(Processor):
  20
  21     def __init__(self, config, shared):
  22         Processor.__init__(self)
  23
  24         self.shared = shared
  25         self.config = config
  26         self.up_to_date = False
  27         self.watched_addresses = []
  28         self.history_cache = {}
  29         self.chunk_cache = {}
  30         self.cache_lock = threading.Lock()
  31         self.headers_data = ''
  32
  33         self.mempool_addresses = {}
  34         self.mempool_hist = {}
  35         self.mempool_hashes = []
  36         self.mempool_lock = threading.Lock()
  37
  38         self.address_queue = Queue()
  39         self.dbpath = config.get('leveldb', 'path')
  40
  41         self.dblock = threading.Lock()
  42         try:
  43             self.db = leveldb.LevelDB(self.dbpath)
  44         except:
  45             traceback.print_exc(file=sys.stdout)
  46             self.shared.stop()
  47
  48         self.bitcoind_url = 'http://%s:%s@%s:%s/' % (
  49             config.get('bitcoind', 'user'),
  50             config.get('bitcoind', 'password'),
  51             config.get('bitcoind', 'host'),
  52             config.get('bitcoind', 'port'))
  53
  54         self.height = 0
  55         self.is_test = False
  56         self.sent_height = 0
  57         self.sent_header = None
  58
  59         try:
  60             hist = self.deserialize(self.db.Get('height'))
  61             self.last_hash, self.height, _ = hist[0]
  62             print_log("hist", hist)
  63         except:
  64             #traceback.print_exc(file=sys.stdout)
  65             print_log('initializing database')
  66             self.height = 0
  67             self.last_hash = '000000000019d6689c085ae165831e934ff763ae46a2a6c172b3f1b60a8ce26f'
  68
  69         # catch_up headers
  70         self.init_headers(self.height)
  71
  72         threading.Timer(0, lambda: self.catch_up(sync=False)).start()
  73         while not shared.stopped() and not self.up_to_date:
  74             try:
  75                 time.sleep(1)
  76             except:
  77                 print "keyboard interrupt: stopping threads"
  78                 shared.stop()
  79                 sys.exit(0)
  80
  81         print_log("blockchain is up to date.")
  82
  83         threading.Timer(10, self.main_iteration).start()
  84
  85     def bitcoind(self, method, params=[]):
  86         postdata = dumps({"method": method, 'params': params, 'id': 'jsonrpc'})
  87         try:
  88             respdata = urllib.urlopen(self.bitcoind_url, postdata).read()
  89         except:
  90             traceback.print_exc(file=sys.stdout)
  91             self.shared.stop()
  92
  93         r = loads(respdata)
  94         if r['error'] is not None:
  95             raise BaseException(r['error'])
  96         return r.get('result')
  97
  98     def serialize(self, h):
  99         s = ''
 100         for txid, txpos, height in h:
 101             s += txid + int_to_hex(txpos, 4) + int_to_hex(height, 4)
 102         return s.decode('hex')
 103
 104     def deserialize(self, s):
 105         h = []
 106         while s:
 107             txid = s[0:32].encode('hex')
 108             txpos = int(rev_hex(s[32:36].encode('hex')), 16)
 109             height = int(rev_hex(s[36:40].encode('hex')), 16)
 110             h.append((txid, txpos, height))
 111             s = s[40:]
 112         return h
 113
 114     def block2header(self, b):
 115         return {
 116             "block_height": b.get('height'),
 117             "version": b.get('version'),
 118             "prev_block_hash": b.get('previousblockhash'),
 119             "merkle_root": b.get('merkleroot'),
 120             "timestamp": b.get('time'),
 121             "bits": int(b.get('bits'), 16),
 122             "nonce": b.get('nonce'),
 123         }
 124
 125     def get_header(self, height):
 126         block_hash = self.bitcoind('getblockhash', [height])
 127         b = self.bitcoind('getblock', [block_hash])
 128         return self.block2header(b)
 129
 130     def init_headers(self, db_height):
 131         self.chunk_cache = {}
 132         self.headers_filename = os.path.join(self.dbpath, 'blockchain_headers')
 133
 134         if os.path.exists(self.headers_filename):
 135             height = os.path.getsize(self.headers_filename)/80 - 1   # the current height
 136             if height > 0:
 137                 prev_hash = self.hash_header(self.read_header(height))
 138             else:
 139                 prev_hash = None
 140         else:
 141             open(self.headers_filename, 'wb').close()
 142             prev_hash = None
 143             height = -1
 144
 145         if height < db_height:
 146             print_log("catching up missing headers:", height, db_height)
 147
 148         try:
 149             while height < db_height:
 150                 height = height + 1
 151                 header = self.get_header(height)
 152                 if height > 1:
 153                     assert prev_hash == header.get('prev_block_hash')
 154                 self.write_header(header, sync=False)
 155                 prev_hash = self.hash_header(header)
 156                 if (height % 1000) == 0:
 157                     print_log("headers file:", height)
 158         except KeyboardInterrupt:
 159             self.flush_headers()
 160             sys.exit()
 161
 162         self.flush_headers()
 163
 164     def hash_header(self, header):
 165         return rev_hex(Hash(header_to_string(header).decode('hex')).encode('hex'))
 166
 167     def read_header(self, block_height):
 168         if os.path.exists(self.headers_filename):
 169             with open(self.headers_filename, 'rb') as f:
 170                 f.seek(block_height * 80)
 171                 h = f.read(80)
 172             if len(h) == 80:
 173                 h = header_from_string(h)
 174                 return h
 175
 176     def read_chunk(self, index):
 177         with open(self.headers_filename, 'rb') as f:
 178             f.seek(index*2016*80)
 179             chunk = f.read(2016*80)
 180         return chunk.encode('hex')
 181
 182     def write_header(self, header, sync=True):
 183         if not self.headers_data:
 184             self.headers_offset = header.get('block_height')
 185
 186         self.headers_data += header_to_string(header).decode('hex')
 187         if sync or len(self.headers_data) > 40*100:
 188             self.flush_headers()
 189
 190         with self.cache_lock:
 191             chunk_index = header.get('block_height')/2016
 192             if self.chunk_cache.get(chunk_index):
 193                 self.chunk_cache.pop(chunk_index)
 194
 195     def pop_header(self):
 196         # we need to do this only if we have not flushed
 197         if self.headers_data:
 198             self.headers_data = self.headers_data[:-40]
 199
 200     def flush_headers(self):
 201         if not self.headers_data:
 202             return
 203         with open(self.headers_filename, 'rb+') as f:
 204             f.seek(self.headers_offset*80)
 205             f.write(self.headers_data)
 206         self.headers_data = ''
 207
 208     def get_chunk(self, i):
 209         # store them on disk; store the current chunk in memory
 210         with self.cache_lock:
 211             chunk = self.chunk_cache.get(i)
 212             if not chunk:
 213                 chunk = self.read_chunk(i)
 214                 self.chunk_cache[i] = chunk
 215
 216         return chunk
 217
 218     def get_mempool_transaction(self, txid):
 219         try:
 220             raw_tx = self.bitcoind('getrawtransaction', [txid, 0, -1])
 221         except:
 222             return None
 223
 224         vds = deserialize.BCDataStream()
 225         vds.write(raw_tx.decode('hex'))
 226
 227         return deserialize.parse_Transaction(vds, is_coinbase=False)
 228
 229     def get_history(self, addr, cache_only=False):
 230         with self.cache_lock:
 231             hist = self.history_cache.get(addr)
 232         if hist is not None:
 233             return hist
 234         if cache_only:
 235             return -1
 236
 237         with self.dblock:
 238             try:
 239                 hash_160 = bc_address_to_hash_160(addr)
 240                 hist = self.deserialize(self.db.Get(hash_160))
 241                 is_known = True
 242             except:
 243                 hist = []
 244                 is_known = False
 245
 246         # should not be necessary
 247         hist.sort(key=lambda tup: tup[1])
 248         # check uniqueness too...
 249
 250         # add memory pool
 251         with self.mempool_lock:
 252             for txid in self.mempool_hist.get(addr, []):
 253                 hist.append((txid, 0, 0))
 254
 255         hist = map(lambda x: {'tx_hash': x[0], 'height': x[2]}, hist)
 256         # add something to distinguish between unused and empty addresses
 257         if hist == [] and is_known:
 258             hist = ['*']
 259
 260         with self.cache_lock:
 261             self.history_cache[addr] = hist
 262         return hist
 263
 264     def get_status(self, addr, cache_only=False):
 265         tx_points = self.get_history(addr, cache_only)
 266         if cache_only and tx_points == -1:
 267             return -1
 268
 269         if not tx_points:
 270             return None
 271         if tx_points == ['*']:
 272             return '*'
 273         status = ''
 274         for tx in tx_points:
 275             status += tx.get('tx_hash') + ':%d:' % tx.get('height')
 276         return hashlib.sha256(status).digest().encode('hex')
 277
 278     def get_merkle(self, tx_hash, height):
 279
 280         block_hash = self.bitcoind('getblockhash', [height])
 281         b = self.bitcoind('getblock', [block_hash])
 282         tx_list = b.get('tx')
 283         tx_pos = tx_list.index(tx_hash)
 284
 285         merkle = map(hash_decode, tx_list)
 286         target_hash = hash_decode(tx_hash)
 287         s = []
 288         while len(merkle) != 1:
 289             if len(merkle) % 2:
 290                 merkle.append(merkle[-1])
 291             n = []
 292             while merkle:
 293                 new_hash = Hash(merkle[0] + merkle[1])
 294                 if merkle[0] == target_hash:
 295                     s.append(hash_encode(merkle[1]))
 296                     target_hash = new_hash
 297                 elif merkle[1] == target_hash:
 298                     s.append(hash_encode(merkle[0]))
 299                     target_hash = new_hash
 300                 n.append(new_hash)
 301                 merkle = merkle[2:]
 302             merkle = n
 303
 304         return {"block_height": height, "merkle": s, "pos": tx_pos}
 305
 306     def add_to_history(self, addr, tx_hash, tx_pos, tx_height):
 307         # keep it sorted
 308         s = (tx_hash + int_to_hex(tx_pos, 4) + int_to_hex(tx_height, 4)).decode('hex')
 309
 310         serialized_hist = self.batch_list[addr]
 311
 312         l = len(serialized_hist)/40
 313         for i in range(l-1, -1, -1):
 314             item = serialized_hist[40*i:40*(i+1)]
 315             item_height = int(rev_hex(item[36:40].encode('hex')), 16)
 316             if item_height < tx_height:
 317                 serialized_hist = serialized_hist[0:40*(i+1)] + s + serialized_hist[40*(i+1):]
 318                 break
 319         else:
 320             serialized_hist = s + serialized_hist
 321
 322         self.batch_list[addr] = serialized_hist
 323
 324         # backlink
 325         txo = (tx_hash + int_to_hex(tx_pos, 4)).decode('hex')
 326         self.batch_txio[txo] = addr
 327
 328     def remove_from_history(self, addr, tx_hash, tx_pos):
 329         txi = (tx_hash + int_to_hex(tx_pos, 4)).decode('hex')
 330
 331         if addr is None:
 332             try:
 333                 addr = self.batch_txio[txi]
 334             except:
 335                 raise BaseException(tx_hash, tx_pos)
 336
 337         serialized_hist = self.batch_list[addr]
 338
 339         l = len(serialized_hist)/40
 340         for i in range(l):
 341             item = serialized_hist[40*i:40*(i+1)]
 342             if item[0:36] == txi:
 343                 height = int(rev_hex(item[36:40].encode('hex')), 16)
 344                 serialized_hist = serialized_hist[0:40*i] + serialized_hist[40*(i+1):]
 345                 break
 346         else:
 347             hist = self.deserialize(serialized_hist)
 348             raise BaseException("prevout not found", addr, hist, tx_hash, tx_pos)
 349
 350         self.batch_list[addr] = serialized_hist
 351         return height, addr
 352
 353     def deserialize_block(self, block):
 354         txlist = block.get('tx')
 355         tx_hashes = []  # ordered txids
 356         txdict = {}     # deserialized tx
 357         is_coinbase = True
 358         for raw_tx in txlist:
 359             tx_hash = hash_encode(Hash(raw_tx.decode('hex')))
 360             tx_hashes.append(tx_hash)
 361             vds = deserialize.BCDataStream()
 362             vds.write(raw_tx.decode('hex'))
 363             tx = deserialize.parse_Transaction(vds, is_coinbase)
 364             txdict[tx_hash] = tx
 365             is_coinbase = False
 366         return tx_hashes, txdict
 367
 368     def get_undo_info(self, height):
 369         s = self.db.Get("undo%d" % (height % 100))
 370         return eval(s)
 371
 372     def write_undo_info(self, batch, height, undo_info):
 373         if self.is_test or height > self.bitcoind_height - 100:
 374             batch.Put("undo%d" % (height % 100), repr(undo_info))
 375
 376     def import_block(self, block, block_hash, block_height, sync, revert=False):
 377
 378         self.batch_list = {}  # address -> history
 379         self.batch_txio = {}  # transaction i/o -> address
 380
 381         block_inputs = []
 382         block_outputs = []
 383         addr_to_read = []
 384
 385         # deserialize transactions
 386         t0 = time.time()
 387         tx_hashes, txdict = self.deserialize_block(block)
 388
 389         t00 = time.time()
 390
 391         if not revert:
 392             # read addresses of tx inputs
 393             for tx in txdict.values():
 394                 for x in tx.get('inputs'):
 395                     txi = (x.get('prevout_hash') + int_to_hex(x.get('prevout_n'), 4)).decode('hex')
 396                     block_inputs.append(txi)
 397
 398             block_inputs.sort()
 399             for txi in block_inputs:
 400                 try:
 401                     addr = self.db.Get(txi)
 402                 except:
 403                     # the input could come from the same block
 404                     continue
 405                 self.batch_txio[txi] = addr
 406                 addr_to_read.append(addr)
 407
 408         else:
 409             for txid, tx in txdict.items():
 410                 for x in tx.get('outputs'):
 411                     txo = (txid + int_to_hex(x.get('index'), 4)).decode('hex')
 412                     block_outputs.append(txo)
 413
 414         # read histories of addresses
 415         for txid, tx in txdict.items():
 416             for x in tx.get('outputs'):
 417                 hash_160 = bc_address_to_hash_160(x.get('address'))
 418                 addr_to_read.append(hash_160)
 419
 420         addr_to_read.sort()
 421         for addr in addr_to_read:
 422             try:
 423                 self.batch_list[addr] = self.db.Get(addr)
 424             except:
 425                 self.batch_list[addr] = ''
 426
 427         if revert:
 428             undo_info = self.get_undo_info(block_height)
 429             # print "undo", block_height, undo_info
 430         else:
 431             undo_info = {}
 432
 433         # process
 434         t1 = time.time()
 435
 436         if revert:
 437             tx_hashes = tx_hashes[::-1]
 438         for txid in tx_hashes:  # must be ordered
 439             tx = txdict[txid]
 440             if not revert:
 441
 442                 undo = []
 443                 for x in tx.get('inputs'):
 444                     prevout_height, prevout_addr = self.remove_from_history(None, x.get('prevout_hash'), x.get('prevout_n'))
 445                     undo.append((prevout_height, prevout_addr))
 446                 undo_info[txid] = undo
 447
 448                 for x in tx.get('outputs'):
 449                     hash_160 = bc_address_to_hash_160(x.get('address'))
 450                     self.add_to_history(hash_160, txid, x.get('index'), block_height)
 451
 452             else:
 453                 for x in tx.get('outputs'):
 454                     hash_160 = bc_address_to_hash_160(x.get('address'))
 455                     self.remove_from_history(hash_160, txid, x.get('index'))
 456
 457                 i = 0
 458                 for x in tx.get('inputs'):
 459                     prevout_height, prevout_addr = undo_info.get(txid)[i]
 460                     i += 1
 461
 462                     # read the history into batch list
 463                     if self.batch_list.get(prevout_addr) is None:
 464                         self.batch_list[prevout_addr] = self.db.Get(prevout_addr)
 465
 466                     # re-add them to the history
 467                     self.add_to_history(prevout_addr, x.get('prevout_hash'), x.get('prevout_n'), prevout_height)
 468                     # print_log("new hist for", hash_160_to_bc_address(prevout_addr), self.deserialize(self.batch_list[prevout_addr]) )
 469
 470         # write
 471         max_len = 0
 472         max_addr = ''
 473         t2 = time.time()
 474
 475         batch = leveldb.WriteBatch()
 476         for addr, serialized_hist in self.batch_list.items():
 477             batch.Put(addr, serialized_hist)
 478             l = len(serialized_hist)
 479             if l > max_len:
 480                 max_len = l
 481                 max_addr = addr
 482
 483         if not revert:
 484             # add new created outputs
 485             for txio, addr in self.batch_txio.items():
 486                 batch.Put(txio, addr)
 487             # delete spent inputs
 488             for txi in block_inputs:
 489                 batch.Delete(txi)
 490             # add undo info
 491             self.write_undo_info(batch, block_height, undo_info)
 492         else:
 493             # restore spent inputs
 494             for txio, addr in self.batch_txio.items():
 495                 batch.Put(txio, addr)
 496             # delete spent outputs
 497             for txo in block_outputs:
 498                 batch.Delete(txo)
 499
 500         # add the max
 501         batch.Put('height', self.serialize([(block_hash, block_height, 0)]))
 502
 503         # actual write
 504         self.db.Write(batch, sync=sync)
 505
 506         t3 = time.time()
 507         if t3 - t0 > 10 and not sync:
 508             print_log("block", block_height,
 509                       "parse:%0.2f " % (t00 - t0),
 510                       "read:%0.2f " % (t1 - t00),
 511                       "proc:%.2f " % (t2-t1),
 512                       "write:%.2f " % (t3-t2),
 513                       "max:", max_len, hash_160_to_bc_address(max_addr))
 514
 515         for h160 in self.batch_list.keys():
 516             addr = hash_160_to_bc_address(h160)
 517             self.invalidate_cache(addr)
 518
 519     def add_request(self, request):
 520         # see if we can get if from cache. if not, add to queue
 521         if self.process(request, cache_only=True) == -1:
 522             self.queue.put(request)
 523
 524     def process(self, request, cache_only=False):
 525         #print "abe process", request
 526
 527         message_id = request['id']
 528         method = request['method']
 529         params = request.get('params', [])
 530         result = None
 531         error = None
 532
 533         if method == 'blockchain.numblocks.subscribe':
 534             result = self.height
 535
 536         elif method == 'blockchain.headers.subscribe':
 537             result = self.header
 538
 539         elif method == 'blockchain.address.subscribe':
 540             try:
 541                 address = params[0]
 542                 result = self.get_status(address, cache_only)
 543                 self.watch_address(address)
 544             except BaseException, e:
 545                 error = str(e) + ': ' + address
 546                 print_log("error:", error)
 547
 548         elif method == 'blockchain.address.unsubscribe':
 549             try:
 550                 password = params[0]
 551                 address = params[1]
 552                 if password == self.config.get('server', 'password'):
 553                     self.watched_addresses.remove(address)
 554                     print_log('unsubscribed', address)
 555                     result = "ok"
 556                 else:
 557                     print_log('incorrect password')
 558                     result = "authentication error"
 559             except BaseException, e:
 560                 error = str(e) + ': ' + address
 561                 print_log("error:", error)
 562
 563         elif method == 'blockchain.address.get_history':
 564             try:
 565                 address = params[0]
 566                 result = self.get_history(address, cache_only)
 567             except BaseException, e:
 568                 error = str(e) + ': ' + address
 569                 print_log("error:", error)
 570
 571         elif method == 'blockchain.block.get_header':
 572             if cache_only:
 573                 result = -1
 574             else:
 575                 try:
 576                     height = params[0]
 577                     result = self.get_header(height)
 578                 except BaseException, e:
 579                     error = str(e) + ': %d' % height
 580                     print_log("error:", error)
 581
 582         elif method == 'blockchain.block.get_chunk':
 583             if cache_only:
 584                 result = -1
 585             else:
 586                 try:
 587                     index = params[0]
 588                     result = self.get_chunk(index)
 589                 except BaseException, e:
 590                     error = str(e) + ': %d' % index
 591                     print_log("error:", error)
 592
 593         elif method == 'blockchain.transaction.broadcast':
 594             try:
 595                 txo = self.bitcoind('sendrawtransaction', params)
 596                 print_log("sent tx:", txo)
 597                 result = txo
 598             except BaseException, e:
 599                 result = str(e)  # do not send an error
 600                 print_log("error:", result, params)
 601
 602         elif method == 'blockchain.transaction.get_merkle':
 603             if cache_only:
 604                 result = -1
 605             else:
 606                 try:
 607                     tx_hash = params[0]
 608                     tx_height = params[1]
 609                     result = self.get_merkle(tx_hash, tx_height)
 610                 except BaseException, e:
 611                     error = str(e) + ': ' + tx_hash
 612                     print_log("error:", error)
 613
 614         elif method == 'blockchain.transaction.get':
 615             try:
 616                 tx_hash = params[0]
 617                 height = params[1]
 618                 result = self.bitcoind('getrawtransaction', [tx_hash, 0, height])
 619             except BaseException, e:
 620                 error = str(e) + ': ' + tx_hash
 621                 print_log("error:", error)
 622
 623         else:
 624             error = "unknown method:%s" % method
 625
 626         if cache_only and result == -1:
 627             return -1
 628
 629         if error:
 630             self.push_response({'id': message_id, 'error': error})
 631         elif result != '':
 632             self.push_response({'id': message_id, 'result': result})
 633
 634     def watch_address(self, addr):
 635         if addr not in self.watched_addresses:
 636             self.watched_addresses.append(addr)
 637
 638     def catch_up(self, sync=True):
 639         t1 = time.time()
 640
 641         while not self.shared.stopped():
 642             # are we done yet?
 643             info = self.bitcoind('getinfo')
 644             self.bitcoind_height = info.get('blocks')
 645             bitcoind_block_hash = self.bitcoind('getblockhash', [self.bitcoind_height])
 646             if self.last_hash == bitcoind_block_hash:
 647                 self.up_to_date = True
 648                 break
 649
 650             # not done..
 651             self.up_to_date = False
 652             next_block_hash = self.bitcoind('getblockhash', [self.height + 1])
 653             next_block = self.bitcoind('getblock', [next_block_hash, 1])
 654
 655             # fixme: this is unsafe, if we revert when the undo info is not yet written
 656             revert = (random.randint(1, 100) == 1) if self.is_test else False
 657
 658             if (next_block.get('previousblockhash') == self.last_hash) and not revert:
 659
 660                 self.import_block(next_block, next_block_hash, self.height+1, sync)
 661                 self.height = self.height + 1
 662                 self.write_header(self.block2header(next_block), sync)
 663                 self.last_hash = next_block_hash
 664
 665                 if self.height % 100 == 0 and not sync:
 666                     t2 = time.time()
 667                     print_log("catch_up: block %d (%.3fs)" % (self.height, t2 - t1))
 668                     t1 = t2
 669
 670             else:
 671                 # revert current block
 672                 block = self.bitcoind('getblock', [self.last_hash, 1])
 673                 print_log("blockchain reorg", self.height, block.get('previousblockhash'), self.last_hash)
 674                 self.import_block(block, self.last_hash, self.height, sync, revert=True)
 675                 self.pop_header()
 676                 self.flush_headers()
 677
 678                 self.height -= 1
 679
 680                 # read previous header from disk
 681                 self.header = self.read_header(self.height)
 682                 self.last_hash = self.hash_header(self.header)
 683
 684         self.header = self.block2header(self.bitcoind('getblock', [self.last_hash]))
 685
 686     def memorypool_update(self):
 687         mempool_hashes = self.bitcoind('getrawmempool')
 688
 689         for tx_hash in mempool_hashes:
 690             if tx_hash in self.mempool_hashes:
 691                 continue
 692
 693             tx = self.get_mempool_transaction(tx_hash)
 694             if not tx:
 695                 continue
 696
 697             for x in tx.get('inputs'):
 698                 txi = (x.get('prevout_hash') + int_to_hex(x.get('prevout_n'), 4)).decode('hex')
 699                 try:
 700                     h160 = self.db.Get(txi)
 701                     addr = hash_160_to_bc_address(h160)
 702                 except:
 703                     continue
 704                 l = self.mempool_addresses.get(tx_hash, [])
 705                 if addr not in l:
 706                     l.append(addr)
 707                     self.mempool_addresses[tx_hash] = l
 708
 709             for x in tx.get('outputs'):
 710                 addr = x.get('address')
 711                 l = self.mempool_addresses.get(tx_hash, [])
 712                 if addr not in l:
 713                     l.append(addr)
 714                     self.mempool_addresses[tx_hash] = l
 715
 716             self.mempool_hashes.append(tx_hash)
 717
 718         # remove older entries from mempool_hashes
 719         self.mempool_hashes = mempool_hashes
 720
 721         # remove deprecated entries from mempool_addresses
 722         for tx_hash, addresses in self.mempool_addresses.items():
 723             if tx_hash not in self.mempool_hashes:
 724                 self.mempool_addresses.pop(tx_hash)
 725
 726         # rebuild histories
 727         new_mempool_hist = {}
 728         for tx_hash, addresses in self.mempool_addresses.items():
 729             for addr in addresses:
 730                 h = new_mempool_hist.get(addr, [])
 731                 if tx_hash not in h:
 732                     h.append(tx_hash)
 733                 new_mempool_hist[addr] = h
 734
 735         for addr in new_mempool_hist.keys():
 736             if addr in self.mempool_hist.keys():
 737                 if self.mempool_hist[addr] != new_mempool_hist[addr]:
 738                     self.invalidate_cache(addr)
 739             else:
 740                 self.invalidate_cache(addr)
 741
 742         with self.mempool_lock:
 743             self.mempool_hist = new_mempool_hist
 744
 745     def invalidate_cache(self, address):
 746         with self.cache_lock:
 747             if 'address' in self.history_cache:
 748                 print_log("cache: invalidating", address)
 749                 self.history_cache.pop(address)
 750
 751         if address in self.watched_addresses:
 752             self.address_queue.put(address)
 753
 754     def main_iteration(self):
 755         if self.shared.stopped():
 756             print_log("blockchain processor terminating")
 757             return
 758
 759         with self.dblock:
 760             t1 = time.time()
 761             self.catch_up()
 762             t2 = time.time()
 763
 764         self.memorypool_update()
 765         t3 = time.time()
 766         # print "mempool:", len(self.mempool_addresses), len(self.mempool_hist), "%.3fs"%(t3 - t2)
 767
 768         if self.sent_height != self.height:
 769             self.sent_height = self.height
 770             self.push_response({
 771                 'id': None,
 772                 'method': 'blockchain.numblocks.subscribe',
 773                 'params': [self.height],
 774             })
 775
 776         if self.sent_header != self.header:
 777             print_log("blockchain: %d (%.3fs)" % (self.height, t2 - t1))
 778             self.sent_header = self.header
 779             self.push_response({
 780                 'id': None,
 781                 'method': 'blockchain.headers.subscribe',
 782                 'params': [self.header],
 783             })
 784
 785         while True:
 786             try:
 787                 addr = self.address_queue.get(False)
 788             except:
 789                 break
 790             if addr in self.watched_addresses:
 791                 status = self.get_status(addr)
 792                 self.push_response({
 793                     'id': None,
 794                     'method': 'blockchain.address.subscribe',
 795                     'params': [addr, status],
 796                 })
 797
 798         if not self.shared.stopped():
 799             threading.Timer(10, self.main_iteration).start()
 800         else:
 801             print_log("blockchain processor terminating")