Idea: Introduce additional parallelism within application of a single shard chunk

[nagisa]

When executing the transactions and receipts within a chunk in apply, the flow of execution is currently very sequential and is roughly as such:

[prefetch A] → [vm_runner::run A] → [prefetch B] → [vm_runner::run B] ... → [commit]

I've already talked about the fact that we don't necessarily need to have commit be as serial as it is currently and we can start applying the next chunk before the [commit] for the previous chunk has fully concluded in #11118. If we look closer and notice that [vm_runner::run] is composed of multiple currently sequential steps, namely [vm_runner::load_and_instantiate] and [vm_runner::invoke_method], and that only the invoke_method calls need to be executed serially, we could pipeline/parallelize the processing of a single chunk much more to look more like this (each horizontal line is a separate "thread" of execution):

[prefetch A] → [vm_runner::instantiate A] → [vm_runner::invoke_method A] → [vm_runner::invoke_method B] → [vm_runner::invoke_method C] ... → [commit]
                                 [prefetch B] [vm_runner::instantiate B]-^ 
                                                                [prefetch C] [vm_runner::instantiate C]-^

Notice how the latency of prefetching – and much more importantly – loading and instantiating the contract gets hidden "behind" the time the runtime spends actually executing the contract code in the background.

This sort of pipelining can be dynamic (automatically adjusting how many contracts are preloaded in advance), or static (preloading N contracts in advance.) Either have benefits and downsides:

• Static:
  • Pro: simpler to implement
  • Pro: simpler to reason about
  • Con: with deeper pipelines we may "waste" some work when gas/compute limits prevent us from invoking methods for which we've already preloaded the contracts for;
  • Con: therefore we need to carefully select an optimal pipeline depth...
• Dynamic:
  • Pro: we don't need to fiddle with pipeline depth parameters
  • Pro: adaptive to the time contract methods consume executing, always maintaining roughly an optimal pipeline depth;
  • Pro: less wasted work
  • Con: more difficult to reason about and implement.

Ultimately I think if we work towards implementing something like this we should start with a static and configurable pipeline depth. There's a good chance that an optimal depth will turn out to be one or two calls deep, in which case throwing away that work isn't going to be particularly painful. Especially knowing that doing this work might place some of the necessary data in the caches so that the next time around these preparatory steps get executed faster and the pipeline is quicker to fill.

@bowenwang1996 had raised another idea that we could consider preparing all the contracts even before we head into the apply call (also in parallel.) This is also an option we could consider, although it remains to be seen how much more of an improvement it would be over a basic localized and static pipeline. There's also an unsolved question of how many of these "all" we would need, since the number of contracts that get executed in a single apply is dynamic and dependent on gas and compute cost limits.

cc @Ekleog-NEAR this approach is another idea to deal with the cost of allocating memory maps for the contract data. Thoughts appreciated.

[nagisa]

In an effort to implement this, the very first thing that will need to happen is a refactor of the VM runner.

There are a couple of issues to resolve:

• Split up the near_vm_runner::run method into an instantiate and run_method;
  • not quite as written, but vm: split preparation and running of a contract #11667 implements this in a way that's agnostic over where exactly the split occurs.
• Instantiation must be split up so that the start method can be run at a later time. This may be particularly troublesome for VMs that we don't control implementation of. For instance wasmtime provides "pre-instantiate" sort of method, but that does less work than we'd ideally like.
• In near_vm_runner we have the split here already, but the instantiation is mixed up together with the loading of the executable, which is probably something we should fix as well.
  • All the caches make things more difficult than they ideally should be...
  • We may need to keep the cache of loaded contracts for use-cases like ReadRPC which will not be able to predict upcoming requests, but we could possibly drop that cache in nearcore.
• Figure out what type represents an instance: do we store it within a VM (and make the VM only able to run a single method?) or do we do some type erasure shenanigans to obtain a type that makes sense to return from an object-safe trait?
• Implementing this requires significant-enough changes to the code structure to require a protocol version change. How do we maintain backward compatibility when the old runtimes cannot allow for the same sort of logic flow...?

[nagisa]

cc #11808

[bowenwang1996]

I have a question: instead of thinking about dynamic vs. static, could we just use an approach where at the beginning of apply, we eagerly go over receipts and load contracts into a queue, then we when need to execute a function call receipt, we pop prepared contracts from the front of queue. We need a concurrent queue here, but otherwise it should be easier to implement.

[nagisa]

I would argue that the described solution is a form of a static approach where N of preloaded contracts is implicitly bounded by available memory or the maximum number of delayed+incoming+local receipts. If we don't carefully manage the number of prepared receipts we'd be looking at >1 TiB of memory to store the prepared contracts (assuming ~20k of delayed receipts and 64MiB of contract data memory per prepared contract.)

Traditional queues are somewhat poor of a data structure for this problem too. For instance local receipts are constructed within apply and processed right after they are constructed before delayed or incoming receipts. But then if those local receipts do not fit into the current chunk, they get pushed to the back of the delayed receipt queue. As thus the ordering in which the receipts get executed is quite dynamic in the grand scheme of things, and is somewhat ambiguous at the beginning of apply.

For these reasons I don't think ordered data structures or starting preparation work in apply are the right way forward.

[nagisa]

The proof of concept (wherein we only pipeline contract preparation and block any processing for accounts which have had a contract deployment in the chunk) is showing some mildly positive results. We're successfully moving about 80% of the time spent on preparation to other worker threads.

I have initially thought that there's a significant amount of overhead in thread pool management introduced by rayon, but it turns out to be mostly due to sleeping when the thread in a pool goes idle.

We are still spending ~20% of the original time on the critical path doing those same preparations. This is -- to an extent -- intended currently, this is how we keep the latency minimal in case the pipeline implementation can't get to processing the action before it is time to execute it. We could improve on this

For an MVP what's remaining is to also offload deployment action processing into the pipeline implementation here. This is mostly necessary in order to establish the necessary data dependency/ordering information between deployment and function call actions. As a bonus this should also move the arguably most expensive (and unpredictable, if rare) action away from the critical path, leaving more time for other tasks. Ultimately, this is required to bring down the worst-case execution time (so that we can reduce the gas fees.) Otherwise any receipt that begins with a deployment action immediately disables any improvements seen here.

[nagisa]

Pipelines deeper than 1 action with a function call per shard give no observable benefits.

[nagisa]

I don't have a separate issue for parallelizing receipts themselves within a chunk, but one interesting though came up in my discussion with @akhi3030 that I wanted to write it down. There's a problem that we can parallelize receipts all we want, but if all of them target the same account, they would still need to be executed sequentially. So we wouldn't be able to reduce gas costs for that reason.

Akhi mentioned that we could instead have chunk producers determine a number of "threads" of execution and then we could raise the gas limit to a multiple of determined threads. This way even if all receipts were targetting the same account, we would put all of the receipts on one thread and this account would not get any more than 1Pgas per chunk anyway. But we would have N-1 threads worth of spare capacity for other accounts and their actions.

So while the throughput for e.g. large.near would not necessarily improve, we could put humongous.near on that same shard without worrying much about it taking capacity away from largecontract.near.