Merge branch 'kyuupichan-offline-2'

lib/verifier.py

@@ -25,48 +25,25 @@ import util
 from bitcoin import *
 
 
-
-
 class SPV(util.DaemonThread):
     """ Simple Payment Verification """
 
-    def __init__(self, network, storage):
+    def __init__(self, network, wallet):
         util.DaemonThread.__init__(self)
-        self.storage = storage
+        self.wallet = wallet
         self.network = network
-        self.transactions    = {}                                 # requested verifications (with height sent by the requestor)
-        self.verified_tx     = storage.get('verified_tx3',{})      # height, timestamp of verified transactions
         self.merkle_roots    = {}                                  # hashed by me
-        self.lock = threading.Lock()
         self.queue = Queue.Queue()
 
-    def get_height(self, tx_hash):
-        with self.lock:
-            v = self.verified_tx.get(tx_hash)
-        height = v[0] if v else None
-        return height
-
-
-    def add(self, tx_hash, tx_height):
-        """ add a transaction to the list of monitored transactions. """
-        assert tx_height > 0
-        with self.lock:
-            if tx_hash not in self.transactions.keys():
-                self.transactions[tx_hash] = tx_height
-
     def run(self):
-        requested_merkle = []
+        requested_merkle = set()
         while self.is_running():
-            # request missing tx
-            for tx_hash, tx_height in self.transactions.items():
-                if tx_hash not in self.verified_tx:
-                    # do not request merkle branch before headers are available
-                    if tx_height > self.network.get_local_height():
-                        continue
-                    if self.merkle_roots.get(tx_hash) is None and tx_hash not in requested_merkle:
-                        if self.network.send([ ('blockchain.transaction.get_merkle',[tx_hash, tx_height]) ], self.queue.put):
-                            self.print_error('requesting merkle', tx_hash)
-                            requested_merkle.append(tx_hash)
+            unverified = self.wallet.get_unverified_txs()
+            for (tx_hash, tx_height) in unverified:
+                if self.merkle_roots.get(tx_hash) is None and tx_hash not in requested_merkle:
+                    if self.network.send([ ('blockchain.transaction.get_merkle',[tx_hash, tx_height]) ], self.queue.put):
+                        self.print_error('requesting merkle', tx_hash)
+                        requested_merkle.add(tx_hash)
             try:
                 r = self.queue.get(timeout=0.1)
             except Queue.Empty:
@@ -102,12 +79,8 @@ class SPV(util.DaemonThread):
 
         # we passed all the tests
         self.merkle_roots[tx_hash] = merkle_root
-        timestamp = header.get('timestamp')
-        with self.lock:
-            self.verified_tx[tx_hash] = (tx_height, timestamp, pos)
-        self.print_error("verified %s"%tx_hash)
-        self.storage.put('verified_tx3', self.verified_tx, True)
-        self.network.trigger_callback('updated')
+        self.print_error("verified %s" % tx_hash)
+        self.wallet.add_verified_tx(tx_hash, (tx_height, header.get('timestamp'), pos))
 
 
     def hash_merkle_root(self, merkle_s, target_hash, pos):
@@ -118,15 +91,8 @@ class SPV(util.DaemonThread):
         return hash_encode(h)
 
 
-
     def undo_verifications(self, height):
-        with self.lock:
-            items = self.verified_tx.items()[:]
-        for tx_hash, item in items:
-            tx_height, timestamp, pos = item
-            if tx_height >= height:
-                self.print_error("redoing", tx_hash)
-                with self.lock:
-                    self.verified_tx.pop(tx_hash)
-                    if tx_hash in self.merkle_roots:
-                        self.merkle_roots.pop(tx_hash)
+        tx_hashes = selt.wallet.undo_verifications(height)
+        for tx_hash in tx_hashes:
+            self.print_error("redoing", tx_hash)
+            self.merkle_roots.pop(tx_hash, None)

lib/wallet.py

@@ -182,6 +182,11 @@ class Abstract_Wallet(object):
 
         # spv
         self.verifier = None
+        # Transactions pending verification.  Each value is the transaction height.  Access with self.lock.
+        self.unverified_tx = {}
+        # Verified transactions.  Each value is a (height, timestamp, block_pos) tuple.  Access with self.lock.
+        self.verified_tx   = storage.get('verified_tx3',{})
+        
         # there is a difference between wallet.up_to_date and interface.is_up_to_date()
         # interface.is_up_to_date() returns true when all requests have been answered and processed
         # wallet.up_to_date is true when the wallet is synchronized (stronger requirement)
@@ -388,19 +393,45 @@ class Abstract_Wallet(object):
         return decrypted
 
     def add_unverified_tx(self, tx_hash, tx_height):
-        if self.verifier and tx_height > 0:
-            self.verifier.add(tx_hash, tx_height)
+        if tx_height > 0:
+            with self.lock:
+                self.unverified_tx[tx_hash] = tx_height
+
+    def add_verified_tx(self, tx_hash, info):
+        with self.lock:
+            self.verified_tx[tx_hash] = info  # (tx_height, timestamp, pos)
+        self.storage.put('verified_tx3', self.verified_tx, True)
+        self.network.trigger_callback('updated')
+
+    def get_unverified_txs(self):
+        '''Returns a list of tuples (tx_hash, height) that are unverified and not beyond local height'''
+        txs = []
+        with self.lock:
+            for tx_hash, tx_height in self.unverified_tx.items():
+                # do not request merkle branch before headers are available
+                if tx_hash not in self.verified_tx and tx_height <= self.network.get_local_height():
+                    txs.append((tx_hash, tx_height))
+        return txs
+    
+    def undo_verifications(self, height):
+        '''Used by the verifier when a reorg has happened'''
+        txs = []
+        with self.lock:
+            for tx_hash, item in self.verified_tx:
+                tx_height, timestamp, pos = item
+                if tx_height >= height:
+                    self.verified_tx.pop(tx_hash, None)
+                    txs.append(tx_hash)
+        return txs
 
     def get_confirmations(self, tx):
         """ return the number of confirmations of a monitored transaction. """
-        if not self.verifier:
-            return (None, None)
-        with self.verifier.lock:
-            if tx in self.verifier.verified_tx:
-                height, timestamp, pos = self.verifier.verified_tx[tx]
+        with self.lock:
+            if tx in self.verified_tx:
+                height, timestamp, pos = self.verified_tx[tx]
                 conf = (self.network.get_local_height() - height + 1)
                 if conf <= 0: timestamp = None
-            elif tx in self.verifier.transactions:
+            elif tx in self.unverified_tx:
                 conf = -1
                 timestamp = None
             else:
@@ -411,9 +442,9 @@ class Abstract_Wallet(object):
 
     def get_txpos(self, tx_hash):
         "return position, even if the tx is unverified"
-        with self.verifier.lock:
-            x = self.verifier.verified_tx.get(tx_hash)
-            y = self.verifier.transactions.get(tx_hash)
+        with self.lock:
+            x = self.verified_tx.get(tx_hash)
+            y = self.unverified_tx.get(tx_hash)
         if x:
             height, timestamp, pos = x
             return height, pos
@@ -1000,7 +1031,8 @@ class Abstract_Wallet(object):
                 self.add_unverified_tx (tx_hash, tx_height)
 
         # if we are on a pruning server, remove unverified transactions
-        vr = self.verifier.transactions.keys() + self.verifier.verified_tx.keys()
+        with self.lock:
+            vr = self.verified_tx.keys() + self.unverified_tx.keys()
         for tx_hash in self.transactions.keys():
             if tx_hash not in vr:
                 print_error("removing transaction", tx_hash)
@@ -1035,8 +1067,9 @@ class Abstract_Wallet(object):
                 if not tx: continue
 
                 # already verified?
-                if self.verifier.get_height(tx_hash):
-                    continue
+                with self.lock:
+                    if tx_hash in self.verified_tx:
+                        continue
                 # unconfirmed tx
                 print_error("new history is orphaning transaction:", tx_hash)
                 # check that all outputs are not mine, request histories
@@ -1065,7 +1098,7 @@ class Abstract_Wallet(object):
         from verifier import SPV
         self.network = network
         if self.network is not None:
-            self.verifier = SPV(self.network, self.storage)
+            self.verifier = SPV(self.network, self)
             self.verifier.start()
             self.set_verifier(self.verifier)
             self.synchronizer = WalletSynchronizer(self, network)