XFA/Using Transfers

Since the transfer architecture is designed to be used from Perl, this document describes the Perl API. The C interface is similar, and the comments and declarations in xfer.h and xfer-element.h should provide enough detail to translate this article into C.

You should be familiar with Amanda's asynchronous programming support; see Amanda::MainLoop.

Transfer Operation

A transfer runs entirely asynchronously, which means that the main thread continues to run in parallel to the data transfer. Any information from the transfer elements or the transfer itself comes back to the main thread in the form of XMsgs, received via an Amanda::MainLoop callback.

Creating a Transfer

To create a transfer, call the Amanda::Xfer constructor with an arrayref of elements:

 my $xfer = Amanda::Xfer->new([
   Amanda::Xfer::Source::Fd->new($infd),
   Amanda::Xfer::Dest::Fd->new($outfd)
 ]); 

Note that the transfer is not yet started.

Available Transfer Elements

Amanda makes a wide array of transfer elements available. They are categorized as source, destination, or filter, and listed in the perldoc for Amanda::Xfer. Each has a constructor with arguments particular to the element. Some elements may have additional methods, which are also described in the perldoc.

 my $src = Amanda::Xfer::Source::Random->new(10240, 0xF00);

Any combination of transfer elements can become a transfer, as long as the first element is a source, the last is a destination, and any intervening elements are filters. The XFA will select the most efficient means of connecting the elements.

Running a Transfer

Before starting a transfer, an application should set up a callaback to receive XMsgs:

 $xfer->get_source()->set_callback(sub {
     my ($src, $xmsg, $xfer) = @_;
      if ($xmsg->{type} == $Amanda::Xfer::XMSG_ERROR) {
         print STDERR $xmsg->{message}, "\n";
     }   
     if ($xfer->get_status() == $XFER_DONE) {
         $src->remove();
         Amanda::MainLoop::quit();
     }
 });
 
 $xfer->start();
 Amanda::MainLoop::run();

Breaking this down, the call to get_source gets the MainLoop event source for this transfer. The set_callback invocation associates the anonymous sub as a callback for the event source, so it will be called for every message sent by $xfer. The parameters for this callback are the event source itself, $src; the incoming message, $xmsg; and the transfer that generated the message, $xfer. The next conditional examines the message, and if it represents an error, prints it to the error stream. The second conditional checks whether the entire transfer is done. This is the standard way to detect completion of a transfer. Note that, even when an error has occurred, an XMSG_DONE message will be sent when the transfer is completely finished. In this case, the snippet removes the event source from MainLoop and causes the Amanda::MainLoop::run call to return.

Finally, the start method sets the transfer in motion. Amanda::MainLoop::run loops waiting for new events until Amanda::MainLoop::quit is invoked.

Canceling a Transfer

An ongoing transfer can be cancelled, either by a fatal error in one of the elements, or by the main thread. The syntax is simple:

$xfer->cancel();

What happens underneath is not so simple. First, this function queues an asynchronous XMSG_CANCEL message, so the cancellation may not begin immediately. If desired, this message can be caught in the callback. When the message is received, the cancellation is passed on to each of the elements in the transfer, but some of those elements may not be able to cancel immediately (because they are in the midst of a large block of data, for example), but will do so as soon as possible. Once all elements have stopped transferring data, the transfer will set its status to XFER_DONE and send an XMSG_DONE. To summarize, then:

• somthing calls $xfer->cancel();
• the transfer waits until XMSG_CANCEL is received
• the transfer enters state "XFER_CANCELLING" while all elements cancel and drain out any buffered data
• the transfer enters state XFER_DONE and sends an XMSG_DONE

Examples

Transfer with a Timeout

sub try_xfer {
    my ($elements, $timeout, $finished_cb) = @_;
    my $xfer = Amanda::Xfer->new($elements);
    my $xfer_src = $xfer->get_source();
    my $timeout_src = Amanda::MainLoop::timeout_source($timeout);

    sub finish {
        my ($success) = @_;

        # clean up the sources
        $xfer_src->remove();
        $timeout_src->remove();

        # call the function our caller gave us
        $finished_cb->($success);
    }

    $xfer_src->set_callback(sub {
         if ($xmsg->{type} == $Amanda::Xfer::XMSG_ERROR) {
            print "Operation failed: ", $xmsg->{message}, "\n";
        }   
        if ($xfer->get_status() == $XFER_DONE) {
            print "Operation complete\n";
            finish(1);
        }
    });

    $timeout_src->set_callback(sub {
        print "Operation timed out\n";
        finish(0);
    });

    $xfer->start();
}

Note that this example extends the idea of callbacks using simple coderefs. This function could be used in a simple application like this:

 try_xfer($elements, sub {
   my ($success) = @_;
   Amanda::MainLoop::quit();
 });
 Amanda::MainLoop::run();

but a more complex application may want to chain multiple transfers together:

 my @elementlist = ( ... );
 sub start_next_xfer {
   my ($success) = @_;
   if (!$success or !@elementlist) {
     Amanda::MainLoop::quit();
   }
   my $elements = pop @elementlist;
   try_xfer($elements, &start_next_xfer);
 }
 start_next_xfer(1);
 Amanda::MainLoop::run();