starnix_core/vfs/
aio.rs

1// Copyright 2024 The Fuchsia Authors. All rights reserved.
2// Use of this source code is governed by a BSD-style license that can be
3// found in the LICENSE file.
4
5use crate::mm::{
6    DesiredAddress, IOVecPtr, MappingName, MappingOptions, MemoryAccessorExt, MemoryManager,
7    ProtectionFlags, TaskMemoryAccessor,
8};
9use crate::task::dynamic_thread_spawner::SpawnRequestBuilder;
10use crate::task::{CurrentTask, SimpleWaiter, Task, TaskFlags, WaitQueue};
11use crate::vfs::eventfd::EventFdFileObject;
12use crate::vfs::syscalls::IocbPtr;
13use crate::vfs::{
14    FdNumber, FileHandle, InputBuffer, UserBuffersInputBuffer, VecInputBuffer, VecOutputBuffer,
15    WeakFileHandle, checked_add_offset_and_length,
16};
17use smallvec::smallvec;
18use starnix_logging::track_stub;
19use starnix_sync::{AioEventsLock, AioPendingOperationsLock, InterruptibleEvent, LockDepMutex};
20use starnix_syscalls::SyscallResult;
21use starnix_types::user_buffer::{UserBuffer, UserBuffers};
22use starnix_uapi::errors::{EINTR, ETIMEDOUT, Errno};
23use starnix_uapi::{
24    IOCB_CMD_PREAD, IOCB_CMD_PREADV, IOCB_CMD_PWRITE, IOCB_CMD_PWRITEV, IOCB_FLAG_RESFD,
25    aio_context_t, errno, error, io_event, iocb,
26};
27use std::collections::VecDeque;
28use std::sync::{Arc, OnceLock, Weak};
29use zerocopy::IntoBytes;
30
31/// From aio.go in gVisor.
32const AIO_RING_SIZE: usize = 32;
33
34/// Kernel state-machine-based implementation of asynchronous I/O.
35/// See https://man7.org/linux/man-pages/man7/aio.7.html#NOTES
36pub struct AioContext {
37    inner: Arc<AioContextInner>,
38}
39
40impl AioContext {
41    pub fn create(
42        current_task: &CurrentTask,
43        max_operations: usize,
44    ) -> Result<aio_context_t, Errno> {
45        let mm = current_task.mm()?;
46        let inner = AioContextInner::new(max_operations);
47        inner.spawn_worker_for(current_task, WorkerType::Read);
48        inner.spawn_worker_for(current_task, WorkerType::Write);
49        let context = Arc::new(AioContext { inner });
50        let context_addr = mm.map_anonymous(
51            DesiredAddress::Any,
52            AIO_RING_SIZE,
53            ProtectionFlags::READ | ProtectionFlags::WRITE,
54            MappingOptions::ANONYMOUS | MappingOptions::DONT_EXPAND,
55            MappingName::AioContext(context),
56        )?;
57        Ok(context_addr.ptr() as aio_context_t)
58    }
59
60    pub fn get_events(
61        &self,
62        current_task: &CurrentTask,
63        min_results: usize,
64        max_results: usize,
65        deadline: zx::MonotonicInstant,
66    ) -> Result<Vec<io_event>, Errno> {
67        self.inner.get_events(current_task, min_results, max_results, deadline)
68    }
69
70    pub fn submit(
71        self: &Arc<Self>,
72        current_task: &CurrentTask,
73        control_block: iocb,
74        iocb_addr: IocbPtr,
75    ) -> Result<(), Errno> {
76        self.inner.submit(current_task, control_block, iocb_addr)
77    }
78
79    pub fn cancel(
80        self: &Arc<Self>,
81        _current_task: &CurrentTask,
82        control_block: iocb,
83        iocb_addr: IocbPtr,
84    ) -> Result<(), Errno> {
85        self.inner.cancel(control_block, iocb_addr)
86    }
87}
88
89impl std::fmt::Debug for AioContext {
90    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
91        f.debug_struct("AioContext").finish()
92    }
93}
94
95impl std::cmp::PartialEq for AioContext {
96    fn eq(&self, other: &AioContext) -> bool {
97        Arc::ptr_eq(&self.inner, &other.inner)
98    }
99}
100
101impl std::cmp::Eq for AioContext {}
102
103impl Drop for AioContext {
104    fn drop(&mut self) {
105        self.inner.stop();
106    }
107}
108
109struct WorkerCompletion {
110    thread_id: OnceLock<std::thread::ThreadId>,
111    done: zx::Event,
112}
113
114impl Default for WorkerCompletion {
115    fn default() -> Self {
116        Self { thread_id: OnceLock::new(), done: zx::Event::create() }
117    }
118}
119
120impl WorkerCompletion {
121    fn record_current_thread(&self) {
122        self.thread_id
123            .set(std::thread::current().id())
124            .expect("worker thread id should only be set once");
125    }
126
127    fn is_current_thread(&self) -> bool {
128        self.thread_id.get() == Some(&std::thread::current().id())
129    }
130
131    fn signal(&self) {
132        let _ = self.done.signal(zx::Signals::NONE, zx::Signals::USER_0);
133    }
134
135    fn wait(&self) {
136        // `AioContext` is owned by `MemoryManager` (via `MappingName::AioContext`) and is normally
137        // dropped by the user task on `io_destroy`, `munmap`, or task exit (`CurrentTask::exit`).
138        // However, because Starnix uses a single `Arc<MemoryManager>` reference count (rather than
139        // separate `mm_users` / `mm_count` with an explicit `exit_aio` step on last user exit), a
140        // worker transiently upgrading `Weak<MemoryManager>` to write into user memory in
141        // `IoOperation::do_read` can race with `CurrentTask::exit` and end up dropping the last
142        // `Arc<MemoryManager>` (and thus `AioContext`) on the worker thread itself. Skip waiting if
143        // `stop()` is running on this worker thread to avoid self-deadlock.
144        if !self.is_current_thread() {
145            let _ = self.done.wait_one(zx::Signals::USER_0, zx::MonotonicInstant::INFINITE);
146        }
147    }
148}
149
150struct AioContextInner {
151    operations: OperationQueue,
152    results: ResultQueue,
153    read_worker: WorkerCompletion,
154    write_worker: WorkerCompletion,
155}
156
157impl AioContextInner {
158    fn new(max_operations: usize) -> Arc<Self> {
159        Arc::new(Self {
160            operations: OperationQueue::new(max_operations),
161            results: Default::default(),
162            read_worker: Default::default(),
163            write_worker: Default::default(),
164        })
165    }
166
167    fn worker_for(&self, worker_type: WorkerType) -> &WorkerCompletion {
168        match worker_type {
169            WorkerType::Read => &self.read_worker,
170            WorkerType::Write => &self.write_worker,
171        }
172    }
173
174    fn stop(&self) {
175        self.operations.stop();
176        self.write_worker.wait();
177        self.read_worker.wait();
178    }
179
180    fn get_events(
181        &self,
182        current_task: &CurrentTask,
183        min_results: usize,
184        max_results: usize,
185        deadline: zx::MonotonicInstant,
186    ) -> Result<Vec<io_event>, Errno> {
187        let mut events = self.results.dequeue(max_results);
188        if events.len() >= min_results {
189            return Ok(events);
190        }
191        let event = InterruptibleEvent::new();
192        loop {
193            let (mut waiter, guard) = SimpleWaiter::new(&event);
194            self.results.waiters.wait_async_simple(&mut waiter);
195            events.extend(self.results.dequeue(max_results - events.len()));
196            if events.len() >= min_results {
197                return Ok(events);
198            }
199            match current_task.block_until(guard, deadline) {
200                Err(err) if err == ETIMEDOUT => {
201                    return Ok(events);
202                }
203                Err(err) if err == EINTR => {
204                    if events.is_empty() {
205                        Err(err)
206                    } else {
207                        return Ok(events);
208                    }
209                }
210                result => result,
211            }?;
212        }
213    }
214
215    fn submit(
216        self: &Arc<Self>,
217        current_task: &CurrentTask,
218        control_block: iocb,
219        iocb_addr: IocbPtr,
220    ) -> Result<(), Errno> {
221        let op = IoOperation::new(current_task, control_block, iocb_addr)?;
222        self.operations.enqueue(op)
223    }
224
225    fn cancel(self: &Arc<Self>, control_block: iocb, iocb_addr: IocbPtr) -> Result<(), Errno> {
226        let op_type: OpType = (control_block.aio_lio_opcode as u32).try_into()?;
227        self.operations.remove(op_type.worker_type(), iocb_addr)
228    }
229
230    fn spawn_worker_for(self: &Arc<Self>, current_task: &CurrentTask, worker_type: WorkerType) {
231        let creds = current_task.current_creds().clone();
232        let inner = self.clone();
233        let closure = move |current_task: &CurrentTask| {
234            let worker = inner.worker_for(worker_type);
235            worker.record_current_thread();
236            let _guard = scopeguard::guard((), |_| {
237                if matches!(worker_type, WorkerType::Read) {
238                    current_task.write().set_flags(TaskFlags::WORK_CANCELLED, false);
239                }
240                worker.signal();
241            });
242            current_task
243                .override_creds(creds, || inner.perform_next_action(current_task, worker_type))
244        };
245        let req = SpawnRequestBuilder::new()
246            .with_debug_name("aio-worker")
247            .with_sync_closure(closure)
248            .build();
249        current_task.kernel().kthreads.spawner().spawn_from_request(req);
250    }
251
252    fn perform_next_action(&self, current_task: &CurrentTask, worker_type: WorkerType) {
253        while let Ok(IoAction::Op(mut op)) =
254            self.operations.block_until_dequeue(current_task, worker_type)
255        {
256            let result = op.execute(current_task);
257            if matches!(worker_type, WorkerType::Read) {
258                // Clear `active_read_task` now that execution is complete so subsequent `stop()`
259                // calls do not interrupt the worker during eventfd notifications or file cleanup.
260                // Also reset `WORK_CANCELLED` if this operation was interrupted, ensuring
261                // `trigger_delayed_releaser()` / `FileOps::close` does not fail with `EINTR`.
262                self.operations.finish_active_read_op(current_task);
263            }
264            if let Some(result) = result {
265                self.results.enqueue(op.complete(result));
266
267                if let Some(eventfd) = op.eventfd.as_ref().and_then(Weak::upgrade) {
268                    let mut input_buffer = VecInputBuffer::new(1u64.as_bytes());
269                    let _ = eventfd.write(current_task, &mut input_buffer);
270                }
271            }
272            drop(op);
273            // Release closed files immediately instead of waiting for the worker thread to exit.
274            current_task.trigger_delayed_releaser();
275        }
276    }
277}
278
279#[derive(Debug, Clone, Copy)]
280enum WorkerType {
281    Read,
282    Write,
283}
284
285#[derive(Debug, Clone, Copy)]
286enum OpType {
287    PRead,
288    PReadV,
289    // TODO: IOCB_CMD_FSYNC
290    // TODO: IOCB_CMD_FDSYNC
291    // TODO: IOCB_CMD_POLL
292    // TODO: IOCB_CMD_NOOP
293    PWrite,
294    PWriteV,
295}
296
297impl OpType {
298    fn worker_type(self) -> WorkerType {
299        match self {
300            OpType::PRead | OpType::PReadV => WorkerType::Read,
301            OpType::PWrite | OpType::PWriteV => WorkerType::Write,
302        }
303    }
304}
305
306impl TryFrom<u32> for OpType {
307    type Error = Errno;
308
309    fn try_from(opcode: u32) -> Result<Self, Self::Error> {
310        match opcode {
311            IOCB_CMD_PREAD => Ok(Self::PRead),
312            IOCB_CMD_PREADV => Ok(Self::PReadV),
313            IOCB_CMD_PWRITE => Ok(Self::PWrite),
314            IOCB_CMD_PWRITEV => Ok(Self::PWriteV),
315            _ => {
316                track_stub!(TODO("https://fxbug.dev/297433877"), "io_submit opcode", opcode);
317                return error!(ENOSYS);
318            }
319        }
320    }
321}
322
323enum IoOperationType {
324    Read { file: WeakFileHandle, mm: Weak<MemoryManager>, buffers: UserBuffers },
325    Write { file: FileHandle, buffer: Result<VecInputBuffer, Errno> },
326}
327
328struct IoOperation {
329    op: IoOperationType,
330    offset: usize,
331    id: u64,
332    iocb_addr: IocbPtr,
333    eventfd: Option<WeakFileHandle>,
334}
335
336impl IoOperation {
337    fn new(
338        current_task: &CurrentTask,
339        control_block: iocb,
340        iocb_addr: IocbPtr,
341    ) -> Result<Self, Errno> {
342        if control_block.aio_reserved2 != 0 {
343            return error!(EINVAL);
344        }
345        let file = current_task.files().get(FdNumber::from_raw(control_block.aio_fildes as i32))?;
346        let op_type = (control_block.aio_lio_opcode as u32).try_into()?;
347        let offset = control_block.aio_offset.try_into().map_err(|_| errno!(EINVAL))?;
348        let flags = control_block.aio_flags;
349
350        match op_type {
351            OpType::PRead | OpType::PReadV => {
352                if !file.can_read() {
353                    return error!(EBADF);
354                }
355            }
356            OpType::PWrite | OpType::PWriteV => {
357                if !file.can_write() {
358                    return error!(EBADF);
359                }
360            }
361        }
362        let mut buffers = match op_type {
363            OpType::PRead | OpType::PWrite => smallvec![UserBuffer {
364                address: control_block.aio_buf.into(),
365                length: control_block.aio_nbytes as usize,
366            }],
367            OpType::PReadV | OpType::PWriteV => {
368                let iovec_addr = IOVecPtr::new(current_task, control_block.aio_buf);
369                let count: i32 = control_block.aio_nbytes.try_into().map_err(|_| errno!(EINVAL))?;
370                current_task.read_iovec(iovec_addr, count.into())?
371            }
372        };
373
374        // Validate the user buffers and offset synchronously.
375        let buffer_length = UserBuffer::cap_buffers_to_max_rw_count(
376            current_task.maximum_valid_address().ok_or_else(|| errno!(EINVAL))?,
377            &mut buffers,
378        )?;
379        checked_add_offset_and_length(offset, buffer_length)?;
380
381        let eventfd = if flags & IOCB_FLAG_RESFD != 0 {
382            let eventfd =
383                current_task.files().get(FdNumber::from_raw(control_block.aio_resfd as i32))?;
384            if eventfd.downcast_file::<EventFdFileObject>().is_none() {
385                return error!(EINVAL);
386            }
387            Some(Arc::downgrade(&eventfd))
388        } else {
389            None
390        };
391
392        let op = match op_type {
393            OpType::PRead | OpType::PReadV => {
394                let mm = current_task.mm()?;
395                IoOperationType::Read {
396                    file: Arc::downgrade(&file),
397                    mm: Arc::downgrade(&mm),
398                    buffers,
399                }
400            }
401            OpType::PWrite | OpType::PWriteV => {
402                let buffer = UserBuffersInputBuffer::syscall_new(current_task, buffers)
403                    .and_then(|mut source| source.read_all())
404                    .map(VecInputBuffer::from);
405                IoOperationType::Write { file, buffer }
406            }
407        };
408
409        Ok(IoOperation { op, offset, id: control_block.aio_data, iocb_addr, eventfd })
410    }
411
412    fn worker_type(&self) -> WorkerType {
413        match &self.op {
414            IoOperationType::Read { .. } => WorkerType::Read,
415            IoOperationType::Write { .. } => WorkerType::Write,
416        }
417    }
418
419    fn execute(&mut self, current_task: &CurrentTask) -> Option<Result<SyscallResult, Errno>> {
420        let offset = self.offset;
421        let result = match &mut self.op {
422            IoOperationType::Read { file, mm, buffers } => {
423                let Some(file) = file.upgrade() else {
424                    // The FileHandle can close while async IO operations are ongoing.
425                    // Ignore this operation when this happens.
426                    return None;
427                };
428                Self::do_read(current_task, file, mm, buffers, offset).map(Into::into)
429            }
430            IoOperationType::Write { file, buffer } => match buffer {
431                Ok(input_buffer) => {
432                    Self::do_write(current_task, file, offset, input_buffer).map(Into::into)
433                }
434                Err(errno) => Err(errno.clone()),
435            },
436        };
437        Some(result)
438    }
439
440    fn complete(&self, result: Result<SyscallResult, Errno>) -> io_event {
441        let res = match result {
442            Ok(return_value) => return_value.value() as i64,
443            Err(errno) => errno.return_value() as i64,
444        };
445
446        io_event { data: self.id, obj: self.iocb_addr.addr().into(), res, ..Default::default() }
447    }
448
449    fn do_read(
450        current_task: &CurrentTask,
451        file: FileHandle,
452        mm: &Weak<MemoryManager>,
453        buffers: &UserBuffers,
454        offset: usize,
455    ) -> Result<usize, Errno> {
456        let mut output_buffer = VecOutputBuffer::new(buffers.iter().map(|b| b.length).sum());
457
458        if file.ops().is_seekable() {
459            file.read_at(current_task, offset, &mut output_buffer)?;
460        } else {
461            file.read(current_task, &mut output_buffer)?;
462        }
463        drop(file);
464
465        let mm = mm.upgrade().ok_or_else(|| errno!(EFAULT))?;
466        let mut bytes = output_buffer.data();
467        let mut bytes_written = 0;
468        for buffer in buffers {
469            if buffer.is_null() {
470                continue;
471            }
472            let bytes_len = std::cmp::min(bytes.len(), buffer.length);
473            let (to_write, remaining) = bytes.split_at(bytes_len);
474            bytes = remaining;
475            bytes_written += mm.syscall_write_memory(buffer.address, to_write)?;
476            if bytes_len < buffer.length {
477                break;
478            }
479        }
480        Ok(bytes_written)
481    }
482
483    fn do_write(
484        current_task: &CurrentTask,
485        file: &FileHandle,
486        offset: usize,
487        input_buffer: &mut dyn InputBuffer,
488    ) -> Result<usize, Errno> {
489        if file.ops().is_seekable() {
490            file.write_at(current_task, offset, input_buffer)
491        } else {
492            file.write(current_task, input_buffer)
493        }
494    }
495}
496
497enum IoAction {
498    Op(IoOperation),
499    Stop,
500}
501
502#[derive(Default)]
503struct PendingOperations {
504    is_stopped: bool,
505    // We currently queue the read and write operations to separate threads.
506    // That behavior is incorrect, but it keeps our clients working well enough while we work on
507    // getting the correct parallelism.
508    read_ops: VecDeque<IoOperation>,
509    write_ops: VecDeque<IoOperation>,
510    active_read_task: Weak<Task>,
511}
512
513impl PendingOperations {
514    fn ops_for(&mut self, worker_type: WorkerType) -> &mut VecDeque<IoOperation> {
515        match worker_type {
516            WorkerType::Read => &mut self.read_ops,
517            WorkerType::Write => &mut self.write_ops,
518        }
519    }
520
521    fn ops_len(&self) -> usize {
522        self.read_ops.len() + self.write_ops.len()
523    }
524}
525
526struct OperationQueue {
527    max_operations: usize,
528    pending: LockDepMutex<PendingOperations, AioPendingOperationsLock>,
529    read_waiters: WaitQueue,
530    write_waiters: WaitQueue,
531}
532
533impl OperationQueue {
534    fn new(max_operations: usize) -> Self {
535        Self {
536            max_operations,
537            pending: Default::default(),
538            read_waiters: Default::default(),
539            write_waiters: Default::default(),
540        }
541    }
542
543    fn waiters_for(&self, worker_type: WorkerType) -> &WaitQueue {
544        match worker_type {
545            WorkerType::Read => &self.read_waiters,
546            WorkerType::Write => &self.write_waiters,
547        }
548    }
549
550    fn enqueue(&self, op: IoOperation) -> Result<(), Errno> {
551        let worker_type = op.worker_type();
552        {
553            let mut pending = self.pending.lock();
554            if pending.is_stopped {
555                return error!(EINVAL);
556            }
557            if pending.ops_len() >= self.max_operations {
558                return error!(EAGAIN);
559            }
560            pending.ops_for(worker_type).push_back(op);
561        }
562        self.waiters_for(worker_type).notify_unordered_count(1);
563        Ok(())
564    }
565
566    fn stop(&self) {
567        {
568            let mut pending = self.pending.lock();
569            pending.is_stopped = true;
570            pending.read_ops.clear();
571            if let Some(task) = std::mem::take(&mut pending.active_read_task).upgrade() {
572                task.write().set_flags(TaskFlags::WORK_CANCELLED, true);
573                task.interrupt();
574            }
575        }
576        self.read_waiters.notify_all();
577        self.write_waiters.notify_all();
578    }
579
580    fn finish_active_read_op(&self, current_task: &CurrentTask) {
581        let mut pending = self.pending.lock();
582        pending.active_read_task = Default::default();
583        current_task.write().set_flags(TaskFlags::WORK_CANCELLED, false);
584    }
585
586    fn dequeue(&self, current_task: &CurrentTask, worker_type: WorkerType) -> Option<IoAction> {
587        let mut pending = self.pending.lock();
588        match worker_type {
589            WorkerType::Read => {
590                if pending.is_stopped {
591                    return Some(IoAction::Stop);
592                }
593                if let Some(op) = pending.read_ops.pop_front() {
594                    pending.active_read_task = current_task.weak_task();
595                    Some(IoAction::Op(op))
596                } else {
597                    None
598                }
599            }
600            WorkerType::Write => {
601                if let Some(op) = pending.write_ops.pop_front() {
602                    return Some(IoAction::Op(op));
603                }
604                if pending.is_stopped {
605                    return Some(IoAction::Stop);
606                }
607                None
608            }
609        }
610    }
611
612    fn remove(&self, worker_type: WorkerType, iocb_addr: IocbPtr) -> Result<(), Errno> {
613        {
614            let mut pending = self.pending.lock();
615            if pending.is_stopped {
616                return error!(EINVAL);
617            }
618            // TODO: Use pop_front_if when available.
619            if let Some(idx) = pending
620                .ops_for(worker_type)
621                .iter()
622                .position(|value| value.iocb_addr.addr() == iocb_addr.addr())
623            {
624                let _ = pending.ops_for(worker_type).remove(idx);
625            } else {
626                return error!(EAGAIN);
627            }
628        }
629        Ok(())
630    }
631
632    fn block_until_dequeue(
633        &self,
634        current_task: &CurrentTask,
635        worker_type: WorkerType,
636    ) -> Result<IoAction, Errno> {
637        if let Some(action) = self.dequeue(current_task, worker_type) {
638            return Ok(action);
639        }
640        loop {
641            let event = InterruptibleEvent::new();
642            let (mut waiter, guard) = SimpleWaiter::new(&event);
643            self.waiters_for(worker_type).wait_async_simple(&mut waiter);
644            if let Some(action) = self.dequeue(current_task, worker_type) {
645                return Ok(action);
646            }
647            match current_task.block_until(guard, zx::MonotonicInstant::INFINITE) {
648                Err(err) if err == EINTR => continue,
649                result => result?,
650            }
651        }
652    }
653}
654
655#[derive(Default)]
656struct ResultQueue {
657    waiters: WaitQueue,
658    events: LockDepMutex<VecDeque<io_event>, AioEventsLock>,
659}
660
661impl ResultQueue {
662    fn enqueue(&self, event: io_event) {
663        self.events.lock().push_back(event);
664        self.waiters.notify_unordered_count(1);
665    }
666
667    fn dequeue(&self, limit: usize) -> Vec<io_event> {
668        let mut events = self.events.lock();
669        let len = std::cmp::min(events.len(), limit);
670        events.drain(..len).collect()
671    }
672}
673
674#[cfg(test)]
675mod tests {
676    use super::*;
677    use crate::mm::{MemoryAccessor, PAGE_SIZE};
678    use crate::task::Waiter;
679    use crate::testing::{map_memory, spawn_kernel_and_run};
680    use crate::vfs::{
681        Anon, FdFlags, FileObject, FileOps, OutputBuffer, fileops_impl_noop_sync,
682        fileops_impl_seekless,
683    };
684    use starnix_sync::Mutex;
685    use starnix_uapi::open_flags::OpenFlags;
686    use starnix_uapi::user_address::{UserAddress, UserRef};
687    use starnix_uapi::{CLONE_SIGHAND, CLONE_THREAD, CLONE_VM};
688
689    struct TestSinkFile {
690        written: Arc<Mutex<Vec<u8>>>,
691    }
692
693    impl FileOps for TestSinkFile {
694        fileops_impl_seekless!();
695        fileops_impl_noop_sync!();
696
697        fn read(
698            &self,
699            _file: &FileObject,
700            _current_task: &CurrentTask,
701            _offset: usize,
702            _data: &mut dyn OutputBuffer,
703        ) -> Result<usize, Errno> {
704            error!(EINVAL)
705        }
706
707        fn write(
708            &self,
709            _file: &FileObject,
710            _current_task: &CurrentTask,
711            _offset: usize,
712            data: &mut dyn InputBuffer,
713        ) -> Result<usize, Errno> {
714            let bytes = data.read_all()?;
715            let len = bytes.len();
716            self.written.lock().extend_from_slice(&bytes);
717            Ok(len)
718        }
719    }
720
721    struct TestBlockedReadFile {
722        read_started: Arc<zx::Event>,
723        file_released: Arc<zx::Event>,
724    }
725
726    impl Drop for TestBlockedReadFile {
727        fn drop(&mut self) {
728            let _ = self.file_released.signal(zx::Signals::NONE, zx::Signals::USER_0);
729        }
730    }
731
732    impl FileOps for TestBlockedReadFile {
733        fileops_impl_seekless!();
734        fileops_impl_noop_sync!();
735
736        fn read(
737            &self,
738            _file: &FileObject,
739            current_task: &CurrentTask,
740            _offset: usize,
741            _data: &mut dyn OutputBuffer,
742        ) -> Result<usize, Errno> {
743            let _ = self.read_started.signal(zx::Signals::NONE, zx::Signals::USER_0);
744            let waiter = Waiter::new();
745            waiter.wait(current_task)?;
746            unreachable!("blocked read should be interrupted on context drop");
747        }
748
749        fn write(
750            &self,
751            _file: &FileObject,
752            _current_task: &CurrentTask,
753            _offset: usize,
754            _data: &mut dyn InputBuffer,
755        ) -> Result<usize, Errno> {
756            error!(EINVAL)
757        }
758    }
759
760    struct TestGatedReadFile {
761        unblock: Arc<zx::Event>,
762        payload: Vec<u8>,
763    }
764
765    impl FileOps for TestGatedReadFile {
766        fileops_impl_seekless!();
767        fileops_impl_noop_sync!();
768
769        fn read(
770            &self,
771            _file: &FileObject,
772            _current_task: &CurrentTask,
773            _offset: usize,
774            data: &mut dyn OutputBuffer,
775        ) -> Result<usize, Errno> {
776            self.unblock
777                .wait_one(zx::Signals::USER_0, zx::MonotonicInstant::INFINITE)
778                .map_err(|_| errno!(EINTR))?;
779            data.write(&self.payload)
780        }
781
782        fn write(
783            &self,
784            _file: &FileObject,
785            _current_task: &CurrentTask,
786            _offset: usize,
787            _data: &mut dyn InputBuffer,
788        ) -> Result<usize, Errno> {
789            error!(EINVAL)
790        }
791    }
792
793    #[::fuchsia::test]
794    async fn test_pwrite_completes_after_fd_closed_and_buffer_unmapped() {
795        spawn_kernel_and_run(async |current_task| {
796            let header = b"A_WRTE_HEADER:";
797            let payload = b"restarting adbd as root\n";
798            let user_buf = map_memory(current_task, UserAddress::default(), *PAGE_SIZE);
799            current_task.write_memory(user_buf, header).expect("write header");
800            let payload_buf = (user_buf + header.len()).expect("payload addr");
801            current_task.write_memory(payload_buf, payload).expect("write user payload");
802
803            let iocb_addr1 = map_memory(current_task, UserAddress::default(), *PAGE_SIZE);
804            let iocb_ref1 = IocbPtr::new(current_task, iocb_addr1);
805            let iocb_addr2 = (iocb_addr1 + std::mem::size_of::<iocb>()).expect("iocb2 addr");
806            let iocb_ref2 = IocbPtr::new(current_task, iocb_addr2);
807
808            let written = Arc::new(Mutex::new(Vec::new()));
809            let write_file = Anon::new_private_file(
810                current_task,
811                Box::new(TestSinkFile { written: Arc::clone(&written) }),
812                OpenFlags::WRONLY,
813                "[test-sink]",
814            );
815            let write_fd = current_task.add_file(write_file, FdFlags::empty()).expect("add file");
816
817            let cb1 = iocb {
818                aio_fildes: write_fd.raw() as u32,
819                aio_lio_opcode: IOCB_CMD_PWRITE as u16,
820                aio_buf: user_buf.ptr() as u64,
821                aio_nbytes: header.len() as u64,
822                aio_offset: 0,
823                aio_data: 0x1234,
824                ..Default::default()
825            };
826            let cb2 = iocb {
827                aio_fildes: write_fd.raw() as u32,
828                aio_lio_opcode: IOCB_CMD_PWRITE as u16,
829                aio_buf: payload_buf.ptr() as u64,
830                aio_nbytes: payload.len() as u64,
831                aio_offset: 0,
832                aio_data: 0x1235,
833                ..Default::default()
834            };
835
836            let ctx_id = AioContext::create(current_task, 16).expect("create aio context");
837            let aio_ctx = current_task
838                .mm()
839                .expect("mm")
840                .get_aio_context(ctx_id.into())
841                .expect("get aio context");
842            aio_ctx.submit(current_task, cb1, iocb_ref1).expect("submit header op");
843            aio_ctx.submit(current_task, cb2, iocb_ref2).expect("submit payload op");
844
845            // Simulate adbd exiting immediately after io_submit returns: close the FD table entry,
846            // unmap userspace buffers, and stop the AIO context before waiting for background writes.
847            current_task.files().close(write_fd).expect("close write fd");
848            current_task
849                .mm()
850                .expect("mm")
851                .unmap(user_buf, (*PAGE_SIZE) as usize)
852                .expect("unmap user buf");
853            aio_ctx.inner.stop();
854
855            let mut expected = Vec::new();
856            expected.extend_from_slice(header);
857            expected.extend_from_slice(payload);
858            assert_eq!(written.lock().as_slice(), expected.as_slice());
859        })
860        .await;
861    }
862
863    #[::fuchsia::test]
864    async fn test_drop_cancels_blocked_read_and_releases_file() {
865        spawn_kernel_and_run(async |current_task| {
866            let user_buf = map_memory(current_task, UserAddress::default(), *PAGE_SIZE);
867            let iocb_addr = map_memory(current_task, UserAddress::default(), *PAGE_SIZE);
868            let iocb_ref = IocbPtr::new(current_task, iocb_addr);
869
870            let read_started = Arc::new(zx::Event::create());
871            let file_released = Arc::new(zx::Event::create());
872            let read_file = Anon::new_private_file(
873                current_task,
874                Box::new(TestBlockedReadFile {
875                    read_started: Arc::clone(&read_started),
876                    file_released: Arc::clone(&file_released),
877                }),
878                OpenFlags::RDONLY,
879                "[test-blocked-read]",
880            );
881            let weak_read_file = Arc::downgrade(&read_file);
882            let read_fd = current_task.add_file(read_file, FdFlags::empty()).expect("add file");
883
884            let cb = iocb {
885                aio_fildes: read_fd.raw() as u32,
886                aio_lio_opcode: IOCB_CMD_PREAD as u16,
887                aio_buf: user_buf.ptr() as u64,
888                aio_nbytes: 64,
889                aio_offset: 0,
890                aio_data: 0x5678,
891                ..Default::default()
892            };
893
894            let ctx_id = AioContext::create(current_task, 16).expect("create aio context");
895            let aio_ctx = current_task
896                .mm()
897                .expect("mm")
898                .get_aio_context(ctx_id.into())
899                .expect("get aio context");
900            aio_ctx.submit(current_task, cb, iocb_ref).expect("submit read op");
901            std::mem::drop(aio_ctx);
902
903            // Wait until the read worker has upgraded its `WeakFileHandle` and entered `read()`.
904            read_started
905                .wait_one(zx::Signals::USER_0, zx::MonotonicInstant::INFINITE)
906                .expect("read started");
907
908            current_task.files().close(read_fd).expect("close read fd");
909            current_task.trigger_delayed_releaser();
910            assert!(
911                weak_read_file.upgrade().is_some(),
912                "blocked read worker should hold strong FileHandle before context teardown"
913            );
914
915            let aio_ctx = current_task
916                .mm()
917                .expect("mm")
918                .destroy_aio_context(ctx_id.into())
919                .expect("destroy aio context");
920            std::mem::drop(aio_ctx);
921
922            file_released
923                .wait_one(zx::Signals::USER_0, zx::MonotonicInstant::INFINITE_PAST)
924                .expect("read file released synchronously before context drop returns");
925
926            assert!(
927                weak_read_file.upgrade().is_none(),
928                "tearing down AioContext must interrupt blocked read and release FileHandle"
929            );
930        })
931        .await;
932    }
933
934    #[::fuchsia::test]
935    async fn test_pread_completes_after_submitting_thread_exits() {
936        spawn_kernel_and_run(async |current_task| {
937            let payload = b"pread-after-submitting-thread-exit";
938            let user_buf = map_memory(current_task, UserAddress::default(), *PAGE_SIZE);
939            let iocb_addr = map_memory(current_task, UserAddress::default(), *PAGE_SIZE);
940            let iocb_ref = IocbPtr::new(current_task, iocb_addr);
941
942            let unblock = Arc::new(zx::Event::create());
943            let read_file = Anon::new_private_file(
944                current_task,
945                Box::new(TestGatedReadFile {
946                    unblock: Arc::clone(&unblock),
947                    payload: payload.to_vec(),
948                }),
949                OpenFlags::RDONLY,
950                "[test-gated-pread]",
951            );
952            let read_fd = current_task.add_file(read_file, FdFlags::empty()).expect("add file");
953
954            let cb = iocb {
955                aio_fildes: read_fd.raw() as u32,
956                aio_lio_opcode: IOCB_CMD_PREAD as u16,
957                aio_buf: user_buf.ptr() as u64,
958                aio_nbytes: payload.len() as u64,
959                aio_offset: 0,
960                aio_data: 0x9abc,
961                ..Default::default()
962            };
963
964            let ctx_id = AioContext::create(current_task, 16).expect("create aio context");
965            let aio_ctx = current_task
966                .mm()
967                .expect("mm")
968                .get_aio_context(ctx_id.into())
969                .expect("get aio context");
970
971            let submitting_thread = current_task
972                .clone_task_for_test((CLONE_THREAD | CLONE_VM | CLONE_SIGHAND) as u64, None);
973            let weak_submitting_task = Arc::downgrade(&submitting_thread.task);
974            aio_ctx.submit(&submitting_thread, cb, iocb_ref).expect("submit pread op");
975            std::mem::drop(submitting_thread);
976            assert!(
977                weak_submitting_task.upgrade().is_none(),
978                "submitting thread Task should be dropped before read completes"
979            );
980
981            unblock.signal(zx::Signals::NONE, zx::Signals::USER_0).expect("signal unblock");
982            let events = aio_ctx
983                .get_events(current_task, 1, 1, zx::MonotonicInstant::INFINITE)
984                .expect("get_events");
985            assert_eq!(events.len(), 1);
986            assert_eq!(events[0].data, 0x9abc);
987            assert_eq!(events[0].res, payload.len() as i64);
988
989            let read_bytes =
990                current_task.read_memory_to_vec(user_buf, payload.len()).expect("read user_buf");
991            assert_eq!(read_bytes.as_slice(), payload);
992        })
993        .await;
994    }
995
996    #[::fuchsia::test]
997    async fn test_preadv_completes_after_submitting_thread_exits() {
998        spawn_kernel_and_run(async |current_task| {
999            let part1 = b"preadv-part1:";
1000            let part2 = b"preadv-part2";
1001            let mut payload = Vec::new();
1002            payload.extend_from_slice(part1);
1003            payload.extend_from_slice(part2);
1004
1005            let part1_buf = map_memory(current_task, UserAddress::default(), *PAGE_SIZE);
1006            let part2_buf = (part1_buf + part1.len()).expect("part2_buf addr");
1007
1008            let iovec_addr = map_memory(current_task, UserAddress::default(), *PAGE_SIZE);
1009            let iovecs = [
1010                UserBuffer { address: part1_buf, length: part1.len() },
1011                UserBuffer { address: part2_buf, length: part2.len() },
1012            ];
1013            current_task
1014                .write_objects(UserRef::new(iovec_addr), &iovecs)
1015                .expect("write iovecs to user memory");
1016
1017            let iocb_addr = map_memory(current_task, UserAddress::default(), *PAGE_SIZE);
1018            let iocb_ref = IocbPtr::new(current_task, iocb_addr);
1019
1020            let unblock = Arc::new(zx::Event::create());
1021            let read_file = Anon::new_private_file(
1022                current_task,
1023                Box::new(TestGatedReadFile {
1024                    unblock: Arc::clone(&unblock),
1025                    payload: payload.clone(),
1026                }),
1027                OpenFlags::RDONLY,
1028                "[test-gated-preadv]",
1029            );
1030            let read_fd = current_task.add_file(read_file, FdFlags::empty()).expect("add file");
1031
1032            let cb = iocb {
1033                aio_fildes: read_fd.raw() as u32,
1034                aio_lio_opcode: IOCB_CMD_PREADV as u16,
1035                aio_buf: iovec_addr.ptr() as u64,
1036                aio_nbytes: iovecs.len() as u64,
1037                aio_offset: 0,
1038                aio_data: 0x9abd,
1039                ..Default::default()
1040            };
1041
1042            let ctx_id = AioContext::create(current_task, 16).expect("create aio context");
1043            let aio_ctx = current_task
1044                .mm()
1045                .expect("mm")
1046                .get_aio_context(ctx_id.into())
1047                .expect("get aio context");
1048
1049            let submitting_thread = current_task
1050                .clone_task_for_test((CLONE_THREAD | CLONE_VM | CLONE_SIGHAND) as u64, None);
1051            let weak_submitting_task = Arc::downgrade(&submitting_thread.task);
1052            aio_ctx.submit(&submitting_thread, cb, iocb_ref).expect("submit preadv op");
1053            std::mem::drop(submitting_thread);
1054            assert!(
1055                weak_submitting_task.upgrade().is_none(),
1056                "submitting thread Task should be dropped before read completes"
1057            );
1058
1059            unblock.signal(zx::Signals::NONE, zx::Signals::USER_0).expect("signal unblock");
1060            let events = aio_ctx
1061                .get_events(current_task, 1, 1, zx::MonotonicInstant::INFINITE)
1062                .expect("get_events");
1063            assert_eq!(events.len(), 1);
1064            assert_eq!(events[0].data, 0x9abd);
1065            assert_eq!(events[0].res, payload.len() as i64);
1066
1067            let read_part1 =
1068                current_task.read_memory_to_vec(part1_buf, part1.len()).expect("read part1_buf");
1069            assert_eq!(read_part1.as_slice(), part1);
1070
1071            let read_part2 =
1072                current_task.read_memory_to_vec(part2_buf, part2.len()).expect("read part2_buf");
1073            assert_eq!(read_part2.as_slice(), part2);
1074        })
1075        .await;
1076    }
1077}
starnix_core/vfs/aio.rs

starnix_core/vfs/
aio.rs
