tests/worker-task-abort.rs

   1 use anyhow::{bail, Error};
   2
   3 extern crate proxmox_backup;
   4
   5 extern crate tokio;
   6 extern crate nix;
   7
   8 use proxmox_sys::fs::CreateOptions;
   9 use proxmox_lang::try_block;
  10 use proxmox_sys::{task_log, WorkerTaskContext};
  11
  12 use pbs_api_types::{Authid, UPID};
  13
  14 use proxmox_rest_server::{CommandSocket, WorkerTask};
  15
  16 fn garbage_collection(worker: &WorkerTask) -> Result<(), Error> {
  17
  18     task_log!(worker, "start garbage collection");
  19
  20     for i in 0..50 {
  21         worker.check_abort()?;
  22
  23         task_log!(worker, "progress {}", i);
  24
  25         std::thread::sleep(std::time::Duration::from_millis(10));
  26     }
  27
  28     task_log!(worker, "end garbage collection");
  29
  30     Ok(())
  31 }
  32
  33
  34 #[test]
  35 #[ignore]
  36 fn worker_task_abort() -> Result<(), Error> {
  37     let uid = nix::unistd::Uid::current();
  38     let gid = nix::unistd::Gid::current();
  39
  40     let file_opts = CreateOptions::new().owner(uid).group(gid);
  41     proxmox_rest_server::init_worker_tasks("./target/tasklogtestdir".into(), file_opts.clone())?;
  42
  43     use std::sync::{Arc, Mutex};
  44
  45     let errmsg: Arc<Mutex<Option<String>>> = Arc::new(Mutex::new(None));
  46     let errmsg1 = errmsg.clone();
  47
  48     let rt = tokio::runtime::Runtime::new().unwrap();
  49     rt.block_on(async move {
  50
  51         let mut commando_sock = CommandSocket::new(
  52             proxmox_rest_server::our_ctrl_sock(), nix::unistd::Gid::current());
  53
  54         let init_result: Result<(), Error> = try_block!({
  55             proxmox_rest_server::register_task_control_commands(&mut commando_sock)?;
  56             Ok(())
  57         });
  58
  59         if let Err(err) = init_result {
  60             eprintln!("unable to start daemon - {}", err);
  61             return;
  62         }
  63
  64        if let Err(err) = commando_sock.spawn() {
  65             eprintln!("unable to spawn command socket - {}", err);
  66             return;
  67         }
  68
  69         let errmsg = errmsg1.clone();
  70         let res = WorkerTask::new_thread(
  71             "garbage_collection",
  72             None,
  73             Authid::root_auth_id().to_string(),
  74             true,
  75             move |worker| {
  76                 println!("WORKER {}", worker);
  77
  78                 let result = garbage_collection(&worker);
  79                 proxmox_rest_server::request_shutdown();
  80
  81                 if let Err(err) = result {
  82                     println!("got expected error: {}", err);
  83                 } else {
  84                     let mut data = errmsg.lock().unwrap();
  85                     *data = Some(String::from("thread finished - seems abort did not work as expected"));
  86                 }
  87
  88                 Ok(())
  89             },
  90         );
  91
  92         match res {
  93             Err(err) => {
  94                 println!("unable to start worker - {}", err);
  95             }
  96             Ok(wid) => {
  97                 println!("WORKER: {}", wid);
  98                 proxmox_rest_server::abort_worker_nowait(wid.parse::<UPID>().unwrap());
  99                 proxmox_rest_server::wait_for_local_worker(&wid).await.unwrap();
 100              }
 101         }
 102     });
 103
 104     let data = errmsg.lock().unwrap();
 105     match *data {
 106         Some(ref err) => bail!("Error: {}", err),
 107         None => {},
 108     }
 109
 110     Ok(())
 111 }