Я пишу HTTP-сервер и хочу иметь один поток на процессор, и каждый поток контролирует отдельный экземпляр epoll.Я реализовал тестовую версию эполла, запускаемого с помощью края, с помощью ящика nix, но я не понимаю, что с ним не так.
Когда я тестирую это с ab (ab -n 200000 -c 1000) работает быстро, но в конце тестирования epoll зависает.Я не знаю, что не так, может быть, я нарушил состояние epoll или мне нужно использовать внутренние состояния клиента?
extern crate nix;
use nix::errno::Errno;
use nix::fcntl::*;
use nix::sys::epoll::EpollCreateFlags;
use nix::sys::epoll::*;
use nix::sys::socket::sockopt::ReuseAddr;
use nix::sys::socket::*;
use nix::unistd::close;
use nix::unistd::{read, write};
use nix::Error;
use std::collections::HashMap;
use std::os::unix::io::RawFd;
use std::thread;
const BUFSIZE: usize = 16384;
#[derive(Hash, Eq, PartialEq, Debug, Clone, Copy)]
pub enum State {
Read,
Write,
}
#[derive(Clone, Debug)]
pub struct Client {
pub readed: Vec<u8>,
pub state: State,
}
impl Client {
fn new() -> Self {
Client {
readed: Vec::new(),
state: State::Read,
}
}
}
fn socket_to_nonblock(sock: RawFd) -> Result<(), nix::Error> {
let flags = match fcntl(sock, F_GETFL) {
Ok(fl) => fl,
Err(err) => return Err(err),
};
let new_flags = flags | OFlag::O_NONBLOCK.bits();
let new_flags = OFlag::from_bits(new_flags).unwrap();
match fcntl(sock, F_SETFL(new_flags)) {
Ok(_) => return Ok(()),
Err(err) => return Err(err),
}
}
fn epoll_loop(server_sock: RawFd) -> nix::Result<()> {
let mut clients: HashMap<RawFd, Client> = HashMap::new(); //SockFlag::EPOLLET
let epfd = epoll_create1(EpollCreateFlags::from_bits(0).unwrap())?;
let mut ev = EpollEvent::new(
EpollFlags::EPOLLIN | EpollFlags::EPOLLOUT,
server_sock as u64,
);
epoll_ctl(epfd, EpollOp::EpollCtlAdd, server_sock, &mut ev)?;
let mut epoll_events = vec![EpollEvent::empty(); 1024];
let buffer = [0; BUFSIZE];
let critical_error = false;
let mut accepted = 0;
let mut closed = 0;
let mut refused = 0;
loop {
let nfds = match epoll_wait(epfd, &mut epoll_events, -1) {
Ok(n) => n,
Err(e) => {
println!("Err wait {:?}", e);
panic!();
}
};
for i in 0..nfds {
let cur_socket = epoll_events[i].data() as i32;
let cur_event = epoll_events[i].events();
if cur_event == cur_event & EpollFlags::EPOLLERR
|| cur_event == cur_event & EpollFlags::EPOLLHUP
|| cur_event != cur_event & (EpollFlags::EPOLLIN | EpollFlags::EPOLLOUT)
{
println!("error big if {:?}", cur_event);
close(epoll_events[i].data() as i32);
continue;
}
if server_sock == cur_socket {
loop {
let client_fd = match accept(server_sock) {
Ok(client) => {
accepted += 1;
if (client < 0) {
println!("JOPA");
panic!();
}
println!(
"Accepted {:?} Closed {:?} Dif: {:?} Refused {:?} Events len: {:?}",
accepted,
closed,
accepted - closed,
refused,
nfds
);
client
}
Err(err) => {
refused += 1;
println!("Error accept {:?}", err);
break;
}
};
match socket_to_nonblock(client_fd) {
Ok(_) => {}
Err(err) => println!("Non block err {:?}", err),
}
let mut ev = EpollEvent::new(
EpollFlags::EPOLLIN | EpollFlags::EPOLLET,
client_fd as u64,
);
match epoll_ctl(epfd, EpollOp::EpollCtlAdd, client_fd, &mut ev) {
Ok(e) => {}
Err(err) => println!("Server accept ctl {:?}", err),
}
println!("{:?}", clients.insert(client_fd, Client::new()));
break;
}
println!("loop breaked");
continue;
}
if cur_event == cur_event & EpollFlags::EPOLLIN {
let mut buf: [u8; 2048] = [0; 2048];
let client = clients.get_mut(&cur_socket).unwrap();
loop {
let mut total_len = client.readed.len();
let size = match read(cur_socket, &mut buf[total_len..]) {
Ok(size) => {
client.readed.extend_from_slice(&buf[..size]);
if (size == 0) {
break;
}
}
Err(e) => break,
};
}
let req = std::str::from_utf8(&client.readed.as_slice())
.unwrap()
.to_string();
if !(req.find("\n\n").is_some() || req.find("\r\n\r\n").is_some()) {
continue;
}
let mut ev = EpollEvent::new(
EpollFlags::EPOLLOUT | EpollFlags::EPOLLET,
cur_socket as u64,
);
match epoll_ctl(epfd, EpollOp::EpollCtlMod, cur_socket, &mut ev) {
Ok(e) => {}
Err(err) => println!("Read ctl Err {:?}", err),
}
client.state = State::Write;
}
if cur_event == cur_event & EpollFlags::EPOLLOUT {
let buf = "HTTP/1.1 200 Ok\nConnection: close\nContent-Type: text/plain\n\nha?\n\n";
let buf_len = buf.len();
let mut sended = 0;
let mut need_to_close = false;
loop {
match write(cur_socket, &buf.as_bytes()[sended..]) {
Ok(len) => {
sended += len;
if sended >= buf_len {
need_to_close = true;
break;
}
}
Err(e) => {
need_to_close = true;
break;
}
}
}
if (need_to_close) {
match epoll_ctl(
epfd,
EpollOp::EpollCtlDel,
cur_socket as i32,
&mut epoll_events[i],
) {
Ok(e) => {}
Err(err) => println!("Err epollctl write {:?}", err),
}
clients.remove(&cur_socket);
match shutdown(cur_socket as i32, Shutdown::Both) {
Ok(e) => {}
Err(err) => println!("Err shutdown {:?}", err),
}
close(cur_socket as i32)?;
closed += 1;
println!("closed: {:?}", closed);
}
}
continue;
}
}
}
fn main() {
let server_sock = match socket(
AddressFamily::Inet,
SockType::Stream,
SockFlag::SOCK_NONBLOCK,
SockProtocol::Tcp,
) {
Ok(sock) => sock,
Err(err) => panic!("{:?}", err),
};
match setsockopt(server_sock, ReuseAddr, &true) {
Ok(_) => {}
Err(err) => panic!("Error set sock opt {:?}", err),
}
let addr = SockAddr::new_inet(InetAddr::new(IpAddr::new_v4(127, 0, 0, 1), 5468));
match bind(server_sock, &addr) {
Ok(_) => {}
Err(err) => panic!("Error bind: {:?}", err),
}
socket_to_nonblock(server_sock);
match listen(server_sock, 1024) {
Ok(_) => {}
Err(err) => panic!("Error listen: {:?}", err),
}
epoll_loop(server_sock.clone());
}