X-Git-Url: https://git.novaco.in/?p=electrum-server.git;a=blobdiff_plain;f=backends%2Flibbitcoin%2F__init__.py;h=ef2eb1d81cde7cbc78f89f7db140d8ee4b5fdf9f;hp=247eae7b5fdf7f3cb5af034d3eadf3cd0c03e405;hb=240db7c28d2e6a0d0078c42f81b2b110615e88c6;hpb=27bd2d1754b5949678d1ee7ff5efad4b3c8bef6f diff --git a/backends/libbitcoin/__init__.py b/backends/libbitcoin/__init__.py index 247eae7..ef2eb1d 100644 --- a/backends/libbitcoin/__init__.py +++ b/backends/libbitcoin/__init__.py @@ -1,10 +1,13 @@ +import threading +import time + import bitcoin from bitcoin import bind, _1, _2, _3 + from processor import Processor -import threading -import time +import history1 as history +import membuf -import history class HistoryCache: @@ -26,25 +29,27 @@ class HistoryCache: def clear(self, addresses): with self.lock: for address in addresses: - if self.cache.has_key(address): + if address in self.cache: del self.cache[address] + class MonitorAddress: - def __init__(self, processor, cache): + def __init__(self, processor, cache, backend): self.processor = processor self.cache = cache + self.backend = backend self.lock = threading.Lock() # key is hash:index, value is address self.monitor_output = {} # key is address self.monitor_address = set() - # affected - self.affected = {} + + backend.memory_buffer.set_handles(self.tx_stored, self.tx_confirmed) def monitor(self, address, result): for info in result: - if not info.has_key("raw_output_script"): + if "raw_output_script" not in info: continue assert info["is_input"] == 0 tx_hash = info["tx_hash"] @@ -55,55 +60,63 @@ class MonitorAddress: with self.lock: self.monitor_address.add(address) - def tx_stored(self, tx_desc): - tx_hash, prevouts, addrs = tx_desc + def unpack(self, tx): + tx_hash = bitcoin.hash_transaction(tx) + previous_outputs = [] + for input in tx.inputs: + prevout = input.previous_output + prevout = "%s:%s" % (prevout.hash, prevout.index) + previous_outputs.append(prevout) + addrs = [] + for output_index, output in enumerate(tx.outputs): + address = bitcoin.payment_address() + if address.extract(output.output_script): + addrs.append((output_index, str(address))) + return tx_hash, previous_outputs, addrs + + def effect_notify(self, tx, delete_outs): affected_addrs = set() - for prevout_hash, prevout_index in prevouts: - prevout = "%s:%s" % (prevout_hash, prevout_index) - with self.lock: - if self.monitor_output.has_key(prevout): + tx_hash, previous_outputs, addrs = self.unpack(tx) + for prevout in previous_outputs: + try: + with self.lock: affected_addrs.add(self.monitor_output[prevout]) + if delete_outs: + del self.monitor_output[prevout] + except KeyError: + pass for idx, address in addrs: with self.lock: if address in self.monitor_address: affected_addrs.add(address) - with self.lock: - self.affected[tx_hash] = affected_addrs self.cache.clear(affected_addrs) self.notify(affected_addrs) + # Used in confirmed txs + return tx_hash, addrs, affected_addrs - def tx_confirmed(self, tx_desc): - tx_hash, prevouts, addrs = tx_desc - with self.lock: - affected_addrs = self.affected[tx_hash] - del self.affected[tx_hash] - self.cache.clear(affected_addrs) - self.notify(affected_addrs) + def tx_stored(self, tx): + self.effect_notify(tx, False) + + def tx_confirmed(self, tx): + tx_hash, addrs, affected_addrs = self.effect_notify(tx, True) # add new outputs to monitor for idx, address in addrs: outpoint = "%s:%s" % (tx_hash, idx) - with self.lock: - if address in affected_addrs: + if address in affected_addrs: + with self.lock: self.monitor_output[outpoint] = address - # delete spent outpoints - for prevout_hash, prevout_index in prevouts: - prevout = "%s:%s" % (prevout_hash, prevout_index) - with self.lock: - if self.monitor_output.has_key(prevout): - del self.monitor_output[prevout] def notify(self, affected_addrs): - templ_response = {"id": None, - "method": "blockchain.address.subscribe", - "params": []} + service = self.backend.mempool_service chain = self.backend.blockchain txpool = self.backend.transaction_pool - membuf = self.backend.pool_buffer + memory_buff = self.backend.memory_buffer for address in affected_addrs: - response = templ_response.copy() - response["params"].append(address) - history.payment_history(chain, txpool, membuf, address, - bind(self.send_notify, _1, response)) + response = {"id": None, + "method": "blockchain.address.subscribe", + "params": [str(address)]} + history.payment_history(service, chain, txpool, memory_buff, address, + bind(self.send_notify, _1, _2, response)) def mempool_n(self, result): assert result is not None @@ -118,13 +131,18 @@ class MonitorAddress: last_id = last_info["block_hash"] return last_id - def send_notify(self, result, response): + def send_notify(self, ec, result, response): + if ec: + print "Error: Monitor.send_notify()", ec + return + assert len(response["params"]) == 1 response["params"].append(self.mempool_n(result)) self.processor.push_response(response) + class Backend: - def __init__(self, monitor): + def __init__(self): # Create 3 thread-pools each with 1 thread self.network_service = bitcoin.async_service(1) self.disk_service = bitcoin.async_service(1) @@ -139,7 +157,7 @@ class Backend: db_prefix = "/home/genjix/libbitcoin/database" self.blockchain = bitcoin.bdb_blockchain(self.disk_service, db_prefix, self.blockchain_started) - self.poller = bitcoin.poller(self.blockchain) + self.poller = bitcoin.poller(self.mempool_service, self.blockchain) self.transaction_pool = \ bitcoin.transaction_pool(self.mempool_service, self.blockchain) @@ -150,9 +168,10 @@ class Backend: self.poller, self.transaction_pool) self.session.start(self.handle_start) - self.pool_buffer = \ - history.MemoryPoolBuffer(self.transaction_pool, - self.blockchain, monitor) + self.memory_buffer = \ + membuf.memory_buffer(self.mempool_service.internal_ptr, + self.blockchain.internal_ptr, + self.transaction_pool.internal_ptr) def handle_start(self, ec): if ec: @@ -183,7 +202,7 @@ class Backend: print "Error with new transaction:", ec return tx_hash = bitcoin.hash_transaction(tx) - self.pool_buffer.recv_tx(tx, bind(self.store_tx, _1, tx_hash)) + self.memory_buffer.receive(tx, bind(self.store_tx, _1, tx_hash)) # Re-subscribe to new transactions from node node.subscribe_transaction(bind(self.recv_tx, _1, _2, node)) @@ -193,6 +212,7 @@ class Backend: else: print "Accepted transaction", tx_hash + class GhostValue: def __init__(self): @@ -207,6 +227,7 @@ class GhostValue: self.value = value self.event.set() + class NumblocksSubscribe: def __init__(self, backend, processor): @@ -227,18 +248,18 @@ class NumblocksSubscribe: latest = fork_point + len(arrivals) self.latest.set(latest) response = {"id": None, "method": "blockchain.numblocks.subscribe", - "result": latest} + "params": [latest]} self.processor.push_response(response) self.backend.blockchain.subscribe_reorganize(self.reorganize) def subscribe(self, request): latest = self.latest.get() response = {"id": request["id"], - "method": "blockchain.numblocks.subscribe", "result": latest, "error": None} self.processor.push_response(response) + class AddressGetHistory: def __init__(self, backend, processor): @@ -247,20 +268,22 @@ class AddressGetHistory: def get(self, request): address = str(request["params"][0]) + service = self.backend.mempool_service chain = self.backend.blockchain txpool = self.backend.transaction_pool - membuf = self.backend.pool_buffer - history.payment_history(chain, txpool, membuf, address, - bind(self.respond, _1, request)) + memory_buff = self.backend.memory_buffer + history.payment_history(service, chain, txpool, memory_buff, address, + bind(self.respond, _1, _2, request)) - def respond(self, result, request): - if result is None: + def respond(self, ec, result, request): + if ec: response = {"id": request["id"], "result": None, - "error": {"message": "Error", "code": -4}} + "error": {"message": str(ec), "code": -4}} else: response = {"id": request["id"], "result": result, "error": None} self.processor.push_response(response) + class AddressSubscribe: def __init__(self, backend, processor, cache, monitor): @@ -269,20 +292,20 @@ class AddressSubscribe: self.cache = cache self.monitor = monitor - self.backend.pool_buffer.cheat = self - def subscribe(self, request): address = str(request["params"][0]) + service = self.backend.mempool_service chain = self.backend.blockchain txpool = self.backend.transaction_pool - membuf = self.backend.pool_buffer - history.payment_history(chain, txpool, membuf, address, - bind(self.construct, _1, request)) + memory_buff = self.backend.memory_buffer + history.payment_history(service, chain, txpool, memory_buff, address, + bind(self.construct, _1, _2, request)) - def construct(self, result, request): - if result is None: + def construct(self, ec, result, request): + if ec: response = {"id": request["id"], "result": None, - "error": {"message": "Error", "code": -4}} + "error": {"message": str(ec), "code": -4}} + self.processor.push_response(response) return last_id = self.monitor.mempool_n(result) response = {"id": request["id"], "result": last_id, "error": None} @@ -301,14 +324,14 @@ class AddressSubscribe: self.processor.push_response(response) return True + class BlockchainProcessor(Processor): def __init__(self, config): Processor.__init__(self) cache = HistoryCache() - monitor = MonitorAddress(self, cache) - self.backend = Backend(monitor) - monitor.backend = self.backend + self.backend = Backend() + monitor = MonitorAddress(self, cache, self.backend) self.numblocks_subscribe = NumblocksSubscribe(self.backend, self) self.address_get_history = AddressGetHistory(self.backend, self) self.address_subscribe = \ @@ -330,23 +353,21 @@ class BlockchainProcessor(Processor): self.broadcast_transaction(request) def broadcast_transaction(self, request): - raw_tx = bitcoin.data_chunk(str(request["params"])) + raw_tx = bitcoin.data_chunk(str(request["params"][0])) exporter = bitcoin.satoshi_exporter() try: tx = exporter.load_transaction(raw_tx) except RuntimeError: - response = {"id": request["id"], "result": None, - "error": {"message": - "Exception while parsing the transaction data.", - "code": -4}} + response = { + "id": request["id"], + "result": None, + "error": { + "message": "Exception while parsing the transaction data.", + "code": -4, + } + } else: self.backend.protocol.broadcast_transaction(tx) tx_hash = str(bitcoin.hash_transaction(tx)) response = {"id": request["id"], "result": tx_hash, "error": None} self.push_response(response) - - def run(self): - print "Warning: pre-alpha prototype. Full of bugs." - while not self.shared.stopped(): - time.sleep(1) -