Update Component trait methods to allow state sharing during execution

[hdevalence]

Is your feature request related to a problem? Please describe.

Currently, the Component trait is defined as:

#[async_trait]
pub trait Component {
    async fn init_chain<S: StateWrite>(state: S, app_state: &genesis::AppState);
    async fn begin_block<S: StateWrite>(state: S, begin_block: &abci::request::BeginBlock);
    async fn end_block<S: StateWrite>(state: S, end_block: &abci::request::EndBlock);
    async fn end_epoch<S: StateWrite>(state: S) -> Result<()>;
}

This gives each component an owned StateWrite, usually instantiated as a &mut reference to a StateDelta with pending changes.

However, it doesn't allow a component the ability to share its state with subtasks, because the state isn't clonable, and because a borrowed state might not be 'static. This is a problem for the DEX component in particular, which would like to explore possible routes in parallel.

Describe the solution you'd like

We want to give each component the ability to share concurrent read access with subtasks, while retaining the ability to get exclusive access for mutation. One way to achieve this is with Arc::get_mut, which returns Some(&mut T) if the refcount is 1 and exclusive access is possible.

This probably looks like changing the trait methods to

#[async_trait]
pub trait Component {
    async fn init_chain<S: StateWrite>(state: S, app_state: &genesis::AppState);
    async fn begin_block<S: StateWrite + 'static>(state: &mut Arc<S>, begin_block: &abci::request::BeginBlock);
    async fn end_block<S: StateWrite + 'static>(state: &mut Arc<S>, end_block: &abci::request::EndBlock);
    async fn end_epoch<S: StateWrite + 'static>(state: &mut Arc<S>) -> Result<()>;
}

The signature is a little weird, but I think it's what we want:

• In order to allow implementors to call get_mut(), we must have state.strong_count() == 1 and state.weak_count() == 0 when called, i.e., the Arc must not actually be shared.
• This means we must pass either ownership of the Arc<S> or unique reference &mut Arc<S> into the component.
• But no sharing means that the caller can't retain a copy, so passing ownership would prevent the caller from calling more than one component's methods. So it must be a &mut Arc<S>.

Having the state be a &mut Arc<S> means that the "root Arc" will be owned by the caller, so the generated futures will have a lifetime bound and won't be 'static. This is actually not so dissimilar from the situation we have now, where we end up desugaring the trait method to

-> Pin<Box<dyn Future<Output = ()> + Send + 'async_trait>>
    where S: 'async_trait + StateWrite,

but now, instead of requiring that S outlives 'async_trait (the lifetime of the generated future), we'll require that S: 'static, and have the lifetime bound on the &mut Arc<S>. This gives us the escape hatch of cloning the borrowed &mut Arc<S> to get an owned Arc<S>, which will then be 'static (since S: 'static), as long as we promise to restore uniqueness by dropping any extra references before we complete the component's processing.

We probably want to add some documentation about invariants, like

    /// # Invariants
    /// 
    /// The `&mut Arc<S>` allows the implementor to optionally share state with
    /// its subtasks.  The implementor SHOULD assume that when the method is
    /// called, `state.get_mut().is_some()`, i.e., the `Arc` is not shared.  The
    /// implementor MUST ensure that any clones of the `Arc` are dropped before
    /// it returns, so that `state.get_mut().is_some()` on completion.

In every existing component, where we don't care about concurrent read access for subtasks, we can do something like

let state = state.get_mut().unwrap();

at the top of the impl, to avoid having to change any of the existing code.

Describe alternatives you've considered

I don't think we need to change the init_chain methods, but if we're going to do a change to every Component implementation, we might as well roll over begin_block and end_epoch at the same time for future-proofing.

Additional context

Let's do this as an isolated change before #2398.