Update Amt API and prep release

blockchain/chain_sync/src/lib.rs

@@ -18,6 +18,5 @@ extern crate serde;
 pub use self::bad_block_cache::BadBlockCache;
 pub use self::errors::Error;
 pub use self::network_context::SyncNetworkContext;
-pub use self::sync::{ChainSyncer, SyncConfig};
+pub use self::sync::{compute_msg_meta, ChainSyncer, SyncConfig};
 pub use self::sync_state::{SyncStage, SyncState};
-pub use self::sync_worker::compute_msg_meta;

blockchain/chain_sync/src/sync.rs

@@ -639,8 +639,8 @@ pub fn compute_msg_meta<DB: BlockStore>(
     let secp_cids = cids_from_messages(secp_msgs)?;
 
     // generate Amt and batch set message values
-    let bls_root = Amt::new_from_slice(blockstore, &bls_cids)?;
-    let secp_root = Amt::new_from_slice(blockstore, &secp_cids)?;
+    let bls_root = Amt::new_from_iter(blockstore, bls_cids)?;
+    let secp_root = Amt::new_from_iter(blockstore, secp_cids)?;
 
     let meta = TxMeta {
         bls_message_root: bls_root,

blockchain/chain_sync/src/sync_worker.rs

@@ -7,20 +7,22 @@ mod full_sync_test;
 mod validate_block_test;
 
 use super::sync_state::{SyncStage, SyncState};
-use super::{bad_block_cache::BadBlockCache, sync::ChainSyncState};
+use super::{
+    bad_block_cache::BadBlockCache,
+    sync::{compute_msg_meta, ChainSyncState},
+};
 use super::{Error, SyncNetworkContext};
 use actor::{is_account_actor, power};
 use address::Address;
-use amt::Amt;
 use async_std::channel::Receiver;
 use async_std::sync::{Mutex, RwLock};
 use async_std::task::{self, JoinHandle};
 use beacon::{Beacon, BeaconEntry, BeaconSchedule, IGNORE_DRAND_VAR};
-use blocks::{Block, BlockHeader, FullTipset, Tipset, TipsetKeys, TxMeta};
+use blocks::{Block, BlockHeader, FullTipset, Tipset, TipsetKeys};
 use chain::{persist_objects, ChainStore};
-use cid::{Cid, Code::Blake2b256};
+use cid::Cid;
 use crypto::{verify_bls_aggregate, DomainSeparationTag};
-use encoding::{Cbor, Error as EncodingError};
+use encoding::Cbor;
 use fil_types::{
     verifier::ProofVerifier, NetworkVersion, Randomness, ALLOWABLE_CLOCK_DRIFT, BLOCK_GAS_LIMIT,
     TICKET_RANDOMNESS_LOOKBACK,
@@ -30,7 +32,7 @@ use futures::stream::{FuturesUnordered, StreamExt};
 use interpreter::price_list_by_epoch;
 use ipld_blockstore::BlockStore;
 use log::{debug, info, warn};
-use message::{Message, SignedMessage, UnsignedMessage};
+use message::{Message, UnsignedMessage};
 use networks::{get_network_version_default, BLOCK_DELAY_SECS, UPGRADE_SMOKE_HEIGHT};
 use state_manager::StateManager;
 use state_tree::StateTree;
@@ -980,37 +982,6 @@ fn block_sanity_checks(header: &BlockHeader) -> Result<(), &'static str> {
     Ok(())
 }
 
-/// Returns message root CID from bls and secp message contained in the param Block
-pub fn compute_msg_meta<DB: BlockStore>(
-    blockstore: &DB,
-    bls_msgs: &[UnsignedMessage],
-    secp_msgs: &[SignedMessage],
-) -> Result<Cid, Error> {
-    // collect bls and secp cids
-    let bls_cids = cids_from_messages(bls_msgs)?;
-    let secp_cids = cids_from_messages(secp_msgs)?;
-
-    // generate Amt and batch set message values
-    let bls_root = Amt::new_from_slice(blockstore, &bls_cids)?;
-    let secp_root = Amt::new_from_slice(blockstore, &secp_cids)?;
-
-    let meta = TxMeta {
-        bls_message_root: bls_root,
-        secp_message_root: secp_root,
-    };
-
-    // store message roots and receive meta_root cid
-    let meta_root = blockstore
-        .put(&meta, Blake2b256)
-        .map_err(|e| Error::Other(e.to_string()))?;
-
-    Ok(meta_root)
-}
-
-fn cids_from_messages<T: Cbor>(messages: &[T]) -> Result<Vec<Cid>, EncodingError> {
-    messages.iter().map(Cbor::cid).collect()
-}
-
 #[cfg(test)]
 mod tests {
     use super::*;

blockchain/state_manager/src/lib.rs

@@ -243,8 +243,7 @@ where
         let receipts = vm.apply_block_messages(messages, parent_epoch, epoch, callback)?;
 
         // Construct receipt root from receipts
-        // TODO this should use an amt relative to the version. This doesn't matter for us yet
-        let rect_root = Amt::new_from_slice(self.blockstore(), &receipts)?;
+        let rect_root = Amt::new_from_iter(self.blockstore(), receipts)?;
 
         // Flush changes to blockstore
         let state_root = vm.flush()?;

ipld/amt/Cargo.toml

@@ -1,7 +1,6 @@
 [package]
 name = "ipld_amt"
 description = "Sharded IPLD Array implementation."
-# TODO bump the major version when releasing v2 actors
 version = "0.2.0"
 license = "MIT OR Apache-2.0"
 authors = ["ChainSafe Systems <info@chainsafe.io>"]
@@ -28,4 +27,4 @@ ipld_blockstore = { version = "0.1", features = ["tracking"] }
 [[bench]]
 name = "amt_benchmark"
 path = "benches/amt_benchmark.rs"
-harness = false
+harness = false

ipld/amt/benches/amt_benchmark.rs

@@ -105,14 +105,14 @@ fn from_slice(c: &mut Criterion) {
     c.bench_function("AMT initialization from slice", |b| {
         b.iter(|| {
             let db = db::MemoryDB::default();
-            Amt::new_from_slice(&db, black_box(VALUES)).unwrap();
+            Amt::new_from_iter(&db, black_box(VALUES.iter().copied())).unwrap();
         })
     });
 }
 
 fn for_each(c: &mut Criterion) {
     let db = db::MemoryDB::default();
-    let cid = Amt::new_from_slice(&db, black_box(VALUES)).unwrap();
+    let cid = Amt::new_from_iter(&db, black_box(VALUES.iter().copied())).unwrap();
 
     c.bench_function("AMT for_each function", |b| {
         b.iter(|| {

ipld/amt/src/amt.rs

@@ -83,10 +83,10 @@ where
     }
 
     /// Generates an AMT with block store and array of cbor marshallable objects and returns Cid
-    pub fn new_from_slice(block_store: &'db BS, vals: &[V]) -> Result<Cid, Error>
-    where
-        V: Clone,
-    {
+    pub fn new_from_iter(
+        block_store: &'db BS,
+        vals: impl IntoIterator<Item = V>,
+    ) -> Result<Cid, Error> {
         let mut t = Self::new(block_store);
 
         t.batch_set(vals)?;
@@ -167,12 +167,9 @@ where
 
     /// Batch set (naive for now)
     // TODO Implement more efficient batch set to not have to traverse tree and keep cache for each
-    pub fn batch_set(&mut self, vals: &[V]) -> Result<(), Error>
-    where
-        V: Clone,
-    {
-        for (i, val) in vals.iter().enumerate() {
-            self.set(i as u64, val.clone())?;
+    pub fn batch_set(&mut self, vals: impl IntoIterator<Item = V>) -> Result<(), Error> {
+        for (i, val) in vals.into_iter().enumerate() {
+            self.set(i as u64, val)?;
         }
 
         Ok(())

node/rpc/src/state_api.rs

@@ -669,8 +669,9 @@ pub(crate) async fn miner_create_block<
         }
     }
 
-    let bls_msg_root = Amt::new_from_slice(data.chain_store.blockstore(), &bls_cids)?;
-    let secp_msg_root = Amt::new_from_slice(data.chain_store.blockstore(), &secp_cids)?;
+    let bls_msg_root = Amt::new_from_iter(data.chain_store.blockstore(), bls_cids.iter().copied())?;
+    let secp_msg_root =
+        Amt::new_from_iter(data.chain_store.blockstore(), secp_cids.iter().copied())?;
 
     let mmcid = data.chain_store.blockstore().put(
         &TxMeta {