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
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
//! Tokio-based asynchronous datalink channel (feature `tokio`).
//!
//! [`AsyncSender`] and [`AsyncReceiver`] mirror the synchronous [`Sender`](crate::channel::Sender)
//! / [`Receiver`](crate::channel::Receiver), but await readiness on the Tokio reactor instead of
//! blocking in `poll`. Readiness is tracked with [`tokio::io::unix::AsyncFd`] registered on the
//! channel's file descriptor; the actual zero-copy reads/writes go straight through the backend.
use std::io;
use std::os::fd::{AsFd, AsRawFd, RawFd};
use tokio::io::Interest;
use tokio::io::unix::AsyncFd;
use crate::block::{Block, Frame, FrameMeta, PacketType};
use crate::channel::{ChannelBuilder, RxBackend, TxBackend};
use crate::error::Result;
use crate::sys::Stats;
/// A bare file descriptor wrapper so [`AsyncFd`] can register interest without owning (and later
/// closing) the descriptor — ownership stays with the backend.
#[derive(Debug)]
struct FdHolder(RawFd);
impl AsRawFd for FdHolder {
fn as_raw_fd(&self) -> RawFd {
self.0
}
}
impl ChannelBuilder {
/// Opens the channel for asynchronous use on the current Tokio runtime.
///
/// Must be called from within a Tokio runtime context (the file descriptors are registered
/// with the reactor on construction).
pub fn build_async(&self) -> Result<(AsyncSender, AsyncReceiver)> {
let parts = self.open()?;
let tx = AsyncSender::new(parts.tx)?;
let rx = AsyncReceiver::new(parts.rx)?;
Ok((tx, rx))
}
/// Opens `count` async receivers load-balanced across a `PACKET_FANOUT` group for multi-core
/// RX. Each [`AsyncReceiver`] is registered with the current Tokio runtime; spawn a task per
/// receiver. Fanout is receive-only.
pub fn build_fanout_rx_async(&self, count: usize) -> Result<Vec<AsyncReceiver>> {
self.build_fanout_backends(count)?.into_iter().map(AsyncReceiver::new).collect()
}
}
/// The asynchronous transmit half of a datalink channel.
#[derive(Debug)]
pub struct AsyncSender {
// `readiness` is declared first so it deregisters from the reactor before `backend` closes
// the underlying fd on drop.
readiness: AsyncFd<FdHolder>,
backend: TxBackend,
}
impl AsyncSender {
pub(crate) fn new(backend: TxBackend) -> Result<Self> {
let fd = backend.borrow_fd().as_raw_fd();
let readiness = AsyncFd::with_interest(FdHolder(fd), Interest::WRITABLE)
.map_err(crate::Error::Register)?;
Ok(AsyncSender { readiness, backend })
}
/// Sends a single raw frame, awaiting transmit capacity if the ring/queue is full.
pub async fn send(&mut self, frame: &[u8]) -> Result<()> {
if let Some(max) = self.backend.max_frame_len() {
if frame.len() > max {
return Err(crate::Error::FrameTooLarge { len: frame.len(), max });
}
}
loop {
match self.backend.try_send(frame) {
Ok(_) => break,
Err(e) if e.kind() == io::ErrorKind::WouldBlock => {
// A full ring may be waiting on a kick the kernel refused earlier; complete
// it, or writability may never arrive.
self.wait_for_pickup().await?;
let mut guard =
self.readiness.writable().await.map_err(crate::Error::Send)?;
guard.clear_ready();
}
Err(e) => return Err(crate::Error::Send(e)),
}
}
self.wait_for_pickup().await
}
/// Awaits until the kernel has picked up the accepted frames. The TX ring is only walked
/// inside `send(2)`, so a kick refused by a full send buffer would otherwise leave frames
/// parked in the ring indefinitely after `send` returned `Ok`.
///
/// The follow-up kick is a *blocking* zero-length send — `POLLOUT` cannot signal this state
/// (see `TxRing::finish_kick`) — so it runs on the blocking pool with a duplicated fd.
async fn wait_for_pickup(&mut self) -> Result<()> {
if !self.backend.has_unsent() {
return Ok(());
}
let fd =
self.backend.borrow_fd().try_clone_to_owned().map_err(crate::Error::Send)?;
tokio::task::spawn_blocking(move || crate::sys::linux::ring::blocking_kick(fd.as_fd()))
.await
.map_err(io::Error::other)
.map_err(crate::Error::Send)?
.map_err(crate::Error::Send)?;
Ok(())
}
/// Sends a batch of frames, awaiting capacity if needed, returning how many were accepted.
pub async fn send_batch(&mut self, frames: &[&[u8]]) -> Result<usize> {
// A full queue surfaces as `WouldBlock`, never `Ok(0)`, so `Ok(0)` here would mean an
// empty input — awaiting writability for it would sleep forever on an idle socket.
if frames.is_empty() {
return Ok(0);
}
crate::channel::check_batch_frame_sizes(frames, self.backend.max_frame_len())?;
let sent = loop {
match self.backend.try_send_batch(frames) {
Ok(n) => break n,
Err(e) if e.kind() == io::ErrorKind::WouldBlock => {
self.wait_for_pickup().await?;
let mut guard =
self.readiness.writable().await.map_err(crate::Error::Send)?;
guard.clear_ready();
}
Err(e) => return Err(crate::Error::Send(e)),
}
};
self.wait_for_pickup().await?;
Ok(sent)
}
/// Borrows the underlying file descriptor.
pub fn as_fd(&self) -> std::os::fd::BorrowedFd<'_> {
self.backend.borrow_fd()
}
}
#[cfg(test)]
mod tests {
use std::collections::VecDeque;
use std::sync::{Arc, Mutex};
use std::time::Duration;
use super::*;
/// Builds an [`AsyncSender`] over the in-memory dummy backend (no privileges needed).
fn dummy_async_sender() -> (AsyncSender, crate::dummy::SentQueue, std::os::fd::OwnedFd) {
let (read, write) = crate::dummy::make_pipe().expect("pipe");
let sent: crate::dummy::SentQueue = Arc::new(Mutex::new(VecDeque::new()));
let tx = AsyncSender::new(TxBackend::Dummy { sent: sent.clone(), fd: write })
.expect("async sender");
(tx, sent, read)
}
#[tokio::test]
#[cfg_attr(miri, ignore = "tokio's reactor uses epoll, which Miri cannot interpret")]
async fn empty_send_batch_returns_zero_without_hanging() {
let (mut tx, _sent, _read) = dummy_async_sender();
let n = tokio::time::timeout(Duration::from_millis(500), tx.send_batch(&[]))
.await
.expect("send_batch(&[]) must resolve, not await readiness forever")
.expect("empty batch is not an error");
assert_eq!(n, 0);
}
#[tokio::test]
#[cfg_attr(miri, ignore = "tokio's reactor uses epoll, which Miri cannot interpret")]
async fn send_batch_delivers_frames() {
let (mut tx, sent, _read) = dummy_async_sender();
let n = tx.send_batch(&[b"one".as_slice(), b"two".as_slice()]).await.expect("send");
assert_eq!(n, 2);
assert_eq!(sent.lock().unwrap().len(), 2);
}
}
/// The asynchronous receive half of a datalink channel.
#[derive(Debug)]
pub struct AsyncReceiver {
readiness: AsyncFd<FdHolder>,
backend: RxBackend,
stats_total: Stats,
}
impl AsyncReceiver {
pub(crate) fn new(backend: RxBackend) -> Result<Self> {
let fd = backend.borrow_fd().as_raw_fd();
let readiness = AsyncFd::with_interest(FdHolder(fd), Interest::READABLE)
.map_err(crate::Error::Register)?;
Ok(AsyncReceiver { readiness, backend, stats_total: Stats::default() })
}
/// Receives the next batch of frames, awaiting until at least one is available.
///
/// The returned [`Block`] borrows zero-copy from kernel memory (ring backend) or an internal
/// buffer (basic backend); it must be dropped before the next call.
pub async fn recv_block(&mut self) -> Result<Block<'_>> {
match &self.backend {
RxBackend::Ring(_) => {
// Wait until the current block is owned by user space, then borrow it.
loop {
if let RxBackend::Ring(ring) = &self.backend {
if ring.block_ready() {
break;
}
}
let mut guard =
self.readiness.readable().await.map_err(crate::Error::Recv)?;
// If readiness was spurious (block still not closed), clear so epoll re-arms.
let ready = matches!(&self.backend, RxBackend::Ring(r) if r.block_ready());
if !ready {
guard.clear_ready();
}
}
match &mut self.backend {
RxBackend::Ring(ring) => Ok(ring.recv_block().expect("block reported ready")),
_ => unreachable!("backend kind is stable"),
}
}
RxBackend::Basic { .. } => {
// Read once we are readable; loop only over the wait, then build the frame.
let (n, pkttype) = loop {
let attempt = match &mut self.backend {
RxBackend::Basic { sock, buf } => sock.recv_into(buf),
_ => unreachable!("backend kind is stable"),
};
match attempt {
Ok(v) => break v,
Err(e) if e.kind() == io::ErrorKind::WouldBlock => {
let mut g =
self.readiness.readable().await.map_err(crate::Error::Recv)?;
g.clear_ready();
}
Err(e) => return Err(crate::Error::Recv(e)),
}
};
let RxBackend::Basic { buf, .. } = &self.backend else {
unreachable!("backend kind is stable")
};
let captured = n.min(buf.len());
let meta = FrameMeta {
wire_len: n,
timestamp: None,
vlan: None,
packet_type: PacketType::from_raw(pkttype),
};
Ok(Block::single(Some(Frame::new(&buf[..captured], meta))))
}
RxBackend::Dummy { .. } => {
// Await an inject (signalled via the pipe), then deliver the queued frame.
let frame = loop {
let popped = match &mut self.backend {
RxBackend::Dummy { queue, .. } => {
queue.lock().expect("dummy queue poisoned").pop_front()
}
_ => unreachable!("backend kind is stable"),
};
match popped {
Some(Ok(v)) => break v,
Some(Err(e)) => return Err(crate::Error::Recv(e)),
None => {
let mut g =
self.readiness.readable().await.map_err(crate::Error::Recv)?;
let eof = match &self.backend {
RxBackend::Dummy { readiness, .. } => {
crate::dummy::drain_readiness(readiness.as_fd())
}
_ => false,
};
g.clear_ready();
let empty = match &self.backend {
RxBackend::Dummy { queue, .. } => {
queue.lock().expect("dummy queue poisoned").is_empty()
}
_ => true,
};
if eof && empty {
return Err(crate::Error::Recv(io::Error::from(
io::ErrorKind::UnexpectedEof,
)));
}
}
}
};
let RxBackend::Dummy { last, .. } = &mut self.backend else {
unreachable!("backend kind is stable")
};
*last = frame;
let meta = FrameMeta {
wire_len: last.len(),
timestamp: None,
vlan: None,
packet_type: crate::channel::dummy_packet_type(last),
};
Ok(Block::single(Some(Frame::new(&last[..], meta))))
}
}
}
/// Returns cumulative receive/drop statistics for this channel since it was opened.
///
/// See [`Receiver::stats`](crate::Receiver::stats); reading also clears the kernel
/// `TP_STATUS_LOSING` flag observed via [`Block::is_losing`](crate::Block::is_losing).
pub fn stats(&mut self) -> Result<Stats> {
let delta = match &mut self.backend {
RxBackend::Ring(ring) => ring.stats().map_err(crate::Error::Stats)?,
RxBackend::Basic { sock, .. } => sock.statistics().map_err(crate::Error::Stats)?,
RxBackend::Dummy { .. } => Stats::default(),
};
self.stats_total.accumulate(delta);
Ok(self.stats_total)
}
/// Enables or disables promiscuous mode on the interface.
pub fn set_promiscuous(&self, on: bool) -> Result<()> {
match &self.backend {
RxBackend::Ring(ring) => ring.set_promiscuous(on),
RxBackend::Basic { sock, .. } => sock.set_promiscuous(on),
RxBackend::Dummy { .. } => Ok(()),
}
}
/// Borrows the underlying file descriptor.
pub fn as_fd(&self) -> std::os::fd::BorrowedFd<'_> {
self.backend.borrow_fd()
}
}