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::try_block;
   9 use proxmox::tools::fs::CreateOptions;
  10
  11 use pbs_api_types::{Authid, UPID};
  12 use pbs_tools::{task_log, task::WorkerTaskContext};
  13
  14 use proxmox_rest_server::{CommandoSocket, 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 = CommandoSocket::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             proxmox_rest_server::server_state_init()?;
  57             Ok(())
  58         });
  59
  60         if let Err(err) = init_result {
  61             eprintln!("unable to start daemon - {}", err);
  62             return;
  63         }
  64
  65        if let Err(err) = commando_sock.spawn() {
  66             eprintln!("unable to spawn command socket - {}", err);
  67             return;
  68         }
  69
  70         let errmsg = errmsg1.clone();
  71         let res = WorkerTask::new_thread(
  72             "garbage_collection",
  73             None,
  74             Authid::root_auth_id().to_string(),
  75             true,
  76             move |worker| {
  77                 println!("WORKER {}", worker);
  78
  79                 let result = garbage_collection(&worker);
  80                 proxmox_rest_server::request_shutdown();
  81
  82                 if let Err(err) = result {
  83                     println!("got expected error: {}", err);
  84                 } else {
  85                     let mut data = errmsg.lock().unwrap();
  86                     *data = Some(String::from("thread finished - seems abort did not work as expected"));
  87                 }
  88
  89                 Ok(())
  90             },
  91         );
  92
  93         match res {
  94             Err(err) => {
  95                 println!("unable to start worker - {}", err);
  96             }
  97             Ok(wid) => {
  98                 println!("WORKER: {}", wid);
  99                 proxmox_rest_server::abort_worker_async(wid.parse::<UPID>().unwrap());
 100                 proxmox_rest_server::wait_for_local_worker(&wid).await.unwrap();
 101              }
 102         }
 103     });
 104
 105     let data = errmsg.lock().unwrap();
 106     match *data {
 107         Some(ref err) => bail!("Error: {}", err),
 108         None => {},
 109     }
 110
 111     Ok(())
 112 }