[pbs-devel] [PATCH proxmox-backup 1/2] termproxy: let users stop the termproxy task

[pbs-devel] [PATCH proxmox-backup 1/2] termproxy: let users stop the termproxy task

[Dominik Csapak]

for that we have to do a select on the workers abort_future

Signed-off-by: Dominik Csapak <d.csapak@proxmox.com>
---
 src/api2/node.rs | 35 +++++++++++++++++++++--------------
 1 file changed, 21 insertions(+), 14 deletions(-)

diff --git a/src/api2/node.rs b/src/api2/node.rs
index 431627c..b572a9c 100644
--- a/src/api2/node.rs
+++ b/src/api2/node.rs
@@ -4,7 +4,7 @@ use std::os::unix::io::AsRawFd;
 use anyhow::{bail, format_err, Error};
 use futures::{
     future::{FutureExt, TryFutureExt},
-    try_join,
+    select,
 };
 use hyper::body::Body;
 use hyper::http::request::Parts;
@@ -169,9 +169,10 @@ async fn termproxy(
 
             let mut cmd = tokio::process::Command::new("/usr/bin/termproxy");
 
-            cmd.args(&arguments);
-            cmd.stdout(std::process::Stdio::piped());
-            cmd.stderr(std::process::Stdio::piped());
+            cmd.args(&arguments)
+                .kill_on_drop(true)
+                .stdout(std::process::Stdio::piped())
+                .stderr(std::process::Stdio::piped());
 
             let mut child = cmd.spawn().expect("error executing termproxy");
 
@@ -184,7 +185,7 @@ async fn termproxy(
                 while let Some(line) = reader.next_line().await? {
                     worker_stdout.log(line);
                 }
-                Ok(())
+                Ok::<(), Error>(())
             };
 
             let worker_stderr = worker.clone();
@@ -193,18 +194,24 @@ async fn termproxy(
                 while let Some(line) = reader.next_line().await? {
                     worker_stderr.warn(line);
                 }
-                Ok(())
+                Ok::<(), Error>(())
             };
 
-            let (exit_code, _, _) = try_join!(child, stdout_fut, stderr_fut)?;
-            if !exit_code.success() {
-                match exit_code.code() {
-                    Some(code) => bail!("termproxy exited with {}", code),
-                    None => bail!("termproxy exited by signal"),
-                }
+            select!{
+                res = child.fuse() => {
+                    let exit_code = res?;
+                    if !exit_code.success() {
+                        match exit_code.code() {
+                            Some(code) => bail!("termproxy exited with {}", code),
+                            None => bail!("termproxy exited by signal"),
+                        }
+                    }
+                    Ok(())
+                },
+                res = stdout_fut.fuse() => res,
+                res = stderr_fut.fuse() => res,
+                res = worker.abort_future().fuse() => res.map_err(Error::from),
             }
-
-            Ok(())
         },
     )?;
 
-- 
2.20.1

[pbs-devel] [RFC PATCH proxmox-backup 2/2] server/state: add spawn_internal_task and use it for websockets

[Dominik Csapak]

is a helper to spawn an internal tokio task without it showing up
in the task list

it is still tracked for reload and notifies the last_worker_listeners

this enables the console to survive a reload of proxmox-backup-proxy

Signed-off-by: Dominik Csapak <d.csapak@proxmox.com>
---
i am not completely sure if this is the right way...
alternatively, we could of course have WorkerTasks that simply do not
show up in the task list... though for what we are doing here,
it is much overhead thats not really needed

 src/api2/node.rs    |  2 +-
 src/server/state.rs | 36 +++++++++++++++++++++++++++++-------
 2 files changed, 30 insertions(+), 8 deletions(-)

diff --git a/src/api2/node.rs b/src/api2/node.rs
index b572a9c..b80a661 100644
--- a/src/api2/node.rs
+++ b/src/api2/node.rs
@@ -267,7 +267,7 @@ fn upgrade_to_websocket(
 
         let (ws, response) = WebSocket::new(parts.headers)?;
 
-        tokio::spawn(async move {
+        crate::server::spawn_internal_task(async move {
             let conn: Upgraded = match req_body.on_upgrade().map_err(Error::from).await {
                 Ok(upgraded) => upgraded,
                 _ => bail!("error"),
diff --git a/src/server/state.rs b/src/server/state.rs
index 41e910d..d073243 100644
--- a/src/server/state.rs
+++ b/src/server/state.rs
@@ -19,6 +19,7 @@ pub struct ServerState {
     pub shutdown_listeners: BroadcastData<()>,
     pub last_worker_listeners: BroadcastData<()>,
     pub worker_count: usize,
+    pub task_count: usize,
     pub reload_request: bool,
 }
 
@@ -28,6 +29,7 @@ lazy_static! {
         shutdown_listeners: BroadcastData::new(),
         last_worker_listeners: BroadcastData::new(),
         worker_count: 0,
+        task_count: 0,
         reload_request: false,
     });
 }
@@ -101,20 +103,40 @@ pub fn last_worker_future() ->  impl Future<Output = Result<(), Error>> {
 }
 
 pub fn set_worker_count(count: usize) {
+    SERVER_STATE.lock().unwrap().worker_count = count;
+
+    check_last_worker();
+}
+
+pub fn check_last_worker() {
     let mut data = SERVER_STATE.lock().unwrap();
-    data.worker_count = count;
 
-    if !(data.mode == ServerMode::Shutdown && data.worker_count == 0) { return; }
+    if !(data.mode == ServerMode::Shutdown && data.worker_count == 0 && data.task_count == 0) { return; }
 
     data.last_worker_listeners.notify_listeners(Ok(()));
 }
 
-
-pub fn check_last_worker() {
-
+/// Spawns a tokio task that will be tracked for reload
+/// and if it is finished, notify the last_worker_listener if we
+/// are in shutdown mode
+pub fn spawn_internal_task<T>(task: T)
+where
+    T: Future + Send + 'static,
+    T::Output: Send + 'static,
+{
     let mut data = SERVER_STATE.lock().unwrap();
+    data.task_count += 1;
 
-    if !(data.mode == ServerMode::Shutdown && data.worker_count == 0) { return; }
+    tokio::spawn(async move {
+        let _ = tokio::spawn(task).await; // ignore errors
 
-    data.last_worker_listeners.notify_listeners(Ok(()));
+        { // drop mutex
+            let mut data = SERVER_STATE.lock().unwrap();
+            if data.task_count > 0 {
+                data.task_count -= 1;
+            }
+        }
+
+        check_last_worker();
+    });
 }
-- 
2.20.1

Re: [pbs-devel] [RFC PATCH proxmox-backup 2/2] server/state: add spawn_internal_task and use it for websockets

[Dietmar Maurer]

> is a helper to spawn an internal tokio task without it showing up
> in the task list

Can't we use a normal worker task?

Re: [pbs-devel] [RFC PATCH proxmox-backup 2/2] server/state: add spawn_internal_task and use it for websockets

[Dominik Csapak]

On 7/24/20 9:29 AM, Dietmar Maurer wrote:
>> is a helper to spawn an internal tokio task without it showing up
>> in the task list
> 
> Can't we use a normal worker task?
> 

we could, but as i wrote in the mail, we'd either have

* 2 visible WorkerTasks per terminal connection (one for the process, 
one for the websocket) -> this would be confusing for users..

* would have to make some workertasks invisible somehow (no upid? how to 
keep track of it then) -> much work for little gain?

* we would have to refactor the whole terminal handling with termproxy,
so that in only gets started during the websocket connection
-> this is a non-trivial refactoring and would also have to be done on
pve/pmg since they all share some code (at least the gui, which
also has to be aware of some of this)

so i think a small wrapper around tokio::spawn is not so bad

Re: [pbs-devel] [PATCH proxmox-backup 1/2] termproxy: let users stop the termproxy task

[Dominik Csapak]

just fyi before anyone complains ^^

if the last console process is stopped, tokio currently
leaves a zombie (that gets reaped when another console is started)

this seems to be a bug in tokio, so i opened a github issue:

https://github.com/tokio-rs/tokio/issues/2685

[pbs-devel] applied-series: [PATCH proxmox-backup 1/2] termproxy: let users stop the termproxy task

[Thomas Lamprecht]

Am 7/23/20 um 3:20 PM schrieb Dominik Csapak:
> for that we have to do a select on the workers abort_future
> 
> Signed-off-by: Dominik Csapak <d.csapak@proxmox.com>
> ---
>  src/api2/node.rs | 35 +++++++++++++++++++++--------------
>  1 file changed, 21 insertions(+), 14 deletions(-)
> 

applied series for now, it's an non-invasive solution fixing the sole issues
we have for moving the current version to the public for testing.

It's not set in stone and can be improved or replaced by a mechanism
deemed more appropriate at any time.