backends/bitcoind/blockchain_processor.py

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