git.gag.com Git - debian/amanda/blob - perl/Amanda/Taper/Worker.pm

   1 #! @PERL@
   2 # Copyright (c) 2009, 2010 Zmanda Inc.  All Rights Reserved.
   3 #
   4 # This program is free software; you can redistribute it and/or modify it
   5 # under the terms of the GNU General Public License version 2 as published
   6 # by the Free Software Foundation.
   7 #
   8 # This program is distributed in the hope that it will be useful, but
   9 # WITHOUT ANY WARRANTY; without even the implied warranty of MERCHANTABILITY
  10 # or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU General Public License
  11 # for more details.
  12 #
  13 # You should have received a copy of the GNU General Public License along
  14 # with this program; if not, write to the Free Software Foundation, Inc.,
  15 # 59 Temple Place, Suite 330, Boston, MA  02111-1307 USA
  16 #
  17 # Contact information: Zmanda Inc., 465 S. Mathilda Ave., Suite 300
  18 # Sunnyvale, CA 94086, USA, or: http://www.zmanda.com
  19
  20 =head1 NAME
  21
  22 Amanda::Taper::Worker
  23
  24 =head1 DESCRIPTION
  25
  26 This package is a component of the Amanda taper, and is not intended for use by
  27 other scripts or applications.
  28
  29 This package interface between L<Amanda::Taper::Controller> and L<Amanda::Taper::Scribe>.
  30
  31 The worker use an L<Amanda::Taper::Scribe> object to execute the request
  32 received from the L<Amanda::Taper::Controller>.
  33
  34 =cut
  35
  36 use lib '@amperldir@';
  37 use strict;
  38 use warnings;
  39
  40 package Amanda::Taper::Worker;
  41
  42 use POSIX qw( :errno_h );
  43 use Amanda::Changer;
  44 use Amanda::Config qw( :getconf config_dir_relative );
  45 use Amanda::Header;
  46 use Amanda::Holding;
  47 use Amanda::MainLoop qw( :GIOCondition );
  48 use Amanda::MainLoop;
  49 use Amanda::Taper::Protocol;
  50 use Amanda::Taper::Scan;
  51 use Amanda::Taper::Scribe qw( get_splitting_args_from_config );
  52 use Amanda::Logfile qw( :logtype_t log_add make_stats );
  53 use Amanda::Xfer qw( :constants );
  54 use Amanda::Util qw( quote_string );
  55 use Amanda::Tapelist;
  56 use File::Temp;
  57
  58 use base qw( Amanda::Taper::Scribe::Feedback );
  59
  60 our $tape_num = 0;
  61
  62 sub new {
  63     my $class           = shift;
  64     my $worker_name     = shift;
  65     my $controller      = shift;
  66     my $write_timestamp = shift;
  67
  68     my $self = bless {
  69         state       => "init",
  70         worker_name => $worker_name,
  71         controller  => $controller,
  72         scribe      => undef,
  73         timestamp   => $write_timestamp,
  74
  75         # filled in when a write starts:
  76         xfer => undef,
  77         xfer_source => undef,
  78         xfer_dest => undef,
  79         handle => undef,
  80         hostname => undef,
  81         diskname => undef,
  82         datestamp => undef,
  83         level => undef,
  84         header => undef,
  85         doing_port_write => undef,
  86         input_errors => [],
  87
  88         # periodic status updates
  89         timer => undef,
  90         status_filename => undef,
  91         status_fh => undef,
  92
  93         # filled in after the header is available
  94         header => undef,
  95
  96         # filled in when a new tape is started:
  97         label => undef
  98     }, $class;
  99
 100     my $scribe = Amanda::Taper::Scribe->new(
 101         taperscan => $controller->{'taperscan'},
 102         feedback => $self,
 103         debug => $Amanda::Config::debug_taper);
 104
 105     $self->{'scribe'} = $scribe;
 106     $self->{'scribe'}->start(write_timestamp => $write_timestamp,
 107         finished_cb => sub { $self->_scribe_started_cb(@_); });
 108
 109     return $self;
 110 }
 111
 112 # called when the scribe is fully started up and ready to go
 113 sub _scribe_started_cb {
 114     my $self = shift;
 115     my ($err) = @_;
 116
 117     if ($err) {
 118         $self->{'controller'}->{'proto'}->send(Amanda::Taper::Protocol::TAPE_ERROR,
 119                 worker_name  => $self->{'worker_name'},
 120                 message => "$err");
 121         $self->{'state'} = "error";
 122
 123         # log the error (note that the message is intentionally not quoted)
 124         log_add($L_ERROR, "no-tape error [$err]");
 125
 126     } else {
 127         $self->{'controller'}->{'proto'}->send(Amanda::Taper::Protocol::TAPER_OK,
 128                 worker_name => $self->{'worker_name'});
 129         $self->{'state'} = "idle";
 130     }
 131 }
 132
 133
 134 sub FILE_WRITE {
 135     my $self = shift;
 136     my ($msgtype, %params) = @_;
 137     $self->_assert_in_state("idle") or return;
 138
 139     $self->{'doing_port_write'} = 0;
 140
 141     $self->setup_and_start_dump($msgtype,
 142         dump_cb => sub { $self->dump_cb(@_); },
 143         %params);
 144 }
 145
 146 sub PORT_WRITE {
 147     my $self = shift;
 148     my ($msgtype, %params) = @_;
 149
 150     my $read_cb;
 151
 152     $self->_assert_in_state("idle") or return;
 153
 154     $self->{'doing_port_write'} = 1;
 155
 156     $self->setup_and_start_dump($msgtype,
 157         dump_cb => sub { $self->dump_cb(@_); },
 158         %params);
 159 }
 160
 161 sub START_SCAN {
 162     my $self = shift;
 163     my ($msgtype, %params) = @_;
 164
 165     $self->{'scribe'}->start_scan(undef);
 166 }
 167
 168 sub NEW_TAPE {
 169     my $self = shift;
 170     my ($msgtype, %params) = @_;
 171
 172     $self->_assert_in_state("writing") or return;
 173
 174     $self->{'perm_cb'}->(allow => 1);
 175 }
 176
 177 sub NO_NEW_TAPE {
 178     my $self = shift;
 179     my ($msgtype, %params) = @_;
 180
 181     $self->_assert_in_state("writing") or return;
 182
 183     # log the error (note that the message is intentionally not quoted)
 184     log_add($L_ERROR, "no-tape config [$params{reason}]");
 185
 186     $self->{'perm_cb'}->(cause => "config", message => $params{'reason'});
 187 }
 188
 189 sub TAKE_SCRIBE_FROM {
 190     my $self = shift;
 191     my ($worker1, $msgtype, %params) = @_;
 192
 193     $self->_assert_in_state("writing") or return;
 194     $worker1->_assert_in_state("idle") or return;
 195
 196     my $scribe = $self->{'scribe'};
 197     my $scribe1 = $worker1->{'scribe'};
 198     $self->{'scribe'} = $scribe1;
 199     $worker1->{'scribe'} = $scribe;
 200     # Change the callback to call the new scribe
 201     $self->{'xfer'}->set_callback(sub {
 202         my ($src, $msg, $xfer) = @_;
 203         $scribe1->handle_xmsg($src, $msg, $xfer);
 204
 205         # if this is an error message that's not from the scribe's element, then
 206         # we'll need to keep track of it ourselves
 207         if ($msg->{'type'} == $XMSG_ERROR and $msg->{'elt'} != $self->{'xfer_dest'}) {
 208             push @{$self->{'input_errors'}}, $msg->{'message'};
 209         }
 210     });
 211
 212     $self->{'label'} = $worker1->{'label'};
 213     $self->{'perm_cb'}->(scribe => $scribe1);
 214     delete $worker1->{'scribe'};
 215     $worker1->{'state'} = 'error';
 216     $scribe->quit(finished_cb => sub {});
 217  }
 218
 219 sub DONE {
 220     my $self = shift;
 221     my ($msgtype, %params) = @_;
 222
 223     $self->_assert_in_state("writing") or return;
 224     $self->{'dumper_status'} = "DONE";
 225     $self->{'orig_kb'} = $params{'orig_kb'};
 226     if (defined $self->{'result'}) {
 227         $self->result_cb(undef);
 228     }
 229 }
 230
 231 sub FAILED {
 232     my $self = shift;
 233     my ($msgtype, %params) = @_;
 234
 235     $self->_assert_in_state("writing") or return;
 236
 237     $self->{'dumper_status'} = "FAILED";
 238     if (defined $self->{'result'}) {
 239         $self->result_cb(undef);
 240     }
 241 }
 242
 243 sub result_cb {
 244     my $self = shift;
 245     my %params = %{$self->{'dump_params'}};
 246     my $msgtype;
 247     my $logtype;
 248
 249     if ($params{'result'} eq 'DONE') {
 250         if (!$self->{'doing_port_write'} or $self->{'dumper_status'} eq "DONE") {
 251             $msgtype = Amanda::Taper::Protocol::DONE;
 252             $logtype = $L_DONE;
 253         } else {
 254             $msgtype = Amanda::Taper::Protocol::DONE;
 255             $logtype = $L_PARTIAL;
 256         }
 257     } elsif ($params{'result'} eq 'PARTIAL') {
 258         $msgtype = Amanda::Taper::Protocol::PARTIAL;
 259         $logtype = $L_PARTIAL;
 260     } elsif ($params{'result'} eq 'FAILED') {
 261         $msgtype = Amanda::Taper::Protocol::FAILED;
 262         $logtype = $L_FAIL;
 263     }
 264
 265     if ($self->{timer}) {
 266         $self->{timer}->remove();
 267         undef $self->{timer};
 268         $self->{status_fh}->close();
 269         undef $self->{status_fh};
 270         unlink($self->{status_filename});
 271         undef $self->{status_filename};
 272     }
 273
 274     # note that we use total_duration here, which is the total time between
 275     # start_dump and dump_cb, so the kps generated here is much less than the
 276     # actual tape write speed.  Think of this as the *taper* speed, rather than
 277     # the *tape* speed.
 278     my $stats = make_stats($params{'size'}, $params{'total_duration'}, $self->{'orig_kb'});
 279
 280     # consider this a config-derived failure only if there were no errors
 281     my $failure_from = (@{$params{'device_errors'}})?  'error' : 'config';
 282
 283     my @all_messages = (@{$params{'device_errors'}}, @{$self->{'input_errors'}});
 284     push @all_messages, $params{'config_denial_message'} if $params{'config_denial_message'};
 285     my $msg = quote_string(join("; ", @all_messages));
 286
 287     # write a DONE/PARTIAL/FAIL log line
 288     if ($logtype == $L_FAIL) {
 289         log_add($L_FAIL, sprintf("%s %s %s %s %s %s",
 290             quote_string($self->{'hostname'}.""), # " is required for SWIG..
 291             quote_string($self->{'diskname'}.""),
 292             $self->{'datestamp'},
 293             $self->{'level'},
 294             $failure_from,
 295             $msg));
 296     } else {
 297         log_add($logtype, sprintf("%s %s %s %s %s %s%s",
 298             quote_string($self->{'hostname'}.""), # " is required for SWIG..
 299             quote_string($self->{'diskname'}.""),
 300             $self->{'datestamp'},
 301             $params{'nparts'},
 302             $self->{'level'},
 303             $stats,
 304             ($logtype == $L_PARTIAL and @all_messages)? " $msg" : ""));
 305     }
 306
 307     # and send a message back to the driver
 308     my %msg_params = (
 309         handle => $self->{'handle'},
 310     );
 311
 312     # reflect errors in our own elements in INPUT-ERROR or INPUT-GOOD
 313     if (@{$self->{'input_errors'}}) {
 314         $msg_params{'input'} = 'INPUT-ERROR';
 315         $msg_params{'inputerr'} = join("; ", @{$self->{'input_errors'}});
 316     } else {
 317         $msg_params{'input'} = 'INPUT-GOOD';
 318         $msg_params{'inputerr'} = '';
 319     }
 320
 321     # and errors from the scribe in TAPE-ERROR or TAPE-GOOD
 322     if (@{$params{'device_errors'}}) {
 323         $msg_params{'taper'} = 'TAPE-ERROR';
 324         $msg_params{'tapererr'} = join("; ", @{$params{'device_errors'}});
 325     } elsif ($params{'config_denial_message'}) {
 326         $msg_params{'taper'} = 'TAPE-ERROR';
 327         $msg_params{'tapererr'} = $params{'config_denial_message'};
 328     } else {
 329         $msg_params{'taper'} = 'TAPE-GOOD';
 330         $msg_params{'tapererr'} = '';
 331     }
 332
 333     if ($msgtype ne Amanda::Taper::Protocol::FAILED) {
 334         $msg_params{'stats'} = $stats;
 335     }
 336
 337     # reset things to 'idle' before sending the message
 338     $self->{'xfer'} = undef;
 339     $self->{'xfer_source'} = undef;
 340     $self->{'xfer_dest'} = undef;
 341     $self->{'handle'} = undef;
 342     $self->{'header'} = undef;
 343     $self->{'hostname'} = undef;
 344     $self->{'diskname'} = undef;
 345     $self->{'datestamp'} = undef;
 346     $self->{'level'} = undef;
 347     $self->{'header'} = undef;
 348     $self->{'state'} = 'idle';
 349     delete $self->{'result'};
 350     delete $self->{'dumper_status'};
 351     delete $self->{'dump_params'};
 352
 353     $self->{'controller'}->{'proto'}->send($msgtype, %msg_params);
 354 }
 355
 356
 357 ##
 358 # Scribe feedback
 359
 360 sub request_volume_permission {
 361     my $self = shift;
 362     my %params = @_;
 363
 364     $self->{'perm_cb'} = $params{'perm_cb'};
 365     # and send the request to the driver
 366     $self->{'controller'}->{'proto'}->send(Amanda::Taper::Protocol::REQUEST_NEW_TAPE,
 367         handle => $self->{'handle'});
 368 }
 369
 370 sub scribe_notif_new_tape {
 371     my $self = shift;
 372     my %params = @_;
 373
 374     # TODO: if $params{error} is set, report it back to the driver
 375     # (this will be a change to the protocol)
 376     if ($params{'volume_label'}) {
 377         $self->{'label'} = $params{'volume_label'};
 378
 379         # add to the trace log
 380         log_add($L_START, sprintf("datestamp %s label %s tape %s",
 381                 $self->{'timestamp'},
 382                 quote_string($self->{'label'}),
 383                 ++$tape_num));
 384
 385         # and the amdump log
 386         print STDERR "taper: wrote label '$self->{label}'\n";
 387
 388         # and inform the driver
 389         $self->{'controller'}->{'proto'}->send(Amanda::Taper::Protocol::NEW_TAPE,
 390             handle => $self->{'handle'},
 391             label => $params{'volume_label'});
 392     } else {
 393         $self->{'label'} = undef;
 394
 395         $self->{'controller'}->{'proto'}->send(Amanda::Taper::Protocol::NO_NEW_TAPE,
 396             handle => $self->{'handle'});
 397     }
 398 }
 399
 400 sub scribe_notif_part_done {
 401     my $self = shift;
 402     my %params = @_;
 403
 404     $self->_assert_in_state("writing") or return;
 405
 406     my $stats = make_stats($params{'size'}, $params{'duration'}, $self->{'orig_kb'});
 407
 408     # log the part, using PART or PARTPARTIAL
 409     my $logbase = sprintf("%s %s %s %s %s %s/%s %s %s",
 410         quote_string($self->{'label'}),
 411         $params{'fileno'},
 412         quote_string($self->{'header'}->{'name'}.""), # " is required for SWIG..
 413         quote_string($self->{'header'}->{'disk'}.""),
 414         $self->{'datestamp'},
 415         $params{'partnum'}, -1, # totalparts is always -1
 416         $self->{'level'},
 417         $stats);
 418     if ($params{'successful'}) {
 419         log_add($L_PART, $logbase);
 420     } else {
 421         log_add($L_PARTPARTIAL, "$logbase \"No space left on device\"");
 422     }
 423
 424     # only send a PARTDONE if it was successful
 425     if ($params{'successful'}) {
 426         $self->{'controller'}->{'proto'}->send(Amanda::Taper::Protocol::PARTDONE,
 427             handle => $self->{'handle'},
 428             label => $self->{'label'},
 429             fileno => $params{'fileno'},
 430             stats => $stats,
 431             kb => $params{'size'} / 1024);
 432     }
 433 }
 434
 435 sub scribe_notif_log_info {
 436     my $self = shift;
 437     my %params = @_;
 438
 439     log_add($L_INFO, $params{'message'});
 440 }
 441
 442 ##
 443 # Utilities
 444
 445 sub _assert_in_state {
 446     my $self = shift;
 447     my ($state) = @_;
 448     if ($self->{'state'} eq $state) {
 449         return 1;
 450     } else {
 451         $self->{'controller'}->{'proto'}->send(Amanda::Taper::Protocol::BAD_COMMAND,
 452             message => "command not appropriate in state '$self->{state}' : '$state'");
 453         return 0;
 454     }
 455 }
 456
 457 sub create_status_file {
 458     my $self = shift;
 459
 460     # create temporary file
 461     ($self->{status_fh}, $self->{status_filename}) =
 462         File::Temp::tempfile("taper_status_file_XXXXXX",
 463                                 DIR => $Amanda::Paths::AMANDA_TMPDIR,
 464                                 UNLINK => 1);
 465
 466     # tell amstatus about it by writing it to the dump log
 467     my $qdisk = Amanda::Util::quote_string($self->{'diskname'});
 468     my $qhost = Amanda::Util::quote_string($self->{'hostname'});
 469     print STDERR "taper: status file $qhost $qdisk:" .
 470                     "$self->{status_filename}\n";
 471     print {$self->{status_fh}} "0";
 472
 473     # create timer callback, firing every 5s (=5000msec)
 474     $self->{timer} = Amanda::MainLoop::timeout_source(5000);
 475     $self->{timer}->set_callback(sub {
 476         my $size = $self->{scribe}->get_bytes_written();
 477         seek $self->{status_fh}, 0, 0;
 478         print {$self->{status_fh}} $size;
 479         $self->{status_fh}->flush();
 480     });
 481 }
 482
 483 sub send_port_and_get_header {
 484     my $self = shift;
 485     my ($finished_cb) = @_;
 486
 487     my $header_xfer;
 488     my ($xsrc, $xdst);
 489     my $errmsg;
 490
 491     my $steps = define_steps
 492         cb_ref => \$finished_cb;
 493
 494     step send_port => sub {
 495         # get the ip:port pairs for the data connection from the data xfer source,
 496         # which should be an Amanda::Xfer::Source::DirectTCPListen
 497         my $data_addrs = $self->{'xfer_source'}->get_addrs();
 498         $data_addrs = join ";", map { $_->[0] . ':' . $_->[1] } @$data_addrs;
 499
 500         # and set up an xfer for the header, too, using DirectTCP as an easy
 501         # way to implement a listen/accept/read process.  Note that this does
 502         # not enforce a maximum size, so this portion of Amanda at least can
 503         # handle any size header
 504         ($xsrc, $xdst) = (
 505             Amanda::Xfer::Source::DirectTCPListen->new(),
 506             Amanda::Xfer::Dest::Buffer->new(0));
 507         $header_xfer = Amanda::Xfer->new([$xsrc, $xdst]);
 508         $header_xfer->start($steps->{'header_xfer_xmsg_cb'});
 509
 510         my $header_addrs = $xsrc->get_addrs();
 511         $header_addrs = [ grep { $_->[0] eq '127.0.0.1' } @$header_addrs ];
 512         die "Source::DirectTCPListen did not return a localhost address"
 513             unless @$header_addrs;
 514         my $header_port = $header_addrs->[0][1];
 515
 516         # and tell the driver which ports we're listening on
 517         $self->{'controller'}->{'proto'}->send(Amanda::Taper::Protocol::PORT,
 518             worker_name => $self->{'worker_name'},
 519             handle => $self->{'handle'},
 520             port => $header_port,
 521             ipports => $data_addrs);
 522     };
 523
 524     step header_xfer_xmsg_cb => sub {
 525         my ($src, $xmsg, $xfer) = @_;
 526         if ($xmsg->{'type'} == $XMSG_INFO) {
 527             info($xmsg->{'message'});
 528         } elsif ($xmsg->{'type'} == $XMSG_ERROR) {
 529             $errmsg = $xmsg->{'messsage'};
 530         } elsif ($xmsg->{'type'} == $XMSG_DONE) {
 531             if ($errmsg) {
 532                 $finished_cb->($errmsg);
 533             } else {
 534                 $steps->{'got_header'}->();
 535             }
 536         }
 537     };
 538
 539     step got_header => sub {
 540         my $hdr_buf = $xdst->get();
 541
 542         # close stuff up
 543         $header_xfer = $xsrc = $xdst = undef;
 544
 545         if (!defined $hdr_buf) {
 546             return $finished_cb->("Got empty header");
 547         }
 548
 549         # parse the header, finally!
 550         $self->{'header'} = Amanda::Header->from_string($hdr_buf);
 551
 552         $finished_cb->(undef);
 553     };
 554 }
 555
 556 # do the work of starting a new xfer; this contains the code common to
 557 # msg_PORT_WRITE and msg_FILE_WRITE.
 558 sub setup_and_start_dump {
 559     my $self = shift;
 560     my ($msgtype, %params) = @_;
 561     my %get_xfer_dest_args;
 562
 563     # setting up the dump is a bit complex, due to the requirements of
 564     # a directtcp port_write.  This function:
 565     # 1. creates and starts a transfer (make_xfer)
 566     # 2. gets the header
 567     # 3. calls the scribe's start_dump method with the new header
 568
 569     my $steps = define_steps
 570         cb_ref => \$params{'dump_cb'};
 571
 572     step setup => sub {
 573         $self->{'handle'} = $params{'handle'};
 574         $self->{'hostname'} = $params{'hostname'};
 575         $self->{'diskname'} = $params{'diskname'};
 576         $self->{'datestamp'} = $params{'datestamp'};
 577         $self->{'level'} = $params{'level'};
 578         $self->{'header'} = undef; # no header yet
 579         $self->{'orig_kb'} = $params{'orig_kb'};
 580         $self->{'input_errors'} = [];
 581
 582         if ($msgtype eq Amanda::Taper::Protocol::PORT_WRITE &&
 583             (my $err = $self->{'scribe'}->check_data_path($params{'data_path'}))) {
 584             return $params{'dump_cb'}->(
 585                 result => "FAILED",
 586                 device_errors => [ ['error', "$err"] ],
 587                 size => 0,
 588                 duration => 0.0,
 589                 total_duration => 0);
 590         }
 591         $steps->{'process_args'}->();
 592     };
 593
 594     step process_args => sub {
 595         # extract the splitting-related parameters, stripping out empty strings
 596         my %splitting_args = map {
 597             ($params{$_} ne '')? ($_, $params{$_}) : ()
 598         } qw(
 599             dle_tape_splitsize dle_split_diskbuffer dle_fallback_splitsize dle_allow_split
 600             part_size part_cache_type part_cache_dir part_cache_max_size
 601         );
 602
 603         # convert numeric values to BigInts
 604         for (qw(dle_tape_splitsize dle_fallback_splitsize part_size part_cache_max_size)) {
 605             $splitting_args{$_} = Math::BigInt->new($splitting_args{$_})
 606                 if (exists $splitting_args{$_});
 607         }
 608
 609         # and convert those to get_xfer_dest args
 610         %get_xfer_dest_args = get_splitting_args_from_config(
 611                 %splitting_args);
 612         $get_xfer_dest_args{'max_memory'} = getconf($CNF_DEVICE_OUTPUT_BUFFER_SIZE);
 613         if (!getconf_seen($CNF_DEVICE_OUTPUT_BUFFER_SIZE)) {
 614             my $device = $self->{'scribe'}->get_device();
 615             my $block_size4 = $device->block_size * 4;
 616             if ($block_size4 > $get_xfer_dest_args{'max_memory'}) {
 617                 $get_xfer_dest_args{'max_memory'} = $block_size4;
 618             }
 619         }
 620         $get_xfer_dest_args{'can_cache_inform'} = ($msgtype eq Amanda::Taper::Protocol::FILE_WRITE);
 621
 622         # if we're unable to fulfill the user's splitting needs, we can still give
 623         # the dump a shot - but we'll warn them about the problem
 624         if ($get_xfer_dest_args{'warning'}) {
 625             log_add($L_WARNING, sprintf("%s:%s: %s",
 626                     $params{'hostname'}, $params{'diskname'},
 627                     $get_xfer_dest_args{'warning'}));
 628             delete $get_xfer_dest_args{'warning'};
 629         }
 630
 631         $steps->{'make_xfer'}->();
 632     };
 633
 634     step make_xfer => sub {
 635         $self->_assert_in_state("idle") or return;
 636         $self->{'state'} = 'making_xfer';
 637
 638         $self->{'xfer_dest'} = $self->{'scribe'}->get_xfer_dest(%get_xfer_dest_args);
 639
 640         my $xfer_source;
 641         if ($msgtype eq Amanda::Taper::Protocol::PORT_WRITE) {
 642             $xfer_source = Amanda::Xfer::Source::DirectTCPListen->new();
 643         } else {
 644             $xfer_source = Amanda::Xfer::Source::Holding->new($params{'filename'});
 645         }
 646         $self->{'xfer_source'} = $xfer_source;
 647
 648         $self->{'xfer'} = Amanda::Xfer->new([$xfer_source, $self->{'xfer_dest'}]);
 649         $self->{'xfer'}->start(sub {
 650             my ($src, $msg, $xfer) = @_;
 651             $self->{'scribe'}->handle_xmsg($src, $msg, $xfer);
 652
 653             # if this is an error message that's not from the scribe's element, then
 654             # we'll need to keep track of it ourselves
 655             if ($msg->{'type'} == $XMSG_ERROR and $msg->{'elt'} != $self->{'xfer_dest'}) {
 656                 push @{$self->{'input_errors'}}, $msg->{'message'};
 657             }
 658         });
 659
 660         # we've started the xfer now, but the destination won't actually write
 661         # any data until we call start_dump.  And we'll need a header for that.
 662
 663         $steps->{'get_header'}->();
 664     };
 665
 666     step get_header => sub {
 667         $self->_assert_in_state("making_xfer") or return;
 668         $self->{'state'} = 'getting_header';
 669
 670         if ($msgtype eq Amanda::Taper::Protocol::FILE_WRITE) {
 671             # getting the header is easy for FILE-WRITE..
 672             my $hdr = $self->{'header'} = Amanda::Holding::get_header($params{'filename'});
 673
 674             # stip out header fields we don't need
 675             $hdr->{'cont_filename'} = '';
 676
 677             if (!defined $hdr || $hdr->{'type'} != $Amanda::Header::F_DUMPFILE) {
 678                 die("Could not read header from '$params{filename}'");
 679             }
 680             $steps->{'start_dump'}->(undef);
 681         } else {
 682             # ..but quite a bit harder for PORT-WRITE; this method will send the
 683             # proper PORT command, then read the header from the dumper and parse
 684             # it, placing the result in $self->{'header'}
 685             $self->send_port_and_get_header($steps->{'start_dump'});
 686         }
 687     };
 688
 689     step start_dump => sub {
 690         my ($err) = @_;
 691
 692         $self->_assert_in_state("getting_header") or return;
 693         $self->{'state'} = 'writing';
 694
 695         # if $err is set, cancel the dump, treating it as a input error
 696         if ($err) {
 697             push @{$self->{'input_errors'}}, $err;
 698             return $self->{'scribe'}->cancel_dump(
 699                 xfer => $self->{'xfer'},
 700                 dump_cb => $params{'dump_cb'});
 701         }
 702
 703         # sanity check the header..
 704         my $hdr = $self->{'header'};
 705         if ($hdr->{'dumplevel'} != $params{'level'}
 706             or $hdr->{'name'} ne $params{'hostname'}
 707             or $hdr->{'disk'} ne $params{'diskname'}
 708             or $hdr->{'datestamp'} ne $params{'datestamp'}) {
 709             die("Header of dumpfile does not match command from driver");
 710         }
 711
 712         # start producing status
 713         $self->create_status_file();
 714
 715         # and fix it up before writing it
 716         $hdr->{'totalparts'} = -1;
 717         $hdr->{'type'} = $Amanda::Header::F_SPLIT_DUMPFILE;
 718
 719         $self->{'scribe'}->start_dump(
 720             xfer => $self->{'xfer'},
 721             dump_header => $hdr,
 722             dump_cb => $params{'dump_cb'});
 723     };
 724 }
 725
 726 sub dump_cb {
 727     my $self = shift;
 728     my %params = @_;
 729
 730     $self->{'dump_params'} = \%params;
 731     $self->{'result'} = $params{'result'};
 732
 733     # if we need to the dumper status (to differentiate a dropped network
 734     # connection from a normal EOF) and have not done so yet, then send a
 735     # DUMPER_STATUS message and re-call this method (dump_cb) with the result.
 736     if ($params{'result'} eq "DONE"
 737             and $self->{'doing_port_write'}
 738             and !exists $self->{'dumper_status'}) {
 739         my $controller = $self->{'controller'};
 740         my $proto = $controller->{'proto'};
 741         my $handle = $self->{'handle'};
 742         $proto->send(Amanda::Taper::Protocol::DUMPER_STATUS,
 743                 handle => "$handle");
 744     } else {
 745         $self->result_cb();
 746     }
 747 }
 748
 749 1;