btrbk/btrbk

#!/usr/bin/perl -T

=head1 NAME

btrbk - backup tool for btrfs volumes

=head1 SYNOPSIS

    btrbk --help

=head1 DESCRIPTION

Backup tool for btrfs subvolumes, taking advantage of btrfs specific
send-receive mechanism, allowing incremental backups at file-system
level.

The full btrbk documentation is available at L<http://www.digint.ch/btrbk/>.

=head1 AUTHOR

Axel Burri <axel@tty0.ch>

=head1 COPYRIGHT AND LICENSE

Copyright (c) 2014-2015 Axel Burri. All rights reserved.

This program is free software: you can redistribute it and/or modify
it under the terms of the GNU General Public License as published by
the Free Software Foundation, either version 3 of the License, or
(at your option) any later version.

This program is distributed in the hope that it will be useful,
but WITHOUT ANY WARRANTY; without even the implied warranty of
MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
GNU General Public License for more details.

You should have received a copy of the GNU General Public License
along with this program.  If not, see <http://www.gnu.org/licenses/>.

=cut

use strict;
use warnings FATAL => qw( all );

use Carp qw(confess);
use Date::Calc qw(Today_and_Now Delta_Days Day_of_Week);
use Getopt::Long qw(GetOptions);
use Data::Dumper;

our $VERSION       = "0.21.0-dev";
our $AUTHOR        = 'Axel Burri <axel@tty0.ch>';
our $PROJECT_HOME  = '<http://www.digint.ch/btrbk/>';

my $version_info   = "btrbk command line client, version $VERSION";

my @config_src = ("/etc/btrbk.conf", "/etc/btrbk/btrbk.conf");

my $ip_addr_match = qr/(([0-9]|[1-9][0-9]|1[0-9]{2}|2[0-4][0-9]|25[0-5])\.){3}([0-9]|[1-9][0-9]|1[0-9]{2}|2[0-4][0-9]|25[0-5])/;
my $host_name_match = qr/(([a-zA-Z0-9]|[a-zA-Z0-9][a-zA-Z0-9\-]*[a-zA-Z0-9])\.)*([A-Za-z0-9]|[A-Za-z0-9][A-Za-z0-9\-]*[A-Za-z0-9])/;
my $file_match = qr/[0-9a-zA-Z_@\+\-\.\/]+/;  # note: ubuntu uses '@' in the subvolume layout: <https://help.ubuntu.com/community/btrfs>
my $ssh_prefix_match = qr/ssh:\/\/($ip_addr_match|$host_name_match)/;
my $uuid_match = qr/[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}/;
my $timestamp_postfix_match = qr/\.(?<YYYY>[0-9]{4})(?<MM>[0-9]{2})(?<DD>[0-9]{2})(T(?<hh>[0-9]{2})(?<mm>[0-9]{2}))?(_(?<NN>[0-9]+))?/;  # matches "YYYYMMDD[Thhmm][_NN]"
my $raw_postfix_match = qr/--(?<received_uuid>$uuid_match)(\@(?<parent_uuid>$uuid_match))\.btrfs?(\.(?<compress>(gz|bz2|xz)))?(\.(?<encrypt>gpg))?/;  # matches ".btrfs_<received_uuid>[@<parent_uuid>][.gz|bz2|xz][.gpg]"
my $group_match = qr/[a-zA-Z0-9_:-]+/;
my $ssh_cipher_match = qr/[a-z0-9][a-z0-9@.-]+/;

my %day_of_week_map = ( monday => 1, tuesday => 2, wednesday => 3, thursday => 4, friday => 5, saturday => 6, sunday => 7 );

my %config_options = (
  # NOTE: the parser always maps "no" to undef
  # NOTE: keys "volume", "subvolume" and "target" are hardcoded
  timestamp_format          => { default => "short",  accept => [ "short", "long" ], context => [ "root", "volume", "subvolume" ] },
  snapshot_dir              => { default => undef,    accept_file => { relative => 1 } },
  snapshot_name             => { default => undef,    accept_file => { name_only => 1 }, context => [ "subvolume" ] },   # NOTE: defaults to the subvolume name (hardcoded)
  snapshot_create           => { default => "always", accept => [ "no", "always", "ondemand", "onchange" ] },
  incremental               => { default => "yes",    accept => [ "yes", "no", "strict" ] },
  resume_missing            => { default => "yes",    accept => [ "yes", "no" ] },
  preserve_day_of_week      => { default => "sunday", accept => [ (keys %day_of_week_map) ] },
  snapshot_preserve_daily   => { default => "all",    accept => [ "all" ], accept_numeric => 1  },
  snapshot_preserve_weekly  => { default => 0,        accept => [ "all" ], accept_numeric => 1  },
  snapshot_preserve_monthly => { default => "all",    accept => [ "all" ], accept_numeric => 1  },
  target_preserve_daily     => { default => "all",    accept => [ "all" ], accept_numeric => 1  },
  target_preserve_weekly    => { default => 0,        accept => [ "all" ], accept_numeric => 1  },
  target_preserve_monthly   => { default => "all",    accept => [ "all" ], accept_numeric => 1  },
  btrfs_commit_delete       => { default => undef,    accept => [ "after", "each", "no" ] },
  ssh_identity              => { default => undef,    accept_file => { absolute => 1 } },
  ssh_user                  => { default => "root",   accept_regexp => qr/^[a-z_][a-z0-9_-]*$/ },
  ssh_port                  => { default => "default", accept => [ "default" ], accept_numeric => 1 },
  ssh_compression           => { default => undef,    accept => [ "yes", "no" ] },
  ssh_cipher_spec           => { default => "default", accept_regexp => qr/^$ssh_cipher_match(,$ssh_cipher_match)*$/ },

  raw_target_compress       => { default => undef,    accept => [ "no", "gzip", "bzip2", "xz" ] },
  raw_target_encrypt        => { default => undef,    accept => [ "no", "gpg" ] },
  gpg_keyring               => { default => undef,    accept_file => { absolute => 1 } },
  gpg_recipient             => { default => undef,    accept_regexp => qr/^[0-9a-zA-Z_@\+\-\.]+$/ },

  btrfs_progs_compat        => { default => undef,    accept => [ "yes", "no" ] },
  group                     => { default => undef,    accept_regexp => qr/^$group_match(\s*,\s*$group_match)*$/, split => qr/\s*,\s*/ },

  # deprecated options
  snapshot_create_always    => { default => undef, accept => [ "yes", "no" ],
                                 deprecated => { yes => { warn => "Please use \"snapshot_create always\"",
                                                          replace_key   => "snapshot_create",
                                                          replace_value => "always",
                                                         },
                                                 no  => { warn => "Please use \"snapshot_create no\" or \"snapshot_create ondemand\"",
                                                          replace_key   => "snapshot_create",
                                                          replace_value => "ondemand",
                                                         }
                                                },
                               },
  receive_log               => { default => undef, accept => [ "sidecar", "no" ], accept_file => { absolute => 1 },
                                 deprecated => { DEFAULT => { warn => "ignoring" } },
                               }
 );

my @config_target_types = qw(send-receive raw);

my %table_formats = (
  list_volume => { table => [ qw( volume ) ],
                   long  => [ qw( volume_host volume_path ) ],
                   raw   => [ qw( volume_url volume_host volume_path volume_rsh ) ],
                 },
  list_source => { table => [ qw( source_host source_path snapshot_path snapshot_name ) ],
                   long  => [ qw( source_host source_path snapshot_path snapshot_name ) ],
                   raw   => [ qw( source_url source_host source_path snapshot_path snapshot_name source_rsh ) ],
                 },
  list_target => { table => [ qw( target ) ],
                   long  => [ qw( target_host target_path ) ],
                   raw   => [ qw( target_url target_host target_path target_rsh ) ],
                 },
  list        => { table => [ qw( source snapshot_path snapshot_name target ) ],
                   long  => [ qw( source_host source_path snapshot_path snapshot_name snapshot_preserve target_host target_path target_preserve ) ],
                   raw   => [ qw( source_url source_host source_path snapshot_path snapshot_name snapshot_preserve target_url target_host target_path target_preserve source_rsh target_rsh ) ],
                 },

  tree        => { table => [ qw( source snapshot btrbk_flags received ) ],
                   long  => [ qw( source_host source_path snapshot_path snapshot_name btrbk_flags received_host received_path ) ],
                   raw   => [ qw( source_host source_path snapshot_path snapshot_name btrbk_flags received_host received_path source_rsh ) ],
                 },

  schedule    => { table => [ qw( action target scheme reason ) ],
                   long  => [ qw( action host root_path name scheme reason ) ],
                   raw   => [ qw( topic action url host path dow d m w) ],
                 },

  action_log  => { table => [ qw( type status target source parent ) ],
                   long  => [ qw( localtime type status target_host target_path source_host source_path parent_path error_message ) ],
                   raw   => [ qw( time localtime type status target_url source_url parent_url error_message ) ],
                 },
);

my %root_tree_cache;    # map URL to SUBTREE (needed since "btrfs subvolume list" does not provide us with the uuid of the btrfs root node)
my %vinfo_cache;        # map URL to vinfo
my %uuid_info;          # map UUID to btr_tree node
my %uuid_fs_map;        # map UUID to URL

my @action_log;

my $dryrun;
my $loglevel = 1;
my $show_progress = 0;
my $err = "";
my $output_format = undef;


$SIG{__DIE__} = sub {
  print STDERR "\nERROR: process died unexpectedly (btrbk v$VERSION)";
  print STDERR "\nPlease contact the author: $AUTHOR\n\n";
  print STDERR "Stack Trace:\n----------------------------------------\n";
  Carp::confess @_;
};

sub VERSION_MESSAGE
{
  print STDERR $version_info . "\n\n";
}

sub HELP_MESSAGE
{
  print STDERR "usage: btrbk [options] <command>\n";
  print STDERR "\n";
  print STDERR "options:\n";
  #            "--------------------------------------------------------------------------------"; # 80
  print STDERR "   -h, --help            display this help message\n";
  print STDERR "       --version         display version information\n";
  print STDERR "   -c, --config=FILE     specify configuration file\n";
  print STDERR "   -p, --preserve        preserve all backups (do not delete any old targets)\n";
  print STDERR "   -r, --resume-only     resume only (do not create new snapshots, only resume\n";
  print STDERR "                         missing backups)\n";
  print STDERR "   -v, --verbose         be verbose (set loglevel=info)\n";
  print STDERR "   -q, --quiet           be quiet (do not print summary for the \"run\" command)\n";
  print STDERR "   -l, --loglevel=LEVEL  set logging level (warn, info, debug, trace)\n";
  print STDERR "   -t, --table           change output to table format\n";
  print STDERR "       --format=FORMAT   change output format, FORMAT=table|long|raw\n";
  print STDERR "       --progress        show progress bar on send-receive operation\n";
  print STDERR "\n";
  print STDERR "commands:\n";
  print STDERR "   run    [filter...]  perform backup operations as defined in the config file\n";
  print STDERR "   dryrun [filter...]  don't run btrfs commands; show what would be executed\n";
  print STDERR "   tree   [filter...]  shows backup tree\n";
  print STDERR "   list   [filter...]  print source/snapshot/target relations\n";
  print STDERR "   info   [filter...]  print useful filesystem information\n";
  print STDERR "   origin <subvol>     print origin information for subvolume\n";
  print STDERR "   diff   <from> <to>  shows new files since subvolume <from> for subvolume <to>\n";
  print STDERR "\n";
  print STDERR "For additional information, see $PROJECT_HOME\n";
}

sub TRACE { my $t = shift; print STDERR "... $t\n" if($loglevel >= 4);  }
sub DEBUG { my $t = shift; print STDERR "$t\n" if($loglevel >= 3);  }
sub INFO  { my $t = shift; print STDERR "$t\n" if($loglevel >= 2);  }
sub WARN  { my $t = shift; print STDERR "WARNING: $t\n" if($loglevel >= 1);  }
sub ERROR { my $t = shift; print STDERR "ERROR: $t\n";  }

sub action($@)
{
  my $type = shift // die;
  my %h = @_;
  $h{type} = $type;
  $h{time} = time;
  push @action_log, \%h;
}

sub ABORTED($$)
{
  my $config = shift;
  my $t = shift;
  $config->{ABORTED} = $t;
  unless($t eq "USER_SKIP") {
    action("abort_" . ($config->{CONTEXT} || "undef"),
           status => "ABORT",
           vinfo_prefixed_keys("target", vinfo($config->{url}, $config)),
           error_message => $t,
          );
  }
}

sub run_cmd(@)
{
  my @commands = (ref($_[0]) eq "HASH") ? @_ : { @_ };
  $err = "";

  my $cmd = "";
  my $name = "";
  my $destructive = 0;
  my $pipe = "";
  my $catch_stderr = 0;
  my $filter_stderr = undef;
  foreach (@commands) {
    $_->{rsh} //= [];
    $_->{cmd} = [ @{$_->{rsh}}, @{$_->{cmd}} ];
    $_->{cmd_text} = join(' ', map { s/\n/\\n/g; "'$_'" } @{$_->{cmd}});  # ugly escape of \n, do we need to escape others?
    $name = $_->{name} // $_->{cmd_text};
    $_->{_buf} = '';
    $cmd .= $pipe . $_->{cmd_text};
    $pipe = ' | ';
    if($_->{catch_stderr}) {
      $cmd .= ' 2>&1';
      $catch_stderr = 1;
      $filter_stderr = $_->{filter_stderr};
    }
    $destructive = 1 unless($_->{non_destructive});
  }

  if($dryrun && $destructive) {
    DEBUG "### (dryrun) $cmd";
    return "";
  }
  DEBUG "### $cmd";

  my $ret = "";
  $ret = `$cmd`;
  chomp($ret);
  TRACE "Command output:\n$ret";
  if($?) {
    my $exitcode= $? >> 8;
    my $signal = $? & 127;
    DEBUG "Command execution failed (exitcode=$exitcode" . ($signal ? ", signal=$signal" : "") . "): \"$cmd\"";

    if($catch_stderr) {
      $_ = $ret;
      &{$filter_stderr} ($cmd) if($filter_stderr);
      ERROR "[$cmd] $_" if($_);
    }
    return undef;
  }
  else {
    DEBUG "Command execution successful";
  }
  return $ret;
}


sub vinfo($$)
{
  my $url = shift // die;
  my $config = shift || die;

  my $name = $url;
  $name =~ s/^.*\///;
  my %info = (
    URL => $url,
    NAME => $name,
   );

  if($url =~ /^ssh:\/\/(\S+?)(\/\S+)$/) {
    my ($host, $path)   = ($1, $2);
    my $ssh_port        = config_key($config, "ssh_port");
    my $ssh_user        = config_key($config, "ssh_user");
    my $ssh_identity    = config_key($config, "ssh_identity");
    my $ssh_compression = config_key($config, "ssh_compression");
    my $ssh_cipher_spec = config_key($config, "ssh_cipher_spec") // "default";
    my @ssh_options;
    push(@ssh_options, '-p', $ssh_port) if($ssh_port ne "default");
    push(@ssh_options, '-C') if($ssh_compression);
    push(@ssh_options, '-c', $ssh_cipher_spec) if($ssh_cipher_spec ne "default");
    if($ssh_identity) {
      push(@ssh_options, '-i', $ssh_identity);
    } else {
      WARN "No SSH identity provided (option ssh_identity is not set) for: $url";
    }
    %info = (
      %info,
      HOST         => $host,
      PATH         => $path,
      PRINT        => "{$host}$path",
      RSH_TYPE     => "ssh",
      SSH_USER     => $ssh_user,
      SSH_IDENTITY => $ssh_identity,
      SSH_PORT     => $ssh_port,
      RSH          => ['/usr/bin/ssh', @ssh_options, $ssh_user . '@' . $host ],
     );
  }
  elsif(($url =~ /^\//) && ($url =~ /^$file_match$/)) {
    %info = (
      %info,
      PATH   => $url,
      PRINT  => $url,
     );
  }
  else {
    die "Ambiguous vinfo url: $url";
  }

  my $btrfs_progs_compat = config_key($config, "btrfs_progs_compat");
  $info{BTRFS_PROGS_COMPAT} = $btrfs_progs_compat if($btrfs_progs_compat);

  TRACE "vinfo created: $url";
  return \%info;
}


sub vinfo_child($$)
{
  my $parent = shift || die;
  my $rel_path = shift // die;

  my $name = $rel_path;
  $name =~ s/^.*\///;
  my %info = (
    NAME         => $name,
    URL          => "$parent->{URL}/$rel_path",
    PATH         => "$parent->{PATH}/$rel_path",
    PRINT        => "$parent->{PRINT}/$rel_path",
    SUBVOL_PATH  => $rel_path,
   );
  foreach (qw( HOST
               RSH_TYPE
               SSH_USER
               SSH_IDENTITY
               SSH_PORT
               RSH
               BTRFS_PROGS_COMPAT ) )
  {
    $info{$_} = $parent->{$_} if(exists $parent->{$_});
  }

  TRACE "vinfo child created from \"$parent->{PRINT}\": $info{PRINT}";
  return \%info;
}


sub vinfo_root($)
{
  my $vol = shift;

  my $detail = btrfs_subvolume_detail($vol);
  return undef unless $detail;
  vinfo_set_detail($vol, $detail);

  # read (and cache) the subvolume list
  return undef unless vinfo_subvol_list($vol);

  TRACE "vinfo root created: $vol->{PRINT}";
  return $vol;
}


sub vinfo_set_detail($$)
{
  my $vol = shift || die;
  my $detail = shift || die;

  # add detail data to vinfo hash
  foreach(keys %$detail) {
    next if($_ eq "REL_PATH");
    next if($_ eq "TOP_LEVEL");
    next if($_ eq "SUBTREE");
    next if($_ eq "path");
    $vol->{$_} = $detail->{$_};
  }

  if($vol->{REAL_PATH}) {
    if($vol->{RSH_TYPE} && ($vol->{RSH_TYPE} eq "ssh")) {
      $vol->{REAL_URL} = "ssh://$vol->{HOST}$detail->{REAL_PATH}";
    } else {
      $vol->{REAL_URL} = $vol->{REAL_PATH};
    }
  }

  # update cache
  $vinfo_cache{$vol->{URL}} = $vol;
  $vinfo_cache{$vol->{REAL_URL}} = $vol if($vol->{REAL_URL});

  TRACE "vinfo updated for: $vol->{PRINT}";
  TRACE(Data::Dumper->Dump([$vol], ["vinfo{$vol->{PRINT}}"]));
  return $vol;
}


# returns hash: ( $prefix_{url,path,host,name,subvol_path,rsh} => value, ... )
sub vinfo_prefixed_keys($$)
{
  my $prefix = shift || die;
  my $vinfo = shift || die;
  my %ret;
  foreach (qw( URL PATH HOST NAME SUBVOL_PATH )) {
    $ret{$prefix . '_' . lc($_)} = $vinfo->{$_};
  }
  $ret{$prefix} = $vinfo->{PRINT};
  $ret{$prefix . "_rsh"} = ($vinfo->{RSH} ? join(" ", @{$vinfo->{RSH}}) : undef),
  return %ret;
}


sub config_key($$;@)
{
  my $node = shift || die;
  my $key = shift || die;
  my %opts = @_;
  TRACE "config_key: context=$node->{CONTEXT}, key=$key";
  while(not exists($node->{$key})) {
    # note: while all config keys exist in root context (at least with default values),
    #       we also allow fake configs (CONTEXT="cmdline") which have no PARENT.
    return undef unless($node->{PARENT});
    $node = $node->{PARENT};
  }
  TRACE "config_key: found value=" . ($node->{$key} // "<undef>");
  my $retval = $node->{$key};
  $retval = $opts{prefix} . $retval if(defined($opts{prefix}) && defined($retval));
  $retval .= $opts{postfix} if(defined($opts{postfix}) && defined($retval));
  return $retval;
}


sub config_dump_keys($;@)
{
  my $config = shift || die;
  my %opts = @_;
  my @ret;
  my $maxlen = 0;

  foreach my $key (sort keys %config_options)
  {
    my $val;
    if($opts{resolve}) {
      $val = config_key($config, $key);
    } else {
      next unless exists($config->{$key});
      $val = $config->{$key};
    }
    if($opts{skip_defaults}) {
      if(defined($config_options{$key}->{default}) && defined($val)) {
        next if($val eq $config_options{$key}->{default});
      }
      if((not defined($config_options{$key}->{default})) && (not (defined($val)))) {
        next; # both undef, skip
      }
    }
    if(ref($val) eq "ARRAY") {
      my $val2 = join(',', @$val);
      $val = $val2;
    }
    $val //= "<unset>";
    my $len = length($key);
    $maxlen = $len if($len > $maxlen);
    push @ret, { key => $key, val => $val, len => $len };
  }
  # print as table
  return map { ($opts{prefix} // "") . $_->{key} . (' ' x (1 + $maxlen - $_->{len})) . ' ' . $_->{val} } @ret;
}


sub check_file($$;$$)
{
  my $file = shift // die;
  my $accept = shift || die;
  my $key = shift;  # only for error text
  my $config_file = shift;  # only for error text

  if($accept->{ssh} && ($file =~ /^ssh:\/\//)) {
    unless($file =~ /^$ssh_prefix_match\/$file_match$/) {
      ERROR "Ambiguous ssh url for option \"$key\" in \"$config_file\" line $.: $file" if($key && $config_file);
      return undef;
    }
  }
  elsif($file =~ /^$file_match$/) {
    if($accept->{absolute}) {
      unless($file =~ /^\//) {
        ERROR "Only absolute files allowed for option \"$key\" in \"$config_file\" line $.: $file" if($key && $config_file);
        return undef;
      }
    }
    elsif($accept->{relative}) {
      if($file =~ /^\//) {
        ERROR "Only relative files allowed for option \"$key\" in \"$config_file\" line $.: $file" if($key && $config_file);
        return undef;
      }
    }
    elsif($accept->{name_only}) {
      if($file =~ /\//) {
        ERROR "Option \"$key\" is not a valid file name in \"$config_file\" line $.: $file" if($key && $config_file);
        return undef;
      }
    }
    else {
      die("accept_type must contain either 'relative' or 'absolute'");
    }
  }
  else {
    ERROR "Ambiguous file for option \"$key\" in \"$config_file\" line $.: $file" if($key && $config_file);
    return undef;
  }
  return 1;
}


sub parse_config(@)
{
  my @config_files = @_;
  my $file = undef;
  foreach(@config_files) {
    TRACE "config: checking for file: $_";
    if(-r "$_") {
      $file = $_;
      last;
    }
  }
  unless($file) {
    ERROR "Configuration file not found: " . join(', ', @config_files);
    return undef;
  }

  my $root = { CONTEXT => "root", SRC_FILE => $file };
  my $cur = $root;
  # set defaults
  foreach (keys %config_options) {
    next if $config_options{$_}->{deprecated};  # don't pollute hash with deprecated options
    $root->{$_} = $config_options{$_}->{default};
  }

  INFO "Using configuration: $file";
  open(FILE, '<', $file) or die $!;
  while (<FILE>) {
    chomp;
    next if /^\s*#/; # ignore comments
    next if /^\s*$/; # ignore empty lines
    TRACE "config: parsing line $. with context=$cur->{CONTEXT}: \"$_\"";
    if(/^(\s*)([a-zA-Z_]+)\s+(.*)$/)
    {
      my ($indent, $key, $value) = (length($1), lc($2), $3);
      $value =~ s/\s*$//;
      # NOTE: we do not perform checks on indentation!

      if($key eq "volume")
      {
        $cur = $root;
        TRACE "config: context forced to: $cur->{CONTEXT}";

        # be very strict about file options, for security sake
        return undef unless(check_file($value, { absolute => 1, ssh => 1 }, $key, $file));
        $value =~ s/\/+$// unless($value =~ /^\/+$/);   # remove trailing slash
        $value =~ s/^\/+/\//; # sanitize leading slash
        TRACE "config: adding volume \"$value\" to root context";
        my $volume = { CONTEXT => "volume",
                       PARENT => $cur,
                       url => $value,
                      };
        $cur->{VOLUME} //= [];
        push(@{$cur->{VOLUME}}, $volume);
        $cur = $volume;
      }
      elsif($key eq "subvolume")
      {
        while($cur->{CONTEXT} ne "volume") {
          if(($cur->{CONTEXT} eq "root") || (not $cur->{PARENT})) {
            ERROR "Subvolume keyword outside volume context, in \"$file\" line $.";
            return undef;
          }
          $cur = $cur->{PARENT} || die;
          TRACE "config: context changed to: $cur->{CONTEXT}";
        }
        # be very strict about file options, for security sake
        return undef unless(check_file($value, { relative => 1 }, $key, $file));
        $value =~ s/\/+$//;    # remove trailing slash
        $value =~ s/^\/+//;    # remove leading slash

        TRACE "config: adding subvolume \"$value\" to volume context: $cur->{url}";
        my $snapshot_name = $value;
        $snapshot_name =~ s/^.*\///; # snapshot_name defaults to subvolume name
        my $subvolume = { CONTEXT => "subvolume",
                          PARENT => $cur,
                          rel_path => $value,
                          url => $cur->{url} . '/' . $value,
                          snapshot_name => $snapshot_name,
                         };
        $cur->{SUBVOLUME} //= [];
        push(@{$cur->{SUBVOLUME}}, $subvolume);
        $cur = $subvolume;
      }
      elsif($key eq "target")
      {
        if($cur->{CONTEXT} eq "target") {
          $cur = $cur->{PARENT} || die;
          TRACE "config: context changed to: $cur->{CONTEXT}";
        }
        if($cur->{CONTEXT} ne "subvolume") {
          ERROR "Target keyword outside subvolume context, in \"$file\" line $.";
          return undef;
        }
        if($value =~ /^(\S+)\s+(\S+)$/)
        {
          my ($target_type, $droot) = ($1, $2);
          unless(grep(/^$target_type$/, @config_target_types)) {
            ERROR "Unknown target type \"$target_type\" in \"$file\" line $.";
            return undef;
          }
          # be very strict about file options, for security sake
          return undef unless(check_file($droot, { absolute => 1, ssh => 1 }, $key, $file));

          $droot =~ s/\/+$//;   # remove trailing slash
          $droot =~ s/^\/+/\//; # sanitize leading slash
          TRACE "config: adding target \"$droot\" (type=$target_type) to subvolume context: $cur->{url}";
          my $target = { CONTEXT => "target",
                         PARENT => $cur,
                         target_type => $target_type,
                         url => $droot,
                       };
          $cur->{TARGET} //= [];
          push(@{$cur->{TARGET}}, $target);
          $cur = $target;
        }
        else
        {
          ERROR "Ambiguous target configuration, in \"$file\" line $.";
          return undef;
        }
      }
      elsif(grep(/^$key$/, keys %config_options))  # accept only keys listed in %config_options
      {
        if(grep(/^$value$/, @{$config_options{$key}->{accept}})) {
          TRACE "option \"$key=$value\" found in accept list";
        }
        elsif($config_options{$key}->{accept_numeric} && ($value =~ /^[0-9]+$/)) {
          TRACE "option \"$key=$value\" is numeric, accepted";
        }
        elsif($config_options{$key}->{accept_file})
        {
          # be very strict about file options, for security sake
          return undef unless(check_file($value, $config_options{$key}->{accept_file}, $key, $file));

          TRACE "option \"$key=$value\" is a valid file, accepted";
          $value =~ s/\/+$//;   # remove trailing slash
          $value =~ s/^\/+/\//; # sanitize leading slash
        }
        elsif($config_options{$key}->{accept_regexp}) {
          my $match = $config_options{$key}->{accept_regexp};
          if($value =~ m/$match/) {
            TRACE "option \"$key=$value\" matched regexp, accepted";
          }
          else {
            ERROR "Value \"$value\" failed input validation for option \"$key\" in \"$file\" line $.";
            return undef;
          }
        }
        else
        {
          ERROR "Unsupported value \"$value\" for option \"$key\" in \"$file\" line $.";
          return undef;
        }

        if($config_options{$key}->{split}) {
          $value = [ split($config_options{$key}->{split}, $value) ];
          TRACE "splitted option \"$key\": " . join(',', @$value);
        }

        if($config_options{$key}->{context} && !grep(/^$cur->{CONTEXT}$/, @{$config_options{$key}->{context}})) {
          ERROR "Option \"$key\" is only allowed in " . join(" or ", map("\"$_\"", @{$config_options{$key}->{context}})) . " context, in \"$file\" line $.";
          return undef;
        }

        if($config_options{$key}->{deprecated}) {
          WARN "Found deprecated option \"$key $value\" in \"$file\" line $.: " .
               ($config_options{$key}->{deprecated}->{$value}->{warn} // $config_options{$key}->{deprecated}->{DEFAULT}->{warn});
          my $replace_key   = $config_options{$key}->{deprecated}->{$value}->{replace_key};
          my $replace_value = $config_options{$key}->{deprecated}->{$value}->{replace_value};
          if(defined($replace_key)) {
            $key = $replace_key;
            $value = $replace_value;
            WARN "Using \"$key $value\"";
          }
        }

        TRACE "config: adding option \"$key=$value\" to $cur->{CONTEXT} context";
        $value = undef if($value eq "no");  # we don't want to check for "no" all the time
        $cur->{$key} = $value;
      }
      else
      {
        ERROR "Unknown option \"$key\" in \"$file\" line $.";
        return undef;
      }

      TRACE "line processed: new context=$cur->{CONTEXT}";
    }
    else
    {
      ERROR "Parse error in \"$file\" line $.";
      return undef;
    }
  }

  TRACE(Data::Dumper->Dump([$root], ["config{$file}"]));
  return $root;
}


sub btrfs_filesystem_show_all_local()
{
  return run_cmd( cmd => [ qw(btrfs filesystem show) ],
                  non_destructive => 1
                 );
}


sub btrfs_filesystem_show($)
{
  my $vol = shift || die;
  my $path = $vol->{PATH} // die;
  return run_cmd( cmd => [ qw(btrfs filesystem show), $path ],
                  rsh => $vol->{RSH},
                  non_destructive => 1
                 );
}


sub btrfs_filesystem_df($)
{
  my $vol = shift || die;
  my $path = $vol->{PATH} // die;
  return run_cmd( cmd => [qw(btrfs filesystem df), $path],
                  rsh => $vol->{RSH},
                  non_destructive => 1
                 );
}


sub btrfs_filesystem_usage($)
{
  my $vol = shift || die;
  my $path = $vol->{PATH} // die;
  return run_cmd( cmd => [ qw(btrfs filesystem usage), $path ],
                  rsh => $vol->{RSH},
                  non_destructive => 1
                 );
}


sub btrfs_subvolume_detail($)
{
  my $vol = shift || die;
  my $path = $vol->{PATH} // die;
  my $ret = run_cmd(cmd => [ qw(btrfs subvolume show), $path],
                    rsh => $vol->{RSH},
                    non_destructive => 1,
                    catch_stderr => 1, # hack for shell-based run_cmd()
                    filter_stderr => sub {
                      if(/ssh command rejected/) {
                        # catch errors from ssh_filter_btrbk.sh
                        $err = "ssh command rejected (please fix ssh_filter_btrbk.sh)";
                      }
                      elsif(/^ERROR: (.*)/) {
                        # catch errors from btrfs command
                        $err = $1;
                      }
                      else {
                        DEBUG "Unparsed error: $_";
                        $err = $_;
                      }
                      # consume stderr line, as $err will be displayed as a user-friendly WARNING
                      $_ = undef;
                    }
                   );

  return undef unless(defined($ret));

  # workaround for btrfs-progs < 3.17.3 (returns exit status 0 on errors)
  if($ret =~ /^ERROR: (.*)/) {
    $err = $1;
    return undef;
  }

  my $real_path;
  if($ret =~ /^($file_match)/) {
    $real_path = $1;
    DEBUG "Real path for subvolume \"$vol->{PRINT}\" is: $real_path" if($real_path ne $path);
    return undef unless(check_file($real_path, { absolute => 1 }));
  }
  else {
    $real_path = $path;
    WARN "No real path provided by \"btrfs subvolume show\" for subvolume \"$vol->{PRINT}\", using: $path";
  }
  my %detail = ( REAL_PATH  => $real_path );

  if($ret =~ /^\Q$real_path\E is btrfs root/) {
    DEBUG "found btrfs root: $vol->{PRINT}";
    $detail{id}      = 5;
    $detail{is_root} = 1;
  }
  elsif($ret =~ /^$real_path/) {
    TRACE "btr_detail: found btrfs subvolume: $vol->{PRINT}";
    my %trans = (
      "Name"                  => "name",
      "uuid"                  => "uuid",
      "UUID"                  => "uuid",            # btrfs-progs >= 4.1
      "Parent uuid"           => "parent_uuid",
      "Parent UUID"           => "parent_uuid",     # btrfs-progs >= 4.1
      "Received UUID"         => "received_uuid",   # btrfs-progs >= 4.1
      "Creation time"         => "creation_time",
      "Object ID"             => "id",
      "Subvolume ID"          => "id",              # btrfs-progs >= 4.1
      "Generation (Gen)"      => "gen",
      "Generation"            => "gen",             # btrfs-progs >= 4.1
      "Gen at creation"       => "cgen",
      "Parent"                => "parent_id",
      "Parent ID"             => "parent_id",       # btrfs-progs >= 4.1
      "Top Level"             => "top_level",
      "Top level ID"          => "top_level",       # btrfs-progs >= 4.1
      "Flags"                 => "flags",
     );
    foreach (split("\n", $ret)) {
      next unless /^\s+(.+):\s+(.*)$/;
      my ($key, $value) = ($1, $2);
      if($trans{$key}) {
        $detail{$trans{$key}} = $value;
      } else {
        WARN "Failed to parse subvolume detail \"$key: $value\" for: $vol->{PRINT}";
      }
    }
    DEBUG "Parsed " . scalar(keys %detail) . " subvolume detail items: $vol->{PRINT}";
    TRACE(Data::Dumper->Dump([$vol], ["btrfs_subvolume_detail($vol->{URL})"]));
  }
  return \%detail;
}


sub btrfs_subvolume_list($;@)
{
  my $vol = shift || die;
  my %opts = @_;
  my $path = $vol->{PATH} // die;
  my $btrfs_progs_compat = $vol->{BTRFS_PROGS_COMPAT} || $opts{btrfs_progs_compat};
  my @filter_options = ('-a');
  push(@filter_options, '-o') if($opts{subvol_only});
  my @display_options = ('-c', '-u', '-q');
  push(@display_options, '-R') unless($btrfs_progs_compat);
  my $ret = run_cmd(cmd => [ qw(btrfs subvolume list), @filter_options, @display_options, $path ],
                    rsh => $vol->{RSH},
                    non_destructive => 1,
                   );
  return undef unless(defined($ret));

  my @nodes;
  foreach (split(/\n/, $ret))
  {
    # ID <ID> top level <ID> path <path> where path is the relative path
    # of the subvolume to the top level subvolume. The subvolume?s ID may
    # be used by the subvolume set-default command, or at mount time via
    # the subvolid= option. If -p is given, then parent <ID> is added to
    # the output between ID and top level. The parent?s ID may be used at
    # mount time via the subvolrootid= option.

    # NOTE: btrfs-progs prior to v1.17 do not support the -R flag
    my %node;
    if($btrfs_progs_compat) {
      die("Failed to parse line: \"$_\"") unless(/^ID ([0-9]+) gen ([0-9]+) cgen ([0-9]+) top level ([0-9]+) parent_uuid ([0-9a-z-]+) uuid ([0-9a-z-]+) path (.+)$/);
      %node = (
        id            => $1,
        gen           => $2,
        cgen          => $3,
        top_level     => $4,
        parent_uuid   => $5, # note: parent_uuid="-" if no parent
        # received_uuid => $6,
        uuid          => $6,
        path          => $7  # btrfs path, NOT filesystem path
     );
    } else {
      die("Failed to parse line: \"$_\"") unless(/^ID ([0-9]+) gen ([0-9]+) cgen ([0-9]+) top level ([0-9]+) parent_uuid ([0-9a-z-]+) received_uuid ([0-9a-z-]+) uuid ([0-9a-z-]+) path (.+)$/);
      %node = (
      id            => $1,
      gen           => $2,
      cgen          => $3,
      top_level     => $4,
      parent_uuid   => $5, # note: parent_uuid="-" if no parent
      received_uuid => $6,
      uuid          => $7,
      path          => $8  # btrfs path, NOT filesystem path
     );
    }

    # NOTE: "btrfs subvolume list <path>" prints <FS_TREE> prefix only if
    # the subvolume is reachable within <path>. (as of btrfs-progs-3.18.2)
    #
    # NOTE: Be prepared for this to change in btrfs-progs!
    $node{path} =~ s/^<FS_TREE>\///;     # remove "<FS_TREE>/" portion from "path".

    push @nodes, \%node;
  }
  DEBUG "Parsed " . scalar(@nodes) . " total subvolumes for filesystem at: $vol->{PRINT}";
  return \@nodes;
}


sub btrfs_subvolume_find_new($$;$)
{
  my $vol = shift || die;
  my $path = $vol->{PATH} // die;
  my $lastgen = shift // die;
  my $ret = run_cmd(cmd => [ qw(btrfs subvolume find-new), $path, $lastgen ],
                    rsh => $vol->{RSH},
                    non_destructive => 1,
                   );
  unless(defined($ret)) {
    ERROR "Failed to fetch modified files for: $vol->{PRINT}";
    return undef;
  }

  my %files;
  my $parse_errors = 0;
  my $transid_marker;
  foreach (split(/\n/, $ret))
  {
    if(/^inode \S+ file offset (\S+) len (\S+) disk start \S+ offset \S+ gen (\S+) flags (\S+) (.+)$/) {
      my $file_offset = $1;
      my $len         = $2;
      my $gen         = $3;
      my $flags       = $4;
      my $name        = $5;
      $files{$name}->{len} += $len;
      $files{$name}->{new} = 1 if($file_offset == 0);
      $files{$name}->{gen}->{$gen} = 1;  # count the generations
      if($flags eq "COMPRESS") {
        $files{$name}->{flags}->{compress} = 1;
      }
      elsif($flags eq "COMPRESS|INLINE") {
        $files{$name}->{flags}->{compress} = 1;
        $files{$name}->{flags}->{inline} = 1;
      }
      elsif($flags eq "INLINE") {
        $files{$name}->{flags}->{inline} = 1;
      }
      elsif($flags eq "NONE") {
      }
      else {
        WARN "unparsed flags: $flags";
      }
    }
    elsif(/^transid marker was (\S+)$/) {
      $transid_marker = $1;
    }
    else {
      $parse_errors++;
    }
  }

  return { files => \%files,
           transid_marker => $transid_marker,
           parse_errors => $parse_errors,
          };
}


# returns $target, or undef on error
sub btrfs_subvolume_snapshot($$)
{
  my $svol = shift || die;
  my $target_vol = shift // die;
  my $target_path = $target_vol->{PATH} // die;
  my $src_path = $svol->{PATH} // die;
  DEBUG "[btrfs] snapshot (ro):";
  DEBUG "[btrfs]   host  : $svol->{HOST}" if($svol->{HOST});
  DEBUG "[btrfs]   source: $src_path";
  DEBUG "[btrfs]   target: $target_path";
  INFO ">>> $target_vol->{PRINT}";
  my $ret = run_cmd(cmd => [ qw(btrfs subvolume snapshot), '-r', $src_path, $target_path ],
                    rsh => $svol->{RSH},
                   );
  action("snapshot",
         status => ($dryrun ? "DRYRUN" : (defined($ret) ? "success" : "ERROR")),
         vinfo_prefixed_keys("target", $target_vol),
         vinfo_prefixed_keys("source", $svol),
        );
  ERROR "Failed to create btrfs subvolume snapshot: $svol->{PRINT} -> $target_path" unless(defined($ret));
  return defined($ret) ? $target_path : undef;
}


sub btrfs_subvolume_delete($@)
{
  my $targets = shift // die;
  my %opts = @_;
  my $commit = $opts{commit};
  die if($commit && ($commit ne "after") && ($commit ne "each"));
  $targets = [ $targets ] unless(ref($targets) eq "ARRAY");
  return 0 unless(scalar(@$targets));
  my $rsh = $targets->[0]->{RSH};
  my $rsh_host_check = $targets->[0]->{HOST} || "";
  foreach (@$targets) {
    # make sure all targets share same HOST
    my $host = $_->{HOST} || "";
    die if($rsh_host_check ne $host);
  }
  DEBUG "[btrfs] delete" . ($commit ? " (commit-$commit):" : ":");
  DEBUG "[btrfs]   subvolume: $_->{PRINT}" foreach(@$targets);
  my @options;
  @options = ("--commit-$commit") if($commit);
  my @target_paths = map( { $_->{PATH} } @$targets);
  my $ret = run_cmd(cmd => [ qw(btrfs subvolume delete), @options, @target_paths ],
                    rsh => $rsh,
                   );
  action($opts{type} // "delete",
         status => ($dryrun ? "DRYRUN" : (defined($ret) ? "success" : "ERROR")),
         vinfo_prefixed_keys("target", $_),
        ) foreach (@$targets);
  ERROR "Failed to delete btrfs subvolumes: " . join(' ', map( { $_->{PRINT} } @$targets)) unless(defined($ret));
  return defined($ret) ? scalar(@$targets) : undef;
}


sub btrfs_send_receive($$$$)
{
  my $snapshot = shift || die;
  my $target = shift || die;
  my $parent = shift;
  my $ret_vol_received = shift;
  my $snapshot_path = $snapshot->{PATH} // die;
  my $target_path = $target->{PATH} // die;
  my $parent_path = $parent ? $parent->{PATH} : undef;

  my $vol_received = vinfo_child($target, $snapshot->{NAME});
  $$ret_vol_received = $vol_received if(ref $ret_vol_received);

  INFO ">>> $vol_received->{PRINT}";
  print STDOUT "Receiving subvol: $vol_received->{PRINT}\n" if($show_progress && (not $dryrun));

  DEBUG "[btrfs] send/receive" . ($parent ? " (incremental)" : " (complete)") . ":";
  DEBUG "[btrfs]   source: $snapshot->{PRINT}";
  DEBUG "[btrfs]   parent: $parent->{PRINT}" if($parent);
  DEBUG "[btrfs]   target: $target->{PRINT}";

  my @send_options;
  my @receive_options;
  push(@send_options, '-p', $parent_path) if($parent_path);
  push(@send_options, '-v') if($loglevel >= 3);
  push(@receive_options, '-v') if($loglevel >= 3);

  my @cmd_pipe;
  push @cmd_pipe, {
    cmd => [ qw(btrfs send), @send_options, $snapshot_path ],
    rsh => $snapshot->{RSH},
    name => "btrfs send",
  };
  push @cmd_pipe, {
    cmd => [ '/usr/bin/pv' ],
  } if($show_progress);
  push @cmd_pipe, {
    cmd => [ qw(btrfs receive), @receive_options, $target_path . '/' ],
    rsh => $target->{RSH},
    name => "btrfs receive",
  };
  my $ret = run_cmd(@cmd_pipe);
  unless(defined($ret)) {
    ERROR "Failed to send/receive btrfs subvolume: $snapshot->{PRINT} " . ($parent_path ? "[$parent_path]" : "") . " -> $target->{PRINT}";

    # NOTE: btrfs-progs v3.19.1 does not delete garbled received subvolume,
    #       we need to do this by hand.
    # TODO: remove this as soon as btrfs-progs handle receive errors correctly.
    DEBUG "send/received failed, deleting (possibly present and garbled) received subvolume: $vol_received->{PRINT}";
    my $ret = btrfs_subvolume_delete($vol_received, commit => "after", type => "delete_garbled");
    if(defined($ret)) {
      WARN "Deleted partially received (garbled) subvolume: $vol_received->{PRINT}";
    }
    else {
      WARN "Deletion of partially received (garbled) subvolume failed, assuming clean environment: $vol_received->{PRINT}";
    }

    return undef;
  }
  return 1;
}


sub btrfs_send_to_file($$$$;@)
{
  my $snapshot = shift || die;
  my $target = shift || die;
  my $parent = shift;
  my $ret_vol_received = shift;
  my %opts = @_;
  my $snapshot_path = $snapshot->{PATH} // die;
  my $target_path   = $target->{PATH} // die;
  my $parent_path   = $parent ? $parent->{PATH} : undef;
  my $parent_uuid   = $parent ? $parent->{uuid} : undef ;
  my $received_uuid = $snapshot->{uuid};
  $received_uuid = "XXXXXXXX-XXXX-XXXX-XXXX-XXXXXXXXXXXX" if((not $received_uuid) && $dryrun);
  die unless($received_uuid);
  die if($parent && !$parent_uuid);

  my $target_filename = $snapshot->{NAME} || die;
  $target_filename .= "--$received_uuid";
  $target_filename .= '@' . $parent_uuid if($parent_uuid);
  $target_filename .= ".btrfs";

  my %compress = ( gzip  => { pipe => { cmd => [ 'gzip'  ], name => 'gzip'  }, postfix => '.gz'  },
                   bzip2 => { pipe => { cmd => [ 'bzip2' ], name => 'bzip2' }, postfix => '.bz2' },
                   xz    => { pipe => { cmd => [ 'xz'    ], name => 'xz'    }, postfix => '.xz'  },
                  );

  my @send_options;
  push(@send_options, '-v') if($loglevel >= 3);
  push(@send_options, '-p', $parent_path) if($parent_path);

  my @cmd_pipe;
  push @cmd_pipe, {
    cmd => [ qw(btrfs send), @send_options, $snapshot_path ],
    rsh => $snapshot->{RSH},
    name => "btrfs send",
  };
  push @cmd_pipe, {
    cmd => [ '/usr/bin/pv' ],
  } if($show_progress);
  if($opts{compress}) {
    die unless($compress{$opts{compress}});
    $target_filename .= $compress{$opts{compress}}->{postfix};
    push @cmd_pipe, $compress{$opts{compress}}->{pipe};
  }
  if($opts{encrypt}) {
    die unless($opts{encrypt}->{type} eq "gpg");
    $target_filename .= '.gpg';
    my @gpg_options = ( '--batch', '--no-tty', '--trust-model', 'always' );
    push(@gpg_options, ( '--no-default-keyring', '--keyring', $opts{encrypt}->{keyring} )) if($opts{encrypt}->{keyring});
    push(@gpg_options, ( '--default-recipient', $opts{encrypt}->{recipient} )) if($opts{encrypt}->{recipient});
    push @cmd_pipe, {
      cmd => [ 'gpg', @gpg_options, '--encrypt' ],
      name => 'gpg',
    };
  }
  push @cmd_pipe, {
    cmd => [ 'dd', 'status=none', "of=$target_path/$target_filename" ],
    rsh => $target->{RSH},
    name => 'dd',
  };

  my $vol_received = vinfo_child($target, $target_filename);
  $$ret_vol_received = $vol_received if(ref $ret_vol_received);

  INFO ">>> $vol_received->{PRINT}";
  print STDOUT "Receiving subvol (raw): $vol_received->{PRINT}\n" if($show_progress && (not $dryrun));

  DEBUG "[btrfs] send-to-file" . ($parent ? " (incremental)" : " (complete)") . ":";
  DEBUG "[btrfs]   source: $snapshot->{PRINT}";
  DEBUG "[btrfs]   parent: $parent->{PRINT}" if($parent);
  DEBUG "[btrfs]   target: $target->{PRINT}";

  my $ret = run_cmd(@cmd_pipe);
  if(defined($ret)) {
    # Test target file for "exists and size > 0" after writing,
    # as we can not rely on the exit status of 'dd'
    DEBUG "Testing target file (non-zero size): $target->{PRINT}";
    $ret = run_cmd({
      cmd => ['test', '-s', "$target_path/$target_filename"],
      rsh => $target->{RSH},
      name => "test",
    });
  }
  unless(defined($ret)) {
    ERROR "Failed to send btrfs subvolume to raw file: $snapshot->{PRINT} " . ($parent_path ? "[$parent_path]" : "") . " -> $vol_received->{PRINT}";
    return undef;
  }
  return 1;
}


sub btr_tree($)
{
  my $vol = shift;

  # return cached info if present
  return $root_tree_cache{$vol->{URL}} if($vol->{is_root} && $root_tree_cache{$vol->{URL}});
  return $root_tree_cache{$vol->{REAL_URL}} if($vol->{is_root} && $vol->{REAL_URL} && $root_tree_cache{$vol->{REAL_URL}});
  return $uuid_info{$vol->{uuid}} if($vol->{uuid} && $uuid_info{$vol->{uuid}});

  # man btrfs-subvolume:
  #   Also every btrfs filesystem has a default subvolume as its initially
  #   top-level subvolume, whose subvolume id is 5(FS_TREE).
  my %tree = ( id => 5, SUBTREE => {} );
  my %id = ( 5 => \%tree );

  my $subvol_list = btrfs_subvolume_list($vol);
  return undef unless(ref($subvol_list) eq "ARRAY");

  TRACE "btr_tree: processing subvolume list of: $vol->{PRINT}";

  foreach my $node (@$subvol_list)
  {
    $node->{SUBTREE} //= {};

    $id{$node->{id}} = $node;
    $uuid_info{$node->{uuid}} = $node;
  }

  # note: it is possible that id < top_level, e.g. after restoring
  foreach my $node (@$subvol_list)
  {
    # set SUBTREE / TOP_LEVEL node
    die unless exists($id{$node->{top_level}});
    my $top_level = $id{$node->{top_level}};

    die if exists($top_level->{SUBTREE}->{$node->{id}});
    $top_level->{SUBTREE}->{$node->{id}} = $node;
    $node->{TOP_LEVEL} = $top_level;

    # "path" always starts with set REL_PATH
    my $rel_path = $node->{path};
    if($node->{top_level} != 5) {
      die unless($rel_path =~ s/^$top_level->{path}\///);
    }

    $node->{REL_PATH} = $rel_path;  # relative to {TOP_LEVEL}->{path}
  }

  if($vol->{is_root}) {
    $root_tree_cache{$vol->{URL}} = \%tree;
    $root_tree_cache{$vol->{REAL_URL}} = \%tree if($vol->{REAL_URL});
    return \%tree;
  }
  else {
    die unless($uuid_info{$vol->{uuid}});
    return $uuid_info{$vol->{uuid}};
  }
}


sub _subtree_list
{
  my $tree = shift;
  my $list = shift // [];
  my $prefix = shift // "";

  $tree = $tree->{SUBTREE};
  foreach(values %$tree) {
    my $path = $prefix . $_->{REL_PATH};
    push(@$list, { SUBVOL_PATH => $path,
                   node    => $_,
                 });

    _subtree_list($_, $list, $path . '/');
  }
  return $list;
}


sub vinfo_subvol_list($)
{
  my $vol = shift || die;
  return $vol->{SUBVOL_LIST} if($vol->{SUBVOL_LIST});

  my $tree_root = btr_tree($vol);
  return undef unless($tree_root);

  # recurse into $tree_root, returns list of href: { SUBVOL_PATH, node }
  my $list = _subtree_list($tree_root);

  # return a hash of relative subvolume path
  my %ret;
  foreach(@$list) {
    my $subvol_path = $_->{SUBVOL_PATH};
    die if exists $ret{$subvol_path};

    my $subvol = vinfo_child($vol, $subvol_path);
    vinfo_set_detail($subvol, $_->{node});

    $uuid_fs_map{$subvol->{uuid}}->{$subvol->{URL}} = $subvol;

    $ret{$subvol_path} = $subvol;
  }

  DEBUG "Found " . scalar(keys %ret) . " subvolume children of: $vol->{PRINT}";
  TRACE(Data::Dumper->Dump([\%ret], ["vinfo_subvol_list{$vol->{URL}}"]));

  $vol->{SUBVOL_LIST} = \%ret;
  return \%ret;
}


# returns list of uuids for ALL subvolumes in the btrfs filesystem of $vol
sub vinfo_fs_list($)
{
  my $vol = shift || die;
  my $tree_root = btr_tree($vol);
  return undef unless($tree_root);

  $tree_root = $tree_root->{TOP_LEVEL} while($tree_root->{TOP_LEVEL});
  my $list = _subtree_list($tree_root);
  my %ret = map { $_->{node}->{uuid} => $_->{node} } @$list;
  return \%ret;
}


sub vinfo_subvol($$)
{
  my $vol = shift || die;
  my $rel_path = shift // die;

  my $subvols = vinfo_subvol_list($vol);
  return $subvols->{$rel_path};
}


# sets $config->{ABORTED} on failure
# sets $config->{SUBVOL_RECEIVED}
sub macro_send_receive($@)
{
  my $config_target = shift || die;
  my %info = @_;
  my $snapshot = $info{snapshot} || die;
  my $target = $info{target} || die;
  my $parent = $info{parent};
  my $target_type = $config_target->{target_type} || die;
  my $incremental = config_key($config_target, "incremental");

  INFO "Receiving from snapshot: $snapshot->{PRINT}";

  # check for existing target subvolume
  if(my $err_vol = vinfo_subvol($target, $snapshot->{NAME})) {
    ABORTED($config_target, "Target subvolume \"$err_vol->{PRINT}\" already exists");
    $config_target->{UNRECOVERABLE} = "Please delete stray subvolume: $err_vol->{PRINT}";
    ERROR $config_target->{ABORTED} . ", aborting send/receive of: $snapshot->{PRINT}";
    ERROR $config_target->{UNRECOVERABLE};
    $info{ERROR} = 1;
    return undef;
  }

  if($incremental)
  {
    # create backup from latest common
    if($parent) {
      INFO "Incremental from parent snapshot: $parent->{PRINT}";
    }
    elsif($incremental ne "strict") {
      INFO "No common parent subvolume present, creating full backup";
    }
    else {
      WARN "Backup to $target->{PRINT} failed: no common parent subvolume found, and option \"incremental\" is set to \"strict\"";
      $info{ERROR} = 1;
      ABORTED($config_target, "No common parent subvolume found, and option \"incremental\" is set to \"strict\"");
      return undef;
    }
  }
  else {
    INFO "Option \"incremental\" is not set, creating full backup";
    $parent = undef;
    delete $info{parent};
  }

  my $ret;
  my $vol_received;
  if($target_type eq "send-receive")
  {
    $ret = btrfs_send_receive($snapshot, $target, $parent, \$vol_received);
    ABORTED($config_target, "Failed to send/receive subvolume") unless($ret);
  }
  elsif($target_type eq "raw")
  {
    unless($dryrun) {
      # make sure we know the snapshot uuid
      unless($snapshot->{uuid}) {
        DEBUG "Fetching uuid of new snapshot: $snapshot->{PRINT}";
        my $detail = btrfs_subvolume_detail($snapshot);
        die unless($detail->{uuid});
        vinfo_set_detail($snapshot, { uuid => $detail->{uuid} });
      }
    }

    my $compress = config_key($config_target, "raw_target_compress");
    my $encrypt = undef;
    my $encrypt_type = config_key($config_target, "raw_target_encrypt");
    if($encrypt_type) {
      die unless($encrypt_type eq "gpg");
      $encrypt = { type => $encrypt_type,
                   keyring => config_key($config_target, "gpg_keyring"),
                   recipient => config_key($config_target, "gpg_recipient"),
                 }
    }
    $ret = btrfs_send_to_file($snapshot, $target, $parent, \$vol_received, compress => $compress, encrypt => $encrypt);
    ABORTED($config_target, "Failed to send subvolume to raw file") unless($ret);
  }
  else
  {
    die "Illegal target type \"$target_type\"";
  }

  # add info to $config->{SUBVOL_RECEIVED}
  $info{received_type} = $target_type || die;
  $info{received_subvolume} = $vol_received || die;
  $config_target->{SUBVOL_RECEIVED} //= [];
  push(@{$config_target->{SUBVOL_RECEIVED}}, \%info);

  action("send-receive",
         status => ($dryrun ? "DRYRUN" : ($ret ? "success" : "ERROR")),
         vinfo_prefixed_keys("target", $vol_received),
         vinfo_prefixed_keys("source", $snapshot),
         vinfo_prefixed_keys("parent", $parent),
        );
  unless($ret) {
    $info{ERROR} = 1;
    return undef;
  }
  return 1;
}


# returns { btrbk_date => [ yyyy, mm, dd, hh, mm, <date_ext> ] } or undef
# fixed array length of 6, all individually defaulting to 0
sub parse_filename($$;$)
{
  my $file = shift;
  my $name_match = shift;
  my $raw_format = shift || 0;
  my %raw_info;
  if($raw_format)
  {
    return undef unless($file =~ /^\Q$name_match\E$timestamp_postfix_match$raw_postfix_match$/);
    die unless($+{YYYY} && $+{MM} && $+{DD});
    return { btrbk_date    => [ $+{YYYY}, $+{MM}, $+{DD}, ($+{hh} // 0), ($+{mm} // 0), ($+{NN} // 0) ],
             received_uuid => $+{received_uuid} // die,
             parent_uuid   => $+{parent_uuid} // '-',
             ENCRYPT       => $+{encrypt} // "",
             COMPRESS      => $+{compress} // "",
            };
  }
  else
  {
    return undef unless($file =~ /^\Q$name_match\E$timestamp_postfix_match$/);
    die unless($+{YYYY} && $+{MM} && $+{DD});
    return { btrbk_date => [ $+{YYYY}, $+{MM}, $+{DD}, ($+{hh} // 0), ($+{mm} // 0), ($+{NN} // 0) ] };
  }
}


sub get_snapshot_children($$)
{
  my $sroot = shift || die;
  my $svol = shift // die;
  my @ret;

  my $sroot_subvols = vinfo_subvol_list($sroot);
  foreach (values %$sroot_subvols) {
    next unless($_->{parent_uuid} eq $svol->{uuid});
    TRACE "get_snapshot_children: found: $_->{PRINT}";
    push(@ret, $_);
  }
  DEBUG "Found " . scalar(@ret) . " snapshot children of: $svol->{PRINT}";
  return @ret;
}


sub get_receive_targets($$)
{
  my $droot = shift || die;
  my $src_vol = shift || die;
  my $droot_subvols = vinfo_subvol_list($droot);
  my @ret;

  if($droot->{BTRFS_PROGS_COMPAT})
  {
    # guess matches by subvolume name (node->received_uuid is not available if BTRFS_PROGS_COMPAT is set)
    DEBUG "Fallback to compatibility mode (get_receive_targets)";
    foreach my $target (values %$droot_subvols) {
      if($target->{NAME} eq $src_vol->{NAME}) {
        TRACE "get_receive_targets: by-name: Found receive target: $target->{SUBVOL_PATH}";
        push(@ret, $target);
      }
    }
  }
  else
  {
    # find matches by comparing uuid / received_uuid
    my $uuid = $src_vol->{uuid};
    die("subvolume info not present: $uuid") unless($uuid_info{$uuid});
    foreach (values %$droot_subvols) {
      next unless($_->{received_uuid} eq $uuid);
      TRACE "get_receive_targets: by-uuid: Found receive target: $_->{SUBVOL_PATH}";
      push(@ret, $_);
    }
  }
  DEBUG "Found " . scalar(@ret) . " receive targets in \"$droot->{PRINT}/\" for: $src_vol->{PRINT}";
  return @ret;
}


sub get_latest_common($$$;$)
{
  my $sroot = shift || die;
  my $svol = shift // die;
  my $droot = shift || die;
  my $threshold_gen = shift;  # skip all snapshot children with generation (cgen) >= $threshold_gen

  die("source subvolume info not present: $sroot->{URL}") unless($sroot->{URL});
  die("target subvolume info not present: $droot->{URL}") unless($droot->{URL});

  my $debug_src = $svol->{URL};
  $debug_src .= "#" . $threshold_gen if($threshold_gen);

  # sort children of svol descending by generation
  foreach my $child (sort { $b->{cgen} <=> $a->{cgen} } get_snapshot_children($sroot, $svol)) {
    TRACE "get_latest_common: checking source snapshot: $child->{SUBVOL_PATH}";
    if($threshold_gen && ($child->{cgen} >= $threshold_gen)) {
      TRACE "get_latest_common: skipped gen=$child->{cgen} >= $threshold_gen: $child->{SUBVOL_PATH}";
      next;
    }

    if($child->{RECEIVE_TARGET_PRESENT} && ($child->{RECEIVE_TARGET_PRESENT} eq $droot->{URL})) {
      # little hack to keep track of previously received subvolumes
      DEBUG("Latest common snapshots for: $debug_src: src=$child->{PRINT}  target=<previously received>");
      return ($child, undef);
    }

    foreach (get_receive_targets($droot, $child)) {
      TRACE "get_latest_common: found receive target: $_->{PRINT}";
      DEBUG("Latest common snapshots for: $debug_src: src=$child->{PRINT}  target=$_->{PRINT}");
      return ($child, $_);
    }
    TRACE "get_latest_common: no matching targets found for: $child->{PRINT}";
  }
  DEBUG("No common snapshots of \"$debug_src\" found in src=\"$sroot->{PRINT}/\", target=\"$droot->{PRINT}/\"");
  return (undef, undef);
}


sub get_latest_snapshot_child($$)
{
  my $sroot = shift || die;
  my $svol = shift // die;
  my $latest = undef;
  my $gen = -1;
  foreach (get_snapshot_children($sroot, $svol)) {
    if($_->{cgen} > $gen) {
      $latest = $_;
      $gen = $_->{cgen};
    }
  }
  if($latest) {
    DEBUG "Latest snapshot child for \"$svol->{PRINT}#$svol->{gen}\" is: $latest->{PRINT}#$latest->{cgen}";
  } else {
    DEBUG "No latest snapshots found for: $svol->{PRINT}";
  }
  return $latest;
}


sub _origin_tree
{
  my $prefix = shift;
  my $uuid = shift;
  my $lines = shift;
  my $node = $uuid_info{$uuid};
  unless($node) {
    push(@$lines, ["$prefix<orphaned>", $uuid]);
    return 0;
  }
  if($uuid_fs_map{$uuid}) {
    push(@$lines, ["$prefix" . join(" === ", sort map { $_->{PRINT} } values %{$uuid_fs_map{$uuid}}), $uuid]);
  } else {
    push(@$lines, ["$prefix<BTRFS_ROOT>/$node->{path}", $uuid]);
  }

  $prefix =~ s/./ /g;
  if($node->{received_uuid}) {
    if($node->{received_uuid} ne '-') {
      _origin_tree("${prefix}^-- ", $node->{received_uuid}, $lines);
    }
  } else {
    # printed if "btrfs_progs_compat" is set
    push(@$lines,  ["$prefix^-- <missing_received_uuid>", $uuid]);
  }
  if($node->{parent_uuid} ne '-') {
    _origin_tree("${prefix}", $node->{parent_uuid}, $lines);
  }
}


sub schedule(@)
{
  my %args = @_;
  my $schedule             = $args{schedule}             || die;
  my @today                = @{$args{today}};
  my $preserve_day_of_week = $args{preserve_day_of_week} || die;
  my $preserve_daily       = $args{preserve_daily}       // die;
  my $preserve_weekly      = $args{preserve_weekly}      // die;
  my $preserve_monthly     = $args{preserve_monthly}     // die;
  my $preserve_latest      = $args{preserve_latest}      || 0;
  my $results_list         = $args{results};
  my $result_hints         = $args{result_hints} // {};

  DEBUG "Filter scheme: preserving all within $preserve_daily days";
  DEBUG "Filter scheme: preserving first in week (starting on $preserve_day_of_week), for $preserve_weekly weeks";
  DEBUG "Filter scheme: preserving last weekly of month, for $preserve_monthly months";

  # sort the schedule, ascending by date
  my @sorted_schedule = sort { ($a->{btrbk_date}->[0] <=> $b->{btrbk_date}->[0]) ||
                               ($a->{btrbk_date}->[1] <=> $b->{btrbk_date}->[1]) ||
                               ($a->{btrbk_date}->[2] <=> $b->{btrbk_date}->[2]) ||
                               ($a->{btrbk_date}->[3] <=> $b->{btrbk_date}->[3]) ||
                               ($a->{btrbk_date}->[4] <=> $b->{btrbk_date}->[4]) ||
                               ($a->{btrbk_date}->[5] <=> $b->{btrbk_date}->[5])
                             } @$schedule;

  # first, do our calendar calculations
  # note: our week starts on $preserve_day_of_week
  my $delta_days_to_eow_from_today = $day_of_week_map{$preserve_day_of_week} - Day_of_Week(@today) - 1;
  $delta_days_to_eow_from_today = $delta_days_to_eow_from_today + 7 if($delta_days_to_eow_from_today < 0);
  TRACE "last day before next $preserve_day_of_week is in $delta_days_to_eow_from_today days";
  foreach my $href (@sorted_schedule)
  {
    my @date = @{$href->{btrbk_date}}[0..2];  # Date::Calc takes: @date = ( yy, mm, dd )
    my $delta_days = Delta_Days(@date, @today);
    my $delta_days_to_eow = $delta_days + $delta_days_to_eow_from_today;
    {
      use integer; # do integer arithmetics
      $href->{delta_days}   = $delta_days;
      $href->{delta_weeks}  = $delta_days_to_eow / 7;
      $href->{err_days}     = 6 - ( $delta_days_to_eow % 7 );
      $href->{delta_months} = ($today[0] - $date[0]) * 12 + ($today[1] - $date[1]);
      $href->{month}        = "$date[0]-$date[1]";
    }
  }

  if($preserve_latest && (scalar @sorted_schedule)) {
    my $href = $sorted_schedule[-1];
    $href->{preserve} ||= $preserve_latest;
  }

  # filter daily, weekly, monthly
  my %first_in_delta_weeks;
  my %last_weekly_in_delta_months;
  foreach my $href (@sorted_schedule) {
    if($preserve_daily && (($preserve_daily eq "all") || ($href->{delta_days} <= $preserve_daily))) {
      $href->{preserve} ||= "preserved daily: $href->{delta_days} days ago";
    }
    $first_in_delta_weeks{$href->{delta_weeks}} //= $href;
  }
  foreach (reverse sort keys %first_in_delta_weeks) {
    my $href = $first_in_delta_weeks{$_} || die;
    if($preserve_weekly && (($preserve_weekly eq "all") || ($href->{delta_weeks} <= $preserve_weekly))) {
      $href->{preserve} ||= "preserved weekly: $href->{delta_weeks} weeks ago, " . ($href->{err_days} ? "+$href->{err_days} days after " : "on ") . "$preserve_day_of_week";
    }
    $last_weekly_in_delta_months{$href->{delta_months}} = $href;
  }
  foreach (reverse sort keys %last_weekly_in_delta_months) {
    my $href = $last_weekly_in_delta_months{$_} || die;
    if($preserve_monthly && (($preserve_monthly eq "all") || ($href->{delta_months} <= $preserve_monthly))) {
      $href->{preserve} ||= "preserved monthly: " . ($href->{err_days} ? "$href->{err_days} days after " : "") . "last $preserve_day_of_week of month $href->{month} (age: $href->{delta_months} months)";
    }
  }

  # assemble results
  my @delete;
  my @preserve;
  my %preserve_matrix = ( d   => $preserve_daily,
                          w   => $preserve_weekly,
                          m   => $preserve_monthly,
                          dow => $preserve_day_of_week,
                         );
  my %result_base = ( %preserve_matrix,
                      scheme => format_preserve_matrix(%preserve_matrix, format => "short"),
                      %$result_hints,
                     );
  foreach my $href (@sorted_schedule)
  {
    if($href->{preserve}) {
      push(@preserve, $href->{value});
      DEBUG "=== $href->{name}: $href->{preserve}" if($href->{name});
      push @$results_list, { %result_base,
                             # action => "preserve",
                             reason => $href->{preserve},
                             value => $href->{value},
                           } if($results_list);

    }
    else {
      push(@delete, $href->{value});
      DEBUG "<<< $href->{name}" if($href->{name});
      push @$results_list, { %result_base,
                             action => "delete",
                             value => $href->{value},
                           } if($results_list);;
    }
  }
  DEBUG "Preserving " . @preserve . "/" . @$schedule . " items";
  return (\@preserve, \@delete);
}


sub format_preserve_matrix(@)
{
  my %args = @_;
  my $dow = $args{dow} // config_key($args{config}, "preserve_day_of_week");
  my $d   = $args{d}   // config_key($args{config}, "$args{prefix}_preserve_daily");
  my $w   = $args{w}   // config_key($args{config}, "$args{prefix}_preserve_weekly");
  my $m   = $args{m}   // config_key($args{config}, "$args{prefix}_preserve_monthly");
  my $format = $args{format} // "long";
  $d =~ s/^all$/-1/;
  $w =~ s/^all$/-1/;
  $m =~ s/^all$/-1/;
  if($format eq "short") {
    # short format
    return sprintf("%2sd %2sw %2sm", $d, $w, $m);
  }
  # long format
  return sprintf("%2sd %2sw %2sm ($dow)", $d, $w, $m);
}


sub print_header(@)
{
  my %args = @_;
  my $config = $args{config};

  print "--------------------------------------------------------------------------------\n";
  print "$args{title} ($version_info)\n\n";
  if($args{time}) {
    print "    Date:   " . localtime($args{time}) . "\n";
  }
  if($config) {
    print "    Config: $config->{SRC_FILE}\n";
  }
  if($dryrun) {
    print "    Dryrun: YES\n";
  }
  if($config && $config->{CMDLINE_FILTER_LIST}) {
    my @list = sort @{$config->{CMDLINE_FILTER_LIST}};
    my @sorted = ( grep(/^group/,     @list),
                   grep(/^volume/,    @list),
                   grep(/^subvolume/, @list),
                   grep(/^target/,    @list) );
    die unless(scalar(@list) == scalar(@sorted));
    print "    Filter: ";
    print join("\n            ", @sorted);
    print "\n";
  }
  if($args{info}) {
    print "\n" . join("\n", grep(defined, @{$args{info}})) . "\n";
  }
  if($args{legend}) {
    print "\nLegend:\n    ";
    print join("\n    ", @{$args{legend}});
    print "\n";
  }
  print "--------------------------------------------------------------------------------\n";
}


sub print_formatted(@)
{
  my $format_key = shift || die;
  my $data = shift || die;
  my $default_format = "table";
  my %args = @_;
  my $title = $args{title};
  my $format = $args{output_format} || $output_format || $default_format;
  my $keys = $table_formats{$format_key}->{$format};

  unless($keys) {
    WARN "Unsupported output format \"$format\", defaulting to \"$default_format\" format.";
    $keys = $table_formats{$format_key}->{$default_format} || die;
    $format = $default_format;
  }

  print "$title\n" if($title);
  if($format eq "raw")
  {
    # output: key0="value0" key1="value1" ...
    foreach my $row (@$data) {
      print "list_type=\"$format_key\" ";
      print join(' ', map { "$_=\"" . ($row->{$_} // "") . "\""; } @$keys) . "\n";
    }
  }
  else
  {
    # sanitize and calculate maxlen for each column
    # NOTE: this is destructive on data!
    my %maxlen;
    my @sane_data;
    foreach my $key (@$keys) {
      $maxlen{$key} = length($key); # initialize with size of key
    }
    foreach my $row (@$data) {
      foreach my $key (@$keys) {
        my $val = $row->{$key};
        if(ref $val eq "ARRAY") {
          $val = join(',', @{$val});
        }
        $val //= "-";
        $val = "-" if($val eq "");
        $row->{$key} = $val;  # write back the sanitized value
        $maxlen{$key} = length($val) if($maxlen{$key} < length($val));
      }
    }

    # print keys (headings)
    my $fill = '';
    foreach (@$keys) {
      print $fill . $_;
      $fill = ' ' x (2 + $maxlen{$_} - length($_));
    }
    print "\n";
    print join("  ", map { '-' x ($maxlen{$_}) } @$keys) . "\n";

    # print values
    foreach my $row (@$data) {
      my $fill = '';
      foreach (@$keys) {
        my $val = $row->{$_};
        print $fill . $val;
        $fill = ' ' x (2 + $maxlen{$_} - length($val));
      }
      print "\n";
    }
  }
}


MAIN:
{
  # set PATH instead of using absolute "/sbin/btrfs" (for now), as
  # different distros (and even different versions of btrfs-progs)
  # install the "btrfs" executable to different locations.
  $ENV{PATH} = '/sbin:/bin:/usr/sbin:/usr/bin';

  Getopt::Long::Configure qw(gnu_getopt);
  $Data::Dumper::Sortkeys = 1;
  my $start_time = time;
  my @today_and_now = Today_and_Now();
  my @today = @today_and_now[0..2];


  my ($config_cmdline, $quiet, $verbose, $preserve_backups, $resume_only);
  unless(GetOptions(
    'help|h'        => sub { VERSION_MESSAGE(); HELP_MESSAGE(0); exit 0; },
    'version'       => sub { VERSION_MESSAGE(); exit 0; },
    'config|c=s'    => \$config_cmdline,
    'preserve|p'    => \$preserve_backups,
    'resume-only|r' => \$resume_only,
    'quiet|q'       => \$quiet,
    'verbose|v'     => sub { $loglevel = 2; },
    'loglevel|l=s'  => \$loglevel,
    'progress'      => \$show_progress,
    'table|t'       => sub { $output_format = "table" },
    'format=s'      => \$output_format,
   ))
  {
    VERSION_MESSAGE();
    HELP_MESSAGE(0);
    exit 2;
  }
  my $command = shift @ARGV;
  unless($command) {
    VERSION_MESSAGE();
    HELP_MESSAGE(0);
    exit 2;
  }

  # assign command line options
  if   (lc($loglevel) eq "warn")  { $loglevel = 1; }
  elsif(lc($loglevel) eq "info")  { $loglevel = 2; }
  elsif(lc($loglevel) eq "debug") { $loglevel = 3; }
  elsif(lc($loglevel) eq "trace") { $loglevel = 4; }
  elsif($loglevel =~ /^[0-9]+$/)  { ; }
  else                            { $loglevel = 1; }
  @config_src = ( $config_cmdline ) if($config_cmdline);

  # check command line options
  if($show_progress && (not -e '/usr/bin/pv')) {
    WARN 'found option "--progress", but "pv" is not present: (please install "pv")';
    $show_progress = 0;
  }
  my ($action_run, $action_info, $action_tree, $action_diff, $action_origin, $action_config_print, $action_list);
  my @filter_args;
  my $args_allow_group = 0;
  my ($args_expected_min, $args_expected_max) = (0, 0);
  if(($command eq "run") || ($command eq "dryrun")) {
    $action_run = 1;
    $dryrun = 1 if($command eq "dryrun");
    $args_expected_min = 0;
    $args_expected_max = 9999;
    $args_allow_group = 1;
    @filter_args = @ARGV;
  }
  elsif ($command eq "info") {
    $action_info = 1;
    $args_expected_min = 0;
    $args_expected_max = 9999;
    $args_allow_group = 1;
    @filter_args = @ARGV;
  }
  elsif ($command eq "tree") {
    $action_tree = 1;
    $args_expected_min = 0;
    $args_expected_max = 9999;
    $args_allow_group = 1;
    @filter_args = @ARGV;
  }
  elsif ($command eq "diff") {
    $action_diff = 1;
    $args_expected_min = $args_expected_max = 2;
    @filter_args = @ARGV;
  }
  elsif ($command eq "origin") {
    $action_origin = 1;
    $args_expected_min = $args_expected_max = 1;
    @filter_args = @ARGV;
  }
  elsif($command eq "list") {
    my $subcommand = shift @ARGV;
    $action_list = "target-all";
    if(defined($subcommand)) {
      if(($subcommand eq "volume") ||
         ($subcommand eq "source") ||
         ($subcommand eq "target"))
      {
        $action_list = $subcommand;
      }
      else {
        unshift @ARGV, $subcommand;
      }
    }
    $args_expected_min = 0;
    $args_expected_max = 9999;
    $args_allow_group = 1;
    @filter_args = @ARGV;
  }
  elsif ($command eq "config") {
    my $subcommand = shift @ARGV // "";
    if(($subcommand eq "print") || ($subcommand eq "print-all")) {
      $action_config_print = $subcommand;
      $args_expected_min = 0;
      $args_expected_max = 9999;
      $args_allow_group = 1;
      @filter_args = @ARGV;
    }
    else {
      ERROR "Unknown subcommand for \"config\" command: $subcommand";
      HELP_MESSAGE(0);
      exit 2;
    }
  }
  else {
    ERROR "Unrecognized command: $command";
    HELP_MESSAGE(0);
    exit 2;
  }
  if(($args_expected_min > scalar(@ARGV)) || ($args_expected_max < scalar(@ARGV))) {
    ERROR "Incorrect number of arguments";
    HELP_MESSAGE(0);
    exit 2;
  }

  # input validation
  foreach (@filter_args) {
    s/\/+$//;   # remove trailing slash
    if($args_allow_group && /^($group_match)$/) {  # matches group
      $_ = $1; # untaint argument
    }
    elsif(/^(($ssh_prefix_match)?\/$file_match)$/) {  # matches ssh statement or absolute file
      $_ = $1; # untaint argument
    }
    elsif(/^(?<host>$ip_addr_match|$host_name_match):\/(?<file>$file_match)$/) {  # convert "my.host.com:/my/path" to ssh url
      $_ = "ssh://$+{host}/$+{file}";
    }
    elsif(/^\{(?<host>$ip_addr_match|$host_name_match)\}\/(?<file>$file_match)$/) {  # convert "{my.host.com}/my/path" to ssh url
      $_ = "ssh://$+{host}/$+{file}";
    }
    else {
      ERROR "Bad argument: not a subvolume" . ($args_allow_group ? "/group" : "") . " declaration: $_";
      HELP_MESSAGE(0);
      exit 2;
    }
  }


  INFO "$version_info  (" . localtime($start_time) . ")";

  if($action_diff)
  {
    #
    # print snapshot diff
    #
    my $src_url    = $filter_args[0] || die;
    my $target_url = $filter_args[1] || die;
    # FIXME: allow ssh:// src/dest (does not work since the configuration is not yet read).

    my $src_vol = vinfo($src_url, { CONTEXT => "cmdline" });
    unless(vinfo_root($src_vol)) { ERROR "Failed to fetch subvolume detail for '$src_vol->{PRINT}'" . ($err ? ": $err" : ""); exit 1; }
    if($src_vol->{is_root})  { ERROR "Subvolume at \"$src_url\" is btrfs root!"; exit 1; }
    unless($src_vol->{cgen}) { ERROR "Subvolume at \"$src_url\" does not provide cgen"; exit 1; }

    my $target_vol = vinfo($target_url, { CONTEXT => "cmdline" });
    unless(vinfo_root($target_vol)) { ERROR "Failed to fetch subvolume detail for '$target_vol->{PRINT}'" . ($err ? ": $err" : ""); exit 1; }
    unless($target_vol->{cgen}) { ERROR "Subvolume at \"$target_url\" does not provide cgen"; exit 1; }

    my $uuid_list = vinfo_fs_list($src_vol);
    unless($uuid_list->{$target_vol->{uuid}}) {
      ERROR "Target subvolume is not on the same btrfs filesystem!";
      exit 1;
    }

    my $lastgen;

    # check if given src and target share same parent
    if($src_vol->{parent_uuid} eq $target_vol->{uuid}) {
      DEBUG "target subvolume is direct parent of source subvolume";
    }
    elsif($src_vol->{parent_uuid} eq $target_vol->{parent_uuid}) {
      DEBUG "target subvolume and source subvolume share same parent";
    }
    else {
      # TODO: this rule only applies to snapshots. find a way to distinguish snapshots from received backups
      # ERROR "Subvolumes \"$target_url\" and \"$src_url\" do not share the same parents";
      # exit 1;
    }

    # NOTE: in some cases "cgen" differs from "gen", even for read-only snapshots (observed: gen=cgen+1)
    $lastgen = $src_vol->{cgen} + 1;

    # dump files, sorted and unique
    my $ret = btrfs_subvolume_find_new($target_vol, $lastgen);
    exit 1 unless(ref($ret));

    print_header(title => "Subvolume Diff",
                 time => $start_time,
                 info => [
                   "Showing changed files for subvolume:",
                   "  $target_vol->{PRINT}  (gen=$target_vol->{gen})",
                   "",
                   "Starting at creation generation of subvolume:",
                   "  $src_vol->{PRINT}  (cgen=$src_vol->{cgen})",
                   "",
                   "This will show all files modified within generation range: [$lastgen..$target_vol->{gen}]",
                   "Newest file generation (transid marker) was: $ret->{transid_marker}",
                   ($ret->{parse_errors} ? "Parse errors: $ret->{parse_errors}" : undef),
                  ],
                 legend => [
                   "+..     file accessed at offset 0 (at least once)",
                   ".c.     flags COMPRESS or COMPRESS|INLINE set (at least once)",
                   "..i     flags INLINE or COMPRESS|INLINE set (at least once)",
                   "<count> file was modified in <count> generations",
                   "<size>  file was modified for a total of <size> bytes",
                  ]
                );

    my $files = $ret->{files};

    # calculate the character offsets
    my $len_charlen = 0;
    my $gen_charlen = 0;
    foreach (values %$files) {
      my $len = length($_->{len});
      my $gen = length(scalar(keys(%{$_->{gen}})));
      $len_charlen = $len if($len > $len_charlen);
      $gen_charlen = $gen if($gen > $gen_charlen);
    }

    # finally print the output
    foreach my $name (sort keys %$files) {
      print ($files->{$name}->{new}               ? '+' : '.');
      print ($files->{$name}->{flags}->{compress} ? 'c' : '.');
      print ($files->{$name}->{flags}->{inline}   ? 'i' : '.');

      # make nice table
      my $gens = scalar(keys(%{$files->{$name}->{gen}}));
      my $len = $files->{$name}->{len};
      print "  " . (' ' x ($gen_charlen - length($gens))) .  $gens;
      print "  " . (' ' x ($len_charlen - length($len))) .  $len;

      print "  $name\n";
    }

    exit 0;
  }


  #
  # parse config file
  #
  my $config = parse_config(@config_src);
  unless($config) {
    ERROR "Failed to parse configuration file";
    exit 2;
  }
  unless(ref($config->{VOLUME}) eq "ARRAY") {
    ERROR "No volumes defined in configuration file";
    exit 2;
  }


  #
  # filter subvolumes matching command line arguments
  #
  if(($action_run || $action_tree || $action_info || $action_list || $action_config_print) && scalar(@filter_args))
  {
    my %match;
    foreach my $config_vol (@{$config->{VOLUME}}) {
      my $vol_url = $config_vol->{url} // die;
      my $found_vol = 0;
      foreach my $filter (@filter_args) {
        if(($vol_url eq $filter) || (map { ($filter eq $_) || () } @{$config_vol->{group}})) {
          TRACE "filter argument \"$filter\" matches volume: $vol_url\n";
          $match{$filter} = ($vol_url eq $filter) ? "volume=" . vinfo($vol_url, $config_vol)->{PRINT} : "group=$filter";
          $found_vol = 1;
          # last; # need to cycle through all filter_args for correct %match
        }
      }
      next if($found_vol);

      my @filter_subvol;
      foreach my $config_subvol (@{$config_vol->{SUBVOLUME}}) {
        my $subvol_url = $config_subvol->{url} // die;
        my $found_subvol = 0;
        foreach my $filter (@filter_args) {
          if(($subvol_url eq $filter) || (map { ($filter eq $_) || () } @{$config_subvol->{group}})) {
            TRACE "filter argument \"$filter\" matches subvolume: $subvol_url\n";
            $match{$filter} = ($subvol_url eq $filter) ? "subvolume=" . vinfo($subvol_url, $config_subvol)->{PRINT} : "group=$filter";
            $found_subvol = 1;
            $found_vol = 1;
            # last; # need to cycle through all filter_args for correct %match
          }
        }
        next if($found_subvol);

        my $snapshot_name = $config_subvol->{snapshot_name} // die;
        foreach my $config_target (@{$config_subvol->{TARGET}}) {
          my $target_url = $config_target->{url} // die;
          my $found_target = 0;
          foreach my $filter (@filter_args) {
            if(($filter eq $target_url) ||
               ($filter eq "$target_url/$snapshot_name") ||
               (map { ($filter eq $_) || () } @{$config_target->{group}})) {
              TRACE "filter argument \"$filter\" matches target: $target_url\n";
              $match{$filter} = ($target_url eq $filter) ? "target=" . vinfo($target_url, $config_target)->{PRINT} : "group=$filter";
              $found_target = 1;
              $found_subvol = 1;
              $found_vol = 1;
              # last; # need to cycle through all filter_args for correct %match
            }
          }
          unless($found_target) {
            DEBUG "No match on filter command line argument, skipping target: $target_url";
            ABORTED($config_target, "USER_SKIP");
          }
        }
        unless($found_subvol) {
          DEBUG "No match on filter command line argument, skipping subvolume: $subvol_url";
          ABORTED($config_subvol, "USER_SKIP");
        }
      }
      unless($found_vol) {
        DEBUG "No match on filter command line argument, skipping volume: $vol_url";
        ABORTED($config_vol, "USER_SKIP");
      }
    }
    # make sure all args have a match
    my @nomatch = map { $match{$_} ? () : $_ } @filter_args;
    if(@nomatch) {
      foreach(@nomatch) {
        ERROR "Command line argument does not match any volume, subvolume, target or group declaration: $_";
      }
      exit 2;
    }
    $config->{CMDLINE_FILTER_LIST} = [ values %match ];
  }


  if($action_info)
  {
    #
    # print filesystem information
    #
    print "================================================================================\n";
    print "Filesystem information ($version_info)\n\n";
    print "    Date:   " . localtime($start_time) . "\n";
    print "    Config: $config->{SRC_FILE}\n";
    print "================================================================================\n";

    my %processed;
    foreach my $config_vol (@{$config->{VOLUME}})
    {
      next if($config_vol->{ABORTED});
      my $sroot = vinfo($config_vol->{url}, $config_vol);
      unless($processed{$sroot->{URL}})
      {
        print "\n--------------------------------------------------------------------------------\n";
        print "Source volume: $sroot->{PRINT}\n";
        print "--------------------------------------------------------------------------------\n";
        print (btrfs_filesystem_usage($sroot) // "");
        print "\n";
        $processed{$sroot->{URL}} = 1;
      }
    }

    foreach my $config_vol (@{$config->{VOLUME}}) {
      next if($config_vol->{ABORTED});
      my $sroot = vinfo($config_vol->{url}, $config_vol);
      foreach my $config_subvol (@{$config_vol->{SUBVOLUME}}) {
        next if($config_subvol->{ABORTED});
        foreach my $config_target (@{$config_subvol->{TARGET}})
        {
          my $droot = vinfo($config_target->{url}, $config_target);
          unless($processed{$droot->{URL}})
          {
            print "\n--------------------------------------------------------------------------------\n";
            print "Target volume: $droot->{PRINT}\n";
            print "               ^--- $sroot->{PRINT}\n";
            print "--------------------------------------------------------------------------------\n";
            print (btrfs_filesystem_usage($droot) // "");
            print "\n";
            $processed{$droot->{URL}} = 1;
          }
        }
      }
    }
    exit 0;
  }


  if($action_config_print)
  {
    my $resolve = ($action_config_print eq "print-all");
    #
    # print configuration lines, machine readable
    #
    my @out;
    push @out, config_dump_keys($config, skip_defaults => 1);
    foreach my $config_vol (@{$config->{VOLUME}}) {
      next if($config_vol->{ABORTED});
      my $sroot = vinfo($config_vol->{url}, $config_vol);
      push @out, "\nvolume $sroot->{URL}";
      push @out, config_dump_keys($config_vol, prefix => "\t", resolve => $resolve);

      foreach my $config_subvol (@{$config_vol->{SUBVOLUME}}) {
        next if($config_subvol->{ABORTED});
        my $svol = vinfo_child($sroot, $config_subvol->{rel_path});
        # push @out, "\n  subvolume $svol->{URL}";
        push @out, "\n\tsubvolume $svol->{SUBVOL_PATH}";
        push @out, config_dump_keys($config_subvol, prefix => "\t\t", resolve => $resolve);

        foreach my $config_target (@{$config_subvol->{TARGET}})
        {
          next if($config_target->{ABORTED});
          my $droot = vinfo($config_target->{url}, $config_target);
          push @out, "\n\t\ttarget $config_target->{target_type} $droot->{URL}";
          push @out, config_dump_keys($config_target, prefix => "\t\t\t", resolve => $resolve);
        }
      }
    }

    print_header(title => "Configuration Dump",
                 config => $config,
                 time => $start_time,
                );

    print join("\n", @out) . "\n";
    exit 0;
  }


  if($action_list)
  {
    my @vol_data;
    my @subvol_data;
    my @target_data;
    my @mixed_data;
    my %target_uniq;

    #
    # print configuration lines, machine readable
    #
    foreach my $config_vol (@{$config->{VOLUME}}) {
      next if($config_vol->{ABORTED});
      my $sroot = vinfo($config_vol->{url}, $config_vol);
      my $volh = { vinfo_prefixed_keys("volume", $sroot) };
      push @vol_data, $volh;

      foreach my $config_subvol (@{$config_vol->{SUBVOLUME}}) {
        next if($config_subvol->{ABORTED});
        my $svol = vinfo_child($sroot, $config_subvol->{rel_path});
        my $subvolh = { %$volh,
                        vinfo_prefixed_keys("source", $svol),
                        snapshot_path     => $sroot->{PATH} . (config_key($config_subvol, "snapshot_dir", prefix => '/') // ""),
                        snapshot_name     => config_key($config_subvol, "snapshot_name"),
                        snapshot_preserve => format_preserve_matrix(config => $config_subvol, prefix => "snapshot"),
                      };
        push @subvol_data, $subvolh;

        my $found = 0;
        foreach my $config_target (@{$config_subvol->{TARGET}})
        {
          next if($config_target->{ABORTED});
          my $droot = vinfo($config_target->{url}, $config_target);
          my $targeth = { %$subvolh,
                          vinfo_prefixed_keys("target", $droot),
                          target_preserve => format_preserve_matrix(config => $config_target, prefix => "target"),
                        };
          if($action_list eq "target") {
            next if($target_uniq{$droot->{URL}});
            $target_uniq{$droot->{URL}} = 1;
          }
          push @target_data, $targeth;
          push @mixed_data, $targeth;
          $found = 1;
        }
        # make sure the subvol is always printed (even if no targets around)
        push @mixed_data, $subvolh unless($found);
      }
    }
    if($action_list eq "volume") {
      print_formatted("list_volume", \@vol_data);
    }
    elsif($action_list eq "source") {
      print_formatted("list_source", \@subvol_data);
    }
    elsif($action_list eq "target") {
      print_formatted("list_target", \@target_data);
    }
    else {
      # default format
      print_formatted("list", \@mixed_data);
    }
    exit 0;
  }


  #
  # fill vinfo hash, basic checks on configuration
  #
  my %snapshot_check;
  my %backup_check;
  foreach my $config_vol (@{$config->{VOLUME}})
  {
    next if($config_vol->{ABORTED});
    my $sroot = vinfo($config_vol->{url}, $config_vol);
    unless(vinfo_root($sroot)) {
      ABORTED($config_vol, "Failed to fetch subvolume detail" . ($err ? ": $err" : ""));
      WARN "Skipping volume \"$sroot->{PRINT}\": $config_vol->{ABORTED}";
      next;
    }
    $config_vol->{sroot} = $sroot;

    foreach my $config_subvol (@{$config_vol->{SUBVOLUME}})
    {
      next if($config_subvol->{ABORTED});

      my $svol = vinfo_subvol($sroot, $config_subvol->{rel_path});
      unless($svol) {
        # configured subvolume is not present in btrfs subvolume list.
        # try to read subvolume detail, as configured subvolume could be a symlink.
        DEBUG "Subvolume \"$config_subvol->{rel_path}\" not present in btrfs subvolume list for \"$sroot->{PRINT}\"";
        $svol = vinfo_child($sroot, $config_subvol->{rel_path});
        my $detail = btrfs_subvolume_detail($svol);
        unless($detail) {
          ABORTED($config_subvol, "Failed to fetch subvolume detail" . ($err ? ": $err" : ""));
          WARN "Skipping subvolume \"$svol->{PRINT}\": $config_subvol->{ABORTED}";
          next;
        }
        if($detail->{is_root}) {
          ABORTED($config_subvol, "Subvolume is btrfs root");
          WARN "Skipping subvolume \"$svol->{PRINT}\": $config_subvol->{ABORTED}";
          next;
        }
        if(grep { $_->{uuid} eq $detail->{uuid} } values %{vinfo_subvol_list($sroot)}) {
          vinfo_set_detail($svol, $uuid_info{$detail->{uuid}});
        } else {
          ABORTED($config_subvol, "Not a child subvolume of: $sroot->{PRINT}");
          WARN "Skipping subvolume \"$svol->{PRINT}\": $config_subvol->{ABORTED}";
          next;
        }
      }
      $config_subvol->{svol} = $svol;

      # check for duplicate snapshot locations
      my $snapdir = config_key($config_subvol, "snapshot_dir", postfix => '/') // "";
      my $snapshot_basename = config_key($config_subvol, "snapshot_name") // die;
      my $snapshot_target = "$sroot->{REAL_URL}/$snapdir$snapshot_basename";
      if(my $prev = $snapshot_check{$snapshot_target}) {
        ERROR "Subvolume \"$prev\" and \"$svol->{PRINT}\" will create same snapshot: $snapshot_target";
        ERROR "Please fix \"snapshot_name\" configuration options!";
        exit 1;
      }
      $snapshot_check{$snapshot_target} = $svol->{PRINT};

      foreach my $config_target (@{$config_subvol->{TARGET}})
      {
        next if($config_target->{ABORTED});
        my $droot = vinfo($config_target->{url}, $config_target);

        my $target_type = $config_target->{target_type} || die;
        if($target_type eq "send-receive")
        {
          unless(vinfo_root($droot)) {
            ABORTED($config_target, "Failed to fetch subvolume detail" . ($err ? ": $err" : ""));
            WARN "Skipping target \"$droot->{PRINT}\": $config_target->{ABORTED}";
            next;
          }
        }
        elsif($target_type eq "raw")
        {
          DEBUG "Creating raw subvolume list: $droot->{PRINT}";
          my $ret = run_cmd(
            # NOTE: check for file size >0, which causes bad (zero-sized) images to be resumed
            # TODO: fix btrfs_send_to_file() to never create bad images
            cmd => [ 'find', $droot->{PATH} . '/', '-maxdepth', '1', '-type', 'f', '-size', '+0' ],
            rsh => $droot->{RSH},
            # note: use something like this to get the real (link resolved) path
            # cmd => [ "find", $droot->{PATH} . '/', "-maxdepth", "1", "-name", "$snapshot_basename.\*.raw\*", '-printf', '%f\0', '-exec', 'realpath', '-z', '{}', ';' ],
            non_destructive => 1,
           );
          unless(defined($ret)) {
            ABORTED($config_target, "Failed to list files from: $droot->{PATH}");
            WARN  "Skipping target \"$droot->{PRINT}\": $config_target->{ABORTED}";
            next;
          }

          my %subvol_list;
          my %parent_uuid_list;
          foreach my $file (split("\n", $ret))
          {
            unless($file =~ /^$file_match$/) {
              DEBUG "Skipping non-parseable file: \"$file\"";
              next;
            }
            unless($file =~ s/^\Q$droot->{PATH}\E\///) {
              ABORTED($config_target, "Unexpected result from 'find': file \"$file\" is not under \"$droot->{PATH}\"");
              last;
            }
            my $filename_info = parse_filename($file, $snapshot_basename, 1);
            unless($filename_info) {
              DEBUG "Skipping file (not btrbk raw): \"$file\"";
              next;
            }

            # Fake btrfs subvolume information (received_uuid, parent_uuid) from filename info.
            #
            # NOTE: parent_uuid in $filename_info is the "parent of the source subvolume", NOT the
            # "parent of the received subvolume". We fake the real parent_uuid with the one from
            # the filename here.
            my $subvol = vinfo_child($droot, $file);
            vinfo_set_detail($subvol, $filename_info);

            $subvol_list{$file} = $subvol;
            $parent_uuid_list{$filename_info->{parent_uuid}} = $subvol if($filename_info->{parent_uuid} ne '-');
          }
          if($config_target->{ABORTED}) {
            WARN "Skipping target \"$droot->{PRINT}\": $config_target->{ABORTED}";
            next;
          }
          DEBUG "Found " . scalar(keys %subvol_list) . " raw subvolume backups of: $svol->{PRINT}";
          $droot->{SUBVOL_LIST} = \%subvol_list;
          $droot->{REAL_URL} = $droot->{URL}; # ignore links here

          # Make sure that incremental backup chains are never broken:
          foreach my $subvol (values %subvol_list)
          {
            # If restoring a backup from raw btrfs images (using "incremental yes|strict"):
            # "btrfs send -p parent source > svol.btrfs", the backups
            # on the target will get corrupted (unusable!)  as soon as
            # an any files in the chain gets deleted.
            #
            # We need to make sure btrbk will NEVER delete those:
            # - svol.<timestamp>--<received_uuid_0>.btrfs                        : root (full) image
            # - svol.<timestamp>--<received_uuid-n>[@<received_uuid_n-1>].btrfs  : incremental image

            if(my $child = $parent_uuid_list{$subvol->{received_uuid}}) {
              DEBUG "Found parent/child partners, forcing preserve of: \"$subvol->{PRINT}\", \"$child->{PRINT}\"";
              $subvol->{FORCE_PRESERVE} = "preserve forced: parent of another raw target";
              $child->{FORCE_PRESERVE} ||= "preserve forced: child of another raw target";
            }
          }

          # TRACE(Data::Dumper->Dump([\%subvol_list], ["vinfo_raw_subvol_list{$droot}"]));
        }
        $config_target->{droot} = $droot;

        # check for duplicate snapshot locations
        my $snapshot_backup_target = "$droot->{REAL_URL}/$snapshot_basename";
        if(my $prev = $backup_check{$snapshot_backup_target}) {
          ERROR "Subvolume \"$prev\" and \"$svol->{PRINT}\" will create same backup target: $snapshot_target";
          ERROR "Please fix \"snapshot_name\" or \"target\" configuration options!";
          exit 1;
        }
        $backup_check{$snapshot_backup_target} = $svol->{PRINT};
      }
    }
  }


  if($action_origin)
  {
    #
    # print origin information
    #
    my $url = $filter_args[0] || die;
    my $dump_uuid = 0;

    my $vol = $vinfo_cache{$url};
    unless($vol) {
      # specified volume is not in config
      DEBUG "Subvolume not parsed yet, fetching info: $url";
      $vol = vinfo($url, { CONTEXT => "cmdline" });
      unless(vinfo_root($vol)) {
        ERROR "Failed to fetch subvolume detail for: $url" . ($err ? ": $err" : "");
        exit 1;
      }
    }
    if($vol->{is_root}) {
      ERROR "Subvolume is btrfs root: $url\n";
      exit 1;
    }

    my $lines = [];
    _origin_tree("", $vol->{uuid}, $lines);

    print_header(title => "Origin Tree",
                 config => $config,
                 time => $start_time,
                 legend => [
                   "^--     : received from subvolume",
                   "newline : parent subvolume",
                   "orphaned: subvolume uuid could not be resolved (probably deleted)",
                  ]
                );

    my $len = 0;
    if($dump_uuid) {
      $len = (length($_->[0]) > $len ? length($_->[0]) : $len) foreach(@$lines);
    }
    foreach(@$lines) {
      print "$_->[0]";
      print ' ' x ($len - length($_->[0]) + 4) . "$_->[1]" if($dump_uuid);
      print "\n";
    }
    exit 0;
  }


  if($action_tree)
  {
    #
    # print snapshot tree
    #
    # TODO: reverse tree: print all backups from $droot and their corresponding source snapshots
    my @tree_out;
    my @raw_out;
    foreach my $config_vol (@{$config->{VOLUME}})
    {
      next if($config_vol->{ABORTED});
      my %droot_compat;
      my $sroot = $config_vol->{sroot} || die;
      push @tree_out, "$sroot->{PRINT}";
      foreach my $config_subvol (@{$config_vol->{SUBVOLUME}})
      {
        next if($config_subvol->{ABORTED});
        my $svol = $config_subvol->{svol} || die;
        push @tree_out, "|-- $svol->{PRINT}";
        foreach my $snapshot (sort { $a->{cgen} cmp $b->{cgen} } get_snapshot_children($sroot, $svol))
        {
          my $raw_data = { type => "snapshot",
                           btrbk_flags => [ ],
                           vinfo_prefixed_keys("source", $svol),
                           vinfo_prefixed_keys("snapshot", $snapshot),
                           snapshot_name => config_key($config_subvol, "snapshot_name"),
                         };
          if($snapshot->{cgen} == $svol->{gen}) {
            push @tree_out, "|   ^== $snapshot->{PATH}";
            push @{$raw_data->{btrbk_flags}}, "up-to-date";
          } else {
            push @tree_out, "|   ^-- $snapshot->{PATH}";
          }
          push @raw_out, $raw_data;
          foreach my $config_target (@{$config_subvol->{TARGET}})
          {
            next if($config_target->{ABORTED});
            my $droot = $config_target->{droot} || die;
            $droot_compat{$droot->{URL}} = 1 if($droot->{BTRFS_PROGS_COMPAT});
            foreach (sort { $a->{SUBVOL_PATH} cmp $b->{SUBVOL_PATH} } get_receive_targets($droot, $snapshot)) {
              push @tree_out, "|   |   >>> $_->{PRINT}";
              push @raw_out, { %$raw_data,
                               type => "received",
                               vinfo_prefixed_keys("received", $_),
                             };
            }
          }
        }
      }
      if(keys %droot_compat) {
        push @tree_out, "\nNOTE: Received subvolumes (backups) are guessed by subvolume name for targets:";
        push @tree_out, "      - " . join("\n      - ", (sort keys %droot_compat));
      }
      push @tree_out, "";
    }

    $output_format ||= "tree";
    if($output_format eq "tree") {
      print_header(title => "Backup Tree",
                   config => $config,
                   time => $start_time,
                   legend => [
                     "^--  snapshot",
                     "^==  snapshot (up-to-date)",
                     ">>>  received subvolume (backup)",
                    ]
                  );
      print join("\n", @tree_out);
    }
    else {
      print_formatted("tree", \@raw_out);
    }
    exit 0;
  }


  if($action_run)
  {
    if($resume_only) {
      INFO "Skipping snapshot creation (option \"-r\" present)";
    }
    else
    {
      #
      # create snapshots
      #
      foreach my $config_vol (@{$config->{VOLUME}})
      {
        next if($config_vol->{ABORTED});
        my $sroot = $config_vol->{sroot} || die;
        foreach my $config_subvol (@{$config_vol->{SUBVOLUME}})
        {
          next if($config_subvol->{ABORTED});
          my $svol = $config_subvol->{svol} || die;
          my $snapdir = config_key($config_subvol, "snapshot_dir", postfix => '/') // "";
          my $snapshot_basename = config_key($config_subvol, "snapshot_name") // die;

          # check if we need to create a snapshot
          my $snapshot_create = config_key($config_subvol, "snapshot_create");
          if(not $snapshot_create) {
            DEBUG "Snapshot creation disabled (snapshot_create=no)";
            next;
          }
          elsif($snapshot_create eq "always") {
            DEBUG "Snapshot creation enabled (snapshot_create=always)";
          }
          elsif($snapshot_create eq "onchange") {
            # check if latest snapshot is up-to-date with source subvolume (by generation)
            my $latest = get_latest_snapshot_child($sroot, $svol);
            if($latest) {
              if($latest->{cgen} == $svol->{gen}) {
                INFO "Snapshot creation skipped: snapshot_create=onchange, snapshot is up-to-date: $latest->{PRINT}";
                $config_subvol->{SNAPSHOT_UP_TO_DATE} = $latest;
                next;
              }
              DEBUG "Snapshot creation enabled: snapshot_create=onchange, gen=$svol->{gen} > snapshot_cgen=$latest->{cgen}";
            }
            else {
              DEBUG "Snapshot creation enabled: snapshot_create=onchange, no snapshots found";
            }
          }
          elsif($snapshot_create eq "ondemand") {
            # check if at least one target is present
            if(scalar grep { not $_->{ABORTED} } @{$config_subvol->{TARGET}}) {
              DEBUG "Snapshot creation enabled (snapshot_create=ondemand): at least one target is present";
            }
            else {
              INFO "Snapshot creation skipped: snapshot_create=ondemand, and no target is present for: $svol->{PRINT}";
              next;
            }
          }
          else {
            die "illegal value for snapshot_create configuration option: $snapshot_create";
          }

          # find unique snapshot name
          my $timestamp = ((config_key($config_subvol, "timestamp_format") eq "short") ?
                           sprintf("%04d%02d%02d", @today) :
                           sprintf("%04d%02d%02dT%02d%02d", @today_and_now));
          my @unconfirmed_target_name;
          my @lookup = keys %{vinfo_subvol_list($sroot)};
          @lookup = grep s/^\Q$snapdir\E// , @lookup;
          foreach my $config_target (@{$config_subvol->{TARGET}}) {
            if($config_target->{ABORTED}) {
              push(@unconfirmed_target_name, vinfo($config_target->{url}, $config_target));
              next;
            }
            my $droot = $config_target->{droot} || die;
            push(@lookup, keys %{vinfo_subvol_list($droot)});
          }
          @lookup = grep /^\Q$snapshot_basename.$timestamp\E(_[0-9]+)?$/ ,@lookup;
          TRACE "Present snapshot names for \"$svol->{PRINT}\": " . join(', ', @lookup);
          @lookup = map { /_([0-9]+)$/ ? $1 : 0 } @lookup;
          @lookup = sort { $b <=> $a } @lookup;
          my $postfix_counter = $lookup[0] // -1;
          $postfix_counter++;
          my $snapshot_name = $snapshot_basename . '.' . $timestamp . ($postfix_counter ? "_$postfix_counter" : "");

          if(@unconfirmed_target_name) {
            INFO "Assuming non-present subvolume \"$snapshot_name\" in skipped targets: " . join(", ", map { "\"$_->{PRINT}\"" } @unconfirmed_target_name);
          }

          # finally create the snapshot
          INFO "Creating subvolume snapshot for: $svol->{PRINT}";
          my $snapshot = vinfo_child($sroot, "$snapdir$snapshot_name");
          if(btrfs_subvolume_snapshot($svol, $snapshot)) {
            $config_subvol->{SNAPSHOT} = $snapshot;
          }
          else {
            ABORTED($config_subvol, "Failed to create snapshot: $svol->{PRINT} -> $sroot->{PRINT}/$snapdir$snapshot_name");
            WARN "Skipping subvolume section: $config_subvol->{ABORTED}";
          }
        }
      }
    }

    #
    # create backups
    #
    foreach my $config_vol (@{$config->{VOLUME}})
    {
      next if($config_vol->{ABORTED});
      my $sroot = $config_vol->{sroot} || die;
      foreach my $config_subvol (@{$config_vol->{SUBVOLUME}})
      {
        next if($config_subvol->{ABORTED});
        my $svol = $config_subvol->{svol} || die;
        my $snapdir = config_key($config_subvol, "snapshot_dir", postfix => '/') // "";
        my $snapshot_basename = config_key($config_subvol, "snapshot_name") // die;
        my $preserve_latest = $config_subvol->{SNAPSHOT} ? 0 : 1;

        foreach my $config_target (@{$config_subvol->{TARGET}})
        {
          next if($config_target->{ABORTED});
          my $droot = $config_target->{droot} || die;

          #
          # resume missing backups (resume_missing)
          #
          if(config_key($config_target, "resume_missing"))
          {
            INFO "Checking for missing backups of subvolume \"$svol->{PRINT}\" in: $droot->{PRINT}/";
            my @schedule;
            my $resume_total = 0;
            my $resume_success = 0;

            foreach my $child (sort { $a->{cgen} <=> $b->{cgen} } get_snapshot_children($sroot, $svol))
            {
              my $filename_info = parse_filename($child->{SUBVOL_PATH}, $snapdir . $snapshot_basename);
              next unless($filename_info);  # ignore non-btrbk files

              if(scalar get_receive_targets($droot, $child)) {
                DEBUG "Found matching receive target, skipping: $child->{PRINT}";
              }
              else {
                DEBUG "No matching receive targets found, adding resume candidate: $child->{PRINT}";

                if(my $err_vol = vinfo_subvol($droot, $child->{NAME})) {
                  WARN "Target subvolume \"$err_vol->{PRINT}\" exists, but is not a receive target of \"$child->{PRINT}\"";
                }

                # check if the target would be preserved
                push(@schedule, { value      => $child,
                                  btrbk_date => $filename_info->{btrbk_date},
                                  preserve   => $child->{FORCE_PRESERVE},
                                 }),
              }
            }

            if(scalar @schedule)
            {
              DEBUG "Checking schedule for resume candidates";
              # add all present backups to schedule, with no value
              # these are needed for correct results of schedule()
              foreach my $vol (values %{vinfo_subvol_list($droot)}) {
                my $filename_info = parse_filename($vol->{SUBVOL_PATH}, $snapshot_basename, ($config_target->{target_type} eq "raw"));
                next unless($filename_info);  # ignore non-btrbk files
                push(@schedule, { value      => undef,
                                  btrbk_date => $filename_info->{btrbk_date},
                                  preserve   => $vol->{FORCE_PRESERVE},
                                 });
              }
              my ($preserve, undef) = schedule(
                schedule             => \@schedule,
                today                => \@today,
                preserve_day_of_week => config_key($config_target, "preserve_day_of_week"),
                preserve_daily       => config_key($config_target, "target_preserve_daily"),
                preserve_weekly      => config_key($config_target, "target_preserve_weekly"),
                preserve_monthly     => config_key($config_target, "target_preserve_monthly"),
                preserve_latest      => $preserve_latest,
               );
              my @resume = grep defined, @$preserve;   # remove entries with no value from list (target subvolumes)
              $resume_total = scalar @resume;

              foreach my $child (sort { $a->{cgen} <=> $b->{cgen} } @resume)
              {
                INFO "Resuming subvolume backup (send-receive) for: $child->{PRINT}";
                my ($latest_common_src, $latest_common_target) = get_latest_common($sroot, $svol, $droot, $child->{cgen});
                if(macro_send_receive($config_target,
                                      snapshot  => $child,
                                      target    => $droot,
                                      parent    => $latest_common_src,  # this is <undef> if no common found
                                      resume    => 1,     # propagated to $config_target->{SUBVOL_RECEIVED}
                                     ))
                {
                  # tag the source snapshot, so that get_latest_common() above can make use of the newly received subvolume
                  $child->{RECEIVE_TARGET_PRESENT} = $droot->{URL};
                  $resume_success++;
                }
                else {
                  # note: ABORTED flag is already set by macro_send_receive()
                  ERROR("Error while resuming backups, aborting");
                  last;
                }
              }
            }

            if($resume_total) {
              INFO "Resumed $resume_success/$resume_total missing backups";
            } else {
              INFO "No missing backups found";
            }
          } # /resume_missing

          unless($resume_only)
          {
            # skip creation if resume_missing failed
            next if($config_target->{ABORTED});
            next unless($config_subvol->{SNAPSHOT});

            # finally receive the previously created snapshot
            INFO "Creating subvolume backup (send-receive) for: $svol->{PRINT}";
            my ($latest_common_src, $latest_common_target) = get_latest_common($sroot, $svol, $droot);
            macro_send_receive($config_target,
                               snapshot => $config_subvol->{SNAPSHOT},
                               target   => $droot,
                               parent   => $latest_common_src,  # this is <undef> if no common found
                              );
          }
        }
      }
    }


    #
    # remove backups following a preserve daily/weekly/monthly scheme
    #
    my $schedule_results = [];
    if($preserve_backups || $resume_only) {
      INFO "Preserving all backups (option \"-p\" or \"-r\" present)";
    }
    else
    {
      foreach my $config_vol (@{$config->{VOLUME}})
      {
        next if($config_vol->{ABORTED});
        my $sroot = $config_vol->{sroot} || die;
        foreach my $config_subvol (@{$config_vol->{SUBVOLUME}})
        {
          next if($config_subvol->{ABORTED});
          my $svol = $config_subvol->{svol} || die;
          my $snapdir = config_key($config_subvol, "snapshot_dir", postfix => '/') // "";
          my $snapshot_basename = config_key($config_subvol, "snapshot_name") // die;
          my $preserve_latest_snapshot = $config_subvol->{SNAPSHOT} ? 0 : "preserve forced: latest in list";
          my $preserve_latest_backup = $preserve_latest_snapshot;
          my $target_aborted = 0;

          foreach my $config_target (@{$config_subvol->{TARGET}})
          {
            if($config_target->{ABORTED}) {
              if($config_target->{ABORTED} eq "USER_SKIP") {
                $target_aborted ||= -1;
              } else {
                $target_aborted = 1;
              }
              next;
            }
            my $droot = $config_target->{droot} || die;
            if($config_target->{target_type} eq "raw") {
              if(config_key($config_target, "incremental")) {
                # In incremental mode, the latest backup is most certainly our parent.
                # (see note on FORCE_PRESERVE above)
                $preserve_latest_backup ||= "preserve forced: possibly parent of latest backup";
                # Note that we could check against $config_subvol->{SNAPSHOT}->{parent_uuid} to be certain,
                # but this information is not available in $dryrun:
                #  foreach my $vol (values %{vinfo_subvol_list($droot)}) {
                #    $vol->{FORCE_PRESERVE} = 1 if($vol->{received_uuid} eq $config_subvol->{SNAPSHOT}->{parent_uuid});
                #  }
              }
            }

            #
            # delete backups
            #
            INFO "Cleaning backups of subvolume \"$svol->{PRINT}\": $droot->{PRINT}/$snapshot_basename.*";
            my @schedule;
            foreach my $vol (values %{vinfo_subvol_list($droot)}) {
              my $filename_info = parse_filename($vol->{SUBVOL_PATH}, $snapshot_basename, ($config_target->{target_type} eq "raw"));
              next unless($filename_info);  # ignore non-btrbk files

              # NOTE: checking received_uuid does not make much sense, as this received_uuid is propagated to snapshots
              # if($vol->{received_uuid} && ($vol->{received_uuid} eq '-')) {
              #   INFO "Target subvolume is not a received backup, skipping deletion of: $vol->{PRINT}";
              #   next;
              # }
              push(@schedule, { value      => $vol,
                                name       => $vol->{PRINT},  # only for logging
                                btrbk_date => $filename_info->{btrbk_date},
                                preserve   => $vol->{FORCE_PRESERVE}
                               });
            }
            my (undef, $delete) = schedule(
              schedule             => \@schedule,
              today                => \@today,
              preserve_day_of_week => config_key($config_target, "preserve_day_of_week"),
              preserve_daily       => config_key($config_target, "target_preserve_daily"),
              preserve_weekly      => config_key($config_target, "target_preserve_weekly"),
              preserve_monthly     => config_key($config_target, "target_preserve_monthly"),
              preserve_latest      => $preserve_latest_backup,
              results              => $schedule_results,
              result_hints         => { topic => "backup", root_path => $droot->{PATH} },
             );
            my $ret = btrfs_subvolume_delete($delete, commit => config_key($config_target, "btrfs_commit_delete"), type => "delete_target");
            if(defined($ret)) {
              INFO "Deleted $ret subvolumes in: $droot->{PRINT}/$snapshot_basename.*";
              $config_target->{SUBVOL_DELETED} = $delete;
            }
            else {
              ABORTED($config_target, "Failed to delete subvolume");
              $target_aborted = -1;
            }
          }

          #
          # delete snapshots
          #
          if($target_aborted) {
            if($target_aborted == -1) {
              INFO "Skipping cleanup of snapshots for subvolume \"$svol->{PRINT}\", as at least one target is skipped by command line argument";
            } else {
              WARN "Skipping cleanup of snapshots for subvolume \"$svol->{PRINT}\", as at least one target aborted earlier";
            }
            next;
          }
          INFO "Cleaning snapshots: $sroot->{PRINT}/$snapdir$snapshot_basename.*";
          my @schedule;
          foreach my $vol (values %{vinfo_subvol_list($sroot)}) {
            my $filename_info = parse_filename($vol->{SUBVOL_PATH}, $snapdir . $snapshot_basename);
            next unless($filename_info);  # ignore non-btrbk files
            push(@schedule, { value      => $vol,
                              name       => $vol->{PRINT},  # only for logging
                              btrbk_date => $filename_info->{btrbk_date}
                             });
          }
          my (undef, $delete) = schedule(
            schedule             => \@schedule,
            today                => \@today,
            preserve_day_of_week => config_key($config_subvol, "preserve_day_of_week"),
            preserve_daily       => config_key($config_subvol, "snapshot_preserve_daily"),
            preserve_weekly      => config_key($config_subvol, "snapshot_preserve_weekly"),
            preserve_monthly     => config_key($config_subvol, "snapshot_preserve_monthly"),
            preserve_latest      => $preserve_latest_snapshot,
            results              => $schedule_results,
            result_hints         => { topic => "snapshot", root_path => $sroot->{PATH} },
           );
          my $ret = btrfs_subvolume_delete($delete, commit => config_key($config_subvol, "btrfs_commit_delete"), type => "delete_snapshot");
          if(defined($ret)) {
            INFO "Deleted $ret subvolumes in: $sroot->{PRINT}/$snapdir$snapshot_basename.*";
            $config_subvol->{SUBVOL_DELETED} = $delete;
          }
          else {
            ABORTED($config_subvol, "Failed to delete subvolume");
          }
        }
      }
    }

    my $time_elapsed = time - $start_time;
    INFO "Completed within: ${time_elapsed}s  (" . localtime(time) . ")";


    unless($quiet)
    {
      #
      # print scheduling results
      #
      if($loglevel >= 2) {
        my @data = map { $_->{url} = $_->{value}->{URL};
                         $_->{host} = $_->{value}->{HOST};
                         $_->{path} = $_->{value}->{PATH};
                         $_->{name} = $_->{value}->{SUBVOL_PATH};
                         $_->{target} = $_->{value}->{PRINT};
                         $_;
                       } @$schedule_results;
        my @data_backup = map { $_->{topic} eq "backup" ? $_ : () } @data;
        my @data_snapshot = map { $_->{topic} eq "snapshot" ? $_ : () } @data;

        print_formatted("schedule", \@data_snapshot, title => "SNAPSHOT SCHEDULE");
        print "\n";
        print_formatted("schedule", \@data_backup, title => "BACKUP SCHEDULE");
        print "\n";
      }


      #
      # print summary
      #
      $output_format ||= "custom";
      if($output_format eq "custom")
      {
        my @unrecoverable;
        my @out;
        my @raw_data;
        my $err_count = 0;
        foreach my $config_vol (@{$config->{VOLUME}})
        {
          my $sroot = $config_vol->{sroot} || vinfo($config_vol->{url}, $config_vol);
          foreach my $config_subvol (@{$config_vol->{SUBVOLUME}})
          {
            my @subvol_out;
            my $svol = $config_subvol->{svol} || vinfo_child($sroot, $config_subvol->{rel_path});

            if($config_subvol->{SNAPSHOT_UP_TO_DATE}) {
              push @subvol_out, "=== $config_subvol->{SNAPSHOT_UP_TO_DATE}->{PRINT}";
            }
            if($config_subvol->{SNAPSHOT}) {
              push @subvol_out, "+++ $config_subvol->{SNAPSHOT}->{PRINT}";
            }
            if($config_subvol->{SUBVOL_DELETED}) {
              foreach(sort { $a->{PATH} cmp $b->{PATH} } @{$config_subvol->{SUBVOL_DELETED}}) {
                push @subvol_out, "--- $_->{PRINT}";
              }
            }
            foreach my $config_target (@{$config_subvol->{TARGET}})
            {
              my $droot = $config_target->{droot} || vinfo($config_target->{url}, $config_target);
              foreach(@{$config_target->{SUBVOL_RECEIVED} // []}) {
                my $create_mode = "***";
                $create_mode = ">>>" if($_->{parent});
                # substr($create_mode, 0, 1, '%') if($_->{resume});
                $create_mode = "!!!" if($_->{ERROR});
                push @subvol_out, "$create_mode $_->{received_subvolume}->{PRINT}";
              }

              if($config_target->{SUBVOL_DELETED}) {
                foreach(sort { $a->{PATH} cmp $b->{PATH} } @{$config_target->{SUBVOL_DELETED}}) {
                  push @subvol_out, "--- $_->{PRINT}";
                }
              }

              if($config_target->{ABORTED} && ($config_target->{ABORTED} ne "USER_SKIP")) {
                push @subvol_out, "!!! Target \"$droot->{PRINT}\" aborted: $config_target->{ABORTED}";
                $err_count++;
              }

              push(@unrecoverable, $config_target->{UNRECOVERABLE}) if($config_target->{UNRECOVERABLE});
            }
            if($config_vol->{ABORTED} && ($config_vol->{ABORTED} ne "USER_SKIP")) {
              # repeat volume errors in subvolume context ($err_count is increased in volume context below)
              push @subvol_out, "!!! Volume \"$sroot->{PRINT}\" aborted: $config_vol->{ABORTED}";
            }
            if($config_subvol->{ABORTED} && ($config_subvol->{ABORTED} ne "USER_SKIP")) {
              push @subvol_out, "!!! Aborted: $config_subvol->{ABORTED}";
              $err_count++;
            }

            if(@subvol_out) {
              push @out, "$svol->{PRINT}", @subvol_out, "";
            }
            elsif($config_subvol->{ABORTED} && ($config_subvol->{ABORTED} eq "USER_SKIP")) {
              # don't print "<no_action>" on USER_SKIP
            }
            else {
              push @out, "$svol->{PRINT}", "<no_action>", "";
            }
          }
          if($config_vol->{ABORTED} && ($config_vol->{ABORTED} ne "USER_SKIP")) {
            push @raw_data, { type => "btrbk_volume",
                              status => "ABORT",
                              target_url => $sroot->{URL},
                              error_message => $config_vol->{ABORTED},
                              SORT => 1,
                            };
            $err_count++;
          }
        }

        print_header(title => "Backup Summary",
                     config => $config,
                     time => $start_time,
                     legend => [
                       "===  up-to-date subvolume (source snapshot)",
                       "+++  created subvolume (source snapshot)",
                       "---  deleted subvolume",
                       "***  received subvolume (non-incremental)",
                       ">>>  received subvolume (incremental)",
                       # "%>>  received subvolume (incremental, resume_missing)",
                      ],
                    );

        print join("\n", @out);

        if($resume_only) {
          print "\nNOTE: No snapshots created (option -r present)\n";
        }
        if($preserve_backups || $resume_only) {
          print "\nNOTE: Preserved all backups (option -p or -r present)\n";
        }
        if($err_count) {
          print "\nNOTE: Some errors occurred, which may result in missing backups!\n";
          print "Please check warning and error messages above.\n";
          print join("\n", @unrecoverable) . "\n" if(@unrecoverable);
        }
        if($dryrun) {
          print "\nNOTE: Dryrun was active, none of the operations above were actually executed!\n";
        }
      }
      else
      {
        # print action log
        @action_log = map { $_->{localtime} = localtime($_->{time}); $_; } @action_log;
        print_formatted("action_log", \@action_log, title => "BACKUP SUMMARY");
      }
    }

    foreach my $config_vol (@{$config->{VOLUME}}) {
      exit 10 if($config_vol->{ABORTED} && ($config_vol->{ABORTED} ne "USER_SKIP"));
      foreach my $config_subvol (@{$config_vol->{SUBVOLUME}}) {
        exit 10 if($config_subvol->{ABORTED} && ($config_subvol->{ABORTED} ne "USER_SKIP"));
        foreach my $config_target (@{$config_subvol->{TARGET}}) {
          exit 10 if($config_target->{ABORTED} && ($config_target->{ABORTED} ne "USER_SKIP"));
        }
      }
    }
  }
}


1;