backends/libbitcoin/__init__.py

   1 import bitcoin
   2 from bitcoin import bind, _1, _2, _3
   3 from processor import Processor
   4 import threading
   5 import time
   6
   7 import history1 as history
   8 import membuf
   9
  10 class HistoryCache:
  11
  12     def __init__(self):
  13         self.lock = threading.Lock()
  14         self.cache = {}
  15
  16     def store(self, address, result):
  17         with self.lock:
  18             self.cache[address] = result
  19
  20     def fetch(self, address):
  21         try:
  22             with self.lock:
  23                 return self.cache[address]
  24         except KeyError:
  25             return None
  26
  27     def clear(self, addresses):
  28         with self.lock:
  29             for address in addresses:
  30                 if self.cache.has_key(address):
  31                     del self.cache[address]
  32
  33 class MonitorAddress:
  34
  35     def __init__(self, processor, cache, backend):
  36         self.processor = processor
  37         self.cache = cache
  38         self.backend = backend
  39         self.lock = threading.Lock()
  40         # key is hash:index, value is address
  41         self.monitor_output = {}
  42         # key is address
  43         self.monitor_address = set()
  44         # affected
  45         self.affected = {}
  46
  47         backend.memory_buffer.set_handles(self.tx_stored, self.tx_confirmed)
  48
  49     def monitor(self, address, result):
  50         for info in result:
  51             if not info.has_key("raw_output_script"):
  52                 continue
  53             assert info["is_input"] == 0
  54             tx_hash = info["tx_hash"]
  55             output_index = info["index"]
  56             outpoint = "%s:%s" % (tx_hash, output_index)
  57             with self.lock:
  58                 self.monitor_output[outpoint] = address
  59         with self.lock:
  60             self.monitor_address.add(address)
  61
  62     def unpack(self, tx):
  63         tx_hash = bitcoin.hash_transaction(tx)
  64         previous_outputs = []
  65         for input in tx.inputs:
  66             prevout = input.previous_output
  67             prevout = "%s:%s" % (prevout.hash, prevout.index)
  68             previous_outputs.append(prevout)
  69         addrs = []
  70         for output_index, output in enumerate(tx.outputs):
  71             address = bitcoin.payment_address()
  72             if address.extract(output.output_script):
  73                 addrs.append((output_index, str(address)))
  74         return tx_hash, previous_outputs, addrs
  75
  76     def tx_stored(self, tx):
  77         affected_addrs = set()
  78         tx_hash, previous_outputs, addrs = self.unpack(tx)
  79         for prevout in previous_outputs:
  80             with self.lock:
  81                 if self.monitor_output.has_key(prevout):
  82                     affected_addrs.add(self.monitor_output[prevout])
  83         for idx, address in addrs:
  84             with self.lock:
  85                 if address in self.monitor_address:
  86                     affected_addrs.add(address)
  87         with self.lock:
  88             self.affected[tx_hash] = affected_addrs
  89         self.cache.clear(affected_addrs)
  90         self.notify(affected_addrs)
  91
  92     def tx_confirmed(self, tx):
  93         tx_hash, previous_outputs, addrs = self.unpack(tx)
  94         with self.lock:
  95             affected_addrs = self.affected[tx_hash]
  96             del self.affected[tx_hash]
  97         self.cache.clear(affected_addrs)
  98         self.notify(affected_addrs)
  99         # add new outputs to monitor
 100         for idx, address in addrs:
 101             outpoint = "%s:%s" % (tx_hash, idx)
 102             with self.lock:
 103                 if address in affected_addrs:
 104                     self.monitor_output[outpoint] = address
 105         # delete spent outpoints
 106         for prevout in previous_outputs:
 107             with self.lock:
 108                 if self.monitor_output.has_key(prevout):
 109                     del self.monitor_output[prevout]
 110
 111     def notify(self, affected_addrs):
 112         service = self.backend.mempool_service
 113         chain = self.backend.blockchain
 114         txpool = self.backend.transaction_pool
 115         memory_buff = self.backend.memory_buffer
 116         for address in affected_addrs:
 117             response = {"id": None,
 118                         "method": "blockchain.address.subscribe",
 119                         "params": [str(address)]}
 120             history.payment_history(service, chain, txpool, memory_buff,
 121                 address, bind(self.send_notify, _1, _2, response))
 122
 123     def mempool_n(self, result):
 124         assert result is not None
 125         if len(result) == 0:
 126             return None
 127         # mempool:n status
 128         # Order by time, grab last item (latest)
 129         last_info = sorted(result, key=lambda k: k['timestamp'])[-1]
 130         if last_info["block_hash"] == "mempool":
 131             last_id = "mempool:%s" % len(result)
 132         else:
 133             last_id = last_info["block_hash"]
 134         return last_id
 135
 136     def send_notify(self, ec, result, response):
 137         if ec:
 138             print "Error: Monitor.send_notify()", ec
 139             return
 140         assert len(response["params"]) == 1
 141         response["params"].append(self.mempool_n(result))
 142         self.processor.push_response(response)
 143
 144 class Backend:
 145
 146     def __init__(self):
 147         # Create 3 thread-pools each with 1 thread
 148         self.network_service = bitcoin.async_service(1)
 149         self.disk_service = bitcoin.async_service(1)
 150         self.mempool_service = bitcoin.async_service(1)
 151
 152         self.hosts = bitcoin.hosts(self.network_service)
 153         self.handshake = bitcoin.handshake(self.network_service)
 154         self.network = bitcoin.network(self.network_service)
 155         self.protocol = bitcoin.protocol(self.network_service, self.hosts,
 156                                          self.handshake, self.network)
 157
 158         db_prefix = "/home/genjix/libbitcoin/database"
 159         self.blockchain = bitcoin.bdb_blockchain(self.disk_service, db_prefix,
 160                                                  self.blockchain_started)
 161         self.poller = bitcoin.poller(self.mempool_service, self.blockchain)
 162         self.transaction_pool = \
 163             bitcoin.transaction_pool(self.mempool_service, self.blockchain)
 164
 165         self.protocol.subscribe_channel(self.monitor_tx)
 166         self.session = \
 167             bitcoin.session(self.network_service, self.hosts, self.handshake,
 168                             self.network, self.protocol, self.blockchain,
 169                             self.poller, self.transaction_pool)
 170         self.session.start(self.handle_start)
 171
 172         self.memory_buffer = \
 173             membuf.memory_buffer(self.mempool_service.internal_ptr,
 174                                  self.blockchain.internal_ptr,
 175                                  self.transaction_pool.internal_ptr)
 176
 177     def handle_start(self, ec):
 178         if ec:
 179             print "Error starting backend:", ec
 180
 181     def blockchain_started(self, ec, chain):
 182         print "Blockchain initialisation:", ec
 183
 184     def stop(self):
 185         self.session.stop(self.handle_stop)
 186
 187     def handle_stop(self, ec):
 188         if ec:
 189             print "Error stopping backend:", ec
 190         print "Stopped backend"
 191
 192     def monitor_tx(self, node):
 193         # We will be notified here when connected to new bitcoin nodes
 194         # Here we subscribe to new transactions from them which we
 195         # add to the transaction_pool. That way we can track which
 196         # transactions we are interested in.
 197         node.subscribe_transaction(bind(self.recv_tx, _1, _2, node))
 198         # Re-subscribe to next new node
 199         self.protocol.subscribe_channel(self.monitor_tx)
 200
 201     def recv_tx(self, ec, tx, node):
 202         if ec:
 203             print "Error with new transaction:", ec
 204             return
 205         tx_hash = bitcoin.hash_transaction(tx)
 206         self.memory_buffer.receive(tx, bind(self.store_tx, _1, tx_hash))
 207         # Re-subscribe to new transactions from node
 208         node.subscribe_transaction(bind(self.recv_tx, _1, _2, node))
 209
 210     def store_tx(self, ec, tx_hash):
 211         if ec:
 212             print "Error storing memory pool transaction", tx_hash, ec
 213         else:
 214             print "Accepted transaction", tx_hash
 215
 216 class GhostValue:
 217
 218     def __init__(self):
 219         self.event = threading.Event()
 220         self.value = None
 221
 222     def get(self):
 223         self.event.wait()
 224         return self.value
 225
 226     def set(self, value):
 227         self.value = value
 228         self.event.set()
 229
 230 class NumblocksSubscribe:
 231
 232     def __init__(self, backend, processor):
 233         self.backend = backend
 234         self.processor = processor
 235         self.lock = threading.Lock()
 236         self.backend.blockchain.subscribe_reorganize(self.reorganize)
 237         self.backend.blockchain.fetch_last_depth(self.set_last_depth)
 238         self.latest = GhostValue()
 239
 240     def set_last_depth(self, ec, last_depth):
 241         if ec:
 242             print "Error retrieving last depth", ec
 243         else:
 244             self.latest.set(last_depth)
 245
 246     def reorganize(self, ec, fork_point, arrivals, replaced):
 247         latest = fork_point + len(arrivals)
 248         self.latest.set(latest)
 249         response = {"id": None, "method": "blockchain.numblocks.subscribe",
 250                     "result": latest}
 251         self.processor.push_response(response)
 252         self.backend.blockchain.subscribe_reorganize(self.reorganize)
 253
 254     def subscribe(self, request):
 255         latest = self.latest.get()
 256         response = {"id": request["id"],
 257                     "method": "blockchain.numblocks.subscribe",
 258                     "result": latest,
 259                     "error": None}
 260         self.processor.push_response(response)
 261
 262 class AddressGetHistory:
 263
 264     def __init__(self, backend, processor):
 265         self.backend = backend
 266         self.processor = processor
 267
 268     def get(self, request):
 269         address = str(request["params"][0])
 270         service = self.backend.mempool_service
 271         chain = self.backend.blockchain
 272         txpool = self.backend.transaction_pool
 273         memory_buff = self.backend.memory_buffer
 274         history.payment_history(service, chain, txpool, memory_buff,
 275             address, bind(self.respond, _1, _2, request))
 276
 277     def respond(self, ec, result, request):
 278         if ec:
 279             response = {"id": request["id"], "result": None,
 280                         "error": {"message": str(ec), "code": -4}}
 281         else:
 282             response = {"id": request["id"], "result": result, "error": None}
 283         self.processor.push_response(response)
 284
 285 class AddressSubscribe:
 286
 287     def __init__(self, backend, processor, cache, monitor):
 288         self.backend = backend
 289         self.processor = processor
 290         self.cache = cache
 291         self.monitor = monitor
 292
 293     def subscribe(self, request):
 294         address = str(request["params"][0])
 295         service = self.backend.mempool_service
 296         chain = self.backend.blockchain
 297         txpool = self.backend.transaction_pool
 298         memory_buff = self.backend.memory_buffer
 299         history.payment_history(service, chain, txpool, memory_buff,
 300             address, bind(self.construct, _1, _2, request))
 301
 302     def construct(self, ec, result, request):
 303         if ec:
 304             response = {"id": request["id"], "result": None,
 305                         "error": {"message": str(ec), "code": -4}}
 306             self.processor.push_response(response)
 307             return
 308         last_id = self.monitor.mempool_n(result)
 309         response = {"id": request["id"], "result": last_id, "error": None}
 310         self.processor.push_response(response)
 311         address = request["params"][0]
 312         self.monitor.monitor(address, result)
 313         # Cache result for get_history
 314         self.cache.store(address, result)
 315
 316     def fetch_cached(self, request):
 317         address = request["params"][0]
 318         cached = self.cache.fetch(address)
 319         if cached is None:
 320             return False
 321         response = {"id": request["id"], "result": cached, "error": None}
 322         self.processor.push_response(response)
 323         return True
 324
 325 class BlockchainProcessor(Processor):
 326
 327     def __init__(self, config):
 328         Processor.__init__(self)
 329         cache = HistoryCache()
 330         self.backend = Backend()
 331         monitor = MonitorAddress(self, cache, self.backend)
 332         self.numblocks_subscribe = NumblocksSubscribe(self.backend, self)
 333         self.address_get_history = AddressGetHistory(self.backend, self)
 334         self.address_subscribe = \
 335             AddressSubscribe(self.backend, self, cache, monitor)
 336
 337     def stop(self):
 338         self.backend.stop()
 339
 340     def process(self, request):
 341         print "New request (lib)", request
 342         if request["method"] == "blockchain.numblocks.subscribe":
 343             self.numblocks_subscribe.subscribe(request)
 344         elif request["method"] == "blockchain.address.subscribe":
 345             self.address_subscribe.subscribe(request)
 346         elif request["method"] == "blockchain.address.get_history":
 347             if not self.address_subscribe.fetch_cached(request):
 348                 self.address_get_history.get(request)
 349         elif request["method"] == "blockchain.transaction.broadcast":
 350             self.broadcast_transaction(request)
 351
 352     def broadcast_transaction(self, request):
 353         raw_tx = bitcoin.data_chunk(str(request["params"][0]))
 354         exporter = bitcoin.satoshi_exporter()
 355         try:
 356             tx = exporter.load_transaction(raw_tx)
 357         except RuntimeError:
 358             response = {"id": request["id"], "result": None,
 359                         "error": {"message":
 360                             "Exception while parsing the transaction data.",
 361                             "code": -4}}
 362         else:
 363             self.backend.protocol.broadcast_transaction(tx)
 364             tx_hash = str(bitcoin.hash_transaction(tx))
 365             response = {"id": request["id"], "result": tx_hash, "error": None}
 366         self.push_response(response)
 367
 368     def run(self):
 369         print "Warning: pre-alpha prototype. Full of bugs."
 370         while not self.shared.stopped():
 371             time.sleep(1)
 372