Implement shep store lock

chaind/adapters/base.py

@@ -6,29 +6,22 @@ import time
 from chainqueue import Store as QueueStore
 
 # local imports
-from chaind.error import BackendIntegrityError
+from chaind.lock import StoreLock
 
 logg = logging.getLogger(__name__)
 
 
 class ChaindAdapter:
 
-    race_delay = 0.1
-
     def __init__(self, chain_spec, state_store, index_store, counter_store, cache_adapter, dispatcher, cache=None, pending_retry_threshold=0, error_retry_threshold=0):
         self.cache_adapter = cache_adapter
         self.dispatcher = dispatcher
-        err = None
+        store_lock = StoreLock()
-        for i in range(3):
+        while True:
             try:
                 self.store = QueueStore(chain_spec, state_store, index_store, counter_store, cache=cache)
-                err = None
                 break
             except FileNotFoundError as e:
                 logg.debug('queuestore instantiation failed, possible race condition (will try again): {}'.format(e))
-                err = e
+                store_lock.again()
-                time.sleep(self.race_delay)
                 continue
-
-        if err != None:
-            raise BackendIntegrityError(err)

chaind/adapters/fs.py

@@ -12,11 +12,14 @@ from chainqueue.store.fs import (
         CounterStore,
         )
 from shep.store.file import SimpleFileStoreFactory
-from shep.error import StateInvalid
+from shep.error import (
+        StateInvalid,
+        StateLockedKey,
+        )
 
 # local imports
 from .base import ChaindAdapter
-from chaind.error import BackendIntegrityError
+from chaind.lock import StoreLock
 
 logg = logging.getLogger(__name__)
 
@@ -24,7 +27,7 @@ logg = logging.getLogger(__name__)
 class ChaindFsAdapter(ChaindAdapter):
 
     def __init__(self, chain_spec, path, cache_adapter, dispatcher, cache=None, pending_retry_threshold=0, error_retry_threshold=0, digest_bytes=32, event_callback=None):
-        factory = SimpleFileStoreFactory(path).add
+        factory = SimpleFileStoreFactory(path, use_lock=True).add
         state_store = Status(factory, allow_invalid=True, event_callback=event_callback)
         index_path = os.path.join(path, 'tx')
         index_store = IndexStore(index_path, digest_bytes=digest_bytes)
@@ -39,22 +42,23 @@ class ChaindFsAdapter(ChaindAdapter):
 
     def get(self, tx_hash):
         v = None
-        err = None
+        store_lock = StoreLock()
-        for i in range(3):
+        while True:
             try:
                 v = self.store.get(tx_hash)
-                err = None
                 break
             except StateInvalid as e:
                 logg.error('I am just a simple syncer and do not know how to handle the state which the tx {} is in: {}'.format(tx_hash, e))
                 return None
             except FileNotFoundError as e:
-                err = e
-                time.sleep(self.race_delay)
                 logg.debug('queuestore get {} failed, possible race condition (will try again): {}'.format(tx_hash, e))
+                store_lock.again()
                 continue
-        if err != None:
+            except StateLockedKey as e:
-            raise BackendIntegrityError(tx_hash)
+                logg.debug('queuestore get {} failed, possible race condition (will try again): {}'.format(tx_hash, e))
+                store_lock.again()
+                continue
+
         return v[1]
 
 
@@ -104,20 +108,20 @@ class ChaindFsAdapter(ChaindAdapter):
 
     def dispatch(self, tx_hash):
         entry = None
-        err = None
+
-        for i in range(3):
+        store_lock = StoreLock()
+        while True:
             try:
                 entry = self.store.send_start(tx_hash)
-                err = None
                 break
             except FileNotFoundError as e:
-                logg.debug('dispatch failed to find {} in backend, will try again: {}'.format(tx_hash, err))
+                logg.debug('dispatch failed to find {} in backend, will try again: {}'.format(tx_hash, e))
-                err = e
+                store_lock.again()
-                time.sleep(self.race_delay)
+                continue
+            except StateLockedKey as e:
+                logg.debug('dispatch failed to find {} in backend, will try again: {}'.format(tx_hash, e))
+                store_lock.again()
                 continue
-
-        if err != None:
-            raise BackendIntegrityError('dispatch failed to find {} in backend: {}'.format(tx_hash, err))
 
         tx_wire = entry.serialize()
 
@@ -128,5 +132,18 @@ class ChaindFsAdapter(ChaindAdapter):
             self.store.fail(tx_hash)
             return False
 
-        self.store.send_end(tx_hash)
+        store_lock = StoreLock()
+        while True:
+            try:
+                self.store.send_end(tx_hash)
+                break
+            except FileNotFoundError as e:
+                logg.debug('dispatch failed to find {} in backend, will try again: {}'.format(tx_hash, e))
+                store_lock.again(e)
+                continue
+            except StateLockedKey as e:
+                logg.debug('dispatch failed to find {} in backend, will try again: {}'.format(tx_hash, e))
+                store_lock.again(e)
+                continue
+
         return True

chaind/error.py

@@ -22,5 +22,5 @@ class QueueLockError(Exception):
     pass
 
 
-class BackendIntegrityError(Exception):
+class BackendError(Exception):
     pass

chaind/filter.py

@@ -7,21 +7,20 @@ from chainlib.status import Status as TxStatus
 from chainsyncer.filter import SyncFilter
 from chainqueue.error import NotLocalTxError
 from chaind.adapters.fs import ChaindFsAdapter
+from shep.error import StateLockedKey
 
 # local imports
 from .error import (
         QueueLockError,
-        BackendIntegrityError,
+        BackendError,
         )
+from chaind.lock import StoreLock
 
 logg = logging.getLogger(__name__)
 
 
 class StateFilter(SyncFilter):
 
-    delay_limit = 3.0
-    race_delay = 0.1
-
     def __init__(self, chain_spec, adapter_path, tx_adapter, throttler=None):
         self.chain_spec = chain_spec
         self.adapter_path = adapter_path
@@ -31,8 +30,9 @@ class StateFilter(SyncFilter):
 
     def filter(self, conn, block, tx, session=None):
         cache_tx = None
-        for i in range(3):
+        store_lock = StoreLock()
-            queue_adapter = None
+        queue_adapter = None
+        while True:
             try:
                 queue_adapter = ChaindFsAdapter(
                     self.chain_spec,
@@ -40,62 +40,52 @@ class StateFilter(SyncFilter):
                     self.tx_adapter,
                     None,
                     )
-            except BackendIntegrityError as e:
+            except BackendError as e:
                 logg.error('adapter instantiation failed: {}, one more try'.format(e))
-                time.sleep(self.race_delay)
+                store_lock.again()
                 continue
 
+            store_lock.reset()
+
             try:
                 cache_tx = queue_adapter.get(tx.hash)
+                break
             except NotLocalTxError:
                 logg.debug('skipping not local transaction {}'.format(tx.hash))
                 return False
-            except BackendIntegrityError as e:
+            except BackendError as e:
                 logg.error('adapter instantiation failed: {}, one more try'.format(e))
-                time.sleep(self.race_delay)
+                queue_adapter = None
+                store_lock.again()
                 continue
 
-            break
-
         if cache_tx == None:
             raise NotLocalTxError(tx.hash)
 
-        delay = 0.01
+        store_lock = StoreLock()
-        race_attempts = 0
+        queue_lock = StoreLock(error=QueueLockError)
-        err = None
         while True:
-            if delay > self.delay_limit:
-                raise QueueLockError('The queue lock for tx {} seems to be stuck. Human meddling needed.'.format(tx.hash))
-            elif race_attempts >= 3:
-                break
             try:
                 if tx.status == TxStatus.SUCCESS:
                     queue_adapter.succeed(block, tx)
                 else:
                     queue_adapter.fail(block, tx)
-                err = None
                 break
             except QueueLockError as e:
                 logg.debug('queue item {} is blocked, will retry: {}'.format(tx.hash, e))
-                time.sleep(delay)
+                queue_lock.again()
-                delay *= 2
-                race_attempts = 0
-                err = None
             except FileNotFoundError as e:
-                err = e
                 logg.debug('queue item {} not found, possible race condition, will retry: {}'.format(tx.hash, e))
-                race_attempts += 1
+                store_lock.again()
-                time.sleep(self.race_delay)
                 continue
             except NotLocalTxError as e:
-                err = e
                 logg.debug('queue item {} not found, possible race condition, will retry: {}'.format(tx.hash, e))
-                race_attempts += 1
+                store_lock.again()
-                time.sleep(self.race_delay)
+                continue
+            except StateLockedKey as e:
+                logg.debug('queue item {} not found, possible race condition, will retry: {}'.format(tx.hash, e))
+                store_lock.again()
                 continue
-
-        if err != None:
-            raise BackendIntegrityError('cannot find queue item {} in backend: {}'.format(tx.hash, err))
 
         logg.info('filter registered {} for {} in {}'.format(tx.status.name, tx.hash, block))
 

chaind/session.py

@@ -8,12 +8,14 @@ import stat
 from hexathon import strip_0x
 
 # local imports
-from chaind.error import (
+from .error import (
         NothingToDoError,
         ClientGoneError,
         ClientBlockError,
         ClientInputError,
         )
+from .lock import StoreLock
+from .error import BackendError
 
 logg = logging.getLogger(__name__)
 
@@ -59,7 +61,16 @@ class SessionController:
 
 
     def process(self, conn):
-        r = self.processor(self.chain_spec, self.adapter, conn)
+        state_lock = StoreLock()
+        r = None
+        while True:
+            try:
+                r = self.processor(self.chain_spec, self.adapter, conn)
+                break
+            except BackendError as e:
+                state_lock.again(e)
+                continue
+                
         if r > 0:
             self.srv.settimeout(0.1)
         else: