runtime: temporarily workaround the LimitedMemoryPool limiting the co…

…ncurrency of the contract runtime too much (#10733)

See individual commits and their descriptions for an easier review
experience.

This PR is intended to be easily backportable with a minimal risk, with
perhaps only the 2nd commit being superfluous in that context. There are
definitely things that we can do to make the reliability and sensibility
of this area much better, but they would result in a much more invasive
change. My belief is that we should land both this workaround *and* fix
#10732 as a follow-up.

cc #10732
cc @posvyatokum we will want this backported to 1.37.1, let me know if
you'd like me to prepare a backport and/or where you want it to end up
at.

---

runtime: hotfix to allow up-to 256 concurrent runtime invocations

This is not a proper fix as certain configuration settings can still
make this number to be exceeded, but the changes there are a little more
involved than what I'd be comfortable with making in a rush, especially
for a backport to the 1.37 release.

In order to keep the initial memory use reasonably bounded, I have
chosen to reduce the initial size of each map to a smaller number.
Since the runtime reuses most recently returned memories first, in
use-cases where there are few concurrent accesses, only that many
memories will get resized to fit all the required contracts over time,
thus keeping the memory use reasonably limited. This will result in
slightly less predictable function call performance for the first
invocation of the very few first contract invocations that
require more memory, but this seems like a required tradeoff to avoid
preallocating dozens of GB of mappings that will later go mostly unused.

---

runtime: avoid losing memories when failing to resize them

This is not something that I am aware of happening but a fresh look at
the code immediately spotted it as icky part of the code. The new
ordering will still fail to resize and most likely lead to the
termination of the node, but if the caller chooses to do something else
about the error, this new setup will correctly handle returning the
previous memory to the memory pool.

---


runtime: switch LimitedMemoryPool to a LIFO queue

crossbeam's implementation is a FIFO queue, which will repeatedly touch
all of the memories, regardless of the concurrency which then can result
in more memory use than strictly necessary in low-concurrency scenarios.

Cargo.toml

@@ -149,7 +149,6 @@ cpu-time = "1.0"
 criterion = { version = "0.5.1", default_features = false, features = ["html_reports", "cargo_bench_support"] }
 crossbeam = "0.8"
 crossbeam-channel = "0.5.8"
-crossbeam-queue = "0.3.8"
 csv = "1.2.1"
 curve25519-dalek = { version = "4.1.1", default-features = false, features = ["alloc", "precomputed-tables", "rand_core"] }
 derive-enum-from-into = "0.1.1"

runtime/near-vm-runner/src/near_vm_runner.rs

@@ -258,7 +258,7 @@ impl NearVM {
                 // that particular function call will be slower. Not to mention there isn't a
                 // strong guarantee on the upper bound of the memory that the contract runtime may
                 // require.
-                LimitedMemoryPool::new(8, 64 * 1024 * 1024).unwrap_or_else(|e| {
+                LimitedMemoryPool::new(256, 1 * 1024 * 1024).unwrap_or_else(|e| {
                     panic!("could not pre-allocate resources for the runtime: {e}");
                 })
             })
@@ -719,7 +719,12 @@ impl crate::runner::VM for NearVM {
     }
 }
 
-#[test]
-fn test_memory_like() {
-    crate::logic::test_utils::test_memory_like(|| Box::new(NearVmMemory::new(1, 1).unwrap()));
+#[cfg(test)]
+mod tests {
+    #[test]
+    fn test_memory_like() {
+        crate::logic::test_utils::test_memory_like(|| {
+            Box::new(super::NearVmMemory::new(1, 1).unwrap())
+        });
+    }
 }

runtime/near-vm-runner/src/tests/regression_tests.rs

@@ -80,3 +80,39 @@ fn gas_intrinsic_did_not_multiply_by_opcode_cost() {
             "#]],
         ]);
 }
+
+#[test]
+fn parallel_runtime_invocations() {
+    let mut join_handles = Vec::new();
+    for _ in 0..128 {
+        let handle = std::thread::spawn(|| {
+            for _ in 0..16 {
+                test_builder()
+                .only_near_vm()
+                .wat(
+                    r#"
+                      (module
+                        (type (func))
+                        (func (type 0)
+                          loop
+                            br 0
+                          end
+                        )
+                        (export "foo" (func 0)))
+                    "#,
+                )
+                .method("foo")
+                .expects(&[
+                    expect![[r#"
+                    VMOutcome: balance 4 storage_usage 12 return data None burnt gas 100000000000000 used gas 100000000000000
+                    Err: Exceeded the prepaid gas.
+                    "#]],
+                ]);
+            }
+        });
+        join_handles.push(handle);
+    }
+    for handle in join_handles {
+        handle.join().unwrap();
+    }
+}

runtime/near-vm/engine/Cargo.toml

@@ -30,7 +30,6 @@ target-lexicon.workspace = true
 thiserror.workspace = true
 cfg-if.workspace = true
 tracing.workspace = true
-crossbeam-queue.workspace = true
 rustix = { workspace = true, features = ["param", "mm"] }
 
 [badges]

runtime/near-vm/engine/src/universal/code_memory.rs

@@ -91,7 +91,7 @@ impl<'a> CodeMemoryWriter<'a> {
 /// Mappings to regions of memory storing the executable JIT code.
 pub struct CodeMemory {
     /// Where to return this memory to when dropped.
-    source_pool: Option<Arc<crossbeam_queue::ArrayQueue<Self>>>,
+    source_pool: Option<Arc<std::sync::Mutex<Vec<Self>>>>,
 
     /// The mapping
     map: *mut u8,
@@ -138,16 +138,10 @@ impl CodeMemory {
         if self.size < size {
             // Ideally we would use mremap, but see
             // https://bugzilla.kernel.org/show_bug.cgi?id=8691
-            let source_pool = unsafe {
-                mm::munmap(self.map.cast(), self.size)?;
-                let source_pool = self.source_pool.take();
-                std::mem::forget(self);
-                source_pool
-            };
-            Self::create(size).map(|mut m| {
-                m.source_pool = source_pool;
-                m
-            })
+            let mut result = Self::create(size)?;
+            result.source_pool = self.source_pool.take();
+            drop(self);
+            Ok(result)
         } else {
             self.executable_end = 0;
             Ok(self)
@@ -218,12 +212,13 @@ impl Drop for CodeMemory {
                     );
                 }
             }
-            drop(source_pool.push(Self {
+            let mut guard = source_pool.lock().expect("unreachable due to panic=abort");
+            guard.push(Self {
                 source_pool: None,
                 map: self.map,
                 size: self.size,
                 executable_end: 0,
-            }));
+            });
         } else {
             unsafe {
                 if let Err(e) = mm::munmap(self.map.cast(), self.size) {
@@ -248,25 +243,24 @@ unsafe impl Send for CodeMemory {}
 /// However it is possible for the mappings inside to grow to accomodate larger code.
 #[derive(Clone)]
 pub struct LimitedMemoryPool {
-    pool: Arc<crossbeam_queue::ArrayQueue<CodeMemory>>,
+    pool: Arc<std::sync::Mutex<Vec<CodeMemory>>>,
 }
 
 impl LimitedMemoryPool {
     /// Create a new pool with `count` mappings initialized to `default_memory_size` each.
     pub fn new(count: usize, default_memory_size: usize) -> rustix::io::Result<Self> {
-        let pool = Arc::new(crossbeam_queue::ArrayQueue::new(count));
-        let this = Self { pool };
+        let mut pool = Vec::with_capacity(count);
         for _ in 0..count {
-            this.pool
-                .push(CodeMemory::create(default_memory_size)?)
-                .unwrap_or_else(|_| panic!("ArrayQueue could not accomodate {count} memories!"));
+            pool.push(CodeMemory::create(default_memory_size)?);
         }
-        Ok(this)
+        let pool = Arc::new(std::sync::Mutex::new(pool));
+        Ok(Self { pool })
     }
 
     /// Get a memory mapping, at least `size` bytes large.
     pub fn get(&self, size: usize) -> rustix::io::Result<CodeMemory> {
-        let mut memory = self.pool.pop().ok_or(rustix::io::Errno::NOMEM)?;
+        let mut guard = self.pool.lock().expect("unreachable due to panic=abort");
+        let mut memory = guard.pop().ok_or(rustix::io::Errno::NOMEM)?;
         memory.source_pool = Some(Arc::clone(&self.pool));
         if memory.size < size {
             Ok(memory.resize(size)?)