berkus
/
actix-net
mirror of https://github.com/fafhrd91/actix-net
1
0
Fork 0
Code Issues Releases Wiki Activity

add tests for socket module

This commit is contained in:
fakeshadow 2020-10-22 10:56:37 +08:00
parent 5a0e236939
commit 30d8971209
16 changed files with 301 additions and 255 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
LICENSE-APACHE
View File

@ -1,4 +1,4 @@
Apache License ../LICENSE-APACHE
Apache License
Version 2.0, January 2004
http://www.apache.org/licenses/

2
LICENSE-MIT
View File

@ -1,4 +1,4 @@
Copyright (c) 2017 Nikolay Kim ../LICENSE-MIT
Copyright (c) 2017 Nikolay Kim
Permission is hereby granted, free of charge, to any
person obtaining a copy of this software and associated

1
actix-codec/LICENSE-APACHE
View File

@ -1 +0,0 @@
../LICENSE-APACHE

1
actix-codec/LICENSE-APACHE Symbolic link
View File

@ -0,0 +1 @@
../LICENSE-APACHE

1
actix-codec/LICENSE-MIT
View File

@ -1 +0,0 @@
../LICENSE-MIT

1
actix-codec/LICENSE-MIT Symbolic link
View File

@ -0,0 +1 @@
../LICENSE-MIT

1
actix-rt/LICENSE-APACHE
View File

@ -1 +0,0 @@
../LICENSE-APACHE

1
actix-rt/LICENSE-APACHE Symbolic link
View File

@ -0,0 +1 @@
../LICENSE-APACHE

1
actix-rt/LICENSE-MIT
View File

@ -1 +0,0 @@
../LICENSE-MIT

1
actix-rt/LICENSE-MIT Symbolic link
View File

@ -0,0 +1 @@
../LICENSE-MIT

1
actix-server/LICENSE-APACHE
View File

@ -1 +0,0 @@
../LICENSE-APACHE

1
actix-server/LICENSE-APACHE Symbolic link
View File

@ -0,0 +1 @@
../LICENSE-APACHE

1
actix-server/LICENSE-MIT
View File

@ -1 +0,0 @@
../LICENSE-MIT

1
actix-server/LICENSE-MIT Symbolic link
View File

@ -0,0 +1 @@
../LICENSE-MIT

43
actix-server/src/accept.rs
View File

@ -14,17 +14,21 @@ use crate::worker::{Conn, WorkerHandle};
use crate::Token;
struct ServerSocketInfo {
// addr for socket. mainly used for logging.
addr: SocketAddr,
// be ware this is the crate token for identify socket and should not be confused with
// mio::Token
token: Token,
sock: MioSocketListener,
// timeout is used to mark the time this socket should be reregistered after an error.
lst: MioSocketListener,
// timeout is used to mark the deadline when this socket's listener should be registered again
// after an error.
timeout: Option<Instant>,
}
/// Accept loop would live with `ServerBuilder`.
///
/// It's tasked with construct `Poll` instance and `WakerQueue` which would be distributed to
/// `Accept` and `WorkerClient` accordingly.
/// `Accept` and `Worker`.
///
/// It would also listen to `ServerCommand` and push interests to `WakerQueue`.
pub(crate) struct AcceptLoop {
@ -138,7 +142,7 @@ impl Accept {
entry.insert(ServerSocketInfo {
addr,
token: hnd_token,
sock,
lst: sock,
timeout: None,
});
}
@ -193,11 +197,11 @@ impl Accept {
Ok(WakerInterest::Stop) => {
return self.deregister_all(&mut sockets);
}
// a new worker thread is made and it's client would be added to Accept
Ok(WakerInterest::Worker(worker)) => {
// a new worker thread is made and it's handle would be added to Accept
Ok(WakerInterest::Worker(handle)) => {
// maybe we want to recover from a backpressure.
self.maybe_backpressure(&mut sockets, false);
self.workers.push(worker);
self.workers.push(handle);
}
// got timer interest and it's time to try register socket(s) again.
Ok(WakerInterest::Timer) => self.process_timer(&mut sockets),
@ -221,8 +225,8 @@ impl Accept {
fn process_timer(&self, sockets: &mut Slab<ServerSocketInfo>) {
let now = Instant::now();
for (token, info) in sockets.iter_mut() {
// only the sockets have an associate timeout value was de registered.
sockets.iter_mut().for_each(|(token, info)| {
// only the ServerSocketInfo have an associate timeout value was de registered.
if let Some(inst) = info.timeout.take() {
if now > inst {
self.register_logged(token, info);
@ -230,13 +234,13 @@ impl Accept {
info.timeout = Some(inst);
}
}
}
});
}
#[cfg(not(target_os = "windows"))]
fn register(&self, token: usize, info: &mut ServerSocketInfo) -> io::Result<()> {
self.poll.registry().register(
&mut info.sock,
&mut info.lst,
MioToken(token + DELTA),
Interest::READABLE,
)
@ -250,13 +254,13 @@ impl Accept {
self.poll
.registry()
.register(
&mut info.sock,
&mut info.lst,
mio::Token(token + DELTA),
Interest::READABLE,
)
.or_else(|_| {
self.poll.registry().reregister(
&mut info.sock,
&mut info.lst,
mio::Token(token + DELTA),
Interest::READABLE,
)
@ -271,7 +275,7 @@ impl Accept {
}
fn deregister(&self, info: &mut ServerSocketInfo) -> io::Result<()> {
self.poll.registry().deregister(&mut info.sock)
self.poll.registry().deregister(&mut info.lst)
}
fn deregister_all(&self, sockets: &mut Slab<ServerSocketInfo>) {
@ -305,7 +309,7 @@ impl Accept {
Err(tmp) => {
// worker lost contact and could be gone. a message is sent to
// `ServerBuilder` future to notify it a new worker should be made.
// after that remove the fault worker and enter backpressure if necessary.
// after that remove the fault worker.
self.srv.worker_faulted(self.workers[self.next].idx);
msg = tmp;
self.workers.swap_remove(self.next);
@ -363,7 +367,7 @@ impl Accept {
fn accept(&mut self, sockets: &mut Slab<ServerSocketInfo>, token: usize) {
loop {
let msg = if let Some(info) = sockets.get_mut(token) {
match info.sock.accept() {
match info.lst.accept() {
Ok(Some((io, addr))) => Conn {
io,
token: info.token,
@ -373,14 +377,15 @@ impl Accept {
Err(ref e) if e.kind() == io::ErrorKind::WouldBlock => return,
Err(ref e) if connection_error(e) => continue,
Err(e) => {
// deregister socket temporary
// deregister socket listener temporary
error!("Error accepting connection: {}", e);
if let Err(err) = self.poll.registry().deregister(&mut info.sock) {
if let Err(err) = self.poll.registry().deregister(&mut info.lst) {
error!("Can not deregister server socket {}", err);
}
// sleep after error. write the timeout to socket info as later the poll
// would need it mark which socket and when it should be registered.
// would need it mark which socket and when it's listener should be
// registered.
info.timeout = Some(Instant::now() + Duration::from_millis(500));
// after the sleep a Timer interest is sent to Accept Poll

26
actix-server/src/builder.rs
View File

@ -300,8 +300,7 @@ impl ServerBuilder {
fn start_worker(&self, idx: usize, waker: WakerQueue) -> WorkerHandle {
let avail = WorkerAvailability::new(waker);
let services: Vec<Box<dyn InternalServiceFactory>> =
self.services.iter().map(|v| v.clone_factory()).collect();
let services = self.services.iter().map(|v| v.clone_factory()).collect();
Worker::start(idx, services, avail, self.shutdown_timeout)
}
@ -376,16 +375,13 @@ impl ServerBuilder {
let _ = tx.send(());
}
if exit {
spawn(
async {
spawn(async {
sleep_until(
Instant::now() + Duration::from_millis(300),
)
.await;
System::current().stop();
}
.boxed(),
);
});
}
ready(())
}),
@ -393,14 +389,10 @@ impl ServerBuilder {
} else {
// we need to stop system if server was spawned
if self.exit {
spawn(
sleep_until(Instant::now() + Duration::from_millis(300)).then(
|_| {
spawn(async {
sleep_until(Instant::now() + Duration::from_millis(300)).await;
System::current().stop();
ready(())
},
),
);
});
}
if let Some(tx) = completion {
let _ = tx.send(());
@ -434,9 +426,9 @@ impl ServerBuilder {
break;
}
let worker = self.start_worker(new_idx, self.accept.waker_owned());
self.workers.push((new_idx, worker.clone()));
self.accept.wake(WakerInterest::Worker(worker));
let handle = self.start_worker(new_idx, self.accept.waker_owned());
self.workers.push((new_idx, handle.clone()));
self.accept.wake(WakerInterest::Worker(handle));
}
}
}

18
actix-server/src/config.rs
View File

@ -5,14 +5,15 @@ use std::{fmt, io, net};
use actix_rt::net::TcpStream;
use actix_service as actix;
use actix_utils::counter::CounterGuard;
use futures_util::future::{ready, FutureExt, LocalBoxFuture};
use futures_util::future::ready;
use log::error;
use super::builder::bind_addr;
use super::service::{
use crate::builder::bind_addr;
use crate::service::{
BoxedServerService, InternalServiceFactory, ServerMessage, StreamService,
};
use super::Token;
use crate::LocalBoxFuture;
use crate::Token;
pub struct ServiceConfig {
pub(crate) services: Vec<(String, net::TcpListener)>,
@ -233,7 +234,7 @@ impl ServiceRuntime {
where
F: Future<Output = ()> + 'static,
{
self.onstart.push(fut.boxed_local())
self.onstart.push(Box::pin(fut))
}
}
@ -264,14 +265,14 @@ where
type Request = (Option<CounterGuard>, ServerMessage);
type Response = ();
type Error = ();
type InitError = ();
type Config = ();
type Service = BoxedServerService;
type InitError = ();
type Future = LocalBoxFuture<'static, Result<BoxedServerService, ()>>;
fn new_service(&self, _: ()) -> Self::Future {
let fut = self.inner.new_service(());
async move {
Box::pin(async move {
match fut.await {
Ok(s) => Ok(Box::new(StreamService::new(s)) as BoxedServerService),
Err(e) => {
@ -279,7 +280,6 @@ where
Err(())
}
}
}
.boxed_local()
})
}
}

69
actix-server/src/socket.rs
View File

@ -30,6 +30,26 @@ pub(crate) enum StdListener {
Uds(StdUnixListener),
}
impl fmt::Debug for StdListener {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
match *self {
StdListener::Tcp(ref lst) => write!(f, "{:?}", lst),
#[cfg(all(unix))]
StdListener::Uds(ref lst) => write!(f, "{:?}", lst),
}
}
}
impl fmt::Display for StdListener {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
match *self {
StdListener::Tcp(ref lst) => write!(f, "{}", lst.local_addr().ok().unwrap()),
#[cfg(unix)]
StdListener::Uds(ref lst) => write!(f, "{:?}", lst.local_addr().ok().unwrap()),
}
}
}
pub(crate) enum SocketAddr {
Tcp(StdTcpSocketAddr),
#[cfg(unix)]
@ -64,16 +84,6 @@ impl fmt::Debug for SocketAddr {
}
}
impl fmt::Display for StdListener {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
match *self {
StdListener::Tcp(ref lst) => write!(f, "{}", lst.local_addr().ok().unwrap()),
#[cfg(unix)]
StdListener::Uds(ref lst) => write!(f, "{:?}", lst.local_addr().ok().unwrap()),
}
}
}
impl StdListener {
pub(crate) fn local_addr(&self) -> SocketAddr {
match self {
@ -226,3 +236,42 @@ impl FromStream for UnixStream {
}
}
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn socket_addr() {
use socket2::{Domain, SockAddr, Socket, Type};
let addr = SocketAddr::Tcp("127.0.0.1:8080".parse().unwrap());
assert!(format!("{:?}", addr).contains("127.0.0.1:8080"));
assert_eq!(format!("{}", addr), "127.0.0.1:8080");
let addr: StdTcpSocketAddr = "127.0.0.1:0".parse().unwrap();
let socket = Socket::new(Domain::ipv4(), Type::stream(), None).unwrap();
socket.set_reuse_address(true).unwrap();
socket.bind(&SockAddr::from(addr)).unwrap();
let tcp = socket.into_tcp_listener();
let lst = StdListener::Tcp(tcp);
assert!(format!("{:?}", lst).contains("TcpListener"));
assert!(format!("{}", lst).contains("127.0.0.1"));
}
#[test]
#[cfg(unix)]
fn uds() {
let _ = std::fs::remove_file("/tmp/sock.xxxxx");
if let Ok(socket) = StdUnixListener::bind("/tmp/sock.xxxxx") {
let addr = socket.local_addr().expect("Couldn't get local address");
let a = SocketAddr::Uds(addr);
assert!(format!("{:?}", a).contains("/tmp/sock.xxxxx"));
assert!(format!("{}", a).contains("/tmp/sock.xxxxx"));
let lst = StdListener::Uds(socket);
assert!(format!("{:?}", lst).contains("/tmp/sock.xxxxx"));
assert!(format!("{}", lst).contains("/tmp/sock.xxxxx"));
}
}
}

4
actix-server/src/waker_queue.rs
View File

@ -57,8 +57,8 @@ pub(crate) enum WakerInterest {
Resume,
Stop,
/// `Timer` is an interest sent as a delayed future. When an error happens on accepting
/// connection `Accept` would deregister sockets temporary and wake up the poll and register
/// them again after the delayed future resolve.
/// connection `Accept` would deregister socket listener temporary and wake up the poll and
/// register them again after the delayed future resolve.
Timer,
/// `Worker` ins an interest happen after a worker runs into faulted state(This is determined by
/// if work can be sent to it successfully).`Accept` would be waked up and add the new

1
actix-utils/LICENSE-APACHE
View File

@ -1 +0,0 @@
../LICENSE-APACHE

1
actix-utils/LICENSE-APACHE Symbolic link
View File

@ -0,0 +1 @@
../LICENSE-APACHE

1
actix-utils/LICENSE-MIT
View File

@ -1 +0,0 @@
../LICENSE-MIT

1
actix-utils/LICENSE-MIT Symbolic link
View File

@ -0,0 +1 @@
../LICENSE-MIT

2
actix-utils/src/order.rs
View File

@ -94,8 +94,8 @@ where
type Request = S::Request;
type Response = S::Response;
type Error = InOrderError<S::Error>;
type InitError = Infallible;
type Transform = InOrderService<S>;
type InitError = Infallible;
type Future = Ready<Result<Self::Transform, Self::InitError>>;
fn new_transform(&self, service: S) -> Self::Future {