update to tokio 0.2.0-alpha.4

Signed-off-by: Wolfgang Bumiller <w.bumiller@proxmox.com>
This commit is contained in:
Wolfgang Bumiller 2019-09-02 15:16:21 +02:00
parent 149b9d997f
commit 5d6ebfb8dc

View File

@ -1,13 +1,15 @@
//! Helpers for daemons/services. //! Helpers for daemons/services.
use std::ffi::CString; use std::ffi::CString;
use std::future::Future;
use std::os::raw::{c_char, c_int}; use std::os::raw::{c_char, c_int};
use std::os::unix::io::{AsRawFd, FromRawFd, IntoRawFd, RawFd}; use std::os::unix::io::{AsRawFd, FromRawFd, IntoRawFd, RawFd};
use std::os::unix::ffi::OsStrExt; use std::os::unix::ffi::OsStrExt;
use std::panic::UnwindSafe; use std::panic::UnwindSafe;
use std::pin::Pin;
use std::task::{Context, Poll};
use failure::*; use failure::*;
use tokio::prelude::*;
use proxmox::tools::io::{ReadExt, WriteExt}; use proxmox::tools::io::{ReadExt, WriteExt};
@ -48,14 +50,15 @@ impl Reloader {
/// the function provided in the `or_create` parameter to instantiate the new "first" instance. /// the function provided in the `or_create` parameter to instantiate the new "first" instance.
/// ///
/// Values created via this method will be remembered for later re-execution. /// Values created via this method will be remembered for later re-execution.
pub fn restore<T, F>(&mut self, name: &'static str, or_create: F) -> Result<T, Error> pub async fn restore<T, F, U>(&mut self, name: &'static str, or_create: F) -> Result<T, Error>
where where
T: Reloadable, T: Reloadable,
F: FnOnce() -> Result<T, Error>, F: FnOnce() -> U,
U: Future<Output = Result<T, Error>>,
{ {
let res = match std::env::var(name) { let res = match std::env::var(name) {
Ok(varstr) => T::restore(&varstr)?, Ok(varstr) => T::restore(&varstr)?,
Err(std::env::VarError::NotPresent) => or_create()?, Err(std::env::VarError::NotPresent) => or_create().await?,
Err(_) => bail!("variable {} has invalid value", name), Err(_) => bail!("variable {} has invalid value", name),
}; };
@ -194,48 +197,56 @@ impl Reloadable for tokio::net::TcpListener {
} }
} }
pub struct NotifyReady;
impl Future for NotifyReady {
type Output = Result<(), Error>;
fn poll(self: Pin<&mut Self>, _cx: &mut Context) -> Poll<Result<(), Error>> {
systemd_notify(SystemdNotify::Ready)?;
Poll::Ready(Ok(()))
}
}
/// This creates a future representing a daemon which reloads itself when receiving a SIGHUP. /// This creates a future representing a daemon which reloads itself when receiving a SIGHUP.
/// If this is started regularly, a listening socket is created. In this case, the file descriptor /// If this is started regularly, a listening socket is created. In this case, the file descriptor
/// number will be remembered in `PROXMOX_BACKUP_LISTEN_FD`. /// number will be remembered in `PROXMOX_BACKUP_LISTEN_FD`.
/// If the variable already exists, its contents will instead be used to restore the listening /// If the variable already exists, its contents will instead be used to restore the listening
/// socket. The finished listening socket is then passed to the `create_service` function which /// socket. The finished listening socket is then passed to the `create_service` function which
/// can be used to setup the TLS and the HTTP daemon. /// can be used to setup the TLS and the HTTP daemon.
pub fn create_daemon<F, S>( pub async fn create_daemon<F, S>(
address: std::net::SocketAddr, address: std::net::SocketAddr,
create_service: F, create_service: F,
) -> Result<impl Future<Output = ()>, Error> ) -> Result<(), Error>
where where
F: FnOnce(tokio::net::TcpListener) -> Result<S, Error>, F: FnOnce(tokio::net::TcpListener, NotifyReady) -> Result<S, Error>,
S: Future<Output = ()>, S: Future<Output = ()>,
{ {
let mut reloader = Reloader::new(); let mut reloader = Reloader::new();
let listener: tokio::net::TcpListener = reloader.restore( let listener: tokio::net::TcpListener = reloader.restore(
"PROXMOX_BACKUP_LISTEN_FD", "PROXMOX_BACKUP_LISTEN_FD",
move || Ok(tokio::net::TcpListener::bind(&address)?), move || async move { Ok(tokio::net::TcpListener::bind(&address).await?) },
)?; ).await?;
let service = create_service(listener)?; create_service(listener, NotifyReady)?.await;
let mut reloader = Some(reloader); let mut reloader = Some(reloader);
Ok(service crate::tools::request_shutdown(); // make sure we are in shutdown mode
.map(move |_| { if server::is_reload_request() {
crate::tools::request_shutdown(); // make sure we are in shutdown mode log::info!("daemon reload...");
if server::is_reload_request() { if let Err(e) = systemd_notify(SystemdNotify::Reloading) {
log::info!("daemon reload..."); log::error!("failed to notify systemd about the state change: {}", e);
if let Err(e) = systemd_notify(SystemdNotify::Reloading) { }
log::error!("failed to notify systemd about the state change: {}", e); if let Err(e) = reloader.take().unwrap().fork_restart() {
} log::error!("error during reload: {}", e);
if let Err(e) = reloader.take().unwrap().fork_restart() { let _ = systemd_notify(SystemdNotify::Status(format!("error during reload")));
log::error!("error during reload: {}", e); }
let _ = systemd_notify(SystemdNotify::Status(format!("error during reload"))); } else {
} log::info!("daemon shutting down...");
} else { }
log::info!("daemon shutting down..."); Ok(())
}
})
)
} }
#[link(name = "systemd")] #[link(name = "systemd")]