karmaio 0.1.0

A modern, fast, share-nothing asynchronous runtime for Rust
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148

//! Offset-less write operations for stream-like file descriptors (pipes,
//! sockets, char devices). See [`crate::driver::ops::read`] for why the
//! offset-based [`crate::driver::ops::write_at`] ops are unsuitable for pipes.

use std::io;

#[cfg(unix)]
use std::os::fd::{AsFd, AsRawFd};
#[cfg(windows)]
use std::os::windows::io::AsRawHandle;

#[cfg(windows)]
use crate::driver::backends::iocp::{IocpOperation, IocpSubmission};
#[cfg(target_os = "linux")]
use crate::driver::backends::iouring::{Submission as UringSubmission, UringOperation};
#[cfg(any(
    target_os = "macos",
    target_os = "freebsd",
    target_os = "netbsd",
    target_os = "openbsd",
    target_os = "dragonfly"
))]
use crate::driver::backends::kqueue::{KqueueAttempt, KqueueOperation};

use crate::{
    buf::{BufResult, IoBuf},
    driver::{
        helpers::io_handle::SharedIoHandle,
        ops::{Completion, Op},
    },
    runtime::local::CURRENT_DRIVER,
};

pub(crate) struct Write<T, B> {
    // Holds a strong ref to the fd, preventing the pipe from being closed while
    // an operation is in-flight.
    #[allow(dead_code)]
    io_handle: SharedIoHandle<T>,

    pub(crate) buf: B,
}

#[cfg(unix)]
impl<T, B> Op<Write<T, B>>
where
    T: AsFd + AsRawFd + 'static,
    B: IoBuf + 'static,
{
    pub(crate) fn write(io_handle: &SharedIoHandle<T>, buf: B) -> io::Result<Op<Write<T, B>>> {
        let data = Write {
            io_handle: io_handle.clone(),
            buf,
        };
        CURRENT_DRIVER.with(|handle| handle.upgrade().expect("not in a runtime context").submit_op(data))
    }
}

#[cfg(windows)]
impl<T, B> Op<Write<T, B>>
where
    T: AsRawHandle + 'static,
    B: IoBuf + 'static,
{
    pub(crate) fn write(io_handle: &SharedIoHandle<T>, buf: B) -> io::Result<Op<Write<T, B>>> {
        let data = Write {
            io_handle: io_handle.clone(),
            buf,
        };
        CURRENT_DRIVER.with(|handle| handle.upgrade().expect("not in a runtime context").submit_op(data))
    }
}

#[cfg(target_os = "linux")]
unsafe impl<T: AsRawFd + 'static, B: IoBuf + 'static> UringOperation for Write<T, B> {
    type Output = BufResult<usize, B>;
    fn submit(&mut self) -> UringSubmission {
        use io_uring::{opcode, types};

        let ptr = self.buf.as_init().as_ptr();
        let len = self.buf.as_init().len();
        opcode::Write::new(types::Fd(self.io_handle.raw_fd()), ptr, len as _).build()
    }

    fn complete(self, completion: Completion) -> Self::Output {
        match completion.result {
            Ok(res) => BufResult(Ok(res as usize), self.buf),
            Err(err) => BufResult(Err(err), self.buf),
        }
    }
}

#[cfg(any(
    target_os = "macos",
    target_os = "freebsd",
    target_os = "netbsd",
    target_os = "openbsd",
    target_os = "dragonfly"
))]
impl<T: AsFd + AsRawFd + 'static, B: IoBuf + 'static> KqueueOperation for Write<T, B> {
    type Output = BufResult<usize, B>;
    fn attempt(&mut self) -> KqueueAttempt {
        kqueue_syscall_submit!(
            self.io_handle.raw_fd(),
            crate::driver::backends::kqueue::Direction::Write,
            {
                // Safety: the operation owns this buffer for the duration of the
                // syscall, and the slice covers exactly its initialized bytes.
                let buf = unsafe { std::slice::from_raw_parts(self.buf.as_init().as_ptr(), self.buf.as_init().len()) };
                rustix::io::write(&self.io_handle, buf)
                    .map(|n| n as u32)
                    .map_err(std::io::Error::from)
            }
        )
    }

    fn complete(self, completion: Completion) -> Self::Output {
        match completion.result {
            Ok(res) => BufResult(Ok(res as usize), self.buf),
            Err(err) => BufResult(Err(err), self.buf),
        }
    }
}

#[cfg(windows)]
unsafe impl<T: AsRawHandle + 'static, B: IoBuf + 'static> IocpOperation for Write<T, B> {
    type Output = BufResult<usize, B>;

    fn submit(&mut self) -> IocpSubmission {
        use crate::driver::backends::iocp::Interest;
        use windows_sys::Win32::Storage::FileSystem::WriteFile;

        let ptr = self.buf.as_init().as_ptr() as *const u8;
        let len = self.buf.as_init().len() as u32;
        let handle = self.io_handle.raw_handle();

        let mut interest = Interest::new(handle as _);
        windows_syscall_submit_overlapped!(interest, file, {
            WriteFile(handle as _, ptr, len, std::ptr::null_mut(), interest.as_mut_ptr())
        })
    }

    fn complete(self, completion: Completion) -> Self::Output {
        match completion.result {
            Ok(res) => BufResult(Ok(res as usize), self.buf),
            Err(err) => BufResult(Err(err), self.buf),
        }
    }
}