X-Git-Url: https://git.novaco.in/?a=blobdiff_plain;f=backends%2Flibbitcoin%2F__init__.py;h=e6092f779b9b85ef548daade899a995011ce526d;hb=2e04e2c499984d5a8f510b33ca040058bec59c8c;hp=cf2a4629515f8105236d8f7445256e6cdb3be85b;hpb=2618bc9616b6eb39c2213af601a4eee0ebc9f04c;p=electrum-server.git diff --git a/backends/libbitcoin/__init__.py b/backends/libbitcoin/__init__.py index cf2a462..e6092f7 100644 --- a/backends/libbitcoin/__init__.py +++ b/backends/libbitcoin/__init__.py @@ -4,7 +4,8 @@ from processor import Processor import threading import time -import history +import history1 as history +import membuf class HistoryCache: @@ -31,16 +32,17 @@ class HistoryCache: 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: @@ -55,55 +57,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 +128,17 @@ 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) @@ -150,9 +164,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 +198,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)) @@ -227,14 +242,13 @@ 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) @@ -247,16 +261,17 @@ 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) @@ -269,20 +284,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} @@ -306,9 +321,8 @@ 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,7 +344,7 @@ 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) @@ -345,8 +359,3 @@ class BlockchainProcessor(Processor): 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) -