adjust find_sectors_by_expiration for not re-quantizing

Author: zhiqiangxu

actors/miner/src/expiration_queue.rs

@@ -6,6 +6,7 @@ use std::convert::TryInto;
 
 use anyhow::{anyhow, Context};
 use cid::Cid;
+use fil_actors_runtime::network::EPOCHS_IN_DAY;
 use fil_actors_runtime::runtime::Policy;
 use fil_actors_runtime::{ActorDowncast, Array};
 use fvm_ipld_amt::{Error as AmtError, ValueMut};
@@ -776,47 +777,63 @@ impl<'db, BS: Blockstore> ExpirationQueue<'db, BS> {
         sector_size: SectorSize,
         sectors: &[SectorOnChainInfo],
     ) -> anyhow::Result<Vec<SectorExpirationSet>> {
-        let mut declared_expirations = BTreeMap::<ChainEpoch, bool>::new();
+        if sectors.is_empty() {
+            return Ok(Vec::new());
+        }
+
         let mut sectors_by_number = BTreeMap::<u64, &SectorOnChainInfo>::new();
         let mut all_remaining = BTreeSet::<u64>::new();
 
+        let mut min_expire = i64::MAX;
+        let mut max_expire = i64::MIN;
         for sector in sectors {
-            let q_expiration = self.quant.quantize_up(sector.expiration);
-            declared_expirations.insert(q_expiration, true);
+            if sector.expiration < min_expire {
+                min_expire = sector.expiration;
+            }
+            if sector.expiration > max_expire {
+                max_expire = sector.expiration;
+            }
             all_remaining.insert(sector.sector_number);
             sectors_by_number.insert(sector.sector_number, sector);
         }
 
-        let mut expiration_groups =
-            Vec::<SectorExpirationSet>::with_capacity(declared_expirations.len());
+        let mut expiration_groups = Vec::<SectorExpirationSet>::with_capacity(sectors.len());
 
-        for (&expiration, _) in declared_expirations.iter() {
-            let es = self.may_get(expiration)?;
+        let end_epoch = (max_expire + EPOCHS_IN_DAY) as u64;
+        self.amt.for_each_while_ranged(Some(min_expire as u64), None, |epoch, es| {
+            if epoch > end_epoch {
+                return Ok(false);
+            }
 
             let group = group_expiration_set(
                 sector_size,
                 &sectors_by_number,
                 &mut all_remaining,
                 es,
-                expiration,
+                epoch as ChainEpoch,
             );
             if !group.sector_epoch_set.sectors.is_empty() {
                 expiration_groups.push(group);
             }
-        }
+
+            Ok(!all_remaining.is_empty())
+        })?;
 
         // If sectors remain, traverse next in epoch order. Remaining sectors should be
         // rescheduled to expire soon, so this traversal should exit early.
         if !all_remaining.is_empty() {
+            // avoid type cast in the for_each_while below
+            let min_expire = min_expire as u64;
             self.amt.for_each_while(|epoch, es| {
-                let epoch = epoch as ChainEpoch;
-                // If this set's epoch is one of our declared epochs, we've already processed it
+                // If this set's epoch is in range [min_expire, end_epoch], we've already processed it
                 // in the loop above, so skip processing here. Sectors rescheduled to this epoch
                 // would have been included in the earlier processing.
-                if declared_expirations.contains_key(&epoch) {
+                if epoch >= min_expire && epoch <= end_epoch {
                     return Ok(true);
                 }
 
+                let epoch = epoch as ChainEpoch;
+
                 // Sector should not be found in EarlyExpirations which holds faults. An implicit assumption
                 // of grouping is that it only returns sectors with active power. ExpirationQueue should not
                 // provide operations that allow this to happen.
@@ -826,7 +843,7 @@ impl<'db, BS: Blockstore> ExpirationQueue<'db, BS> {
                     sector_size,
                     &sectors_by_number,
                     &mut all_remaining,
-                    es.clone(),
+                    es,
                     epoch,
                 );
 
@@ -911,7 +928,7 @@ fn group_expiration_set(
     sector_size: SectorSize,
     sectors: &BTreeMap<u64, &SectorOnChainInfo>,
     include_set: &mut BTreeSet<u64>,
-    es: ExpirationSet,
+    es: &ExpirationSet,
     expiration: ChainEpoch,
 ) -> SectorExpirationSet {
     let mut sector_numbers = Vec::new();
@@ -927,14 +944,26 @@ fn group_expiration_set(
         }
     }
 
-    SectorExpirationSet {
-        sector_epoch_set: SectorEpochSet {
-            epoch: expiration,
-            sectors: sector_numbers,
-            power: total_power,
-            pledge: total_pledge,
-        },
-        expiration_set: es,
+    if sector_numbers.is_empty() {
+        SectorExpirationSet {
+            sector_epoch_set: SectorEpochSet {
+                epoch: expiration,
+                sectors: sector_numbers,
+                power: total_power,
+                pledge: total_pledge,
+            },
+            expiration_set: ExpirationSet::default(),
+        }
+    } else {
+        SectorExpirationSet {
+            sector_epoch_set: SectorEpochSet {
+                epoch: expiration,
+                sectors: sector_numbers,
+                power: total_power,
+                pledge: total_pledge,
+            },
+            expiration_set: es.clone(), // lazy clone
+        }
     }
 }