File Coverage

blib/lib/Net/CascadeCopy.pm

Criterion	Covered	Total	%
statement	243	257	94.5
branch	71	92	77.1
condition	7	11	63.6
subroutine	28	29	96.5
pod	4	4	100.0
total	353	393	89.8

package Net::CascadeCopy;
use strict;
use warnings;

our $VERSION = '0.2.6'; # VERSION

use Mouse;

use Benchmark;
use Log::Log4perl qw(:easy);
use POSIX ":sys_wait_h"; # imports WNOHANG
use Proc::Queue size => 32, debug => 0, trace => 0, delay => 1;

my $logger = get_logger( 'default' );

has data         => ( is => 'ro', isa => 'HashRef', default => sub { return {} } );

has total_time   => ( is => 'rw', isa => 'Num', default => 0 );

has ssh          => ( is => 'ro', isa => 'Str', default => "ssh"   );
has ssh_args     => ( is => 'ro', isa => 'Str', default => "-x -A" );

has command      => ( is => 'ro', isa => 'Str', required => 1  );
has command_args => ( is => 'ro', isa => 'Str', default  => "" );

has source_path  => ( is => 'ro', isa => 'Str', required => 1 );
has target_path  => ( is => 'ro', isa => 'Str', required => 1 );

has output       => ( is => 'ro', isa => 'Str', default => "" );

# maximum number of failures per server
has max_failures => ( is => 'ro', isa => 'Num', default => 3 );

# maximum processes per remote server
has max_forks    => ( is => 'ro', isa => 'Num', default => 2 );

# keep track of child processes
has children     => ( is => 'ro', isa => 'HashRef', default => sub { return {} } );

# for testing purposes
has transfer_map => ( is => 'ro', isa => 'HashRef', default => sub { return {} } );

# sort order
has sort_order   => ( is => 'ro', isa => 'HashRef', default => sub { return {} } );

sub add_group {
    my ( $self, $group, $servers_a ) = @_;

    $logger->info( "Adding group: $group: ",
                   join( ", ", @$servers_a ),
               );

    # initialize data structures
    for my $server ( @{ $servers_a } ) {
        $self->data->{remaining}->{ $group }->{$server} = 1;

        unless ( defined $self->sort_order->{ $group }->{ $server } ) {
            $self->sort_order->{ $group }->{ $server } = scalar keys %{ $self->sort_order->{ $group } };
        }
    }

    # first server to transfer from is the current server
    $self->data->{available}->{ $group }->{localhost} = 1;

    # initialize data structures
    $self->data->{completed}->{ $group } = [];

}

sub get_groups {
    my ( $self ) = @_;

    my @groups;

    for my $group ( keys %{ $self->sort_order } ) {
        push @groups, $group;
    }

    return @groups;
}

sub transfer {
    my ( $self ) = @_;

    my $transfer_start = new Benchmark;

  LOOP:
    while ( 1 ) {
        last LOOP unless $self->_transfer_loop( $transfer_start );
        sleep 1;
    }
}

sub _transfer_loop {
    my ( $self, $transfer_start ) = @_;

    $self->_check_for_completed_processes();

    # keep track if there are any remaining servers in any groups
    my ( $remaining_flag, $available_flag );

    # handle completed processes
    if ( ! scalar keys %{ $self->data->{remaining} } && ! $self->data->{running} ) {
        my $transfer_end = new Benchmark;
        my $transfer_diff = timediff( $transfer_end, $transfer_start );
        my $transfer_time = $self->_human_friendly_time( $transfer_diff->[0] );
        $logger->warn( "Job completed in $transfer_time" );

        my $total_time = $self->_human_friendly_time( $self->total_time );
        $logger->info ( "Cumulative tansfer time of all jobs: $total_time" );

        my $savings = $self->total_time - $transfer_diff->[0];
        if ( $savings ) {
            $savings = $self->_human_friendly_time( $savings );
            $logger->info( "Approximate Time Saved: $savings" );
        }

        my $failure_count;
        for my $group ( $self->get_groups() ) {
            my ( $errors, $failures ) = $self->_get_failed_count( $group );
            $failure_count += $failures;
        }

        if ( $failure_count ) {
            $logger->fatal( "$failure_count Fatal Errors" );
        }
        else {
            $logger->warn( "Completed successfully" );
        }
        return;
    }

    # start new transfers to remaining servers
    for ( 0, 1 ) {
        # code_smell: start 2 transfers each round to keep the
        # test cases from breaking.  need to refactor test cases!
        for my $group ( $self->_get_remaining_groups() ) {
            # group contains servers that still need to be tranferred

            if ( $self->_get_available_servers( $group )  ) {
                # reserve a server to start a new transfer from
                my $source = $self->_reserve_available_server( $group );
                my $target = $self->_reserve_remaining_server( $group );
                $self->_start_process( $group, $source, $target );
            }
        }
    }

    return 1;
}

sub _human_friendly_time {
    my ( $self, $seconds ) = @_;

    return "0 secs" unless $seconds;

    my @time_string;

    if ( $seconds > 3600 ) {
        my $hours = int( $seconds / 3600 );
        $seconds = $seconds % 3600;
        push @time_string, "$hours hrs";
    }
    if ( $seconds > 60 ) {
        my $minutes = int( $seconds / 60 );
        $seconds = $seconds % 60;
        push @time_string, "$minutes mins";
    }
    if ( $seconds ) {
        push @time_string, "$seconds secs";
    }

    return join " ", @time_string;
}

sub _print_status {
    my ( $self, $group ) = @_;

    # completed procs
    my $completed = 0;
    if ( $self->data->{completed}->{ $group } ) {
        $completed = scalar @{ $self->data->{completed}->{ $group } };
    }

    # running procs
    my $running = 0;
    if ( $self->data->{running} ) {
        for my $pid ( keys %{ $self->data->{running} } ) {
            if ( $self->data->{running}->{ $pid }->{group} eq $group ) {
                $running++;
            }
        }
    }

    # unstarted
    my $unstarted = 0;
    if ( $self->data->{remaining}->{ $group } ) {
        $unstarted = scalar keys %{ $self->data->{remaining}->{ $group } };
    }

    # failed
    my ( $errors, $failures ) = $self->_get_failed_count( $group );

    $logger->info( "\U$group: ",
                   "completed:$completed ",
                   "running:$running ",
                   "left:$unstarted ",
                   "errors:$errors ",
                   "failures:$failures ",
               );
}

sub _start_process {
    my ( $self, $group, $source, $target ) = @_;

    $self->transfer_map->{$source}->{$target} += 1;

    my $f=fork;
    if (defined ($f) and $f==0) {

        my $target_path = $self->target_path;

        my $command;
        if ( $source eq "localhost" ) {
            $command = join( " ",
                             $self->command,
                             $self->command_args,
                             $self->source_path,
                             "$target:$target_path"
                         );
        } else {
            $command = join( " ",
                             $self->ssh,
                             $self->ssh_args,
                             $source,
                             $self->command,
                             $self->command_args,
                             $self->target_path,
                             "$target:$target_path"
                         );
        }

        print "COMMAND: $command\n";

        my $output = $self->output || "";
        if ( $output eq "stdout" ) {
            # don't modify command
        } elsif ( $output eq "log" ) {
            # redirect all child output to log
            $command = "$command >> ccp.$source.$target.log 2>&1"
        } else {
            # default is to redirectout stdout to /dev/null
            $command = "$command >/dev/null"
        }

        $logger->info( "Starting: ($group) $source => $target" );
        $logger->debug( "Starting new child: $command" );

        system( $command );

        if ($? == -1) {
            $logger->logconfess( "failed to execute: $!" );
        } elsif ($? & 127) {
            $logger->logconfess( sprintf "child died with signal %d, %s coredump",
                                 ($? & 127),  ($? & 128) ? 'with' : 'without'
                             );
        } else {
            my $exit_status = $? >> 8;
            if ( $exit_status ) {
                $logger->error( "child exit status: $exit_status" );
            }
            exit $exit_status;
        }
    } else {
        my $start = new Benchmark;
        $self->data->{running}->{ $f } = { group  => $group,
                                           source => $source,
                                           target => $target,
                                           start  => $start,
                                       };
    }
}

sub _check_for_completed_processes {
    my ( $self ) = @_;

    return unless $self->data->{running};

    # find any processes that ended and reschedule the source and
    # target servers in the available pool
    for my $pid ( keys %{ $self->data->{running} } ) {
        if ( waitpid( $pid, WNOHANG) ) {

            # check the exit status of the command.
            if ( $? ) {
                $self->_failed_process( $pid );
            } else {
                $self->_succeeded_process( $pid );
            }
        }
    }

    unless ( keys %{ $self->data->{running} } ) {
        delete $self->data->{running};
    }
}

sub _succeeded_process {
    my ( $self, $pid ) = @_;

    my $group  = $self->data->{running}->{ $pid }->{group};
    my $source = $self->data->{running}->{ $pid }->{source};
    my $target = $self->data->{running}->{ $pid }->{target};
    my $start  = $self->data->{running}->{ $pid }->{start};

    # calculate time for this transfer
    my $end = new Benchmark;
    my $diff = timediff( $end, $start );

    # keep track of transfer time totals
    $self->total_time( $self->total_time + $diff->[0] );

    my $time = $self->_human_friendly_time( $diff->[0] );
    $logger->warn( "Succeeded: ($group) $source => $target ($time)" );

    $self->_mark_available( $group, $source );
    $self->_mark_completed( $group, $target );
    $self->_mark_available( $group, $target );

    delete $self->data->{running}->{ $pid };

    $self->_print_status( $group );
}

sub _failed_process {
    my ( $self, $pid ) = @_;

    my $group  = $self->data->{running}->{ $pid }->{group};
    my $source = $self->data->{running}->{ $pid }->{source};
    my $target = $self->data->{running}->{ $pid }->{target};
    my $start  = $self->data->{running}->{ $pid }->{start};

    # calculate time for this transfer
    my $end = new Benchmark;
    my $diff = timediff( $end, $start );
    # keep track of transfer time totals
    $self->total_time( $self->total_time + $diff->[0] );

    $logger->warn( "Failed: ($group) $source => $target ($diff->[0] seconds)" );

    # there was an error during the transfer, reschedule
    # it at the end of the list
    $self->_mark_available( $group, $source );
    my $fail_count = $self->_mark_failed( $group, $target );
    if (  ! $self->_get_available_servers( $group ) ) {
        $logger->fatal( "Error: no available servers in '$group' to handle $target" );
        $self->_mark_failed( $group, $target, 1 );
    }
    elsif ( $fail_count >= $self->max_failures ) {
        $logger->fatal( "Error: giving up on ($group) $target" );
    } else {
        $self->_mark_remaining( $group, $target );
    }

    delete $self->data->{running}->{ $pid };

    $self->_print_status( $group );
}

sub _get_available_servers {
    my ( $self, $group ) = @_;
    return unless $self->data->{available};
    return unless $self->data->{available}->{ $group };

    my @available;
    for my $host ( $self->_sort_servers( $group, keys %{ $self->data->{available}->{ $group } } ) ) {
        if ( $self->children->{$host} ) {
            next if $self->children->{$host} >= $self->max_forks;
        }
        push @available, $host;
    }

    return @available;
}

sub _reserve_available_server {
    my ( $self, $group ) = @_;

    my ( $server ) = $self->_get_available_servers( $group );
    $logger->debug( "Reserving ($group) $server" );

    # only one transfer from localhost to each dc
    if ( $server eq "localhost" ) {
        delete $self->data->{available}->{$group}->{localhost};
    }

    $self->children->{ $server }++;
    return $server;
}

sub _get_remaining_servers {
    my ( $self, $group ) = @_;

    return unless $self->data->{remaining};

    return unless $self->data->{remaining}->{ $group };

    my @hosts = $self->_sort_servers( $group, keys %{ $self->data->{remaining}->{ $group } } );
    return @hosts;
}

sub _sort_servers {
    my ( $self, $group, @servers ) = @_;

    # sort servers based on original insertion order
    @servers = sort { $self->sort_order->{$group}->{$a} <=> $self->sort_order->{$group}->{$b} } @servers;

    return @servers;
}

sub _reserve_remaining_server {
    my ( $self, $group ) = @_;

    if ( $self->_get_remaining_servers( $group ) ) {
        my $server = ( $self->_sort_servers( $group, keys %{ $self->data->{remaining}->{ $group } } ) )[0];
        delete $self->data->{remaining}->{ $group }->{$server};
        $logger->debug( "Reserving ($group) $server" );

        # delete remaining data structure as groups are completed
        unless ( scalar keys %{ $self->data->{remaining}->{ $group } } ) {
            $logger->debug( "Group empty: $group" );
            delete $self->data->{remaining}->{ $group };
            unless ( scalar ( keys %{ $self->data->{remaining} } ) ) {
                $logger->debug( "No servers remaining" );
                delete $self->data->{remaining};
            }
        }
        return $server;
    }
}

sub _get_remaining_groups {
    my ( $self ) = @_;
    return unless $self->data->{remaining};
    my @keys = sort keys %{ $self->data->{remaining} };
    return unless scalar @keys;
    return @keys;
}

sub _mark_available {
    my ( $self, $group, $server ) = @_;

    # only the initial transfer to each dc comes from localhost.  iff
    # first transfer (from localhost) succeeded, then don't reschedule
    # for future syncs
    if ( $server eq "localhost" ) {

        if ( $self->data->{completed}->{ $group } ) {

            delete $self->data->{available}->{ $group }->{$server};
            return;
        }
    }

    $logger->debug( "Server available: ($group) $server" );
    $self->data->{available}->{ $group }->{$server} = 1;

    # reduce count of number of active processes on this server
    if ( $self->children->{$server} ) {
        $self->children->{$server}--;
    }
}

sub _mark_remaining {
    my ( $self, $group, $server ) = @_;

    $logger->debug( "Server remaining: ($group) $server" );
    $self->data->{remaining}->{ $group }->{$server} = 1;
}

sub _mark_completed {
    my ( $self, $group, $server ) = @_;

    $logger->debug( "Server completed: ($group) $server" );
    push @{ $self->data->{completed}->{ $group } }, $server;
}

sub _mark_failed {
    my ( $self, $group, $server, $fatal ) = @_;

    $logger->debug( "Server failed: ($group) $server" );

    if ( $fatal ) {
        $self->data->{failed}->{ $group }->{ $server } = $self->max_failures;
    }
    else {
        $self->data->{failed}->{ $group }->{ $server }++;
    }

    my $failures = $self->data->{failed}->{ $group }->{ $server };
    $logger->debug( "$failures failures for ($group) $server" );
    return $failures;
}

sub _get_failed_count {
    my ( $self, $group ) = @_;

    my $errors = 0;
    my $failures = 0;

    if ( $self->data->{failed} && $self->data->{failed}->{ $group } ) {
        for my $server ( keys %{ $self->data->{failed}->{ $group }} ) {
            $errors += $self->data->{failed}->{ $group }->{ $server };
            if ( $self->data->{failed}->{ $group }->{ $server } >= $self->max_failures ) {
                $failures++;
            }
        }
    }

    return ( $errors, $failures );
}

# tracing all the attempted transfers
sub get_transfer_map {
    my ( $self ) = @_;

    return $self->transfer_map;
}


1;

__END__