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
//! Io filter integration tests.
use std::{cell::Cell, io, io::Read, io::Write, net, time::Duration};
use ntex::io::{FilterBuf, FilterLayer, Io};
use ntex::{codec::BytesCodec, fn_service, server::test_server, util::Bytes};
/// Must be greater than or equal to `IoConfig::write_buf_threshold`
/// (8192 by default), so that the io stream initiates a direct
/// (in-place) write while the read buffer is being processed.
const BURST_SIZE: usize = 16 * 1024;
/// Filter that emits a large burst of write data while processing
/// incoming read data, similar to a TLS filter that writes handshake
/// or control data in response to incoming records.
#[derive(Debug, Default)]
struct BurstWriteFilter {
sent: Cell<bool>,
}
impl FilterLayer for BurstWriteFilter {
fn process_read_buf(&self, buf: &FilterBuf<'_>) -> io::Result<()> {
// pass incoming data through
let got_data = buf.with_read_buffers(|src, dst| {
if let Some(src) = src.take() {
dst.extend_from_slice(&src);
!src.is_empty()
} else {
false
}
});
// the first incoming data triggers a large write
if got_data && !self.sent.get() {
self.sent.set(true);
buf.with_write_buffers(|_, dst| {
dst.extend_from_slice(&[b'x'; BURST_SIZE]);
});
}
Ok(())
}
fn process_write_buf(&self, buf: &FilterBuf<'_>) -> io::Result<()> {
// pass outgoing data through
buf.with_write_buffers(|src, dst| {
if !src.is_empty() {
src.move_to(dst);
}
});
Ok(())
}
}
/// A filter that produces a write burst (>= `write_buf_threshold`)
/// during read-buffer processing must not break the io stream.
///
/// Regression test for the polling (neon) backend: the driver used to
/// panic on reentrant slab access ("called `Option::unwrap()` on a
/// `None` value") when a filter generated enough write data during
/// read processing to trigger a direct write.
#[ntex::test]
async fn test_filter_large_write_during_read_processing() {
let srv = test_server(async || {
fn_service(async move |io: Io| {
let io = io.add_filter(BurstWriteFilter::default());
// notify the client that the filter is installed
io.send(Bytes::from_static(b"hi"), &BytesCodec)
.await
.unwrap();
// echo service
while let Ok(Some(msg)) = io.recv(&BytesCodec).await {
io.send(msg, &BytesCodec).await.unwrap();
}
Ok::<_, io::Error>(())
})
});
let mut client = net::TcpStream::connect(srv.addr()).unwrap();
client
.set_read_timeout(Some(Duration::from_secs(5)))
.unwrap();
// wait for the server-side filter to be installed
let mut greeting = [0u8; 2];
client.read_exact(&mut greeting).unwrap();
assert_eq!(&greeting, b"hi");
// this write triggers the filter burst during read processing
// on the server side
client.write_all(b"ping").unwrap();
// burst must arrive first (it is written during read processing),
// followed by the echoed message
let mut data = vec![0u8; BURST_SIZE + 4];
client
.read_exact(&mut data)
.expect("server did not send data, io driver is broken");
assert!(data[..BURST_SIZE].iter().all(|b| *b == b'x'));
assert_eq!(&data[BURST_SIZE..], b"ping");
}