hedgewars: comparison rust/hedgewars-server/src/server/network.rs

equal deleted inserted replaced

-:ed3b510b860c
+:6af892a0a4b8
 collections::HashSet,
 io,
 io::{Error, ErrorKind, Read, Write},
 mem::{replace, swap},
 net::{IpAddr, Ipv4Addr, SocketAddr},
+num::NonZeroU32,
+time::Duration,
+time::Instant,
 };
 use log::*;
 use mio::{
+event::Source,
 net::{TcpListener, TcpStream},
-Evented, Poll, PollOpt, Ready, Token,
+Interest, Poll, Token, Waker,
 };
-use mio_extras::timer;
 use netbuf;
 use slab::Slab;
 use crate::{
-core::types::ClientId,
+core::{
+events::{TimedEvents, Timeout},
+types::ClientId,
+},
 handlers,
 handlers::{IoResult, IoTask, ServerState},
 protocol::{messages::HwServerMessage::Redirect, messages::*, ProtocolDecoder},
 utils,
 };
 ssl::{
 HandshakeError, MidHandshakeSslStream, Ssl, SslContext, SslContextBuilder, SslFiletype,
 SslMethod, SslOptions, SslStream, SslStreamBuilder, SslVerifyMode,
 },
 };
-use std::time::Duration;
 const MAX_BYTES_PER_READ: usize = 2048;
-const SEND_PING_TIMEOUT: Duration = Duration::from_secs(30);
+const SEND_PING_TIMEOUT: Duration = Duration::from_secs(5);
-const DROP_CLIENT_TIMEOUT: Duration = Duration::from_secs(30);
+const DROP_CLIENT_TIMEOUT: Duration = Duration::from_secs(5);
+const MAX_TIMEOUT: usize = DROP_CLIENT_TIMEOUT.as_secs() as usize;
 const PING_PROBES_COUNT: u8 = 2;
 #[derive(Hash, Eq, PartialEq, Copy, Clone)]
 pub enum NetworkClientState {
 Idle,
 #[cfg(feature = "tls-connections")]
 SslStream(SslStream<TcpStream>),
 }
 impl ClientSocket {
-fn inner(&self) -> &TcpStream {
+fn inner_mut(&mut self) -> &mut TcpStream {
 match self {
 ClientSocket::Plain(stream) => stream,
 #[cfg(feature = "tls-connections")]
-ClientSocket::SslHandshake(Some(builder)) => builder.get_ref(),
+ClientSocket::SslHandshake(Some(builder)) => builder.get_mut(),
 #[cfg(feature = "tls-connections")]
 ClientSocket::SslHandshake(None) => unreachable!(),
 #[cfg(feature = "tls-connections")]
-ClientSocket::SslStream(ssl_stream) => ssl_stream.get_ref(),
+ClientSocket::SslStream(ssl_stream) => ssl_stream.get_mut(),
 }
 }
 }
 pub struct NetworkClient {
 id: ClientId,
 socket: ClientSocket,
 peer_addr: SocketAddr,
 decoder: ProtocolDecoder,
 buf_out: netbuf::Buf,
-timeout: timer::Timeout,
 pending_close: bool,
+timeout: Timeout,
+last_rx_time: Instant,
 }
 impl NetworkClient {
 pub fn new(
 id: ClientId,
 socket: ClientSocket,
 peer_addr: SocketAddr,
-timeout: timer::Timeout,
+timeout: Timeout,
 ) -> NetworkClient {
 NetworkClient {
 id,
 socket,
 peer_addr,
 decoder: ProtocolDecoder::new(),
 buf_out: netbuf::Buf::new(),
+pending_close: false,
 timeout,
-pending_close: false,
+last_rx_time: Instant::now(),
 }
 }
 #[cfg(feature = "tls-connections")]
 fn handshake_impl(
 };
 result
 }
 pub fn read(&mut self) -> NetworkResult<Vec<HwProtocolMessage>> {
-match self.socket {
+let result = match self.socket {
 ClientSocket::Plain(ref mut stream) => {
 NetworkClient::read_impl(&mut self.decoder, stream, self.id, &self.peer_addr)
 }
 #[cfg(feature = "tls-connections")]
 ClientSocket::SslHandshake(ref mut handshake_opt) => {
 }
 #[cfg(feature = "tls-connections")]
 ClientSocket::SslStream(ref mut stream) => {
 NetworkClient::read_impl(&mut self.decoder, stream, self.id, &self.peer_addr)
 }
-}
+};
+if let Ok(_) = result {
+self.last_rx_time = Instant::now();
+}
+result
 }
 fn write_impl<W: Write>(
 buf_out: &mut netbuf::Buf,
 destination: &mut W,
 ClientSocket::SslStream(ref mut stream) => {
 NetworkClient::write_impl(&mut self.buf_out, stream, self.pending_close)
 }
 };
-self.socket.inner().flush()?;
+self.socket.inner_mut().flush()?;
 result
 }
 pub fn send_raw_msg(&mut self, msg: &[u8]) {
 self.buf_out.write_all(msg).unwrap();
 pub fn send_string(&mut self, msg: &str) {
 self.send_raw_msg(&msg.as_bytes());
 }
-pub fn replace_timeout(&mut self, timeout: timer::Timeout) -> timer::Timeout {
+pub fn replace_timeout(&mut self, timeout: Timeout) -> Timeout {
 replace(&mut self.timeout, timeout)
 }
 pub fn has_pending_sends(&self) -> bool {
 !self.buf_out.is_empty()
 io_thread: IoThread,
 }
 #[cfg(feature = "official-server")]
 impl IoLayer {
-fn new() -> Self {
+fn new(waker: Waker) -> Self {
 Self {
 next_request_id: 0,
 request_queue: vec![],
-io_thread: IoThread::new(),
+io_thread: IoThread::new(waker),
 }
 }
 fn send(&mut self, client_id: ClientId, task: IoTask) {
 let request_id = self.next_request_id;
 SendPing { probes_count: u8 },
 DropClient,
 }
 struct TimerData(TimeoutEvent, ClientId);
+type NetworkTimeoutEvents = TimedEvents<TimerData, MAX_TIMEOUT>;
 pub struct NetworkLayer {
 listener: TcpListener,
 server_state: ServerState,
 clients: Slab<NetworkClient>,
 pending_cache: Vec<(ClientId, NetworkClientState)>,
 #[cfg(feature = "tls-connections")]
 ssl: ServerSsl,
 #[cfg(feature = "official-server")]
 io: IoLayer,
-timer: timer::Timer<TimerData>,
+timeout_events: NetworkTimeoutEvents,
 }
-fn register_read<E: Evented>(poll: &Poll, evented: &E, token: mio::Token) -> io::Result<()> {
+fn register_read<S: Source>(poll: &Poll, source: &mut S, token: mio::Token) -> io::Result<()> {
-poll.register(evented, token, Ready::readable(), PollOpt::edge())
+poll.registry().register(source, token, Interest::READABLE)
 }
 fn create_ping_timeout(
-timer: &mut timer::Timer<TimerData>,
+timeout_events: &mut NetworkTimeoutEvents,
 probes_count: u8,
 client_id: ClientId,
-) -> timer::Timeout {
+) -> Timeout {
-timer.set_timeout(
+timeout_events.set_timeout(
-SEND_PING_TIMEOUT,
+NonZeroU32::new(SEND_PING_TIMEOUT.as_secs() as u32).unwrap(),
 TimerData(TimeoutEvent::SendPing { probes_count }, client_id),
 )
 }
-fn create_drop_timeout(timer: &mut timer::Timer<TimerData>, client_id: ClientId) -> timer::Timeout {
+fn create_drop_timeout(timeout_events: &mut NetworkTimeoutEvents, client_id: ClientId) -> Timeout {
-timer.set_timeout(
+timeout_events.set_timeout(
-DROP_CLIENT_TIMEOUT,
+NonZeroU32::new(DROP_CLIENT_TIMEOUT.as_secs() as u32).unwrap(),
 TimerData(TimeoutEvent::DropClient, client_id),
 )
 }
 impl NetworkLayer {
-pub fn register(&self, poll: &Poll) -> io::Result<()> {
+pub fn register(&mut self, poll: &Poll) -> io::Result<()> {
-register_read(poll, &self.listener, utils::SERVER_TOKEN)?;
+register_read(poll, &mut self.listener, utils::SERVER_TOKEN)?;
 #[cfg(feature = "tls-connections")]
-register_read(poll, &self.ssl.listener, utils::SECURE_SERVER_TOKEN)?;
+register_read(poll, &mut self.ssl.listener, utils::SECURE_SERVER_TOKEN)?;
-register_read(poll, &self.timer, utils::TIMER_TOKEN)?;
-#[cfg(feature = "official-server")]
-self.io.io_thread.register_rx(poll, utils::IO_TOKEN)?;
 Ok(())
 }
 fn deregister_client(&mut self, poll: &Poll, id: ClientId, is_error: bool) {
 if let Some(ref mut client) = self.clients.get_mut(id) {
-poll.deregister(client.socket.inner())
+poll.registry()
+.deregister(client.socket.inner_mut())
 .expect("could not deregister socket");
 if client.has_pending_sends() && !is_error {
 info!(
 "client {} ({}) pending removal",
 client.id, client.peer_addr
 );
 client.pending_close = true;
-poll.register(
+poll.registry()
-client.socket.inner(),
+.register(client.socket.inner_mut(), Token(id), Interest::WRITABLE)
-Token(id),
+.unwrap_or_else(|_| {
-Ready::writable(),
+self.clients.remove(id);
-PollOpt::edge(),
+});
-)
-.unwrap_or_else(|_| {
-self.clients.remove(id);
-});
 } else {
 info!("client {} ({}) removed", client.id, client.peer_addr);
 self.clients.remove(id);
 }
 #[cfg(feature = "official-server")]
 }
 fn register_client(
 &mut self,
 poll: &Poll,
-client_socket: ClientSocket,
+mut client_socket: ClientSocket,
 addr: SocketAddr,
 ) -> io::Result<ClientId> {
 let entry = self.clients.vacant_entry();
 let client_id = entry.key();
-poll.register(
+poll.registry().register(
-client_socket.inner(),
+client_socket.inner_mut(),
 Token(client_id),
-Ready::readable() | Ready::writable(),
+Interest::READABLE | Interest::WRITABLE,
-PollOpt::edge(),
 )?;
 let client = NetworkClient::new(
 client_id,
 client_socket,
 addr,
-create_ping_timeout(&mut self.timer, PING_PROBES_COUNT - 1, client_id),
+create_ping_timeout(&mut self.timeout_events, PING_PROBES_COUNT - 1, client_id),
 );
 info!("client {} ({}) added", client.id, client.peer_addr);
 entry.insert(client);
 Ok(client_id)
 self.io.send(client_id, task);
 }
 }
 }
-pub fn handle_timeout(&mut self, poll: &Poll) -> io::Result<()> {
+pub fn handle_timeout(&mut self, poll: &mut Poll) -> io::Result<()> {
-while let Some(TimerData(event, client_id)) = self.timer.poll() {
+for TimerData(event, client_id) in self.timeout_events.poll(Instant::now()) {
-match event {
+if let Some(client) = self.clients.get_mut(client_id) {
-TimeoutEvent::SendPing { probes_count } => {
+if client.last_rx_time.elapsed() > SEND_PING_TIMEOUT {
-if let Some(ref mut client) = self.clients.get_mut(client_id) {
+match event {
-client.send_string(&HwServerMessage::Ping.to_raw_protocol());
+TimeoutEvent::SendPing { probes_count } => {
-client.write()?;
+client.send_string(&HwServerMessage::Ping.to_raw_protocol());
-let timeout = if probes_count != 0 {
+client.write()?;
-create_ping_timeout(&mut self.timer, probes_count - 1, client_id)
+let timeout = if probes_count != 0 {
-} else {
+create_ping_timeout(
-create_drop_timeout(&mut self.timer, client_id)
+&mut self.timeout_events,
-};
+probes_count - 1,
-client.replace_timeout(timeout);
+client_id,
+)
+} else {
+create_drop_timeout(&mut self.timeout_events, client_id)
+};
+client.replace_timeout(timeout);
+}
+TimeoutEvent::DropClient => {
+client.send_string(
+&HwServerMessage::Bye("Ping timeout".to_string()).to_raw_protocol(),
+);
+let _res = client.write();
+self.operation_failed(
+poll,
+client_id,
+&ErrorKind::TimedOut.into(),
+"No ping response",
+)?;
+}
 }
-}
+} else {
-TimeoutEvent::DropClient => {
+client.replace_timeout(create_ping_timeout(
-if let Some(ref mut client) = self.clients.get_mut(client_id) {
+&mut self.timeout_events,
-client.send_string(
+PING_PROBES_COUNT - 1,
-&HwServerMessage::Bye("Ping timeout".to_string()).to_raw_protocol(),
-);
-let _res = client.write();
-}
-self.operation_failed(
-poll,
 client_id,
-&ErrorKind::TimedOut.into(),
+));
-"No ping response",
-)?;
 }
 }
 }
 Ok(())
 }
 self.client_error(poll, client_id)
 }
 pub fn client_readable(&mut self, poll: &Poll, client_id: ClientId) -> io::Result<()> {
 let messages = if let Some(ref mut client) = self.clients.get_mut(client_id) {
-let timeout = client.replace_timeout(create_ping_timeout(
-&mut self.timer,
-PING_PROBES_COUNT - 1,
-client_id,
-));
-self.timer.cancel_timeout(&timeout);
 client.read()
 } else {
 warn!("invalid readable client: {}", client_id);
 Ok((Vec::new(), NetworkClientState::Idle))
 };
 Ok(())
 }
 pub fn has_pending_operations(&self) -> bool {
-!self.pending.is_empty()
+!self.pending.is_empty() || !self.timeout_events.is_empty()
 }
 pub fn on_idle(&mut self, poll: &Poll) -> io::Result<()> {
 if self.has_pending_operations() {
 let mut cache = replace(&mut self.pending_cache, Vec::new());
 listener,
 context: builder.build(),
 }
 }
-pub fn build(self) -> NetworkLayer {
+pub fn build(self, poll: &Poll) -> NetworkLayer {
 let server_state = ServerState::new(self.clients_capacity, self.rooms_capacity);
 let clients = Slab::with_capacity(self.clients_capacity);
 let pending = HashSet::with_capacity(2 * self.clients_capacity);
 let pending_cache = Vec::with_capacity(2 * self.clients_capacity);
-let timer = timer::Builder::default().build();
+let timeout_events = NetworkTimeoutEvents::new();
+#[cfg(feature = "official-server")]
+let waker = Waker::new(poll.registry(), utils::IO_TOKEN)
+.expect("Unable to create a waker for the IO thread");
 NetworkLayer {
 listener: self.listener.expect("No listener provided"),
 server_state,
 clients,
 #[cfg(feature = "tls-connections")]
 ssl: Self::create_ssl_context(
 self.secure_listener.expect("No secure listener provided"),
 ),
 #[cfg(feature = "official-server")]
-io: IoLayer::new(),
+io: IoLayer::new(waker),
-timer,
+timeout_events,
 }
 }
 }

changeset 15800	6af892a0a4b8
parent 15531	ede5f4ec48f3
child 15804	747278149393