#!/usr/bin/env perl
# PROJECT: TaskManager
# AUTHOR: Lukasz Szajkowski
#
# Copyright (c) 2008 Illumina
# This software is covered by the "Illumina Genome Analyzer Software
# License Agreement" and the "Illumina Source Code License Agreement",
# and certain third party copyright/licenses, and any user of this
# source file is bound by the terms therein (see accompanying files
# Illumina_Genome_Analyzer_Software_License_Agreement.pdf and
# Illumina_Source_Code_License_Agreement.pdf and third party
# copyright/license notices).
#use warnings FATAL => 'all';
#use strict;
use warnings FATAL => 'all';
use strict;
use POSIX qw(strftime);
use Sys::Hostname;
use IO::Socket::INET;
use Getopt::Long;

use lib '/home/psgendb/local/pkg/CASAVA_v1.8.2-build/lib/CASAVA-1.8.2/perl';
use Casava::TaskManager qw(%taskFields);

my $help        = 0;
my $argvStr     = join ' ', @ARGV;
my $worflowFile = "";
my $port;
my $delay = 60; #seconds
my $retries=3;
my $them;
my $processType = '';
my $timeFile = undef;
my $hostname     = hostname();
my $verbose = 2;
my $logDateFormat = "%F %T";
my $shell = "bash -o pipefail";

my $result      = GetOptions(
    "worflowFile|w=s" => \$worflowFile,
    "port|p=s"        => \$port,
    "host|h=s"        => \$them,
    "type=s"          => \$processType,
    "delay|d=i"       => \$delay,
    'verbose|v=i'     => \$verbose,
    "retries|r=i"     => \$retries,
    "logDateFormat=s" => \$logDateFormat,
    "shell=s"         => \$shell,
    "help"          => \$help
);
$port = 2345        unless $port;
$them = 'localhost' unless $them;
my $child;

sub dokill {
    kill 9, $child if $child;
}
my $sock = undef;
while (!$sock && $retries--)
{
    sleep(rand($delay));
    $sock = new IO::Socket::INET(
        PeerAddr => $them,
        PeerPort => $port,
        Proto    => 'tcp'
      );
    print STDERR "[" . ( strftime $logDateFormat, localtime ) . "]\t[$hostname]\t[taskAgent.pl ]\tWARNING: could not connect to $them:$port - $@\n" unless $sock;
}
die "[" . ( strftime $logDateFormat, localtime ) . "]\t[$hostname]\t[taskAgent.pl ]\tERROR: ran out of retries while connecting to $them:$port - $@\n" unless $sock;

$sock->autoflush(1);
my $wait = 0;
my $buf;
my $setStatus   = 'setStatus';
my $getNextTask = 'getNextTask';

if ( $processType eq 'sge' ) {
    $setStatus   = 'setStatusSge';
    $getNextTask = 'getNextTaskSge';
}
print $sock "system\t$getNextTask\t$hostname\n";
while ( $buf = $sock->getline ) {
    chomp($buf);
    if ( !defined $buf ) {
        print STDERR "No line\n";
        next;
    }

    #print "Server1: $buf\n";
    my @taskDesc = split '\t', $buf;
    if ( scalar(@taskDesc) == 3 ) {
        print STDERR "Badly formed line = [$_]\n" if 2 < $verbose;
        next;
    }
    my $jobtype = shift @taskDesc;
    if ( $jobtype eq 'msg' ) {
        print STDERR "Msg:[$buf]\n" if 2 < $verbose;
    } elsif ( $jobtype eq 'task' ) {
        my $status   = 'finished';
        my $queueId  = $taskDesc[ $taskFields{queueId} ];
        my $taskId   = $taskDesc[ $taskFields{taskId} ];
        my $cmd      = $taskDesc[ $taskFields{cmd} ];
        my $taskType = $taskDesc[ $taskFields{taskType} ];
        if ( $taskType eq 'task' ) {

            system('/home/psgendb/local/pkg/CASAVA_v1.8.2-build/libexec/CASAVA-1.8.2/loggingShell.sh',
                   "$verbose", "$logDateFormat", 'taskAgent', "$queueId:$taskId",
                   '', '', #empty prerequisites info 
                   $shell, '-c', "$cmd");

            #sleep $delay;
            my $exit_value  = $? >> 8;
            my $signal_num  = $? & 127;
            my $dumped_core = $? & 128;
            if ( $signal_num > 0 ) {
                $status = "ERROR Process killed by SIG($signal_num):$hostname\n";
            }    # if
            if ( $dumped_core > 0 ) {
                $status = "ERROR core dumped:$hostname\n";
            }    # if
            if ( $exit_value != 0 ) {
                $status = "ERROR EXIT_CODE($exit_value):$hostname\n";
            }    # if
            print $sock "status\t$setStatus\t$hostname\t$queueId\[$taskId]\t$status\t\n" or
                die "[" . ( strftime $logDateFormat, localtime ) . "]\t[$hostname]\t[taskAgent.pl ]\tERROR: failed to send task $queueId\[$taskId] status to $them:$port - $@\n";
            if ( $processType eq 'sge' ) {
                print STDERR $hostname."\t".$status . "\t".$buf. "\n" if 2 < $verbose;
                #sleep(1);
                #close $sock;
                #exit(0);
            }
            $wait = 0;
        } elsif ( $taskType eq 'synch' ) {
            $status = 'finished';
            print $sock "status\t$setStatus\t$hostname\t$queueId\[$taskId]\t$status\n" or
                die "[" . ( strftime $logDateFormat, localtime ) . "]\t[$hostname]\t[taskAgent.pl ]\tERROR: failed to send synch $queueId\[$taskId] status to $them:$port - $@\n";
            if ( $processType eq 'sge' ) {
                print STDERR $hostname."\t".$status . "\t".$buf. "\n" if 2 < $verbose;
                #sleep(1);
                #close $sock;
                #exit(0);
            }
            $wait = 0;
        }
    } elsif ( $jobtype eq 'wait' ) {
        if ( $wait == 0 ) {

            #print "Server3: $buf\n";
            $wait = 1;
        }
        sleep $taskDesc[0];
        if ( $processType eq 'sge' ) {
            print $sock "status\tsge_no_job\t$hostname\n" or
                die "[" . ( strftime $logDateFormat, localtime ) . "]\t[$hostname]\t[taskAgent.pl ]\tERROR: failed to send sge task request to $them:$port - $@\n";
            print STDERR $hostname."\t "."tsge_no_job" . "\t".$buf. "\n" if 2 < $verbose;
            #sleep(1);
            #close $sock;
            #exit(0);
        } else {
            print $sock "system\t$getNextTask\t$hostname\n" or
                die "[" . ( strftime $logDateFormat, localtime ) . "]\t[$hostname]\t[taskAgent.pl ]\tERROR: failed to send local task request to $them:$port - $@\n";
        }
    } elsif ( $jobtype eq 'exit' ) {
        close $sock;
        exit(0);
    }
}
close $sock;
1;
__END__

