rename TaskState to WorkerTaskContext

This commit is contained in:
Dietmar Maurer 2021-09-24 07:40:49 +02:00
parent f7348a23cd
commit c8449217dc
10 changed files with 31 additions and 29 deletions

View File

@ -9,7 +9,7 @@ use proxmox::tools::fs::{CreateOptions, create_path, create_dir};
use pbs_api_types::GarbageCollectionStatus; use pbs_api_types::GarbageCollectionStatus;
use pbs_tools::process_locker::{self, ProcessLocker}; use pbs_tools::process_locker::{self, ProcessLocker};
use pbs_tools::{task_log, task::TaskState}; use pbs_tools::{task_log, task::WorkerTaskContext};
use crate::DataBlob; use crate::DataBlob;
@ -65,7 +65,7 @@ impl ChunkStore {
&self.base &self.base
} }
pub fn create<P>(name: &str, path: P, uid: nix::unistd::Uid, gid: nix::unistd::Gid, worker: Option<&dyn TaskState>) -> Result<Self, Error> pub fn create<P>(name: &str, path: P, uid: nix::unistd::Uid, gid: nix::unistd::Gid, worker: Option<&dyn WorkerTaskContext>) -> Result<Self, Error>
where where
P: Into<PathBuf>, P: Into<PathBuf>,
{ {
@ -285,7 +285,7 @@ impl ChunkStore {
oldest_writer: i64, oldest_writer: i64,
phase1_start_time: i64, phase1_start_time: i64,
status: &mut GarbageCollectionStatus, status: &mut GarbageCollectionStatus,
worker: &dyn TaskState, worker: &dyn WorkerTaskContext,
fail_on_shutdown: F, fail_on_shutdown: F,
) -> Result<(), Error> { ) -> Result<(), Error> {
use nix::sys::stat::fstatat; use nix::sys::stat::fstatat;

View File

@ -1,7 +1,7 @@
use anyhow::Error; use anyhow::Error;
/// `WorkerTask` methods commonly used from contexts otherwise not related to the API server. /// `WorkerTask` methods commonly used from contexts otherwise not related to the API server.
pub trait TaskState { pub trait WorkerTaskContext {
/// If the task should be aborted, this should fail with a reasonable error message. /// If the task should be aborted, this should fail with a reasonable error message.
fn check_abort(&self) -> Result<(), Error>; fn check_abort(&self) -> Result<(), Error>;
@ -10,47 +10,47 @@ pub trait TaskState {
} }
/// Convenience implementation: /// Convenience implementation:
impl<T: TaskState + ?Sized> TaskState for std::sync::Arc<T> { impl<T: WorkerTaskContext + ?Sized> WorkerTaskContext for std::sync::Arc<T> {
fn check_abort(&self) -> Result<(), Error> { fn check_abort(&self) -> Result<(), Error> {
<T as TaskState>::check_abort(&*self) <T as WorkerTaskContext>::check_abort(&*self)
} }
fn log(&self, level: log::Level, message: &std::fmt::Arguments) { fn log(&self, level: log::Level, message: &std::fmt::Arguments) {
<T as TaskState>::log(&*self, level, message) <T as WorkerTaskContext>::log(&*self, level, message)
} }
} }
#[macro_export] #[macro_export]
macro_rules! task_error { macro_rules! task_error {
($task:expr, $($fmt:tt)+) => {{ ($task:expr, $($fmt:tt)+) => {{
$crate::task::TaskState::log(&*$task, log::Level::Error, &format_args!($($fmt)+)) $crate::task::WorkerTaskContext::log(&*$task, log::Level::Error, &format_args!($($fmt)+))
}}; }};
} }
#[macro_export] #[macro_export]
macro_rules! task_warn { macro_rules! task_warn {
($task:expr, $($fmt:tt)+) => {{ ($task:expr, $($fmt:tt)+) => {{
$crate::task::TaskState::log(&*$task, log::Level::Warn, &format_args!($($fmt)+)) $crate::task::WorkerTaskContext::log(&*$task, log::Level::Warn, &format_args!($($fmt)+))
}}; }};
} }
#[macro_export] #[macro_export]
macro_rules! task_log { macro_rules! task_log {
($task:expr, $($fmt:tt)+) => {{ ($task:expr, $($fmt:tt)+) => {{
$crate::task::TaskState::log(&*$task, log::Level::Info, &format_args!($($fmt)+)) $crate::task::WorkerTaskContext::log(&*$task, log::Level::Info, &format_args!($($fmt)+))
}}; }};
} }
#[macro_export] #[macro_export]
macro_rules! task_debug { macro_rules! task_debug {
($task:expr, $($fmt:tt)+) => {{ ($task:expr, $($fmt:tt)+) => {{
$crate::task::TaskState::log(&*$task, log::Level::Debug, &format_args!($($fmt)+)) $crate::task::WorkerTaskContext::log(&*$task, log::Level::Debug, &format_args!($($fmt)+))
}}; }};
} }
#[macro_export] #[macro_export]
macro_rules! task_trace { macro_rules! task_trace {
($task:expr, $($fmt:tt)+) => {{ ($task:expr, $($fmt:tt)+) => {{
$crate::task::TaskState::log(&*$task, log::Level::Trace, &format_args!($($fmt)+)) $crate::task::WorkerTaskContext::log(&*$task, log::Level::Trace, &format_args!($($fmt)+))
}}; }};
} }

View File

@ -20,6 +20,7 @@ use proxmox::try_block;
use proxmox::tools::fs::{create_path, replace_file, atomic_open_or_create_file, CreateOptions}; use proxmox::tools::fs::{create_path, replace_file, atomic_open_or_create_file, CreateOptions};
use proxmox::api::upid::UPID; use proxmox::api::upid::UPID;
use pbs_tools::task::WorkerTaskContext;
use pbs_tools::logrotate::{LogRotate, LogRotateFiles}; use pbs_tools::logrotate::{LogRotate, LogRotateFiles};
use crate::{CommandoSocket, FileLogger, FileLogOptions}; use crate::{CommandoSocket, FileLogger, FileLogOptions};
@ -834,7 +835,7 @@ impl WorkerTask {
} }
/// Fail if abort was requested. /// Fail if abort was requested.
pub fn fail_on_abort(&self) -> Result<(), Error> { pub fn check_abort(&self) -> Result<(), Error> {
if self.abort_requested() { if self.abort_requested() {
bail!("abort requested - aborting task"); bail!("abort requested - aborting task");
} }
@ -859,9 +860,9 @@ impl WorkerTask {
} }
} }
impl pbs_tools::task::TaskState for WorkerTask { impl WorkerTaskContext for WorkerTask {
fn check_abort(&self) -> Result<(), Error> { fn check_abort(&self) -> Result<(), Error> {
self.fail_on_abort() self.check_abort()
} }
fn log(&self, level: log::Level, message: &std::fmt::Arguments) { fn log(&self, level: log::Level, message: &std::fmt::Arguments) {

View File

@ -16,7 +16,6 @@ use pbs_api_types::{
PRIV_DATASTORE_ALLOCATE, PRIV_DATASTORE_AUDIT, PRIV_DATASTORE_MODIFY, PRIV_DATASTORE_ALLOCATE, PRIV_DATASTORE_AUDIT, PRIV_DATASTORE_MODIFY,
DataStoreConfig, DataStoreConfigUpdater, DataStoreConfig, DataStoreConfigUpdater,
}; };
use pbs_tools::task::TaskState;
use crate::api2::config::sync::delete_sync_job; use crate::api2::config::sync::delete_sync_job;
use crate::api2::config::verify::delete_verification_job; use crate::api2::config::verify::delete_verification_job;
@ -26,6 +25,8 @@ use crate::api2::admin::{
verify::list_verification_jobs, verify::list_verification_jobs,
}; };
use pbs_config::CachedUserInfo; use pbs_config::CachedUserInfo;
use pbs_tools::task::WorkerTaskContext;
use proxmox_rest_server::WorkerTask; use proxmox_rest_server::WorkerTask;
use crate::server::jobstate; use crate::server::jobstate;
@ -69,7 +70,7 @@ pub(crate) fn do_create_datastore(
_lock: BackupLockGuard, _lock: BackupLockGuard,
mut config: SectionConfigData, mut config: SectionConfigData,
datastore: DataStoreConfig, datastore: DataStoreConfig,
worker: Option<&dyn TaskState>, worker: Option<&dyn WorkerTaskContext>,
) -> Result<(), Error> { ) -> Result<(), Error> {
let path: PathBuf = datastore.path.clone().into(); let path: PathBuf = datastore.path.clone().into();

View File

@ -22,7 +22,7 @@ use pbs_api_types::{
use pbs_datastore::StoreProgress; use pbs_datastore::StoreProgress;
use pbs_datastore::backup_info::{BackupDir, BackupInfo}; use pbs_datastore::backup_info::{BackupDir, BackupInfo};
use pbs_tools::{task_log, task_warn, task::TaskState}; use pbs_tools::{task_log, task_warn};
use pbs_config::CachedUserInfo; use pbs_config::CachedUserInfo;
use proxmox_rest_server::WorkerTask; use proxmox_rest_server::WorkerTask;

View File

@ -45,7 +45,7 @@ use pbs_tape::{
TapeRead, BlockReadError, MediaContentHeader, TapeRead, BlockReadError, MediaContentHeader,
PROXMOX_BACKUP_CONTENT_HEADER_MAGIC_1_0, PROXMOX_BACKUP_CONTENT_HEADER_MAGIC_1_0,
}; };
use pbs_tools::{task_log, task_warn, task::TaskState}; use pbs_tools::{task_log, task_warn};
use proxmox_rest_server::WorkerTask; use proxmox_rest_server::WorkerTask;
use crate::{ use crate::{

View File

@ -26,7 +26,7 @@ use pbs_datastore::manifest::{
use pbs_tools::format::HumanByte; use pbs_tools::format::HumanByte;
use pbs_tools::fs::{lock_dir_noblock, DirLockGuard}; use pbs_tools::fs::{lock_dir_noblock, DirLockGuard};
use pbs_tools::process_locker::ProcessLockSharedGuard; use pbs_tools::process_locker::ProcessLockSharedGuard;
use pbs_tools::{task_log, task_warn, task::TaskState}; use pbs_tools::{task_log, task_warn, task::WorkerTaskContext};
use pbs_config::{open_backup_lockfile, BackupLockGuard}; use pbs_config::{open_backup_lockfile, BackupLockGuard};
use proxmox_rest_server::fail_on_shutdown; use proxmox_rest_server::fail_on_shutdown;
@ -498,7 +498,7 @@ impl DataStore {
index: I, index: I,
file_name: &Path, // only used for error reporting file_name: &Path, // only used for error reporting
status: &mut GarbageCollectionStatus, status: &mut GarbageCollectionStatus,
worker: &dyn TaskState, worker: &dyn WorkerTaskContext,
) -> Result<(), Error> { ) -> Result<(), Error> {
status.index_file_count += 1; status.index_file_count += 1;
@ -534,7 +534,7 @@ impl DataStore {
fn mark_used_chunks( fn mark_used_chunks(
&self, &self,
status: &mut GarbageCollectionStatus, status: &mut GarbageCollectionStatus,
worker: &dyn TaskState, worker: &dyn WorkerTaskContext,
) -> Result<(), Error> { ) -> Result<(), Error> {
let image_list = self.list_images()?; let image_list = self.list_images()?;
@ -611,7 +611,7 @@ impl DataStore {
!matches!(self.gc_mutex.try_lock(), Ok(_)) !matches!(self.gc_mutex.try_lock(), Ok(_))
} }
pub fn garbage_collection(&self, worker: &dyn TaskState, upid: &UPID) -> Result<(), Error> { pub fn garbage_collection(&self, worker: &dyn WorkerTaskContext, upid: &UPID) -> Result<(), Error> {
if let Ok(ref mut _mutex) = self.gc_mutex.try_lock() { if let Ok(ref mut _mutex) = self.gc_mutex.try_lock() {

View File

@ -12,7 +12,7 @@ use pbs_datastore::backup_info::{BackupGroup, BackupDir, BackupInfo};
use pbs_datastore::index::IndexFile; use pbs_datastore::index::IndexFile;
use pbs_datastore::manifest::{archive_type, ArchiveType, BackupManifest, FileInfo}; use pbs_datastore::manifest::{archive_type, ArchiveType, BackupManifest, FileInfo};
use pbs_tools::fs::lock_dir_noblock_shared; use pbs_tools::fs::lock_dir_noblock_shared;
use pbs_tools::{task_log, task::TaskState}; use pbs_tools::{task_log, task::WorkerTaskContext};
use crate::{ use crate::{
backup::DataStore, backup::DataStore,
@ -22,7 +22,7 @@ use crate::{
/// A VerifyWorker encapsulates a task worker, datastore and information about which chunks have /// A VerifyWorker encapsulates a task worker, datastore and information about which chunks have
/// already been verified or detected as corrupt. /// already been verified or detected as corrupt.
pub struct VerifyWorker { pub struct VerifyWorker {
worker: Arc<dyn TaskState + Send + Sync>, worker: Arc<dyn WorkerTaskContext + Send + Sync>,
datastore: Arc<DataStore>, datastore: Arc<DataStore>,
verified_chunks: Arc<Mutex<HashSet<[u8; 32]>>>, verified_chunks: Arc<Mutex<HashSet<[u8; 32]>>>,
corrupt_chunks: Arc<Mutex<HashSet<[u8; 32]>>>, corrupt_chunks: Arc<Mutex<HashSet<[u8; 32]>>>,
@ -30,7 +30,7 @@ pub struct VerifyWorker {
impl VerifyWorker { impl VerifyWorker {
/// Creates a new VerifyWorker for a given task worker and datastore. /// Creates a new VerifyWorker for a given task worker and datastore.
pub fn new(worker: Arc<dyn TaskState + Send + Sync>, datastore: Arc<DataStore>) -> Self { pub fn new(worker: Arc<dyn WorkerTaskContext + Send + Sync>, datastore: Arc<DataStore>) -> Self {
Self { Self {
worker, worker,
datastore, datastore,
@ -73,7 +73,7 @@ fn verify_blob(
fn rename_corrupted_chunk( fn rename_corrupted_chunk(
datastore: Arc<DataStore>, datastore: Arc<DataStore>,
digest: &[u8;32], digest: &[u8;32],
worker: &dyn TaskState, worker: &dyn WorkerTaskContext,
) { ) {
let (path, digest_str) = datastore.chunk_path(digest); let (path, digest_str) = datastore.chunk_path(digest);

View File

@ -30,7 +30,7 @@ use proxmox::{
use pbs_api_types::{VirtualTapeDrive, LtoTapeDrive, Fingerprint}; use pbs_api_types::{VirtualTapeDrive, LtoTapeDrive, Fingerprint};
use pbs_config::key_config::KeyConfig; use pbs_config::key_config::KeyConfig;
use pbs_tools::{task_log, task::TaskState}; use pbs_tools::task_log;
use pbs_tape::{ use pbs_tape::{
TapeWrite, TapeRead, BlockReadError, MediaContentHeader, TapeWrite, TapeRead, BlockReadError, MediaContentHeader,

View File

@ -17,7 +17,7 @@ fn garbage_collection(worker: &WorkerTask) -> Result<(), Error> {
worker.log("start garbage collection"); worker.log("start garbage collection");
for i in 0..50 { for i in 0..50 {
worker.fail_on_abort()?; worker.check_abort()?;
flog!(worker, "progress {}", i); flog!(worker, "progress {}", i);