[pbs-devel] [RFC proxmox{, -backup} 0/6] add support for configuring max

[pbs-devel] [RFC proxmox{, -backup} 0/6] add support for configuring max

[Hannes Laimer]

This uses POSIX semaphores to control how many backups run at any given
time. Semaphores seemed like a fitting choice given the problem, so I 
added a basic wrapper for named ones to proxmox-sys. Also, the changes
needed to add concurrency limits in PBS were minimal.

A few questions I still have and would be great to get some feedback on,
 - do we want to use semaphores like proposed here? Are there any good
   reasons not to?
 - should we include things like verify/gc/prune? Would an extra sem for
   reading make sense(so separate limits for r and w, idk if there's a
   use-case)?

Nothing UI-wise is included, and for changes to apply the proxy has to
be restarted. Not sure if restarting the proxy is "ok-ish" ux wise, I
guess not :P But changing it involves re-creating the semaphore, which
is fine, just that there won't exist one for a really short time...


proxmox:

Hannes Laimer (2):
  sys: add wrapper for POSIX semaphores
  pbs-api-types: add concurrency_limit to DataStoreConfig

 pbs-api-types/src/datastore.rs |   5 +
 proxmox-sys/Cargo.toml         |   1 +
 proxmox-sys/src/lib.rs         |   2 +
 proxmox-sys/src/semaphore.rs   | 164 +++++++++++++++++++++++++++++++++
 4 files changed, 172 insertions(+)
 create mode 100644 proxmox-sys/src/semaphore.rs


proxmox-backup:

Hannes Laimer (4):
  api: config: update/delete concurrency_limit on datastore
  Cargo.toml: add 'semaphore' feature to proxmox-sys dep
  bin: proxy: initialize concurrency semaphores for datastores
  api: backup: wait for semaphore if one exists

 Cargo.toml                      |  2 +-
 src/api2/backup/mod.rs          | 24 ++++++++++++++++++++++--
 src/api2/config/datastore.rs    |  9 +++++++++
 src/bin/proxmox-backup-proxy.rs | 11 +++++++++++
 4 files changed, 43 insertions(+), 3 deletions(-)


Summary over all repositories:
  8 files changed, 215 insertions(+), 3 deletions(-)

-- 
Generated by git-murpp 0.8.1


_______________________________________________
pbs-devel mailing list
pbs-devel@lists.proxmox.com
https://lists.proxmox.com/cgi-bin/mailman/listinfo/pbs-devel

[pbs-devel] [PATCH proxmox 1/2] sys: add wrapper for POSIX semaphores

[Hannes Laimer]

Signed-off-by: Hannes Laimer <h.laimer@proxmox.com>
---
 proxmox-sys/Cargo.toml       |   1 +
 proxmox-sys/src/lib.rs       |   2 +
 proxmox-sys/src/semaphore.rs | 164 +++++++++++++++++++++++++++++++++++
 3 files changed, 167 insertions(+)
 create mode 100644 proxmox-sys/src/semaphore.rs

diff --git a/proxmox-sys/Cargo.toml b/proxmox-sys/Cargo.toml
index 64cd7cc6..500009b1 100644
--- a/proxmox-sys/Cargo.toml
+++ b/proxmox-sys/Cargo.toml
@@ -31,3 +31,4 @@ logrotate = ["dep:zstd"]
 acl = []
 crypt = ["dep:openssl"]
 timer = []
+semaphore = []
diff --git a/proxmox-sys/src/lib.rs b/proxmox-sys/src/lib.rs
index d8213438..be7471a7 100644
--- a/proxmox-sys/src/lib.rs
+++ b/proxmox-sys/src/lib.rs
@@ -18,6 +18,8 @@ pub mod logrotate;
 pub mod macros;
 pub mod mmap;
 pub mod process_locker;
+#[cfg(feature = "semaphore")]
+pub mod semaphore;
 pub mod systemd;
 
 /// Returns the hosts node name (UTS node name)
diff --git a/proxmox-sys/src/semaphore.rs b/proxmox-sys/src/semaphore.rs
new file mode 100644
index 00000000..61774607
--- /dev/null
+++ b/proxmox-sys/src/semaphore.rs
@@ -0,0 +1,164 @@
+use std::ffi::CString;
+use std::time::{Duration, SystemTime};
+
+use nix::errno::Errno;
+use nix::libc;
+
+use crate::c_try;
+
+/// POSIX named semaphore wrapper using `sem_open(3)`, `sem_wait(3)`, `sem_post(3)`,
+/// `sem_close(3)`, and `sem_unlink(3)`.
+///
+/// Notes:
+/// - On Linux, named semaphores reside in `/dev/shm` as `sem.<name>`.
+/// - Names start with a leading slash (for example `"/mysem"`). If
+///   omitted, we prepend one automatically.
+/// - The `mode` passed to [`Self::create`] is filtered by the creating process `umask`.
+/// - [`Self::close`] only drops this process' handle; use [`Self::unlink`] to remove the name.
+/// - The object is freed once the last open handle is closed and it has been unlinked.
+/// - Errors are returned as [`std::io::Error`] mapped from `errno`; callers can
+///   use [`crate::error::SysError`].
+
+pub struct PosixSemaphore {
+    sem: *mut libc::sem_t,
+}
+
+unsafe impl Send for PosixSemaphore {}
+unsafe impl Sync for PosixSemaphore {}
+
+impl PosixSemaphore {
+    /// Open an existing named semaphore.
+    ///
+    /// Equivalent to `sem_open(name, 0)`. Fails with `ENOENT` (as
+    /// `io::ErrorKind::NotFound`) if the name does not exist.
+    pub fn open(name: &str) -> std::io::Result<Self> {
+        let cname = if name.starts_with('/') {
+            name.to_string()
+        } else {
+            format!("/{}", name)
+        };
+        let cstr = CString::new(cname.clone()).map_err(std::io::Error::other)?;
+        let sem_ptr = unsafe { libc::sem_open(cstr.as_ptr(), 0) };
+        if sem_ptr == libc::SEM_FAILED {
+            return Err(std::io::Error::from_raw_os_error(Errno::last() as i32));
+        }
+        Ok(Self { sem: sem_ptr })
+    }
+
+    /// Create a named semaphore if missing, with `mode` and initial `value`.
+    ///
+    /// Equivalent to `sem_open(name, O_CREAT, mode, value)`. If the semaphore
+    /// already exists, POSIX opens it and ignores `value`.
+    /// The effective permissions are subject to the current process' `umask`.
+    pub fn create(name: &str, mode: u32, value: u32) -> std::io::Result<Self> {
+        let cname = if name.starts_with('/') {
+            name.to_string()
+        } else {
+            format!("/{}", name)
+        };
+        let cstr = CString::new(cname.clone()).map_err(std::io::Error::other)?;
+        let sem_ptr = unsafe { libc::sem_open(cstr.as_ptr(), libc::O_CREAT, mode, value) };
+        if sem_ptr == libc::SEM_FAILED {
+            return Err(std::io::Error::from_raw_os_error(Errno::last() as i32));
+        }
+        Ok(Self { sem: sem_ptr })
+    }
+
+    /// Decrement the semaphore, blocking until a permit is available.
+    ///
+    /// Equivalent to `sem_wait(3)`. May return `Interrupted` on `EINTR`.
+    pub fn wait(&self) -> std::io::Result<()> {
+        c_try!(unsafe { libc::sem_wait(self.sem) });
+        Ok(())
+    }
+
+    /// Try to decrement the semaphore without blocking (`sem_trywait`).
+    /// Returns `Ok(true)` if acquired, `Ok(false)` on `EAGAIN`.
+    pub fn try_wait(&self) -> std::io::Result<bool> {
+        let rc = unsafe { libc::sem_trywait(self.sem) };
+        if rc == 0 {
+            return Ok(true);
+        }
+        let err = Errno::last();
+        if err == Errno::EAGAIN {
+            return Ok(false);
+        }
+        Err(std::io::Error::from_raw_os_error(err as i32))
+    }
+
+    /// Decrement the semaphore or time out after `timeout` (`sem_timedwait`).
+    /// Returns `Ok(true)` if acquired, `Ok(false)` on `ETIMEDOUT`.
+    pub fn timed_wait(&self, timeout: Duration) -> std::io::Result<bool> {
+        let abs = SystemTime::now() + timeout;
+        let d = abs
+            .duration_since(SystemTime::UNIX_EPOCH)
+            .unwrap_or(Duration::from_secs(0));
+        let ts = libc::timespec {
+            tv_sec: d.as_secs() as i64,
+            tv_nsec: d.subsec_nanos() as i64,
+        };
+
+        let rc = unsafe { libc::sem_timedwait(self.sem, &ts as *const libc::timespec) };
+        if rc == 0 {
+            return Ok(true);
+        }
+        let err = Errno::last();
+        if err == Errno::ETIMEDOUT {
+            return Ok(false);
+        }
+        Err(std::io::Error::from_raw_os_error(err as i32))
+    }
+
+    /// Increment the semaphore (`sem_post`).
+    /// Can fail with `EOVERFLOW` if exceeding `SEM_VALUE_MAX`.
+    pub fn post(&self) -> std::io::Result<()> {
+        c_try!(unsafe { libc::sem_post(self.sem) });
+        Ok(())
+    }
+
+    /// Close this process' handle (`sem_close`).
+    /// Use [`Self::unlink`] to remove the name from the namespace.
+    pub fn close(&self) -> std::io::Result<()> {
+        c_try!(unsafe { libc::sem_close(self.sem) });
+        Ok(())
+    }
+
+    /// Unlink the named semaphore (`sem_unlink`). Removes the name so new opens fail.
+    /// Existing handles remain valid until closed.
+    pub fn unlink(name: &str) -> std::io::Result<()> {
+        let cname = if name.starts_with('/') {
+            name.to_string()
+        } else {
+            format!("/{}", name)
+        };
+        let cstr = CString::new(cname).map_err(std::io::Error::other)?;
+        c_try!(unsafe { libc::sem_unlink(cstr.as_ptr()) });
+        Ok(())
+    }
+}
+
+impl Drop for PosixSemaphore {
+    fn drop(&mut self) {
+        unsafe { libc::sem_close(self.sem) };
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn basic_ops() -> std::io::Result<()> {
+        let name = format!("test-semaphore-{}", std::process::id());
+        let _ = PosixSemaphore::unlink(&name);
+        let sem = PosixSemaphore::create(&name, 0o600, 1)?;
+        // Immediately unlink the name so the semaphore is cleaned up once all handles close.
+        PosixSemaphore::unlink(&name)?;
+        assert!(sem.try_wait()?);
+        assert!(!sem.try_wait()?);
+        sem.post()?;
+        assert!(sem.try_wait()?);
+        sem.close()?;
+        Ok(())
+    }
+}
-- 
2.47.2



_______________________________________________
pbs-devel mailing list
pbs-devel@lists.proxmox.com
https://lists.proxmox.com/cgi-bin/mailman/listinfo/pbs-devel

[pbs-devel] [PATCH proxmox 2/2] pbs-api-types: add concurrency_limit to DataStoreConfig

[Hannes Laimer]

Signed-off-by: Hannes Laimer <h.laimer@proxmox.com>
---
 pbs-api-types/src/datastore.rs | 5 +++++
 1 file changed, 5 insertions(+)

diff --git a/pbs-api-types/src/datastore.rs b/pbs-api-types/src/datastore.rs
index ee94ccad..fc79af0a 100644
--- a/pbs-api-types/src/datastore.rs
+++ b/pbs-api-types/src/datastore.rs
@@ -499,6 +499,10 @@ pub struct DataStoreConfig {
     #[updater(skip)]
     #[serde(skip_serializing_if = "Option::is_none")]
     pub backend: Option<String>,
+
+    /// Limit of how many backups can run at the same time
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub concurrency_limit: Option<u32>,
 }
 
 #[api]
@@ -535,6 +539,7 @@ impl DataStoreConfig {
             maintenance_mode: None,
             backing_device: None,
             backend: None,
+            concurrency_limit: None,
         }
     }
 
-- 
2.47.2



_______________________________________________
pbs-devel mailing list
pbs-devel@lists.proxmox.com
https://lists.proxmox.com/cgi-bin/mailman/listinfo/pbs-devel

[pbs-devel] [PATCH proxmox-backup 1/4] api: config: update/delete concurrency_limit on datastore

[Hannes Laimer]

Signed-off-by: Hannes Laimer <h.laimer@proxmox.com>
---
 src/api2/config/datastore.rs | 9 +++++++++
 1 file changed, 9 insertions(+)

diff --git a/src/api2/config/datastore.rs b/src/api2/config/datastore.rs
index e86ded0d..23d56393 100644
--- a/src/api2/config/datastore.rs
+++ b/src/api2/config/datastore.rs
@@ -447,6 +447,8 @@ pub enum DeletableProperty {
     Tuning,
     /// Delete the maintenance-mode property
     MaintenanceMode,
+    /// Delete the concurrency-limit property
+    ConcurrencyLimit,
 }
 
 #[api(
@@ -545,6 +547,9 @@ pub fn update_datastore(
                 DeletableProperty::MaintenanceMode => {
                     data.set_maintenance_mode(None)?;
                 }
+                DeletableProperty::ConcurrencyLimit => {
+                    data.concurrency_limit = None;
+                }
             }
         }
     }
@@ -617,6 +622,10 @@ pub fn update_datastore(
         data.tuning = update.tuning;
     }
 
+    if update.concurrency_limit.is_some() {
+        data.concurrency_limit = update.concurrency_limit;
+    }
+
     let mut maintenance_mode_changed = false;
     if update.maintenance_mode.is_some() {
         maintenance_mode_changed = data.maintenance_mode != update.maintenance_mode;
-- 
2.47.2



_______________________________________________
pbs-devel mailing list
pbs-devel@lists.proxmox.com
https://lists.proxmox.com/cgi-bin/mailman/listinfo/pbs-devel

[pbs-devel] [PATCH proxmox-backup 2/4] Cargo.toml: add 'semaphore' feature to proxmox-sys dep

[Hannes Laimer]

Signed-off-by: Hannes Laimer <h.laimer@proxmox.com>
---
 Cargo.toml | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/Cargo.toml b/Cargo.toml
index 19974da2..645c131f 100644
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -88,7 +88,7 @@ proxmox-shared-cache = "1"
 proxmox-shared-memory = "1"
 proxmox-sortable-macro = "1"
 proxmox-subscription = { version = "1", features = [ "api-types" ] }
-proxmox-sys = "1"
+proxmox-sys = { version = "1", features = [ "semaphore" ] }
 proxmox-systemd = "1"
 proxmox-tfa = { version = "6.0.3", features = [ "api", "api-types" ] }
 proxmox-time = "2"
-- 
2.47.2



_______________________________________________
pbs-devel mailing list
pbs-devel@lists.proxmox.com
https://lists.proxmox.com/cgi-bin/mailman/listinfo/pbs-devel

[pbs-devel] [PATCH proxmox-backup 3/4] bin: proxy: initialize concurrency semaphores for datastores

[Hannes Laimer]

... which have a `concurrency_limit` configured.

Signed-off-by: Hannes Laimer <h.laimer@proxmox.com>
---
 src/bin/proxmox-backup-proxy.rs | 11 +++++++++++
 1 file changed, 11 insertions(+)

diff --git a/src/bin/proxmox-backup-proxy.rs b/src/bin/proxmox-backup-proxy.rs
index cfd93f92..1678be66 100644
--- a/src/bin/proxmox-backup-proxy.rs
+++ b/src/bin/proxmox-backup-proxy.rs
@@ -251,6 +251,17 @@ async fn run() -> Result<(), Error> {
     let acceptor = make_tls_acceptor()?;
     let acceptor = Arc::new(Mutex::new(acceptor));
 
+    // create concurrent tasks semaphore
+    let (config, _digest) = pbs_config::datastore::config()?;
+    let all_stores: Vec<DataStoreConfig> = config.convert_to_typed_array("datastore")?;
+    for store in all_stores {
+        let sem_name = format!("/{}_concurrent", store.name);
+        let _ = proxmox_sys::semaphore::PosixSemaphore::unlink(&sem_name);
+        if let Some(c) = store.concurrency_limit {
+            let _ = proxmox_sys::semaphore::PosixSemaphore::create(&sem_name, 0o600, c);
+        }
+    }
+
     // to renew the acceptor we just add a command-socket handler
     command_sock.register_command("reload-certificate".to_string(), {
         let acceptor = Arc::clone(&acceptor);
-- 
2.47.2



_______________________________________________
pbs-devel mailing list
pbs-devel@lists.proxmox.com
https://lists.proxmox.com/cgi-bin/mailman/listinfo/pbs-devel

[pbs-devel] [PATCH proxmox-backup 4/4] api: backup: wait for semaphore if one exists

[Hannes Laimer]

Backups will start in the order they started waiting, so
basically a FIFO queue.

Signed-off-by: Hannes Laimer <h.laimer@proxmox.com>
---
 src/api2/backup/mod.rs | 24 ++++++++++++++++++++++--
 1 file changed, 22 insertions(+), 2 deletions(-)

diff --git a/src/api2/backup/mod.rs b/src/api2/backup/mod.rs
index ae61ff69..a52a736e 100644
--- a/src/api2/backup/mod.rs
+++ b/src/api2/backup/mod.rs
@@ -19,6 +19,8 @@ use proxmox_router::{
 };
 use proxmox_schema::*;
 use proxmox_sortable_macro::sortable;
+use proxmox_sys::error::SysError;
+use proxmox_sys::semaphore::PosixSemaphore;
 
 use pbs_api_types::{
     ArchiveType, Authid, BackupNamespace, BackupType, Operation, VerifyState,
@@ -226,6 +228,20 @@ fn upgrade_to_backup_protocol(
                     Some(ip) => format!(" from {ip}"),
                     None => "".into(),
                 };
+                let sem = match PosixSemaphore::open(&format!("/{}_concurrent", &store)) {
+                    Ok(_) if benchmark => None,
+                    Ok(sem) => Some(sem),
+                    Err(ref err) if err.not_found() => None,
+                    Err(err) => bail!("could not open semaphore: {err}"),
+                };
+
+                if let Some(ref sem) = sem {
+                    if let Ok(false) = sem.try_wait() {
+                        log::info!("max amount of concurrent tasks reached, waiting for one to finish...");
+                        sem.wait()?;
+                    }
+                }
+
                 env.log(format!(
                     "starting new {worker_type} on datastore '{store}'{origin}: {path:?}",
                 ));
@@ -295,7 +311,7 @@ fn upgrade_to_backup_protocol(
                     }
                 };
 
-                match (res, env.ensure_finished()) {
+                let r = match (res, env.ensure_finished()) {
                     (Ok(_), Ok(())) => {
                         env.log("backup finished successfully");
                         verify(env);
@@ -319,7 +335,11 @@ fn upgrade_to_backup_protocol(
                         proxmox_async::runtime::block_in_place(|| env.remove_backup())?;
                         Err(err)
                     }
-                }
+                };
+                if let Some(ref sem) = sem {
+                    let _ = sem.post();
+                };
+                r
             },
         )?;
 
-- 
2.47.2



_______________________________________________
pbs-devel mailing list
pbs-devel@lists.proxmox.com
https://lists.proxmox.com/cgi-bin/mailman/listinfo/pbs-devel