--- /dev/null
+# Copyright (c) 2009, 2010 Zmanda Inc. 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 version 2 as published
+# by the Free Software Foundation.
+#
+# 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, write to the Free Software Foundation, Inc.,
+# 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA
+#
+# Contact information: Zmanda Inc., 465 S. Mathilda Ave., Suite 300
+# Sunnyvale, CA 94086, USA, or: http://www.zmanda.com
+
+=head1 NAME
+
+Amanda::Taper::Worker
+
+=head1 DESCRIPTION
+
+This package is a component of the Amanda taper, and is not intended for use by
+other scripts or applications.
+
+This package interface between L<Amanda::Taper::Controller> and L<Amanda::Taper::Scribe>.
+
+The worker use an L<Amanda::Taper::Scribe> object to execute the request
+received from the L<Amanda::Taper::Controller>.
+
+=cut
+
+use lib '@amperldir@';
+use strict;
+use warnings;
+
+package Amanda::Taper::Worker;
+
+use POSIX qw( :errno_h );
+use Amanda::Changer;
+use Amanda::Config qw( :getconf config_dir_relative );
+use Amanda::Debug qw( :logging );
+use Amanda::Header;
+use Amanda::Holding;
+use Amanda::MainLoop qw( :GIOCondition );
+use Amanda::MainLoop;
+use Amanda::Taper::Protocol;
+use Amanda::Taper::Scan;
+use Amanda::Taper::Scribe qw( get_splitting_args_from_config );
+use Amanda::Logfile qw( :logtype_t log_add make_stats );
+use Amanda::Xfer qw( :constants );
+use Amanda::Util qw( quote_string );
+use Amanda::Tapelist;
+use File::Temp;
+
+use base qw( Amanda::Taper::Scribe::Feedback );
+
+our $tape_num = 0;
+
+sub new {
+ my $class = shift;
+ my $worker_name = shift;
+ my $controller = shift;
+ my $write_timestamp = shift;
+
+ my $self = bless {
+ state => "init",
+ worker_name => $worker_name,
+ controller => $controller,
+ scribe => undef,
+ timestamp => $write_timestamp,
+
+ # filled in when a write starts:
+ xfer => undef,
+ xfer_source => undef,
+ xfer_dest => undef,
+ handle => undef,
+ hostname => undef,
+ diskname => undef,
+ datestamp => undef,
+ level => undef,
+ header => undef,
+ doing_port_write => undef,
+ input_errors => [],
+
+ # periodic status updates
+ timer => undef,
+ status_filename => undef,
+ status_fh => undef,
+
+ # filled in after the header is available
+ header => undef,
+
+ # filled in when a new tape is started:
+ label => undef
+ }, $class;
+
+ my $scribe = Amanda::Taper::Scribe->new(
+ taperscan => $controller->{'taperscan'},
+ feedback => $self,
+ debug => $Amanda::Config::debug_taper,
+ eject_volume => getconf($CNF_EJECT_VOLUME));
+
+ $self->{'scribe'} = $scribe;
+ $self->{'scribe'}->start(write_timestamp => $write_timestamp,
+ finished_cb => sub { $self->_scribe_started_cb(@_); });
+
+ return $self;
+}
+
+# called when the scribe is fully started up and ready to go
+sub _scribe_started_cb {
+ my $self = shift;
+ my ($err) = @_;
+
+ if ($err) {
+ $self->{'controller'}->{'proto'}->send(Amanda::Taper::Protocol::TAPE_ERROR,
+ worker_name => $self->{'worker_name'},
+ message => "$err");
+ $self->{'state'} = "error";
+
+ # log the error (note that the message is intentionally not quoted)
+ log_add($L_ERROR, "no-tape error [$err]");
+
+ } else {
+ $self->{'controller'}->{'proto'}->send(Amanda::Taper::Protocol::TAPER_OK,
+ worker_name => $self->{'worker_name'});
+ $self->{'state'} = "idle";
+ }
+}
+
+
+sub FILE_WRITE {
+ my $self = shift;
+ my ($msgtype, %params) = @_;
+ $self->_assert_in_state("idle") or return;
+
+ $self->{'doing_port_write'} = 0;
+
+ $self->setup_and_start_dump($msgtype,
+ dump_cb => sub { $self->dump_cb(@_); },
+ %params);
+}
+
+sub PORT_WRITE {
+ my $self = shift;
+ my ($msgtype, %params) = @_;
+
+ my $read_cb;
+
+ $self->_assert_in_state("idle") or return;
+
+ $self->{'doing_port_write'} = 1;
+
+ $self->setup_and_start_dump($msgtype,
+ dump_cb => sub { $self->dump_cb(@_); },
+ %params);
+}
+
+sub START_SCAN {
+ my $self = shift;
+ my ($msgtype, %params) = @_;
+
+ $self->{'scribe'}->start_scan(undef);
+}
+
+sub NEW_TAPE {
+ my $self = shift;
+ my ($msgtype, %params) = @_;
+
+ $self->_assert_in_state("writing") or return;
+
+ $self->{'perm_cb'}->(allow => 1);
+}
+
+sub NO_NEW_TAPE {
+ my $self = shift;
+ my ($msgtype, %params) = @_;
+
+ $self->_assert_in_state("writing") or return;
+
+ # log the error (note that the message is intentionally not quoted)
+ log_add($L_ERROR, "no-tape config [$params{reason}]");
+
+ $self->{'perm_cb'}->(cause => "config", message => $params{'reason'});
+}
+
+sub TAKE_SCRIBE_FROM {
+ my $self = shift;
+ my ($worker1, $msgtype, %params) = @_;
+
+ $self->_assert_in_state("writing") or return;
+ $worker1->_assert_in_state("idle") or return;
+
+ my $scribe = $self->{'scribe'};
+ my $scribe1 = $worker1->{'scribe'};
+ $self->{'scribe'} = $scribe1;
+ $worker1->{'scribe'} = $scribe;
+ # Change the callback to call the new scribe
+ $self->{'xfer'}->set_callback(sub {
+ my ($src, $msg, $xfer) = @_;
+ $scribe1->handle_xmsg($src, $msg, $xfer);
+
+ # if this is an error message that's not from the scribe's element, then
+ # we'll need to keep track of it ourselves
+ if ($msg->{'type'} == $XMSG_ERROR and $msg->{'elt'} != $self->{'xfer_dest'}) {
+ push @{$self->{'input_errors'}}, $msg->{'message'};
+ }
+ });
+
+ $self->{'label'} = $worker1->{'label'};
+ $self->{'perm_cb'}->(scribe => $scribe1);
+ delete $worker1->{'scribe'};
+ $worker1->{'state'} = 'error';
+ $scribe->quit(finished_cb => sub {});
+ }
+
+sub DONE {
+ my $self = shift;
+ my ($msgtype, %params) = @_;
+
+ $self->_assert_in_state("writing") or return;
+ $self->{'dumper_status'} = "DONE";
+ $self->{'orig_kb'} = $params{'orig_kb'};
+ if (defined $self->{'result'}) {
+ $self->result_cb(undef);
+ }
+}
+
+sub FAILED {
+ my $self = shift;
+ my ($msgtype, %params) = @_;
+
+ $self->_assert_in_state("writing") or return;
+
+ $self->{'dumper_status'} = "FAILED";
+ if (defined $self->{'result'}) {
+ $self->result_cb(undef);
+ }
+}
+
+sub CLOSE_VOLUME {
+ my $self = shift;
+ my ($msgtype, %params) = @_;
+
+ $self->_assert_in_state("idle") or return;
+
+ $self->{'scribe'}->close_volume();
+}
+
+sub result_cb {
+ my $self = shift;
+ my %params = %{$self->{'dump_params'}};
+ my $msgtype;
+ my $logtype;
+
+ if ($params{'result'} eq 'DONE') {
+ if (!$self->{'doing_port_write'} or $self->{'dumper_status'} eq "DONE") {
+ $msgtype = Amanda::Taper::Protocol::DONE;
+ $logtype = $L_DONE;
+ } else {
+ $msgtype = Amanda::Taper::Protocol::DONE;
+ $logtype = $L_PARTIAL;
+ }
+ } elsif ($params{'result'} eq 'PARTIAL') {
+ $msgtype = Amanda::Taper::Protocol::PARTIAL;
+ $logtype = $L_PARTIAL;
+ } elsif ($params{'result'} eq 'FAILED') {
+ $msgtype = Amanda::Taper::Protocol::FAILED;
+ $logtype = $L_FAIL;
+ }
+
+ if ($self->{timer}) {
+ $self->{timer}->remove();
+ undef $self->{timer};
+ $self->{status_fh}->close();
+ undef $self->{status_fh};
+ unlink($self->{status_filename});
+ undef $self->{status_filename};
+ }
+
+ # note that we use total_duration here, which is the total time between
+ # start_dump and dump_cb, so the kps generated here is much less than the
+ # actual tape write speed. Think of this as the *taper* speed, rather than
+ # the *tape* speed.
+ my $stats = make_stats($params{'size'}, $params{'total_duration'}, $self->{'orig_kb'});
+
+ # consider this a config-derived failure only if there were no errors
+ my $failure_from = (@{$params{'device_errors'}})? 'error' : 'config';
+
+ my @all_messages = (@{$params{'device_errors'}}, @{$self->{'input_errors'}});
+ push @all_messages, $params{'config_denial_message'} if $params{'config_denial_message'};
+ my $msg = quote_string(join("; ", @all_messages));
+
+ # write a DONE/PARTIAL/FAIL log line
+ if ($logtype == $L_FAIL) {
+ log_add($L_FAIL, sprintf("%s %s %s %s %s %s",
+ quote_string($self->{'hostname'}.""), # " is required for SWIG..
+ quote_string($self->{'diskname'}.""),
+ $self->{'datestamp'},
+ $self->{'level'},
+ $failure_from,
+ $msg));
+ } else {
+ log_add($logtype, sprintf("%s %s %s %s %s %s%s",
+ quote_string($self->{'hostname'}.""), # " is required for SWIG..
+ quote_string($self->{'diskname'}.""),
+ $self->{'datestamp'},
+ $params{'nparts'},
+ $self->{'level'},
+ $stats,
+ ($logtype == $L_PARTIAL and @all_messages)? " $msg" : ""));
+ }
+
+ # and send a message back to the driver
+ my %msg_params = (
+ handle => $self->{'handle'},
+ );
+
+ # reflect errors in our own elements in INPUT-ERROR or INPUT-GOOD
+ if (@{$self->{'input_errors'}}) {
+ $msg_params{'input'} = 'INPUT-ERROR';
+ $msg_params{'inputerr'} = join("; ", @{$self->{'input_errors'}});
+ } else {
+ $msg_params{'input'} = 'INPUT-GOOD';
+ $msg_params{'inputerr'} = '';
+ }
+
+ # and errors from the scribe in TAPE-ERROR or TAPE-GOOD
+ if (@{$params{'device_errors'}}) {
+ $msg_params{'taper'} = 'TAPE-ERROR';
+ $msg_params{'tapererr'} = join("; ", @{$params{'device_errors'}});
+ } elsif ($params{'config_denial_message'}) {
+ $msg_params{'taper'} = 'TAPE-ERROR';
+ $msg_params{'tapererr'} = $params{'config_denial_message'};
+ } else {
+ $msg_params{'taper'} = 'TAPE-GOOD';
+ $msg_params{'tapererr'} = '';
+ }
+
+ if ($msgtype ne Amanda::Taper::Protocol::FAILED) {
+ $msg_params{'stats'} = $stats;
+ }
+
+ # reset things to 'idle' before sending the message
+ $self->{'xfer'} = undef;
+ $self->{'xfer_source'} = undef;
+ $self->{'xfer_dest'} = undef;
+ $self->{'handle'} = undef;
+ $self->{'header'} = undef;
+ $self->{'hostname'} = undef;
+ $self->{'diskname'} = undef;
+ $self->{'datestamp'} = undef;
+ $self->{'level'} = undef;
+ $self->{'header'} = undef;
+ $self->{'state'} = 'idle';
+ delete $self->{'result'};
+ delete $self->{'dumper_status'};
+ delete $self->{'dump_params'};
+
+ $self->{'controller'}->{'proto'}->send($msgtype, %msg_params);
+}
+
+
+##
+# Scribe feedback
+
+sub request_volume_permission {
+ my $self = shift;
+ my %params = @_;
+
+ $self->{'perm_cb'} = $params{'perm_cb'};
+ # and send the request to the driver
+ $self->{'controller'}->{'proto'}->send(Amanda::Taper::Protocol::REQUEST_NEW_TAPE,
+ handle => $self->{'handle'});
+}
+
+sub scribe_notif_new_tape {
+ my $self = shift;
+ my %params = @_;
+
+ # TODO: if $params{error} is set, report it back to the driver
+ # (this will be a change to the protocol)
+ log_add($L_INFO, "$params{'error'}") if defined $params{'error'};
+
+ if ($params{'volume_label'}) {
+ $self->{'label'} = $params{'volume_label'};
+
+ # add to the trace log
+ log_add($L_START, sprintf("datestamp %s label %s tape %s",
+ $self->{'timestamp'},
+ quote_string($self->{'label'}),
+ ++$tape_num));
+
+ # and the amdump log
+ print STDERR "taper: wrote label '$self->{label}'\n";
+
+ # and inform the driver
+ $self->{'controller'}->{'proto'}->send(Amanda::Taper::Protocol::NEW_TAPE,
+ handle => $self->{'handle'},
+ label => $params{'volume_label'});
+ } else {
+ $self->{'label'} = undef;
+
+ $self->{'controller'}->{'proto'}->send(Amanda::Taper::Protocol::NO_NEW_TAPE,
+ handle => $self->{'handle'});
+ }
+}
+
+sub scribe_notif_part_done {
+ my $self = shift;
+ my %params = @_;
+
+ $self->_assert_in_state("writing") or return;
+
+ my $stats = make_stats($params{'size'}, $params{'duration'}, $self->{'orig_kb'});
+
+ # log the part, using PART or PARTPARTIAL
+ my $logbase = sprintf("%s %s %s %s %s %s/%s %s %s",
+ quote_string($self->{'label'}),
+ $params{'fileno'},
+ quote_string($self->{'header'}->{'name'}.""), # " is required for SWIG..
+ quote_string($self->{'header'}->{'disk'}.""),
+ $self->{'datestamp'},
+ $params{'partnum'}, -1, # totalparts is always -1
+ $self->{'level'},
+ $stats);
+ if ($params{'successful'}) {
+ log_add($L_PART, $logbase);
+ } else {
+ log_add($L_PARTPARTIAL, "$logbase \"No space left on device\"");
+ }
+
+ # only send a PARTDONE if it was successful
+ if ($params{'successful'}) {
+ $self->{'controller'}->{'proto'}->send(Amanda::Taper::Protocol::PARTDONE,
+ handle => $self->{'handle'},
+ label => $self->{'label'},
+ fileno => $params{'fileno'},
+ stats => $stats,
+ kb => $params{'size'} / 1024);
+ }
+}
+
+sub scribe_notif_log_info {
+ my $self = shift;
+ my %params = @_;
+
+ debug("$params{'message'}");
+ log_add($L_INFO, "$params{'message'}");
+}
+
+##
+# Utilities
+
+sub _assert_in_state {
+ my $self = shift;
+ my ($state) = @_;
+ if ($self->{'state'} eq $state) {
+ return 1;
+ } else {
+ $self->{'controller'}->{'proto'}->send(Amanda::Taper::Protocol::BAD_COMMAND,
+ message => "command not appropriate in state '$self->{state}' : '$state'");
+ return 0;
+ }
+}
+
+sub create_status_file {
+ my $self = shift;
+
+ # create temporary file
+ ($self->{status_fh}, $self->{status_filename}) =
+ File::Temp::tempfile("taper_status_file_XXXXXX",
+ DIR => $Amanda::Paths::AMANDA_TMPDIR,
+ UNLINK => 1);
+
+ # tell amstatus about it by writing it to the dump log
+ my $qdisk = Amanda::Util::quote_string($self->{'diskname'});
+ my $qhost = Amanda::Util::quote_string($self->{'hostname'});
+ print STDERR "taper: status file $qhost $qdisk:" .
+ "$self->{status_filename}\n";
+ print {$self->{status_fh}} "0";
+
+ # create timer callback, firing every 5s (=5000msec)
+ $self->{timer} = Amanda::MainLoop::timeout_source(5000);
+ $self->{timer}->set_callback(sub {
+ my $size = $self->{scribe}->get_bytes_written();
+ seek $self->{status_fh}, 0, 0;
+ print {$self->{status_fh}} $size;
+ $self->{status_fh}->flush();
+ });
+}
+
+sub send_port_and_get_header {
+ my $self = shift;
+ my ($finished_cb) = @_;
+
+ my $header_xfer;
+ my ($xsrc, $xdst);
+ my $errmsg;
+
+ my $steps = define_steps
+ cb_ref => \$finished_cb;
+
+ step send_port => sub {
+ # get the ip:port pairs for the data connection from the data xfer source,
+ # which should be an Amanda::Xfer::Source::DirectTCPListen
+ my $data_addrs = $self->{'xfer_source'}->get_addrs();
+ $data_addrs = join ";", map { $_->[0] . ':' . $_->[1] } @$data_addrs;
+
+ # and set up an xfer for the header, too, using DirectTCP as an easy
+ # way to implement a listen/accept/read process. Note that this does
+ # not enforce a maximum size, so this portion of Amanda at least can
+ # handle any size header
+ ($xsrc, $xdst) = (
+ Amanda::Xfer::Source::DirectTCPListen->new(),
+ Amanda::Xfer::Dest::Buffer->new(0));
+ $header_xfer = Amanda::Xfer->new([$xsrc, $xdst]);
+ $header_xfer->start($steps->{'header_xfer_xmsg_cb'});
+
+ my $header_addrs = $xsrc->get_addrs();
+ my $header_port = $header_addrs->[0][1];
+
+ # and tell the driver which ports we're listening on
+ $self->{'controller'}->{'proto'}->send(Amanda::Taper::Protocol::PORT,
+ worker_name => $self->{'worker_name'},
+ handle => $self->{'handle'},
+ port => $header_port,
+ ipports => $data_addrs);
+ };
+
+ step header_xfer_xmsg_cb => sub {
+ my ($src, $xmsg, $xfer) = @_;
+ if ($xmsg->{'type'} == $XMSG_INFO) {
+ info($xmsg->{'message'});
+ } elsif ($xmsg->{'type'} == $XMSG_ERROR) {
+ $errmsg = $xmsg->{'message'};
+ } elsif ($xmsg->{'type'} == $XMSG_DONE) {
+ if ($errmsg) {
+ $finished_cb->($errmsg);
+ } else {
+ $steps->{'got_header'}->();
+ }
+ }
+ };
+
+ step got_header => sub {
+ my $hdr_buf = $xdst->get();
+
+ # close stuff up
+ $header_xfer = $xsrc = $xdst = undef;
+
+ if (!defined $hdr_buf) {
+ return $finished_cb->("Got empty header");
+ }
+
+ # parse the header, finally!
+ $self->{'header'} = Amanda::Header->from_string($hdr_buf);
+
+ $finished_cb->(undef);
+ };
+}
+
+# do the work of starting a new xfer; this contains the code common to
+# msg_PORT_WRITE and msg_FILE_WRITE.
+sub setup_and_start_dump {
+ my $self = shift;
+ my ($msgtype, %params) = @_;
+ my %get_xfer_dest_args;
+
+ # setting up the dump is a bit complex, due to the requirements of
+ # a directtcp port_write. This function:
+ # 1. creates and starts a transfer (make_xfer)
+ # 2. gets the header
+ # 3. calls the scribe's start_dump method with the new header
+
+ my $steps = define_steps
+ cb_ref => \$params{'dump_cb'};
+
+ step setup => sub {
+ $self->{'handle'} = $params{'handle'};
+ $self->{'hostname'} = $params{'hostname'};
+ $self->{'diskname'} = $params{'diskname'};
+ $self->{'datestamp'} = $params{'datestamp'};
+ $self->{'level'} = $params{'level'};
+ $self->{'header'} = undef; # no header yet
+ $self->{'orig_kb'} = $params{'orig_kb'};
+ $self->{'input_errors'} = [];
+
+ if ($msgtype eq Amanda::Taper::Protocol::PORT_WRITE &&
+ (my $err = $self->{'scribe'}->check_data_path($params{'data_path'}))) {
+ return $params{'dump_cb'}->(
+ result => "FAILED",
+ device_errors => [ ['error', "$err"] ],
+ size => 0,
+ duration => 0.0,
+ total_duration => 0);
+ }
+ $steps->{'process_args'}->();
+ };
+
+ step process_args => sub {
+ # extract the splitting-related parameters, stripping out empty strings
+ my %splitting_args = map {
+ ($params{$_} ne '')? ($_, $params{$_}) : ()
+ } qw(
+ dle_tape_splitsize dle_split_diskbuffer dle_fallback_splitsize dle_allow_split
+ part_size part_cache_type part_cache_dir part_cache_max_size
+ );
+
+ # convert numeric values to BigInts
+ for (qw(dle_tape_splitsize dle_fallback_splitsize part_size part_cache_max_size)) {
+ $splitting_args{$_} = Math::BigInt->new($splitting_args{$_})
+ if (exists $splitting_args{$_});
+ }
+
+ my $device = $self->{'scribe'}->get_device();
+ if (!defined $device) {
+ die "no device is available to create an xfer_dest";
+ }
+ $splitting_args{'leom_supported'} = $device->property_get("leom");
+ # and convert those to get_xfer_dest args
+ %get_xfer_dest_args = get_splitting_args_from_config(
+ %splitting_args);
+ $get_xfer_dest_args{'max_memory'} = getconf($CNF_DEVICE_OUTPUT_BUFFER_SIZE);
+ if (!getconf_seen($CNF_DEVICE_OUTPUT_BUFFER_SIZE)) {
+ my $block_size4 = $device->block_size * 4;
+ if ($block_size4 > $get_xfer_dest_args{'max_memory'}) {
+ $get_xfer_dest_args{'max_memory'} = $block_size4;
+ }
+ }
+ $device = undef;
+ $get_xfer_dest_args{'can_cache_inform'} = ($msgtype eq Amanda::Taper::Protocol::FILE_WRITE and $get_xfer_dest_args{'allow_split'});
+
+ # if we're unable to fulfill the user's splitting needs, we can still give
+ # the dump a shot - but we'll warn them about the problem
+ if ($get_xfer_dest_args{'warning'}) {
+ log_add($L_WARNING, sprintf("%s:%s: %s",
+ $params{'hostname'}, $params{'diskname'},
+ $get_xfer_dest_args{'warning'}));
+ delete $get_xfer_dest_args{'warning'};
+ }
+
+ $steps->{'make_xfer'}->();
+ };
+
+ step make_xfer => sub {
+ $self->_assert_in_state("idle") or return;
+ $self->{'state'} = 'making_xfer';
+
+ $self->{'xfer_dest'} = $self->{'scribe'}->get_xfer_dest(%get_xfer_dest_args);
+
+ my $xfer_source;
+ if ($msgtype eq Amanda::Taper::Protocol::PORT_WRITE) {
+ $xfer_source = Amanda::Xfer::Source::DirectTCPListen->new();
+ } else {
+ $xfer_source = Amanda::Xfer::Source::Holding->new($params{'filename'});
+ }
+ $self->{'xfer_source'} = $xfer_source;
+
+ $self->{'xfer'} = Amanda::Xfer->new([$xfer_source, $self->{'xfer_dest'}]);
+ $self->{'xfer'}->start(sub {
+ my ($src, $msg, $xfer) = @_;
+ $self->{'scribe'}->handle_xmsg($src, $msg, $xfer);
+
+ # if this is an error message that's not from the scribe's element, then
+ # we'll need to keep track of it ourselves
+ if ($msg->{'type'} == $XMSG_ERROR and $msg->{'elt'} != $self->{'xfer_dest'}) {
+ push @{$self->{'input_errors'}}, $msg->{'message'};
+ }
+ });
+
+ # we've started the xfer now, but the destination won't actually write
+ # any data until we call start_dump. And we'll need a header for that.
+
+ $steps->{'get_header'}->();
+ };
+
+ step get_header => sub {
+ $self->_assert_in_state("making_xfer") or return;
+ $self->{'state'} = 'getting_header';
+
+ if ($msgtype eq Amanda::Taper::Protocol::FILE_WRITE) {
+ # getting the header is easy for FILE-WRITE..
+ my $hdr = $self->{'header'} = Amanda::Holding::get_header($params{'filename'});
+
+ # stip out header fields we don't need
+ $hdr->{'cont_filename'} = '';
+
+ if (!defined $hdr || $hdr->{'type'} != $Amanda::Header::F_DUMPFILE) {
+ die("Could not read header from '$params{filename}'");
+ }
+ $steps->{'start_dump'}->(undef);
+ } else {
+ # ..but quite a bit harder for PORT-WRITE; this method will send the
+ # proper PORT command, then read the header from the dumper and parse
+ # it, placing the result in $self->{'header'}
+ $self->send_port_and_get_header($steps->{'start_dump'});
+ }
+ };
+
+ step start_dump => sub {
+ my ($err) = @_;
+
+ $self->_assert_in_state("getting_header") or return;
+ $self->{'state'} = 'writing';
+
+ # if $err is set, cancel the dump, treating it as a input error
+ if ($err) {
+ push @{$self->{'input_errors'}}, $err;
+ return $self->{'scribe'}->cancel_dump(
+ xfer => $self->{'xfer'},
+ dump_cb => $params{'dump_cb'});
+ }
+
+ # sanity check the header..
+ my $hdr = $self->{'header'};
+ if ($hdr->{'dumplevel'} != $params{'level'}
+ or $hdr->{'name'} ne $params{'hostname'}
+ or $hdr->{'disk'} ne $params{'diskname'}
+ or $hdr->{'datestamp'} ne $params{'datestamp'}) {
+ die("Header of dumpfile does not match command from driver");
+ }
+
+ # start producing status
+ $self->create_status_file();
+
+ # and fix it up before writing it
+ $hdr->{'totalparts'} = -1;
+ $hdr->{'type'} = $Amanda::Header::F_SPLIT_DUMPFILE;
+
+ $self->{'scribe'}->start_dump(
+ xfer => $self->{'xfer'},
+ dump_header => $hdr,
+ dump_cb => $params{'dump_cb'});
+ };
+}
+
+sub dump_cb {
+ my $self = shift;
+ my %params = @_;
+
+ $self->{'dump_params'} = \%params;
+ $self->{'result'} = $params{'result'};
+
+ # if we need to the dumper status (to differentiate a dropped network
+ # connection from a normal EOF) and have not done so yet, then send a
+ # DUMPER_STATUS message and re-call this method (dump_cb) with the result.
+ if ($params{'result'} eq "DONE"
+ and $self->{'doing_port_write'}
+ and !exists $self->{'dumper_status'}) {
+ my $controller = $self->{'controller'};
+ my $proto = $controller->{'proto'};
+ my $handle = $self->{'handle'};
+ $proto->send(Amanda::Taper::Protocol::DUMPER_STATUS,
+ handle => "$handle");
+ } else {
+ $self->result_cb();
+ }
+}
+
+1;
projects / debian / amanda / blobdiff