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

Merge branch 'master' into refactor/simplify_server_future

This commit is contained in:
fakeshadow 2021-04-04 08:50:00 -07:00 committed by GitHub
parent 429b3132d0 8079c50ddb
commit d533ca38b0
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
3 changed files with 16 additions and 23 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -243,7 +243,7 @@ impl ServiceRuntime {
type BoxedNewService = Box< type BoxedNewService = Box<
dyn BaseServiceFactory< dyn BaseServiceFactory<
(Option<CounterGuard>, MioStream), (CounterGuard, MioStream),
Response = (), Response = (),
Error = (), Error = (),
InitError = (), InitError = (),
@ -257,7 +257,7 @@ struct ServiceFactory<T> {
inner: T, inner: T,
} }
impl<T> BaseServiceFactory<(Option<CounterGuard>, MioStream)> for ServiceFactory<T> impl<T> BaseServiceFactory<(CounterGuard, MioStream)> for ServiceFactory<T>
where where
T: BaseServiceFactory<TcpStream, Config = ()>, T: BaseServiceFactory<TcpStream, Config = ()>,
T::Future: 'static, T::Future: 'static,

6
actix-server/src/service.rs
View File

@ -29,7 +29,7 @@ pub(crate) trait InternalServiceFactory: Send {
pub(crate) type BoxedServerService = Box< pub(crate) type BoxedServerService = Box<
dyn Service< dyn Service<
(Option<CounterGuard>, MioStream), (CounterGuard, MioStream),
Response = (), Response = (),
Error = (), Error = (),
Future = Ready<Result<(), ()>>, Future = Ready<Result<(), ()>>,
@ -50,7 +50,7 @@ impl<S, I> StreamService<S, I> {
} }
} }
impl<S, I> Service<(Option<CounterGuard>, MioStream)> for StreamService<S, I> impl<S, I> Service<(CounterGuard, MioStream)> for StreamService<S, I>
where where
S: Service<I>, S: Service<I>,
S::Future: 'static, S::Future: 'static,
@ -65,7 +65,7 @@ where
self.service.poll_ready(ctx).map_err(|_| ()) self.service.poll_ready(ctx).map_err(|_| ())
} }
fn call(&self, (guard, req): (Option<CounterGuard>, MioStream)) -> Self::Future { fn call(&self, (guard, req): (CounterGuard, MioStream)) -> Self::Future {
ready(match FromStream::from_mio(req) { ready(match FromStream::from_mio(req) {
Ok(stream) => { Ok(stream) => {
let f = self.service.call(stream); let f = self.service.call(stream);

29
actix-server/src/worker.rs
View File

@ -273,6 +273,13 @@ impl ServerWorker {
WorkerHandle::new(idx, tx1, tx2, avail) WorkerHandle::new(idx, tx1, tx2, avail)
} }
fn restart_service(&mut self, token: Token, idx: usize) {
let factory = &self.factories[idx];
trace!("Service {:?} failed, restarting", factory.name(token));
self.services[token.0].status = WorkerServiceStatus::Restarting;
self.state = WorkerState::Restarting(idx, token, factory.create());
}
fn shutdown(&mut self, force: bool) { fn shutdown(&mut self, force: bool) {
self.services self.services
.iter_mut() .iter_mut()
@ -387,13 +394,7 @@ impl Future for ServerWorker {
} }
Ok(false) => Poll::Pending, Ok(false) => Poll::Pending,
Err((token, idx)) => { Err((token, idx)) => {
trace!( self.restart_service(token, idx);
"Service {:?} failed, restarting",
self.factories[idx].name(token)
);
self.services[token.0].status = WorkerServiceStatus::Restarting;
self.state =
WorkerState::Restarting(idx, token, self.factories[idx].create());
self.poll(cx) self.poll(cx)
} }
}, },
@ -446,7 +447,7 @@ impl Future for ServerWorker {
// actively poll stream and handle worker command // actively poll stream and handle worker command
WorkerState::Available => loop { WorkerState::Available => loop {
match self.check_readiness(cx) { match self.check_readiness(cx) {
Ok(true) => (), Ok(true) => {}
Ok(false) => { Ok(false) => {
trace!("Worker is unavailable"); trace!("Worker is unavailable");
self.availability.set(false); self.availability.set(false);
@ -454,14 +455,8 @@ impl Future for ServerWorker {
return self.poll(cx); return self.poll(cx);
} }
Err((token, idx)) => { Err((token, idx)) => {
trace!( self.restart_service(token, idx);
"Service {:?} failed, restarting",
self.factories[idx].name(token)
);
self.availability.set(false); self.availability.set(false);
self.services[token.0].status = WorkerServiceStatus::Restarting;
self.state =
WorkerState::Restarting(idx, token, self.factories[idx].create());
return self.poll(cx); return self.poll(cx);
} }
} }
@ -470,9 +465,7 @@ impl Future for ServerWorker {
// handle incoming io stream // handle incoming io stream
Some(WorkerCommand(msg)) => { Some(WorkerCommand(msg)) => {
let guard = self.conns.get(); let guard = self.conns.get();
let _ = self.services[msg.token.0] let _ = self.services[msg.token.0].service.call((guard, msg.io));
.service
.call((Some(guard), msg.io));
} }
None => return Poll::Ready(()), None => return Poll::Ready(()),
}; };