[pve-devel] [PATCH v3 pve-zsync 1/2] Refactor locking
Fabian Ebner
f.ebner at proxmox.com
Mon Oct 7 11:16:10 CEST 2019
This introduces a new locked() mechanism allowing to enclose locked
sections in a cleaner way. There's only two types of locks namely one
for state and cron (they are always read together and almost always
written together) and one for sync.
Signed-off-by: Fabian Ebner <f.ebner at proxmox.com>
---
Changes from v2:
* Split into two patches
Changes from v1:
* Refactored locking as Thomas and Fabian suggested
pve-zsync | 239 +++++++++++++++++++++++++++---------------------------
1 file changed, 119 insertions(+), 120 deletions(-)
diff --git a/pve-zsync b/pve-zsync
index 425ffa2..f7bf5bd 100755
--- a/pve-zsync
+++ b/pve-zsync
@@ -18,7 +18,6 @@ my $PATH = "/usr/sbin";
my $PVE_DIR = "/etc/pve/local";
my $QEMU_CONF = "${PVE_DIR}/qemu-server";
my $LXC_CONF = "${PVE_DIR}/lxc";
-my $LOCKFILE = "$CONFIG_PATH/${PROGNAME}.lock";
my $PROG_PATH = "$PATH/${PROGNAME}";
my $INTERVAL = 15;
my $DEBUG;
@@ -110,14 +109,20 @@ sub cut_target_width {
return "$head/" . $path . "/$tail";
}
-sub lock {
- my ($fh) = @_;
- flock($fh, LOCK_EX) || die "Can't lock config - $!\n";
-}
+sub locked {
+ my ($lock_fn, $code) = @_;
+
+ my $lock_fh = IO::File->new("> $lock_fn");
+
+ flock($lock_fh, LOCK_EX) || die "Couldn't acquire lock - $!\n";
+ my $res = eval { $code->() };
+ my $err = $@;
+
+ flock($lock_fh, LOCK_UN) || warn "Error unlocking - $!\n";
+ die "$err" if $err;
-sub unlock {
- my ($fh) = @_;
- flock($fh, LOCK_UN) || die "Can't unlock config- $!\n";
+ close($lock_fh);
+ return $res;
}
sub get_status {
@@ -338,13 +343,11 @@ sub update_state {
my $text;
my $in_fh;
- eval {
-
+ if (-e $STATE) {
$in_fh = IO::File->new("< $STATE");
die "Could not open file $STATE: $!\n" if !$in_fh;
- lock($in_fh);
$text = <$in_fh>;
- };
+ }
my $out_fh = IO::File->new("> $STATE.new");
die "Could not open file ${STATE}.new: $!\n" if !$out_fh;
@@ -376,9 +379,7 @@ sub update_state {
close($out_fh);
rename "$STATE.new", $STATE;
- eval {
- close($in_fh);
- };
+ close($in_fh);
return $states;
}
@@ -395,7 +396,6 @@ sub update_cron {
my $fh = IO::File->new("< $CRONJOBS");
die "Could not open file $CRONJOBS: $!\n" if !$fh;
- lock($fh);
my @test = <$fh>;
@@ -502,43 +502,45 @@ sub vm_exists {
sub init {
my ($param) = @_;
- my $cfg = read_cron();
+ locked("$CONFIG_PATH/cron_and_state.lock", sub {
+ my $cfg = read_cron();
- my $job = param_to_job($param);
+ my $job = param_to_job($param);
- $job->{state} = "ok";
- $job->{lsync} = 0;
+ $job->{state} = "ok";
+ $job->{lsync} = 0;
- my $source = parse_target($param->{source});
- my $dest = parse_target($param->{dest});
+ my $source = parse_target($param->{source});
+ my $dest = parse_target($param->{dest});
- if (my $ip = $dest->{ip}) {
- run_cmd(['ssh-copy-id', '-i', '/root/.ssh/id_rsa.pub', "$param->{dest_user}\@$ip"]);
- }
+ if (my $ip = $dest->{ip}) {
+ run_cmd(['ssh-copy-id', '-i', '/root/.ssh/id_rsa.pub', "$param->{dest_user}\@$ip"]);
+ }
- if (my $ip = $source->{ip}) {
- run_cmd(['ssh-copy-id', '-i', '/root/.ssh/id_rsa.pub', "$param->{source_user}\@$ip"]);
- }
+ if (my $ip = $source->{ip}) {
+ run_cmd(['ssh-copy-id', '-i', '/root/.ssh/id_rsa.pub', "$param->{source_user}\@$ip"]);
+ }
- die "Pool $dest->{all} does not exists\n" if !check_pool_exists($dest, $param->{dest_user});
+ die "Pool $dest->{all} does not exists\n" if !check_pool_exists($dest, $param->{dest_user});
- if (!defined($source->{vmid})) {
- die "Pool $source->{all} does not exists\n" if !check_pool_exists($source, $param->{source_user});
- }
+ if (!defined($source->{vmid})) {
+ die "Pool $source->{all} does not exists\n" if !check_pool_exists($source, $param->{source_user});
+ }
- my $vm_type = vm_exists($source, $param->{source_user});
- $job->{vm_type} = $vm_type;
- $source->{vm_type} = $vm_type;
+ my $vm_type = vm_exists($source, $param->{source_user});
+ $job->{vm_type} = $vm_type;
+ $source->{vm_type} = $vm_type;
- die "VM $source->{vmid} doesn't exist\n" if $source->{vmid} && !$vm_type;
+ die "VM $source->{vmid} doesn't exist\n" if $source->{vmid} && !$vm_type;
- die "Config already exists\n" if $cfg->{$job->{source}}->{$job->{name}};
+ die "Config already exists\n" if $cfg->{$job->{source}}->{$job->{name}};
- #check if vm has zfs disks if not die;
- get_disks($source, $param->{source_user}) if $source->{vmid};
+ #check if vm has zfs disks if not die;
+ get_disks($source, $param->{source_user}) if $source->{vmid};
- update_cron($job);
- update_state($job);
+ update_cron($job);
+ update_state($job);
+ }); #cron and state lock
eval {
sync($param) if !$param->{skip};
@@ -568,96 +570,89 @@ sub get_job {
sub destroy_job {
my ($param) = @_;
- my $job = get_job($param);
- $job->{state} = "del";
-
- update_cron($job);
- update_state($job);
+ locked("$CONFIG_PATH/cron_and_state.lock", sub {
+ my $job = get_job($param);
+ $job->{state} = "del";
+ update_cron($job);
+ update_state($job);
+ });
}
sub sync {
my ($param) = @_;
- my $lock_fh = IO::File->new("> $LOCKFILE");
- die "Can't open Lock File: $LOCKFILE $!\n" if !$lock_fh;
- lock($lock_fh);
-
my $date = get_date();
my $job;
- eval {
- $job = get_job($param);
- };
-
- if ($job && defined($job->{state}) && $job->{state} eq "syncing") {
- die "Job --source $param->{source} --name $param->{name} is syncing at the moment";
- }
-
- my $dest = parse_target($param->{dest});
- my $source = parse_target($param->{source});
+ my $dest;
+ my $source;
+ my $vm_type;
- my $sync_path = sub {
- my ($source, $dest, $job, $param, $date) = @_;
+ locked("$CONFIG_PATH/sync.lock", sub {
+ locked("$CONFIG_PATH/cron_and_state.lock", sub {
+ eval { $job = get_job($param); };
- ($source->{old_snap}, $source->{last_snap}) = snapshot_get($source, $dest, $param->{maxsnap}, $param->{name}, $param->{source_user});
-
- snapshot_add($source, $dest, $param->{name}, $date, $param->{source_user}, $param->{dest_user});
-
- send_image($source, $dest, $param);
-
- snapshot_destroy($source, $dest, $param->{method}, $source->{old_snap}, $param->{source_user}, $param->{dest_user}) if ($source->{destroy} && $source->{old_snap});
+ if ($job && defined($job->{state}) && $job->{state} eq "syncing") {
+ die "Job --source $param->{source} --name $param->{name} is syncing at the moment";
+ }
- };
+ $dest = parse_target($param->{dest});
+ $source = parse_target($param->{source});
- my $vm_type = vm_exists($source, $param->{source_user});
- $source->{vm_type} = $vm_type;
+ $vm_type = vm_exists($source, $param->{source_user});
+ $source->{vm_type} = $vm_type;
- if ($job) {
- $job->{state} = "syncing";
- $job->{vm_type} = $vm_type if !$job->{vm_type};
- update_state($job);
- }
+ if ($job) {
+ $job->{state} = "syncing";
+ $job->{vm_type} = $vm_type if !$job->{vm_type};
+ update_state($job);
+ }
+ }); #cron and state lock
+
+ my $sync_path = sub {
+ my ($source, $dest, $job, $param, $date) = @_;
+ ($source->{old_snap}, $source->{last_snap}) = snapshot_get($source, $dest, $param->{maxsnap}, $param->{name}, $param->{source_user});
+ snapshot_add($source, $dest, $param->{name}, $date, $param->{source_user}, $param->{dest_user});
+ send_image($source, $dest, $param);
+ snapshot_destroy($source, $dest, $param->{method}, $source->{old_snap}, $param->{source_user}, $param->{dest_user}) if ($source->{destroy} && $source->{old_snap});
+ };
- eval{
- if ($source->{vmid}) {
- die "VM $source->{vmid} doesn't exist\n" if !$vm_type;
- die "source-user has to be root for syncing VMs\n" if ($param->{source_user} ne "root");
- my $disks = get_disks($source, $param->{source_user});
-
- foreach my $disk (sort keys %{$disks}) {
- $source->{all} = $disks->{$disk}->{all};
- $source->{pool} = $disks->{$disk}->{pool};
- $source->{path} = $disks->{$disk}->{path} if $disks->{$disk}->{path};
- $source->{last_part} = $disks->{$disk}->{last_part};
+ eval{
+ if ($source->{vmid}) {
+ die "VM $source->{vmid} doesn't exist\n" if !$vm_type;
+ die "source-user has to be root for syncing VMs\n" if ($param->{source_user} ne "root");
+ my $disks = get_disks($source, $param->{source_user});
+
+ foreach my $disk (sort keys %{$disks}) {
+ $source->{all} = $disks->{$disk}->{all};
+ $source->{pool} = $disks->{$disk}->{pool};
+ $source->{path} = $disks->{$disk}->{path} if $disks->{$disk}->{path};
+ $source->{last_part} = $disks->{$disk}->{last_part};
+ &$sync_path($source, $dest, $job, $param, $date);
+ }
+ if ($param->{method} eq "ssh" && ($source->{ip} || $dest->{ip})) {
+ send_config($source, $dest,'ssh', $param->{source_user}, $param->{dest_user}, $param->{dest_config_path});
+ } else {
+ send_config($source, $dest,'local', $param->{source_user}, $param->{dest_user}, $param->{dest_config_path});
+ }
+ } else {
&$sync_path($source, $dest, $job, $param, $date);
}
- if ($param->{method} eq "ssh" && ($source->{ip} || $dest->{ip})) {
- send_config($source, $dest,'ssh', $param->{source_user}, $param->{dest_user}, $param->{dest_config_path});
- } else {
- send_config($source, $dest,'local', $param->{source_user}, $param->{dest_user}, $param->{dest_config_path});
+ };
+ if (my $err = $@) {
+ if ($job) {
+ $job->{state} = "error";
+ locked("$CONFIG_PATH/cron_and_state.lock", sub { update_state($job); });
+ print "Job --source $param->{source} --name $param->{name} got an ERROR!!!\nERROR Message:\n";
}
- } else {
- &$sync_path($source, $dest, $job, $param, $date);
+ die "$err\n";
}
- };
- if(my $err = $@) {
+
if ($job) {
- $job->{state} = "error";
- update_state($job);
- unlock($lock_fh);
- close($lock_fh);
- print "Job --source $param->{source} --name $param->{name} got an ERROR!!!\nERROR Message:\n";
+ $job->{state} = "ok";
+ $job->{lsync} = $date;
+ locked("$CONFIG_PATH/cron_and_state.lock", sub { update_state($job); });
}
- die "$err\n";
- }
-
- if ($job) {
- $job->{state} = "ok";
- $job->{lsync} = $date;
- update_state($job);
- }
-
- unlock($lock_fh);
- close($lock_fh);
+ }); #sync lock
}
sub snapshot_get{
@@ -1031,19 +1026,23 @@ sub status {
sub enable_job {
my ($param) = @_;
- my $job = get_job($param);
- $job->{state} = "ok";
- update_state($job);
- update_cron($job);
+ locked("$CONFIG_PATH/cron_and_state.lock", sub {
+ my $job = get_job($param);
+ $job->{state} = "ok";
+ update_state($job);
+ update_cron($job);
+ });
}
sub disable_job {
my ($param) = @_;
- my $job = get_job($param);
- $job->{state} = "stopped";
- update_state($job);
- update_cron($job);
+ locked("$CONFIG_PATH/cron_and_state.lock", sub {
+ my $job = get_job($param);
+ $job->{state} = "stopped";
+ update_state($job);
+ update_cron($job);
+ });
}
my $cmd_help = {
--
2.20.1
More information about the pve-devel
mailing list