Swap to ThreadPoolExecutor and use chia_rs version of validate_clvm_and_signature()

benchmarks/mempool-long-lived.py

@@ -114,7 +114,7 @@ async def get_coin_record(coin_ids: Collection[bytes32]) -> List[CoinRecord]:
                 coin.name(): CoinRecord(coin, uint32(height // 2), uint32(0), False, uint64(timestamp // 2))
             }
             spend_bundle_id = sb.name()
-            sbc = await mempool.pre_validate_spendbundle(sb, None, spend_bundle_id)
+            sbc = await mempool.pre_validate_spendbundle(sb, spend_bundle_id)
             assert sbc is not None
             await mempool.add_spend_bundle(sb, sbc, spend_bundle_id, uint32(height))
 

benchmarks/mempool.py

@@ -173,7 +173,7 @@ async def get_unspent_lineage_info_for_puzzle_hash(_: bytes32) -> Optional[Unspe
         async def add_spend_bundles(spend_bundles: List[SpendBundle]) -> None:
             for tx in spend_bundles:
                 spend_bundle_id = tx.name()
-                npc = await mempool.pre_validate_spendbundle(tx, None, spend_bundle_id)
+                npc = await mempool.pre_validate_spendbundle(tx, spend_bundle_id)
                 assert npc is not None
                 info = await mempool.add_spend_bundle(tx, npc, spend_bundle_id, height)
                 assert info.status == MempoolInclusionStatus.SUCCESS

chia/_tests/core/full_node/test_full_node.py

@@ -935,8 +935,8 @@ async def test_new_transaction_and_mempool(self, wallet_nodes, self_hostname, se
         # these numbers reflect the capacity of the mempool. In these
         # tests MEMPOOL_BLOCK_BUFFER is 1. The other factors are COST_PER_BYTE
         # and MAX_BLOCK_COST_CLVM
-        assert included_tx == 23
-        assert not_included_tx == 10
+        assert included_tx == 27
+        assert not_included_tx == 6
         assert seen_bigger_transaction_has_high_fee
 
         # Mempool is full

chia/_tests/core/mempool/test_mempool_manager.py

@@ -387,7 +387,7 @@ def spend_bundle_from_conditions(
 async def add_spendbundle(
     mempool_manager: MempoolManager, sb: SpendBundle, sb_name: bytes32
 ) -> Tuple[Optional[uint64], MempoolInclusionStatus, Optional[Err]]:
-    sbc = await mempool_manager.pre_validate_spendbundle(sb, None, sb_name)
+    sbc = await mempool_manager.pre_validate_spendbundle(sb, sb_name)
     ret = await mempool_manager.add_spend_bundle(sb, sbc, sb_name, TEST_HEIGHT)
     invariant_check_mempool(mempool_manager.mempool)
     return ret.cost, ret.status, ret.error
@@ -461,7 +461,7 @@ async def test_empty_spend_bundle() -> None:
     mempool_manager = await instantiate_mempool_manager(zero_calls_get_coin_records)
     sb = SpendBundle([], G2Element())
     with pytest.raises(ValidationError, match="INVALID_SPEND_BUNDLE"):
-        await mempool_manager.pre_validate_spendbundle(sb, None, sb.name())
+        await mempool_manager.pre_validate_spendbundle(sb, sb.name())
 
 
 @pytest.mark.anyio
@@ -470,7 +470,7 @@ async def test_negative_addition_amount() -> None:
     conditions = [[ConditionOpcode.CREATE_COIN, IDENTITY_PUZZLE_HASH, -1]]
     sb = spend_bundle_from_conditions(conditions)
     with pytest.raises(ValidationError, match="COIN_AMOUNT_NEGATIVE"):
-        await mempool_manager.pre_validate_spendbundle(sb, None, sb.name())
+        await mempool_manager.pre_validate_spendbundle(sb, sb.name())
 
 
 @pytest.mark.anyio
@@ -481,7 +481,7 @@ async def test_valid_addition_amount() -> None:
     coin = Coin(IDENTITY_PUZZLE_HASH, IDENTITY_PUZZLE_HASH, max_amount)
     sb = spend_bundle_from_conditions(conditions, coin)
     # ensure this does not throw
-    _ = await mempool_manager.pre_validate_spendbundle(sb, None, sb.name())
+    _ = await mempool_manager.pre_validate_spendbundle(sb, sb.name())
 
 
 @pytest.mark.anyio
@@ -491,7 +491,7 @@ async def test_too_big_addition_amount() -> None:
     conditions = [[ConditionOpcode.CREATE_COIN, IDENTITY_PUZZLE_HASH, max_amount + 1]]
     sb = spend_bundle_from_conditions(conditions)
     with pytest.raises(ValidationError, match="COIN_AMOUNT_EXCEEDS_MAXIMUM"):
-        await mempool_manager.pre_validate_spendbundle(sb, None, sb.name())
+        await mempool_manager.pre_validate_spendbundle(sb, sb.name())
 
 
 @pytest.mark.anyio
@@ -503,18 +503,18 @@ async def test_duplicate_output() -> None:
     ]
     sb = spend_bundle_from_conditions(conditions)
     with pytest.raises(ValidationError, match="DUPLICATE_OUTPUT"):
-        await mempool_manager.pre_validate_spendbundle(sb, None, sb.name())
+        await mempool_manager.pre_validate_spendbundle(sb, sb.name())
 
 
 @pytest.mark.anyio
 async def test_block_cost_exceeds_max() -> None:
     mempool_manager = await instantiate_mempool_manager(zero_calls_get_coin_records)
     conditions = []
-    for i in range(2400):
+    for i in range(3800):
         conditions.append([ConditionOpcode.CREATE_COIN, IDENTITY_PUZZLE_HASH, i])
     sb = spend_bundle_from_conditions(conditions)
     with pytest.raises(ValidationError, match="BLOCK_COST_EXCEEDS_MAX"):
-        await mempool_manager.pre_validate_spendbundle(sb, None, sb.name())
+        await mempool_manager.pre_validate_spendbundle(sb, sb.name())
 
 
 @pytest.mark.anyio
@@ -524,31 +524,31 @@ async def test_double_spend_prevalidation() -> None:
     sb = spend_bundle_from_conditions(conditions)
     sb_twice: SpendBundle = SpendBundle.aggregate([sb, sb])
     with pytest.raises(ValidationError, match="DOUBLE_SPEND"):
-        await mempool_manager.pre_validate_spendbundle(sb_twice, None, sb_twice.name())
+        await mempool_manager.pre_validate_spendbundle(sb_twice, sb_twice.name())
 
 
 @pytest.mark.anyio
 async def test_minting_coin() -> None:
     mempool_manager = await instantiate_mempool_manager(zero_calls_get_coin_records)
     conditions = [[ConditionOpcode.CREATE_COIN, IDENTITY_PUZZLE_HASH, TEST_COIN_AMOUNT]]
     sb = spend_bundle_from_conditions(conditions)
-    _ = await mempool_manager.pre_validate_spendbundle(sb, None, sb.name())
+    _ = await mempool_manager.pre_validate_spendbundle(sb, sb.name())
     conditions = [[ConditionOpcode.CREATE_COIN, IDENTITY_PUZZLE_HASH, TEST_COIN_AMOUNT + 1]]
     sb = spend_bundle_from_conditions(conditions)
     with pytest.raises(ValidationError, match="MINTING_COIN"):
-        await mempool_manager.pre_validate_spendbundle(sb, None, sb.name())
+        await mempool_manager.pre_validate_spendbundle(sb, sb.name())
 
 
 @pytest.mark.anyio
 async def test_reserve_fee_condition() -> None:
     mempool_manager = await instantiate_mempool_manager(zero_calls_get_coin_records)
     conditions = [[ConditionOpcode.RESERVE_FEE, TEST_COIN_AMOUNT]]
     sb = spend_bundle_from_conditions(conditions)
-    _ = await mempool_manager.pre_validate_spendbundle(sb, None, sb.name())
+    _ = await mempool_manager.pre_validate_spendbundle(sb, sb.name())
     conditions = [[ConditionOpcode.RESERVE_FEE, TEST_COIN_AMOUNT + 1]]
     sb = spend_bundle_from_conditions(conditions)
     with pytest.raises(ValidationError, match="RESERVE_FEE_CONDITION_FAILED"):
-        await mempool_manager.pre_validate_spendbundle(sb, None, sb.name())
+        await mempool_manager.pre_validate_spendbundle(sb, sb.name())
 
 
 @pytest.mark.anyio
@@ -581,7 +581,7 @@ async def test_same_sb_twice_with_eligible_coin() -> None:
     sb = SpendBundle.aggregate([sb1, sb2])
     sb_name = sb.name()
     result = await add_spendbundle(mempool_manager, sb, sb_name)
-    expected_cost = uint64(10268283)
+    expected_cost = uint64(6_600_088)
     assert result == (expected_cost, MempoolInclusionStatus.SUCCESS, None)
     assert mempool_manager.get_spendbundle(sb_name) == sb
     result = await add_spendbundle(mempool_manager, sb, sb_name)
@@ -614,7 +614,7 @@ async def test_sb_twice_with_eligible_coin_and_different_spends_order() -> None:
     assert mempool_manager.get_spendbundle(sb_name) is None
     assert mempool_manager.get_spendbundle(reordered_sb_name) is None
     result = await add_spendbundle(mempool_manager, sb, sb_name)
-    expected_cost = uint64(13091510)
+    expected_cost = uint64(7_800_132)
     assert result == (expected_cost, MempoolInclusionStatus.SUCCESS, None)
     assert mempool_manager.get_spendbundle(sb_name) == sb
     assert mempool_manager.get_spendbundle(reordered_sb_name) is None
@@ -1053,7 +1053,7 @@ async def make_and_send_big_cost_sb(coin: Coin) -> None:
         g1 = sk.get_g1()
         sig = AugSchemeMPL.sign(sk, IDENTITY_PUZZLE_HASH, g1)
         aggsig = G2Element()
-        for _ in range(169):
+        for _ in range(318):
             conditions.append([ConditionOpcode.AGG_SIG_UNSAFE, g1, IDENTITY_PUZZLE_HASH])
             aggsig += sig
         conditions.append([ConditionOpcode.CREATE_COIN, IDENTITY_PUZZLE_HASH, coin.amount - 10_000_000])

chia/clvm/spend_sim.py

@@ -268,7 +268,6 @@ async def farm_block(
                     get_unspent_lineage_info_for_puzzle_hash=self.coin_store.get_unspent_lineage_info_for_puzzle_hash,
                     item_inclusion_filter=item_inclusion_filter,
                 )
-
                 if result is not None:
                     bundle, additions = result
                     generator_bundle = bundle
@@ -337,7 +336,7 @@ def __init__(self, service: SpendSim) -> None:
     async def push_tx(self, spend_bundle: SpendBundle) -> Tuple[MempoolInclusionStatus, Optional[Err]]:
         try:
             spend_bundle_id = spend_bundle.name()
-            sbc = await self.service.mempool_manager.pre_validate_spendbundle(spend_bundle, None, spend_bundle_id)
+            sbc = await self.service.mempool_manager.pre_validate_spendbundle(spend_bundle, spend_bundle_id)
         except ValidationError as e:
             return MempoolInclusionStatus.FAILED, e.code
         assert self.service.mempool_manager.peak is not None

chia/full_node/full_node.py

@@ -279,7 +279,6 @@ async def manage(self) -> AsyncIterator[None]:
             self._mempool_manager = MempoolManager(
                 get_coin_records=self.coin_store.get_coin_records,
                 consensus_constants=self.constants,
-                multiprocessing_context=self.multiprocessing_context,
                 single_threaded=single_threaded,
             )
 
@@ -2356,7 +2355,7 @@ async def add_transaction(
         else:
             try:
                 cost_result = await self.mempool_manager.pre_validate_spendbundle(
-                    transaction, tx_bytes, spend_name, self._bls_cache
+                    transaction, spend_name, self._bls_cache
                 )
             except ValidationError as e:
                 self.mempool_manager.remove_seen(spend_name)

chia/full_node/mempool.py

@@ -521,6 +521,7 @@ async def create_bundle_from_mempool_items(
                         unique_additions.extend(spend_data.additions)
                     cost_saving = 0
                 else:
+                    # Update this item with fast forwarded spends, if it has any.
                     await eligible_coin_spends.process_fast_forward_spends(
                         mempool_item=item,
                         get_unspent_lineage_info_for_puzzle_hash=get_unspent_lineage_info_for_puzzle_hash,

chia/full_node/mempool_manager.py

@@ -3,24 +3,21 @@
 import asyncio
 import logging
 import time
-from concurrent.futures import Executor
-from concurrent.futures.process import ProcessPoolExecutor
+from concurrent.futures import Executor, ThreadPoolExecutor
 from dataclasses import dataclass
-from multiprocessing.context import BaseContext
 from typing import Awaitable, Callable, Collection, Dict, List, Optional, Set, Tuple, TypeVar
 
-from chia_rs import ELIGIBLE_FOR_DEDUP, ELIGIBLE_FOR_FF, BLSCache, G1Element, supports_fast_forward
+from chia_rs import ELIGIBLE_FOR_DEDUP, ELIGIBLE_FOR_FF, BLSCache, supports_fast_forward, validate_clvm_and_signature
 from chiabip158 import PyBIP158
 
 from chia.consensus.block_record import BlockRecordProtocol
 from chia.consensus.constants import ConsensusConstants
 from chia.consensus.cost_calculator import NPCResult
 from chia.full_node.bitcoin_fee_estimator import create_bitcoin_fee_estimator
-from chia.full_node.bundle_tools import simple_solution_generator
 from chia.full_node.fee_estimation import FeeBlockInfo, MempoolInfo, MempoolItemInfo
 from chia.full_node.fee_estimator_interface import FeeEstimatorInterface
 from chia.full_node.mempool import MEMPOOL_ITEM_FEE_LIMIT, Mempool, MempoolRemoveInfo, MempoolRemoveReason
-from chia.full_node.mempool_check_conditions import get_name_puzzle_conditions, mempool_check_time_locks
+from chia.full_node.mempool_check_conditions import mempool_check_time_locks
 from chia.full_node.pending_tx_cache import ConflictTxCache, PendingTxCache
 from chia.types.blockchain_format.coin import Coin
 from chia.types.blockchain_format.sized_bytes import bytes32
@@ -32,7 +29,6 @@
 from chia.types.mempool_item import BundleCoinSpend, MempoolItem
 from chia.types.spend_bundle import SpendBundle
 from chia.types.spend_bundle_conditions import SpendBundleConditions
-from chia.util.condition_tools import pkm_pairs
 from chia.util.db_wrapper import SQLITE_INT_MAX
 from chia.util.errors import Err, ValidationError
 from chia.util.inline_executor import InlineExecutor
@@ -46,49 +42,6 @@
 MEMPOOL_MIN_FEE_INCREASE = uint64(10000000)
 
 
-# TODO: once the 1.8.0 soft-fork has activated, we don't really need to pass
-# the constants through here
-def validate_clvm_and_signature(
-    spend_bundle_bytes: bytes, max_cost: int, constants: ConsensusConstants, height: uint32
-) -> Tuple[Optional[Err], bytes, List[Tuple[bytes, bytes]], float]:
-    """
-    Validates CLVM and aggregate signature for a spendbundle. This is meant to be called under a ProcessPoolExecutor
-    in order to validate the heavy parts of a transaction in a different thread. Returns an optional error,
-    the NPCResult and a cache of the new pairings validated (if not error)
-    """
-
-    start_time = time.monotonic()
-    additional_data = constants.AGG_SIG_ME_ADDITIONAL_DATA
-
-    try:
-        bundle: SpendBundle = SpendBundle.from_bytes(spend_bundle_bytes)
-        program = simple_solution_generator(bundle)
-        # npc contains names of the coins removed, puzzle_hashes and their spend conditions
-        result: NPCResult = get_name_puzzle_conditions(
-            program, max_cost, mempool_mode=True, constants=constants, height=height
-        )
-
-        if result.error is not None:
-            return Err(result.error), b"", [], time.monotonic() - start_time
-
-        pks: List[G1Element] = []
-        msgs: List[bytes] = []
-        assert result.conds is not None
-        pks, msgs = pkm_pairs(result.conds, additional_data)
-
-        # Verify aggregated signature
-        cache = BLSCache(10000)
-        if not cache.aggregate_verify(pks, msgs, bundle.aggregated_signature):
-            return Err.BAD_AGGREGATE_SIGNATURE, b"", [], time.monotonic() - start_time
-        new_cache_entries: List[Tuple[bytes, bytes]] = cache.items()
-    except ValidationError as e:
-        return e.code, b"", [], time.monotonic() - start_time
-    except Exception:
-        return Err.UNKNOWN, b"", [], time.monotonic() - start_time
-
-    return None, bytes(result), new_cache_entries, time.monotonic() - start_time
-
-
 @dataclass
 class TimelockConditions:
     assert_height: uint32 = uint32(0)
@@ -188,7 +141,6 @@ def __init__(
         self,
         get_coin_records: Callable[[Collection[bytes32]], Awaitable[List[CoinRecord]]],
         consensus_constants: ConsensusConstants,
-        multiprocessing_context: Optional[BaseContext] = None,
         *,
         single_threaded: bool = False,
         max_tx_clvm_cost: Optional[uint64] = None,
@@ -220,9 +172,8 @@ def __init__(
         if single_threaded:
             self.pool = InlineExecutor()
         else:
-            self.pool = ProcessPoolExecutor(
+            self.pool = ThreadPoolExecutor(
                 max_workers=2,
-                mp_context=multiprocessing_context,
                 initializer=setproctitle,
                 initargs=(f"{getproctitle()}_mempool_worker",),
             )
@@ -306,16 +257,13 @@ def remove_seen(self, bundle_hash: bytes32) -> None:
     async def pre_validate_spendbundle(
         self,
         new_spend: SpendBundle,
-        new_spend_bytes: Optional[bytes],
         spend_name: bytes32,
         bls_cache: Optional[BLSCache] = None,
     ) -> SpendBundleConditions:
         """
         Errors are included within the cached_result.
         This runs in another process so we don't block the main thread
         """
-        if new_spend_bytes is None:
-            new_spend_bytes = bytes(new_spend)
 
         if new_spend.coin_spends == []:
             raise ValidationError(Err.INVALID_SPEND_BUNDLE, "Empty SpendBundle")
@@ -324,23 +272,28 @@ async def pre_validate_spendbundle(
 
         self._worker_queue_size += 1
         try:
-            err, cached_result_bytes, new_cache_entries, duration = await asyncio.get_running_loop().run_in_executor(
+            sbc, new_cache_entries, duration = await asyncio.get_running_loop().run_in_executor(
                 self.pool,
                 validate_clvm_and_signature,
-                new_spend_bytes,
+                new_spend,
                 self.max_tx_clvm_cost,
                 self.constants,
                 self.peak.height,
             )
+        except Exception as e:  # take returned TypeError and turn raise it as a ValidationError
+            if len(e.args) > 0:
+                error = Err(e.args[0])
+                raise ValidationError(error)
+            else:
+                raise ValidationError(Err.UNKNOWN)
         finally:
             self._worker_queue_size -= 1
 
-        if err is not None:
-            raise ValidationError(err)
         if bls_cache is not None:
-            bls_cache.update(new_cache_entries)
+            bls_cache.update([(e[0], bytes(e[1])) for e in new_cache_entries])
+
+        ret = NPCResult(None, sbc)
 
-        ret: NPCResult = NPCResult.from_bytes(cached_result_bytes)
         log.log(
             logging.DEBUG if duration < 2 else logging.WARNING,
             f"pre_validate_spendbundle took {duration:0.4f} seconds "
@@ -349,7 +302,9 @@ async def pre_validate_spendbundle(
         if ret.error is not None:
             raise ValidationError(Err(ret.error), "pre_validate_spendbundle failed")
         assert ret.conds is not None
-        return ret.conds
+        # TODO: this cost padding is temporary, until we update the mempool's
+        # block creation logic to take the block overhead into account
+        return ret.conds.replace(cost=ret.conds.cost + 10000000)
 
     async def add_spend_bundle(
         self,

chia/rpc/full_node_rpc_api.py

@@ -868,7 +868,7 @@ async def _validate_fee_estimate_cost(self, request: Dict[str, Any]) -> uint64:
             spend_bundle: SpendBundle = SpendBundle.from_json_dict(request["spend_bundle"])
             spend_name = spend_bundle.name()
             conds: SpendBundleConditions = await self.service.mempool_manager.pre_validate_spendbundle(
-                spend_bundle, None, spend_name
+                spend_bundle, spend_name
             )
             cost = conds.cost
         elif "cost" in request: