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
  13 use proxmox_rest_server::{flog, CommandoSocket, WorkerTask};
  14
  15 fn garbage_collection(worker: &WorkerTask) -> Result<(), Error> {
  16
  17     worker.log("start garbage collection");
  18
  19     for i in 0..50 {
  20         worker.fail_on_abort()?;
  21
  22         flog!(worker, "progress {}", i);
  23
  24         std::thread::sleep(std::time::Duration::from_millis(10));
  25     }
  26
  27     worker.log("end garbage collection");
  28
  29     Ok(())
  30 }
  31
  32
  33 #[test]
  34 #[ignore]
  35 fn worker_task_abort() -> Result<(), Error> {
  36     let uid = nix::unistd::Uid::current();
  37     let gid = nix::unistd::Gid::current();
  38
  39     let file_opts = CreateOptions::new().owner(uid).group(gid);
  40     proxmox_rest_server::init_worker_tasks("./target/tasklogtestdir".into(), file_opts.clone())?;
  41
  42     use std::sync::{Arc, Mutex};
  43
  44     let errmsg: Arc<Mutex<Option<String>>> = Arc::new(Mutex::new(None));
  45     let errmsg1 = errmsg.clone();
  46
  47     let rt = tokio::runtime::Runtime::new().unwrap();
  48     rt.block_on(async move {
  49
  50         let mut commando_sock = CommandoSocket::new(
  51             proxmox_rest_server::our_ctrl_sock(), nix::unistd::Gid::current());
  52
  53         let init_result: Result<(), Error> = try_block!({
  54             proxmox_rest_server::register_task_control_commands(&mut commando_sock)?;
  55             proxmox_rest_server::server_state_init()?;
  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_async(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 }