sim extended

Cargo.toml

@@ -158,7 +158,7 @@ ahash = { version = "0.8", default-features = false, features = ["std"] }
 anyhow = "1.0.68"
 anymap = "0.12"
 arrayvec = "0.7.2"
-async-channel = "2.5"
+async-channel = { version = "2.5", default-features = false }
 async-stream = "0.3.6"
 async-trait = "0.1.68"
 axum = { version = "0.7", features = ["tracing"] }

crates/durability/Cargo.toml

@@ -13,7 +13,7 @@ fallocate = ["spacetimedb-commitlog/fallocate"]
 
 [dependencies]
 anyhow.workspace = true
-async-channel.workspace = true
+
 futures.workspace = true
 itertools.workspace = true
 log.workspace = true

crates/durability/src/imp/local.rs

@@ -100,7 +100,7 @@ where
     ///
     /// The queue is bounded to
     /// `Options::QUEUE_CAPACITY_MULTIPLIER * Options::batch_capacity`.
-    queue: async_channel::Sender<PreparedTx<Txdata<T>>>,
+    queue: spacetimedb_runtime::channel::Sender<PreparedTx<Txdata<T>>>,
     /// How many transactions are pending durability, including items buffered
     /// in the queue and items currently being written by the actor.
     ///
@@ -167,7 +167,7 @@ where
         lock: Option<LockedFile>,
     ) -> Result<Self, OpenError> {
         let queue_capacity = opts.queue_capacity();
-        let (queue, txdata_rx) = async_channel::bounded(queue_capacity);
+        let (queue, txdata_rx) = spacetimedb_runtime::channel::bounded(queue_capacity, rt.clone());
         let queue_depth = Arc::new(AtomicU64::new(0));
         let (durable_tx, durable_rx) = watch::channel(clog.max_committed_offset());
         let actor = rt.spawn(
@@ -253,7 +253,7 @@ where
     R: Repo + Send + Sync + 'static,
 {
     #[instrument(name = "durability::local::actor", skip_all)]
-    async fn run(self, transactions_rx: async_channel::Receiver<PreparedTx<Txdata<T>>>) {
+    async fn run(self, transactions_rx: spacetimedb_runtime::channel::Receiver<PreparedTx<Txdata<T>>>) {
         info!("starting durability actor");
 
         let mut tx_buf = Vec::with_capacity(self.batch_capacity.get());
@@ -419,8 +419,8 @@ where
     }
 }
 
-/// Implement tokio's `recv_many` for an `async_channel` receiver.
-async fn recv_many<T>(chan: &async_channel::Receiver<T>, buf: &mut Vec<T>, limit: usize) -> usize {
+/// Implement tokio's `recv_many` for a `spacetimedb_runtime::channel::Receiver` receiver.
+async fn recv_many<T>(chan: &spacetimedb_runtime::channel::Receiver<T>, buf: &mut Vec<T>, limit: usize) -> usize {
     let mut n = 0;
     if !chan.is_empty() {
         buf.reserve(chan.len().min(limit));

crates/runtime/Cargo.toml

@@ -12,13 +12,15 @@ workspace = true
 [dependencies]
 tokio = { workspace = true, optional = true }
 async-task = { version = "4.4", default-features = false, optional = true }
-spin = { version = "0.9", default-features = false, features = ["mutex", "spin_mutex"], optional = true }
+spin = { version = "0.9", default-features = false, features = ["mutex", "spin_mutex", "once"], optional = true }
 libc = { version = "0.2", optional = true }
+async-channel.workspace = true
 
 [dev-dependencies]
 futures.workspace = true
+parking_lot = "0.12"
 
 [features]
 default = ["tokio"]
-tokio = ["dep:tokio"]
+tokio = ["dep:tokio", "async-channel/std"]
 simulation = ["dep:async-task", "dep:spin", "dep:libc"]

crates/runtime/src/channel.rs

@@ -0,0 +1,88 @@
+use async_channel::SendError;
+
+#[cfg(feature = "simulation")]
+use async_channel::TrySendError;
+
+/// Sending end of a bounded channel.
+///
+/// Production: uses `send_blocking` (futex, true backpressure).
+/// Simulation: uses `try_send` + executor tick when full (no futex).
+pub struct Sender<T> {
+    inner: async_channel::Sender<T>,
+    rt: crate::Handle,
+}
+
+/// Receiving end of a bounded channel.
+///
+/// Identical to `async_channel::Receiver` in both modes.
+pub struct Receiver<T> {
+    inner: async_channel::Receiver<T>,
+}
+
+impl<T> Receiver<T> {
+    pub fn recv(&self) -> async_channel::Recv<'_, T> {
+        self.inner.recv()
+    }
+
+    pub fn try_recv(&self) -> Result<T, async_channel::TryRecvError> {
+        self.inner.try_recv()
+    }
+
+    pub fn close(&self) {
+        self.inner.close();
+    }
+
+    pub fn is_empty(&self) -> bool {
+        self.inner.is_empty()
+    }
+
+    pub fn len(&self) -> usize {
+        self.inner.len()
+    }
+}
+
+/// Create a bounded channel with the given capacity and runtime handle.
+///
+/// The returned `Sender` adapts its send strategy based on the runtime variant:
+/// - `Handle::Tokio` → `send_blocking` (OS thread park)
+/// - `Handle::Simulation` → `try_send` + executor tick on full (no futex)
+pub fn bounded<T>(cap: usize, rt: crate::Handle) -> (Sender<T>, Receiver<T>) {
+    let (tx, rx) = async_channel::bounded(cap);
+    (Sender { inner: tx, rt }, Receiver { inner: rx })
+}
+
+impl<T> Sender<T> {
+    /// Close the sender, signalling that no more messages will be sent.
+    pub fn close(&self) {
+        self.inner.close();
+    }
+
+    /// Send a message, applying backpressure.
+    ///
+    /// In production (`Tokio`) this parks the OS thread via futex.
+    /// In simulation it loops on `try_send`, calling `sim::Handle::run_all_ready()`
+    /// when the channel is full so the actor can make progress without
+    /// actually parking the (sole) thread.
+    pub fn send_blocking(&self, msg: T) -> Result<(), SendError<T>> {
+        match &self.rt {
+            #[cfg(feature = "tokio")]
+            crate::Handle::Tokio(_) => self.inner.send_blocking(msg),
+            #[cfg(feature = "simulation")]
+            crate::Handle::Simulation(sim) => {
+                let mut msg = msg;
+                loop {
+                    match self.inner.try_send(msg) {
+                        Ok(()) => return Ok(()),
+                        Err(TrySendError::Full(m)) => {
+                            msg = m;
+                            sim.run_all_ready();
+                        }
+                        Err(TrySendError::Closed(m)) => return Err(SendError(m)),
+                    }
+                }
+            }
+            #[cfg(not(any(feature = "tokio", feature = "simulation")))]
+            _ => unreachable!("runtime::channel::send called with no backend enabled"),
+        }
+    }
+}