diff --git a/Cargo.toml b/Cargo.toml index cb5ae8f..b441f44 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -23,7 +23,6 @@ default-features = false features = ["map"] [dev-dependencies] -log = "0.3" env_logger = "0.4" getopts = "0.2" rand = "0.3" @@ -69,3 +68,9 @@ required-features = ["std", "phy-tap_interface", "socket-tcp", "socket-udp"] [[example]] name = "loopback" required-features = ["alloc", "proto-ipv4"] + +# This is really a test, but it requires root privileges for setup (the tap interface) +# so it is built as an example. +[[example]] +name = "stress" +required-features = ["std", "phy-tap_interface", "socket-tcp"] diff --git a/examples/stress.rs b/examples/stress.rs new file mode 100644 index 0000000..66cf2ab --- /dev/null +++ b/examples/stress.rs @@ -0,0 +1,139 @@ +#[cfg(feature = "log")] +extern crate log; +#[cfg(feature = "log")] +extern crate env_logger; +extern crate getopts; +extern crate smoltcp; + +mod utils; + +use std::cmp; +use std::collections::BTreeMap; +use std::sync::atomic::{Ordering, AtomicBool, ATOMIC_BOOL_INIT}; +use std::time::Instant; +use std::thread; +use std::io::{Read, Write}; +use std::net::TcpStream; +use std::os::unix::io::AsRawFd; +use smoltcp::phy::wait as phy_wait; +use smoltcp::wire::{EthernetAddress, IpAddress, IpCidr}; +use smoltcp::iface::{NeighborCache, EthernetInterfaceBuilder}; +use smoltcp::socket::SocketSet; +use smoltcp::socket::{TcpSocket, TcpSocketBuffer}; + +const AMOUNT: usize = 10_000_000; + +enum Client { Reader, Writer } + +fn client(kind: Client) { + let port = match kind { Client::Reader => 1234, Client::Writer => 1235 }; + let mut stream = TcpStream::connect(("192.168.69.1", port)).unwrap(); + let mut buffer = vec![0; 64]; + + let mut processed = 0; + while processed < AMOUNT { + let length = cmp::min(buffer.len(), AMOUNT - processed); + let result = match kind { + Client::Reader => stream.read(&mut buffer[..length]), + Client::Writer => stream.write(&buffer[..length]), + }; + match result { + Ok(0) => break, + Ok(result) => { + print!("(P:{})", result); + processed += result + } + Err(err) => panic!("cannot process: {}", err) + } + } + println!("client done"); + CLIENT_DONE.store(true, Ordering::SeqCst); +} + +static CLIENT_DONE: AtomicBool = ATOMIC_BOOL_INIT; + +fn main() { + let (mut opts, mut free) = utils::create_options(); + utils::add_tap_options(&mut opts, &mut free); + utils::add_middleware_options(&mut opts, &mut free); + free.push("MODE"); + + let mut matches = utils::parse_options(&opts, free); + let device = utils::parse_tap_options(&mut matches); + let fd = device.as_raw_fd(); + let device = utils::parse_middleware_options(&mut matches, device, /*loopback=*/false); + let mode = match matches.free[0].as_ref() { + "reader" => Client::Reader, + "writer" => Client::Writer, + _ => panic!("invalid mode") + }; + + thread::spawn(move || client(mode)); + + let startup_time = Instant::now(); + + let neighbor_cache = NeighborCache::new(BTreeMap::new()); + + let tcp1_rx_buffer = TcpSocketBuffer::new(vec![0; 65535]); + let tcp1_tx_buffer = TcpSocketBuffer::new(vec![0; 65535]); + let tcp1_socket = TcpSocket::new(tcp1_rx_buffer, tcp1_tx_buffer); + + let tcp2_rx_buffer = TcpSocketBuffer::new(vec![0; 65535]); + let tcp2_tx_buffer = TcpSocketBuffer::new(vec![0; 65535]); + let tcp2_socket = TcpSocket::new(tcp2_rx_buffer, tcp2_tx_buffer); + + let ethernet_addr = EthernetAddress([0x02, 0x00, 0x00, 0x00, 0x00, 0x01]); + let ip_addrs = [IpCidr::new(IpAddress::v4(192, 168, 69, 1), 24)]; + let mut iface = EthernetInterfaceBuilder::new(device) + .ethernet_addr(ethernet_addr) + .neighbor_cache(neighbor_cache) + .ip_addrs(ip_addrs) + .finalize(); + + let mut sockets = SocketSet::new(vec![]); + let tcp1_handle = sockets.add(tcp1_socket); + let tcp2_handle = sockets.add(tcp2_socket); + + let mut processed = 0; + while !CLIENT_DONE.load(Ordering::SeqCst) { + // tcp:1234: emit data + { + let mut socket = sockets.get::(tcp1_handle); + if !socket.is_open() { + socket.listen(1234).unwrap(); + } + + if socket.can_send() { + if processed < AMOUNT { + let length = socket.send(|buffer| { + let length = cmp::min(buffer.len(), AMOUNT - processed); + (length, length) + }).unwrap(); + processed += length; + } + } + } + + // tcp:1235: sink data + { + let mut socket = sockets.get::(tcp2_handle); + if !socket.is_open() { + socket.listen(1235).unwrap(); + } + + if socket.can_recv() { + if processed < AMOUNT { + let length = socket.recv(|buffer| { + let length = cmp::min(buffer.len(), AMOUNT - processed); + (length, length) + }).unwrap(); + processed += length; + } + } + } + + let timestamp = utils::millis_since(startup_time); + let poll_at = iface.poll(&mut sockets, timestamp).expect("poll error"); + phy_wait(fd, poll_at.map(|at| at.saturating_sub(timestamp))).expect("wait error"); + } +} diff --git a/examples/utils.rs b/examples/utils.rs index d2ced06..90dc265 100644 --- a/examples/utils.rs +++ b/examples/utils.rs @@ -1,3 +1,5 @@ +#![allow(dead_code)] + use std::cell::RefCell; use std::str::{self, FromStr}; use std::rc::Rc; @@ -6,13 +8,16 @@ use std::fs::File; use std::time::{Instant, SystemTime, UNIX_EPOCH}; use std::env; use std::process; +#[cfg(feature = "log")] use log::{LogLevel, LogLevelFilter, LogRecord}; +#[cfg(feature = "log")] use env_logger::LogBuilder; use getopts::{Options, Matches}; use smoltcp::phy::{Device, EthernetTracer, FaultInjector, TapInterface}; use smoltcp::phy::{PcapWriter, PcapSink, PcapMode, PcapLinkType}; +#[cfg(feature = "log")] pub fn setup_logging_with_clock(filter: &str, since_startup: F) where F: Fn() -> u64 + Send + Sync + 'static { LogBuilder::new() @@ -38,6 +43,7 @@ pub fn setup_logging_with_clock(filter: &str, since_startup: F) .unwrap(); } +#[cfg(feature = "log")] pub fn setup_logging(filter: &str) { let startup_at = Instant::now(); setup_logging_with_clock(filter, move || { @@ -91,8 +97,10 @@ pub fn add_middleware_options(opts: &mut Options, _free: &mut Vec<&str>) { opts.optopt("", "shaping-interval", "Sets the interval for rate limiting (ms)", "RATE"); } -pub fn parse_middleware_options Device<'a>>(matches: &mut Matches, device: D, loopback: bool) - -> FaultInjector>>> { +pub fn parse_middleware_options(matches: &mut Matches, device: D, loopback: bool) + -> FaultInjector>>> + where D: for<'a> Device<'a> +{ let drop_chance = matches.opt_str("drop-chance").map(|s| u8::from_str(&s).unwrap()) .unwrap_or(0); let corrupt_chance = matches.opt_str("corrupt-chance").map(|s| u8::from_str(&s).unwrap()) @@ -118,7 +126,10 @@ pub fn parse_middleware_options Device<'a>>(matches: &mut Matches, de let device = PcapWriter::new(device, Rc::new(RefCell::new(pcap_writer)) as Rc, if loopback { PcapMode::TxOnly } else { PcapMode::Both }, PcapLinkType::Ethernet); - let device = EthernetTracer::new(device, |_timestamp, printer| trace!("{}", printer)); + let device = EthernetTracer::new(device, |_timestamp, _printer| { + #[cfg(feature = "log")] + trace!("{}", _printer); + }); let mut device = FaultInjector::new(device, seed); device.set_drop_chance(drop_chance); device.set_corrupt_chance(corrupt_chance);