syncoid

#!/usr/bin/perl

# this software is licensed for use under the Free Software Foundation's GPL v3.0 license, as retrieved
# from http://www.gnu.org/licenses/gpl-3.0.html on 2014-11-17.  A copy should also be available in this
# project's Git repository at https://github.com/jimsalterjrs/sanoid/blob/master/LICENSE.

$::VERSION = '2.1.0';

use strict;
use warnings;
use Data::Dumper;
use Getopt::Long qw(:config auto_version auto_help);
use Pod::Usage;
use Time::Local;
use Sys::Hostname;
use Capture::Tiny ':all';

my $mbuffer_size = "16M";
my $pvoptions = "-p -t -e -r -b";

# Blank defaults to use ssh client's default
# TODO: Merge into a single "sshflags" option?
my %args = ('sshkey' => '', 'sshport' => '', 'sshcipher' => '', 'sshoption' => [], 'target-bwlimit' => '', 'source-bwlimit' => '');
GetOptions(\%args, "no-command-checks", "monitor-version", "compress=s", "dumpsnaps", "recursive|r", "sendoptions=s", "recvoptions=s",
                   "source-bwlimit=s", "target-bwlimit=s", "sshkey=s", "sshport=i", "sshcipher|c=s", "sshoption|o=s@",
                   "debug", "quiet", "no-stream", "no-sync-snap", "no-resume", "exclude=s@", "skip-parent", "identifier=s",
                   "no-clone-handling", "no-privilege-elevation", "force-delete", "no-clone-rollback", "no-rollback",
                   "create-bookmark", "pv-options=s" => \$pvoptions, "keep-sync-snap", "preserve-recordsize",
                   "mbuffer-size=s" => \$mbuffer_size) or pod2usage(2);

my %compressargs = %{compressargset($args{'compress'} || 'default')}; # Can't be done with GetOptions arg, as default still needs to be set

my @sendoptions = ();
if (length $args{'sendoptions'}) {
	@sendoptions = parsespecialoptions($args{'sendoptions'});
	if (! defined($sendoptions[0])) {
		warn "invalid send options!";
		pod2usage(2);
		exit 127;
	}

	if (defined $args{'recursive'}) {
		foreach my $option(@sendoptions) {
			if ($option->{option} eq 'R') {
				warn "invalid argument combination, zfs send -R and --recursive aren't compatible!";
				pod2usage(2);
				exit 127;
			}
		}
	}
}

my @recvoptions = ();
if (length $args{'recvoptions'}) {
	@recvoptions = parsespecialoptions($args{'recvoptions'});
	if (! defined($recvoptions[0])) {
		warn "invalid receive options!";
		pod2usage(2);
		exit 127;
	}
}


# TODO Expand to accept multiple sources?
if (scalar(@ARGV) != 2) {
	print("Source or target not found!\n");
	pod2usage(2);
	exit 127;
} else {
	$args{'source'} = $ARGV[0];
	$args{'target'} = $ARGV[1];
}

# Could possibly merge these into an options function
if (length $args{'source-bwlimit'}) {
	$args{'source-bwlimit'} = "-R $args{'source-bwlimit'}";
}
if (length $args{'target-bwlimit'}) {
	$args{'target-bwlimit'} = "-r $args{'target-bwlimit'}";
}
$args{'streamarg'} = (defined $args{'no-stream'} ? '-i' : '-I');

my $rawsourcefs = $args{'source'};
my $rawtargetfs = $args{'target'};
my $debug = $args{'debug'};
my $quiet = $args{'quiet'};
my $resume = !$args{'no-resume'};

# for compatibility reasons, older versions used hardcoded command paths
$ENV{'PATH'} = $ENV{'PATH'} . ":/bin:/usr/bin:/sbin";

my $zfscmd = 'zfs';
my $zpoolcmd = 'zpool';
my $sshcmd = 'ssh';
my $pscmd = 'ps';

my $pvcmd = 'pv';
my $mbuffercmd = 'mbuffer';
my $sudocmd = 'sudo';
my $mbufferoptions = "-q -s 128k -m $mbuffer_size 2>/dev/null";
# currently using POSIX compatible command to check for program existence because we aren't depending on perl
# being present on remote machines.
my $checkcmd = 'command -v';

if (length $args{'sshcipher'}) {
	$args{'sshcipher'} = "-c $args{'sshcipher'}";
}
if (length $args{'sshport'}) {
  $args{'sshport'} = "-p $args{'sshport'}";
}
if (length $args{'sshkey'}) {
	$args{'sshkey'} = "-i $args{'sshkey'}";
}
my $sshoptions = join " ", map { "-o " . $_ } @{$args{'sshoption'}}; # deref required

my $identifier = "";
if (length $args{'identifier'}) {
	if ($args{'identifier'} !~ /^[a-zA-Z0-9-_:.]+$/) {
		# invalid extra identifier
		print("CRITICAL: extra identifier contains invalid chars!\n");
		pod2usage(2);
		exit 127;
	}
	$identifier = "$args{'identifier'}_";
}

# figure out if source and/or target are remote.
$sshcmd = "$sshcmd $args{'sshcipher'} $sshoptions $args{'sshport'} $args{'sshkey'}";
if ($debug) { print "DEBUG: SSHCMD: $sshcmd\n"; }
my ($sourcehost,$sourcefs,$sourceisroot) = getssh($rawsourcefs);
my ($targethost,$targetfs,$targetisroot) = getssh($rawtargetfs);

my $sourcesudocmd = $sourceisroot ? '' : $sudocmd;
my $targetsudocmd = $targetisroot ? '' : $sudocmd;

# figure out whether compression, mbuffering, pv
# are available on source, target, local machines.
# warn user of anything missing, then continue with sync.
my %avail = checkcommands();

my %snaps;
my $exitcode = 0;

## break here to call replication individually so that we ##
## can loop across children separately, for recursive     ##
## replication                                            ##

if (!defined $args{'recursive'}) {
	syncdataset($sourcehost, $sourcefs, $targethost, $targetfs, undef);
} else {
	if ($debug) { print "DEBUG: recursive sync of $sourcefs.\n"; }
	my @datasets = getchilddatasets($sourcehost, $sourcefs, $sourceisroot);

	if (!@datasets) {
		warn "CRITICAL ERROR: no datasets found";
		@datasets = ();
		$exitcode = 2;
	}

	my @deferred;

	foreach my $datasetProperties(@datasets) {
		my $dataset = $datasetProperties->{'name'};
		my $origin = $datasetProperties->{'origin'};
		if ($origin eq "-" || defined $args{'no-clone-handling'}) {
			$origin = undef;
		} else {
			# check if clone source is replicated too
			my @values = split(/@/, $origin, 2);
			my $srcdataset = $values[0];

			my $found = 0;
			foreach my $datasetProperties(@datasets) {
				if ($datasetProperties->{'name'} eq $srcdataset) {
					$found = 1;
					last;
				}
			}

			if ($found == 0) {
				# clone source is not replicated, do a full replication
				$origin = undef;
			} else {
				# clone source is replicated, defer until all non clones are replicated
				push @deferred, $datasetProperties;
				next;
			}
		}

		$dataset =~ s/\Q$sourcefs\E//;
		chomp $dataset;
		my $childsourcefs = $sourcefs . $dataset;
		my $childtargetfs = $targetfs . $dataset;
		# print "syncdataset($sourcehost, $childsourcefs, $targethost, $childtargetfs); \n";
		syncdataset($sourcehost, $childsourcefs, $targethost, $childtargetfs, $origin);
	}

	# replicate cloned datasets and if this is the initial run, recreate them on the target
	foreach my $datasetProperties(@deferred) {
		my $dataset = $datasetProperties->{'name'};
		my $origin = $datasetProperties->{'origin'};

		$dataset =~ s/\Q$sourcefs\E//;
		chomp $dataset;
		my $childsourcefs = $sourcefs . $dataset;
		my $childtargetfs = $targetfs . $dataset;
		syncdataset($sourcehost, $childsourcefs, $targethost, $childtargetfs, $origin);
	}
}

# close SSH sockets for master connections as applicable
if ($sourcehost ne '') {
	open FH, "$sshcmd $sourcehost -O exit 2>&1 |";
	close FH;
}
if ($targethost ne '') {
	open FH, "$sshcmd $targethost -O exit 2>&1 |";
	close FH;
}

exit $exitcode;

##############################################################################
##############################################################################
##############################################################################
##############################################################################

sub getchilddatasets {
	my ($rhost,$fs,$isroot,%snaps) = @_;
	my $mysudocmd;
	my $fsescaped = escapeshellparam($fs);

	if ($isroot) { $mysudocmd = ''; } else { $mysudocmd = $sudocmd; }
	if ($rhost ne '') {
		$rhost = "$sshcmd $rhost";
		# double escaping needed
		$fsescaped = escapeshellparam($fsescaped);
	}

	my $getchildrencmd = "$rhost $mysudocmd $zfscmd list -o name,origin -t filesystem,volume -Hr $fsescaped |";
	if ($debug) { print "DEBUG: getting list of child datasets on $fs using $getchildrencmd...\n"; }
	if (! open FH, $getchildrencmd) {
		die "ERROR: list command failed!\n";
	}

	my @children;
	my $first = 1;

	DATASETS: while(<FH>) {
		chomp;

		if (defined $args{'skip-parent'} && $first eq 1) {
			# parent dataset is the first element
			$first = 0;
			next;
		}

		my ($dataset, $origin) = /^([^\t]+)\t([^\t]+)/;

		if (defined $args{'exclude'}) {
			my $excludes = $args{'exclude'};
			foreach (@$excludes) {
				if ($dataset =~ /$_/) {
					if ($debug) { print "DEBUG: excluded $dataset because of $_\n"; }
					next DATASETS;
				}
			}
		}

		my %properties;
		$properties{'name'} = $dataset;
		$properties{'origin'} = $origin;

		push @children, \%properties;
	}
	close FH;

	return @children;
}

sub syncdataset {

	my ($sourcehost, $sourcefs, $targethost, $targetfs, $origin, $skipsnapshot) = @_;

	my $stdout;
	my $exit;

	my $sourcefsescaped = escapeshellparam($sourcefs);
	my $targetfsescaped = escapeshellparam($targetfs);

	# if no rollbacks are allowed, disable forced receive
	my $forcedrecv = "-F";
	if (defined $args{'no-rollback'}) {
		$forcedrecv = "";
	}

	if ($debug) { print "DEBUG: syncing source $sourcefs to target $targetfs.\n"; }

	my ($sync, $error) = getzfsvalue($sourcehost,$sourcefs,$sourceisroot,'syncoid:sync');

	if (!defined $sync) {
		# zfs already printed the corresponding error
		if ($error =~ /\bdataset does not exist\b/) {
			if (!$quiet) { print "WARN Skipping dataset (dataset no longer exists): $sourcefs...\n"; }
			return 0;
		}
		else {
			# print the error out and set exit code
			print "ERROR: $error\n";
			if ($exitcode < 2) { $exitcode = 2 }
		}

		return 0;
	}

	if ($sync eq 'true' || $sync eq '-' || $sync eq '') {
		# empty is handled the same as unset (aka: '-')
		# definitely sync this dataset - if a host is called 'true' or '-', then you're special
	} elsif ($sync eq 'false') {
		if (!$quiet) { print "INFO: Skipping dataset (syncoid:sync=false): $sourcefs...\n"; }
		return 0;
	} else {
		my $hostid = hostname();
		my @hosts = split(/,/,$sync);
		if (!(grep $hostid eq $_, @hosts)) {
			if (!$quiet) { print "INFO: Skipping dataset (syncoid:sync doesn't include $hostid): $sourcefs...\n"; }
			return 0;
		}
	}

	# make sure target is not currently in receive.
	if (iszfsbusy($targethost,$targetfs,$targetisroot)) {
		warn "Cannot sync now: $targetfs is already target of a zfs receive process.\n";
		if ($exitcode < 1) { $exitcode = 1; }
		return 0;
	}

	# does the target filesystem exist yet?
	my $targetexists = targetexists($targethost,$targetfs,$targetisroot);

	my $receiveextraargs = "";
	my $receivetoken;
	if ($resume) {
		# save state of interrupted receive stream
		$receiveextraargs = "-s";

		if ($targetexists) {
			# check remote dataset for receive resume token (interrupted receive)
			$receivetoken = getreceivetoken($targethost,$targetfs,$targetisroot);

			if ($debug && defined($receivetoken)) {
				print "DEBUG: got receive resume token: $receivetoken: \n";
			}
		}
	}

	my $newsyncsnap;

	# skip snapshot checking/creation in case of resumed receive
	if (!defined($receivetoken)) {
		# build hashes of the snaps on the source and target filesystems.

		%snaps = getsnaps('source',$sourcehost,$sourcefs,$sourceisroot);

		if ($targetexists) {
		    my %targetsnaps = getsnaps('target',$targethost,$targetfs,$targetisroot);
		    my %sourcesnaps = %snaps;
		    %snaps = (%sourcesnaps, %targetsnaps);
		}

		if (defined $args{'dumpsnaps'}) {
		    print "merged snapshot list of $targetfs: \n";
		    dumphash(\%snaps);
		    print "\n\n\n";
		}

		if (!defined $args{'no-sync-snap'} && !defined $skipsnapshot) {
			# create a new syncoid snapshot on the source filesystem.
			$newsyncsnap = newsyncsnap($sourcehost,$sourcefs,$sourceisroot);
			if (!$newsyncsnap) {
				# we already whined about the error
				return 0;
			}
		} else {
			# we don't want sync snapshots created, so use the newest snapshot we can find.
			$newsyncsnap = getnewestsnapshot($sourcehost,$sourcefs,$sourceisroot);
			if ($newsyncsnap eq 0) {
				warn "CRITICAL: no snapshots exist on source $sourcefs, and you asked for --no-sync-snap.\n";
				if ($exitcode < 1) { $exitcode = 1; }
				return 0;
			}
		}
	}
	my $newsyncsnapescaped = escapeshellparam($newsyncsnap);

	# there is currently (2014-09-01) a bug in ZFS on Linux
	# that causes readonly to always show on if it's EVER
	# been turned on... even when it's off... unless and
	# until the filesystem is zfs umounted and zfs remounted.
	# we're going to do the right thing anyway.
		# dyking this functionality out for the time being due to buggy mount/unmount behavior
		# with ZFS on Linux (possibly OpenZFS in general) when setting/unsetting readonly.
	#my $originaltargetreadonly;

	my $sendoptions = getoptionsline(\@sendoptions, ('D','L','P','R','c','e','h','p','v','w'));
	my $recvoptions = getoptionsline(\@recvoptions, ('h','o','x','u','v'));

	# sync 'em up.
	if (! $targetexists) {
		# do an initial sync from the oldest source snapshot
		# THEN do an -I to the newest
		if ($debug) {
			if (!defined ($args{'no-stream'}) ) {
				print "DEBUG: target $targetfs does not exist.  Finding oldest available snapshot on source $sourcefs ...\n";
			} else {
				print "DEBUG: target $targetfs does not exist, and --no-stream selected.  Finding newest available snapshot on source $sourcefs ...\n";
			}
		}
		my $oldestsnap = getoldestsnapshot(\%snaps);
		if (! $oldestsnap) {
			if (defined ($args{'no-sync-snap'}) ) {
				# we already whined about the missing snapshots
				return 0;
			}

			# getoldestsnapshot() returned false, so use new sync snapshot
			if ($debug) { print "DEBUG: getoldestsnapshot() returned false, so using $newsyncsnap.\n"; }
			$oldestsnap = $newsyncsnap;
		}

		# if --no-stream is specified, our full needs to be the newest snapshot, not the oldest.
		if (defined $args{'no-stream'}) {
			if (defined ($args{'no-sync-snap'}) ) {
				$oldestsnap = getnewestsnapshot(\%snaps);
			} else {
				$oldestsnap = $newsyncsnap;
			}
		}
		my $oldestsnapescaped = escapeshellparam($oldestsnap);

		if (defined $args{'preserve-recordsize'}) {
			my $type = getzfsvalue($sourcehost,$sourcefs,$sourceisroot,'type');
			if ($type eq "filesystem") {
				my $recordsize = getzfsvalue($sourcehost,$sourcefs,$sourceisroot,'recordsize');
				$recvoptions .= "-o recordsize=$recordsize"
			}
		}

		my $sendcmd = "$sourcesudocmd $zfscmd send $sendoptions $sourcefsescaped\@$oldestsnapescaped";
		my $recvcmd = "$targetsudocmd $zfscmd receive $recvoptions $receiveextraargs $forcedrecv $targetfsescaped";

		my $pvsize;
		if (defined $origin) {
			my $originescaped = escapeshellparam($origin);
			$sendcmd = "$sourcesudocmd $zfscmd send $sendoptions -i $originescaped $sourcefsescaped\@$oldestsnapescaped";
			my $streamargBackup = $args{'streamarg'};
			$args{'streamarg'} = "-i";
			$pvsize = getsendsize($sourcehost,$origin,"$sourcefs\@$oldestsnap",$sourceisroot);
			$args{'streamarg'} = $streamargBackup;
		} else {
			$pvsize = getsendsize($sourcehost,"$sourcefs\@$oldestsnap",0,$sourceisroot);
		}

		my $disp_pvsize = readablebytes($pvsize);
		if ($pvsize == 0) { $disp_pvsize = 'UNKNOWN'; }
		my $synccmd = buildsynccmd($sendcmd,$recvcmd,$pvsize,$sourceisroot,$targetisroot);
		if (!$quiet) {
			if (defined $origin) {
				print "INFO: Clone is recreated on target $targetfs based on $origin\n";
			}
			if (!defined ($args{'no-stream'}) ) {
				print "INFO: Sending oldest full snapshot $sourcefs\@$oldestsnap (~ $disp_pvsize) to new target filesystem:\n";
			} else {
				print "INFO: --no-stream selected; sending newest full snapshot $sourcefs\@$oldestsnap (~ $disp_pvsize) to new target filesystem:\n";
			}
		}
		if ($debug) { print "DEBUG: $synccmd\n"; }

		# make sure target is (still) not currently in receive.
		if (iszfsbusy($targethost,$targetfs,$targetisroot)) {
			warn "Cannot sync now: $targetfs is already target of a zfs receive process.\n";
			if ($exitcode < 1) { $exitcode = 1; }
			return 0;
		}
		system($synccmd) == 0 or do {
			if (defined $origin) {
				print "INFO: clone creation failed, trying ordinary replication as fallback\n";
				syncdataset($sourcehost, $sourcefs, $targethost, $targetfs, undef, 1);
				return 0;
			}

			warn "CRITICAL ERROR: $synccmd failed: $?";
			if ($exitcode < 2) { $exitcode = 2; }
			return 0;
		};

		# now do an -I to the new sync snapshot, assuming there were any snapshots
		# other than the new sync snapshot to begin with, of course - and that we
		# aren't invoked with --no-stream, in which case a full of the newest snap
		# available was all we needed to do
		if (!defined ($args{'no-stream'}) && ($oldestsnap ne $newsyncsnap) ) {

			# get current readonly status of target, then set it to on during sync
				# dyking this functionality out for the time being due to buggy mount/unmount behavior
				# with ZFS on Linux (possibly OpenZFS in general) when setting/unsetting readonly.
			# $originaltargetreadonly = getzfsvalue($targethost,$targetfs,$targetisroot,'readonly');
			# setzfsvalue($targethost,$targetfs,$targetisroot,'readonly','on');

			$sendcmd = "$sourcesudocmd $zfscmd send $sendoptions $args{'streamarg'} $sourcefsescaped\@$oldestsnapescaped $sourcefsescaped\@$newsyncsnapescaped";
			$pvsize = getsendsize($sourcehost,"$sourcefs\@$oldestsnap","$sourcefs\@$newsyncsnap",$sourceisroot);
			$disp_pvsize = readablebytes($pvsize);
			if ($pvsize == 0) { $disp_pvsize = "UNKNOWN"; }
			$synccmd = buildsynccmd($sendcmd,$recvcmd,$pvsize,$sourceisroot,$targetisroot);

			# make sure target is (still) not currently in receive.
			if (iszfsbusy($targethost,$targetfs,$targetisroot)) {
				warn "Cannot sync now: $targetfs is already target of a zfs receive process.\n";
				if ($exitcode < 1) { $exitcode = 1; }
				return 0;
			}

			if (!$quiet) { print "INFO: Updating new target filesystem with incremental $sourcefs\@$oldestsnap ... $newsyncsnap (~ $disp_pvsize):\n"; }
			if ($debug) { print "DEBUG: $synccmd\n"; }

			if ($oldestsnap ne $newsyncsnap) {
				my $ret = system($synccmd);
				if ($ret != 0) {
					warn "CRITICAL ERROR: $synccmd failed: $?";
					if ($exitcode < 1) { $exitcode = 1; }
					return 0;
				}
			} else {
				if (!$quiet) { print "INFO: no incremental sync needed; $oldestsnap is already the newest available snapshot.\n"; }
			}

			# restore original readonly value to target after sync complete
				# dyking this functionality out for the time being due to buggy mount/unmount behavior
				# with ZFS on Linux (possibly OpenZFS in general) when setting/unsetting readonly.
			# setzfsvalue($targethost,$targetfs,$targetisroot,'readonly',$originaltargetreadonly);
		}
	} else {
		# resume interrupted receive if there is a valid resume $token
		# and because this will ony resume the receive to the next
		# snapshot, do a normal sync after that
		if (defined($receivetoken)) {
			$sendoptions = getoptionsline(\@sendoptions, ('P','e','v','w'));
			my $sendcmd = "$sourcesudocmd $zfscmd send $sendoptions -t $receivetoken";
			my $recvcmd = "$targetsudocmd $zfscmd receive $recvoptions $receiveextraargs $forcedrecv $targetfsescaped 2>&1";
			my $pvsize = getsendsize($sourcehost,"","",$sourceisroot,$receivetoken);
			my $disp_pvsize = readablebytes($pvsize);
			if ($pvsize == 0) { $disp_pvsize = "UNKNOWN"; }
			my $synccmd = buildsynccmd($sendcmd,$recvcmd,$pvsize,$sourceisroot,$targetisroot);

			if (!$quiet) { print "Resuming interrupted zfs send/receive from $sourcefs to $targetfs (~ $disp_pvsize remaining):\n"; }
			if ($debug) { print "DEBUG: $synccmd\n"; }

			if ($pvsize == 0) {
				# we need to capture the error of zfs send, this will render pv useless but in this case
				# it doesn't matter because we don't know the estimated send size (probably because
				# the initial snapshot used for resumed send doesn't exist anymore)
				($stdout, $exit) = tee_stderr {
					system("$synccmd")
				};
			} else {
				($stdout, $exit) = tee_stdout {
					system("$synccmd")
				};
			}

			$exit == 0 or do {
				if (
					$stdout =~ /\Qused in the initial send no longer exists\E/ ||
					$stdout =~ /incremental source [0-9xa-f]+ no longer exists/
				) {
					if (!$quiet) { print "WARN: resetting partially receive state because the snapshot source no longer exists\n"; }
					resetreceivestate($targethost,$targetfs,$targetisroot);
					# do an normal sync cycle
					return syncdataset($sourcehost, $sourcefs, $targethost, $targetfs, $origin);
				} else {
					warn "CRITICAL ERROR: $synccmd failed: $?";
					if ($exitcode < 2) { $exitcode = 2; }
					return 0;
				}
			};

			# a resumed transfer will only be done to the next snapshot,
			# so do an normal sync cycle
			return syncdataset($sourcehost, $sourcefs, $targethost, $targetfs, undef);
		}

		# find most recent matching snapshot and do an -I
		# to the new snapshot

		# get current readonly status of target, then set it to on during sync
			# dyking this functionality out for the time being due to buggy mount/unmount behavior
			# with ZFS on Linux (possibly OpenZFS in general) when setting/unsetting readonly.
		# $originaltargetreadonly = getzfsvalue($targethost,$targetfs,$targetisroot,'readonly');
		# setzfsvalue($targethost,$targetfs,$targetisroot,'readonly','on');

		my $targetsize = getzfsvalue($targethost,$targetfs,$targetisroot,'-p used');

		my $bookmark = 0;
		my $bookmarkcreation = 0;

		my $matchingsnap = getmatchingsnapshot($sourcefs, $targetfs, \%snaps);
		if (! $matchingsnap) {
			# no matching snapshots, check for bookmarks as fallback
			my %bookmarks = getbookmarks($sourcehost,$sourcefs,$sourceisroot);

			# check for matching guid of source bookmark and target snapshot (oldest first)
			foreach my $snap ( sort { $snaps{'target'}{$b}{'createtxg'}<=>$snaps{'target'}{$a}{'createtxg'} } keys %{ $snaps{'target'} }) {
				my $guid = $snaps{'target'}{$snap}{'guid'};

				if (defined $bookmarks{$guid}) {
					# found a match
					$bookmark = $bookmarks{$guid}{'name'};
					$bookmarkcreation = $bookmarks{$guid}{'creation'};
					$matchingsnap = $snap;
					last;
				}
			}

			if (! $bookmark) {
				if ($args{'force-delete'}) {
					if (!$quiet) { print "Removing $targetfs because no matching snapshots were found\n"; }

					my $rcommand = '';
					my $mysudocmd = '';
					my $targetfsescaped = escapeshellparam($targetfs);

					if ($targethost ne '') { $rcommand = "$sshcmd $targethost"; }
					if (!$targetisroot) { $mysudocmd = $sudocmd; }

					my $prunecmd = "$mysudocmd $zfscmd destroy -r $targetfsescaped; ";
					if ($targethost ne '') {
						$prunecmd = escapeshellparam($prunecmd);
					}

					my $ret = system("$rcommand $prunecmd");
					if ($ret != 0) {
						warn "WARNING: $rcommand $prunecmd failed: $?";
					} else {
						# redo sync and skip snapshot creation (already taken)
						return syncdataset($sourcehost, $sourcefs, $targethost, $targetfs, undef, 1);
					}
				}

				# if we got this far, we failed to find a matching snapshot/bookmark.
				if ($exitcode < 2) { $exitcode = 2; }

				print "\n";
				print "CRITICAL ERROR: Target $targetfs exists but has no snapshots matching with $sourcefs!\n";
				print "                Replication to target would require destroying existing\n";
				print "                target. Cowardly refusing to destroy your existing target.\n\n";

				# experience tells me we need a mollyguard for people who try to
				# zfs create targetpool/targetsnap ; syncoid sourcepool/sourcesnap targetpool/targetsnap ...

				if ( $targetsize < (64*1024*1024) ) {
					print "          NOTE: Target $targetfs dataset is < 64MB used - did you mistakenly run\n";
					print "                \`zfs create $args{'target'}\` on the target? ZFS initial\n";
					print "                replication must be to a NON EXISTENT DATASET, which will\n";
					print "                then be CREATED BY the initial replication process.\n\n";
				}

				# return false now in case more child datasets need replication.
				return 0;
			}
		}

		# make sure target is (still) not currently in receive.
		if (iszfsbusy($targethost,$targetfs,$targetisroot)) {
			warn "Cannot sync now: $targetfs is already target of a zfs receive process.\n";
			if ($exitcode < 1) { $exitcode = 1; }
			return 0;
		}

		if ($matchingsnap eq $newsyncsnap) {
			# barf some text but don't touch the filesystem
			if (!$quiet) { print "INFO: no snapshots on source newer than $newsyncsnap on target. Nothing to do, not syncing.\n"; }
			return 0;
		} else {
			my $matchingsnapescaped = escapeshellparam($matchingsnap);
			# rollback target to matchingsnap
			if (!defined $args{'no-rollback'}) {
				my $rollbacktype = "-R";
				if (defined $args{'no-clone-rollback'}) {
					$rollbacktype = "-r";
				}
				if ($debug) { print "DEBUG: rolling back target to $targetfs\@$matchingsnap...\n"; }
				if ($targethost ne '') {
					if ($debug) { print "$sshcmd $targethost $targetsudocmd $zfscmd rollback $rollbacktype $targetfsescaped\@$matchingsnapescaped\n"; }
					system ("$sshcmd $targethost " . escapeshellparam("$targetsudocmd $zfscmd rollback $rollbacktype $targetfsescaped\@$matchingsnapescaped"));
				} else {
					if ($debug) { print "$targetsudocmd $zfscmd rollback $rollbacktype $targetfsescaped\@$matchingsnapescaped\n"; }
					system ("$targetsudocmd $zfscmd rollback $rollbacktype $targetfsescaped\@$matchingsnapescaped");
				}
			}

			my $nextsnapshot = 0;

			if ($bookmark) {
				my $bookmarkescaped = escapeshellparam($bookmark);

				if (!defined $args{'no-stream'}) {
					# if intermediate snapshots are needed we need to find the next oldest snapshot,
					# do an replication to it and replicate as always from oldest to newest
					# because bookmark sends doesn't support intermediates directly
					foreach my $snap ( sort { $snaps{'source'}{$a}{'createtxg'}<=>$snaps{'source'}{$b}{'createtxg'} } keys %{ $snaps{'source'} }) {
						if ($snaps{'source'}{$snap}{'createtxg'} >= $bookmarkcreation) {
							$nextsnapshot = $snap;
							last;
						}
					}
				}

				# bookmark stream size can't be determined
				my $pvsize = 0;
				my $disp_pvsize = "UNKNOWN";

				$sendoptions = getoptionsline(\@sendoptions, ('L','c','e','w'));
				if ($nextsnapshot) {
					my $nextsnapshotescaped = escapeshellparam($nextsnapshot);
					my $sendcmd = "$sourcesudocmd $zfscmd send $sendoptions -i $sourcefsescaped#$bookmarkescaped $sourcefsescaped\@$nextsnapshotescaped";
					my $recvcmd = "$targetsudocmd $zfscmd receive $recvoptions $receiveextraargs $forcedrecv $targetfsescaped 2>&1";
					my $synccmd = buildsynccmd($sendcmd,$recvcmd,$pvsize,$sourceisroot,$targetisroot);

					if (!$quiet) { print "Sending incremental $sourcefs#$bookmarkescaped ... $nextsnapshot (~ $disp_pvsize):\n"; }
					if ($debug) { print "DEBUG: $synccmd\n"; }

					($stdout, $exit) = tee_stdout {
						system("$synccmd")
					};

					$exit == 0 or do {
						if (!$resume && $stdout =~ /\Qcontains partially-complete state\E/) {
							if (!$quiet) { print "WARN: resetting partially receive state\n"; }
							resetreceivestate($targethost,$targetfs,$targetisroot);
							system("$synccmd") == 0 or do {
								warn "CRITICAL ERROR: $synccmd failed: $?";
								if ($exitcode < 2) { $exitcode = 2; }
								return 0;
							}
						} else {
							warn "CRITICAL ERROR: $synccmd failed: $?";
							if ($exitcode < 2) { $exitcode = 2; }
							return 0;
						}
					};

					$matchingsnap = $nextsnapshot;
					$matchingsnapescaped = escapeshellparam($matchingsnap);
				} else {
					my $sendcmd = "$sourcesudocmd $zfscmd send $sendoptions -i $sourcefsescaped#$bookmarkescaped $sourcefsescaped\@$newsyncsnapescaped";
					my $recvcmd = "$targetsudocmd $zfscmd receive $recvoptions $receiveextraargs $forcedrecv $targetfsescaped 2>&1";
					my $synccmd = buildsynccmd($sendcmd,$recvcmd,$pvsize,$sourceisroot,$targetisroot);

					if (!$quiet) { print "Sending incremental $sourcefs#$bookmarkescaped ... $newsyncsnap (~ $disp_pvsize):\n"; }
					if ($debug) { print "DEBUG: $synccmd\n"; }

					($stdout, $exit) = tee_stdout {
						system("$synccmd")
					};

					$exit == 0 or do {
						if (!$resume && $stdout =~ /\Qcontains partially-complete state\E/) {
							if (!$quiet) { print "WARN: resetting partially receive state\n"; }
							resetreceivestate($targethost,$targetfs,$targetisroot);
							system("$synccmd") == 0 or do {
								warn "CRITICAL ERROR: $synccmd failed: $?";
								if ($exitcode < 2) { $exitcode = 2; }
								return 0;
							}
						} else {
							warn "CRITICAL ERROR: $synccmd failed: $?";
							if ($exitcode < 2) { $exitcode = 2; }
							return 0;
						}
					};
				}
			}

			# do a normal replication if bookmarks aren't used or if previous
			# bookmark replication was only done to the next oldest snapshot
			if (!$bookmark || $nextsnapshot) {
				if ($matchingsnap eq $newsyncsnap) {
					# edge case: bookmark replication used the latest snapshot
					return 0;
				}

				$sendoptions = getoptionsline(\@sendoptions, ('D','L','P','R','c','e','h','p','v','w'));
				my $sendcmd = "$sourcesudocmd $zfscmd send $sendoptions $args{'streamarg'} $sourcefsescaped\@$matchingsnapescaped $sourcefsescaped\@$newsyncsnapescaped";
				my $recvcmd = "$targetsudocmd $zfscmd receive $recvoptions $receiveextraargs $forcedrecv $targetfsescaped 2>&1";
				my $pvsize = getsendsize($sourcehost,"$sourcefs\@$matchingsnap","$sourcefs\@$newsyncsnap",$sourceisroot);
				my $disp_pvsize = readablebytes($pvsize);
				if ($pvsize == 0) { $disp_pvsize = "UNKNOWN"; }
				my $synccmd = buildsynccmd($sendcmd,$recvcmd,$pvsize,$sourceisroot,$targetisroot);

				if (!$quiet) { print "Sending incremental $sourcefs\@$matchingsnap ... $newsyncsnap (~ $disp_pvsize):\n"; }
				if ($debug) { print "DEBUG: $synccmd\n"; }

				($stdout, $exit) = tee_stdout {
					system("$synccmd")
				};

				$exit == 0 or do {
					# FreeBSD reports "dataset is busy" instead of "contains partially-complete state"
					if (!$resume && ($stdout =~ /\Qcontains partially-complete state\E/ || $stdout =~ /\Qdataset is busy\E/)) {
						if (!$quiet) { print "WARN: resetting partially receive state\n"; }
						resetreceivestate($targethost,$targetfs,$targetisroot);
						system("$synccmd") == 0 or do {
							warn "CRITICAL ERROR: $synccmd failed: $?";
							if ($exitcode < 2) { $exitcode = 2; }
							return 0;
						}
					} else {
						warn "CRITICAL ERROR: $synccmd failed: $?";
						if ($exitcode < 2) { $exitcode = 2; }
						return 0;
					}
				};
			}

			# restore original readonly value to target after sync complete
				# dyking this functionality out for the time being due to buggy mount/unmount behavior
				# with ZFS on Linux (possibly OpenZFS in general) when setting/unsetting readonly.
			#setzfsvalue($targethost,$targetfs,$targetisroot,'readonly',$originaltargetreadonly);
		}
	}

	if (defined $args{'no-sync-snap'}) {
		if (defined $args{'create-bookmark'}) {
			my $bookmarkcmd;
			if ($sourcehost ne '') {
				$bookmarkcmd = "$sshcmd $sourcehost " . escapeshellparam("$sourcesudocmd $zfscmd bookmark $sourcefsescaped\@$newsyncsnapescaped $sourcefsescaped\#$newsyncsnapescaped");
			} else {
				$bookmarkcmd = "$sourcesudocmd $zfscmd bookmark $sourcefsescaped\@$newsyncsnapescaped $sourcefsescaped\#$newsyncsnapescaped";
			}
			if ($debug) { print "DEBUG: $bookmarkcmd\n"; }
			system($bookmarkcmd) == 0 or do {
				# fallback: assume nameing conflict and try again with guid based suffix
				my $guid = $snaps{'source'}{$newsyncsnap}{'guid'};
				$guid = substr($guid, 0, 6);

				if (!$quiet) { print "INFO: bookmark creation failed, retrying with guid based suffix ($guid)...\n"; }

				if ($sourcehost ne '') {
					$bookmarkcmd = "$sshcmd $sourcehost " . escapeshellparam("$sourcesudocmd $zfscmd bookmark $sourcefsescaped\@$newsyncsnapescaped $sourcefsescaped\#$newsyncsnapescaped$guid");
				} else {
					$bookmarkcmd = "$sourcesudocmd $zfscmd bookmark $sourcefsescaped\@$newsyncsnapescaped $sourcefsescaped\#$newsyncsnapescaped$guid";
				}
				if ($debug) { print "DEBUG: $bookmarkcmd\n"; }
				system($bookmarkcmd) == 0 or do {
					warn "CRITICAL ERROR: $bookmarkcmd failed: $?";
					if ($exitcode < 2) { $exitcode = 2; }
					return 0;
				}
			};
		}
	} else {
		if (!defined $args{'keep-sync-snap'}) {
			# prune obsolete sync snaps on source and target (only if this run created ones).
			pruneoldsyncsnaps($sourcehost,$sourcefs,$newsyncsnap,$sourceisroot,keys %{ $snaps{'source'}});
			pruneoldsyncsnaps($targethost,$targetfs,$newsyncsnap,$targetisroot,keys %{ $snaps{'target'}});
		}
	}

} # end syncdataset()

sub compressargset {
	my ($value) = @_;
	my $DEFAULT_COMPRESSION = 'lzo';
	my %COMPRESS_ARGS = (
		'none' => {
			rawcmd      => '',
			args        => '',
			decomrawcmd => '',
			decomargs   => '',
		},
		'gzip' => {
			rawcmd      => 'gzip',
			args        => '-3',
			decomrawcmd => 'zcat',
			decomargs   => '',
		},
		'pigz-fast' => {
			rawcmd      => 'pigz',
			args        => '-3',
			decomrawcmd => 'pigz',
			decomargs   => '-dc',
		},
		'pigz-slow' => {
			rawcmd      => 'pigz',
			args        => '-9',
			decomrawcmd => 'pigz',
			decomargs   => '-dc',
		},
		'zstd-fast' => {
			rawcmd      => 'zstd',
			args        => '-3',
			decomrawcmd => 'zstd',
			decomargs   => '-dc',
		},
		'zstd-slow' => {
			rawcmd      => 'zstd',
			args        => '-19',
			decomrawcmd => 'zstd',
			decomargs   => '-dc',
		},
		'xz' => {
			rawcmd      => 'xz',
			args        => '',
			decomrawcmd => 'xz',
			decomargs   => '-d',
		},
		'lzo' => {
			rawcmd      => 'lzop',
			args        => '',
			decomrawcmd => 'lzop',
			decomargs   => '-dfc',
		},
		'lz4' => {
			rawcmd      => 'lz4',
			args        => '',
			decomrawcmd => 'lz4',
			decomargs   => '-dc',
		},
	);

	if ($value eq 'default') {
		$value = $DEFAULT_COMPRESSION;
	} elsif (!(grep $value eq $_, ('gzip', 'pigz-fast', 'pigz-slow', 'zstd-fast', 'zstd-slow', 'lz4', 'xz', 'lzo', 'default', 'none'))) {
		warn "Unrecognised compression value $value, defaulting to $DEFAULT_COMPRESSION";
		$value = $DEFAULT_COMPRESSION;
	}

	my %comargs = %{$COMPRESS_ARGS{$value}}; # copy
	$comargs{'compress'} = $value;
	$comargs{'cmd'}      = "$comargs{'rawcmd'} $comargs{'args'}";
	$comargs{'decomcmd'} = "$comargs{'decomrawcmd'} $comargs{'decomargs'}";
	return \%comargs;
}

sub checkcommands {
	# make sure compression, mbuffer, and pv are available on
	# source, target, and local hosts as appropriate.

	my %avail;
	my $sourcessh;
	my $targetssh;

	# if --nocommandchecks then assume everything's available and return
	if ($args{'nocommandchecks'}) {
		if ($debug) { print "DEBUG: not checking for command availability due to --nocommandchecks switch.\n"; }
		$avail{'compress'} = 1;
		$avail{'localpv'} = 1;
		$avail{'localmbuffer'} = 1;
		$avail{'sourcembuffer'} = 1;
		$avail{'targetmbuffer'} = 1;
		$avail{'sourceresume'} = 1;
		$avail{'targetresume'} = 1;
		return %avail;
	}

	if (!defined $sourcehost) { $sourcehost = ''; }
	if (!defined $targethost) { $targethost = ''; }

	if ($sourcehost ne '') { $sourcessh = "$sshcmd $sourcehost"; } else { $sourcessh = ''; }
	if ($targethost ne '') { $targetssh = "$sshcmd $targethost"; } else { $targetssh = ''; }

	# if raw compress command is null, we must have specified no compression. otherwise,
	# make sure that compression is available everywhere we need it
	if ($compressargs{'compress'} eq 'none') {
		if ($debug) { print "DEBUG: compression forced off from command line arguments.\n"; }
	} else {
		if ($debug) { print "DEBUG: checking availability of $compressargs{'rawcmd'} on source...\n"; }
		$avail{'sourcecompress'} = `$sourcessh $checkcmd $compressargs{'rawcmd'} 2>/dev/null`;
		if ($debug) { print "DEBUG: checking availability of $compressargs{'rawcmd'} on target...\n"; }
		$avail{'targetcompress'} = `$targetssh $checkcmd $compressargs{'rawcmd'} 2>/dev/null`;
		if ($debug) { print "DEBUG: checking availability of $compressargs{'rawcmd'} on local machine...\n"; }
		$avail{'localcompress'} = `$checkcmd $compressargs{'rawcmd'} 2>/dev/null`;
	}

	my ($s,$t);
	if ($sourcehost eq '') {
		$s = '[local machine]'
	} else {
		$s = $sourcehost;
		$s =~ s/^\S*\@//;
		$s = "ssh:$s";
	}
	if ($targethost eq '') {
		$t = '[local machine]'
	} else {
		$t = $targethost;
		$t =~ s/^\S*\@//;
		$t = "ssh:$t";
	}

	if (!defined $avail{'sourcecompress'}) { $avail{'sourcecompress'} = ''; }
	if (!defined $avail{'targetcompress'}) { $avail{'targetcompress'} = ''; }
	if (!defined $avail{'localcompress'}) { $avail{'localcompress'} = ''; }
	if (!defined $avail{'sourcembuffer'}) { $avail{'sourcembuffer'} = ''; }
	if (!defined $avail{'targetmbuffer'}) { $avail{'targetmbuffer'} = ''; }


	if ($avail{'sourcecompress'} eq '') {
		if ($compressargs{'rawcmd'} ne '') {
			print "WARN: $compressargs{'rawcmd'} not available on source $s- sync will continue without compression.\n";
		}
		$avail{'compress'} = 0;
	}
	if ($avail{'targetcompress'} eq '') {
		if ($compressargs{'rawcmd'} ne '') {
			print "WARN: $compressargs{'rawcmd'} not available on target $t - sync will continue without compression.\n";
		}
		$avail{'compress'} = 0;
	}
	if ($avail{'targetcompress'} ne '' && $avail{'sourcecompress'} ne '') {
		# compression available - unless source and target are both remote, which we'll check
		# for in the next block and respond to accordingly.
		$avail{'compress'} = 1;
	}

	# corner case - if source AND target are BOTH remote, we have to check for local compress too
	if ($sourcehost ne '' && $targethost ne '' && $avail{'localcompress'} eq '') {
		if ($compressargs{'rawcmd'} ne '') {
			print "WARN: $compressargs{'rawcmd'} not available on local machine - sync will continue without compression.\n";
		}
		$avail{'compress'} = 0;
	}

	if ($debug) { print "DEBUG: checking availability of $mbuffercmd on source...\n"; }
	$avail{'sourcembuffer'} = `$sourcessh $checkcmd $mbuffercmd 2>/dev/null`;
	if ($avail{'sourcembuffer'} eq '') {
		if (!$quiet) { print "WARN: $mbuffercmd not available on source $s - sync will continue without source buffering.\n"; }
		$avail{'sourcembuffer'} = 0;
	} else {
		$avail{'sourcembuffer'} = 1;
	}

	if ($debug) { print "DEBUG: checking availability of $mbuffercmd on target...\n"; }
	$avail{'targetmbuffer'} = `$targetssh $checkcmd $mbuffercmd 2>/dev/null`;
	if ($avail{'targetmbuffer'} eq '') {
		if (!$quiet) { print "WARN: $mbuffercmd not available on target $t - sync will continue without target buffering.\n"; }
		$avail{'targetmbuffer'} = 0;
	} else {
		$avail{'targetmbuffer'} = 1;
	}

	# if we're doing remote source AND remote target, check for local mbuffer as well
	if ($sourcehost ne '' && $targethost ne '') {
		if ($debug) { print "DEBUG: checking availability of $mbuffercmd on local machine...\n"; }
		$avail{'localmbuffer'} = `$checkcmd $mbuffercmd 2>/dev/null`;
		if ($avail{'localmbuffer'} eq '') {
			$avail{'localmbuffer'} = 0;
			if (!$quiet) { print "WARN: $mbuffercmd not available on local machine - sync will continue without local buffering.\n"; }
		}
	}

	if ($debug) { print "DEBUG: checking availability of $pvcmd on local machine...\n"; }
	$avail{'localpv'} = `$checkcmd $pvcmd 2>/dev/null`;
	if ($avail{'localpv'} eq '') {
		if (!$quiet) { print "WARN: $pvcmd not available on local machine - sync will continue without progress bar.\n"; }
		$avail{'localpv'} = 0;
	} else {
		$avail{'localpv'} = 1;
	}

	# check for ZFS resume feature support
	if ($resume) {
		my @parts = split ('/', $sourcefs);
		my $srcpool = $parts[0];
		@parts = split ('/', $targetfs);
		my $dstpool = $parts[0];

		$srcpool = escapeshellparam($srcpool);
		$dstpool = escapeshellparam($dstpool);

		if ($sourcehost ne '') {
			# double escaping needed
			$srcpool = escapeshellparam($srcpool);
		}

		if ($targethost ne '') {
			# double escaping needed
			$dstpool = escapeshellparam($dstpool);
		}

		my $resumechkcmd = "$zpoolcmd get -o value -H feature\@extensible_dataset";

		if ($debug) { print "DEBUG: checking availability of zfs resume feature on source...\n"; }
		$avail{'sourceresume'} = system("$sourcessh $sourcesudocmd $resumechkcmd $srcpool 2>/dev/null | grep '\\(active\\|enabled\\)' >/dev/null 2>&1");
		$avail{'sourceresume'} = $avail{'sourceresume'} == 0 ? 1 : 0;

		if ($debug) { print "DEBUG: checking availability of zfs resume feature on target...\n"; }
		$avail{'targetresume'} = system("$targetssh $targetsudocmd $resumechkcmd $dstpool 2>/dev/null | grep '\\(active\\|enabled\\)' >/dev/null 2>&1");
		$avail{'targetresume'} = $avail{'targetresume'} == 0 ? 1 : 0;

		if ($avail{'sourceresume'} == 0 || $avail{'targetresume'} == 0) {
			# disable resume
			$resume = '';

			my @hosts = ();
			if ($avail{'sourceresume'} == 0) {
				push @hosts, 'source';
			}
			if ($avail{'targetresume'} == 0) {
				push @hosts, 'target';
			}
			my $affected = join(" and ", @hosts);
			print "WARN: ZFS resume feature not available on $affected machine - sync will continue without resume support.\n";
		}
	} else {
		$avail{'sourceresume'} = 0;
		$avail{'targetresume'} = 0;
	}

	return %avail;
}

sub iszfsbusy {
	my ($rhost,$fs,$isroot) = @_;
	if ($rhost ne '') { $rhost = "$sshcmd $rhost"; }
	if ($debug) { print "DEBUG: checking to see if $fs on $rhost is already in zfs receive using $rhost $pscmd -Ao args= ...\n"; }

	open PL, "$rhost $pscmd -Ao args= |";
	my @processes = <PL>;
	close PL;

	foreach my $process (@processes) {
		# if ($debug) { print "DEBUG: checking process $process...\n"; }
		if ($process =~ /zfs *(receive|recv).*\Q$fs\E\Z/) {
			# there's already a zfs receive process for our target filesystem - return true
			if ($debug) { print "DEBUG: process $process matches target $fs!\n"; }
			return 1;
		}
	}

	# no zfs receive processes for our target filesystem found - return false
	return 0;
}

sub setzfsvalue {
	my ($rhost,$fs,$isroot,$property,$value) = @_;

	my $fsescaped = escapeshellparam($fs);

	if ($rhost ne '') {
		$rhost = "$sshcmd $rhost";
		# double escaping needed
		$fsescaped = escapeshellparam($fsescaped);
	}

	if ($debug) { print "DEBUG: setting $property to $value on $fs...\n"; }
	my $mysudocmd;
	if ($isroot) { $mysudocmd = ''; } else { $mysudocmd = $sudocmd; }
	if ($debug) { print "$rhost $mysudocmd $zfscmd set $property=$value $fsescaped\n"; }
	system("$rhost $mysudocmd $zfscmd set $property=$value $fsescaped") == 0
		or warn "WARNING: $rhost $mysudocmd $zfscmd set $property=$value $fsescaped died: $?, proceeding anyway.\n";
	return;
}

sub getzfsvalue {
	my ($rhost,$fs,$isroot,$property) = @_;

	my $fsescaped = escapeshellparam($fs);

	if ($rhost ne '') {
		$rhost = "$sshcmd $rhost";
		# double escaping needed
		$fsescaped = escapeshellparam($fsescaped);
	}

	if ($debug) { print "DEBUG: getting current value of $property on $fs...\n"; }
	my $mysudocmd;
	if ($isroot) { $mysudocmd = ''; } else { $mysudocmd = $sudocmd; }
	if ($debug) { print "$rhost $mysudocmd $zfscmd get -H $property $fsescaped\n"; }
	my ($value, $error, $exit) = capture {
		system("$rhost $mysudocmd $zfscmd get -H $property $fsescaped");
	};

	my @values = split(/\t/,$value);
	$value = $values[2];

	my $wantarray = wantarray || 0;

	# If we are in scalar context and there is an error, print it out.
	# Otherwise we assume the caller will deal with it.
	if (!$wantarray and $error) {
		print "ERROR getzfsvalue $fs $property: $error\n";
	}

	return $wantarray ? ($value, $error) : $value;
}

sub readablebytes {
	my $bytes = shift;
	my $disp;

	if ($bytes > 1024*1024*1024) {
		$disp = sprintf("%.1f",$bytes/1024/1024/1024) . ' GB';
	} elsif ($bytes > 1024*1024) {
		$disp = sprintf("%.1f",$bytes/1024/1024) . ' MB';
	} else {
		$disp = sprintf("%d",$bytes/1024) . ' KB';
	}
	return $disp;
}

sub getoldestsnapshot {
	my $snaps = shift;
	foreach my $snap ( sort { $snaps{'source'}{$a}{'createtxg'}<=>$snaps{'source'}{$b}{'createtxg'} } keys %{ $snaps{'source'} }) {
		# return on first snap found - it's the oldest
		return $snap;
	}
	# must not have had any snapshots on source - luckily, we already made one, amirite?
	if (defined ($args{'no-sync-snap'}) ) {
		# well, actually we set --no-sync-snap, so no we *didn't* already make one. Whoops.
		warn "CRIT: --no-sync-snap is set, and getoldestsnapshot() could not find any snapshots on source!\n";
	}
	return 0;
}

sub getnewestsnapshot {
	my $snaps = shift;
	foreach my $snap ( sort { $snaps{'source'}{$b}{'createtxg'}<=>$snaps{'source'}{$a}{'createtxg'} } keys %{ $snaps{'source'} }) {
		# return on first snap found - it's the newest
		if (!$quiet) { print "NEWEST SNAPSHOT: $snap\n"; }
		return $snap;
	}
	# must not have had any snapshots on source - looks like we'd better create one!
	if (defined ($args{'no-sync-snap'}) ) {
		if (!defined ($args{'recursive'}) ) {
			# well, actually we set --no-sync-snap and we're not recursive, so no we *can't* make one. Whoops.
			die "CRIT: --no-sync-snap is set, and getnewestsnapshot() could not find any snapshots on source!\n";
		}
		# fixme: we need to output WHAT the current dataset IS if we encounter this WARN condition.
		#        we also probably need an argument to mute this WARN, for people who deliberately exclude
		#        datasets from recursive replication this way.
		warn "WARN: --no-sync-snap is set, and getnewestsnapshot() could not find any snapshots on source for current dataset. Continuing.\n";
		if ($exitcode < 2) { $exitcode = 2; }
	}
	return 0;
}

sub buildsynccmd {
	my ($sendcmd,$recvcmd,$pvsize,$sourceisroot,$targetisroot) = @_;
	# here's where it gets fun: figuring out when to compress and decompress.
	# to make this work for all possible combinations, you may have to decompress
	# AND recompress across the pipe viewer. FUN.
	my $synccmd;

	if ($sourcehost eq '' && $targethost eq '') {
		# both sides local. don't compress. do mbuffer, once, on the source side.
		# $synccmd = "$sendcmd | $mbuffercmd | $pvcmd | $recvcmd";
		$synccmd = "$sendcmd |";
		# avoid confusion - accept either source-bwlimit or target-bwlimit as the bandwidth limiting option here
		my $bwlimit = '';
		if (length $args{'source-bwlimit'}) {
			$bwlimit = $args{'source-bwlimit'};
		} elsif (length $args{'target-bwlimit'}) {
			$bwlimit = $args{'target-bwlimit'};
		}

		if ($avail{'sourcembuffer'}) { $synccmd .= " $mbuffercmd $bwlimit $mbufferoptions |"; }
		if ($avail{'localpv'} && !$quiet) { $synccmd .= " $pvcmd $pvoptions -s $pvsize |"; }
		$synccmd .= " $recvcmd";
	} elsif ($sourcehost eq '') {
		# local source, remote target.
		#$synccmd = "$sendcmd | $pvcmd | $compressargs{'cmd'} | $mbuffercmd | $sshcmd $targethost '$compressargs{'decomcmd'} | $mbuffercmd | $recvcmd'";
		$synccmd = "$sendcmd |";
		if ($avail{'localpv'} && !$quiet) { $synccmd .= " $pvcmd $pvoptions -s $pvsize |"; }
		if ($avail{'compress'}) { $synccmd .= " $compressargs{'cmd'} |"; }
		if ($avail{'sourcembuffer'}) { $synccmd .= " $mbuffercmd $args{'source-bwlimit'} $mbufferoptions |"; }
		$synccmd .= " $sshcmd $targethost ";

		my $remotecmd = "";
		if ($avail{'targetmbuffer'}) { $remotecmd .= " $mbuffercmd $args{'target-bwlimit'} $mbufferoptions |"; }
		if ($avail{'compress'}) { $remotecmd .= " $compressargs{'decomcmd'} |"; }
		$remotecmd .= " $recvcmd";

		$synccmd .= escapeshellparam($remotecmd);
	} elsif ($targethost eq '') {
		# remote source, local target.
		#$synccmd = "$sshcmd $sourcehost '$sendcmd | $compressargs{'cmd'} | $mbuffercmd' | $compressargs{'decomcmd'} | $mbuffercmd | $pvcmd | $recvcmd";

		my $remotecmd = $sendcmd;
		if ($avail{'compress'}) { $remotecmd .= " | $compressargs{'cmd'}"; }
		if ($avail{'sourcembuffer'}) { $remotecmd .= " | $mbuffercmd $args{'source-bwlimit'} $mbufferoptions"; }

		$synccmd = "$sshcmd $sourcehost " . escapeshellparam($remotecmd);
		$synccmd .= " | ";
		if ($avail{'targetmbuffer'}) { $synccmd .= "$mbuffercmd $args{'target-bwlimit'} $mbufferoptions | "; }
		if ($avail{'compress'}) { $synccmd .= "$compressargs{'decomcmd'} | "; }
		if ($avail{'localpv'} && !$quiet) { $synccmd .= "$pvcmd $pvoptions -s $pvsize | "; }
		$synccmd .= "$recvcmd";
	} else {
		#remote source, remote target... weird, but whatever, I'm not here to judge you.
		#$synccmd = "$sshcmd $sourcehost '$sendcmd | $compressargs{'cmd'} | $mbuffercmd' | $compressargs{'decomcmd'} | $pvcmd | $compressargs{'cmd'} | $mbuffercmd | $sshcmd $targethost '$compressargs{'decomcmd'} | $mbuffercmd | $recvcmd'";

		my $remotecmd = $sendcmd;
		if ($avail{'compress'}) { $remotecmd .= " | $compressargs{'cmd'}"; }
		if ($avail{'sourcembuffer'}) { $remotecmd .= " | $mbuffercmd $args{'source-bwlimit'} $mbufferoptions"; }

		$synccmd = "$sshcmd $sourcehost " . escapeshellparam($remotecmd);
		$synccmd .= " | ";

		if ($avail{'compress'}) { $synccmd .= "$compressargs{'decomcmd'} | "; }
		if ($avail{'localpv'} && !$quiet) { $synccmd .= "$pvcmd $pvoptions -s $pvsize | "; }
		if ($avail{'compress'}) { $synccmd .= "$compressargs{'cmd'} | "; }
		if ($avail{'localmbuffer'}) { $synccmd .= "$mbuffercmd $mbufferoptions | "; }
		$synccmd .= "$sshcmd $targethost ";

		$remotecmd = "";
		if ($avail{'targetmbuffer'}) { $remotecmd .= " $mbuffercmd $args{'target-bwlimit'} $mbufferoptions |"; }
		if ($avail{'compress'}) { $remotecmd .= " $compressargs{'decomcmd'} |"; }
		$remotecmd .= " $recvcmd";

		$synccmd .= escapeshellparam($remotecmd);
	}
	return $synccmd;
}

sub pruneoldsyncsnaps {
	my ($rhost,$fs,$newsyncsnap,$isroot,@snaps) = @_;

	my $fsescaped = escapeshellparam($fs);

	if ($rhost ne '') { $rhost = "$sshcmd $rhost"; }

	my $hostid = hostname();

	my $mysudocmd;
	if ($isroot) { $mysudocmd=''; } else { $mysudocmd = $sudocmd; }

	my @prunesnaps;

	# only prune snaps beginning with syncoid and our own hostname
	foreach my $snap(@snaps) {
		if ($snap =~ /^syncoid_\Q$identifier$hostid\E/) {
			# no matter what, we categorically refuse to
			# prune the new sync snap we created for this run
			if ($snap ne $newsyncsnap) {
				push (@prunesnaps,$snap);
			}
		}
	}

	# concatenate pruning commands to ten per line, to cut down
	# auth times for any remote hosts that must be operated via SSH
	my $counter;
	my $maxsnapspercmd = 10;
	my $prunecmd;
	foreach my $snap(@prunesnaps) {
		$counter ++;
		$prunecmd .= "$mysudocmd $zfscmd destroy $fsescaped\@$snap; ";
		if ($counter > $maxsnapspercmd) {
			$prunecmd =~ s/\; $//;
			if ($debug) { print "DEBUG: pruning up to $maxsnapspercmd obsolete sync snapshots...\n"; }
			if ($debug) { print "DEBUG: $rhost $prunecmd\n"; }
			if ($rhost ne '') {
				$prunecmd = escapeshellparam($prunecmd);
			}
			system("$rhost $prunecmd") == 0
				or warn "WARNING: $rhost $prunecmd failed: $?";
			$prunecmd = '';
			$counter = 0;
		}
	}
	# if we still have some prune commands stacked up after finishing
	# the loop, commit 'em now
	if ($counter) {
		$prunecmd =~ s/\; $//;
		if ($debug) { print "DEBUG: pruning up to $maxsnapspercmd obsolete sync snapshots...\n"; }
		if ($debug) { print "DEBUG: $rhost $prunecmd\n"; }
		if ($rhost ne '') {
			$prunecmd = escapeshellparam($prunecmd);
		}
		system("$rhost $prunecmd") == 0
			or warn "WARNING: $rhost $prunecmd failed: $?";
	}
	return;
}

sub getmatchingsnapshot {
	my ($sourcefs, $targetfs, $snaps) = @_;
	foreach my $snap ( sort { $snaps{'source'}{$b}{'createtxg'}<=>$snaps{'source'}{$a}{'createtxg'} } keys %{ $snaps{'source'} }) {
		if (defined $snaps{'target'}{$snap}) {
			if ($snaps{'source'}{$snap}{'guid'} == $snaps{'target'}{$snap}{'guid'}) {
				return $snap;
			}
		}
	}

	return 0;
}

sub newsyncsnap {
	my ($rhost,$fs,$isroot) = @_;
	my $fsescaped = escapeshellparam($fs);
	if ($rhost ne '') {
		$rhost = "$sshcmd $rhost";
		# double escaping needed
		$fsescaped = escapeshellparam($fsescaped);
	}
	my $mysudocmd;
	if ($isroot) { $mysudocmd = ''; } else { $mysudocmd = $sudocmd; }
	my $hostid = hostname();
	my %date = getdate();
	my $snapname = "syncoid\_$identifier$hostid\_$date{'stamp'}";
	my $snapcmd = "$rhost $mysudocmd $zfscmd snapshot $fsescaped\@$snapname\n";
	if ($debug) { print "DEBUG: creating sync snapshot using \"$snapcmd\"...\n"; }
	system($snapcmd) == 0 or do {
		warn "CRITICAL ERROR: $snapcmd failed: $?";
		if ($exitcode < 2) { $exitcode = 2; }
		return 0;
	};

	return $snapname;
}

sub targetexists {
	my ($rhost,$fs,$isroot) = @_;
	my $fsescaped = escapeshellparam($fs);
	if ($rhost ne '') {
		$rhost = "$sshcmd $rhost";
		# double escaping needed
		$fsescaped = escapeshellparam($fsescaped);
	}
	my $mysudocmd;
	if ($isroot) { $mysudocmd = ''; } else { $mysudocmd = $sudocmd; }
	my $checktargetcmd = "$rhost $mysudocmd $zfscmd get -H name $fsescaped";
	if ($debug) { print "DEBUG: checking to see if target filesystem exists using \"$checktargetcmd 2>&1 |\"...\n"; }
	open FH, "$checktargetcmd 2>&1 |";
	my $targetexists = <FH>;
	close FH;
	my $exit = $?;
	$targetexists = ( $targetexists =~ /^\Q$fs\E/ && $exit == 0 );
	return $targetexists;
}

sub getssh {
	my $fs = shift;

	my $rhost = "";
	my $isroot;
	my $socket;
	my $remoteuser = "";

	# if we got passed something with an @ in it, we assume it's an ssh connection, eg root@myotherbox
	if ($fs =~ /\@/) {
		$rhost = $fs;
		$fs =~ s/^[^\@:]*\@[^\@:]*://;
		$rhost =~ s/:\Q$fs\E$//;
		$remoteuser = $rhost;
		$remoteuser =~ s/\@.*$//;
		# do not require a username to be specified
		$rhost =~ s/^@//;
	} elsif ($fs =~ m{^[^/]*:}) {
		# if we got passed something with an : in it, BEFORE any forward slash
		# (i.e., not in a dataset name) it MAY be an ssh connection
		# but we need to check if there is a pool with that name
		my $pool = $fs;
		$pool =~ s%/.*$%%;
		my ($pools, $error, $exit) = capture {
			system("$zfscmd list -d0 -H -oname");
		};
		$rhost = $fs;
		if ($exit != 0) {
			warn "Unable to enumerate pools (is zfs available?)";
		} else {
			foreach (split(/\n/,$pools)) {
				if ($_ eq $pool) {
					# there's a pool with this name.
					$rhost = "";
					last;
				}
			}
		}
		if ($rhost ne "") {
			# there's no pool that might conflict with this
			$rhost =~ s/:.*$//;
			$fs =~ s/\Q$rhost\E://;
		}
	}

	if ($rhost ne "") {
		if ($remoteuser eq 'root' || $args{'no-privilege-elevation'}) { $isroot = 1; } else { $isroot = 0; }
		# now we need to establish a persistent master SSH connection
		$socket = "/tmp/syncoid-$rhost-" . time();

		open FH, "$sshcmd -M -S $socket -o ControlPersist=1m $args{'sshport'} $rhost exit |";
		close FH;

		system("$sshcmd -S $socket $rhost echo -n") == 0 or do {
			my $code = $? >> 8;
			warn "CRITICAL ERROR: ssh connection echo test failed for $rhost with exit code $code";
			exit(2);
		};

		$rhost = "-S $socket $rhost";
	} else {
		my $localuid = $<;
		if ($localuid == 0 || $args{'no-privilege-elevation'}) { $isroot = 1; } else { $isroot = 0; }
	}
	# if ($isroot) { print "this user is root.\n"; } else { print "this user is not root.\n"; }
	return ($rhost,$fs,$isroot);
}

sub dumphash() {
	my $hash = shift;
	$Data::Dumper::Sortkeys = 1;
	print Dumper($hash);
}

sub getsnaps() {
	my ($type,$rhost,$fs,$isroot,%snaps) = @_;
	my $mysudocmd;
	my $fsescaped = escapeshellparam($fs);
	if ($isroot) { $mysudocmd = ''; } else { $mysudocmd = $sudocmd; }

	if ($rhost ne '') {
		$rhost = "$sshcmd $rhost";
		# double escaping needed
		$fsescaped = escapeshellparam($fsescaped);
	}

	my $getsnapcmd = "$rhost $mysudocmd $zfscmd get -Hpd 1 -t snapshot guid,creation,createtxg $fsescaped";
	if ($debug) {
		$getsnapcmd = "$getsnapcmd |";
		print "DEBUG: getting list of snapshots on $fs using $getsnapcmd...\n";
	} else {
		$getsnapcmd = "$getsnapcmd 2>/dev/null |";
	}
	open FH, $getsnapcmd;
	my @rawsnaps = <FH>;
	close FH or do {
		# fallback (solaris for example doesn't support the -t option)
		return getsnapsfallback($type,$rhost,$fs,$isroot,%snaps);
	};

	# this is a little obnoxious. get guid,creation returns guid,creation on two separate lines
	# as though each were an entirely separate get command.

	my %creationtimes=();

	foreach my $line (@rawsnaps) {
		# only import snap guids from the specified filesystem
		if ($line =~ /\Q$fs\E\@.*guid/) {
			chomp $line;
			my $guid = $line;
			$guid =~ s/^.*\tguid\t*(\d*).*/$1/;
			my $snap = $line;
			$snap =~ s/^.*\@(.*)\tguid.*$/$1/;
			$snaps{$type}{$snap}{'guid'}=$guid;
		}
	}

	foreach my $line (@rawsnaps) {
		# only import snap creations from the specified filesystem
		if ($line =~ /\Q$fs\E\@.*creation/) {
			chomp $line;
			my $creation = $line;
			$creation =~ s/^.*\tcreation\t*(\d*).*/$1/;
			my $snap = $line;
			$snap =~ s/^.*\@(.*)\tcreation.*$/$1/;

			# the accuracy of the creation timestamp is only for a second, but
			# snapshots in the same second are highly likely. The list command
			# has an ordered output so we append another three digit running number
			# to the creation timestamp and make sure those are ordered correctly
			# for snapshot with the same creation timestamp
			my $counter = 0;
			my $creationsuffix;
			while ($counter < 999) {
				$creationsuffix = sprintf("%s%03d", $creation, $counter);
				if (!defined $creationtimes{$creationsuffix}) {
					$creationtimes{$creationsuffix} = 1;
					last;
				}
				$counter += 1;
			}

			$snaps{$type}{$snap}{'creation'}=$creationsuffix;
		}
	}

	foreach my $line (@rawsnaps) {
                # only import snap createtxg from the specified filesystem
                if ($line =~ /\Q$fs\E\@.*createtxg/) {
                        chomp $line;
                        my $createtxg = $line;
                        $createtxg =~ s/^.*\tcreatetxg\t*(\d*).*/$1/;
                        my $snap = $line;
                        $snap =~ s/^.*\@(.*)\tcreatetxg.*$/$1/;
                        $snaps{$type}{$snap}{'createtxg'}=$createtxg;
                }
        }

	return %snaps;
}

sub getsnapsfallback() {
	# fallback (solaris for example doesn't support the -t option)
	my ($type,$rhost,$fs,$isroot,%snaps) = @_;
	my $mysudocmd;
	my $fsescaped = escapeshellparam($fs);
	if ($isroot) { $mysudocmd = ''; } else { $mysudocmd = $sudocmd; }

	if ($rhost ne '') {
		$rhost = "$sshcmd $rhost";
		# double escaping needed
		$fsescaped = escapeshellparam($fsescaped);
	}

	my $getsnapcmd = "$rhost $mysudocmd $zfscmd get -Hpd 1 type,guid,creation,createtxg $fsescaped |";
	warn "snapshot listing failed, trying fallback command";
	if ($debug) { print "DEBUG: FALLBACK, getting list of snapshots on $fs using $getsnapcmd...\n"; }
	open FH, $getsnapcmd;
	my @rawsnaps = <FH>;
	close FH or die "CRITICAL ERROR: snapshots couldn't be listed for $fs (exit code $?)";

	my %creationtimes=();

	my $state = 0;
	foreach my $line (@rawsnaps) {
		if ($state < 0) {
			$state++;
			next;
		}

		if ($state eq 0) {
			if ($line !~ /\Q$fs\E\@.*type\s*snapshot/) {
				# skip non snapshot type object
				$state = -2;
				next;
			}
		} elsif ($state eq 1) {
			if ($line !~ /\Q$fs\E\@.*guid/) {
				die "CRITICAL ERROR: snapshots couldn't be listed for $fs (guid parser error)";
			}

			chomp $line;
			my $guid = $line;
			$guid =~ s/^.*\tguid\t*(\d*).*/$1/;
			my $snap = $line;
			$snap =~ s/^.*\@(.*)\tguid.*$/$1/;
			$snaps{$type}{$snap}{'guid'}=$guid;
		} elsif ($state eq 2) {
			if ($line !~ /\Q$fs\E\@.*creation/) {
				die "CRITICAL ERROR: snapshots couldn't be listed for $fs (creation parser error)";
			}

			chomp $line;
			my $creation = $line;
			$creation =~ s/^.*\tcreation\t*(\d*).*/$1/;
			my $snap = $line;
			$snap =~ s/^.*\@(.*)\tcreation.*$/$1/;

			# the accuracy of the creation timestamp is only for a second, but
			# snapshots in the same second are highly likely. The list command
			# has an ordered output so we append another three digit running number
			# to the creation timestamp and make sure those are ordered correctly
			# for snapshot with the same creation timestamp
			my $counter = 0;
			my $creationsuffix;
			while ($counter < 999) {
				$creationsuffix = sprintf("%s%03d", $creation, $counter);
				if (!defined $creationtimes{$creationsuffix}) {
					$creationtimes{$creationsuffix} = 1;
					last;
				}
				$counter += 1;
			}

			$snaps{$type}{$snap}{'creation'}=$creationsuffix;
			$state = -1;
		} elsif ($state eq 3) {
			if ($line !~ /\Q$fs\E\@.*createtxg/) {
                        	die "CRITICAL ERROR: snapshots couldn't be listed for $fs (createtxg parser error)";
                        }

                        chomp $line;
                        my $createtxg = $line;
                        $createtxg =~ s/^.*\tcreatetxg\t*(\d*).*/$1/;
                        my $snap = $line;
                        $snap =~ s/^.*\@(.*)\tcreatetxg.*$/$1/;
                        $snaps{$type}{$snap}{'createtxg'}=$createtxg;
		}

		$state++;
	}

	return %snaps;
}

sub getbookmarks() {
	my ($rhost,$fs,$isroot,%bookmarks) = @_;
	my $mysudocmd;
	my $fsescaped = escapeshellparam($fs);
	if ($isroot) { $mysudocmd = ''; } else { $mysudocmd = $sudocmd; }

	if ($rhost ne '') {
		$rhost = "$sshcmd $rhost";
		# double escaping needed
		$fsescaped = escapeshellparam($fsescaped);
	}

	my $error = 0;
	my $getbookmarkcmd = "$rhost $mysudocmd $zfscmd get -Hpd 1 -t bookmark guid,creation $fsescaped 2>&1 |";
	if ($debug) { print "DEBUG: getting list of bookmarks on $fs using $getbookmarkcmd...\n"; }
	open FH, $getbookmarkcmd;
	my @rawbookmarks = <FH>;
	close FH or $error = 1;

	if ($error == 1) {
		if ($rawbookmarks[0] =~ /invalid type/ or $rawbookmarks[0] =~ /operation not applicable to datasets of this type/) {
			# no support for zfs bookmarks, return empty hash
			return %bookmarks;
		}

		die "CRITICAL ERROR: bookmarks couldn't be listed for $fs (exit code $?)";
	}

	# this is a little obnoxious. get guid,creation returns guid,creation on two separate lines
	# as though each were an entirely separate get command.

	my $lastguid;

	foreach my $line (@rawbookmarks) {
		# only import bookmark guids, creation from the specified filesystem
		if ($line =~ /\Q$fs\E\#.*guid/) {
			chomp $line;
			$lastguid = $line;
			$lastguid =~ s/^.*\tguid\t*(\d*).*/$1/;
			my $bookmark = $line;
			$bookmark =~ s/^.*\#(.*)\tguid.*$/$1/;
			$bookmarks{$lastguid}{'name'}=$bookmark;
		} elsif ($line =~ /\Q$fs\E\#.*creation/) {
			chomp $line;
			my $creation = $line;
			$creation =~ s/^.*\tcreation\t*(\d*).*/$1/;
			my $bookmark = $line;
			$bookmark =~ s/^.*\#(.*)\tcreation.*$/$1/;
			$bookmarks{$lastguid}{'creation'}=$creation;
		}
	}

	return %bookmarks;
}

sub getsendsize {
	my ($sourcehost,$snap1,$snap2,$isroot,$receivetoken) = @_;

	my $snap1escaped = escapeshellparam($snap1);
	my $snap2escaped = escapeshellparam($snap2);

	my $mysudocmd;
	if ($isroot) { $mysudocmd = ''; } else { $mysudocmd = $sudocmd; }

	my $sourcessh;
	if ($sourcehost ne '') {
		$sourcessh = "$sshcmd $sourcehost";
		$snap1escaped = escapeshellparam($snap1escaped);
		$snap2escaped = escapeshellparam($snap2escaped);
	} else {
		$sourcessh = '';
	}

	my $snaps;
	if ($snap2) {
		# if we got a $snap2 argument, we want an incremental send estimate from $snap1 to $snap2.
		$snaps = "$args{'streamarg'} $snap1escaped $snap2escaped";
	} else {
		# if we didn't get a $snap2 arg, we want a full send estimate for $snap1.
		$snaps = "$snap1escaped";
	}

	# in case of a resumed receive, get the remaining
	# size based on the resume token
	if (defined($receivetoken)) {
		$snaps = "-t $receivetoken";
	}

	my $sendoptions;
	if (defined($receivetoken)) {
		$sendoptions = getoptionsline(\@sendoptions, ('e'));
	} else {
		$sendoptions = getoptionsline(\@sendoptions, ('D','L','R','c','e','h','p','w'));
	}
	my $getsendsizecmd = "$sourcessh $mysudocmd $zfscmd send $sendoptions -nvP $snaps";
	if ($debug) { print "DEBUG: getting estimated transfer size from source $sourcehost using \"$getsendsizecmd 2>&1 |\"...\n"; }

	open FH, "$getsendsizecmd 2>&1 |";
	my @rawsize = <FH>;
	close FH;
	my $exit = $?;

	# process sendsize: last line of multi-line output is
	# size of proposed xfer in bytes, but we need to remove
	# human-readable crap from it
	my $sendsize = pop(@rawsize);
	# the output format is different in case of
	# a resumed receive
	if (defined($receivetoken)) {
		$sendsize =~ s/.*\t([0-9]+)$/$1/;
	} else {
		$sendsize =~ s/^size\t*//;
	}
	chomp $sendsize;

	# check for valid value
	if ($sendsize !~ /^\d+$/) {
		$sendsize = '';
	}

	# to avoid confusion with a zero size pv, give sendsize
	# a minimum 4K value - or if empty, make sure it reads UNKNOWN
	if ($debug) { print "DEBUG: sendsize = $sendsize\n"; }
	if ($sendsize eq '' || $exit != 0) {
		$sendsize = '0';
	} elsif ($sendsize < 4096) {
		$sendsize = 4096;
	}
	return $sendsize;
}

sub getdate {
	my @time = localtime(time);

	# get timezone info
	my $offset = timegm(@time) - timelocal(@time);
	my $sign = ''; # + is not allowed in a snapshot name
	if ($offset < 0) {
		$sign = '-';
		$offset = abs($offset);
	}
	my $hours = int($offset / 3600);
	my $minutes = int($offset / 60) - $hours * 60;

	my ($sec,$min,$hour,$mday,$mon,$year,$wday,$yday,$isdst) = @time;
	$year += 1900;
	my %date;
	$date{'unix'} = (((((((($year - 1971) * 365) + $yday) * 24) + $hour) * 60) + $min) * 60) + $sec;
	$date{'year'} = $year;
	$date{'sec'} = sprintf ("%02u", $sec);
	$date{'min'} = sprintf ("%02u", $min);
	$date{'hour'} = sprintf ("%02u", $hour);
	$date{'mday'} = sprintf ("%02u", $mday);
	$date{'mon'} = sprintf ("%02u", ($mon + 1));
	$date{'tzoffset'} = sprintf ("GMT%s%02d:%02u", $sign, $hours, $minutes);
	$date{'stamp'} = "$date{'year'}-$date{'mon'}-$date{'mday'}:$date{'hour'}:$date{'min'}:$date{'sec'}-$date{'tzoffset'}";
	return %date;
}

sub escapeshellparam {
	my ($par) = @_;
	# avoid use of uninitialized string in regex
	if (length($par)) {
		# "escape" all single quotes
		$par =~ s/'/'"'"'/g;
	} else {
		# avoid use of uninitialized string in concatenation below
		$par = '';
	}
	# single-quote entire string
	return "'$par'";
}

sub getreceivetoken() {
	my ($rhost,$fs,$isroot) = @_;
	my $token = getzfsvalue($rhost,$fs,$isroot,"receive_resume_token");

	if (defined $token && $token ne '-' && $token ne '') {
		return $token;
	}

	if ($debug) {
        print "DEBUG: no receive token found \n";
    }

	return
}

sub parsespecialoptions {
	my ($line) = @_;

	my @options = ();

	my @values = split(/ /, $line);

	my $optionValue = 0;
	my $lastOption;

	foreach my $value (@values) {
		if ($optionValue ne 0) {
			my %item = (
				"option"  => $lastOption,
				"line" => "-$lastOption $value",
			);

			push @options, \%item;
			$optionValue = 0;
			next;
		}

		for my $char (split //, $value) {
			if ($optionValue ne 0) {
				return undef;
			}

			if ($char eq 'o' || $char eq 'x') {
				$lastOption = $char;
				$optionValue = 1;
			} else {
				my %item = (
					"option"  => $char,
					"line" => "-$char",
				);

				push @options, \%item;
			}
		}
	}

	return @options;
}

sub getoptionsline {
	my ($options_ref, @allowed) = @_;

	my $line = '';

	foreach my $value (@{ $options_ref }) {
		if (@allowed) {
			if (!grep( /^$$value{'option'}$/, @allowed) ) {
				next;
			}
		}

		$line = "$line$$value{'line'} ";
	}

	return $line;
}

sub resetreceivestate {
	my ($rhost,$fs,$isroot) = @_;

	my $fsescaped = escapeshellparam($fs);

	if ($rhost ne '') {
		$rhost = "$sshcmd $rhost";
		# double escaping needed
		$fsescaped = escapeshellparam($fsescaped);
	}

	if ($debug) { print "DEBUG: reset partial receive state of $fs...\n"; }
	my $mysudocmd;
	if ($isroot) { $mysudocmd = ''; } else { $mysudocmd = $sudocmd; }
	my $resetcmd = "$rhost $mysudocmd $zfscmd receive -A $fsescaped";
	if ($debug) { print "$resetcmd\n"; }
	system("$resetcmd") == 0
		or die "CRITICAL ERROR: $resetcmd failed: $?";
}

__END__

=head1 NAME

syncoid - ZFS snapshot replication tool

=head1 SYNOPSIS

 syncoid [options]... SOURCE TARGET
 or   syncoid [options]... SOURCE [[USER]@]HOST:TARGET
 or   syncoid [options]... [[USER]@]HOST:SOURCE TARGET
 or   syncoid [options]... [[USER]@]HOST:SOURCE [[USER]@]HOST:TARGET

 SOURCE                Source ZFS dataset. Can be either local or remote
 TARGET                Target ZFS dataset. Can be either local or remote

Options:

  --compress=FORMAT     Compresses data during transfer. Currently accepted options are gzip, pigz-fast, pigz-slow, zstd-fast, zstd-slow, lz4, xz, lzo (default) & none
  --identifier=EXTRA    Extra identifier which is included in the snapshot name. Can be used for replicating to multiple targets.
  --recursive|r         Also transfers child datasets
  --skip-parent         Skips syncing of the parent dataset. Does nothing without '--recursive' option.
  --source-bwlimit=<limit k|m|g|t>  Bandwidth limit in bytes/kbytes/etc per second on the source transfer
  --target-bwlimit=<limit k|m|g|t>  Bandwidth limit in bytes/kbytes/etc per second on the target transfer
  --mbuffer-size=VALUE  Specify the mbuffer size (default: 16M), please refer to mbuffer(1) manual page.
  --pv-options=OPTIONS  Configure how pv displays the progress bar, default '-p -t -e -r -b'
  --no-stream           Replicates using newest snapshot instead of intermediates
  --no-sync-snap        Does not create new snapshot, only transfers existing
  --keep-sync-snap      Don't destroy created sync snapshots
  --create-bookmark     Creates a zfs bookmark for the newest snapshot on the source after replication succeeds (only works with --no-sync-snap)
  --preserve-recordsize Preserves the recordsize on initial sends to the target
  --no-clone-rollback   Does not rollback clones on target
  --no-rollback         Does not rollback clones or snapshots on target (it probably requires a readonly target)
  --exclude=REGEX       Exclude specific datasets which match the given regular expression. Can be specified multiple times
  --sendoptions=OPTIONS Use advanced options for zfs send (the arguments are filtered as needed), e.g. syncoid --sendoptions="Lc e" sets zfs send -L -c -e ...
  --recvoptions=OPTIONS Use advanced options for zfs receive (the arguments are filtered as needed), e.g. syncoid --recvoptions="ux recordsize o compression=lz4" sets zfs receive -u -x recordsize -o compression=lz4 ...
  --sshkey=FILE         Specifies a ssh key to use to connect
  --sshport=PORT        Connects to remote on a particular port
  --sshcipher|c=CIPHER  Passes CIPHER to ssh to use a particular cipher set
  --sshoption|o=OPTION  Passes OPTION to ssh for remote usage. Can be specified multiple times

  --help                Prints this helptext
  --version             Prints the version number
  --debug               Prints out a lot of additional information during a syncoid run
  --monitor-version     Currently does nothing
  --quiet               Suppresses non-error output
  --dumpsnaps           Dumps a list of snapshots during the run
  --no-command-checks   Do not check command existence before attempting transfer. Not recommended
  --no-resume           Don't use the ZFS resume feature if available
  --no-clone-handling   Don't try to recreate clones on target
  --no-privilege-elevation  Bypass the root check, for use with ZFS permission delegation

  --force-delete        Remove target datasets recursively, if there are no matching snapshots/bookmarks