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, ProtectionFlags,
7    TaskMemoryAccessor,
8};
9use crate::task::dynamic_thread_spawner::SpawnRequestBuilder;
10use crate::task::{CurrentTask, SimpleWaiter, Task, WaitQueue};
11use crate::vfs::eventfd::EventFdFileObject;
12use crate::vfs::syscalls::IocbPtr;
13use crate::vfs::{
14    FdNumber, FileHandle, InputBuffer, OutputBuffer, UserBuffersInputBuffer,
15    UserBuffersOutputBuffer, VecInputBuffer, VecOutputBuffer, WeakFileHandle,
16    checked_add_offset_and_length,
17};
18use smallvec::smallvec;
19use starnix_logging::track_stub;
20use starnix_sync::{AioEventsLock, AioPendingOperationsLock, InterruptibleEvent, LockDepMutex};
21use starnix_syscalls::SyscallResult;
22use starnix_types::user_buffer::{UserBuffer, UserBuffers};
23use starnix_uapi::errors::{EINTR, ETIMEDOUT, Errno};
24use starnix_uapi::{
25    IOCB_CMD_PREAD, IOCB_CMD_PREADV, IOCB_CMD_PWRITE, IOCB_CMD_PWRITEV, IOCB_FLAG_RESFD,
26    aio_context_t, errno, error, io_event, iocb,
27};
28use std::collections::VecDeque;
29use std::sync::{Arc, Weak};
30use zerocopy::IntoBytes;
31
32/// From aio.go in gVisor.
33const AIO_RING_SIZE: usize = 32;
34
35/// Kernel state-machine-based implementation of asynchronous I/O.
36/// See https://man7.org/linux/man-pages/man7/aio.7.html#NOTES
37pub struct AioContext {
38    inner: Arc<AioContextInner>,
39}
40
41impl AioContext {
42    pub fn create(
43        current_task: &CurrentTask,
44        max_operations: usize,
45    ) -> Result<aio_context_t, Errno> {
46        let context = Arc::new(AioContext { inner: AioContextInner::new(max_operations) });
47        context.inner.spawn_worker_for(current_task, WorkerType::Read);
48        context.inner.spawn_worker_for(current_task, WorkerType::Write);
49        let context_addr = current_task.mm()?.map_anonymous(
50            DesiredAddress::Any,
51            AIO_RING_SIZE,
52            ProtectionFlags::READ | ProtectionFlags::WRITE,
53            MappingOptions::ANONYMOUS | MappingOptions::DONT_EXPAND,
54            MappingName::AioContext(context),
55        )?;
56        Ok(context_addr.ptr() as aio_context_t)
57    }
58
59    pub fn get_events(
60        &self,
61        current_task: &CurrentTask,
62        min_results: usize,
63        max_results: usize,
64        deadline: zx::MonotonicInstant,
65    ) -> Result<Vec<io_event>, Errno> {
66        self.inner.get_events(current_task, min_results, max_results, deadline)
67    }
68
69    pub fn submit(
70        self: &Arc<Self>,
71        current_task: &CurrentTask,
72        control_block: iocb,
73        iocb_addr: IocbPtr,
74    ) -> Result<(), Errno> {
75        self.inner.submit(current_task, control_block, iocb_addr)
76    }
77
78    pub fn cancel(
79        self: &Arc<Self>,
80        _current_task: &CurrentTask,
81        control_block: iocb,
82        iocb_addr: IocbPtr,
83    ) -> Result<(), Errno> {
84        self.inner.cancel(control_block, iocb_addr)
85    }
86}
87
88impl std::fmt::Debug for AioContext {
89    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
90        f.debug_struct("AioContext").finish()
91    }
92}
93
94impl std::cmp::PartialEq for AioContext {
95    fn eq(&self, other: &AioContext) -> bool {
96        Arc::ptr_eq(&self.inner, &other.inner)
97    }
98}
99
100impl std::cmp::Eq for AioContext {}
101
102impl Drop for AioContext {
103    fn drop(&mut self) {
104        self.inner.stop();
105    }
106}
107
108struct AioContextInner {
109    operations: OperationQueue,
110    results: ResultQueue,
111    write_worker_done: zx::Event,
112}
113
114impl AioContextInner {
115    fn new(max_operations: usize) -> Arc<Self> {
116        Arc::new(Self {
117            operations: OperationQueue::new(max_operations),
118            results: Default::default(),
119            write_worker_done: zx::Event::create(),
120        })
121    }
122
123    fn stop(&self) {
124        self.operations.stop();
125        let _ =
126            self.write_worker_done.wait_one(zx::Signals::USER_0, zx::MonotonicInstant::INFINITE);
127    }
128
129    fn get_events(
130        &self,
131        current_task: &CurrentTask,
132        min_results: usize,
133        max_results: usize,
134        deadline: zx::MonotonicInstant,
135    ) -> Result<Vec<io_event>, Errno> {
136        let mut events = self.results.dequeue(max_results);
137        if events.len() >= min_results {
138            return Ok(events);
139        }
140        let event = InterruptibleEvent::new();
141        loop {
142            let (mut waiter, guard) = SimpleWaiter::new(&event);
143            self.results.waiters.wait_async_simple(&mut waiter);
144            events.extend(self.results.dequeue(max_results - events.len()));
145            if events.len() >= min_results {
146                return Ok(events);
147            }
148            match current_task.block_until(guard, deadline) {
149                Err(err) if err == ETIMEDOUT => {
150                    return Ok(events);
151                }
152                Err(err) if err == EINTR => {
153                    if events.is_empty() {
154                        Err(err)
155                    } else {
156                        return Ok(events);
157                    }
158                }
159                result => result,
160            }?;
161        }
162    }
163
164    fn submit(
165        self: &Arc<Self>,
166        current_task: &CurrentTask,
167        control_block: iocb,
168        iocb_addr: IocbPtr,
169    ) -> Result<(), Errno> {
170        let op = IoOperation::new(current_task, control_block, iocb_addr)?;
171        self.operations.enqueue(op)
172    }
173
174    fn cancel(self: &Arc<Self>, control_block: iocb, iocb_addr: IocbPtr) -> Result<(), Errno> {
175        let op_type: OpType = (control_block.aio_lio_opcode as u32).try_into()?;
176        self.operations.remove(op_type.worker_type(), iocb_addr)
177    }
178
179    fn spawn_worker_for(self: &Arc<Self>, current_task: &CurrentTask, worker_type: WorkerType) {
180        let creds = current_task.current_creds().clone();
181        let inner = self.clone();
182        let closure = move |current_task: &CurrentTask| {
183            current_task
184                .override_creds(creds, || inner.perform_next_action(current_task, worker_type))
185        };
186        let req = SpawnRequestBuilder::new()
187            .with_debug_name("aio-worker")
188            .with_sync_closure(closure)
189            .build();
190        current_task.kernel().kthreads.spawner().spawn_from_request(req);
191    }
192
193    fn perform_next_action(&self, current_task: &CurrentTask, worker_type: WorkerType) {
194        while let Ok(IoAction::Op(mut op)) =
195            self.operations.block_until_dequeue(current_task, worker_type)
196        {
197            let Some(result) = op.execute(current_task) else {
198                continue;
199            };
200            self.results.enqueue(op.complete(result));
201
202            if let Some(eventfd) = op.eventfd {
203                if let Some(eventfd) = eventfd.upgrade() {
204                    let mut input_buffer = VecInputBuffer::new(1u64.as_bytes());
205                    let _ = eventfd.write(current_task, &mut input_buffer);
206                }
207            }
208        }
209        if matches!(worker_type, WorkerType::Write) {
210            let _ = self.write_worker_done.signal(zx::Signals::NONE, zx::Signals::USER_0);
211        }
212    }
213}
214
215#[derive(Debug, Clone, Copy)]
216enum WorkerType {
217    Read,
218    Write,
219}
220
221#[derive(Debug, Clone, Copy)]
222enum OpType {
223    PRead,
224    PReadV,
225    // TODO: IOCB_CMD_FSYNC
226    // TODO: IOCB_CMD_FDSYNC
227    // TODO: IOCB_CMD_POLL
228    // TODO: IOCB_CMD_NOOP
229    PWrite,
230    PWriteV,
231}
232
233impl OpType {
234    fn worker_type(self) -> WorkerType {
235        match self {
236            OpType::PRead | OpType::PReadV => WorkerType::Read,
237            OpType::PWrite | OpType::PWriteV => WorkerType::Write,
238        }
239    }
240}
241
242impl TryFrom<u32> for OpType {
243    type Error = Errno;
244
245    fn try_from(opcode: u32) -> Result<Self, Self::Error> {
246        match opcode {
247            IOCB_CMD_PREAD => Ok(Self::PRead),
248            IOCB_CMD_PREADV => Ok(Self::PReadV),
249            IOCB_CMD_PWRITE => Ok(Self::PWrite),
250            IOCB_CMD_PWRITEV => Ok(Self::PWriteV),
251            _ => {
252                track_stub!(TODO("https://fxbug.dev/297433877"), "io_submit opcode", opcode);
253                return error!(ENOSYS);
254            }
255        }
256    }
257}
258
259enum IoOperationType {
260    Read { file: WeakFileHandle, task: Weak<Task>, buffers: UserBuffers },
261    Write { file: FileHandle, buffer: Result<VecInputBuffer, Errno> },
262}
263
264struct IoOperation {
265    op: IoOperationType,
266    offset: usize,
267    id: u64,
268    iocb_addr: IocbPtr,
269    eventfd: Option<WeakFileHandle>,
270}
271
272impl IoOperation {
273    fn new(
274        current_task: &CurrentTask,
275        control_block: iocb,
276        iocb_addr: IocbPtr,
277    ) -> Result<Self, Errno> {
278        if control_block.aio_reserved2 != 0 {
279            return error!(EINVAL);
280        }
281        let file = current_task.files().get(FdNumber::from_raw(control_block.aio_fildes as i32))?;
282        let op_type = (control_block.aio_lio_opcode as u32).try_into()?;
283        let offset = control_block.aio_offset.try_into().map_err(|_| errno!(EINVAL))?;
284        let flags = control_block.aio_flags;
285
286        match op_type {
287            OpType::PRead | OpType::PReadV => {
288                if !file.can_read() {
289                    return error!(EBADF);
290                }
291            }
292            OpType::PWrite | OpType::PWriteV => {
293                if !file.can_write() {
294                    return error!(EBADF);
295                }
296            }
297        }
298        let mut buffers = match op_type {
299            OpType::PRead | OpType::PWrite => smallvec![UserBuffer {
300                address: control_block.aio_buf.into(),
301                length: control_block.aio_nbytes as usize,
302            }],
303            OpType::PReadV | OpType::PWriteV => {
304                let iovec_addr = IOVecPtr::new(current_task, control_block.aio_buf);
305                let count: i32 = control_block.aio_nbytes.try_into().map_err(|_| errno!(EINVAL))?;
306                current_task.read_iovec(iovec_addr, count.into())?
307            }
308        };
309
310        // Validate the user buffers and offset synchronously.
311        let buffer_length = UserBuffer::cap_buffers_to_max_rw_count(
312            current_task.maximum_valid_address().ok_or_else(|| errno!(EINVAL))?,
313            &mut buffers,
314        )?;
315        checked_add_offset_and_length(offset, buffer_length)?;
316
317        let eventfd = if flags & IOCB_FLAG_RESFD != 0 {
318            let eventfd =
319                current_task.files().get(FdNumber::from_raw(control_block.aio_resfd as i32))?;
320            if eventfd.downcast_file::<EventFdFileObject>().is_none() {
321                return error!(EINVAL);
322            }
323            Some(Arc::downgrade(&eventfd))
324        } else {
325            None
326        };
327
328        let op = match op_type {
329            OpType::PRead | OpType::PReadV => IoOperationType::Read {
330                file: Arc::downgrade(&file),
331                task: Arc::downgrade(&current_task.task),
332                buffers,
333            },
334            OpType::PWrite | OpType::PWriteV => {
335                let buffer = UserBuffersInputBuffer::syscall_new(current_task, buffers)
336                    .and_then(|mut source| source.read_all())
337                    .map(VecInputBuffer::from);
338                IoOperationType::Write { file, buffer }
339            }
340        };
341
342        Ok(IoOperation { op, offset, id: control_block.aio_data, iocb_addr, eventfd })
343    }
344
345    fn worker_type(&self) -> WorkerType {
346        match &self.op {
347            IoOperationType::Read { .. } => WorkerType::Read,
348            IoOperationType::Write { .. } => WorkerType::Write,
349        }
350    }
351
352    fn execute(&mut self, current_task: &CurrentTask) -> Option<Result<SyscallResult, Errno>> {
353        let offset = self.offset;
354        let result = match &mut self.op {
355            IoOperationType::Read { file, task, buffers } => {
356                let Some(file) = file.upgrade() else {
357                    // The FileHandle can close while async IO operations are ongoing.
358                    // Ignore this operation when this happens.
359                    return None;
360                };
361                Self::do_read(current_task, file, task, buffers, offset).map(Into::into)
362            }
363            IoOperationType::Write { file, buffer } => match buffer {
364                Ok(input_buffer) => {
365                    Self::do_write(current_task, file, offset, input_buffer).map(Into::into)
366                }
367                Err(errno) => Err(errno.clone()),
368            },
369        };
370        Some(result)
371    }
372
373    fn complete(&self, result: Result<SyscallResult, Errno>) -> io_event {
374        let res = match result {
375            Ok(return_value) => return_value.value() as i64,
376            Err(errno) => errno.return_value() as i64,
377        };
378
379        io_event { data: self.id, obj: self.iocb_addr.addr().into(), res, ..Default::default() }
380    }
381
382    fn do_read(
383        current_task: &CurrentTask,
384        file: FileHandle,
385        task: &Weak<Task>,
386        buffers: &UserBuffers,
387        offset: usize,
388    ) -> Result<usize, Errno> {
389        let mut output_buffer = {
390            let task = task.upgrade().ok_or_else(|| errno!(EFAULT))?;
391            let sink = UserBuffersOutputBuffer::syscall_new(&task, buffers.clone())?;
392            VecOutputBuffer::new(sink.available())
393        };
394
395        if file.ops().is_seekable() {
396            file.read_at(current_task, offset, &mut output_buffer)?;
397        } else {
398            file.read(current_task, &mut output_buffer)?;
399        }
400
401        let task = task.upgrade().ok_or_else(|| errno!(EFAULT))?;
402        let mut sink = UserBuffersOutputBuffer::syscall_new(&task, buffers.clone())?;
403        sink.write(&output_buffer.data())
404    }
405
406    fn do_write(
407        current_task: &CurrentTask,
408        file: &FileHandle,
409        offset: usize,
410        input_buffer: &mut dyn InputBuffer,
411    ) -> Result<usize, Errno> {
412        if file.ops().is_seekable() {
413            file.write_at(current_task, offset, input_buffer)
414        } else {
415            file.write(current_task, input_buffer)
416        }
417    }
418}
419
420enum IoAction {
421    Op(IoOperation),
422    Stop,
423}
424
425#[derive(Default)]
426struct PendingOperations {
427    is_stopped: bool,
428    // We currently queue the read and write operations to separate threads.
429    // That behavior is incorrect, but it keeps our clients working well enough while we work on
430    // getting the correct parallelism.
431    read_ops: VecDeque<IoOperation>,
432    write_ops: VecDeque<IoOperation>,
433}
434
435impl PendingOperations {
436    fn ops_for(&mut self, worker_type: WorkerType) -> &mut VecDeque<IoOperation> {
437        match worker_type {
438            WorkerType::Read => &mut self.read_ops,
439            WorkerType::Write => &mut self.write_ops,
440        }
441    }
442
443    fn ops_len(&self) -> usize {
444        self.read_ops.len() + self.write_ops.len()
445    }
446}
447
448struct OperationQueue {
449    max_operations: usize,
450    pending: LockDepMutex<PendingOperations, AioPendingOperationsLock>,
451    read_waiters: WaitQueue,
452    write_waiters: WaitQueue,
453}
454
455impl OperationQueue {
456    fn new(max_operations: usize) -> Self {
457        Self {
458            max_operations,
459            pending: Default::default(),
460            read_waiters: Default::default(),
461            write_waiters: Default::default(),
462        }
463    }
464
465    fn waiters_for(&self, worker_type: WorkerType) -> &WaitQueue {
466        match worker_type {
467            WorkerType::Read => &self.read_waiters,
468            WorkerType::Write => &self.write_waiters,
469        }
470    }
471
472    fn enqueue(&self, op: IoOperation) -> Result<(), Errno> {
473        let worker_type = op.worker_type();
474        {
475            let mut pending = self.pending.lock();
476            if pending.is_stopped {
477                return error!(EINVAL);
478            }
479            if pending.ops_len() >= self.max_operations {
480                return error!(EAGAIN);
481            }
482            pending.ops_for(worker_type).push_back(op);
483        }
484        self.waiters_for(worker_type).notify_unordered_count(1);
485        Ok(())
486    }
487
488    fn stop(&self) {
489        {
490            let mut pending = self.pending.lock();
491            pending.is_stopped = true;
492            pending.read_ops.clear();
493        }
494        self.read_waiters.notify_all();
495        self.write_waiters.notify_all();
496    }
497
498    fn dequeue(&self, worker_type: WorkerType) -> Option<IoAction> {
499        let mut pending = self.pending.lock();
500        match worker_type {
501            WorkerType::Read => {
502                if pending.is_stopped {
503                    return Some(IoAction::Stop);
504                }
505                pending.read_ops.pop_front().map(IoAction::Op)
506            }
507            WorkerType::Write => {
508                if let Some(op) = pending.write_ops.pop_front() {
509                    return Some(IoAction::Op(op));
510                }
511                if pending.is_stopped {
512                    return Some(IoAction::Stop);
513                }
514                None
515            }
516        }
517    }
518
519    fn remove(&self, worker_type: WorkerType, iocb_addr: IocbPtr) -> Result<(), Errno> {
520        {
521            let mut pending = self.pending.lock();
522            if pending.is_stopped {
523                return error!(EINVAL);
524            }
525            // TODO: Use pop_front_if when available.
526            if let Some(idx) = pending
527                .ops_for(worker_type)
528                .iter()
529                .position(|value| value.iocb_addr.addr() == iocb_addr.addr())
530            {
531                let _ = pending.ops_for(worker_type).remove(idx);
532            } else {
533                return error!(EAGAIN);
534            }
535        }
536        Ok(())
537    }
538
539    fn block_until_dequeue(
540        &self,
541        current_task: &CurrentTask,
542        worker_type: WorkerType,
543    ) -> Result<IoAction, Errno> {
544        if let Some(action) = self.dequeue(worker_type) {
545            return Ok(action);
546        }
547        loop {
548            let event = InterruptibleEvent::new();
549            let (mut waiter, guard) = SimpleWaiter::new(&event);
550            self.waiters_for(worker_type).wait_async_simple(&mut waiter);
551            if let Some(action) = self.dequeue(worker_type) {
552                return Ok(action);
553            }
554            match current_task.block_until(guard, zx::MonotonicInstant::INFINITE) {
555                Err(err) if err == EINTR => continue,
556                result => result?,
557            }
558        }
559    }
560}
561
562#[derive(Default)]
563struct ResultQueue {
564    waiters: WaitQueue,
565    events: LockDepMutex<VecDeque<io_event>, AioEventsLock>,
566}
567
568impl ResultQueue {
569    fn enqueue(&self, event: io_event) {
570        self.events.lock().push_back(event);
571        self.waiters.notify_unordered_count(1);
572    }
573
574    fn dequeue(&self, limit: usize) -> Vec<io_event> {
575        let mut events = self.events.lock();
576        let len = std::cmp::min(events.len(), limit);
577        events.drain(..len).collect()
578    }
579}
580
581#[cfg(test)]
582mod tests {
583    use super::*;
584    use crate::mm::{MemoryAccessor, PAGE_SIZE};
585    use crate::testing::{map_memory, spawn_kernel_and_run};
586    use crate::vfs::{
587        Anon, FdFlags, FileObject, FileOps, fileops_impl_noop_sync, fileops_impl_seekless,
588    };
589    use starnix_sync::Mutex;
590    use starnix_uapi::open_flags::OpenFlags;
591    use starnix_uapi::user_address::UserAddress;
592
593    struct TestSinkFile {
594        written: Arc<Mutex<Vec<u8>>>,
595    }
596
597    impl FileOps for TestSinkFile {
598        fileops_impl_seekless!();
599        fileops_impl_noop_sync!();
600
601        fn read(
602            &self,
603            _file: &FileObject,
604            _current_task: &CurrentTask,
605            _offset: usize,
606            _data: &mut dyn OutputBuffer,
607        ) -> Result<usize, Errno> {
608            error!(EINVAL)
609        }
610
611        fn write(
612            &self,
613            _file: &FileObject,
614            _current_task: &CurrentTask,
615            _offset: usize,
616            data: &mut dyn InputBuffer,
617        ) -> Result<usize, Errno> {
618            let bytes = data.read_all()?;
619            let len = bytes.len();
620            self.written.lock().extend_from_slice(&bytes);
621            Ok(len)
622        }
623    }
624
625    #[::fuchsia::test]
626    async fn test_pwrite_completes_after_fd_closed_and_buffer_unmapped() {
627        spawn_kernel_and_run(async |current_task| {
628            let header = b"A_WRTE_HEADER:";
629            let payload = b"restarting adbd as root\n";
630            let user_buf = map_memory(current_task, UserAddress::default(), *PAGE_SIZE);
631            current_task.write_memory(user_buf, header).expect("write header");
632            let payload_buf = (user_buf + header.len()).expect("payload addr");
633            current_task.write_memory(payload_buf, payload).expect("write user payload");
634
635            let iocb_addr1 = map_memory(current_task, UserAddress::default(), *PAGE_SIZE);
636            let iocb_ref1 = IocbPtr::new(current_task, iocb_addr1);
637            let iocb_addr2 = (iocb_addr1 + std::mem::size_of::<iocb>()).expect("iocb2 addr");
638            let iocb_ref2 = IocbPtr::new(current_task, iocb_addr2);
639
640            let written = Arc::new(Mutex::new(Vec::new()));
641            let write_file = Anon::new_private_file(
642                current_task,
643                Box::new(TestSinkFile { written: Arc::clone(&written) }),
644                OpenFlags::WRONLY,
645                "[test-sink]",
646            );
647            let write_fd = current_task.add_file(write_file, FdFlags::empty()).expect("add file");
648
649            let cb1 = iocb {
650                aio_fildes: write_fd.raw() as u32,
651                aio_lio_opcode: IOCB_CMD_PWRITE as u16,
652                aio_buf: user_buf.ptr() as u64,
653                aio_nbytes: header.len() as u64,
654                aio_offset: 0,
655                aio_data: 0x1234,
656                ..Default::default()
657            };
658            let cb2 = iocb {
659                aio_fildes: write_fd.raw() as u32,
660                aio_lio_opcode: IOCB_CMD_PWRITE as u16,
661                aio_buf: payload_buf.ptr() as u64,
662                aio_nbytes: payload.len() as u64,
663                aio_offset: 0,
664                aio_data: 0x1235,
665                ..Default::default()
666            };
667
668            let ctx_id = AioContext::create(current_task, 16).expect("create aio context");
669            let aio_ctx = current_task
670                .mm()
671                .expect("mm")
672                .get_aio_context(ctx_id.into())
673                .expect("get aio context");
674            aio_ctx.submit(current_task, cb1, iocb_ref1).expect("submit header op");
675            aio_ctx.submit(current_task, cb2, iocb_ref2).expect("submit payload op");
676
677            // Simulate adbd exiting immediately after io_submit returns: close the FD table entry,
678            // unmap userspace buffers, and stop the AIO context before waiting for background writes.
679            current_task.files().close(write_fd).expect("close write fd");
680            current_task
681                .mm()
682                .expect("mm")
683                .unmap(user_buf, (*PAGE_SIZE) as usize)
684                .expect("unmap user buf");
685            aio_ctx.inner.stop();
686
687            let mut expected = Vec::new();
688            expected.extend_from_slice(header);
689            expected.extend_from_slice(payload);
690            assert_eq!(written.lock().as_slice(), expected.as_slice());
691        })
692        .await;
693    }
694}
starnix_core/vfs/aio.rs

starnix_core/vfs/
aio.rs
