From 80c5eadd618046dc33d4bd4b9ddd4026299e6e6d Mon Sep 17 00:00:00 2001 From: DaniPopes <57450786+DaniPopes@users.noreply.github.com> Date: Fri, 22 May 2026 14:19:17 +0200 Subject: [PATCH 1/2] feat: add async database fiber support --- Cargo.lock | 13 + Cargo.toml | 6 + crates/context/Cargo.toml | 1 + crates/context/src/evm.rs | 11 + crates/database/interface/Cargo.toml | 3 +- crates/database/interface/src/async_db.rs | 1136 +++++++++++++++-- crates/database/interface/src/lib.rs | 2 +- crates/handler/Cargo.toml | 1 + crates/handler/src/api.rs | 62 + crates/handler/src/lib.rs | 4 + crates/handler/src/mainnet_builder.rs | 4 + crates/handler/src/precompile_provider.rs | 2 + crates/handler/src/system_call.rs | 100 ++ crates/revm/Cargo.toml | 7 +- crates/revm/src/lib.rs | 4 + examples/custom_precompile_journal/Cargo.toml | 3 + .../src/custom_evm.rs | 2 + examples/my_evm/Cargo.toml | 3 + examples/my_evm/src/evm.rs | 2 + 19 files changed, 1275 insertions(+), 91 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index b3b92937ce..4f257355e6 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1482,6 +1482,18 @@ version = "0.8.7" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "773648b94d0e5d620f64f280777445740e61fe701025087ec8b57f45c791888b" +[[package]] +name = "corosensei" +version = "0.3.3" +source = "git+https://github.com/Amanieu/corosensei?rev=3a122508ab1d7c25e4dee8b2b4b4c47046c3c3d8#3a122508ab1d7c25e4dee8b2b4b4c47046c3c3d8" +dependencies = [ + "autocfg", + "cfg-if", + "libc", + "scopeguard", + "windows-sys 0.59.0", +] + [[package]] name = "cpufeatures" version = "0.2.17" @@ -3789,6 +3801,7 @@ name = "revm-database-interface" version = "12.1.0" dependencies = [ "auto_impl", + "corosensei", "either", "revm-primitives", "revm-state", diff --git a/Cargo.toml b/Cargo.toml index 50877eb8aa..6c11a8d687 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -112,6 +112,12 @@ serde_json = { version = "1.0", default-features = false } auto_impl = "1.3.0" bitflags = { version = "2.10", default-features = false } cfg-if = { version = "1.0", default-features = false } +# Uses public `Coroutine::with_stack_unchecked`. +# https://github.com/Amanieu/corosensei/pull/74 +corosensei = { version = "0.3.3", git = "https://github.com/Amanieu/corosensei", rev = "3a122508ab1d7c25e4dee8b2b4b4c47046c3c3d8", default-features = false, features = [ + "default-stack", + "unwind", +] } derive-where = { version = "1.6", default-features = false } rand = "0.9" tokio = "1.47" diff --git a/crates/context/Cargo.toml b/crates/context/Cargo.toml index fbb01e828e..1dd52180a8 100644 --- a/crates/context/Cargo.toml +++ b/crates/context/Cargo.toml @@ -70,6 +70,7 @@ dev = [ "optional_priority_fee_check", "optional_fee_charge", ] +asyncdb = ["std", "database-interface/asyncdb"] memory_limit = [] optional_balance_check = [] optional_block_gas_limit = [] diff --git a/crates/context/src/evm.rs b/crates/context/src/evm.rs index 283fa917b2..f597c3aa33 100644 --- a/crates/context/src/evm.rs +++ b/crates/context/src/evm.rs @@ -22,6 +22,9 @@ pub struct Evm { pub precompiles: P, /// Frame that is going to be executed. pub frame_stack: FrameStack, + /// Reusable async execution stack. + #[cfg(feature = "asyncdb")] + pub async_stack: database_interface::async_db::FiberStack, } impl Evm { @@ -35,6 +38,8 @@ impl Evm { instruction, precompiles, frame_stack: FrameStack::new_prealloc(8), + #[cfg(feature = "asyncdb")] + async_stack: database_interface::async_db::FiberStack::default(), } } } @@ -48,6 +53,8 @@ impl Evm { instruction, precompiles, frame_stack: FrameStack::new_prealloc(8), + #[cfg(feature = "asyncdb")] + async_stack: database_interface::async_db::FiberStack::default(), } } } @@ -62,6 +69,8 @@ impl Evm { instruction: self.instruction, precompiles: self.precompiles, frame_stack: self.frame_stack, + #[cfg(feature = "asyncdb")] + async_stack: self.async_stack, } } @@ -73,6 +82,8 @@ impl Evm { instruction: self.instruction, precompiles, frame_stack: self.frame_stack, + #[cfg(feature = "asyncdb")] + async_stack: self.async_stack, } } diff --git a/crates/database/interface/Cargo.toml b/crates/database/interface/Cargo.toml index bad9d28f1e..639de4b590 100644 --- a/crates/database/interface/Cargo.toml +++ b/crates/database/interface/Cargo.toml @@ -32,6 +32,7 @@ serde = { workspace = true, features = ["derive", "rc"], optional = true } # asyncdb tokio = { workspace = true, optional = true } +corosensei = { workspace = true, optional = true } [dev-dependencies] @@ -45,4 +46,4 @@ std = [ "thiserror/std", ] serde = ["dep:serde", "primitives/serde", "state/serde", "either/serde"] -asyncdb = ["dep:tokio", "tokio/rt-multi-thread"] +asyncdb = ["std", "dep:tokio", "tokio/rt-multi-thread", "dep:corosensei"] diff --git a/crates/database/interface/src/async_db.rs b/crates/database/interface/src/async_db.rs index 33f6b84fc8..27e0615c81 100644 --- a/crates/database/interface/src/async_db.rs +++ b/crates/database/interface/src/async_db.rs @@ -1,17 +1,474 @@ //! Async database interface. -use crate::{DBErrorMarker, Database, DatabaseRef}; -use core::future::Future; -use primitives::{Address, StorageKey, StorageValue, B256}; -use state::{AccountId, AccountInfo, Bytecode}; -use tokio::runtime::{Handle, Runtime}; +use crate::{DBErrorMarker, Database, DatabaseCommit, DatabaseRef}; +use core::{ + convert::Infallible, + future::Future, + marker::PhantomData, + pin::Pin, + ptr::NonNull, + task::{Context, Poll}, +}; +use corosensei::{stack::DefaultStack, Coroutine, CoroutineResult, Yielder}; +use primitives::{Address, AddressMap, StorageKey, StorageValue, B256}; +use state::{Account, AccountId, AccountInfo, Bytecode}; +use std::{cell::Cell, fmt, io}; +use tokio::{ + runtime::{Handle, Runtime}, + task, +}; -/// The async EVM database interface +type Resume = AsyncResult>>; +type Yield = (); +type Complete = AsyncResult; +type DatabaseFiber = Coroutine, DefaultStack>; + +const DEFAULT_STACK_SIZE: usize = 1024 * 1024; + +/// Reusable async EVM fiber stack storage. +#[derive(Default)] +pub struct FiberStack { + stack: Option, +} + +impl Clone for FiberStack { + #[inline] + fn clone(&self) -> Self { + Self::default() + } +} + +impl fmt::Debug for FiberStack { + #[inline] + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + f.debug_struct("FiberStack").finish_non_exhaustive() + } +} + +impl FiberStack { + #[inline] + fn take_or_new(&mut self) -> AsyncResult { + match self.stack.take() { + Some(stack) => Ok(stack), + None => DefaultStack::new(DEFAULT_STACK_SIZE).map_err(AsyncError::Io), + } + } + + #[inline] + fn put(&mut self, stack: DefaultStack) { + debug_assert!(self.stack.is_none()); + self.stack = Some(stack); + } +} + +thread_local! { + static CURRENT: Cell>> = const { Cell::new(None) }; +} + +/// Result type used by async database execution helpers. +pub type AsyncResult = Result>; + +/// Error returned by async database execution helpers. +#[derive(Debug, thiserror::Error)] +#[non_exhaustive] +pub enum AsyncError { + /// The async EVM fiber was cancelled before execution completed. + #[error("async EVM execution was cancelled")] + Cancelled, + /// An async host operation was called outside an async EVM fiber. + #[error("async host operation requires EVM async fiber execution")] + NotOnFiber, + /// Blocking async I/O was requested outside a supported Tokio runtime. + #[error("async host operation requires a Tokio multi-thread runtime")] + Runtime, + /// Async fiber stack setup failed. + #[error(transparent)] + Io(io::Error), + /// The wrapped operation returned an error. + #[error(transparent)] + Inner(#[from] E), +} + +impl AsyncError { + fn with_inner_error(self) -> AsyncError { + match self { + Self::Cancelled => AsyncError::Cancelled, + Self::NotOnFiber => AsyncError::NotOnFiber, + Self::Runtime => AsyncError::Runtime, + Self::Io(error) => AsyncError::Io(error), + Self::Inner(error) => match error {}, + } + } +} + +impl DBErrorMarker for AsyncError { + #[inline] + fn is_fatal(&self) -> bool { + match self { + Self::Inner(error) => error.is_fatal(), + _ => true, + } + } +} + +struct CurrentFiber { + suspend: NonNull>, + future_cx: NonNull>, + cancelled: bool, +} + +impl CurrentFiber { + #[inline] + fn context(&mut self) -> &mut Context<'_> { + unsafe { restore_context_lifetime(self.future_cx.as_mut()) } + } + + #[inline] + fn suspend(&mut self) -> AsyncResult<()> { + match unsafe { self.suspend.as_ref() }.suspend(()) { + Ok(cx) => { + self.future_cx = cx; + Ok(()) + } + Err(error) => { + self.cancelled = true; + Err(error) + } + } + } + + #[inline] + const fn is_cancelled(&self) -> bool { + self.cancelled + } +} + +struct ResetCurrentFiber(Option>); + +impl Drop for ResetCurrentFiber { + fn drop(&mut self) { + CURRENT.set(self.0); + } +} + +/// Runs `func` on a native fiber and awaits its completion. +/// +/// Synchronous code running inside `func` may call [`block_on_current`] to wait for async host +/// operations without blocking the executor thread. +#[cfg(test)] +pub(crate) fn on_fiber_result<'a, R, E>( + func: impl FnOnce() -> Result + 'a, +) -> impl Future> + Send + 'a +where + R: Send + 'a, + E: Send + 'a, +{ + OnFiber::new(func) +} + +/// Runs `func` on a native fiber backed by a reusable EVM stack slot. +/// +/// # Safety +/// +/// `stack` must point to valid stack storage for the lifetime of the returned future. That storage +/// must not be accessed by anything else until the returned future is dropped. +pub unsafe fn on_fiber_result_with_stack<'a, R, E>( + stack: NonNull, + func: impl FnOnce() -> Result + 'a, +) -> impl Future> + Send + 'a +where + R: Send + 'a, + E: Send + 'a, +{ + OnFiber::with_stack(stack, func) +} + +#[cfg(test)] +pub(crate) fn on_fiber<'a, R>( + func: impl FnOnce() -> R + 'a, +) -> impl Future> + Send + 'a +where + R: Send + 'a, +{ + on_fiber_result(move || Ok::<_, Infallible>(func())) +} + +/// Runs `func` on a native fiber backed by a reusable EVM stack slot. +/// +/// # Safety +/// +/// See [`on_fiber_result_with_stack`]. +pub unsafe fn on_fiber_with_stack<'a, R>( + stack: NonNull, + func: impl FnOnce() -> R + 'a, +) -> impl Future> + Send + 'a +where + R: Send + 'a, +{ + unsafe { on_fiber_result_with_stack(stack, move || Ok::<_, Infallible>(func())) } +} + +enum OnFiber<'a, R, E> { + Running(FiberFuture<'a, Result>), + Error(Option), + Done, +} + +impl<'a, R, E> OnFiber<'a, R, E> { + #[cfg(test)] + fn new(func: impl FnOnce() -> Result + 'a) -> Self { + Self::new_inner(None, func) + } + + fn with_stack(stack: NonNull, func: impl FnOnce() -> Result + 'a) -> Self { + Self::new_inner(Some(stack), func) + } + + fn new_inner( + stack: Option>, + func: impl FnOnce() -> Result + 'a, + ) -> Self { + match FiberFuture::new(stack, func) { + Ok(fiber) => Self::Running(fiber), + Err(error) => Self::Error(Some(error)), + } + } +} + +impl Future for OnFiber<'_, R, E> { + type Output = AsyncResult; + + fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll { + let this = self.get_mut(); + match this { + Self::Running(fiber) => match Pin::new(fiber).poll(cx) { + Poll::Ready(Ok(Ok(value))) => { + *this = Self::Done; + Poll::Ready(Ok(value)) + } + Poll::Ready(Ok(Err(error))) => { + *this = Self::Done; + Poll::Ready(Err(AsyncError::Inner(error))) + } + Poll::Ready(Err(error)) => { + *this = Self::Done; + Poll::Ready(Err(error.with_inner_error())) + } + Poll::Pending => Poll::Pending, + }, + Self::Error(error) => { + let error = error + .take() + .expect("async EVM fiber error already returned"); + Poll::Ready(Err(error.with_inner_error())) + } + Self::Done => panic!("async EVM fiber polled after completion"), + } + } +} + +struct FiberFuture<'a, R> { + fiber: Option>, + stack: Option>, + _marker: PhantomData<&'a ()>, +} + +// SAFETY: The future may move between polls, but the coroutine stack itself is heap allocated and +// is only resumed through `poll` with a fresh task context. Values that can remain on the coroutine +// stack across suspension are required to be `Send` by the blocking boundary. +unsafe impl Send for FiberFuture<'_, R> {} + +impl<'a, R> FiberFuture<'a, R> { + fn new( + mut stack: Option>, + func: impl FnOnce() -> R + 'a, + ) -> AsyncResult { + let fiber_stack = match &mut stack { + Some(stack) => unsafe { stack.as_mut() }.take_or_new()?, + None => DefaultStack::new(DEFAULT_STACK_SIZE).map_err(AsyncError::Io)?, + }; + let body = move |suspend: &Yielder, resume| { + let future_cx = resume?; + let mut current = CurrentFiber { + suspend: NonNull::from(suspend), + future_cx, + cancelled: false, + }; + let current = NonNull::from(&mut current); + let previous = CURRENT.replace(Some(current)); + let _reset = ResetCurrentFiber(previous); + Ok(func()) + }; + // SAFETY: The coroutine is stored inside `FiberFuture<'a, R>`, which is tied to the + // borrowed state lifetime and dropped before those borrows can expire. + let fiber = unsafe { Coroutine::with_stack_unchecked(fiber_stack, body) }; + Ok(Self { + fiber: Some(fiber), + stack, + _marker: PhantomData, + }) + } + + fn recycle_stack(&mut self) { + let Some(fiber) = self.fiber.take() else { + return; + }; + debug_assert!(fiber.done()); + let stack = fiber.into_stack(); + if let Some(mut slot) = self.stack { + unsafe { slot.as_mut() }.put(stack); + } + } +} + +impl Future for FiberFuture<'_, R> { + type Output = AsyncResult; + + fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll { + let this = self.get_mut(); + let cx = NonNull::from(unsafe { change_context_lifetime(cx) }); + let fiber = this + .fiber + .as_mut() + .expect("async EVM fiber polled after completion"); + match fiber.resume(Ok(cx)) { + CoroutineResult::Return(result) => { + this.recycle_stack(); + Poll::Ready(result) + } + CoroutineResult::Yield(()) => Poll::Pending, + } + } +} + +impl Drop for FiberFuture<'_, R> { + fn drop(&mut self) { + let Some(fiber) = self.fiber.as_mut() else { + return; + }; + if fiber.done() { + self.recycle_stack(); + } else if matches!( + fiber.resume(Err(AsyncError::Cancelled)), + CoroutineResult::Yield(()) + ) { + // SAFETY: Cancellation already gave the coroutine a chance to return normally. If it + // yields again, the stack is no longer useful to this future. + unsafe { fiber.force_reset() }; + } else { + self.recycle_stack(); + } + } +} + +/// Polls `future` to completion from inside an async EVM fiber. +/// +/// If `future` returns `Poll::Pending`, the current EVM fiber is suspended and the outer async EVM +/// future returns `Poll::Pending`. When the executor wakes and polls the outer future again, the +/// EVM fiber resumes and continues polling `future`. +/// +/// # Errors +/// +/// Returns [`AsyncError::NotOnFiber`] if called outside async EVM execution, or +/// [`AsyncError::Cancelled`] if the outer async EVM execution was dropped. +pub fn block_on_current(future: F) -> AsyncResult { + let mut future = core::pin::pin!(future); + loop { + match with_current(|current| { + if current.is_cancelled() { + return Err(AsyncError::Cancelled); + } + let poll = future.as_mut().poll(current.context()); + if poll.is_pending() { + current.suspend()?; + } + Ok(poll) + })? { + Poll::Ready(value) => return Ok(value), + Poll::Pending => {} + } + } +} + +fn current_tokio_handle() -> Option { + match Handle::try_current() { + Ok(handle) => match handle.runtime_flavor() { + tokio::runtime::RuntimeFlavor::CurrentThread => None, + _ => Some(handle), + }, + Err(_) => None, + } +} + +fn block_on_handle(handle: &Handle, future: F) -> F::Output +where + F: Future + Send, + F::Output: Send, +{ + let should_use_block_in_place = Handle::try_current() + .ok() + .map(|current| { + !matches!( + current.runtime_flavor(), + tokio::runtime::RuntimeFlavor::CurrentThread + ) + }) + .unwrap_or(false); + + if should_use_block_in_place { + task::block_in_place(move || handle.block_on(future)) + } else { + handle.block_on(future) + } +} + +fn block_on_runtime(runtime: Option<&Handle>, future: F) -> AsyncResult +where + F: Future + Send, + F::Output: Send, +{ + if CURRENT.get().is_some() { + return block_on_current(future); + } + + if let Some(runtime) = runtime { + return Ok(block_on_handle(runtime, future)); + } + + Err(AsyncError::Runtime) +} + +fn block_on_runtime_result(runtime: Option<&Handle>, future: F) -> AsyncResult +where + F: Future> + Send, + T: Send, + E: Send, +{ + match block_on_runtime(runtime, future).map_err(AsyncError::with_inner_error)? { + Ok(value) => Ok(value), + Err(error) => Err(AsyncError::Inner(error)), + } +} + +fn with_current(f: impl FnOnce(&mut CurrentFiber) -> AsyncResult) -> AsyncResult { + let mut current = CURRENT.get().ok_or(AsyncError::NotOnFiber)?; + f(unsafe { current.as_mut() }) +} + +unsafe fn change_context_lifetime<'a>(cx: &'a mut Context<'_>) -> &'a mut Context<'static> { + unsafe { core::mem::transmute::<&'a mut Context<'_>, &'a mut Context<'static>>(cx) } +} + +unsafe fn restore_context_lifetime<'a>(cx: &'a mut Context<'static>) -> &'a mut Context<'a> { + unsafe { core::mem::transmute::<&'a mut Context<'static>, &'a mut Context<'a>>(cx) } +} + +/// The async EVM database interface. /// /// Contains the same methods as [Database], but it returns [Future] type instead. /// -/// Use [WrapDatabaseAsync] to provide [Database] implementation for a type that only implements this trait. +/// Use [AsyncDb] to provide [Database] implementation for a type that only implements this trait. pub trait DatabaseAsync { - /// The database error type + /// The database error type. type Error: DBErrorMarker; /// Gets basic account information. @@ -54,13 +511,13 @@ pub trait DatabaseAsync { ) -> impl Future> + Send; } -/// The async EVM database interface +/// The async EVM database interface. /// /// Contains the same methods as [DatabaseRef], but it returns [Future] type instead. /// -/// Use [WrapDatabaseAsync] to provide [DatabaseRef] implementation for a type that only implements this trait. +/// Use [AsyncDb] to provide [DatabaseRef] implementation for a type that only implements this trait. pub trait DatabaseAsyncRef { - /// The database error type + /// The database error type. type Error: DBErrorMarker; /// Gets basic account information. @@ -103,26 +560,208 @@ pub trait DatabaseAsyncRef { ) -> impl Future> + Send; } -/// Wraps a [DatabaseAsync] or [DatabaseAsyncRef] to provide a [`Database`] implementation. +/// Adapter that exposes an async database through the synchronous [`Database`] interface. #[derive(Debug)] -pub struct WrapDatabaseAsync { +pub struct AsyncDb { db: T, - rt: HandleOrRuntime, + rt: Option, +} + +impl AsyncDb { + /// Creates a new async database adapter. + /// + /// This captures the current Tokio runtime handle when one is available. + #[inline] + pub fn new(db: T) -> Self { + Self { + db, + rt: current_tokio_handle().map(HandleOrRuntime::Handle), + } + } + + /// Creates a new async database adapter using the current Tokio runtime handle. + /// + /// Returns `None` if no Tokio runtime is available or the current runtime is current-threaded. + #[inline] + pub fn blocking(db: T) -> Option { + Some(Self { + db, + rt: Some(HandleOrRuntime::Handle(current_tokio_handle()?)), + }) + } + + /// Creates a new async database adapter with a Tokio runtime. + #[inline] + pub const fn with_runtime(db: T, runtime: Runtime) -> Self { + Self { + db, + rt: Some(HandleOrRuntime::Runtime(runtime)), + } + } + + /// Creates a new async database adapter with a Tokio runtime handle. + #[inline] + pub const fn with_handle(db: T, handle: Handle) -> Self { + Self { + db, + rt: Some(HandleOrRuntime::Handle(handle)), + } + } + + /// Returns the wrapped database. + #[inline] + pub const fn inner(&self) -> &T { + &self.db + } + + /// Returns the wrapped database mutably. + #[inline] + pub const fn inner_mut(&mut self) -> &mut T { + &mut self.db + } + + /// Consumes the adapter and returns the wrapped database. + #[inline] + pub fn into_inner(self) -> T { + self.db + } } +impl Database for AsyncDb { + type Error = AsyncError; + + #[inline] + fn basic(&mut self, address: Address) -> Result, Self::Error> { + let Self { db, rt } = self; + block_on_runtime_result( + rt.as_ref().map(HandleOrRuntime::handle), + db.basic_async(address), + ) + } + + #[inline] + fn code_by_hash(&mut self, code_hash: B256) -> Result { + let Self { db, rt } = self; + block_on_runtime_result( + rt.as_ref().map(HandleOrRuntime::handle), + db.code_by_hash_async(code_hash), + ) + } + + #[inline] + fn storage( + &mut self, + address: Address, + index: StorageKey, + ) -> Result { + let Self { db, rt } = self; + block_on_runtime_result( + rt.as_ref().map(HandleOrRuntime::handle), + db.storage_async(address, index), + ) + } + + #[inline] + fn storage_by_account_id( + &mut self, + address: Address, + account_id: AccountId, + storage_key: StorageKey, + ) -> Result { + let Self { db, rt } = self; + block_on_runtime_result( + rt.as_ref().map(HandleOrRuntime::handle), + db.storage_by_account_id_async(address, account_id, storage_key), + ) + } + + #[inline] + fn block_hash(&mut self, number: u64) -> Result { + let Self { db, rt } = self; + block_on_runtime_result( + rt.as_ref().map(HandleOrRuntime::handle), + db.block_hash_async(number), + ) + } +} + +impl DatabaseRef for AsyncDb { + type Error = AsyncError; + + #[inline] + fn basic_ref(&self, address: Address) -> Result, Self::Error> { + block_on_runtime_result( + self.rt.as_ref().map(HandleOrRuntime::handle), + self.db.basic_async_ref(address), + ) + } + + #[inline] + fn code_by_hash_ref(&self, code_hash: B256) -> Result { + block_on_runtime_result( + self.rt.as_ref().map(HandleOrRuntime::handle), + self.db.code_by_hash_async_ref(code_hash), + ) + } + + #[inline] + fn storage_ref( + &self, + address: Address, + index: StorageKey, + ) -> Result { + block_on_runtime_result( + self.rt.as_ref().map(HandleOrRuntime::handle), + self.db.storage_async_ref(address, index), + ) + } + + #[inline] + fn storage_by_account_id_ref( + &self, + address: Address, + account_id: AccountId, + storage_key: StorageKey, + ) -> Result { + block_on_runtime_result( + self.rt.as_ref().map(HandleOrRuntime::handle), + self.db + .storage_by_account_id_async_ref(address, account_id, storage_key), + ) + } + + #[inline] + fn block_hash_ref(&self, number: u64) -> Result { + block_on_runtime_result( + self.rt.as_ref().map(HandleOrRuntime::handle), + self.db.block_hash_async_ref(number), + ) + } +} + +impl DatabaseCommit for AsyncDb { + #[inline] + fn commit(&mut self, changes: AddressMap) { + self.db.commit(changes); + } + + #[inline] + fn commit_iter(&mut self, changes: &mut dyn Iterator) { + self.db.commit_iter(changes); + } +} + +/// Wraps a [DatabaseAsync] or [DatabaseAsyncRef] to provide a [`Database`] implementation. +#[derive(Debug)] +pub struct WrapDatabaseAsync(AsyncDb); + impl WrapDatabaseAsync { /// Wraps a [DatabaseAsync] or [DatabaseAsyncRef] instance. /// /// Returns `None` if no tokio runtime is available or if the current runtime is a current-thread runtime. + #[inline] pub fn new(db: T) -> Option { - let rt = match Handle::try_current() { - Ok(handle) => match handle.runtime_flavor() { - tokio::runtime::RuntimeFlavor::CurrentThread => return None, - _ => HandleOrRuntime::Handle(handle), - }, - Err(_) => return None, - }; - Some(Self { db, rt }) + AsyncDb::blocking(db).map(Self) } /// Wraps a [DatabaseAsync] or [DatabaseAsyncRef] instance, with a runtime. @@ -130,9 +769,9 @@ impl WrapDatabaseAsync { /// Refer to [tokio::runtime::Builder] on how to create a runtime if you are in synchronous world. /// /// If you are already using something like [tokio::main], call [`WrapDatabaseAsync::new`] instead. + #[inline] pub const fn with_runtime(db: T, runtime: Runtime) -> Self { - let rt = HandleOrRuntime::Runtime(runtime); - Self { db, rt } + Self(AsyncDb::with_runtime(db, runtime)) } /// Wraps a [DatabaseAsync] or [DatabaseAsyncRef] instance, with a runtime handle. @@ -141,23 +780,41 @@ impl WrapDatabaseAsync { /// to obtain a handle. /// /// If you are already in asynchronous world, like [tokio::main], use [`WrapDatabaseAsync::new`] instead. + #[inline] pub const fn with_handle(db: T, handle: Handle) -> Self { - let rt = HandleOrRuntime::Handle(handle); - Self { db, rt } + Self(AsyncDb::with_handle(db, handle)) + } + + /// Returns the wrapped database. + #[inline] + pub const fn inner(&self) -> &T { + self.0.inner() + } + + /// Returns the wrapped database mutably. + #[inline] + pub const fn inner_mut(&mut self) -> &mut T { + self.0.inner_mut() + } + + /// Consumes the adapter and returns the wrapped database. + #[inline] + pub fn into_inner(self) -> T { + self.0.into_inner() } } impl Database for WrapDatabaseAsync { - type Error = T::Error; + type Error = AsyncError; #[inline] fn basic(&mut self, address: Address) -> Result, Self::Error> { - self.rt.block_on(self.db.basic_async(address)) + self.0.basic(address) } #[inline] fn code_by_hash(&mut self, code_hash: B256) -> Result { - self.rt.block_on(self.db.code_by_hash_async(code_hash)) + self.0.code_by_hash(code_hash) } #[inline] @@ -166,12 +823,9 @@ impl Database for WrapDatabaseAsync { address: Address, index: StorageKey, ) -> Result { - self.rt.block_on(self.db.storage_async(address, index)) + self.0.storage(address, index) } - /// Gets storage value of account by its id. - /// - /// Wraps [`DatabaseAsync::storage_by_account_id_async`] in a blocking call. #[inline] fn storage_by_account_id( &mut self, @@ -179,29 +833,27 @@ impl Database for WrapDatabaseAsync { account_id: AccountId, storage_key: StorageKey, ) -> Result { - self.rt.block_on( - self.db - .storage_by_account_id_async(address, account_id, storage_key), - ) + self.0 + .storage_by_account_id(address, account_id, storage_key) } #[inline] fn block_hash(&mut self, number: u64) -> Result { - self.rt.block_on(self.db.block_hash_async(number)) + self.0.block_hash(number) } } impl DatabaseRef for WrapDatabaseAsync { - type Error = T::Error; + type Error = AsyncError; #[inline] fn basic_ref(&self, address: Address) -> Result, Self::Error> { - self.rt.block_on(self.db.basic_async_ref(address)) + self.0.basic_ref(address) } #[inline] fn code_by_hash_ref(&self, code_hash: B256) -> Result { - self.rt.block_on(self.db.code_by_hash_async_ref(code_hash)) + self.0.code_by_hash_ref(code_hash) } #[inline] @@ -210,7 +862,7 @@ impl DatabaseRef for WrapDatabaseAsync { address: Address, index: StorageKey, ) -> Result { - self.rt.block_on(self.db.storage_async_ref(address, index)) + self.0.storage_ref(address, index) } #[inline] @@ -220,19 +872,29 @@ impl DatabaseRef for WrapDatabaseAsync { account_id: AccountId, storage_key: StorageKey, ) -> Result { - self.rt.block_on( - self.db - .storage_by_account_id_async_ref(address, account_id, storage_key), - ) + self.0 + .storage_by_account_id_ref(address, account_id, storage_key) } #[inline] fn block_hash_ref(&self, number: u64) -> Result { - self.rt.block_on(self.db.block_hash_async_ref(number)) + self.0.block_hash_ref(number) } } -// Hold a tokio runtime handle or full runtime +impl DatabaseCommit for WrapDatabaseAsync { + #[inline] + fn commit(&mut self, changes: AddressMap) { + self.0.commit(changes); + } + + #[inline] + fn commit_iter(&mut self, changes: &mut dyn Iterator) { + self.0.commit_iter(changes); + } +} + +// Hold a tokio runtime handle or full runtime. #[derive(Debug)] enum HandleOrRuntime { Handle(Handle), @@ -241,48 +903,346 @@ enum HandleOrRuntime { impl HandleOrRuntime { #[inline] - fn block_on(&self, f: F) -> F::Output - where - F: Future + Send, - F::Output: Send, - { + fn handle(&self) -> &Handle { match self { - Self::Handle(handle) => { - // We use a conservative approach: if we're currently in a multi-threaded - // tokio runtime context, we use block_in_place. This works because: - // 1. If we're in the SAME runtime, block_in_place prevents deadlock - // 2. If we're in a DIFFERENT runtime, block_in_place still works safely - // (it just moves the work off the current worker thread before blocking) - // - // This approach is compatible with all tokio versions and doesn't require - // runtime identity comparison (Handle::id() is unstable feature). - let should_use_block_in_place = Handle::try_current() - .ok() - .map(|current| { - // Only use block_in_place for multi-threaded runtimes - // (block_in_place panics on current-thread runtime) - !matches!( - current.runtime_flavor(), - tokio::runtime::RuntimeFlavor::CurrentThread - ) - }) - .unwrap_or(false); - - if should_use_block_in_place { - // We're in a multi-threaded runtime context. - // Use block_in_place to: - // 1. Move the blocking operation off the async worker thread - // 2. Prevent potential deadlock if this is the same runtime - // 3. Allow other tasks to continue executing - tokio::task::block_in_place(move || handle.block_on(f)) - } else { - // Safe to call block_on directly in these cases: - // - We're outside any runtime context - // - We're in a current-thread runtime (where block_in_place doesn't work) - handle.block_on(f) - } + Self::Handle(handle) => handle, + Self::Runtime(runtime) => runtime.handle(), + } + } +} + +#[cfg(test)] +mod tests { + use super::{block_on_current, on_fiber, AsyncDb, AsyncError, DatabaseAsync}; + use crate::Database; + use core::{convert::Infallible, fmt, future::Future, pin::Pin, task::Poll}; + use primitives::{Address, StorageKey, StorageValue, B256}; + use state::{AccountInfo, Bytecode}; + use std::task::{Context, Waker}; + + #[test] + fn block_on_requires_fiber() { + assert!(matches!( + block_on_current(core::future::ready(())), + Err(AsyncError::NotOnFiber) + )); + } + + #[test] + fn fiber_suspends_and_resumes_pending_future() { + let mut state = 1; + let mut future = core::pin::pin!(on_fiber(|| { + state += block_on_current(PendingOnce { pending: true }).unwrap(); + state + })); + let waker = Waker::noop(); + let mut cx = Context::from_waker(waker); + + assert!(matches!(future.as_mut().poll(&mut cx), Poll::Pending)); + assert!(matches!(future.as_mut().poll(&mut cx), Poll::Ready(Ok(3)))); + } + + #[test] + fn fiber_reuses_stack_slot() { + let mut stack = super::FiberStack::default(); + let stack_ptr = core::ptr::NonNull::from(&mut stack); + + poll_ready(unsafe { + super::on_fiber_result_with_stack(stack_ptr, || Ok::<_, Infallible>(1)) + }) + .unwrap(); + assert!(stack.stack.is_some()); + poll_ready(unsafe { + super::on_fiber_result_with_stack(stack_ptr, || Ok::<_, Infallible>(2)) + }) + .unwrap(); + assert!(stack.stack.is_some()); + } + + #[test] + fn async_database_adapts_to_database() { + let mut db = AsyncDb::new(TestDb); + + let value = poll_ready(on_fiber(|| { + Database::storage(&mut db, Address::ZERO, StorageKey::from(7)).unwrap() + })) + .unwrap(); + + assert_eq!(value, StorageValue::from(9)); + } + + #[test] + fn async_database_suspends_until_ready() { + let mut db = AsyncDb::new(PendingDb { pending: true }); + let mut future = core::pin::pin!(on_fiber(|| { + Database::storage(&mut db, Address::ZERO, StorageKey::from(7)).unwrap() + })); + let waker = Waker::noop(); + let mut cx = Context::from_waker(waker); + + assert!(matches!(future.as_mut().poll(&mut cx), Poll::Pending)); + assert!( + matches!(future.as_mut().poll(&mut cx), Poll::Ready(Ok(value)) if value == StorageValue::from(9)) + ); + } + + #[test] + fn async_database_returns_database_error() { + let mut db = AsyncDb::new(FailingDb); + let result = poll_ready(on_fiber(|| { + Database::storage(&mut db, Address::ZERO, StorageKey::from(7)) + })); + + assert!(matches!(result, Ok(Err(AsyncError::Inner(TestError))))); + } + + #[test] + fn synchronous_database_blocks_with_current_tokio_runtime() { + let runtime = tokio::runtime::Runtime::new().unwrap(); + let _guard = runtime.enter(); + let mut db = AsyncDb::new(TokioDb); + + let value = Database::storage(&mut db, Address::ZERO, StorageKey::from(7)).unwrap(); + + assert_eq!(value, StorageValue::from(9)); + } + + #[test] + fn blocking_constructor_uses_current_tokio_runtime() { + let runtime = tokio::runtime::Runtime::new().unwrap(); + let _guard = runtime.enter(); + let mut db = AsyncDb::blocking(TokioDb).unwrap(); + + let value = Database::storage(&mut db, Address::ZERO, StorageKey::from(7)).unwrap(); + + assert_eq!(value, StorageValue::from(9)); + } + + #[test] + fn synchronous_database_blocks_with_stored_tokio_handle() { + let runtime = tokio::runtime::Runtime::new().unwrap(); + let mut db = AsyncDb::with_handle(TokioDb, runtime.handle().clone()); + + let value = Database::storage(&mut db, Address::ZERO, StorageKey::from(7)).unwrap(); + + assert_eq!(value, StorageValue::from(9)); + } + + #[test] + fn synchronous_database_requires_runtime_handle() { + let mut db = AsyncDb::new(TestDb); + + let result = Database::storage(&mut db, Address::ZERO, StorageKey::from(7)); + + assert!(matches!(result, Err(AsyncError::Runtime))); + } + + #[test] + fn dropping_fiber_cancels_blocked_future() { + let mut saw_cancel = false; + { + let mut future = core::pin::pin!(on_fiber(|| { + saw_cancel = matches!(block_on_current(PendingForever), Err(AsyncError::Cancelled)); + })); + let waker = Waker::noop(); + let mut cx = Context::from_waker(waker); + + assert!(matches!(future.as_mut().poll(&mut cx), Poll::Pending)); + } + assert!(saw_cancel); + } + + fn poll_ready(future: F) -> F::Output { + let mut future = core::pin::pin!(future); + let waker = Waker::noop(); + let mut cx = Context::from_waker(waker); + + match future.as_mut().poll(&mut cx) { + Poll::Ready(value) => value, + Poll::Pending => panic!("future unexpectedly pending"), + } + } + + struct PendingOnce { + pending: bool, + } + + impl Future for PendingOnce { + type Output = i32; + + fn poll(mut self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll { + if self.pending { + self.pending = false; + Poll::Pending + } else { + Poll::Ready(2) } - Self::Runtime(rt) => rt.block_on(f), } } + + struct PendingForever; + + impl Future for PendingForever { + type Output = (); + + fn poll(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll { + Poll::Pending + } + } + + struct TestDb; + + impl DatabaseAsync for TestDb { + type Error = Infallible; + + async fn basic_async( + &mut self, + _address: Address, + ) -> Result, Self::Error> { + Ok(None) + } + + async fn code_by_hash_async(&mut self, _code_hash: B256) -> Result { + Ok(Bytecode::default()) + } + + async fn storage_async( + &mut self, + _address: Address, + _index: StorageKey, + ) -> Result { + Ok(StorageValue::from(9)) + } + + async fn block_hash_async(&mut self, _number: u64) -> Result { + Ok(B256::ZERO) + } + } + + struct PendingDb { + pending: bool, + } + + impl DatabaseAsync for PendingDb { + type Error = Infallible; + + async fn basic_async( + &mut self, + _address: Address, + ) -> Result, Self::Error> { + Ok(None) + } + + async fn code_by_hash_async(&mut self, _code_hash: B256) -> Result { + Ok(Bytecode::default()) + } + + async fn storage_async( + &mut self, + _address: Address, + _index: StorageKey, + ) -> Result { + PendingStorage { + pending: &mut self.pending, + } + .await; + Ok(StorageValue::from(9)) + } + + async fn block_hash_async(&mut self, _number: u64) -> Result { + Ok(B256::ZERO) + } + } + + struct PendingStorage<'a> { + pending: &'a mut bool, + } + + impl Future for PendingStorage<'_> { + type Output = (); + + fn poll(mut self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll { + if *self.pending { + *self.pending = false; + Poll::Pending + } else { + Poll::Ready(()) + } + } + } + + struct FailingDb; + + impl DatabaseAsync for FailingDb { + type Error = TestError; + + async fn basic_async( + &mut self, + _address: Address, + ) -> Result, Self::Error> { + Ok(None) + } + + async fn code_by_hash_async(&mut self, _code_hash: B256) -> Result { + Ok(Bytecode::default()) + } + + async fn storage_async( + &mut self, + _address: Address, + _index: StorageKey, + ) -> Result { + Err(TestError) + } + + async fn block_hash_async(&mut self, _number: u64) -> Result { + Ok(B256::ZERO) + } + } + + struct TokioDb; + + impl DatabaseAsync for TokioDb { + type Error = Infallible; + + async fn basic_async( + &mut self, + _address: Address, + ) -> Result, Self::Error> { + tokio::task::yield_now().await; + Ok(None) + } + + async fn code_by_hash_async(&mut self, _code_hash: B256) -> Result { + tokio::task::yield_now().await; + Ok(Bytecode::default()) + } + + async fn storage_async( + &mut self, + _address: Address, + _index: StorageKey, + ) -> Result { + tokio::task::yield_now().await; + Ok(StorageValue::from(9)) + } + + async fn block_hash_async(&mut self, _number: u64) -> Result { + tokio::task::yield_now().await; + Ok(B256::ZERO) + } + } + + #[derive(Clone, Copy, Debug, PartialEq, Eq)] + struct TestError; + + impl fmt::Display for TestError { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + f.write_str("storage read failed") + } + } + + impl core::error::Error for TestError {} + + impl crate::DBErrorMarker for TestError {} } diff --git a/crates/database/interface/src/lib.rs b/crates/database/interface/src/lib.rs index cd39bb9bdc..e8c6a7077e 100644 --- a/crates/database/interface/src/lib.rs +++ b/crates/database/interface/src/lib.rs @@ -39,7 +39,7 @@ pub mod erased_error; pub mod try_commit; #[cfg(feature = "asyncdb")] -pub use async_db::{DatabaseAsync, WrapDatabaseAsync}; +pub use async_db::{AsyncDb, AsyncError, AsyncResult, DatabaseAsync, WrapDatabaseAsync}; pub use empty_db::{EmptyDB, EmptyDBTyped}; pub use erased_error::ErasedError; pub use try_commit::{ArcUpgradeError, TryDatabaseCommit}; diff --git a/crates/handler/Cargo.toml b/crates/handler/Cargo.toml index 8e73f5bb51..fa1ed22c39 100644 --- a/crates/handler/Cargo.toml +++ b/crates/handler/Cargo.toml @@ -65,6 +65,7 @@ serde = [ "interpreter/serde", "derive-where/serde", ] +asyncdb = ["std", "context/asyncdb", "database-interface/asyncdb"] # Deprecated, please use `serde` feature instead. serde-json = ["serde"] diff --git a/crates/handler/src/api.rs b/crates/handler/src/api.rs index f6428781e3..e4754b70bd 100644 --- a/crates/handler/src/api.rs +++ b/crates/handler/src/api.rs @@ -8,9 +8,13 @@ use context::{ }, Block, ContextSetters, ContextTr, Database, Evm, JournalTr, Transaction, }; +#[cfg(feature = "asyncdb")] +use database_interface::async_db::{on_fiber_result_with_stack, AsyncResult}; use database_interface::DatabaseCommit; use interpreter::{interpreter::EthInterpreter, InterpreterResult}; use state::EvmState; +#[cfg(feature = "asyncdb")] +use std::ptr::NonNull; use std::vec::Vec; /// Type alias for the result of transact_many_finalize to reduce type complexity. @@ -170,6 +174,25 @@ pub trait ExecuteCommitEvm: ExecuteEvm { } } +/// Async extension of the [`ExecuteEvm`] trait that runs execution on an async fiber. +#[cfg(feature = "asyncdb")] +pub trait ExecuteEvmAsync: ExecuteEvm { + /// Execute transaction and store state inside journal on an async fiber. + fn transact_one_async( + &mut self, + tx: Self::Tx, + ) -> impl core::future::Future> + Send + '_; + + /// Transact the given transaction and finalize in a single operation on an async fiber. + fn transact_async( + &mut self, + tx: Self::Tx, + ) -> impl core::future::Future< + Output = AsyncResult, Self::Error>, + > + Send + + '_; +} + impl ExecuteEvm for Evm> where @@ -208,6 +231,45 @@ where } } +#[cfg(feature = "asyncdb")] +impl ExecuteEvmAsync + for Evm> +where + CTX: ContextTr> + ContextSetters, + INST: InstructionProvider, + PRECOMPILES: PrecompileProvider, + ExecutionResult: Send, + EvmState: Send, + EVMError<::Error, InvalidTransaction>: Send, + ::Tx: Send, +{ + #[inline] + fn transact_one_async( + &mut self, + tx: Self::Tx, + ) -> impl core::future::Future> + Send + '_ + { + let stack = NonNull::from(&mut self.async_stack); + // SAFETY: The returned future owns the exclusive `&mut self` borrow, so nothing else can + // access the EVM stack slot until that future is dropped. + unsafe { on_fiber_result_with_stack(stack, move || self.transact_one(tx)) } + } + + #[inline] + fn transact_async( + &mut self, + tx: Self::Tx, + ) -> impl core::future::Future< + Output = AsyncResult, Self::Error>, + > + Send + + '_ { + let stack = NonNull::from(&mut self.async_stack); + // SAFETY: The returned future owns the exclusive `&mut self` borrow, so nothing else can + // access the EVM stack slot until that future is dropped. + unsafe { on_fiber_result_with_stack(stack, move || self.transact(tx)) } + } +} + impl ExecuteCommitEvm for Evm> where diff --git a/crates/handler/src/lib.rs b/crates/handler/src/lib.rs index 0436cbcaaa..58eeb1a40b 100644 --- a/crates/handler/src/lib.rs +++ b/crates/handler/src/lib.rs @@ -41,6 +41,8 @@ pub use primitives; pub use state; // Public exports +#[cfg(feature = "asyncdb")] +pub use api::ExecuteEvmAsync; pub use api::{ExecuteCommitEvm, ExecuteEvm}; pub use evm::{EvmTr, FrameTr}; pub use frame::{handle_reservoir_remaining_gas, return_create, ContextTrDbError, EthFrame}; @@ -52,4 +54,6 @@ pub use mainnet_handler::MainnetHandler; pub use precompile_provider::{ precompile_output_to_interpreter_result, EthPrecompiles, PrecompileProvider, }; +#[cfg(feature = "asyncdb")] +pub use system_call::SystemCallEvmAsync; pub use system_call::{SystemCallCommitEvm, SystemCallEvm, SystemCallTx, SYSTEM_ADDRESS}; diff --git a/crates/handler/src/mainnet_builder.rs b/crates/handler/src/mainnet_builder.rs index 46baa54fc8..5426162380 100644 --- a/crates/handler/src/mainnet_builder.rs +++ b/crates/handler/src/mainnet_builder.rs @@ -43,6 +43,8 @@ where instruction: EthInstructions::new_mainnet_with_spec(spec), precompiles: EthPrecompiles::new(spec), frame_stack: FrameStack::new_prealloc(8), + #[cfg(feature = "asyncdb")] + async_stack: database_interface::async_db::FiberStack::default(), } } @@ -57,6 +59,8 @@ where instruction: EthInstructions::new_mainnet_with_spec(spec), precompiles: EthPrecompiles::new(spec), frame_stack: FrameStack::new_prealloc(8), + #[cfg(feature = "asyncdb")] + async_stack: database_interface::async_db::FiberStack::default(), } } } diff --git a/crates/handler/src/precompile_provider.rs b/crates/handler/src/precompile_provider.rs index c8064b840e..302d0fc0cd 100644 --- a/crates/handler/src/precompile_provider.rs +++ b/crates/handler/src/precompile_provider.rs @@ -282,6 +282,8 @@ mod tests { instruction: EthInstructions::::new_mainnet_with_spec(spec), precompiles: OverspendingPrecompiles::new(spec), frame_stack: FrameStack::new_prealloc(8), + #[cfg(feature = "asyncdb")] + async_stack: database_interface::async_db::FiberStack::default(), }; let tx = TxEnv::builder() diff --git a/crates/handler/src/system_call.rs b/crates/handler/src/system_call.rs index 533d9fc20e..27bddc8b46 100644 --- a/crates/handler/src/system_call.rs +++ b/crates/handler/src/system_call.rs @@ -24,10 +24,14 @@ use crate::{ MainnetHandler, PrecompileProvider, }; use context::{result::ExecResultAndState, ContextSetters, ContextTr, Evm, JournalTr, TxEnv}; +#[cfg(feature = "asyncdb")] +use database_interface::async_db::{on_fiber_result_with_stack, AsyncResult}; use database_interface::DatabaseCommit; use interpreter::{interpreter::EthInterpreter, InterpreterResult}; use primitives::{address, eip8037, Address, Bytes, TxKind}; use state::EvmState; +#[cfg(feature = "asyncdb")] +use std::ptr::NonNull; /// The system address used for system calls. pub const SYSTEM_ADDRESS: Address = address!("0xfffffffffffffffffffffffffffffffffffffffe"); @@ -227,6 +231,51 @@ pub trait SystemCallCommitEvm: SystemCallEvm + ExecuteCommitEvm { } } +/// Async extension of the [`SystemCallEvm`] trait that runs execution on an async fiber. +#[cfg(feature = "asyncdb")] +pub trait SystemCallEvmAsync: SystemCallEvm { + /// System call executed on an async fiber. + fn system_call_one_with_caller_async( + &mut self, + caller: Address, + system_contract_address: Address, + data: Bytes, + ) -> impl core::future::Future> + Send + '_; + + /// System call executed on an async fiber. + fn system_call_one_async( + &mut self, + system_contract_address: Address, + data: Bytes, + ) -> impl core::future::Future> + Send + '_ + { + self.system_call_one_with_caller_async(SYSTEM_ADDRESS, system_contract_address, data) + } + + /// System call executed and finalized on an async fiber. + fn system_call_with_caller_async( + &mut self, + caller: Address, + system_contract_address: Address, + data: Bytes, + ) -> impl core::future::Future< + Output = AsyncResult, Self::Error>, + > + Send + + '_; + + /// System call executed and finalized on an async fiber. + fn system_call_async( + &mut self, + system_contract_address: Address, + data: Bytes, + ) -> impl core::future::Future< + Output = AsyncResult, Self::Error>, + > + Send + + '_ { + self.system_call_with_caller_async(SYSTEM_ADDRESS, system_contract_address, data) + } +} + impl SystemCallEvm for Evm> where @@ -251,6 +300,57 @@ where } } +#[cfg(feature = "asyncdb")] +impl SystemCallEvmAsync + for Evm> +where + CTX: ContextTr, Tx: SystemCallTx> + ContextSetters, + INST: InstructionProvider, + PRECOMPILES: PrecompileProvider, + Self: ExecuteEvm, + ::ExecutionResult: Send, + ::State: Send, + ::Error: Send, +{ + #[inline] + fn system_call_one_with_caller_async( + &mut self, + caller: Address, + system_contract_address: Address, + data: Bytes, + ) -> impl core::future::Future> + Send + '_ + { + let stack = NonNull::from(&mut self.async_stack); + // SAFETY: The returned future owns the exclusive `&mut self` borrow, so nothing else can + // access the EVM stack slot until that future is dropped. + unsafe { + on_fiber_result_with_stack(stack, move || { + self.system_call_one_with_caller(caller, system_contract_address, data) + }) + } + } + + #[inline] + fn system_call_with_caller_async( + &mut self, + caller: Address, + system_contract_address: Address, + data: Bytes, + ) -> impl core::future::Future< + Output = AsyncResult, Self::Error>, + > + Send + + '_ { + let stack = NonNull::from(&mut self.async_stack); + // SAFETY: The returned future owns the exclusive `&mut self` borrow, so nothing else can + // access the EVM stack slot until that future is dropped. + unsafe { + on_fiber_result_with_stack(stack, move || { + self.system_call_with_caller(caller, system_contract_address, data) + }) + } + } +} + impl SystemCallCommitEvm for Evm> where diff --git a/crates/revm/Cargo.toml b/crates/revm/Cargo.toml index da5a8c221e..26fdca97c6 100644 --- a/crates/revm/Cargo.toml +++ b/crates/revm/Cargo.toml @@ -70,7 +70,12 @@ serde = [ arbitrary = ["primitives/arbitrary"] asm-keccak = ["primitives/asm-keccak"] sha3-keccak = ["primitives/sha3-keccak"] -asyncdb = ["database-interface/asyncdb"] +asyncdb = [ + "std", + "context/asyncdb", + "database-interface/asyncdb", + "handler/asyncdb", +] # Enables alloydb inside database crate alloydb = ["database/alloydb"] diff --git a/crates/revm/src/lib.rs b/crates/revm/src/lib.rs index 381004090a..c03315bb13 100644 --- a/crates/revm/src/lib.rs +++ b/crates/revm/src/lib.rs @@ -36,10 +36,14 @@ pub use context::{ journal::{Journal, JournalEntry}, Context, }; +#[cfg(feature = "asyncdb")] +pub use database_interface::{AsyncDb, AsyncError, AsyncResult, DatabaseAsync, WrapDatabaseAsync}; pub use database_interface::{Database, DatabaseCommit, DatabaseRef}; pub use handler::{ ExecuteCommitEvm, ExecuteEvm, MainBuilder, MainContext, MainnetEvm, SystemCallCommitEvm, SystemCallEvm, }; +#[cfg(feature = "asyncdb")] +pub use handler::{ExecuteEvmAsync, SystemCallEvmAsync}; pub use inspector::{InspectCommitEvm, InspectEvm, InspectSystemCallEvm, Inspector}; pub use precompile::install_crypto; diff --git a/examples/custom_precompile_journal/Cargo.toml b/examples/custom_precompile_journal/Cargo.toml index 8a052ce301..492749f5e0 100644 --- a/examples/custom_precompile_journal/Cargo.toml +++ b/examples/custom_precompile_journal/Cargo.toml @@ -13,3 +13,6 @@ rust-version.workspace = true [dependencies] revm = { path = "../../crates/revm", features = ["optional_eip3607"] } anyhow = "1.0" + +[features] +asyncdb = ["revm/asyncdb"] diff --git a/examples/custom_precompile_journal/src/custom_evm.rs b/examples/custom_precompile_journal/src/custom_evm.rs index 3b00331e6b..72164ddc59 100644 --- a/examples/custom_precompile_journal/src/custom_evm.rs +++ b/examples/custom_precompile_journal/src/custom_evm.rs @@ -54,6 +54,8 @@ where instruction: EthInstructions::new_mainnet_with_spec(SpecId::CANCUN), precompiles: CustomPrecompileProvider::new_with_spec(SpecId::CANCUN), frame_stack: FrameStack::new(), + #[cfg(feature = "asyncdb")] + async_stack: revm::database_interface::async_db::FiberStack::default(), }) } } diff --git a/examples/my_evm/Cargo.toml b/examples/my_evm/Cargo.toml index b5a335171a..571202d3d7 100644 --- a/examples/my_evm/Cargo.toml +++ b/examples/my_evm/Cargo.toml @@ -19,3 +19,6 @@ workspace = true [dependencies] revm = { workspace = true } + +[features] +asyncdb = ["revm/asyncdb"] diff --git a/examples/my_evm/src/evm.rs b/examples/my_evm/src/evm.rs index fbdeacb8be..c076f3e5d9 100644 --- a/examples/my_evm/src/evm.rs +++ b/examples/my_evm/src/evm.rs @@ -51,6 +51,8 @@ impl MyEvm { instruction: EthInstructions::new_mainnet_with_spec(SpecId::default()), precompiles: EthPrecompiles::new(SpecId::default()), frame_stack: FrameStack::new(), + #[cfg(feature = "asyncdb")] + async_stack: revm::database_interface::async_db::FiberStack::default(), }) } } From b1e7ccc6c35190d264039f5cc2d4ee67995e8ec7 Mon Sep 17 00:00:00 2001 From: DaniPopes <57450786+DaniPopes@users.noreply.github.com> Date: Sun, 24 May 2026 20:15:46 +0200 Subject: [PATCH 2/2] chore: use crates.io corosensei --- Cargo.lock | 5 +++-- Cargo.toml | 4 +--- crates/database/src/states/bundle_account.rs | 2 +- 3 files changed, 5 insertions(+), 6 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 5fa2d13b90..1e6b7f975b 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1484,8 +1484,9 @@ checksum = "773648b94d0e5d620f64f280777445740e61fe701025087ec8b57f45c791888b" [[package]] name = "corosensei" -version = "0.3.3" -source = "git+https://github.com/Amanieu/corosensei?rev=3a122508ab1d7c25e4dee8b2b4b4c47046c3c3d8#3a122508ab1d7c25e4dee8b2b4b4c47046c3c3d8" +version = "0.3.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "6886a0c0f263965933c438626e7179139a62b978a33aa18281cbf0cd5a975f34" dependencies = [ "autocfg", "cfg-if", diff --git a/Cargo.toml b/Cargo.toml index 0bf26a230b..a144ac378b 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -112,9 +112,7 @@ serde_json = { version = "1.0", default-features = false } auto_impl = "1.3.0" bitflags = { version = "2.10", default-features = false } cfg-if = { version = "1.0", default-features = false } -# Uses public `Coroutine::with_stack_unchecked`. -# https://github.com/Amanieu/corosensei/pull/74 -corosensei = { version = "0.3.3", git = "https://github.com/Amanieu/corosensei", rev = "3a122508ab1d7c25e4dee8b2b4b4c47046c3c3d8", default-features = false, features = [ +corosensei = { version = "0.3.4", default-features = false, features = [ "default-stack", "unwind", ] } diff --git a/crates/database/src/states/bundle_account.rs b/crates/database/src/states/bundle_account.rs index 4254770dbd..0c9c537da5 100644 --- a/crates/database/src/states/bundle_account.rs +++ b/crates/database/src/states/bundle_account.rs @@ -372,6 +372,6 @@ impl BundleAccount { } }; - account_revert.and_then(|acc| if acc.is_empty() { None } else { Some(acc) }) + account_revert.filter(|acc| !acc.is_empty()) } }