src/server/worker_task.rs

   1 use std::collections::{HashMap, VecDeque};
   2 use std::fs::File;
   3 use std::io::{Read, Write, BufRead, BufReader};
   4 use std::panic::UnwindSafe;
   5 use std::sync::atomic::{AtomicBool, Ordering};
   6 use std::sync::{Arc, Mutex};
   7
   8 use anyhow::{bail, format_err, Error};
   9 use futures::*;
  10 use lazy_static::lazy_static;
  11 use nix::unistd::Pid;
  12 use serde_json::{json, Value};
  13 use serde::{Serialize, Deserialize};
  14 use tokio::sync::oneshot;
  15
  16 use proxmox::sys::linux::procfs;
  17 use proxmox::try_block;
  18 use proxmox::tools::fs::{create_path, open_file_locked, replace_file, CreateOptions};
  19
  20 use super::UPID;
  21
  22 use crate::tools::logrotate::{LogRotate, LogRotateFiles};
  23 use crate::tools::FileLogger;
  24 use crate::api2::types::Userid;
  25
  26 macro_rules! PROXMOX_BACKUP_VAR_RUN_DIR_M { () => ("/run/proxmox-backup") }
  27 macro_rules! PROXMOX_BACKUP_LOG_DIR_M { () => ("/var/log/proxmox-backup") }
  28 macro_rules! PROXMOX_BACKUP_TASK_DIR_M { () => (concat!( PROXMOX_BACKUP_LOG_DIR_M!(), "/tasks")) }
  29
  30 pub const PROXMOX_BACKUP_VAR_RUN_DIR: &str = PROXMOX_BACKUP_VAR_RUN_DIR_M!();
  31 pub const PROXMOX_BACKUP_LOG_DIR: &str = PROXMOX_BACKUP_LOG_DIR_M!();
  32 pub const PROXMOX_BACKUP_TASK_DIR: &str = PROXMOX_BACKUP_TASK_DIR_M!();
  33 pub const PROXMOX_BACKUP_TASK_LOCK_FN: &str = concat!(PROXMOX_BACKUP_TASK_DIR_M!(), "/.active.lock");
  34 pub const PROXMOX_BACKUP_ACTIVE_TASK_FN: &str = concat!(PROXMOX_BACKUP_TASK_DIR_M!(), "/active");
  35 pub const PROXMOX_BACKUP_INDEX_TASK_FN: &str = concat!(PROXMOX_BACKUP_TASK_DIR_M!(), "/index");
  36 pub const PROXMOX_BACKUP_ARCHIVE_TASK_FN: &str = concat!(PROXMOX_BACKUP_TASK_DIR_M!(), "/archive");
  37
  38 const MAX_INDEX_TASKS: usize = 1000;
  39
  40 lazy_static! {
  41     static ref WORKER_TASK_LIST: Mutex<HashMap<usize, Arc<WorkerTask>>> = Mutex::new(HashMap::new());
  42
  43     static ref MY_PID: i32 = unsafe { libc::getpid() };
  44     static ref MY_PID_PSTART: u64 = procfs::PidStat::read_from_pid(Pid::from_raw(*MY_PID))
  45         .unwrap()
  46         .starttime;
  47 }
  48
  49 /// Test if the task is still running
  50 pub async fn worker_is_active(upid: &UPID) -> Result<bool, Error> {
  51     if (upid.pid == *MY_PID) && (upid.pstart == *MY_PID_PSTART) {
  52         return Ok(WORKER_TASK_LIST.lock().unwrap().contains_key(&upid.task_id));
  53     }
  54
  55     if !procfs::check_process_running_pstart(upid.pid, upid.pstart).is_some() {
  56         return Ok(false);
  57     }
  58
  59     let socketname = format!(
  60         "\0{}/proxmox-task-control-{}.sock", PROXMOX_BACKUP_VAR_RUN_DIR, upid.pid);
  61
  62     let cmd = json!({
  63         "command": "status",
  64         "upid": upid.to_string(),
  65     });
  66
  67     let status = super::send_command(socketname, cmd).await?;
  68
  69     if let Some(active) = status.as_bool() {
  70         Ok(active)
  71     } else {
  72         bail!("got unexpected result {:?} (expected bool)", status);
  73     }
  74 }
  75
  76 /// Test if the task is still running (fast but inaccurate implementation)
  77 ///
  78 /// If the task is spanned from a different process, we simply return if
  79 /// that process is still running. This information is good enough to detect
  80 /// stale tasks...
  81 pub fn worker_is_active_local(upid: &UPID) -> bool {
  82     if (upid.pid == *MY_PID) && (upid.pstart == *MY_PID_PSTART) {
  83         WORKER_TASK_LIST.lock().unwrap().contains_key(&upid.task_id)
  84     } else {
  85         procfs::check_process_running_pstart(upid.pid, upid.pstart).is_some()
  86     }
  87 }
  88
  89 pub fn create_task_control_socket() -> Result<(), Error> {
  90
  91     let socketname = format!(
  92         "\0{}/proxmox-task-control-{}.sock", PROXMOX_BACKUP_VAR_RUN_DIR, *MY_PID);
  93
  94     let control_future = super::create_control_socket(socketname, |param| {
  95         let param = param.as_object()
  96             .ok_or_else(|| format_err!("unable to parse parameters (expected json object)"))?;
  97         if param.keys().count() != 2 { bail!("wrong number of parameters"); }
  98
  99         let command = param["command"].as_str()
 100             .ok_or_else(|| format_err!("unable to parse parameters (missing command)"))?;
 101
 102         // we have only two commands for now
 103         if !(command == "abort-task" || command == "status") { bail!("got unknown command '{}'", command); }
 104
 105         let upid_str = param["upid"].as_str()
 106             .ok_or_else(|| format_err!("unable to parse parameters (missing upid)"))?;
 107
 108         let upid = upid_str.parse::<UPID>()?;
 109
 110         if !((upid.pid == *MY_PID) && (upid.pstart == *MY_PID_PSTART)) {
 111             bail!("upid does not belong to this process");
 112         }
 113
 114         let hash = WORKER_TASK_LIST.lock().unwrap();
 115
 116         match command {
 117             "abort-task" => {
 118                 if let Some(ref worker) = hash.get(&upid.task_id) {
 119                     worker.request_abort();
 120                 } else {
 121                     // assume task is already stopped
 122                 }
 123                 Ok(Value::Null)
 124             }
 125             "status" => {
 126                 let active = hash.contains_key(&upid.task_id);
 127                 Ok(active.into())
 128             }
 129             _ => {
 130                 bail!("got unknown command '{}'", command);
 131             }
 132         }
 133     })?;
 134
 135     tokio::spawn(control_future);
 136
 137     Ok(())
 138 }
 139
 140 pub fn abort_worker_async(upid: UPID) {
 141     tokio::spawn(async move {
 142         if let Err(err) = abort_worker(upid).await {
 143             eprintln!("abort worker failed - {}", err);
 144         }
 145     });
 146 }
 147
 148 pub async fn abort_worker(upid: UPID) -> Result<(), Error> {
 149
 150     let target_pid = upid.pid;
 151
 152     let socketname = format!(
 153         "\0{}/proxmox-task-control-{}.sock", PROXMOX_BACKUP_VAR_RUN_DIR, target_pid);
 154
 155     let cmd = json!({
 156         "command": "abort-task",
 157         "upid": upid.to_string(),
 158     });
 159
 160     super::send_command(socketname, cmd).map_ok(|_| ()).await
 161 }
 162
 163 fn parse_worker_status_line(line: &str) -> Result<(String, UPID, Option<TaskState>), Error> {
 164
 165     let data = line.splitn(3, ' ').collect::<Vec<&str>>();
 166
 167     let len = data.len();
 168
 169     match len {
 170         1 => Ok((data[0].to_owned(), data[0].parse::<UPID>()?, None)),
 171         3 => {
 172             let endtime = i64::from_str_radix(data[1], 16)?;
 173             let state = TaskState::from_endtime_and_message(endtime, data[2])?;
 174             Ok((data[0].to_owned(), data[0].parse::<UPID>()?, Some(state)))
 175         }
 176         _ => bail!("wrong number of components"),
 177     }
 178 }
 179
 180 /// Create task log directory with correct permissions
 181 pub fn create_task_log_dirs() -> Result<(), Error> {
 182
 183     try_block!({
 184         let backup_user = crate::backup::backup_user()?;
 185         let opts = CreateOptions::new()
 186             .owner(backup_user.uid)
 187             .group(backup_user.gid);
 188
 189         create_path(PROXMOX_BACKUP_LOG_DIR, None, Some(opts.clone()))?;
 190         create_path(PROXMOX_BACKUP_TASK_DIR, None, Some(opts.clone()))?;
 191         create_path(PROXMOX_BACKUP_VAR_RUN_DIR, None, Some(opts))?;
 192         Ok(())
 193     }).map_err(|err: Error| format_err!("unable to create task log dir - {}", err))?;
 194
 195     Ok(())
 196 }
 197
 198 /// Read endtime (time of last log line) and exitstatus from task log file
 199 /// If there is not a single line with at valid datetime, we assume the
 200 /// starttime to be the endtime
 201 pub fn upid_read_status(upid: &UPID) -> Result<TaskState, Error> {
 202
 203     let mut status = TaskState::Unknown { endtime: upid.starttime };
 204
 205     let path = upid.log_path();
 206
 207     let mut file = File::open(path)?;
 208
 209     /// speedup - only read tail
 210     use std::io::Seek;
 211     use std::io::SeekFrom;
 212     let _ = file.seek(SeekFrom::End(-8192)); // ignore errors
 213
 214     let mut data = Vec::with_capacity(8192);
 215     file.read_to_end(&mut data)?;
 216
 217     // task logs should end with newline, we do not want it here
 218     if data.len() > 0 && data[data.len()-1] == b'\n' {
 219         data.pop();
 220     }
 221
 222     let last_line = {
 223         let mut start = 0;
 224         for pos in (0..data.len()).rev() {
 225             if data[pos] == b'\n' {
 226                 start = data.len().min(pos + 1);
 227                 break;
 228             }
 229         }
 230         &data[start..]
 231     };
 232
 233     let last_line = std::str::from_utf8(last_line)
 234         .map_err(|err| format_err!("upid_read_status: utf8 parse failed: {}", err))?;
 235
 236     let mut iter = last_line.splitn(2, ": ");
 237     if let Some(time_str) = iter.next() {
 238         if let Ok(endtime) = proxmox::tools::time::parse_rfc3339(time_str) {
 239             if let Some(rest) = iter.next().and_then(|rest| rest.strip_prefix("TASK ")) {
 240                 if let Ok(state) = TaskState::from_endtime_and_message(endtime, rest) {
 241                     status = state;
 242                 }
 243             }
 244         }
 245     }
 246
 247     Ok(status)
 248 }
 249
 250 /// Task State
 251 #[derive(Debug, PartialEq, Eq, Serialize, Deserialize)]
 252 pub enum TaskState {
 253     /// The Task ended with an undefined state
 254     Unknown { endtime: i64 },
 255     /// The Task ended and there were no errors or warnings
 256     OK { endtime: i64 },
 257     /// The Task had 'count' amount of warnings and no errors
 258     Warning { count: u64, endtime: i64 },
 259     /// The Task ended with the error described in 'message'
 260     Error { message: String, endtime: i64 },
 261 }
 262
 263 impl TaskState {
 264     pub fn endtime(&self) -> i64 {
 265         match *self {
 266             TaskState::Unknown { endtime } => endtime,
 267             TaskState::OK { endtime } => endtime,
 268             TaskState::Warning { endtime, .. } => endtime,
 269             TaskState::Error { endtime, .. } => endtime,
 270         }
 271     }
 272
 273     fn result_text(&self) -> String {
 274         match self {
 275             TaskState::Error { message, .. } => format!("TASK ERROR: {}", message),
 276             other => format!("TASK {}", other),
 277         }
 278     }
 279
 280     fn from_endtime_and_message(endtime: i64, s: &str) -> Result<Self, Error> {
 281         if s == "unknown" {
 282             Ok(TaskState::Unknown { endtime })
 283         } else if s == "OK" {
 284             Ok(TaskState::OK { endtime })
 285         } else if s.starts_with("WARNINGS: ") {
 286             let count: u64 = s[10..].parse()?;
 287             Ok(TaskState::Warning{ count, endtime })
 288         } else if s.len() > 0 {
 289             let message = if s.starts_with("ERROR: ") { &s[7..] } else { s }.to_string();
 290             Ok(TaskState::Error{ message, endtime })
 291         } else {
 292             bail!("unable to parse Task Status '{}'", s);
 293         }
 294     }
 295 }
 296
 297 impl std::cmp::PartialOrd for TaskState {
 298     fn partial_cmp(&self, other: &Self) -> Option<std::cmp::Ordering> {
 299         Some(self.endtime().cmp(&other.endtime()))
 300     }
 301 }
 302
 303 impl std::cmp::Ord for TaskState {
 304     fn cmp(&self, other: &Self) -> std::cmp::Ordering {
 305         self.endtime().cmp(&other.endtime())
 306     }
 307 }
 308
 309 impl std::fmt::Display for TaskState {
 310     fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
 311         match self {
 312             TaskState::Unknown { .. } => write!(f, "unknown"),
 313             TaskState::OK { .. }=> write!(f, "OK"),
 314             TaskState::Warning { count, .. } => write!(f, "WARNINGS: {}", count),
 315             TaskState::Error { message, .. } => write!(f, "{}", message),
 316         }
 317     }
 318 }
 319
 320 /// Task details including parsed UPID
 321 ///
 322 /// If there is no `state`, the task is still running.
 323 #[derive(Debug)]
 324 pub struct TaskListInfo {
 325     /// The parsed UPID
 326     pub upid: UPID,
 327     /// UPID string representation
 328     pub upid_str: String,
 329     /// Task `(endtime, status)` if already finished
 330     pub state: Option<TaskState>, // endtime, status
 331 }
 332
 333 fn lock_task_list_files(exclusive: bool) -> Result<std::fs::File, Error> {
 334     let backup_user = crate::backup::backup_user()?;
 335
 336     let lock = open_file_locked(PROXMOX_BACKUP_TASK_LOCK_FN, std::time::Duration::new(10, 0), exclusive)?;
 337     nix::unistd::chown(PROXMOX_BACKUP_TASK_LOCK_FN, Some(backup_user.uid), Some(backup_user.gid))?;
 338
 339     Ok(lock)
 340 }
 341
 342 // atomically read/update the task list, update status of finished tasks
 343 // new_upid is added to the list when specified.
 344 fn update_active_workers(new_upid: Option<&UPID>) -> Result<(), Error> {
 345
 346     let backup_user = crate::backup::backup_user()?;
 347
 348     let lock = lock_task_list_files(true)?;
 349
 350     let mut finish_list: Vec<TaskListInfo> = read_task_file_from_path(PROXMOX_BACKUP_INDEX_TASK_FN)?;
 351     let mut active_list: Vec<TaskListInfo> = read_task_file_from_path(PROXMOX_BACKUP_ACTIVE_TASK_FN)?
 352         .into_iter()
 353         .filter_map(|info| {
 354             if info.state.is_some() {
 355                 // this can happen when the active file still includes finished tasks
 356                 finish_list.push(info);
 357                 return None;
 358             }
 359
 360             if !worker_is_active_local(&info.upid) {
 361                 println!("Detected stopped UPID {}", &info.upid_str);
 362                 let now = proxmox::tools::time::epoch_i64();
 363                 let status = upid_read_status(&info.upid)
 364                     .unwrap_or_else(|_| TaskState::Unknown { endtime: now });
 365                 finish_list.push(TaskListInfo {
 366                     upid: info.upid,
 367                     upid_str: info.upid_str,
 368                     state: Some(status)
 369                 });
 370                 return None;
 371             }
 372
 373             Some(info)
 374         }).collect();
 375
 376     if let Some(upid) = new_upid {
 377         active_list.push(TaskListInfo { upid: upid.clone(), upid_str: upid.to_string(), state: None });
 378     }
 379
 380     let active_raw = render_task_list(&active_list);
 381
 382     replace_file(
 383         PROXMOX_BACKUP_ACTIVE_TASK_FN,
 384         active_raw.as_bytes(),
 385         CreateOptions::new()
 386             .owner(backup_user.uid)
 387             .group(backup_user.gid),
 388     )?;
 389
 390     finish_list.sort_unstable_by(|a, b| {
 391         match (&a.state, &b.state) {
 392             (Some(s1), Some(s2)) => s1.cmp(&s2),
 393             (Some(_), None) => std::cmp::Ordering::Less,
 394             (None, Some(_)) => std::cmp::Ordering::Greater,
 395             _ => a.upid.starttime.cmp(&b.upid.starttime),
 396         }
 397     });
 398
 399     let start = (finish_list.len()-MAX_INDEX_TASKS).max(0);
 400     let end = (start+MAX_INDEX_TASKS).min(finish_list.len());
 401     let index_raw = render_task_list(&finish_list[start..end]);
 402
 403     replace_file(
 404         PROXMOX_BACKUP_INDEX_TASK_FN,
 405         index_raw.as_bytes(),
 406         CreateOptions::new()
 407             .owner(backup_user.uid)
 408             .group(backup_user.gid),
 409     )?;
 410
 411     if !finish_list.is_empty() && start > 0 {
 412         match std::fs::OpenOptions::new().append(true).create(true).open(PROXMOX_BACKUP_ARCHIVE_TASK_FN) {
 413             Ok(mut writer) => {
 414                 for info in &finish_list[0..start] {
 415                     writer.write_all(render_task_line(&info).as_bytes())?;
 416                 }
 417             },
 418             Err(err) => bail!("could not write task archive - {}", err),
 419         }
 420
 421         nix::unistd::chown(PROXMOX_BACKUP_ARCHIVE_TASK_FN, Some(backup_user.uid), Some(backup_user.gid))?;
 422     }
 423
 424     drop(lock);
 425
 426     Ok(())
 427 }
 428
 429 fn render_task_line(info: &TaskListInfo) -> String {
 430     let mut raw = String::new();
 431     if let Some(status) = &info.state {
 432         raw.push_str(&format!("{} {:08X} {}\n", info.upid_str, status.endtime(), status));
 433     } else {
 434         raw.push_str(&info.upid_str);
 435         raw.push('\n');
 436     }
 437
 438     raw
 439 }
 440
 441 fn render_task_list(list: &[TaskListInfo]) -> String {
 442     let mut raw = String::new();
 443     for info in list {
 444         raw.push_str(&render_task_line(&info));
 445     }
 446     raw
 447 }
 448
 449 // note this is not locked, caller has to make sure it is
 450 // this will skip (and log) lines that are not valid status lines
 451 fn read_task_file<R: Read>(reader: R) -> Result<Vec<TaskListInfo>, Error>
 452 {
 453     let reader = BufReader::new(reader);
 454     let mut list = Vec::new();
 455     for line in reader.lines() {
 456         let line = line?;
 457         match parse_worker_status_line(&line) {
 458             Ok((upid_str, upid, state)) => list.push(TaskListInfo {
 459                 upid_str,
 460                 upid,
 461                 state
 462             }),
 463             Err(err) => {
 464                 eprintln!("unable to parse worker status '{}' - {}", line, err);
 465                 continue;
 466             }
 467         };
 468     }
 469
 470     Ok(list)
 471 }
 472
 473 // note this is not locked, caller has to make sure it is
 474 fn read_task_file_from_path<P>(path: P) -> Result<Vec<TaskListInfo>, Error>
 475 where
 476     P: AsRef<std::path::Path> + std::fmt::Debug,
 477 {
 478     let file = match File::open(&path) {
 479         Ok(f) => f,
 480         Err(err) if err.kind() == std::io::ErrorKind::NotFound => return Ok(Vec::new()),
 481         Err(err) => bail!("unable to open task list {:?} - {}", path, err),
 482     };
 483
 484     read_task_file(file)
 485 }
 486
 487 enum TaskFile {
 488     Active,
 489     Index,
 490     Archive,
 491     End,
 492 }
 493
 494 pub struct TaskListInfoIterator {
 495     list: VecDeque<TaskListInfo>,
 496     file: TaskFile,
 497     archive: Option<LogRotateFiles>,
 498     lock: Option<File>,
 499 }
 500
 501 impl TaskListInfoIterator {
 502     pub fn new(active_only: bool) -> Result<Self, Error> {
 503         let (read_lock, active_list) = {
 504             let lock = lock_task_list_files(false)?;
 505             let active_list = read_task_file_from_path(PROXMOX_BACKUP_ACTIVE_TASK_FN)?;
 506
 507             let needs_update = active_list
 508                 .iter()
 509                 .any(|info| info.state.is_none() && !worker_is_active_local(&info.upid));
 510
 511             if needs_update {
 512                 drop(lock);
 513                 update_active_workers(None)?;
 514                 let lock = lock_task_list_files(false)?;
 515                 let active_list = read_task_file_from_path(PROXMOX_BACKUP_ACTIVE_TASK_FN)?;
 516                 (lock, active_list)
 517             } else {
 518                 (lock, active_list)
 519             }
 520         };
 521
 522         let archive = if active_only {
 523             None
 524         } else {
 525             let logrotate = LogRotate::new(PROXMOX_BACKUP_ARCHIVE_TASK_FN, true).ok_or_else(|| format_err!("could not get archive file names"))?;
 526             Some(logrotate.files())
 527         };
 528
 529         let file = if active_only { TaskFile::End } else { TaskFile::Active };
 530         let lock = if active_only { None } else { Some(read_lock) };
 531
 532         Ok(Self {
 533             list: active_list.into(),
 534             file,
 535             archive,
 536             lock,
 537         })
 538     }
 539 }
 540
 541 impl Iterator for TaskListInfoIterator {
 542     type Item = Result<TaskListInfo, Error>;
 543
 544     fn next(&mut self) -> Option<Self::Item> {
 545         loop {
 546             if let Some(element) = self.list.pop_back() {
 547                 return Some(Ok(element));
 548             } else {
 549                 match self.file {
 550                     TaskFile::Active => {
 551                         let index = match read_task_file_from_path(PROXMOX_BACKUP_INDEX_TASK_FN) {
 552                             Ok(index) => index,
 553                             Err(err) => return Some(Err(err)),
 554                         };
 555                         self.list.append(&mut index.into());
 556                         self.file = TaskFile::Index;
 557                     },
 558                     TaskFile::Index | TaskFile::Archive => {
 559                         if let Some(mut archive) = self.archive.take() {
 560                             if let Some(file) = archive.next() {
 561                                 let list = match read_task_file(file) {
 562                                     Ok(list) => list,
 563                                     Err(err) => return Some(Err(err)),
 564                                 };
 565                                 self.list.append(&mut list.into());
 566                                 self.archive = Some(archive);
 567                                 self.file = TaskFile::Archive;
 568                                 continue;
 569                             }
 570                         }
 571                         self.file = TaskFile::End;
 572                         self.lock.take();
 573                         return None;
 574                     }
 575                     TaskFile::End => return None,
 576                 }
 577             }
 578         }
 579     }
 580 }
 581
 582 /// Launch long running worker tasks.
 583 ///
 584 /// A worker task can either be a whole thread, or a simply tokio
 585 /// task/future. Each task can `log()` messages, which are stored
 586 /// persistently to files. Task should poll the `abort_requested`
 587 /// flag, and stop execution when requested.
 588 #[derive(Debug)]
 589 pub struct WorkerTask {
 590     upid: UPID,
 591     data: Mutex<WorkerTaskData>,
 592     abort_requested: AtomicBool,
 593 }
 594
 595 impl std::fmt::Display for WorkerTask {
 596
 597     fn fmt(&self, f: &mut std::fmt::Formatter) -> std::fmt::Result {
 598         self.upid.fmt(f)
 599     }
 600 }
 601
 602 #[derive(Debug)]
 603 struct WorkerTaskData {
 604     logger: FileLogger,
 605     progress: f64, // 0..1
 606     warn_count: u64,
 607     pub abort_listeners: Vec<oneshot::Sender<()>>,
 608 }
 609
 610 impl Drop for WorkerTask {
 611
 612     fn drop(&mut self) {
 613         println!("unregister worker");
 614     }
 615 }
 616
 617 impl WorkerTask {
 618
 619     pub fn new(worker_type: &str, worker_id: Option<String>, userid: Userid, to_stdout: bool) -> Result<Arc<Self>, Error> {
 620         println!("register worker");
 621
 622         let upid = UPID::new(worker_type, worker_id, userid)?;
 623         let task_id = upid.task_id;
 624
 625         let mut path = std::path::PathBuf::from(PROXMOX_BACKUP_TASK_DIR);
 626
 627         path.push(format!("{:02X}", upid.pstart % 256));
 628
 629         let backup_user = crate::backup::backup_user()?;
 630
 631         create_path(&path, None, Some(CreateOptions::new().owner(backup_user.uid).group(backup_user.gid)))?;
 632
 633         path.push(upid.to_string());
 634
 635         println!("FILE: {:?}", path);
 636
 637         let logger = FileLogger::new(&path, to_stdout)?;
 638         nix::unistd::chown(&path, Some(backup_user.uid), Some(backup_user.gid))?;
 639
 640         let worker = Arc::new(Self {
 641             upid: upid.clone(),
 642             abort_requested: AtomicBool::new(false),
 643             data: Mutex::new(WorkerTaskData {
 644                 logger,
 645                 progress: 0.0,
 646                 warn_count: 0,
 647                 abort_listeners: vec![],
 648             }),
 649         });
 650
 651         // scope to drop the lock again after inserting
 652         {
 653             let mut hash = WORKER_TASK_LIST.lock().unwrap();
 654             hash.insert(task_id, worker.clone());
 655             super::set_worker_count(hash.len());
 656         }
 657
 658         update_active_workers(Some(&upid))?;
 659
 660         Ok(worker)
 661     }
 662
 663     /// Spawn a new tokio task/future.
 664     pub fn spawn<F, T>(
 665         worker_type: &str,
 666         worker_id: Option<String>,
 667         userid: Userid,
 668         to_stdout: bool,
 669         f: F,
 670     ) -> Result<String, Error>
 671         where F: Send + 'static + FnOnce(Arc<WorkerTask>) -> T,
 672               T: Send + 'static + Future<Output = Result<(), Error>>,
 673     {
 674         let worker = WorkerTask::new(worker_type, worker_id, userid, to_stdout)?;
 675         let upid_str = worker.upid.to_string();
 676         let f = f(worker.clone());
 677         tokio::spawn(async move {
 678             let result = f.await;
 679             worker.log_result(&result);
 680         });
 681
 682         Ok(upid_str)
 683     }
 684
 685     /// Create a new worker thread.
 686     pub fn new_thread<F>(
 687         worker_type: &str,
 688         worker_id: Option<String>,
 689         userid: Userid,
 690         to_stdout: bool,
 691         f: F,
 692     ) -> Result<String, Error>
 693         where F: Send + UnwindSafe + 'static + FnOnce(Arc<WorkerTask>) -> Result<(), Error>
 694     {
 695         println!("register worker thread");
 696
 697         let worker = WorkerTask::new(worker_type, worker_id, userid, to_stdout)?;
 698         let upid_str = worker.upid.to_string();
 699
 700         let _child = std::thread::Builder::new().name(upid_str.clone()).spawn(move || {
 701             let worker1 = worker.clone();
 702             let result = match std::panic::catch_unwind(move || f(worker1)) {
 703                 Ok(r) => r,
 704                 Err(panic) => {
 705                     match panic.downcast::<&str>() {
 706                         Ok(panic_msg) => {
 707                             Err(format_err!("worker panicked: {}", panic_msg))
 708                         }
 709                         Err(_) => {
 710                             Err(format_err!("worker panicked: unknown type."))
 711                         }
 712                     }
 713                 }
 714             };
 715
 716             worker.log_result(&result);
 717         });
 718
 719         Ok(upid_str)
 720     }
 721
 722     /// create state from self and a result
 723     pub fn create_state(&self, result: &Result<(), Error>) -> TaskState {
 724         let warn_count = self.data.lock().unwrap().warn_count;
 725
 726         let endtime = proxmox::tools::time::epoch_i64();
 727
 728         if let Err(err) = result {
 729             TaskState::Error { message: err.to_string(), endtime }
 730         } else if warn_count > 0 {
 731             TaskState::Warning { count: warn_count, endtime }
 732         } else {
 733             TaskState::OK { endtime }
 734         }
 735     }
 736
 737     /// Log task result, remove task from running list
 738     pub fn log_result(&self, result: &Result<(), Error>) {
 739         let state = self.create_state(result);
 740         self.log(state.result_text());
 741
 742         WORKER_TASK_LIST.lock().unwrap().remove(&self.upid.task_id);
 743         let _ = update_active_workers(None);
 744         super::set_worker_count(WORKER_TASK_LIST.lock().unwrap().len());
 745     }
 746
 747     /// Log a message.
 748     pub fn log<S: AsRef<str>>(&self, msg: S) {
 749         let mut data = self.data.lock().unwrap();
 750         data.logger.log(msg);
 751     }
 752
 753     /// Log a message as warning.
 754     pub fn warn<S: AsRef<str>>(&self, msg: S) {
 755         let mut data = self.data.lock().unwrap();
 756         data.logger.log(format!("WARN: {}", msg.as_ref()));
 757         data.warn_count += 1;
 758     }
 759
 760     /// Set progress indicator
 761     pub fn progress(&self, progress: f64) {
 762         if progress >= 0.0 && progress <= 1.0 {
 763             let mut data = self.data.lock().unwrap();
 764             data.progress = progress;
 765         } else {
 766            // fixme:  log!("task '{}': ignoring strange value for progress '{}'", self.upid, progress);
 767         }
 768     }
 769
 770     /// Request abort
 771     pub fn request_abort(&self) {
 772         eprintln!("set abort flag for worker {}", self.upid);
 773         self.abort_requested.store(true, Ordering::SeqCst);
 774         // noitify listeners
 775         let mut data = self.data.lock().unwrap();
 776         loop {
 777             match data.abort_listeners.pop() {
 778                 None => { break; },
 779                 Some(ch) => {
 780                     let _ = ch.send(()); // ignore erros here
 781                 },
 782             }
 783         }
 784     }
 785
 786     /// Test if abort was requested.
 787     pub fn abort_requested(&self) -> bool {
 788         self.abort_requested.load(Ordering::SeqCst)
 789     }
 790
 791     /// Fail if abort was requested.
 792     pub fn fail_on_abort(&self) -> Result<(), Error> {
 793         if self.abort_requested() {
 794             bail!("abort requested - aborting task");
 795         }
 796         Ok(())
 797     }
 798
 799     /// Get a future which resolves on task abort
 800     pub fn abort_future(&self) ->  oneshot::Receiver<()> {
 801         let (tx, rx) = oneshot::channel::<()>();
 802
 803         let mut data = self.data.lock().unwrap();
 804         if self.abort_requested() {
 805             let _ = tx.send(());
 806         } else {
 807             data.abort_listeners.push(tx);
 808         }
 809         rx
 810     }
 811
 812     pub fn upid(&self) -> &UPID {
 813         &self.upid
 814     }
 815 }