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}
112
113impl AioContextInner {
114    fn new(max_operations: usize) -> Arc<Self> {
115        Arc::new(Self {
116            operations: OperationQueue::new(max_operations),
117            results: Default::default(),
118        })
119    }
120
121    fn stop(&self) {
122        self.operations.stop();
123    }
124
125    fn get_events(
126        &self,
127        current_task: &CurrentTask,
128        min_results: usize,
129        max_results: usize,
130        deadline: zx::MonotonicInstant,
131    ) -> Result<Vec<io_event>, Errno> {
132        let mut events = self.results.dequeue(max_results);
133        if events.len() >= min_results {
134            return Ok(events);
135        }
136        let event = InterruptibleEvent::new();
137        loop {
138            let (mut waiter, guard) = SimpleWaiter::new(&event);
139            self.results.waiters.wait_async_simple(&mut waiter);
140            events.extend(self.results.dequeue(max_results - events.len()));
141            if events.len() >= min_results {
142                return Ok(events);
143            }
144            match current_task.block_until(guard, deadline) {
145                Err(err) if err == ETIMEDOUT => {
146                    return Ok(events);
147                }
148                Err(err) if err == EINTR => {
149                    if events.is_empty() {
150                        Err(err)
151                    } else {
152                        return Ok(events);
153                    }
154                }
155                result => result,
156            }?;
157        }
158    }
159
160    fn submit(
161        self: &Arc<Self>,
162        current_task: &CurrentTask,
163        control_block: iocb,
164        iocb_addr: IocbPtr,
165    ) -> Result<(), Errno> {
166        let op = IoOperation::new(current_task, control_block, iocb_addr)?;
167        self.operations.enqueue(op)
168    }
169
170    fn cancel(self: &Arc<Self>, control_block: iocb, iocb_addr: IocbPtr) -> Result<(), Errno> {
171        let op_type: OpType = (control_block.aio_lio_opcode as u32).try_into()?;
172        self.operations.remove(op_type.worker_type(), iocb_addr)
173    }
174
175    fn spawn_worker_for(self: &Arc<Self>, current_task: &CurrentTask, worker_type: WorkerType) {
176        let creds = current_task.current_creds().clone();
177        let inner = self.clone();
178        let closure = move |current_task: &CurrentTask| {
179            current_task
180                .override_creds(creds, || inner.perform_next_action(current_task, worker_type))
181        };
182        let req = SpawnRequestBuilder::new()
183            .with_debug_name("aio-worker")
184            .with_sync_closure(closure)
185            .build();
186        current_task.kernel().kthreads.spawner().spawn_from_request(req);
187    }
188
189    fn perform_next_action(&self, current_task: &CurrentTask, worker_type: WorkerType) {
190        while let Ok(IoAction::Op(op)) =
191            self.operations.block_until_dequeue(current_task, worker_type)
192        {
193            let Some(result) = op.execute(current_task) else {
194                return;
195            };
196            self.results.enqueue(op.complete(result));
197
198            if let Some(eventfd) = op.eventfd {
199                if let Some(eventfd) = eventfd.upgrade() {
200                    let mut input_buffer = VecInputBuffer::new(1u64.as_bytes());
201                    let _ = eventfd.write(current_task, &mut input_buffer);
202                }
203            }
204        }
205    }
206}
207
208#[derive(Debug, Clone, Copy)]
209enum WorkerType {
210    Read,
211    Write,
212}
213
214#[derive(Debug, Clone, Copy)]
215enum OpType {
216    PRead,
217    PReadV,
218    // TODO: IOCB_CMD_FSYNC
219    // TODO: IOCB_CMD_FDSYNC
220    // TODO: IOCB_CMD_POLL
221    // TODO: IOCB_CMD_NOOP
222    PWrite,
223    PWriteV,
224}
225
226impl OpType {
227    fn worker_type(self) -> WorkerType {
228        match self {
229            OpType::PRead | OpType::PReadV => WorkerType::Read,
230            OpType::PWrite | OpType::PWriteV => WorkerType::Write,
231        }
232    }
233}
234
235impl TryFrom<u32> for OpType {
236    type Error = Errno;
237
238    fn try_from(opcode: u32) -> Result<Self, Self::Error> {
239        match opcode {
240            IOCB_CMD_PREAD => Ok(Self::PRead),
241            IOCB_CMD_PREADV => Ok(Self::PReadV),
242            IOCB_CMD_PWRITE => Ok(Self::PWrite),
243            IOCB_CMD_PWRITEV => Ok(Self::PWriteV),
244            _ => {
245                track_stub!(TODO("https://fxbug.dev/297433877"), "io_submit opcode", opcode);
246                return error!(ENOSYS);
247            }
248        }
249    }
250}
251struct IoOperation {
252    op_type: OpType,
253    file: WeakFileHandle,
254    task: Weak<Task>,
255    buffers: UserBuffers,
256    offset: usize,
257    id: u64,
258    iocb_addr: IocbPtr,
259    eventfd: Option<WeakFileHandle>,
260}
261
262impl IoOperation {
263    fn new(
264        current_task: &CurrentTask,
265        control_block: iocb,
266        iocb_addr: IocbPtr,
267    ) -> Result<Self, Errno> {
268        if control_block.aio_reserved2 != 0 {
269            return error!(EINVAL);
270        }
271        let file = current_task.files().get(FdNumber::from_raw(control_block.aio_fildes as i32))?;
272        let op_type = (control_block.aio_lio_opcode as u32).try_into()?;
273        let offset = control_block.aio_offset.try_into().map_err(|_| errno!(EINVAL))?;
274        let flags = control_block.aio_flags;
275
276        match op_type {
277            OpType::PRead | OpType::PReadV => {
278                if !file.can_read() {
279                    return error!(EBADF);
280                }
281            }
282            OpType::PWrite | OpType::PWriteV => {
283                if !file.can_write() {
284                    return error!(EBADF);
285                }
286            }
287        }
288        let mut buffers = match op_type {
289            OpType::PRead | OpType::PWrite => smallvec![UserBuffer {
290                address: control_block.aio_buf.into(),
291                length: control_block.aio_nbytes as usize,
292            }],
293            OpType::PReadV | OpType::PWriteV => {
294                let iovec_addr = IOVecPtr::new(current_task, control_block.aio_buf);
295                let count: i32 = control_block.aio_nbytes.try_into().map_err(|_| errno!(EINVAL))?;
296                current_task.read_iovec(iovec_addr, count.into())?
297            }
298        };
299
300        // Validate the user buffers and offset synchronously.
301        let buffer_length = UserBuffer::cap_buffers_to_max_rw_count(
302            current_task.maximum_valid_address().ok_or_else(|| errno!(EINVAL))?,
303            &mut buffers,
304        )?;
305        checked_add_offset_and_length(offset, buffer_length)?;
306
307        let eventfd = if flags & IOCB_FLAG_RESFD != 0 {
308            let eventfd =
309                current_task.files().get(FdNumber::from_raw(control_block.aio_resfd as i32))?;
310            if eventfd.downcast_file::<EventFdFileObject>().is_none() {
311                return error!(EINVAL);
312            }
313            Some(Arc::downgrade(&eventfd))
314        } else {
315            None
316        };
317
318        Ok(IoOperation {
319            op_type,
320            file: Arc::downgrade(&file),
321            task: Arc::downgrade(&current_task.task),
322            buffers,
323            offset,
324            id: control_block.aio_data,
325            iocb_addr,
326            eventfd,
327        })
328    }
329
330    fn execute(&self, current_task: &CurrentTask) -> Option<Result<SyscallResult, Errno>> {
331        let Some(file) = self.file.upgrade() else {
332            // The FileHandle can close while async IO operations are ongoing.
333            // Ignore this operation when this happens.
334            return None;
335        };
336
337        let result = match self.op_type {
338            OpType::PRead | OpType::PReadV => self.do_read(current_task, file).map(Into::into),
339            OpType::PWrite | OpType::PWriteV => self.do_write(current_task, file).map(Into::into),
340        };
341        Some(result)
342    }
343
344    fn complete(&self, result: Result<SyscallResult, Errno>) -> io_event {
345        let res = match result {
346            Ok(return_value) => return_value.value() as i64,
347            Err(errno) => errno.return_value() as i64,
348        };
349
350        io_event { data: self.id, obj: self.iocb_addr.addr().into(), res, ..Default::default() }
351    }
352
353    fn do_read(&self, current_task: &CurrentTask, file: FileHandle) -> Result<usize, Errno> {
354        let mut output_buffer = {
355            let task = self.task.upgrade().ok_or_else(|| errno!(EFAULT))?;
356            let sink = UserBuffersOutputBuffer::syscall_new(&task, self.buffers.clone())?;
357            VecOutputBuffer::new(sink.available())
358        };
359
360        file.read_at(current_task, self.offset, &mut output_buffer)?;
361
362        let task = self.task.upgrade().ok_or_else(|| errno!(EFAULT))?;
363        let mut sink = UserBuffersOutputBuffer::syscall_new(&task, self.buffers.clone())?;
364        sink.write(&output_buffer.data())
365    }
366
367    fn do_write(&self, current_task: &CurrentTask, file: FileHandle) -> Result<usize, Errno> {
368        let mut input_buffer = {
369            let task = self.task.upgrade().ok_or_else(|| errno!(EFAULT))?;
370            let mut source = UserBuffersInputBuffer::syscall_new(&task, self.buffers.clone())?;
371            VecInputBuffer::new(&source.read_all()?)
372        };
373
374        file.write_at(current_task, self.offset, &mut input_buffer)
375    }
376}
377
378enum IoAction {
379    Op(IoOperation),
380    Stop,
381}
382
383#[derive(Default)]
384struct PendingOperations {
385    is_stopped: bool,
386    // We currently queue the read and write operations to separate threads.
387    // That behavior is incorrect, but it keeps our clients working well enough while we work on
388    // getting the correct parallelism.
389    read_ops: VecDeque<IoOperation>,
390    write_ops: VecDeque<IoOperation>,
391}
392
393impl PendingOperations {
394    fn ops_for(&mut self, worker_type: WorkerType) -> &mut VecDeque<IoOperation> {
395        match worker_type {
396            WorkerType::Read => &mut self.read_ops,
397            WorkerType::Write => &mut self.write_ops,
398        }
399    }
400
401    fn ops_len(&self) -> usize {
402        self.read_ops.len() + self.write_ops.len()
403    }
404}
405
406struct OperationQueue {
407    max_operations: usize,
408    pending: LockDepMutex<PendingOperations, AioPendingOperationsLock>,
409    read_waiters: WaitQueue,
410    write_waiters: WaitQueue,
411}
412
413impl OperationQueue {
414    fn new(max_operations: usize) -> Self {
415        Self {
416            max_operations,
417            pending: Default::default(),
418            read_waiters: Default::default(),
419            write_waiters: Default::default(),
420        }
421    }
422
423    fn waiters_for(&self, worker_type: WorkerType) -> &WaitQueue {
424        match worker_type {
425            WorkerType::Read => &self.read_waiters,
426            WorkerType::Write => &self.write_waiters,
427        }
428    }
429
430    fn enqueue(&self, op: IoOperation) -> Result<(), Errno> {
431        let worker_type = op.op_type.worker_type();
432        {
433            let mut pending = self.pending.lock();
434            if pending.is_stopped {
435                return error!(EINVAL);
436            }
437            if pending.ops_len() >= self.max_operations {
438                return error!(EAGAIN);
439            }
440            pending.ops_for(worker_type).push_back(op);
441        }
442        self.waiters_for(worker_type).notify_unordered_count(1);
443        Ok(())
444    }
445
446    fn stop(&self) {
447        let mut pending = self.pending.lock();
448        pending.is_stopped = true;
449        pending.read_ops.clear();
450        pending.write_ops.clear();
451        self.read_waiters.notify_all();
452        self.write_waiters.notify_all();
453    }
454
455    fn dequeue(&self, worker_type: WorkerType) -> Option<IoAction> {
456        let mut pending = self.pending.lock();
457        if pending.is_stopped {
458            return Some(IoAction::Stop);
459        }
460        pending.ops_for(worker_type).pop_front().map(IoAction::Op)
461    }
462
463    fn remove(&self, worker_type: WorkerType, iocb_addr: IocbPtr) -> Result<(), Errno> {
464        {
465            let mut pending = self.pending.lock();
466            if pending.is_stopped {
467                return error!(EINVAL);
468            }
469            // TODO: Use pop_front_if when available.
470            if let Some(idx) = pending
471                .ops_for(worker_type)
472                .iter()
473                .position(|value| value.iocb_addr.addr() == iocb_addr.addr())
474            {
475                pending.ops_for(worker_type).remove(idx);
476            } else {
477                return error!(EAGAIN);
478            }
479        }
480        Ok(())
481    }
482
483    fn block_until_dequeue(
484        &self,
485        current_task: &CurrentTask,
486        worker_type: WorkerType,
487    ) -> Result<IoAction, Errno> {
488        if let Some(action) = self.dequeue(worker_type) {
489            return Ok(action);
490        }
491        loop {
492            let event = InterruptibleEvent::new();
493            let (mut waiter, guard) = SimpleWaiter::new(&event);
494            self.waiters_for(worker_type).wait_async_simple(&mut waiter);
495            if let Some(action) = self.dequeue(worker_type) {
496                return Ok(action);
497            }
498            current_task.block_until(guard, zx::MonotonicInstant::INFINITE)?;
499        }
500    }
501}
502
503#[derive(Default)]
504struct ResultQueue {
505    waiters: WaitQueue,
506    events: LockDepMutex<VecDeque<io_event>, AioEventsLock>,
507}
508
509impl ResultQueue {
510    fn enqueue(&self, event: io_event) {
511        self.events.lock().push_back(event);
512        self.waiters.notify_unordered_count(1);
513    }
514
515    fn dequeue(&self, limit: usize) -> Vec<io_event> {
516        let mut events = self.events.lock();
517        let len = std::cmp::min(events.len(), limit);
518        events.drain(..len).collect()
519    }
520}
starnix_core/vfs/aio.rs

starnix_core/vfs/
aio.rs
