[PATCH manager 12/19] vzdump: allow jobs to run alongside each other

[Thomas Lamprecht <t.lamprecht@proxmox.com>]

A node ran one backup job at a time. A short job, for example an
hourly one to a nearby target, had to wait until a long nightly job
finished, possibly for hours, and failed once the lock wait time ran
out.

Jobs with the new 'overlap' option share the global lock and run at
the same time as other such jobs. Every guest backup on the node, no
matter which job it belongs to, holds one of the node's backup slots
while running, so overlapping jobs together stay within the node's
backup concurrency. When the concurrency gets lowered, slots above
the new limit keep counting until their guest backups end. Jobs
without the option keep running alone, and a job waiting for that
takes precedence over overlapping jobs starting later, so these cannot
starve it. Jobs to external backup providers always run alone.

A guest that another job is backing up is deferred until that backup
finished, waiting at most the lock wait time. Before starting a guest
backup, the job also checks whether the guest moved to another node in
the meantime.

All of these are file locks held by the processes doing the work, so
they get released whenever such a process ends. Jobs that do not
overlap still take the global lock exclusively, which keeps it working
for the boot-time cleanup of stale backup locks and against older
versions during an upgrade.

Signed-off-by: Thomas Lamprecht <t.lamprecht@proxmox.com>
---
 PVE/API2/VZDump.pm |   2 +-
 PVE/VZDump.pm      | 193 +++++++++++++++++++++++++++++++++++++++++++--
 2 files changed, 188 insertions(+), 7 deletions(-)

diff --git a/PVE/API2/VZDump.pm b/PVE/API2/VZDump.pm
index 73291189f..3e31505f3 100644
--- a/PVE/API2/VZDump.pm
+++ b/PVE/API2/VZDump.pm
@@ -137,7 +137,7 @@ __PACKAGE__->register_method({
             my $vzdump = PVE::VZDump->new($cmdline, $param, $skiplist);
 
             my $LOCK_FH = eval {
-                my $lock_fh = $vzdump->getlock($upid); # only one process allowed
+                my $lock_fh = $vzdump->getlock($upid);
                 $vzdump->init_job();
                 $lock_fh;
             };
diff --git a/PVE/VZDump.pm b/PVE/VZDump.pm
index 33fe8793e..f597f7ed7 100644
--- a/PVE/VZDump.pm
+++ b/PVE/VZDump.pm
@@ -46,6 +46,7 @@ PVE::JSONSchema::register_standard_option(
 my @posix_filesystems = qw(ext3 ext4 nfs nfs4 reiserfs xfs);
 
 my $lockfile = '/var/run/vzdump.lock';
+my $lockdir = '/run/vzdump';
 my $logdir = '/var/log/vzdump';
 
 my @plugins = qw();
@@ -682,6 +683,14 @@ sub new {
 
     $self->{'default-prune-backups'} = $defaults->{'prune-backups'};
 
+    # external providers are not known to cope with other backups running at the same time
+    my $uses_provider = $opts->{storage} && eval {
+        PVE::Storage::storage_has_feature(
+            PVE::Storage::config(), $opts->{storage}, 'backup-provider',
+        );
+    };
+    $self->{exclusive} = !$opts->{overlap} || $uses_provider;
+
     return $self;
 }
 
@@ -781,14 +790,32 @@ sub getlock {
     my $maxwait = $self->{opts}->{lockwait} || $self->{lockwait};
 
     die "missing UPID" if !$upid; # should not happen
+    $self->{upid} = $upid;
+
+    mkdir $lockdir;
 
     my $SERVER_FLCK;
     if (!open($SERVER_FLCK, '>>', "$lockfile")) {
         debugmsg('err', "can't open lock on file '$lockfile' - $!", undef, 1);
         die "can't open lock on file '$lockfile' - $!";
     }
+    open(my $gate, '>>', "$lockdir/gate.lock")
+        or die "can't open lock on file '$lockdir/gate.lock' - $!\n";
 
-    if (!flock($SERVER_FLCK, LOCK_EX | LOCK_NB)) {
+    # Jobs that may overlap share the global lock. A job waiting to get it exclusively holds the
+    # gate meanwhile, so overlapping jobs starting later cannot keep it waiting forever.
+    my $mode = $self->{exclusive} ? LOCK_EX : LOCK_SH;
+    my $acquire = sub {
+        my ($flags) = @_;
+
+        flock($gate, $mode | $flags) or return 0;
+        my $locked = flock($SERVER_FLCK, $mode | $flags);
+        flock($gate, LOCK_UN);
+
+        return $locked;
+    };
+
+    if (!$acquire->(LOCK_NB)) {
         if (!$maxwait) {
             debugmsg('err', "can't acquire lock '$lockfile' (wait = 0)", undef, 1);
             die "can't acquire lock '$lockfile' (wait = 0)";
@@ -800,7 +827,7 @@ sub getlock {
 
             local $SIG{ALRM} = sub { alarm(0); die "got timeout\n"; };
 
-            if (!flock($SERVER_FLCK, LOCK_EX)) {
+            if (!$acquire->(0)) {
                 my $err = $!;
                 close($SERVER_FLCK);
                 alarm(0);
@@ -1635,12 +1662,145 @@ my sub finish_guest_backup {
     }
 }
 
+my sub try_lock_file {
+    my ($path) = @_;
+
+    open(my $fh, '>>', $path) or die "unable to open lock file '$path' - $!\n";
+    return $fh if flock($fh, LOCK_EX | LOCK_NB);
+
+    close($fh);
+    return undef;
+}
+
+my sub try_guest_lock {
+    my ($self, $vmid) = @_;
+
+    my $fh = try_lock_file("$lockdir/guest-$vmid.lock") or return undef;
+    truncate($fh, 0);
+    print $fh "$self->{upid}\n";
+    $fh->flush();
+
+    return $fh;
+}
+
+# Slots above a lowered limit stay in use until their backups end, no new one starts before.
+my sub try_slot {
+    my ($prefix, $limit) = @_;
+
+    for my $path (glob("$prefix-*.lock")) {
+        my ($slot) = $path =~ m/^\Q$prefix\E-(\d+)\.lock$/ or next;
+        next if $slot <= $limit;
+        my $fh = try_lock_file("$prefix-$slot.lock") or return undef;
+        close($fh);
+    }
+    for my $slot (1 .. $limit) {
+        my $fh = try_lock_file("$prefix-$slot.lock");
+        return $fh if $fh;
+    }
+    return undef;
+}
+
+my sub try_node_slot {
+    return try_slot("$lockdir/slot", node_backup_concurrency());
+}
+
+my sub log_wait {
+    my ($self, $reason) = @_;
+
+    return if ($self->{'waiting-for'} // '') eq $reason;
+    $self->{'waiting-for'} = $reason;
+    debugmsg('info', "waiting for $reason");
+}
+
+my sub guest_moved_to {
+    my ($vmid) = @_;
+
+    PVE::Cluster::cfs_update();
+    my $node = PVE::Cluster::get_vmlist()->{ids}->{$vmid}->{node};
+
+    return defined($node) && $node ne PVE::INotify::nodename() ? $node : undef;
+}
+
+my sub claim_next_guest {
+    my ($self, $queue, $deferred) = @_;
+
+    my $maxwait = ($self->{opts}->{lockwait} || 0) * 60;
+
+    for (my $i = 0; $i < scalar($queue->@*); $i++) {
+        my $task = $queue->[$i];
+        my $vmid = $task->{vmid};
+
+        my $guest_lock = try_guest_lock($self, $vmid);
+        if (!$guest_lock) {
+            my $holder = PVE::Tools::file_read_firstline("$lockdir/guest-$vmid.lock") // '';
+            if (!defined($deferred->{$vmid})) {
+                $deferred->{$vmid} = time();
+                debugmsg('info', "VM $vmid is being backed up by '$holder', deferring");
+            } elsif (time() - $deferred->{$vmid} >= $maxwait) {
+                splice($queue->@*, $i--, 1);
+                skip_task($task, "still being backed up by '$holder'");
+            }
+            next;
+        }
+
+        my $slot = try_node_slot();
+        if (!$slot) {
+            log_wait($self, "a free backup slot on the node");
+            return;
+        }
+        if (my $node = guest_moved_to($vmid)) {
+            splice($queue->@*, $i--, 1);
+            skip_task($task, "guest moved to node '$node'");
+            next;
+        }
+        delete $self->{'waiting-for'};
+
+        splice($queue->@*, $i--, 1);
+        return ($task, [$guest_lock, $slot]);
+    }
+
+    return;
+}
+
+my sub skip_after_error {
+    my ($queue, $err) = @_;
+
+    return if $err =~ m/interrupted by signal$/;
+
+    chomp(my $reason = $err);
+    skip_task($_, "not started due to error - $reason") for splice($queue->@*);
+}
+
+my sub exec_serial_backup_tasks {
+    my ($self, $tasklist) = @_;
+
+    my @queue = $tasklist->@*;
+    my $deferred = {};
+
+    eval {
+        while (scalar(@queue)) {
+            my ($task, $locks) = claim_next_guest($self, \@queue, $deferred);
+            if ($task) {
+                $self->exec_backup_task($task);
+            } elsif (scalar(@queue)) {
+                sleep(1);
+            }
+        }
+    };
+    if (my $err = $@) {
+        skip_after_error(\@queue, $err);
+        die $err;
+    }
+}
+
 my sub exec_parallel_backup_tasks {
     my ($self, $tasklist, $concurrency) = @_;
 
     my @queue = $tasklist->@*;
+    my $deferred = {};
     my $running = {};
     my $interrupted = 0;
+    my $error;
     my $job_pid = $$;
 
     # Stopping a task signals all its processes, so the guest backups stop on their own. Signalling
@@ -1649,8 +1809,17 @@ my sub exec_parallel_backup_tasks {
         sub { $interrupted = 1; };
 
     while (scalar(@queue) || scalar(keys $running->%*)) {
-        while (!$interrupted && scalar(@queue) && scalar(keys $running->%*) < $concurrency) {
-            my $task = shift @queue;
+        while (!$interrupted && !defined($error) && scalar(keys $running->%*) < $concurrency) {
+            my ($task, $locks) = eval { claim_next_guest($self, \@queue, $deferred) };
+            if (my $err = $@) {
+                # let running guest backups finish, they would get aborted otherwise
+                debugmsg('err', "unable to start further guest backups - $err");
+                $error = $err;
+                last;
+            }
+            last if !$task;
+
+            # the backup process keeps holding the locks through its inherited copies of $locks
             my $child = eval { fork_guest_backup($self, $task, $running, \$interrupted) };
             # a guest backup process signalled while it exits dies into this eval
             POSIX::_exit(1) if $$ != $job_pid;
@@ -1662,7 +1831,14 @@ my sub exec_parallel_backup_tasks {
             unshift @queue, $task if $child->{cancelled};
         }
 
-        last if $interrupted && !scalar(keys $running->%*);
+        if ($interrupted || defined($error)) {
+            last if !scalar(keys $running->%*);
+        }
+
+        if (!scalar(keys $running->%*)) {
+            sleep(1);
+            next;
+        }
 
         my $handles = {};
         for my $child (values $running->%*) {
@@ -1681,6 +1857,11 @@ my sub exec_parallel_backup_tasks {
     }
 
     die "interrupted by signal\n" if $interrupted;
+
+    if (defined($error)) {
+        skip_after_error(\@queue, $error);
+        die $error;
+    }
 }
 
 sub exec_backup {
@@ -1716,7 +1897,7 @@ sub exec_backup {
             debugmsg('info', "running up to $concurrency guest backups at the same time");
             exec_parallel_backup_tasks($self, $tasklist, $concurrency);
         } else {
-            $self->exec_backup_task($_) for $tasklist->@*;
+            exec_serial_backup_tasks($self, $tasklist);
         }
 
         $self->{'backup-provider'}->job_cleanup() if $self->{'backup-provider'};
-- 
2.47.3