perf(miner,market): batch settlement of TerminateSectors

actors/market/src/lib.rs

@@ -799,6 +799,15 @@ impl Actor {
             let proposals = st.load_proposals(rt.store())?;
             let states = st.load_deal_states(rt.store())?;
 
+            // Load the balance tables once too, rather than letting each slashed deal below
+            // independently load-and-reflush them: a sector can carry many deals, and most of
+            // that per-deal round trip is redundant when they all touch the same couple of
+            // tables.
+            let mut escrow_table =
+                BalanceTable::from_root(rt.store(), &st.escrow_table, "escrow table")?;
+            let mut locked_table =
+                BalanceTable::from_root(rt.store(), &st.locked_table, "locked table")?;
+
             // The sector deals mapping is removed all at once.
             // Note there may be some deal states that are not removed here,
             // despite deletion of this mapping, e.g. for expired but not-yet-settled deals.
@@ -862,7 +871,8 @@ impl Actor {
                 }
 
                 state.slash_epoch = params.epoch;
-                total_slashed += st.process_slashed_deal(rt.store(), &deal, &state)?;
+                total_slashed +=
+                    st.process_slashed_deal(&mut escrow_table, &mut locked_table, &deal, &state)?;
                 st.remove_completed_deal(rt.store(), id)?;
 
                 emit::deal_terminated(
@@ -873,6 +883,9 @@ impl Actor {
                 )?;
             }
 
+            st.escrow_table = escrow_table.root()?;
+            st.locked_table = locked_table.root()?;
+
             Ok(total_slashed)
         })?;
 

actors/market/src/state.rs

@@ -919,9 +919,16 @@ impl State {
         Ok((TokenAmount::zero(), elapsed_payment, false, false))
     }
 
+    /// Takes already-loaded balance tables rather than loading (and re-flushing) them from
+    /// `self.escrow_table`/`self.locked_table` internally. Callers processing many deals in one
+    /// transaction (e.g. sector termination) should load the tables once, call this once per
+    /// deal, and flush the tables once at the end -- each independent load/flush of a balance
+    /// table is a real HAMT round trip, and there's no need to pay for one per deal when most
+    /// of them touch the same one or two addresses.
     pub fn process_slashed_deal<BS>(
         &mut self,
-        store: &BS,
+        escrow_table: &mut BalanceTable<BS>,
+        locked_table: &mut BalanceTable<BS>,
         proposal: &DealProposal,
         state: &DealState,
     ) -> Result<TokenAmount, ActorError>
@@ -934,19 +941,30 @@ impl State {
         let num_epochs_elapsed = max(0, payment_end_epoch - payment_start_epoch);
         let total_payment = &proposal.storage_price_per_epoch * num_epochs_elapsed;
         if total_payment.is_positive() {
-            self.transfer_balance(store, &proposal.client, &proposal.provider, &total_payment)?;
+            self.transfer_balance_in_tables(
+                escrow_table,
+                locked_table,
+                &proposal.client,
+                &proposal.provider,
+                &total_payment,
+            )?;
         }
 
         // unlock client collateral and locked storage fee
         let payment_remaining = deal_get_payment_remaining(proposal, state.slash_epoch)?;
 
         // Unlock remaining storage fee
-        self.unlock_balance(store, &proposal.client, &payment_remaining, Reason::ClientStorageFee)
-            .context("unlocking client storage fee")?;
+        self.unlock_balance_in_table(
+            locked_table,
+            &proposal.client,
+            &payment_remaining,
+            Reason::ClientStorageFee,
+        )
+        .context("unlocking client storage fee")?;
 
         // Unlock client collateral
-        self.unlock_balance(
-            store,
+        self.unlock_balance_in_table(
+            locked_table,
             &proposal.client,
             &proposal.client_collateral,
             Reason::ClientCollateral,
@@ -955,8 +973,14 @@ impl State {
 
         // slash provider collateral
         let slashed = proposal.provider_collateral.clone();
-        self.slash_balance(store, &proposal.provider, &slashed, Reason::ProviderCollateral)
-            .context("slashing balance")?;
+        self.slash_balance_in_tables(
+            escrow_table,
+            locked_table,
+            &proposal.provider,
+            &slashed,
+            Reason::ProviderCollateral,
+        )
+        .context("slashing balance")?;
 
         Ok(slashed)
     }
@@ -1097,9 +1121,11 @@ impl State {
         Ok(())
     }
 
-    fn unlock_balance<BS>(
+    /// Takes an already-loaded locked-balance table instead of loading (and re-flushing) it
+    /// from `self.locked_table` internally. See `process_slashed_deal` for why this matters.
+    fn unlock_balance_in_table<BS>(
         &mut self,
-        store: &BS,
+        locked_table: &mut BalanceTable<BS>,
         addr: &Address,
         amount: &TokenAmount,
         lock_reason: Reason,
@@ -1111,7 +1137,6 @@ impl State {
             return Err(actor_error!(illegal_state, "unlock negative amount: {}", amount));
         }
 
-        let mut locked_table = BalanceTable::from_root(store, &self.locked_table, "locked table")?;
         locked_table.must_subtract(addr, amount).context("unlocking balance")?;
 
         match lock_reason {
@@ -1126,14 +1151,33 @@ impl State {
             }
         };
 
-        self.locked_table = locked_table.root()?;
         Ok(())
     }
 
-    /// move funds from locked in client to available in provider
-    fn transfer_balance<BS>(
+    fn unlock_balance<BS>(
         &mut self,
         store: &BS,
+        addr: &Address,
+        amount: &TokenAmount,
+        lock_reason: Reason,
+    ) -> Result<(), ActorError>
+    where
+        BS: Blockstore,
+    {
+        let mut locked_table = BalanceTable::from_root(store, &self.locked_table, "locked table")?;
+        self.unlock_balance_in_table(&mut locked_table, addr, amount, lock_reason)?;
+        self.locked_table = locked_table.root()?;
+        Ok(())
+    }
+
+    /// move funds from locked in client to available in provider.
+    /// Takes already-loaded balance tables instead of loading (and re-flushing) them from
+    /// `self.escrow_table`/`self.locked_table` internally. See `process_slashed_deal` for why
+    /// this matters.
+    fn transfer_balance_in_tables<BS>(
+        &mut self,
+        escrow_table: &mut BalanceTable<BS>,
+        locked_table: &mut BalanceTable<BS>,
         from_addr: &Address,
         to_addr: &Address,
         amount: &TokenAmount,
@@ -1145,22 +1189,48 @@ impl State {
             return Err(actor_error!(illegal_state, "transfer negative amount: {}", amount));
         }
 
-        let mut escrow_table = BalanceTable::from_root(store, &self.escrow_table, "escrow table")?;
-
         // Subtract from locked and escrow tables
         escrow_table.must_subtract(from_addr, amount)?;
-        self.unlock_balance(store, from_addr, amount, Reason::ClientStorageFee)
+        self.unlock_balance_in_table(locked_table, from_addr, amount, Reason::ClientStorageFee)
             .context("unlocking client balance")?;
 
         // Add subtracted amount to the recipient
         escrow_table.add(to_addr, amount)?;
-        self.escrow_table = escrow_table.root()?;
         Ok(())
     }
 
-    fn slash_balance<BS>(
+    /// move funds from locked in client to available in provider
+    fn transfer_balance<BS>(
         &mut self,
         store: &BS,
+        from_addr: &Address,
+        to_addr: &Address,
+        amount: &TokenAmount,
+    ) -> Result<(), ActorError>
+    where
+        BS: Blockstore,
+    {
+        let mut escrow_table = BalanceTable::from_root(store, &self.escrow_table, "escrow table")?;
+        let mut locked_table = BalanceTable::from_root(store, &self.locked_table, "locked table")?;
+        self.transfer_balance_in_tables(
+            &mut escrow_table,
+            &mut locked_table,
+            from_addr,
+            to_addr,
+            amount,
+        )?;
+        self.escrow_table = escrow_table.root()?;
+        self.locked_table = locked_table.root()?;
+        Ok(())
+    }
+
+    /// Takes already-loaded balance tables instead of loading (and re-flushing) them from
+    /// `self.escrow_table`/`self.locked_table` internally. See `process_slashed_deal` for why
+    /// this matters.
+    fn slash_balance_in_tables<BS>(
+        &mut self,
+        escrow_table: &mut BalanceTable<BS>,
+        locked_table: &mut BalanceTable<BS>,
         addr: &Address,
         amount: &TokenAmount,
         lock_reason: Reason,
@@ -1172,12 +1242,33 @@ impl State {
             return Err(actor_error!(illegal_state, "negative amount to slash: {}", amount));
         }
 
-        let mut escrow_table = BalanceTable::from_root(store, &self.escrow_table, "escrow table")?;
-
         // Subtract from locked and escrow tables
         escrow_table.must_subtract(addr, amount)?;
+        self.unlock_balance_in_table(locked_table, addr, amount, lock_reason)
+    }
+
+    fn slash_balance<BS>(
+        &mut self,
+        store: &BS,
+        addr: &Address,
+        amount: &TokenAmount,
+        lock_reason: Reason,
+    ) -> Result<(), ActorError>
+    where
+        BS: Blockstore,
+    {
+        let mut escrow_table = BalanceTable::from_root(store, &self.escrow_table, "escrow table")?;
+        let mut locked_table = BalanceTable::from_root(store, &self.locked_table, "locked table")?;
+        self.slash_balance_in_tables(
+            &mut escrow_table,
+            &mut locked_table,
+            addr,
+            amount,
+            lock_reason,
+        )?;
         self.escrow_table = escrow_table.root()?;
-        self.unlock_balance(store, addr, amount, lock_reason)
+        self.locked_table = locked_table.root()?;
+        Ok(())
     }
 }
 

actors/miner/src/deadline_state.rs

@@ -597,6 +597,7 @@ impl Deadline {
         Ok((popped, modified))
     }
 
+    /// Fast path when `record_termination` is false -- see `Partition::terminate_sectors`.
     #[allow(clippy::too_many_arguments)]
     pub fn terminate_sectors<BS: Blockstore>(
         &mut self,
@@ -607,10 +608,12 @@ impl Deadline {
         partition_sectors: &mut PartitionSectorMap,
         sector_size: SectorSize,
         quant: QuantSpec,
-    ) -> anyhow::Result<PowerPair> {
+        record_termination: bool,
+    ) -> anyhow::Result<(PowerPair, Vec<SectorOnChainInfo>)> {
         let mut partitions = self.partitions_amt(store)?;
 
         let mut power_lost = PowerPair::zero();
+        let mut terminated_sector_infos = Vec::new();
         for (partition_idx, sector_numbers) in partition_sectors.iter() {
             let mut partition = partitions
                 .get(partition_idx)
@@ -622,7 +625,7 @@ impl Deadline {
                 )?
                 .clone();
 
-            let (removed, removed_unproven) = partition
+            let (removed, removed_unproven, sector_infos) = partition
                 .terminate_sectors(
                     policy,
                     store,
@@ -631,6 +634,7 @@ impl Deadline {
                     sector_numbers,
                     sector_size,
                     quant,
+                    record_termination,
                 )
                 .map_err(|e| {
                     e.downcast_wrap(format!(
@@ -644,12 +648,14 @@ impl Deadline {
             })?;
 
             if !removed.is_empty() {
-                // Record that partition now has pending early terminations.
-                self.early_terminations.set(partition_idx);
+                if record_termination {
+                    // Record that partition now has pending early terminations.
+                    self.early_terminations.set(partition_idx);
+                } // else: fast path -- the caller settles fees itself, so no queue entry is recorded.
 
                 // Record change to sectors and power
                 self.live_sectors -= removed.len();
-            } // note: we should _always_ have early terminations, unless the early termination bitfield is empty.
+            }
 
             self.faulty_power -= &removed.faulty_power;
             self.live_power -= &removed.active_power;
@@ -659,13 +665,15 @@ impl Deadline {
 
             // Aggregate power lost from active sectors
             power_lost += &removed.active_power;
+
+            terminated_sector_infos.extend(sector_infos);
         }
 
         // save partitions back
         self.partitions =
             partitions.flush().map_err(|e| e.downcast_wrap("failed to persist partitions"))?;
 
-        Ok(power_lost)
+        Ok((power_lost, terminated_sector_infos))
     }
 
     /// RemovePartitions removes the specified partitions, shifting the remaining