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
  18 class BlockchainProcessor(Processor):
  19
  20     def __init__(self, config, shared):
  21         Processor.__init__(self)
  22
  23         self.shared = shared
  24         self.config = config
  25         self.up_to_date = False
  26
  27         self.watch_lock = threading.Lock()
  28         self.watch_blocks = []
  29         self.watch_headers = []
  30         self.watched_addresses = {}
  31
  32         self.history_cache = {}
  33         self.chunk_cache = {}
  34         self.cache_lock = threading.Lock()
  35         self.headers_data = ''
  36
  37         self.mempool_addresses = {}
  38         self.mempool_hist = {}
  39         self.mempool_hashes = set([])
  40         self.mempool_lock = threading.Lock()
  41
  42         self.address_queue = Queue()
  43
  44         try:
  45             self.use_plyvel = config.getboolean('leveldb', 'use_plyvel')
  46         except:
  47             self.use_plyvel = False
  48         print_log('use_plyvel:', self.use_plyvel)
  49
  50         # don't use the same database for plyvel, because python-leveldb uses snappy compression
  51         self.dbpath = config.get('leveldb', 'path_plyvel' if self.use_plyvel else 'path')
  52
  53         self.pruning_limit = config.getint('leveldb', 'pruning_limit')
  54         self.db_version = 1 # increase this when database needs to be updated
  55
  56         self.dblock = threading.Lock()
  57         try:
  58             if self.use_plyvel:
  59                 import plyvel
  60                 self.db = plyvel.DB(self.dbpath, create_if_missing=True, paranoid_checks=None, compression=None)
  61             else:
  62                 import leveldb
  63                 self.db = leveldb.LevelDB(self.dbpath, paranoid_checks=False)
  64         except:
  65             traceback.print_exc(file=sys.stdout)
  66             self.shared.stop()
  67
  68         self.bitcoind_url = 'http://%s:%s@%s:%s/' % (
  69             config.get('bitcoind', 'user'),
  70             config.get('bitcoind', 'password'),
  71             config.get('bitcoind', 'host'),
  72             config.get('bitcoind', 'port'))
  73
  74         while True:
  75             try:
  76                 self.bitcoind('getinfo')
  77                 break
  78             except:
  79                 print_log('cannot contact bitcoind...')
  80                 time.sleep(5)
  81                 continue
  82
  83         self.height = 0
  84         self.is_test = False
  85         self.sent_height = 0
  86         self.sent_header = None
  87
  88         try:
  89             hist = self.deserialize(self.db_get('height'))
  90             self.last_hash, self.height, db_version = hist[0]
  91             print_log("Database version", self.db_version)
  92             print_log("Blockchain height", self.height)
  93         except:
  94             traceback.print_exc(file=sys.stdout)
  95             print_log('initializing database')
  96             self.height = 0
  97             self.last_hash = '000000000019d6689c085ae165831e934ff763ae46a2a6c172b3f1b60a8ce26f'
  98             db_version = self.db_version
  99
 100         # check version
 101         if self.db_version != db_version:
 102             print_log("Your database '%s' is deprecated. Please create a new database"%self.dbpath)
 103             self.shared.stop()
 104             return
 105
 106         # catch_up headers
 107         self.init_headers(self.height)
 108
 109         threading.Timer(0, lambda: self.catch_up(sync=False)).start()
 110         while not shared.stopped() and not self.up_to_date:
 111             try:
 112                 time.sleep(1)
 113             except:
 114                 print "keyboard interrupt: stopping threads"
 115                 shared.stop()
 116                 sys.exit(0)
 117
 118         print_log("Blockchain is up to date.")
 119         self.memorypool_update()
 120         print_log("Memory pool initialized.")
 121
 122         threading.Timer(10, self.main_iteration).start()
 123
 124
 125     def db_get(self, key):
 126         if self.use_plyvel:
 127             return self.db.get(key)
 128         else:
 129             try:
 130                 return self.db.Get(key)
 131             except KeyError:
 132                 return None
 133
 134     def batch_put(self, batch, key, value):
 135         if self.use_plyvel:
 136             batch.put(key, value)
 137         else:
 138             batch.Put(key, value)
 139
 140     def batch_delete(self, batch, key):
 141         if self.use_plyvel:
 142             batch.delete(key)
 143         else:
 144             batch.Delete(key)
 145
 146     def batch_write(self, batch, sync):
 147         if self.use_plyvel:
 148             batch.write()#, sync=sync)
 149         else:
 150             self.db.Write(batch, sync=sync)
 151
 152
 153     def bitcoind(self, method, params=[]):
 154         postdata = dumps({"method": method, 'params': params, 'id': 'jsonrpc'})
 155         try:
 156             respdata = urllib.urlopen(self.bitcoind_url, postdata).read()
 157         except:
 158             traceback.print_exc(file=sys.stdout)
 159             self.shared.stop()
 160
 161         r = loads(respdata)
 162         if r['error'] is not None:
 163             raise BaseException(r['error'])
 164         return r.get('result')
 165
 166     def serialize(self, h):
 167         s = ''
 168         for txid, txpos, height in h:
 169             s += self.serialize_item(txid, txpos, height)
 170         return s
 171
 172     def serialize_item(self, txid, txpos, height, spent=chr(0)):
 173         s = (txid + int_to_hex(txpos, 4) + int_to_hex(height, 3)).decode('hex') + spent
 174         return s
 175
 176     def deserialize_item(self,s):
 177         txid = s[0:32].encode('hex')
 178         txpos = int(rev_hex(s[32:36].encode('hex')), 16)
 179         height = int(rev_hex(s[36:39].encode('hex')), 16)
 180         spent = s[39:40]
 181         return (txid, txpos, height, spent)
 182
 183     def deserialize(self, s):
 184         h = []
 185         while s:
 186             txid, txpos, height, spent = self.deserialize_item(s[0:40])
 187             h.append((txid, txpos, height))
 188             if spent == chr(1):
 189                 txid, txpos, height, spent = self.deserialize_item(s[40:80])
 190                 h.append((txid, txpos, height))
 191             s = s[80:]
 192         return h
 193
 194     def block2header(self, b):
 195         return {
 196             "block_height": b.get('height'),
 197             "version": b.get('version'),
 198             "prev_block_hash": b.get('previousblockhash'),
 199             "merkle_root": b.get('merkleroot'),
 200             "timestamp": b.get('time'),
 201             "bits": int(b.get('bits'), 16),
 202             "nonce": b.get('nonce'),
 203         }
 204
 205     def get_header(self, height):
 206         block_hash = self.bitcoind('getblockhash', [height])
 207         b = self.bitcoind('getblock', [block_hash])
 208         return self.block2header(b)
 209
 210     def init_headers(self, db_height):
 211         self.chunk_cache = {}
 212         self.headers_filename = os.path.join(self.dbpath, 'blockchain_headers')
 213
 214         if os.path.exists(self.headers_filename):
 215             height = os.path.getsize(self.headers_filename)/80 - 1   # the current height
 216             if height > 0:
 217                 prev_hash = self.hash_header(self.read_header(height))
 218             else:
 219                 prev_hash = None
 220         else:
 221             open(self.headers_filename, 'wb').close()
 222             prev_hash = None
 223             height = -1
 224
 225         if height < db_height:
 226             print_log("catching up missing headers:", height, db_height)
 227
 228         try:
 229             while height < db_height:
 230                 height = height + 1
 231                 header = self.get_header(height)
 232                 if height > 1:
 233                     assert prev_hash == header.get('prev_block_hash')
 234                 self.write_header(header, sync=False)
 235                 prev_hash = self.hash_header(header)
 236                 if (height % 1000) == 0:
 237                     print_log("headers file:", height)
 238         except KeyboardInterrupt:
 239             self.flush_headers()
 240             sys.exit()
 241
 242         self.flush_headers()
 243
 244     def hash_header(self, header):
 245         return rev_hex(Hash(header_to_string(header).decode('hex')).encode('hex'))
 246
 247     def read_header(self, block_height):
 248         if os.path.exists(self.headers_filename):
 249             with open(self.headers_filename, 'rb') as f:
 250                 f.seek(block_height * 80)
 251                 h = f.read(80)
 252             if len(h) == 80:
 253                 h = header_from_string(h)
 254                 return h
 255
 256     def read_chunk(self, index):
 257         with open(self.headers_filename, 'rb') as f:
 258             f.seek(index*2016*80)
 259             chunk = f.read(2016*80)
 260         return chunk.encode('hex')
 261
 262     def write_header(self, header, sync=True):
 263         if not self.headers_data:
 264             self.headers_offset = header.get('block_height')
 265
 266         self.headers_data += header_to_string(header).decode('hex')
 267         if sync or len(self.headers_data) > 40*100:
 268             self.flush_headers()
 269
 270         with self.cache_lock:
 271             chunk_index = header.get('block_height')/2016
 272             if self.chunk_cache.get(chunk_index):
 273                 self.chunk_cache.pop(chunk_index)
 274
 275     def pop_header(self):
 276         # we need to do this only if we have not flushed
 277         if self.headers_data:
 278             self.headers_data = self.headers_data[:-40]
 279
 280     def flush_headers(self):
 281         if not self.headers_data:
 282             return
 283         with open(self.headers_filename, 'rb+') as f:
 284             f.seek(self.headers_offset*80)
 285             f.write(self.headers_data)
 286         self.headers_data = ''
 287
 288     def get_chunk(self, i):
 289         # store them on disk; store the current chunk in memory
 290         with self.cache_lock:
 291             chunk = self.chunk_cache.get(i)
 292             if not chunk:
 293                 chunk = self.read_chunk(i)
 294                 self.chunk_cache[i] = chunk
 295
 296         return chunk
 297
 298     def get_mempool_transaction(self, txid):
 299         try:
 300             raw_tx = self.bitcoind('getrawtransaction', [txid, 0])
 301         except:
 302             return None
 303
 304         vds = deserialize.BCDataStream()
 305         vds.write(raw_tx.decode('hex'))
 306         try:
 307             return deserialize.parse_Transaction(vds, is_coinbase=False)
 308         except:
 309             print_log("ERROR: cannot parse", txid)
 310             return None
 311
 312     def get_history(self, addr, cache_only=False):
 313         with self.cache_lock:
 314             hist = self.history_cache.get(addr)
 315         if hist is not None:
 316             return hist
 317         if cache_only:
 318             return -1
 319
 320         with self.dblock:
 321             try:
 322                 hist = self.deserialize(self.db_get(str((addr))))
 323                 is_known = True
 324             except:
 325                 self.shared.stop()
 326                 raise
 327             if hist:
 328                 is_known = True
 329             else:
 330                 hist = []
 331                 is_known = False
 332
 333         # sort history, because redeeming transactions are next to the corresponding txout
 334         hist.sort(key=lambda tup: tup[2])
 335
 336         # add memory pool
 337         with self.mempool_lock:
 338             for txid in self.mempool_hist.get(addr, []):
 339                 hist.append((txid, 0, 0))
 340
 341         # uniqueness
 342         hist = set(map(lambda x: (x[0], x[2]), hist))
 343
 344         # convert to dict
 345         hist = map(lambda x: {'tx_hash': x[0], 'height': x[1]}, hist)
 346
 347         # add something to distinguish between unused and empty addresses
 348         if hist == [] and is_known:
 349             hist = ['*']
 350
 351         with self.cache_lock:
 352             self.history_cache[addr] = hist
 353         return hist
 354
 355     def get_status(self, addr, cache_only=False):
 356         tx_points = self.get_history(addr, cache_only)
 357         if cache_only and tx_points == -1:
 358             return -1
 359
 360         if not tx_points:
 361             return None
 362         if tx_points == ['*']:
 363             return '*'
 364         status = ''
 365         for tx in tx_points:
 366             status += tx.get('tx_hash') + ':%d:' % tx.get('height')
 367         return hashlib.sha256(status).digest().encode('hex')
 368
 369     def get_merkle(self, tx_hash, height):
 370
 371         block_hash = self.bitcoind('getblockhash', [height])
 372         b = self.bitcoind('getblock', [block_hash])
 373         tx_list = b.get('tx')
 374         tx_pos = tx_list.index(tx_hash)
 375
 376         merkle = map(hash_decode, tx_list)
 377         target_hash = hash_decode(tx_hash)
 378         s = []
 379         while len(merkle) != 1:
 380             if len(merkle) % 2:
 381                 merkle.append(merkle[-1])
 382             n = []
 383             while merkle:
 384                 new_hash = Hash(merkle[0] + merkle[1])
 385                 if merkle[0] == target_hash:
 386                     s.append(hash_encode(merkle[1]))
 387                     target_hash = new_hash
 388                 elif merkle[1] == target_hash:
 389                     s.append(hash_encode(merkle[0]))
 390                     target_hash = new_hash
 391                 n.append(new_hash)
 392                 merkle = merkle[2:]
 393             merkle = n
 394
 395         return {"block_height": height, "merkle": s, "pos": tx_pos}
 396
 397
 398     def add_to_history(self, addr, tx_hash, tx_pos, tx_height):
 399         # keep it sorted
 400         s = self.serialize_item(tx_hash, tx_pos, tx_height) + 40*chr(0)
 401         assert len(s) == 80
 402
 403         serialized_hist = self.batch_list[addr]
 404
 405         l = len(serialized_hist)/80
 406         for i in range(l-1, -1, -1):
 407             item = serialized_hist[80*i:80*(i+1)]
 408             item_height = int(rev_hex(item[36:39].encode('hex')), 16)
 409             if item_height <= tx_height:
 410                 serialized_hist = serialized_hist[0:80*(i+1)] + s + serialized_hist[80*(i+1):]
 411                 break
 412         else:
 413             serialized_hist = s + serialized_hist
 414
 415         self.batch_list[addr] = serialized_hist
 416
 417         # backlink
 418         txo = (tx_hash + int_to_hex(tx_pos, 4)).decode('hex')
 419         self.batch_txio[txo] = addr
 420
 421
 422
 423     def revert_add_to_history(self, addr, tx_hash, tx_pos, tx_height):
 424
 425         serialized_hist = self.batch_list[addr]
 426         s = self.serialize_item(tx_hash, tx_pos, tx_height) + 40*chr(0)
 427         if serialized_hist.find(s) == -1: raise
 428         serialized_hist = serialized_hist.replace(s, '')
 429         self.batch_list[addr] = serialized_hist
 430
 431
 432
 433     def prune_history(self, addr, undo):
 434         # remove items that have bit set to one
 435         if undo.get(addr) is None: undo[addr] = []
 436
 437         serialized_hist = self.batch_list[addr]
 438         l = len(serialized_hist)/80
 439         for i in range(l):
 440             if len(serialized_hist)/80 < self.pruning_limit: break
 441             item = serialized_hist[80*i:80*(i+1)]
 442             if item[39:40] == chr(1):
 443                 assert item[79:80] == chr(2)
 444                 serialized_hist = serialized_hist[0:80*i] + serialized_hist[80*(i+1):]
 445                 undo[addr].append(item)  # items are ordered
 446         self.batch_list[addr] = serialized_hist
 447
 448
 449     def revert_prune_history(self, addr, undo):
 450         # restore removed items
 451         serialized_hist = self.batch_list[addr]
 452
 453         if undo.get(addr) is not None:
 454             itemlist = undo.pop(addr)
 455         else:
 456             return
 457
 458         if not itemlist: return
 459
 460         l = len(serialized_hist)/80
 461         tx_item = ''
 462         for i in range(l-1, -1, -1):
 463             if tx_item == '':
 464                 if not itemlist:
 465                     break
 466                 else:
 467                     tx_item = itemlist.pop(-1) # get the last element
 468                     tx_height = int(rev_hex(tx_item[36:39].encode('hex')), 16)
 469
 470             item = serialized_hist[80*i:80*(i+1)]
 471             item_height = int(rev_hex(item[36:39].encode('hex')), 16)
 472
 473             if item_height < tx_height:
 474                 serialized_hist = serialized_hist[0:80*(i+1)] + tx_item + serialized_hist[80*(i+1):]
 475                 tx_item = ''
 476
 477         else:
 478             serialized_hist = ''.join(itemlist) + tx_item + serialized_hist
 479
 480         self.batch_list[addr] = serialized_hist
 481
 482
 483     def set_spent_bit(self, addr, txi, is_spent, txid=None, index=None, height=None):
 484         serialized_hist = self.batch_list[addr]
 485         l = len(serialized_hist)/80
 486         for i in range(l):
 487             item = serialized_hist[80*i:80*(i+1)]
 488             if item[0:36] == txi:
 489                 if is_spent:
 490                     new_item = item[0:39] + chr(1) + self.serialize_item(txid, index, height, chr(2))
 491                 else:
 492                     new_item = item[0:39] + chr(0) + chr(0)*40
 493                 serialized_hist = serialized_hist[0:80*i] + new_item + serialized_hist[80*(i+1):]
 494                 break
 495         else:
 496             self.shared.stop()
 497             hist = self.deserialize(serialized_hist)
 498             raise BaseException("prevout not found", addr, hist, txi.encode('hex'))
 499
 500         self.batch_list[addr] = serialized_hist
 501
 502
 503     def unset_spent_bit(self, addr, txi):
 504         self.set_spent_bit(addr, txi, False)
 505         self.batch_txio[txi] = addr
 506
 507
 508     def deserialize_block(self, block):
 509         txlist = block.get('tx')
 510         tx_hashes = []  # ordered txids
 511         txdict = {}     # deserialized tx
 512         is_coinbase = True
 513         for raw_tx in txlist:
 514             tx_hash = hash_encode(Hash(raw_tx.decode('hex')))
 515             vds = deserialize.BCDataStream()
 516             vds.write(raw_tx.decode('hex'))
 517             try:
 518                 tx = deserialize.parse_Transaction(vds, is_coinbase)
 519             except:
 520                 print_log("ERROR: cannot parse", tx_hash)
 521                 continue
 522             tx_hashes.append(tx_hash)
 523             txdict[tx_hash] = tx
 524             is_coinbase = False
 525         return tx_hashes, txdict
 526
 527     def get_undo_info(self, height):
 528         s = self.db_get("undo%d" % (height % 100))
 529         return eval(s)
 530
 531     def write_undo_info(self, batch, height, undo_info):
 532         if self.is_test or height > self.bitcoind_height - 100:
 533             self.batch_put(batch, "undo%d" % (height % 100), repr(undo_info))
 534
 535     def import_block(self, block, block_hash, block_height, sync, revert=False):
 536
 537         self.batch_list = {}  # address -> history
 538         self.batch_txio = {}  # transaction i/o -> address
 539
 540         block_inputs = set([])
 541         block_outputs = set([])
 542         addr_to_read = set([])
 543
 544         # deserialize transactions
 545         t0 = time.time()
 546         tx_hashes, txdict = self.deserialize_block(block)
 547
 548         t00 = time.time()
 549
 550         # undo info
 551         if revert:
 552             undo_info = self.get_undo_info(block_height)
 553         else:
 554             undo_info = {}
 555
 556
 557         if not revert:
 558             # read addresses of tx inputs
 559             for tx in txdict.values():
 560                 for x in tx.get('inputs'):
 561                     txi = (x.get('prevout_hash') + int_to_hex(x.get('prevout_n'), 4)).decode('hex')
 562                     block_inputs.add(txi)
 563
 564             #block_inputs.sort()
 565             for txi in sorted(block_inputs):
 566                 try:
 567                     addr = self.db_get(txi)
 568                     if addr is None:
 569                         # the input could come from the same block
 570                         continue
 571                 except:
 572                     traceback.print_exc(file=sys.stdout)
 573                     self.shared.stop()
 574                     raise
 575
 576                 self.batch_txio[txi] = addr
 577                 addr_to_read.add(addr)
 578
 579         else:
 580             for txid, tx in txdict.items():
 581                 for x in tx.get('outputs'):
 582                     txo = (txid + int_to_hex(x.get('index'), 4)).decode('hex')
 583                     block_outputs.add(txo)
 584                     addr_to_read.add( x.get('address') )
 585
 586                 undo = undo_info.get(txid)
 587                 for i, x in enumerate(tx.get('inputs')):
 588                     addr = undo['prev_addr'][i]
 589                     addr_to_read.add(addr)
 590
 591
 592         #time spent reading txio
 593         t000 = time.time()
 594
 595         # read histories of addresses
 596         for txid, tx in txdict.items():
 597             for x in tx.get('outputs'):
 598                 addr_to_read.add(x.get('address'))
 599
 600         #addr_to_read.sort()
 601         for addr in sorted(addr_to_read):
 602             try:
 603                 h = self.db_get(addr)
 604                 self.batch_list[addr] = '' if h is None else h
 605             except:
 606                 print "db get error", addr
 607                 traceback.print_exc(file=sys.stdout)
 608                 self.shared.stop()
 609                 raise
 610
 611
 612         # process
 613         t1 = time.time()
 614
 615         if revert:
 616             tx_hashes = tx_hashes[::-1]
 617
 618
 619         for txid in tx_hashes:  # must be ordered
 620             tx = txdict[txid]
 621             if not revert:
 622
 623                 undo = { 'prev_addr':[] } # contains the list of pruned items for each address in the tx; also, 'prev_addr' is a list of prev addresses
 624
 625                 prev_addr = []
 626                 for i, x in enumerate(tx.get('inputs')):
 627                     txi = (x.get('prevout_hash') + int_to_hex(x.get('prevout_n'), 4)).decode('hex')
 628                     addr = self.batch_txio[txi]
 629
 630                     # add redeem item to the history.
 631                     # add it right next to the input txi? this will break history sorting, but it's ok if I neglect tx inputs during search
 632                     self.set_spent_bit(addr, txi, True, txid, i, block_height)
 633
 634                     # when I prune, prune a pair
 635                     self.prune_history(addr, undo)
 636                     prev_addr.append(addr)
 637
 638                 undo['prev_addr'] = prev_addr
 639
 640                 # here I add only the outputs to history; maybe I want to add inputs too (that's in the other loop)
 641                 for x in tx.get('outputs'):
 642                     addr = x.get('address')
 643                     self.add_to_history(addr, txid, x.get('index'), block_height)
 644                     self.prune_history(addr, undo)  # prune here because we increased the length of the history
 645
 646                 undo_info[txid] = undo
 647
 648             else:
 649
 650                 undo = undo_info.pop(txid)
 651
 652                 for x in tx.get('outputs'):
 653                     addr = x.get('address')
 654                     self.revert_prune_history(addr, undo)
 655                     self.revert_add_to_history(addr, txid, x.get('index'), block_height)
 656
 657                 prev_addr = undo.pop('prev_addr')
 658                 for i, x in enumerate(tx.get('inputs')):
 659                     addr = prev_addr[i]
 660                     self.revert_prune_history(addr, undo)
 661                     txi = (x.get('prevout_hash') + int_to_hex(x.get('prevout_n'), 4)).decode('hex')
 662                     self.unset_spent_bit(addr, txi)
 663
 664                 assert undo == {}
 665
 666         if revert:
 667             assert undo_info == {}
 668
 669
 670         # write
 671         max_len = 0
 672         max_addr = ''
 673         t2 = time.time()
 674
 675         if self.use_plyvel:
 676             batch = self.db.write_batch()
 677         else:
 678             import leveldb
 679             batch = leveldb.WriteBatch()
 680
 681         for addr, serialized_hist in self.batch_list.items():
 682             self.batch_put(batch, addr, serialized_hist)
 683             l = len(serialized_hist)/80
 684             if l > max_len:
 685                 max_len = l
 686                 max_addr = addr
 687
 688         if not revert:
 689             # add new created outputs
 690             for txio, addr in self.batch_txio.items():
 691                 self.batch_put(batch, txio, addr)
 692             # delete spent inputs
 693             for txi in block_inputs:
 694                 self.batch_delete(batch, txi)
 695             # add undo info
 696             self.write_undo_info(batch, block_height, undo_info)
 697         else:
 698             # restore spent inputs
 699             for txio, addr in self.batch_txio.items():
 700                 # print "restoring spent input", repr(txio)
 701                 self.batch_put(batch, txio, addr)
 702             # delete spent outputs
 703             for txo in block_outputs:
 704                 self.batch_delete(batch, txo)
 705
 706         # add the max
 707         self.batch_put(batch, 'height', self.serialize([(block_hash, block_height, self.db_version)]))
 708
 709         # actual write
 710         self.batch_write(batch, sync)
 711
 712         t3 = time.time()
 713         if t3 - t0 > 10 and not sync:
 714             print_log("block %d  "%block_height,
 715                       "total:%0.2f  " % (t3 - t0),
 716                       #"parse:%0.2f  " % (t00 - t0),
 717                       "read_txio[%4d]:%0.2f " % (len(block_inputs), t000 - t00),
 718                       "read_addr[%4d]:%0.2f " % (len(addr_to_read), t1 - t000),
 719                       #"proc:%.2f " % (t2-t1),
 720                       "write:%.2f " % (t3-t2),
 721                       "max:", max_len, max_addr)
 722
 723         for addr in self.batch_list.keys():
 724             self.invalidate_cache(addr)
 725
 726     def add_request(self, session, request):
 727         # see if we can get if from cache. if not, add to queue
 728         if self.process(session, request, cache_only=True) == -1:
 729             self.queue.put((session, request))
 730
 731
 732     def do_subscribe(self, method, params, session):
 733         with self.watch_lock:
 734             if method == 'blockchain.numblocks.subscribe':
 735                 if session not in self.watch_blocks:
 736                     self.watch_blocks.append(session)
 737
 738             elif method == 'blockchain.headers.subscribe':
 739                 if session not in self.watch_headers:
 740                     self.watch_headers.append(session)
 741
 742             elif method == 'blockchain.address.subscribe':
 743                 address = params[0]
 744                 l = self.watched_addresses.get(address)
 745                 if l is None:
 746                     self.watched_addresses[address] = [session]
 747                 elif session not in l:
 748                     l.append(session)
 749
 750
 751     def do_unsubscribe(self, method, params, session):
 752         with self.watch_lock:
 753             if method == 'blockchain.numblocks.subscribe':
 754                 if session in self.watch_blocks:
 755                     self.watch_blocks.remove(session)
 756             elif method == 'blockchain.headers.subscribe':
 757                 if session in self.watch_headers:
 758                     self.watch_headers.remove(session)
 759             elif method == "blockchain.address.subscribe":
 760                 addr = params[0]
 761                 l = self.watched_addresses.get(addr)
 762                 if not l:
 763                     return
 764                 if session in l:
 765                     l.remove(session)
 766                 if session in l:
 767                     print "error rc!!"
 768                     self.shared.stop()
 769                 if l == []:
 770                     self.watched_addresses.pop(addr)
 771
 772
 773     def process(self, session, request, cache_only=False):
 774
 775         message_id = request['id']
 776         method = request['method']
 777         params = request.get('params', [])
 778         result = None
 779         error = None
 780
 781         if method == 'blockchain.numblocks.subscribe':
 782             result = self.height
 783
 784         elif method == 'blockchain.headers.subscribe':
 785             result = self.header
 786
 787         elif method == 'blockchain.address.subscribe':
 788             try:
 789                 address = params[0]
 790                 result = self.get_status(address, cache_only)
 791             except BaseException, e:
 792                 error = str(e) + ': ' + address
 793                 print_log("error:", error)
 794
 795         elif method == 'blockchain.address.get_history':
 796             try:
 797                 address = params[0]
 798                 result = self.get_history(address, cache_only)
 799             except BaseException, e:
 800                 error = str(e) + ': ' + address
 801                 print_log("error:", error)
 802
 803         elif method == 'blockchain.block.get_header':
 804             if cache_only:
 805                 result = -1
 806             else:
 807                 try:
 808                     height = params[0]
 809                     result = self.get_header(height)
 810                 except BaseException, e:
 811                     error = str(e) + ': %d' % height
 812                     print_log("error:", error)
 813
 814         elif method == 'blockchain.block.get_chunk':
 815             if cache_only:
 816                 result = -1
 817             else:
 818                 try:
 819                     index = params[0]
 820                     result = self.get_chunk(index)
 821                 except BaseException, e:
 822                     error = str(e) + ': %d' % index
 823                     print_log("error:", error)
 824
 825         elif method == 'blockchain.transaction.broadcast':
 826             try:
 827                 txo = self.bitcoind('sendrawtransaction', params)
 828                 print_log("sent tx:", txo)
 829                 result = txo
 830             except BaseException, e:
 831                 result = str(e)  # do not send an error
 832                 print_log("error:", result, params)
 833
 834         elif method == 'blockchain.transaction.get_merkle':
 835             if cache_only:
 836                 result = -1
 837             else:
 838                 try:
 839                     tx_hash = params[0]
 840                     tx_height = params[1]
 841                     result = self.get_merkle(tx_hash, tx_height)
 842                 except BaseException, e:
 843                     error = str(e) + ': ' + repr(params)
 844                     print_log("get_merkle error:", error)
 845
 846         elif method == 'blockchain.transaction.get':
 847             try:
 848                 tx_hash = params[0]
 849                 result = self.bitcoind('getrawtransaction', [tx_hash, 0])
 850             except BaseException, e:
 851                 error = str(e) + ': ' + repr(params)
 852                 print_log("tx get error:", error)
 853
 854         else:
 855             error = "unknown method:%s" % method
 856
 857         if cache_only and result == -1:
 858             return -1
 859
 860         if error:
 861             self.push_response(session, {'id': message_id, 'error': error})
 862         elif result != '':
 863             self.push_response(session, {'id': message_id, 'result': result})
 864
 865
 866     def getfullblock(self, block_hash):
 867         block = self.bitcoind('getblock', [block_hash])
 868
 869         rawtxreq = []
 870         i = 0
 871         for txid in block['tx']:
 872             rawtxreq.append({
 873                 "method": "getrawtransaction",
 874                 "params": [txid],
 875                 "id": i,
 876             })
 877             i += 1
 878
 879         postdata = dumps(rawtxreq)
 880         try:
 881             respdata = urllib.urlopen(self.bitcoind_url, postdata).read()
 882         except:
 883             traceback.print_exc(file=sys.stdout)
 884             self.shared.stop()
 885
 886         r = loads(respdata)
 887         rawtxdata = []
 888         for ir in r:
 889             if ir['error'] is not None:
 890                 self.shared.stop()
 891                 print_log("Error: make sure you run bitcoind with txindex=1; use -reindex if needed.")
 892                 raise BaseException(ir['error'])
 893             rawtxdata.append(ir['result'])
 894         block['tx'] = rawtxdata
 895         return block
 896
 897     def catch_up(self, sync=True):
 898         t1 = time.time()
 899
 900         while not self.shared.stopped():
 901             # are we done yet?
 902             info = self.bitcoind('getinfo')
 903             self.bitcoind_height = info.get('blocks')
 904             bitcoind_block_hash = self.bitcoind('getblockhash', [self.bitcoind_height])
 905             if self.last_hash == bitcoind_block_hash:
 906                 self.up_to_date = True
 907                 break
 908
 909             # not done..
 910             self.up_to_date = False
 911             next_block_hash = self.bitcoind('getblockhash', [self.height + 1])
 912             next_block = self.getfullblock(next_block_hash)
 913
 914             # fixme: this is unsafe, if we revert when the undo info is not yet written
 915             revert = (random.randint(1, 100) == 1) if self.is_test else False
 916
 917             if (next_block.get('previousblockhash') == self.last_hash) and not revert:
 918
 919                 self.import_block(next_block, next_block_hash, self.height+1, sync)
 920                 self.height = self.height + 1
 921                 self.write_header(self.block2header(next_block), sync)
 922                 self.last_hash = next_block_hash
 923
 924                 if self.height % 100 == 0 and not sync:
 925                     t2 = time.time()
 926                     print_log("catch_up: block %d (%.3fs)" % (self.height, t2 - t1))
 927                     t1 = t2
 928
 929             else:
 930                 # revert current block
 931                 block = self.getfullblock(self.last_hash)
 932                 print_log("blockchain reorg", self.height, block.get('previousblockhash'), self.last_hash)
 933                 self.import_block(block, self.last_hash, self.height, sync, revert=True)
 934                 self.pop_header()
 935                 self.flush_headers()
 936
 937                 self.height -= 1
 938
 939                 # read previous header from disk
 940                 self.header = self.read_header(self.height)
 941                 self.last_hash = self.hash_header(self.header)
 942
 943         self.header = self.block2header(self.bitcoind('getblock', [self.last_hash]))
 944
 945         if self.shared.stopped() and self.use_plyvel:
 946             print_log( "closing database" )
 947             self.db.close()
 948
 949
 950     def memorypool_update(self):
 951         mempool_hashes = set(self.bitcoind('getrawmempool'))
 952         touched_addresses = set([])
 953
 954         for tx_hash in mempool_hashes:
 955             if tx_hash in self.mempool_hashes:
 956                 continue
 957
 958             tx = self.get_mempool_transaction(tx_hash)
 959             if not tx:
 960                 continue
 961
 962             mpa = self.mempool_addresses.get(tx_hash, [])
 963             for x in tx.get('inputs'):
 964                 # we assume that the input address can be parsed by deserialize(); this is true for Electrum transactions
 965                 addr = x.get('address')
 966                 if addr and addr not in mpa:
 967                     mpa.append(addr)
 968                     touched_addresses.add(addr)
 969
 970             for x in tx.get('outputs'):
 971                 addr = x.get('address')
 972                 if addr and addr not in mpa:
 973                     mpa.append(addr)
 974                     touched_addresses.add(addr)
 975
 976             self.mempool_addresses[tx_hash] = mpa
 977             self.mempool_hashes.add(tx_hash)
 978
 979         # remove older entries from mempool_hashes
 980         self.mempool_hashes = mempool_hashes
 981
 982         # remove deprecated entries from mempool_addresses
 983         for tx_hash, addresses in self.mempool_addresses.items():
 984             if tx_hash not in self.mempool_hashes:
 985                 self.mempool_addresses.pop(tx_hash)
 986                 for addr in addresses:
 987                     touched_addresses.add(addr)
 988
 989         # rebuild mempool histories
 990         new_mempool_hist = {}
 991         for tx_hash, addresses in self.mempool_addresses.items():
 992             for addr in addresses:
 993                 h = new_mempool_hist.get(addr, [])
 994                 if tx_hash not in h:
 995                     h.append(tx_hash)
 996                 new_mempool_hist[addr] = h
 997
 998         with self.mempool_lock:
 999             self.mempool_hist = new_mempool_hist
1000
1001         # invalidate cache for touched addresses
1002         for addr in touched_addresses:
1003             self.invalidate_cache(addr)
1004
1005
1006     def invalidate_cache(self, address):
1007         with self.cache_lock:
1008             if address in self.history_cache:
1009                 print_log("cache: invalidating", address)
1010                 self.history_cache.pop(address)
1011
1012         with self.watch_lock:
1013             sessions = self.watched_addresses.get(address)
1014
1015         if sessions:
1016             # TODO: update cache here. if new value equals cached value, do not send notification
1017             self.address_queue.put((address,sessions))
1018
1019     def main_iteration(self):
1020         if self.shared.stopped():
1021             print_log("blockchain processor terminating")
1022             if self.use_plyvel:
1023                 self.db.close()
1024             return
1025
1026         with self.dblock:
1027             t1 = time.time()
1028             self.catch_up()
1029             t2 = time.time()
1030
1031         self.memorypool_update()
1032
1033         if self.sent_height != self.height:
1034             self.sent_height = self.height
1035             for session in self.watch_blocks:
1036                 self.push_response(session, {
1037                         'id': None,
1038                         'method': 'blockchain.numblocks.subscribe',
1039                         'params': [self.height],
1040                         })
1041
1042         if self.sent_header != self.header:
1043             print_log("blockchain: %d (%.3fs)" % (self.height, t2 - t1))
1044             self.sent_header = self.header
1045             for session in self.watch_headers:
1046                 self.push_response(session, {
1047                         'id': None,
1048                         'method': 'blockchain.headers.subscribe',
1049                         'params': [self.header],
1050                         })
1051
1052         while True:
1053             try:
1054                 addr, sessions = self.address_queue.get(False)
1055             except:
1056                 break
1057
1058             status = self.get_status(addr)
1059             for session in sessions:
1060                 self.push_response(session, {
1061                         'id': None,
1062                         'method': 'blockchain.address.subscribe',
1063                         'params': [addr, status],
1064                         })
1065
1066         if not self.shared.stopped():
1067             threading.Timer(10, self.main_iteration).start()
1068         else:
1069             print_log("blockchain processor terminating")