CHIA-2823 move sp broadcast (#19654)

* move sp broadcast

* refactor to avoid double calls

* move up

Author: almogdepaz

chia/consensus/blockchain.py

@@ -613,20 +613,13 @@ async def _reconsider_peak(
             [fork_add.coin for fork_add in fork_info.additions_since_fork.values() if fork_add.is_coinbase],
         )
 
-    def get_next_difficulty(self, header_hash: bytes32, new_slot: bool) -> uint64:
+    def get_next_sub_slot_iters_and_difficulty(self, header_hash: bytes32, new_slot: bool) -> tuple[uint64, uint64]:
         curr = self.try_block_record(header_hash)
         assert curr is not None
         if curr.height <= 2:
-            return self.constants.DIFFICULTY_STARTING
+            return self.constants.SUB_SLOT_ITERS_STARTING, self.constants.DIFFICULTY_STARTING
 
-        return get_next_sub_slot_iters_and_difficulty(self.constants, new_slot, curr, self)[1]
-
-    def get_next_slot_iters(self, header_hash: bytes32, new_slot: bool) -> uint64:
-        curr = self.try_block_record(header_hash)
-        assert curr is not None
-        if curr.height <= 2:
-            return self.constants.SUB_SLOT_ITERS_STARTING
-        return get_next_sub_slot_iters_and_difficulty(self.constants, new_slot, curr, self)[0]
+        return get_next_sub_slot_iters_and_difficulty(self.constants, new_slot, curr, self)
 
     async def get_sp_and_ip_sub_slots(
         self, header_hash: bytes32

chia/full_node/full_node.py

@@ -102,6 +102,7 @@ class PeakPostProcessingResult:
     fns_peak_result: FullNodeStorePeakResult  # The result of calling FullNodeStore.new_peak
     hints: list[tuple[bytes32, bytes]]  # The hints added to the DB
     lookup_coin_ids: list[bytes32]  # The coin IDs that we need to look up to notify wallets of changes
+    signage_points: list[tuple[RespondSignagePoint, WSChiaConnection, Optional[EndOfSubSlotBundle]]]
 
 
 @dataclasses.dataclass(frozen=True)
@@ -837,7 +838,7 @@ async def send_peak_to_timelords(
             peak_block = await self.blockchain.get_full_peak()
         if peak_block is not None:
             peak = self.blockchain.block_record(peak_block.header_hash)
-            difficulty = self.blockchain.get_next_difficulty(peak.header_hash, False)
+            difficulty = self.blockchain.get_next_sub_slot_iters_and_difficulty(peak.header_hash, False)[1]
             ses: Optional[SubEpochSummary] = next_sub_epoch_summary(
                 self.constants,
                 self.blockchain,
@@ -1807,10 +1808,9 @@ async def signage_point_post_processing(
             # Makes sure to potentially update the difficulty if we are past the peak (into a new sub-slot)
             assert ip_sub_slot is not None
             if request.challenge_chain_vdf.challenge != ip_sub_slot.challenge_chain.get_hash():
-                next_difficulty = self.blockchain.get_next_difficulty(peak.header_hash, True)
-                next_sub_slot_iters = self.blockchain.get_next_slot_iters(peak.header_hash, True)
-                difficulty = next_difficulty
-                sub_slot_iters = next_sub_slot_iters
+                sub_slot_iters, difficulty = self.blockchain.get_next_sub_slot_iters_and_difficulty(
+                    peak.header_hash, True
+                )
         else:
             difficulty = self.constants.DIFFICULTY_STARTING
             sub_slot_iters = self.constants.SUB_SLOT_ITERS_STARTING
@@ -1845,8 +1845,7 @@ async def peak_post_processing(
         """
 
         record = state_change_summary.peak
-        difficulty = self.blockchain.get_next_difficulty(record.header_hash, False)
-        sub_slot_iters = self.blockchain.get_next_slot_iters(record.header_hash, False)
+        sub_slot_iters, difficulty = self.blockchain.get_next_sub_slot_iters_and_difficulty(record.header_hash, False)
 
         self.log.info(
             f"🌱 Updated peak to height {record.height}, weight {record.weight}, "
@@ -1895,6 +1894,7 @@ async def peak_post_processing(
             difficulty,
         )
 
+        signage_points: list[tuple[RespondSignagePoint, WSChiaConnection, Optional[EndOfSubSlotBundle]]] = []
         if fns_peak_result.new_signage_points is not None and peer is not None:
             for index, sp in fns_peak_result.new_signage_points:
                 assert (
@@ -1903,8 +1903,13 @@ async def peak_post_processing(
                     and sp.rc_vdf is not None
                     and sp.rc_proof is not None
                 )
-                await self.signage_point_post_processing(
-                    RespondSignagePoint(index, sp.cc_vdf, sp.cc_proof, sp.rc_vdf, sp.rc_proof), peer, sub_slots[1]
+                # Collect the data for networking outside the mutex
+                signage_points.append(
+                    (
+                        RespondSignagePoint(index, sp.cc_vdf, sp.cc_proof, sp.rc_vdf, sp.rc_proof),
+                        peer,
+                        sub_slots[1],
+                    )
                 )
 
         if sub_slots[1] is None:
@@ -1934,6 +1939,7 @@ async def peak_post_processing(
             fns_peak_result,
             hints_to_add,
             lookup_coin_ids,
+            signage_points=signage_points,
         )
 
     async def peak_post_processing_2(
@@ -1948,6 +1954,8 @@ async def peak_post_processing_2(
         with peers
         """
         record = state_change_summary.peak
+        for signage_point in ppp_result.signage_points:
+            await self.signage_point_post_processing(*signage_point)
         for new_peak_item in ppp_result.mempool_peak_result:
             self.log.debug(f"Added transaction to mempool: {new_peak_item.transaction_id}")
             mempool_item = self.mempool_manager.get_mempool_item(new_peak_item.transaction_id)
@@ -2673,8 +2681,9 @@ async def add_end_of_sub_slot(
 
             peak = self.blockchain.get_peak()
             if peak is not None and peak.height > 2:
-                next_sub_slot_iters = self.blockchain.get_next_slot_iters(peak.header_hash, True)
-                next_difficulty = self.blockchain.get_next_difficulty(peak.header_hash, True)
+                next_sub_slot_iters, next_difficulty = self.blockchain.get_next_sub_slot_iters_and_difficulty(
+                    peak.header_hash, True
+                )
             else:
                 next_sub_slot_iters = self.constants.SUB_SLOT_ITERS_STARTING
                 next_difficulty = self.constants.DIFFICULTY_STARTING

chia/full_node/full_node_api.py

@@ -726,7 +726,9 @@ async def respond_signage_point(
                 return None
             peak = self.full_node.blockchain.get_peak()
             if peak is not None and peak.height > self.full_node.constants.MAX_SUB_SLOT_BLOCKS:
-                next_sub_slot_iters = self.full_node.blockchain.get_next_slot_iters(peak.header_hash, True)
+                next_sub_slot_iters = self.full_node.blockchain.get_next_sub_slot_iters_and_difficulty(
+                    peak.header_hash, True
+                )[0]
                 sub_slots_for_peak = await self.full_node.blockchain.get_sp_and_ip_sub_slots(peak.header_hash)
                 assert sub_slots_for_peak is not None
                 ip_sub_slot: Optional[EndOfSubSlotBundle] = sub_slots_for_peak[1]