src/server/worker_task.rs

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