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