#!/usr/bin/env perl

use Carp;
use warnings;
use strict;
use Getopt::Long qw( GetOptions :config auto_help auto_version gnu_getopt );
use Pod::Usage qw( pod2usage );
use File::Basename;
use IPC::Cmd 0.36 qw( can_run run );
use Sys::Hostname;

my $local_raw = '/scratch3/watersc1/raw/';
my $local_tmp = '/scratch3/watersc1/tmp/';
my $threads = 10;
my @hosts = ('ippc20.ipp.ifa.hawaii.edu','ippc24.ipp.ifa.hawaii.edu','ippc28.ipp.ifa.hawaii.edu',
	     'ippc21.ipp.ifa.hawaii.edu','ippc25.ipp.ifa.hawaii.edu','ippc29.ipp.ifa.hawaii.edu',
	     'ippc22.ipp.ifa.hawaii.edu','ippc26.ipp.ifa.hawaii.edu',
	     'ippc23.ipp.ifa.hawaii.edu','ippc27.ipp.ifa.hawaii.edu'    );
@hosts = (@hosts, @hosts, @hosts, @hosts, @hosts);
my $input_file;
my $verbose = 0;
my $fetch = 0;
my $offset = 0;

GetOptions(
    'threads=s'   => \$threads,
    'input=s'     => \$input_file,
    'fetch=s'     => \$fetch,
    'offset=s'    => \$offset,
    'verbose'     => \$verbose,
    ) or pod2usage( 2 );
pod2usage( -msg => "Unknown option: @ARGV", -exitval => 2) if @ARGV;
pod2usage( -msg => "Required options: --input", -exitval => 3) unless
    defined($input_file);

my $hostname = hostname;
print STDERR "Running on $hostname\n";

# split in input file list
unless(-d $local_tmp) { system("mkdir -p $local_tmp"); }
unless(-d $local_raw) { system("mkdir -p $local_raw"); }

open(I,$input_file) || die "Couldn't find input file specified\n";
if ($fetch) {
    open(F,$fetch) || die "Couldn't find fetch file specified\n";
}
my $input_base = basename($input_file);
my @filehandles;
my $i;
my $line = 0;
for ($i = 0; $i < $threads; $i++) {
    open($filehandles[$i], ">${local_tmp}/${input_base}.${i}");
}
$i = 0;
while(<I>) {
    chomp;
    $line++;
    if ($line < $offset) { next; } # I think this is off-by-one in a safe direction

    $i = int(rand($#filehandles + 1));

    my $fline;
    if ($fetch) {
	$fline = <F>;
    }
    if (($fetch)&&(!(-e $_))) {  # We are fetching, and do not already have this file.
	print { $filehandles[$i] } $fline;
    }
    elsif (!($fetch)) { # We are pushing
	print { $filehandles[$i] } "${local_tmp}/$_" . "\n";
    }
}
close(I);

for ($i = 0; $i < $threads; $i++) {
    close($filehandles[$i]);
}

# fork the tars
my @pids = ();
for ($i = 0; $i < $threads; $i++) {
    $pids[$i] = fork();
    if ($pids[$i] == 0) {
	my $host = $hosts[$i];
	my $code = 0;
	if ($fetch) { 
	    $code = fetch_task($host,"${input_base}.${i}", 0);
	}
	else {
	    $code = transfer_task($host,"${local_tmp}/${input_base}.${i}", 0);
	}
	exit($code);
    }
}
for ($i = 0; $i < $threads; $i++) {
    waitpid($pids[$i],0);
}

for ($i = 0; $i < $threads; $i++) {
#   unlink("${local_tmp}/${input_base}.${i}");
}


# distribute bundles to nodes

# execute bundle transfer

sub transfer_task {
    my $destination_host = shift;
    my $transfer_filelist= shift;
    my $error = shift;

    # the transform bit is there because it looks like the ' gets dropped, so the * is interpreted, and why is our tar so out of date?
    my $command = "tar cf - --ignore-failed-read --dereference --files-from=${transfer_filelist} | /usr/projects/cosmo/amd6100/bin/ssh -o NoneSwitch=yes -o NoneEnabled=yes $destination_host tar xf - -C /data/ --transform '" . 's,^.\*/data/,,' . "' --dereference"; 
    print STDERR "$command\n";

    my ( $success, $error_code, $full_buf, $stdout_buf, $stderr_buf ) =
	run(command => $command, verbose => $verbose);
    unless ($success) {

	print STDERR "ERROR:  $error_code\n";
# 	my $std = join "\n", @{ $stdout_buf };
# 	print "STDOUT: $std\n";
# 	$std = join "\n", @{ $stderr_buf };
# 	print "STDERR: $std\n";

	foreach my $line (@{ $stderr_buf }) {  # This is a hack.  A messy ugly hack.
	    if ($line =~ /No such file or directory/) { 
		my $file = (split /\s+/, $line)[1];
		$file =~ s/:$//;
		if ($file !~ /nebulous/) { next; }
		print STDERR "Touching $file so tar can continue.\n";
		system("touch $file");
	    }
	}

	warn("Transfer of $transfer_filelist to $destination_host failed with error $error_code.");
	$error++;
	if ($error < 4) {
	    $error_code = transfer_task($destination_host,$transfer_filelist,$error);
	}
	else {
	    die("Failed too many times $error $destination_host $transfer_filelist");
	}
    }
    return($error_code);
}
    
sub fetch_task {
    my $destination_host = shift;
    my $transfer_filelist= shift;
    my $error = shift;

    system("/usr/projects/cosmo/amd6100/bin/scp -o NoneSwitch=yes -o NoneEnabled=yes ${local_tmp}/${transfer_filelist} ${destination_host}:/tmp/");
    my $command = "/usr/projects/cosmo/amd6100/bin/ssh -o NoneSwitch=yes -o NoneEnabled=yes $destination_host tar cf - --ignore-failed-read  --dereference --files-from=/tmp/${transfer_filelist} | tar xf - -C ${local_raw} --skip-old-files --warning=existing-file --dereference ";
    print STDERR "$command\n";

    my ( $success, $error_code, $full_buf, $stdout_buf, $stderr_buf ) =
	run(command => $command, verbose => $verbose);
    unless ($success) {
	$error_code = (($error_code >> 8) or 4);
	warn("Transfer of $transfer_filelist to $destination_host failed with error $error_code.");
	$error++;
	if ($error < 4) {
	    $error_code = fetch_task($destination_host,$transfer_filelist,$error);
	}
	else {
	    die("Failed too many times $error $destination_host $transfer_filelist");
	}
    }
    return($error_code);
}
