refactor: send (kernel & syscall)

This combines two changes that are somewhat interlinked. Really, they're
just hard to factor out into two different patches without a bunch of work.

Motivation:

1. Remove copying costs from send/return.
2. Avoid calling "syscall" kernel methods in the call manager.
3. I _really_ hated that `Kernel::block_get` method.
4. Remove an extra call on send to "stat" the returned object.

----

1. In the kernel, the `send` function now takes/returns a block ID. This
moves all the parameter-relates logic into the kernel itself. Otherwise,
the syscall has to make three kernel calls, all of which charge gas.

2. In the call manager, instead of passing return values & parameters as
raw bytes, we pass them as `Option<kernel::Block>` where `kernel::Block`
is a reference counted IPLD block (with a codec). This, let's us pass
them around without copying, and lets us be very clear about codecs,
empty parameters, etc.

3. In the syscalls, return the block length/codec from send to avoid a
second syscall _just_ to look those up.

Author: Stebalien

fvm/src/call_manager/default.rs

@@ -1,18 +1,19 @@
 use anyhow::Context;
 use derive_more::{Deref, DerefMut};
-use fvm_ipld_encoding::{RawBytes, DAG_CBOR};
+use fvm_ipld_encoding::{to_vec, DAG_CBOR};
 use fvm_shared::actor::builtin::Type;
 use fvm_shared::address::{Address, Protocol};
 use fvm_shared::econ::TokenAmount;
 use fvm_shared::error::{ErrorNumber, ExitCode};
+use fvm_shared::sys::BlockId;
 use fvm_shared::{ActorID, MethodNum, METHOD_SEND};
 use num_traits::Zero;
 
 use super::{Backtrace, CallManager, InvocationResult, NO_DATA_BLOCK_ID};
 use crate::call_manager::backtrace::Frame;
 use crate::call_manager::FinishRet;
 use crate::gas::GasTracker;
-use crate::kernel::{ExecutionError, Kernel, Result, SyscallError};
+use crate::kernel::{Block, BlockRegistry, ExecutionError, Kernel, Result, SyscallError};
 use crate::machine::Machine;
 use crate::syscalls::error::Abort;
 use crate::syscalls::{charge_for_exec, update_gas_available};
@@ -86,7 +87,7 @@ where
         from: ActorID,
         to: Address,
         method: MethodNum,
-        params: &RawBytes,
+        params: Option<Block>,
         value: &TokenAmount,
     ) -> Result<InvocationResult>
     where
@@ -97,7 +98,10 @@ where
                 from,
                 to,
                 method,
-                params: params.clone(),
+                params: params
+                    .as_ref()
+                    .map(|blk| blk.data().to_owned().into())
+                    .unwrap_or_default(),
                 value: value.clone(),
             }));
         }
@@ -233,15 +237,15 @@ where
 
         // Now invoke the constructor; first create the parameters, then
         // instantiate a new kernel to invoke the constructor.
-        let params = RawBytes::serialize(&addr)
+        let params = to_vec(&addr)
             // TODO(#198) this should be a Sys actor error, but we're copying lotus here.
             .map_err(|e| syscall_error!(Serialization; "failed to serialize params: {}", e))?;
 
         self.send_resolved::<K>(
             account_actor::SYSTEM_ACTOR_ID,
             id,
             fvm_shared::METHOD_CONSTRUCTOR,
-            &params,
+            Some(Block::new(DAG_CBOR, params)),
             &TokenAmount::from(0u32),
         )?;
 
@@ -254,7 +258,7 @@ where
         from: ActorID,
         to: Address,
         method: MethodNum,
-        params: &RawBytes,
+        params: Option<Block>,
         value: &TokenAmount,
     ) -> Result<InvocationResult>
     where
@@ -284,7 +288,7 @@ where
         from: ActorID,
         to: ActorID,
         method: MethodNum,
-        params: &RawBytes,
+        params: Option<Block>,
         value: &TokenAmount,
     ) -> Result<InvocationResult>
     where
@@ -310,31 +314,29 @@ where
             return Ok(InvocationResult::Return(Default::default()));
         }
 
+        // Store the parametrs, and initialize the block registry for the target actor.
+        // TODO: In M2, the block registry may have some form of shared block limit.
+        let mut block_registry = BlockRegistry::new();
+        let params_id = if let Some(blk) = params {
+            block_registry.put(blk)?
+        } else {
+            NO_DATA_BLOCK_ID
+        };
+
         // This is a cheap operation as it doesn't actually clone the struct,
         // it returns a referenced copy.
         let engine = self.engine().clone();
 
         log::trace!("calling {} -> {}::{}", from, to, method);
         self.map_mut(|cm| {
             // Make the kernel.
-            let mut kernel = K::new(cm, from, to, method, value.clone());
-
-            // Store parameters, if any.
-            let param_id = if params.len() > 0 {
-                match kernel.block_create(DAG_CBOR, params) {
-                    Ok(id) => id,
-                    // This could fail if we pass some global memory limit.
-                    Err(err) => return (Err(err), kernel.into_call_manager()),
-                }
-            } else {
-                super::NO_DATA_BLOCK_ID
-            };
+            let kernel = K::new(cm, block_registry, from, to, method, value.clone());
 
             // Make a store.
             let mut store = engine.new_store(kernel);
 
             // From this point on, there are no more syscall errors, only aborts.
-            let result: std::result::Result<RawBytes, Abort> = (|| {
+            let result: std::result::Result<BlockId, Abort> = (|| {
                 // Instantiate the module.
                 let instance = engine
                     .get_instance(&mut store, &state.code)
@@ -358,39 +360,40 @@ where
                 update_gas_available(&mut store)?;
 
                 // Invoke it.
-                let res = invoke.call(&mut store, (param_id,));
+                let res = invoke.call(&mut store, (params_id,));
 
                 // Charge for any remaining uncharged execution gas, returning an error if we run
                 // out.
                 charge_for_exec(&mut store)?;
 
-                // If the invocation failed due to running out of exec_units, we have already detected it and returned OutOfGas above.
-                // Any other invocation failure is returned here as an Abort
-                let return_block_id = res?;
-
-                // Extract the return value, if there is one.
-                let return_value: RawBytes = if return_block_id > NO_DATA_BLOCK_ID {
-                    let (code, ret) = store
-                        .data_mut()
-                        .kernel
-                        .block_get(return_block_id)
-                        .map_err(|e| Abort::from_error(ExitCode::SYS_MISSING_RETURN, e))?;
-                    debug_assert_eq!(code, DAG_CBOR);
-                    RawBytes::new(ret)
-                } else {
-                    RawBytes::default()
-                };
-
-                Ok(return_value)
+                // If the invocation failed due to running out of exec_units, we have already
+                // detected it and returned OutOfGas above. Any other invocation failure is returned
+                // here as an Abort
+                Ok(res?)
             })();
 
             let invocation_data = store.into_data();
             let last_error = invocation_data.last_error;
-            let mut cm = invocation_data.kernel.into_call_manager();
+            let (mut cm, block_registry) = invocation_data.kernel.into_inner();
+
+            // Resolve the return block's ID into an actual block, converting to an abort if it
+            // doesn't exist.
+            let result = result.and_then(|ret_id| {
+                Ok(if ret_id == NO_DATA_BLOCK_ID {
+                    None
+                } else {
+                    Some(block_registry.get(ret_id).map_err(|_| {
+                        Abort::Exit(
+                            ExitCode::SYS_MISSING_RETURN,
+                            String::from("returned block does not exist"),
+                        )
+                    })?)
+                })
+            });
 
             // Process the result, updating the backtrace if necessary.
             let ret = match result {
-                Ok(value) => Ok(InvocationResult::Return(value)),
+                Ok(ret) => Ok(InvocationResult::Return(ret.cloned())),
                 Err(abort) => {
                     if let Some(err) = last_error {
                         cm.backtrace.begin(err);

fvm/src/call_manager/mod.rs

@@ -1,11 +1,10 @@
-use fvm_ipld_encoding::RawBytes;
 use fvm_shared::address::Address;
 use fvm_shared::econ::TokenAmount;
 use fvm_shared::error::ExitCode;
 use fvm_shared::{ActorID, MethodNum};
 
 use crate::gas::{GasCharge, GasTracker, PriceList};
-use crate::kernel::Result;
+use crate::kernel::{self, Result};
 use crate::machine::{Machine, MachineContext};
 use crate::state_tree::StateTree;
 use crate::Kernel;
@@ -52,7 +51,7 @@ pub trait CallManager: 'static {
         from: ActorID,
         to: Address,
         method: MethodNum,
-        params: &RawBytes,
+        params: Option<kernel::Block>,
         value: &TokenAmount,
     ) -> Result<InvocationResult>;
 
@@ -125,7 +124,7 @@ pub trait CallManager: 'static {
 #[derive(Clone, Debug)]
 pub enum InvocationResult {
     /// Indicates that the actor successfully returned. The value may be empty.
-    Return(RawBytes),
+    Return(Option<kernel::Block>),
     /// Indicates that the actor aborted with the given exit code.
     Failure(ExitCode),
 }

fvm/src/executor/default.rs

@@ -3,6 +3,7 @@ use std::result::Result as StdResult;
 
 use anyhow::{anyhow, Result};
 use cid::Cid;
+use fvm_ipld_encoding::{RawBytes, DAG_CBOR};
 use fvm_shared::actor::builtin::Type;
 use fvm_shared::address::Address;
 use fvm_shared::bigint::{BigInt, Sign};
@@ -16,7 +17,7 @@ use num_traits::Zero;
 use super::{ApplyFailure, ApplyKind, ApplyRet, Executor};
 use crate::call_manager::{backtrace, CallManager, InvocationResult};
 use crate::gas::{milligas_to_gas, GasCharge, GasOutputs};
-use crate::kernel::{ClassifyResult, Context as _, ExecutionError, Kernel};
+use crate::kernel::{Block, ClassifyResult, Context as _, ExecutionError, Kernel};
 use crate::machine::{Machine, BURNT_FUNDS_ACTOR_ADDR, REWARD_ACTOR_ADDR};
 
 /// The default [`Executor`].
@@ -67,14 +68,21 @@ where
                 return (Err(e), cm.finish().1);
             }
 
+            let params = if msg.params.is_empty() {
+                None
+            } else {
+                Some(Block::new(DAG_CBOR, msg.params.bytes()))
+            };
+
             let result = cm.with_transaction(|cm| {
                 // Invoke the message.
-                let ret =
-                    cm.send::<K>(sender_id, msg.to, msg.method_num, &msg.params, &msg.value)?;
+                let ret = cm.send::<K>(sender_id, msg.to, msg.method_num, params, &msg.value)?;
 
                 // Charge for including the result (before we end the transaction).
-                if let InvocationResult::Return(data) = &ret {
-                    cm.charge_gas(cm.context().price_list.on_chain_return_value(data.len()))?;
+                if let InvocationResult::Return(value) = &ret {
+                    cm.charge_gas(cm.context().price_list.on_chain_return_value(
+                        value.as_ref().map(|v| v.size() as usize).unwrap_or(0),
+                    ))?;
                 }
 
                 Ok(ret)
@@ -88,7 +96,13 @@ where
 
         // Extract the exit code and build the result of the message application.
         let receipt = match res {
-            Ok(InvocationResult::Return(return_data)) => {
+            Ok(InvocationResult::Return(return_value)) => {
+                // Convert back into a top-level return "value". We throw away the codec here,
+                // unfortunately.
+                let return_data = return_value
+                    .map(|blk| RawBytes::from(blk.data().to_vec()))
+                    .unwrap_or_default();
+
                 backtrace.clear();
                 Receipt {
                     exit_code: ExitCode::OK,

fvm/src/kernel/blocks.rs

@@ -1,4 +1,5 @@
 use std::convert::TryInto;
+use std::rc::Rc;
 
 use fvm_ipld_encoding::DAG_CBOR;
 use thiserror::Error;
@@ -7,7 +8,7 @@ use super::{ExecutionError, SyscallError};
 use crate::syscall_error;
 
 #[derive(Default)]
-pub(crate) struct BlockRegistry {
+pub struct BlockRegistry {
     blocks: Vec<Block>,
 }
 
@@ -25,18 +26,22 @@ pub struct BlockStat {
     pub size: u32,
 }
 
-#[derive(Clone)]
+#[derive(Debug, Clone)]
 pub struct Block {
     codec: u64,
-    data: Box<[u8]>,
+    // Unfortunately, we usually start with a vector/boxed buffer. If we used Rc<[u8]>, we'd have to
+    // copy the bytes. So we accept some indirection for reliable performance.
+    #[allow(clippy::redundant_allocation)]
+    data: Rc<Box<[u8]>>,
 }
 
 impl Block {
     pub fn new(codec: u64, data: impl Into<Box<[u8]>>) -> Self {
-        // TODO: check size on the way in?
+        // This requires an extra allocation (ew) but no extra copy on send.
+        // The extra allocation is basically nothing.
         Self {
             codec,
-            data: data.into(),
+            data: Rc::new(data.into()),
         }
     }