Serve broker calls on per-call lanes

dev_tests/src/ratchet.rs

@@ -41,7 +41,7 @@ fn ratchet_pub_uses() -> Result<()> {
             ("litebox_broker_local_userland/", 2),
             ("litebox_broker_platform_linux_userland/", 3),
             ("litebox_broker_platform_windows_userland/", 1),
-            ("litebox_broker_transport_linux_userland/", 3),
+            ("litebox_broker_transport_linux_userland/", 2),
             ("litebox_broker_transport_windows_userland/", 4),
             ("litebox_common_linux/", 1),
             ("litebox_common_macos/", 2),

litebox_broker_local_userland/src/linux.rs

@@ -17,7 +17,7 @@ use litebox_broker_local::{BrokerLocal, BrokerNotifications};
 use litebox_broker_protocol::message::BrokerNotification;
 use litebox_broker_protocol::process::ProcessStartupData;
 use litebox_broker_protocol::shared_buffer::SHARED_BUFFER_POOL_SIZE;
-use litebox_broker_transport::control_ring::ControlRing;
+use litebox_broker_transport::control_ring::{ControlRing, ControlRingLayout};
 use litebox_broker_transport_linux_userland::unix_socket::{
     UnixControlRingLocalCallChannel, UnixControlRingLocalNotificationChannel,
     UnixControlRingLocalShutdown, UnixStreamLocalSetupChannel,
@@ -84,13 +84,14 @@ fn negotiate(
         BrokerLocal::negotiate(setup_channel, |mut setup| {
             let shared_memory =
                 setup.receive_memfd(SHARED_BUFFER_POOL_SIZE, Some(setup_deadline))?;
-            let control_memory = setup.receive_control_ring(Some(setup_deadline))?;
+            let layout = ControlRingLayout::MULTI_LANE;
+            let control_memory = setup.receive_control_ring(layout, Some(setup_deadline))?;
             let process_image = setup.receive_process_image(Some(setup_deadline))?;
             let positional_io_fds = [
                 shared_memory.as_fd().as_raw_fd(),
                 control_memory.as_fd().as_raw_fd(),
             ];
-            let control_ring = ControlRing::new(control_memory).map_err(|error| {
+            let control_ring = ControlRing::new(control_memory, layout).map_err(|error| {
                 std::io::Error::new(
                     std::io::ErrorKind::InvalidData,
                     format!("invalid broker control ring: {error:?}"),
@@ -268,8 +269,8 @@ mod tests {
     };
     use litebox_broker_transport_linux_userland::memfd::MemfdSharedMemory;
     use litebox_broker_transport_linux_userland::unix_socket::{
-        UnixControlRingHostNotificationChannel, UnixControlRingHostRequestSource,
-        UnixControlRingHostResponseSink, UnixControlRingHostShutdown, UnixStreamHostSetupChannel,
+        UnixControlRingHostCallLaneSource, UnixControlRingHostNotificationChannel,
+        UnixControlRingHostShutdown, UnixStreamHostSetupChannel,
     };
     use litebox_broker_transport_linux_userland::unix_socket::{
         UnixControlRingLocalCallChannel, UnixControlRingLocalNotificationChannel,
@@ -318,18 +319,19 @@ mod tests {
         UnixControlRingLocalCallChannel,
         UnixControlRingLocalShutdown,
         UnixControlRingLocalNotificationChannel,
-        UnixControlRingHostRequestSource,
-        UnixControlRingHostResponseSink,
+        UnixControlRingHostCallLaneSource,
         UnixControlRingHostNotificationChannel,
         UnixControlRingHostShutdown,
     ) {
-        let local_memory = MemfdSharedMemory::create_control_ring().unwrap();
+        let layout = ControlRingLayout::MULTI_LANE;
+        let local_memory = MemfdSharedMemory::create_control_ring(layout).unwrap();
         let host_memory = MemfdSharedMemory::control_ring_from_received_fd(
             local_memory.as_fd().try_clone_to_owned().unwrap(),
+            layout,
         )
         .unwrap();
-        let local_ring = ControlRing::new(local_memory).unwrap();
-        let host_ring = ControlRing::new(host_memory).unwrap();
+        let local_ring = ControlRing::new(local_memory, layout).unwrap();
+        let host_ring = ControlRing::new(host_memory, layout).unwrap();
         let host_activation =
             std::thread::spawn(move || host_channel.into_active(host_ring).unwrap());
         let weak_association_coordinator = Arc::downgrade(association_coordinator);
@@ -340,14 +342,12 @@ mod tests {
                 }
             })
             .unwrap();
-        let (request_source, response_sink, host_notifications, host_shutdown) =
-            host_activation.join().unwrap();
+        let (host_lanes, host_notifications, host_shutdown) = host_activation.join().unwrap();
         (
             local_call,
             local_shutdown,
             local_notifications,
-            request_source,
-            response_sink,
+            host_lanes,
             host_notifications,
             host_shutdown,
         )
@@ -362,8 +362,7 @@ mod tests {
             active_channel,
             association_shutdown,
             notification_channel,
-            host_request_source,
-            host_response_sink,
+            host_lanes,
             _host_notifications,
             host_shutdown,
         ) = activate_control_channel(local_setup, host_control, &association_coordinator);
@@ -380,8 +379,7 @@ mod tests {
         .unwrap();
 
         host_shutdown.shutdown().unwrap();
-        drop(host_request_source);
-        drop(host_response_sink);
+        drop(host_lanes);
 
         failure_receiver
             .recv_timeout(Duration::from_secs(1))
@@ -401,8 +399,7 @@ mod tests {
             active_channel,
             association_shutdown,
             _notification_channel,
-            mut host_request_source,
-            _host_response_sink,
+            mut host_lanes,
             _host_notifications,
             _host_shutdown,
         ) = activate_control_channel(local_setup, host_control, &association_coordinator);
@@ -419,10 +416,9 @@ mod tests {
         let (failure_sender, failure_receiver) = mpsc::sync_channel(1);
         association_coordinator.install_dispatch(move || failure_sender.send(()).unwrap());
         failure_receiver.try_recv().unwrap();
-        assert_eq!(
-            host_request_source.recv_request().unwrap(),
-            HostReceive::PeerClosed
-        );
+        // Lane 0 is always demanded, so the broker may hand it out before it
+        // sees the close.
+        while host_lanes.next_lane().unwrap().is_some() {}
         drop(active_channel);
     }
 
@@ -435,8 +431,7 @@ mod tests {
             active_channel,
             association_shutdown,
             notification_channel,
-            _host_request_source,
-            _host_response_sink,
+            _host_lanes,
             mut host_notifications,
             host_shutdown,
         ) = activate_control_channel(local_setup, host_control, &association_coordinator);

litebox_broker_local_userland/src/windows.rs

@@ -10,7 +10,7 @@ use litebox_broker_local::{BrokerLocal, BrokerNotifications};
 use litebox_broker_protocol::message::BrokerNotification;
 use litebox_broker_protocol::process::ProcessStartupData;
 use litebox_broker_protocol::shared_buffer::SHARED_BUFFER_POOL_SIZE;
-use litebox_broker_transport::control_ring::ControlRing;
+use litebox_broker_transport::control_ring::{ControlRing, ControlRingLayout};
 use litebox_broker_transport_windows_userland::control_ring::{
     WindowsControlRingLocalCallChannel, WindowsControlRingLocalNotificationChannel,
 };
@@ -43,9 +43,10 @@ pub fn connect(control_pipe: &OsStr) -> Result<(BrokerConnection, Option<Process
     let (local, startup, (notifications, process_image)) =
         BrokerLocal::negotiate(setup, |mut setup| {
             let shared_memory = Arc::new(setup.receive_shared_memory(SHARED_BUFFER_POOL_SIZE)?);
-            let control_memory = setup.receive_control_ring()?;
+            let layout = ControlRingLayout::MULTI_LANE;
+            let control_memory = setup.receive_control_ring(layout)?;
             let process_image = setup.receive_process_image()?;
-            let control_ring = ControlRing::new(control_memory).map_err(|error| {
+            let control_ring = ControlRing::new(control_memory, layout).map_err(|error| {
                 std::io::Error::new(
                     std::io::ErrorKind::InvalidData,
                     format!("invalid broker control ring: {error:?}"),

litebox_broker_local_vm_userland/src/lib.rs

@@ -40,10 +40,9 @@ use litebox_broker_protocol::process::ProcessStartupData;
 use litebox_broker_protocol::wire::{self, WireError};
 use litebox_broker_transport::channel::{LocalCallChannel, LocalSetupChannel};
 use litebox_broker_transport::control_ring::{
-    CONTROL_RING_MEMORY_SIZE, CONTROL_RING_NOTIFICATION_SLOT_COUNT, ControlRing,
-    ControlRingConsumer, ControlRingDirection, ControlRingError, ControlRingProducer,
-    ControlRingReadError, ControlRingReadStatus, ControlRingWriteStatus, LocalControlRingEndpoints,
-    MemoryAccessPolicy, WaitableSharedMemory,
+    CONTROL_RING_NOTIFICATION_SLOT_COUNT, ControlRing, ControlRingConsumer, ControlRingError,
+    ControlRingLayout, ControlRingProducer, ControlRingReadError, ControlRingReadStatus,
+    ControlRingWriteStatus, LocalControlRingEndpoints, WaitableSharedMemory,
 };
 use litebox_broker_transport::shared_memory::{ControlRingMemory, SharedMemory, SharedMemoryError};
 use litebox_common_vm_abi::{
@@ -232,8 +231,10 @@ impl SharedMemory for KernelSharedMemory {
     }
 }
 
-/// The control ring at the start of its region. Accesses keep to
-/// [`MemoryAccessPolicy::ControlRing`].
+/// The layout of the control ring the kernel serves.
+const LAYOUT: ControlRingLayout = ControlRingLayout::SINGLE_LANE;
+
+/// The control ring at the start of its region. Accesses keep to [`LAYOUT`].
 struct KernelControlRing {
     start: usize,
 }
@@ -242,7 +243,7 @@ impl KernelControlRing {
     /// `None` if `region` cannot hold the ring.
     fn new(region: UserRange) -> Option<Self> {
         let start: usize = region.start.trunc();
-        (region.len >= CONTROL_RING_MEMORY_SIZE as u64 && start.is_multiple_of(size_of::<u64>()))
+        (region.len >= LAYOUT.memory_size() as u64 && start.is_multiple_of(size_of::<u64>()))
             .then_some(Self { start })
     }
 
@@ -257,26 +258,26 @@ impl KernelControlRing {
     }
 
     fn u32_at(&self, offset: usize) -> Result<*mut u32, SharedMemoryError> {
-        self.word(offset, MemoryAccessPolicy::ControlRing.permits_u32(offset))
+        self.word(offset, LAYOUT.permits_u32(offset))
     }
 
     fn u64_at(&self, offset: usize) -> Result<*mut u64, SharedMemoryError> {
-        self.word(offset, MemoryAccessPolicy::ControlRing.permits_u64(offset))
+        self.word(offset, LAYOUT.permits_u64(offset))
     }
 
     fn bytes_at(&self, offset: usize, len: usize) -> Result<usize, SharedMemoryError> {
-        if !MemoryAccessPolicy::ControlRing.permits_byte_range(offset, len) {
+        if !LAYOUT.permits_byte_range(offset, len) {
             return Err(SharedMemoryError::InvalidRange);
         }
-        checked(self.start, CONTROL_RING_MEMORY_SIZE, offset, len)
+        checked(self.start, LAYOUT.memory_size(), offset, len)
     }
 }
 
 // Safety (below): in bounds and aligned (checked), in a region mapped for the
 // process's lifetime and disjoint from private memory.
 impl SharedMemory for KernelControlRing {
     fn len(&self) -> usize {
-        CONTROL_RING_MEMORY_SIZE
+        LAYOUT.memory_size()
     }
 
     fn read(&self, offset: usize, destination: &mut [u8]) -> Result<(), SharedMemoryError> {
@@ -395,12 +396,16 @@ pub fn connect(
                     .ok_or(ChannelError::Memory(SharedMemoryError::InvalidRange))
             };
             let LocalControlRingEndpoints {
-                request_producer,
-                response_consumer,
+                lanes,
+                lane_demand: _,
                 notification_consumer,
-            } = ControlRing::new(ring()?)
+            } = ControlRing::new(ring()?, LAYOUT)
                 .map_err(ChannelError::Ring)?
                 .into_local();
+            let [lane] = <[_; 1]>::try_from(lanes)
+                .unwrap_or_else(|_| unreachable!("a single-lane layout has one lane"));
+            let (request_producer, response_consumer) =
+                (lane.request_producer, lane.response_consumer);
             let notifications = Notifications {
                 consumer: notification_consumer,
                 words: ring()?,
@@ -442,7 +447,7 @@ impl Notifications {
             if received != 0 {
                 return Ok(received);
             }
-            let epoch = ControlRingDirection::Notifications.producer_epoch_offset();
+            let epoch = LAYOUT.notifications().producer_epoch_offset();
             match broker_wait(self.words.start, epoch, wait_epoch, deadline) {
                 Ok(()) => {}
                 Err(Status::TimedOut) => return Ok(0),
@@ -661,7 +666,7 @@ fn futex(
 /// Has the kernel serve published broker requests: a wake on the requests'
 /// producer epoch of the control ring at `start`.
 fn broker_wake(start: usize) -> Result<(), Status> {
-    let offset = ControlRingDirection::Requests.producer_epoch_offset();
+    let offset = LAYOUT.requests(0).producer_epoch_offset();
     futex(start, |addr| FutexRequest::wake(addr, 1), offset)
 }
 
@@ -691,8 +696,8 @@ fn broker_wait(
 ///
 /// On a failed check.
 pub fn self_check(notifications: &Notifications, platform: &VmUserland) {
-    let epoch = ControlRingDirection::Notifications.producer_epoch_offset();
-    let served = ControlRingDirection::Requests.consumer_epoch_offset();
+    let epoch = LAYOUT.notifications().producer_epoch_offset();
+    let served = LAYOUT.requests(0).consumer_epoch_offset();
     let now = || platform.now();
     let current = |offset| {
         notifications
@@ -745,19 +750,19 @@ mod tests {
     fn ring(memory: &mut [u64]) -> (Notifications, ControlRingProducer<KernelControlRing>) {
         let region = UserRange {
             start: memory.as_mut_ptr() as u64,
-            len: CONTROL_RING_MEMORY_SIZE as u64,
+            len: LAYOUT.memory_size() as u64,
         };
         let ring = || KernelControlRing::new(region).expect("room for the ring");
         let notifications = Notifications {
-            consumer: ControlRing::new(ring())
+            consumer: ControlRing::new(ring(), LAYOUT)
                 .expect("ring")
                 .into_local()
                 .notification_consumer,
             words: ring(),
         };
-        let producer = ControlRing::new(ring())
+        let producer = ControlRing::new(ring(), LAYOUT)
             .expect("ring")
-            .into_broker()
+            .into_host()
             .notification_producer;
         (notifications, producer)
     }
@@ -770,7 +775,7 @@ mod tests {
     }
 
     fn ring_memory() -> Vec<u64> {
-        alloc::vec![0u64; CONTROL_RING_MEMORY_SIZE / size_of::<u64>()]
+        alloc::vec![0u64; LAYOUT.memory_size() / size_of::<u64>()]
     }
 
     #[test]

litebox_broker_transport/Cargo.toml

@@ -3,6 +3,10 @@ name = "litebox_broker_transport"
 version = "0.1.0"
 edition = "2024"
 
+[features]
+# Local call-lane pool.
+std = []
+
 [dependencies]
 litebox_broker_protocol = { path = "../litebox_broker_protocol", version = "0.1.0" }
 litebox_platform = { path = "../litebox_platform", version = "0.1.0" }