non blocking claim producer

lbry/wallet/server/env.py

@@ -46,7 +46,7 @@ class Env:
         self.trending_whale_half_life = math.log2(0.1 ** (1 / (3 + self.integer('TRENDING_WHALE_DECAY_RATE', 24)))) + 1
         self.trending_whale_threshold = float(self.integer('TRENDING_WHALE_THRESHOLD', 10000)) * 1E8
 
-        self.max_query_workers = self.integer('MAX_QUERY_WORKERS', None)
+        self.max_query_workers = self.integer('MAX_QUERY_WORKERS', 4)
         self.individual_tag_indexes = self.boolean('INDIVIDUAL_TAG_INDEXES', True)
         self.track_metrics = self.boolean('TRACK_METRICS', False)
         self.websocket_host = self.default('WEBSOCKET_HOST', self.host)

lbry/wallet/server/leveldb.py

@@ -742,22 +742,47 @@ class LevelDB:
 
     async def claims_producer(self, claim_hashes: Set[bytes]):
         batch = []
-        for claim_hash in claim_hashes:
+        results = []
+
+        loop = asyncio.get_event_loop()
+
+        def produce_claim(claim_hash):
             if claim_hash not in self.claim_to_txo:
                 self.logger.warning("can't sync non existent claim to ES: %s", claim_hash.hex())
-                continue
+                return
             name = self.claim_to_txo[claim_hash].normalized_name
             if not self.prefix_db.claim_takeover.get(name):
                 self.logger.warning("can't sync non existent claim to ES: %s", claim_hash.hex())
-                continue
-            claim = self._fs_get_claim_by_hash(claim_hash)
+                return
+            claim_txo = self.claim_to_txo.get(claim_hash)
+            if not claim_txo:
+                return
+            activation = self.get_activation(claim_txo.tx_num, claim_txo.position)
+            claim = self._prepare_resolve_result(
+                claim_txo.tx_num, claim_txo.position, claim_hash, claim_txo.name, claim_txo.root_tx_num,
+                claim_txo.root_position, activation, claim_txo.channel_signature_is_valid
+            )
             if claim:
                 batch.append(claim)
-        batch.sort(key=lambda x: x.tx_hash)
-        for claim in batch:
+
+        def get_metadata(claim):
             meta = self._prepare_claim_metadata(claim.claim_hash, claim)
             if meta:
-                yield meta
+                results.append(meta)
+
+        if claim_hashes:
+            await asyncio.wait(
+                [loop.run_in_executor(None, produce_claim, claim_hash) for claim_hash in claim_hashes]
+            )
+        batch.sort(key=lambda x: x.tx_hash)
+
+        if batch:
+            await asyncio.wait(
+                [loop.run_in_executor(None, get_metadata, claim) for claim in batch]
+            )
+        for meta in results:
+            yield meta
+
         batch.clear()
 
     def get_activated_at_height(self, height: int) -> DefaultDict[PendingActivationValue, List[PendingActivationKey]]:

lbry/wallet/server/server.py

@@ -69,7 +69,7 @@ class Server:
 
     def run(self):
         loop = asyncio.get_event_loop()
-        executor = ThreadPoolExecutor(4)
+        executor = ThreadPoolExecutor(self.env.max_query_workers)
         loop.set_default_executor(executor)
 
         def __exit():