tmp: split advance_txs 9

Author: SomberNight

src/electrumx/server/block_processor.py

@@ -10,13 +10,10 @@
 
 
 import asyncio
+import concurrent.futures
 from concurrent.futures import ThreadPoolExecutor
 import time
 from typing import Sequence, Tuple, List, Callable, Optional, TYPE_CHECKING, Type, Set
-try:
-    from concurrent.futures import InterpreterPoolExecutor
-except ImportError:
-    InterpreterPoolExecutor = None
 
 from aiorpcx import run_in_thread, CancelledError, ignore_after
 
@@ -41,19 +38,6 @@
     from electrumx.server.controller import Notifications
 
 
-class _myglobals:
-    coin = None  # type: type[Coin]
-
-
-def _init_pool(coin2):
-    _myglobals.coin = coin2
-
-
-def eval_g_coin_block(raw_block: bytes, height: int) -> 'Block':
-    return _myglobals.coin.block(raw_block, height)
-
-
-
 class Prefetcher:
     '''Prefetches blocks (in the forward direction only).'''
 
@@ -80,7 +64,7 @@ def __init__(
         self.cache_size = 0
         self.min_cache_size = 10 * 1024 * 1024
         # This makes the first fetch be 10 blocks
-        self.ave_size = self.min_cache_size // 10
+        self.ave_size = max(1, self.min_cache_size // 10)
         self.polling_delay = polling_delay_secs
 
     async def main_loop(self, bp_height: int) -> None:
@@ -258,7 +242,7 @@ async def check_and_advance_blocks(self, raw_blocks: Sequence[bytes]) -> None:
         if not raw_blocks:
             return
         first = self.height + 1
-        blocks = self.pool_executor.map(self.coin.block, raw_blocks, range(first, first + (len(raw_blocks))))
+        blocks = self.pool_executor1.map(self.coin.block, raw_blocks, range(first, first + (len(raw_blocks))))
         blocks = list(blocks)  # join threads
         headers = [block.header for block in blocks]
         hprevs = [self.coin.header_prevhash_rev(h) for h in headers]
@@ -461,7 +445,6 @@ def advance_blocks(self, blocks: Sequence['Block']) -> None:
         assert self.state_lock.locked()
         assert blocks
         min_height = self.db.min_undo_height(self.daemon.cached_height())
-        genesis_activation = self.coin.GENESIS_ACTIVATION
         coin = self.coin
 
         tx_num = self.tx_count
@@ -475,30 +458,36 @@ def advance_blocks(self, blocks: Sequence['Block']) -> None:
             self.db.tx_counts.append(tx_num)
         self.tx_count = tx_num
 
+        # process tx outputs
+        blk_process_outputs = []  # type: list[concurrent.futures.Future]
         height = self.height
         for block in blocks:
             height += 1
-            tx_num_start = self.db.tx_counts[height-1] if height > 0 else 0
-            is_unspendable = (is_unspendable_genesis if height >= genesis_activation
-                              else is_unspendable_legacy)
-            add_unflushed_hist1 = self.advance_txs_process_outputs(
+            fut = self.pool_executor1.submit(
+                self.advance_txs_process_outputs,
                 block.transactions,
-                tx_num_start=tx_num_start,
-                is_unspendable=is_unspendable,
+                height=height,
             )
-            add_unflushed_hist1()
+            blk_process_outputs.append(fut)
+        for fut in blk_process_outputs:
+            add_unflushed_hist = fut.result()
+            add_unflushed_hist()
 
+        # process tx inputs
+        blk_process_inputs = []  # type: list[concurrent.futures.Future]
         height = self.height
         for block in blocks:
             height += 1
-            tx_num_start = self.db.tx_counts[height - 1] if height > 0 else 0
-            add_unflushed_hist2 = self.advance_txs_process_inputs(
+            fut = self.pool_executor1.submit(
+                self.advance_txs_process_inputs,
                 block.transactions,
-                tx_num_start=tx_num_start,
                 height=height,
                 add_undo_info=height >= min_height,
             )
-            add_unflushed_hist2()
+            blk_process_inputs.append(fut)
+        for fut in blk_process_inputs:
+            add_unflushed_hist = fut.result()
+            add_unflushed_hist()
 
         height = self.height
         for block in blocks:
@@ -519,10 +508,14 @@ def advance_txs_process_outputs(
             self,
             txs: Sequence[Tx],
             *,
-            is_unspendable: Callable[[bytes], bool],
-            tx_num_start: int,
+            height: int,
     ) -> Callable[[], None]:
 
+        tx_num_start = self.db.tx_counts[height - 1] if height > 0 else 0
+        is_unspendable = (
+            is_unspendable_genesis if height >= self.coin.GENESIS_ACTIVATION
+            else is_unspendable_legacy)
+
         # Use local vars for speed in the loops
         script_hashX = self.coin.hashX_from_script
         put_utxo = self.utxo_cache.__setitem__
@@ -556,7 +549,7 @@ def process_txouts_for_chunk(txs_chunk):
             for tx_pos, tx in txs_chunk:
                 process_txouts_for_single_tx(tx_pos=tx_pos, tx=tx)
 
-        list(self.pool_executor.map(
+        list(self.pool_executor2.map(
             process_txouts_for_chunk,
             chunks(list(enumerate(txs)), 200),
         ))
@@ -571,11 +564,12 @@ def advance_txs_process_inputs(
             self,
             txs: Sequence[Tx],
             *,
-            tx_num_start: int,
             height: int,
             add_undo_info: bool,
     ) -> Callable[[], None]:
 
+        tx_num_start = self.db.tx_counts[height - 1] if height > 0 else 0
+
         # Use local vars for speed in the loops
         bl_undo_info = [b"" for _ in txs]  # type: list[bytes]
         spend_utxo = self.spend_utxo
@@ -606,7 +600,7 @@ def process_txins_for_chunk(txs_chunk):
             for tx_pos, tx in txs_chunk:
                 process_txins_for_single_tx(tx_pos=tx_pos, tx=tx)
 
-        list(self.pool_executor.map(
+        list(self.pool_executor2.map(
             process_txins_for_chunk,
             chunks(list(enumerate(txs)), 200),
         ))
@@ -865,19 +859,17 @@ async def fetch_and_process_blocks(self, caught_up_event: asyncio.Event) -> None
         '''
         self._caught_up_event = caught_up_event
         await self._first_open_dbs()
-        XPoolExecutor = ThreadPoolExecutor
-        # XPoolExecutor = InterpreterPoolExecutor or ThreadPoolExecutor
-        self.logger.info(f"using pool executor: {XPoolExecutor}")
-        with XPoolExecutor(initializer=_init_pool, initargs=(self.coin,)) as self.pool_executor:
-            try:
-                async with OldTaskGroup() as group:
-                    await group.spawn(self.prefetcher.main_loop(self.height))
-                    await group.spawn(self._process_prefetched_blocks())
-            # Don't flush for arbitrary exceptions as they might be a cause or consequence of
-            # corrupted data
-            except CancelledError:
-                self.logger.info('flushing to DB for a clean shutdown...')
-                await self.flush(True)
+        with ThreadPoolExecutor() as self.pool_executor1:
+            with ThreadPoolExecutor() as self.pool_executor2:
+                try:
+                    async with OldTaskGroup() as group:
+                        await group.spawn(self.prefetcher.main_loop(self.height))
+                        await group.spawn(self._process_prefetched_blocks())
+                # Don't flush for arbitrary exceptions as they might be a cause or consequence of
+                # corrupted data
+                except CancelledError:
+                    self.logger.info('flushing to DB for a clean shutdown...')
+                    await self.flush(True)
 
     def force_chain_reorg(self, count: int) -> bool:
         '''Force a reorg of the given number of blocks.