src/server/worker_task.rs

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