From mboxrd@z Thu Jan 1 00:00:00 1970 Return-Path: Received: from gate001.proxmox.com (gate001.proxmox.com [IPv6:2a0f:8001:1:32::40]) by lore.proxmox.com (Postfix) with ESMTPS id 5479B1FF09D for ; Mon, 02 Nov 2026 00:01:58 +0100 (CET) Received: from gate001.proxmox.com (localhost.localdomain [127.0.0.1]) by gate001.proxmox.com (Proxmox) with ESMTP id 08FD2218DA; Sun, 01 Nov 2026 23:59:58 +0100 (CET) From: Thomas Lamprecht To: pve-devel@lists.proxmox.com Subject: [PATCH manager 11/19] vzdump: back up guests of a job in parallel Date: Sun, 1 Nov 2026 23:58:38 +0100 Message-ID: <20261101225849.884587-12-t.lamprecht@proxmox.com> X-Mailer: git-send-email 2.47.3 In-Reply-To: <20261101225849.884587-1-t.lamprecht@proxmox.com> References: <20261101225849.884587-1-t.lamprecht@proxmox.com> MIME-Version: 1.0 Content-Transfer-Encoding: 8bit X-Bm-Milter-Handled: 55990f41-d878-4baa-be0a-ee34c49e34d2 X-Bm-Transport-Timestamp: 1793573948508 X-SPAM-LEVEL: Spam detection results: 0 AWL 0.379 Adjusted score from AWL reputation of From: address DMARC_MISSING 0.1 Missing DMARC policy KAM_DMARC_STATUS 0.01 Test Rule for DKIM or SPF Failure with Strict Alignment (newer systems) RCVD_IN_DNSWL_MED -2.3 Sender listed at https://www.dnswl.org/, medium trust SPF_HELO_NONE 0.001 SPF: HELO does not publish an SPF Record SPF_PASS -0.001 SPF: sender matches SPF record Message-ID-Hash: GJB4R7WYBT3BPHJNPLNG2PIK4CUBUORX X-Message-ID-Hash: GJB4R7WYBT3BPHJNPLNG2PIK4CUBUORX X-MailFrom: t.lamprecht@proxmox.com X-Mailman-Rule-Misses: dmarc-mitigation; no-senders; approved; loop; banned-address; emergency; member-moderation; nonmember-moderation; administrivia; implicit-dest; max-recipients; max-size; news-moderation; no-subject; digests; suspicious-header X-Mailman-Version: 3.3.10 Precedence: list List-Id: Proxmox VE development discussion List-Help: List-Owner: List-Post: List-Subscribe: List-Unsubscribe: Each node backed up the guests of a job strictly one after the other. How long a job took thus depended on how its guests were spread over the nodes: guests all on one node were backed up in series, even if that node and the backup target could handle more. Allow backing up several guests at the same time, up to the backup concurrency of the node. It comes from the new node configuration option or the datacenter-wide default, and stays at one guest unless configured. A job can lower it with its 'concurrency' option. Each guest backup then runs in a forked process. The job process collects the results, prefixes the guest output in the task log with the guest ID, keeps the warning count, and sends a single notification as before. A guest backup process that ends without a result is reported as failed. Stopping the task signals all its processes, so the guest backups stop on their own while the job process waits for them. It does not signal them again, which could interrupt their cleanup. A stop request that arrives while a guest backup process is forked does not reach that process, so it only starts its backup once the job process confirmed that there was none; otherwise the guest counts as not started. A signal that only reached the job process keeps it from starting further guests. A guest listed more than once is backed up once, as its backups would otherwise run at the same time. Backups to external providers stay sequential, as providers are not known to cope with backups from multiple processes. With a concurrency of one, guests are backed up in the job process as before. Signed-off-by: Thomas Lamprecht --- PVE/NodeConfig.pm | 7 + PVE/VZDump.pm | 244 +++++++++++++++++++++++++++++++- test/Makefile | 6 +- test/vzdump_concurrency_test.pl | 89 ++++++++++++ 4 files changed, 340 insertions(+), 6 deletions(-) create mode 100755 test/vzdump_concurrency_test.pl diff --git a/PVE/NodeConfig.pm b/PVE/NodeConfig.pm index cc1a2a296..fbd7ecf87 100644 --- a/PVE/NodeConfig.pm +++ b/PVE/NodeConfig.pm @@ -108,6 +108,13 @@ my $confdesc = { default => 80, optional => 1, }, + 'backup-concurrency' => { + description => "Maximal number of guests backed up at the same time on this node." + . " Overrides the datacenter-wide default.", + type => 'integer', + minimum => 1, + optional => 1, + }, }; my $wakeonlan_desc = { diff --git a/PVE/VZDump.pm b/PVE/VZDump.pm index b12281790..33fe8793e 100644 --- a/PVE/VZDump.pm +++ b/PVE/VZDump.pm @@ -11,6 +11,7 @@ use IO::File; use IO::Select; use IPC::Open3; use POSIX qw(strftime); +use Storable; use Time::Local; use PVE::Cluster qw(cfs_read_file); @@ -19,6 +20,7 @@ use PVE::Exception qw(raise_param_exc); use PVE::HA::Config; use PVE::HA::Env::PVE2; use PVE::JSONSchema qw(get_standard_option); +use PVE::NodeConfig; use PVE::Notify; use PVE::RPCEnvironment; use PVE::Storage; @@ -288,10 +290,11 @@ sub check_vmids { my (@vmids) = @_; my $res = []; + my $seen = {}; for my $vmid (sort { $a <=> $b } @vmids) { die "ERROR: strange VM ID '${vmid}'\n" if $vmid !~ m/^\d+$/; $vmid = int($vmid); # remove leading zeros - next if !$vmid; + next if !$vmid || $seen->{$vmid}++; push @$res, $vmid; } @@ -1454,6 +1457,232 @@ sub skip_all_guests { return $tasklist; } +my sub fail_task { + my ($task, $err) = @_; + + $task->{state} = 'err'; + $task->{msg} = $err; + debugmsg('err', "Backup of VM $task->{vmid} failed - $err"); +} + +sub node_backup_concurrency { + my $node_conf = PVE::NodeConfig::load_config(PVE::INotify::nodename()); + + return $node_conf->{'backup-concurrency'} + // cfs_read_file('datacenter.cfg')->{'backup-concurrency'}->{node} // 1; +} + +sub guest_backup_concurrency { + my ($self) = @_; + + # external providers are not known to cope with backups from multiple processes + return 1 if $self->{'backup-provider'}; + + my $node_limit = node_backup_concurrency(); + my $limit = $self->{opts}->{concurrency} // $node_limit; + + return $limit < $node_limit ? $limit : $node_limit; +} + +my $guest_result_keys = [qw(backuptime hostname log msg size state target tmplog vmtype)]; + +my sub fork_guest_backup { + my ($self, $task, $running, $interrupted) = @_; + + pipe(my $output_reader, my $output_writer) or die "unable to create pipe - $!\n"; + pipe(my $result_reader, my $result_writer) or die "unable to create pipe - $!\n"; + pipe(my $start_reader, my $start_writer) or die "unable to create pipe - $!\n"; + + # Until the child set up its own handlers, a signal would only reach the inherited one of the + # job process and get lost. + my $signals = POSIX::SigSet->new( + POSIX::SIGINT(), + POSIX::SIGTERM(), + POSIX::SIGQUIT(), + POSIX::SIGHUP(), + POSIX::SIGPIPE(), + ); + my $old_mask = POSIX::SigSet->new(); + POSIX::sigprocmask(POSIX::SIG_BLOCK(), $signals, $old_mask) + or die "unable to block signals - $!\n"; + + my $pid = fork(); + if (!defined($pid)) { + my $err = $!; + POSIX::sigprocmask(POSIX::SIG_SETMASK(), $old_mask); + die "unable to fork guest backup - $err\n"; + } + + if (!$pid) { + my $interrupted = 0; + $SIG{INT} = $SIG{TERM} = $SIG{QUIT} = $SIG{HUP} = $SIG{PIPE} = sub { + # a second signal must not abort the cleanup the first one started + die "interrupted by signal\n" if !$interrupted++; + }; + + my $exitcode = 1; + eval { + POSIX::sigprocmask(POSIX::SIG_SETMASK(), $old_mask); + close($output_reader); + close($result_reader); + close($start_writer); + # only the job process reads the output of the other guest backups + for my $sibling (values $running->%*) { + close($_) for grep { defined } $sibling->@{qw(output result)}; + } + open(STDOUT, '>&', $output_writer) or die "unable to redirect STDOUT - $!\n"; + open(STDERR, '>&', $output_writer) or die "unable to redirect STDERR - $!\n"; + close($output_writer); + + sysread($start_reader, my $start, 1) or die "interrupted by signal\n"; + close($start_reader); + + PVE::INotify::inotify_close(); + + eval { $self->exec_backup_task($task) }; + my $err = $@; + + my $result = { map { $_ => $task->{$_} } $guest_result_keys->@* }; + print $result_writer Storable::nfreeze($result); + close($result_writer); + + die $err if $err; + $exitcode = 0; + }; + warn $@ if $@ && $@ !~ m/^interrupted by signal$/; + STDOUT->flush(); + POSIX::_exit($exitcode); + } + + POSIX::sigprocmask(POSIX::SIG_SETMASK(), $old_mask); + + close($output_writer); + close($result_writer); + close($start_reader); + + # A stop request that came in while forking reached only the job process, so the backup + # process must not start then. + my $cancelled = $$interrupted; + if (!$cancelled) { + local $SIG{PIPE} = 'IGNORE'; + syswrite($start_writer, '1'); + } + close($start_writer); + + return { + pid => $pid, + cancelled => $cancelled, + task => $task, + output => $output_reader, + result => $result_reader, + 'output-buffer' => '', + 'result-data' => '', + }; +} + +my sub forward_guest_output { + my ($child, $data, $final) = @_; + + my $vmid = $child->{task}->{vmid}; + + $child->{'output-buffer'} .= $data; + $child->{'output-buffer'} .= "\n" if $final && length($child->{'output-buffer'}); + + while ($child->{'output-buffer'} =~ s/^(.*)\n//) { + my $line = $1; + # warnings of a guest backup process count for the task, but its counter ends with it + PVE::RPCEnvironment::get()->{warning_count}++ if $line =~ m/^WARN: /; + print STDERR "$vmid: $line\n"; + } +} + +my sub read_guest_backup_pipe { + my ($child, $kind) = @_; + + my $count = sysread($child->{$kind}, my $data, 64 * 1024); + return if !defined($count) && ($!{EINTR} || $!{EAGAIN}); + + if (!$count) { + close(delete $child->{$kind}); + forward_guest_output($child, '', 1) if $kind eq 'output'; + } elsif ($kind eq 'output') { + forward_guest_output($child, $data); + } else { + $child->{'result-data'} .= $data; + } +} + +my sub finish_guest_backup { + my ($child, $status) = @_; + + # the backup process is gone, but helpers it started might still hold the output pipe + for my $kind (qw(result output)) { + while ($child->{$kind} && IO::Select->new($child->{$kind})->can_read(0)) { + read_guest_backup_pipe($child, $kind); + } + close(delete $child->{$kind}) if $child->{$kind}; + } + forward_guest_output($child, '', 1); + return if $child->{cancelled}; + + my $task = $child->{task}; + my $result = eval { Storable::thaw($child->{'result-data'}) }; + if (ref($result) eq 'HASH') { + $task->{$_} = $result->{$_} for keys $result->%*; + } else { + my $reason = $status & 127 ? "signal " . ($status & 127) : "exit code " . ($status >> 8); + fail_task($task, "backup process ended unexpectedly ($reason)"); + } +} + +my sub exec_parallel_backup_tasks { + my ($self, $tasklist, $concurrency) = @_; + + my @queue = $tasklist->@*; + my $running = {}; + my $interrupted = 0; + my $job_pid = $$; + + # Stopping a task signals all its processes, so the guest backups stop on their own. Signalling + # them again could interrupt their cleanup. + local $SIG{INT} = local $SIG{TERM} = local $SIG{QUIT} = local $SIG{HUP} = local $SIG{PIPE} = + sub { $interrupted = 1; }; + + while (scalar(@queue) || scalar(keys $running->%*)) { + while (!$interrupted && scalar(@queue) && scalar(keys $running->%*) < $concurrency) { + my $task = shift @queue; + 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; + if (my $err = $@) { + fail_task($task, $err); + next; + } + $running->{ $child->{pid} } = $child; + unshift @queue, $task if $child->{cancelled}; + } + + last if $interrupted && !scalar(keys $running->%*); + + my $handles = {}; + for my $child (values $running->%*) { + $handles->{ fileno($child->{$_}) } = [$child, $_] + for grep { $child->{$_} } qw(output result); + } + my $select = IO::Select->new(map { $_->[0]->{ $_->[1] } } values $handles->%*); + for my $fh ($select->can_read(1)) { + read_guest_backup_pipe($handles->{ fileno($fh) }->@*); + } + + for my $pid (keys $running->%*) { + next if waitpid($pid, POSIX::WNOHANG()) != $pid; + finish_guest_backup(delete $running->{$pid}, $?); + } + } + + die "interrupted by signal\n" if $interrupted; +} + sub exec_backup { my ($self, $rpcenv, $authuser) = @_; @@ -1474,22 +1703,27 @@ sub exec_backup { open my $job_start_fd, '>', \$job_start_log; open my $job_end_fd, '>', \$job_end_log; + my $concurrency = $self->guest_backup_concurrency(); + $concurrency = scalar($tasklist->@*) if scalar($tasklist->@*) < $concurrency; + my $starttime = time(); - my $errcount = 0; eval { $self->{'backup-provider'}->job_init($starttime) if $self->{'backup-provider'}; $self->run_hook_script('job-start', undef, $job_start_fd); - foreach my $task (@$tasklist) { - $self->exec_backup_task($task); - $errcount += 1 if $task->{state} ne 'ok'; + if ($concurrency > 1) { + 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->@*; } $self->{'backup-provider'}->job_cleanup() if $self->{'backup-provider'}; $self->run_hook_script('job-end', undef, $job_end_fd); }; my $err = $@; + my $errcount = grep { $_->{state} ne 'ok' } $tasklist->@*; if ($err) { if ($self->{'backup-provider'}) { diff --git a/test/Makefile b/test/Makefile index 2dfd20a41..1eb94f4e0 100644 --- a/test/Makefile +++ b/test/Makefile @@ -18,7 +18,7 @@ replication%.t: replication_test%.pl ./$< .PHONY: test-vzdump -test-vzdump: test-vzdump-guest-included test-vzdump-new +test-vzdump: test-vzdump-guest-included test-vzdump-new test-vzdump-concurrency .PHONY: test-vzdump-guest-included test-vzdump-guest-included: @@ -28,6 +28,10 @@ test-vzdump-guest-included: test-vzdump-new: ./vzdump_new_test.pl +.PHONY: test-vzdump-concurrency +test-vzdump-concurrency: + ./vzdump_concurrency_test.pl + .PHONY: test-osd test-osd: ./OSD_test.pl diff --git a/test/vzdump_concurrency_test.pl b/test/vzdump_concurrency_test.pl new file mode 100755 index 000000000..46d679490 --- /dev/null +++ b/test/vzdump_concurrency_test.pl @@ -0,0 +1,89 @@ +#!/usr/bin/perl + +use strict; +use warnings; + +use lib '..'; + +use Test::More; +use Test::MockModule; + +use PVE::VZDump; + +my $node_config = {}; +my $datacenter_config = {}; + +my $node_config_module = Test::MockModule->new('PVE::NodeConfig'); +$node_config_module->mock(load_config => sub { return $node_config; }); + +my $vzdump_module = Test::MockModule->new('PVE::VZDump'); +$vzdump_module->mock( + cfs_read_file => sub { + my ($filename) = @_; + die "unexpected file '$filename'\n" if $filename ne 'datacenter.cfg'; + return $datacenter_config; + }, +); + +my $tests = [ + { + description => 'defaults keep backing up one guest at a time', + expected => 1, + }, + { + description => 'datacenter default applies to the node', + datacenter => 4, + expected => 4, + }, + { + description => 'node configuration overrides datacenter default', + datacenter => 4, + node => 2, + expected => 2, + }, + { + description => 'node configuration applies without datacenter default', + node => 3, + expected => 3, + }, + { + description => 'job limit lowers node limit', + node => 4, + job => 2, + expected => 2, + }, + { + description => 'job limit cannot raise node limit', + node => 2, + job => 8, + expected => 2, + }, + { + description => 'job limit does not raise the default', + job => 8, + expected => 1, + }, + { + description => 'external backup providers back up one guest at a time', + node => 4, + job => 4, + provider => 1, + expected => 1, + }, +]; + +for my $test ($tests->@*) { + $datacenter_config = + defined($test->{datacenter}) + ? { 'backup-concurrency' => { node => $test->{datacenter} } } + : {}; + $node_config = defined($test->{node}) ? { 'backup-concurrency' => $test->{node} } : {}; + + my $self = bless { opts => {} }, 'PVE::VZDump'; + $self->{opts}->{concurrency} = $test->{job} if defined($test->{job}); + $self->{'backup-provider'} = {} if $test->{provider}; + + is($self->guest_backup_concurrency(), $test->{expected}, $test->{description}); +} + +done_testing(); -- 2.47.3