in the first part, we learned how to build a minimum executor to drive some futures. now let's do that on a 3DS!
the 3ds runs a microkernel-based operating system called Horizon. drivers for peripherals are just slightly special processes: to interface with peripherals, we use Inter-Process Communication (IPC) to send these processes messages.
additionally, the 3ds provides a series of synchronization objects: things that a thread can wait for.
these two things are the basic pieces we need to build an asyncio executor!
so here we go:
we're making a very minimal executor: its job is only to spawn futures and then get told to poll them.
initially, i used IPC for this - including sending rust future objects through IPC, which was a fun challenge - but this has a big problem: sending an IPC request to the same thread that's currently running breaks things very badly. considering futures can ideally wake each other up, this made lockups frequent.
instead, our executor will have three pieces:
here's our queues:
pub static TASK_QUEUE: OnceLock<SyncQueue<BoxFuture<'static, ()>>> = OnceLock::new(); pub static WAKE_QUEUE: OnceLock<SyncQueue<TaskToken>> = OnceLock::new();
these use the 3ds built-in semaphore primitive, to let other tasks signal us to wake up! we can wait on several semaphores at once and get woken up when any one of them is signalled:
let handles = [self.task_semaphore, self.wake_semaphore]; loop { let mut idx = 0; unsafe { svcWaitSynchronizationN(&mut idx, handles.as_ptr(), 2, false, i64::MAX) }; if idx == 1 { // this is the wake semaphore! while let Some(task) = WAKE_QUEUE.get().unwrap().remove() { self.wake(task); } } else if idx == 0 { // this is the spawn semaphore! while let Some(task) = TASK_QUEUE.get().unwrap().remove() { self.spawn(task); } } }
in our previous post, our Waker that we passed to futures was an Arc'd object containing both an id for the current task and a sender channel that linked up to the executor. we can trim this down a lot now!
to make a non-Arc'd waker, we can use the RawWaker type, which is constructed from (a) a pointer to some data, and (b) a set of pointers to functions that take that data as an argument to wake a task.
usually, the pointer is just to an Arc
// snip: <impls for clone_waker, wake_by_ref, drop> pub struct TaskToken(pub u32); unsafe fn wake(data: *const ()) { WAKE_QUEUE.get().unwrap().add(TaskToken(data as u32)); } static VTABLE: RawWakerVTable = RawWakerVTable::new(clone_waker, wake, wake_by_ref, drop); pub fn make_waker(task: TaskToken) -> Waker { unsafe { Waker::from_raw(RawWaker::new(task.0 as *const (), &VTABLE)) } }
we have an executor now! let's make it do something useful, like sleeping.
to implement this, we want to run a thread that:
to run both concurrently, we can do a fun trick. remember how we're using the 3ds' Inter-Process-Communication (IPC) system instead of standard channels?
it turns out, the same system call that listens for IPC requests can also listen for.. anything that can be waited on in the 3ds' operating system? events, mutexes, timers: it can do it all.
struct ReactorHandler { wait_tokens: LiteMap<OSHandle, Vec<TaskToken>>, } impl IPCServerHandler<ReactorCmd, ReactorReply> for ReactorHandler { // handles IPC requests fn handle_request( &mut self, request: ReactorCmd, server: &mut IPCServer<ReactorCmd, ReactorReply>, ) -> ReactorReply { match request { ReactorCmd::AddHandle(task, handle) => { self.wait_tokens.entry(handle).or_default().push(task); server.add_handle_to_list(handle); ReactorReply::Ok } ReactorCmd::PopHandle => todo!(), } } // handles non-IPC (any other synchronization object) fn handle_additional_oshandle( &mut self, handle: OSHandle, _server: &mut IPCServer<ReactorCmd, ReactorReply>, ) { for token in self.wait_tokens.remove(&handle).unwrap() { WAKE_QUEUE.get().unwrap().add(token); } } }
files and sockets is where it gets a little bit more complicated. operations on both are implemented as IPC calls, and these are always synchronous: we can't convert an ongoing operation into a synchronization object that we wait on concurrently with other things; it's more akin to a normal function call.
instead, we use a thread pool: file operations will force a thread to yield if they aren't ready, so we can use them without hogging CPU.
so to do a file operation, we submit it to a thread, alongside the id for our task. the worker then picks it up, and uses the id for our task to wake us up when the operation is done!
more specifically, we send a message containing the file's operating-system handle (a file descriptor, for unix people), the id of our task, a couple of pointers where the worker will store the result of the operation, and a pointer to the buffer where the actual data will be stored.
#[derive(IPCMessage)] #[repr(u32)] pub(crate) enum AsyncFsMsg { Write(#[flatten] FileIoOperation) = 0xA, Read(#[flatten] FileIoOperation) = 0xB, Flush(#[flatten] FileControlOperation) = 0xC, Close(#[flatten] FileControlOperation) = 0xD, } #[derive(IPCSerializable)] pub(crate) struct FileIoOperation { #[normal] pub file: u32, #[normal] pub task: TaskToken, #[normal] pub state: u32, // atomici32 ptr (result code), #[normal] pub result: u32, // atomicu32 ptr (u32::MAX if not done, else bytes read) #[normal] pub offset: u32, #[normal] pub len: u32, #[normal] pub data_ptr: u32, } impl FileIoOperation { fn view<'a>(&'a self) -> FileIoOperationView<'a> { FileIoOperationView { file: ManuallyDrop::new(unsafe { FileHandle::from_raw(self.file) }), task: self.task, state: unsafe { AtomicI32::from_ptr(self.state as *mut i32) }, result: unsafe { AtomicU32::from_ptr(self.result as *mut u32) }, offset: self.offset, data: unsafe { std::slice::from_raw_parts(self.data_ptr as *const u8, self.len as usize) }, } } }
on the worker side:
match task { AsyncFsMsg::Write(op) => { info!("io.worker.{} write fd:{:x}", self.id, op.file); let mut op = op.view(); let res: BunnyResult<usize> = op.file .write(op.offset as u64, op.data, super::WriteOptions::empty()); op.resolve(res); self.executor.wake(op.task).unwrap(); }, _ => ... }
and our future:
impl<'a> Future for Write<'a> { type Output = BunnyResult<usize>; fn poll( mut self: std::pin::Pin<&mut Self>, cx: &mut std::task::Context<'_>, ) -> std::task::Poll<Self::Output> { let bytes_read = self.bytes_written.load(Ordering::Acquire); if bytes_read == u32::MAX && !self.registered { if let Err(e) = self.client.request(&AsyncFsMsg::Write(FileIoOperation { file: self.file.inner.session, task: TaskToken::from_waker(cx.waker()), state: self.state.as_ptr() as u32, result: self.bytes_written.as_ptr() as u32, offset: self.offset, len: self.data.len() as u32, data_ptr: self.data.as_ptr() as u32, })) { Poll::Ready(Err(e.into())) } else { self.registered = true; Poll::Pending } } else if self.registered && bytes_read != u32::MAX { state_resolve(&self.state, bytes_read as usize) } else { Poll::Pending } } }
if we wanted to implement an async-only lock or mutex, that's easy: we just keep a list of tasks that are waiting on the lock. to lock it, we add ourselves to the list; to unlock it, we wake up the next task up on the list if there is one.
making a lock that can interface both synchronous and asynchronous code is a little bit trickier. my approach has been to modify libctru's LightLock, which uses a lightweight synchronization primitive called an AddressArbitrator to signal threads.
the full code is a little bit bulky to put on here, but here's the highlights. if you want to see the full impl, it lives here
impl DsLock { // if we want to use this from a non-async contest, we use TaskToken(u32::MAX) to mark that fn try_lock_and_register(&self, token: TaskToken) -> bool { let state = self.arb.0.load(Ordering::Acquire); if state > 0 { self.arb.0.store(-state, Ordering::Release); true } else { self.arb.0.store(state - 1, Ordering::Release); self.waiters.lock().push(token); false } } unsafe fn unlock(&self) { let lock_state = self.arb.0.load(Ordering::Acquire); if lock_state < 0 { self.arb.0.store(-lock_state, Ordering::Release); let mut waiters = self.waiters.lock(); if waiters.is_empty() { return; } // get the next task to wake up... let task = waiters.remove(0); drop(waiters); if task == TaskToken(u32::MAX) { // it's a non-async one, signal it using the AddressArbitrator let _ = self.arb.signal_one(); } else { // it's an async task, wake it up using the executor executor::WAKE_QUEUE.get().unwrap().add(task); } } } }
here's a tiny example, using egui_citro3d (originally by github.com/LexiBigCheese, with some modifications by me).
it combines an asynchronous TCP socket with a blocking UI task, to display messages sent on the screen!
// full example: https://github.com/kore-signet/bunnyds-egui-example let server = AsyncTcpSocket::bind("0.0.0.0:3027")?; let client = server.accept().await?; let messages: Arc<DSMutex<Vec<String>>> = Arc::new(DSMutex::new(Vec::new())); let messages_tx: Arc<DSMutex<Vec<String>>> = Arc::clone(&messages); let message_task = bunnyds::spawn(async move { let mut rd_buf = [0u8; 1024]; while let Ok(bytes_read) = client.recv(&mut rd_buf).await { if bytes_read == 0 { return; } messages_tx .lock() .await .push(String::from_utf8_lossy(&rd_buf[..bytes_read]).into()); } }); let ui_task = bunnyds::spawn_blocking(Some(0x18), Some(2), move || { let mut hid = Hid::new().unwrap(); let apt = Apt::new().unwrap(); let gfx = Gfx::new().unwrap(); let mut egui_ctx = EguiRenderer::new(&mut hid, &gfx, &apt, &key_mapping); while apt.main_loop() { gfx.wait_for_vblank(); egui_ctx.render_frame( |ui| { egui::CentralPanel::default().show_inside(ui, |ui| { let messages = messages.lock_sync(); for message in messages.iter() { ui.label(message.trim()); } }); }, |_ui| {}, ); } }); futures::future::join(ui_task, message_task).await;
this has been a very fun little adventure! if you have questions about this post or writing rust code on the 3ds, feel free to email me at kore at cat-girl.gay.
you can find the full code for the project at github.com/kore-signet/bunnyds.
an additional side-project of this whole thing has been making a library to abstract the 3ds IPC system into rust primitives, which lives here. if you think an article on that would be fun, let me know!
此内容由惯性聚合(RSS阅读器)自动聚合整理，仅供阅读参考。 原文来自 — 版权归原作者所有。