###############################################################################
#                                                                             #
#  TakTuk, a middleware for adaptive large scale parallel remote executions   #
#  deployment. Perl implementation, copyright(C) 2006 Guillaume Huard.        #
#                                                                             #
#  This program is free software; you can redistribute it and/or modify       #
#  it under the terms of the GNU General Public License as published by       #
#  the Free Software Foundation; either version 2 of the License, or          #
#  (at your option) any later version.                                        #
#                                                                             #
#  This program is distributed in the hope that it will be useful,            #
#  but WITHOUT ANY WARRANTY; without even the implied warranty of             #
#  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the              #
#  GNU General Public License for more details.                               #
#                                                                             #
#  You should have received a copy of the GNU General Public License          #
#  along with this program; if not, write to the Free Software                #
#  Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA  02110-1301  USA #
#                                                                             #
#  Contact: Guillaume.Huard@imag.fr                                           #
#           ENSIMAG - Laboratoire ID                                          #
#           51 avenue Jean Kuntzmann                                          #
#           38330 Montbonnot Saint Martin                                     #
#                                                                             #
###############################################################################

package taktuk;

# Turns autoflush on for the given filehandle
sub no_flush($);
# Unpacks stuff in a buffer begining by something packed
# Returns a couple : the unpacked stuff and the remaining of the buffer
sub unpack($);
# Packs some string so that it can be unpacked when it begining is found into
# a stream of bytes
sub pack($);
# Decode a previously coded message
sub decode($);
# Encode a message code and a message body into a single decodable string
sub encode($$);
# Same as CORE::syswrite with some error coping code
sub syswrite($$);
# Reads data from a given file descriptor and bufferizes it into a buffer
# managed by the taktuk package
sub read_data($);
# Returns the next available message for the given descriptor or the empty
# string if none available
sub get_message($);
# Find the next occurence of the required regexp (second argument) in the
# buffer associated to descriptor (first argument) and returns it
# Returns an empty string if the regexp do not match
# \n is treated as a delimiter by this function and cannot be part of the
# sequence
sub find_sequence($$);
# Returns the content of the buffer associated to a descriptor and empty it
sub flush_buffer($);
# Returns a textual description of any error that occured using send or recv
sub error_msg($);
# Sends a message to another node arguments are the node number and the message
sub send(%);
# Receive a message from another node under the form ($to, $from, $message)
sub recv(%);
# Waits for one of the given TakTuk messages. Returns its code and body
sub wait_message(@);
# Get some info from TakTuk
sub get($);
# Decode a string specifying a set of intervals and transforms it to an array
sub decode_set($);
# Encode an array of intervals into a compact string specification
sub encode_set(@);

our $VERSION = "3.6.1";
our $RELEASE = sprintf "%d", q$Rev: 453 $ =~ /(\d+)/g;
# syscalls default granularity
our $read_size = 32768;
our $write_size = 32768;
our $error = undef;

# These are codes for available messages
our $action="A";
our $eof="D";
our $taktuk_perl="E";
our $get="G";
our $invalid="H";
our $info="I";
our $option="N";
our $timeout="O";
our $put="P";
our $reduce_result="Q";
our $reduce="R";
our $spread="S";
our $taktuk_code="T";
our $update_failed="U";
our $options="V";
our $wait_message="W";
our $resign="X";
our $arguments="a";
our $broadcast="b";
our $down="d";
our $execute="e";
our $file="f";
our $get_info="g";
our $input="i";
our $kill="k";
our $message="m";
our $numbered="n";
our $output="o";
our $position="p";
our $quit="q";
our $ready="r";
our $steal="s";
our $send_to="t";
our $forward_up="u";
our $work="w";
our $synchronize="x";
our $pipe="z";

# Reduce types
our $reduce_count = 'c';
our $reduce_tree = 't';

###############################################
### SCHEDULER                               ###
### dispatching of connectors execution     ###
###############################################

package taktuk;
use strict; use bytes;

our %buffer;

sub no_flush($)
  {
    my $new_fd = shift;
    binmode($new_fd);
    my $old_fd=select($new_fd);
    $|=1;
    select($old_fd);
  }

sub unpack($)
  {
    my $buffer = shift;

    if (length($buffer) >= 4)
      {
        my $size;
        ($size) = CORE::unpack("N",$buffer);
        if (length($buffer) >= $size+4)
          {
            return (substr($buffer, 4, $size), substr($buffer, $size+4));
          }
        else
          {
            return (undef, $buffer);
          }
      }
    else
      {
        return (undef, $buffer);
      }
  }

sub pack($)
  {
    my $full_message = shift;
    my $size = length($full_message);
    return CORE::pack("N",$size).$full_message;
  }

sub decode($)
  {
    my $message = shift;
    my $message_code = substr($message, 0, 1);
    my $body = substr($message, 1);
    return ($message_code, $body);
  }

sub encode($$)
  {
    my $message = shift;
    my $body = shift;
    return ($message).($body);
  }

sub syswrite ($$)
  {
    my $unrecoverable = 0;
    my $write_fd = shift;
    my $full_message = shift;
    my $result;
    my $total_expected = length($full_message);
    my $call_expected = $write_size;
    my $offset = 0;

    while ($total_expected and not $unrecoverable)
      {
        $call_expected = $total_expected if $call_expected > $total_expected;
        $result =
            CORE::syswrite($write_fd, $full_message, $call_expected, $offset);
        if ($result)
          {
            $total_expected -= $result;
            $offset += $result;
          }
        else
          {
            if ($!{EAGAIN})
              {
                # In this case the ressource is temporarily unavailable
                # This happens on a heavily loaded system in which too many
                # writes are pending
                # Here there are two options :
                # 1) we sleep for some time loosing potential write
                # opportunities
                # 2) we make another try right now potentially overloading the
                # system
                # In any case the message server is blocked and it would be
                # better to find another solution
                #print STDERR "Delayed write ... $total_expected more to go";
                #sleep 1;
              }
            else
              {
                # This is more serious, here we probably eventually end badly
                print STDERR "Unrecoverable write error\n";
                $unrecoverable = 1;
                # I guess we should end up even more badly by killing any child
                # connector (to avoid partially deployed crashed instances)
              }
          }
      }
    if ($unrecoverable)
      {
        return undef;
      }
    else
      {
        return 1;
      }
  }

sub read_data ($)
  {
    my $descriptor = shift;
    my $new_data;

    my $result = sysread($descriptor, $new_data, $read_size);
    return undef if not defined($result);

    if ($result and exists($buffer{$descriptor}))
      {
        $buffer{$descriptor} .= $new_data;
      }
    else
      {
        $buffer{$descriptor} = $new_data;
      }
    return $result;
  }

sub get_message ($)
  {
    my $descriptor = shift;

    if (exists($buffer{$descriptor}))
      {
        my ($message, $new_buffer) = taktuk::unpack($buffer{$descriptor});

        if (defined($new_buffer))
          {
            $buffer{$descriptor} = $new_buffer;
          }
        else
          {
            delete($buffer{$descriptor});
          }
        if (defined($message))
          {
            return $message;
          }
        else
          {
            return "";
          }
      }
    else
      {
        return "";
      }
  }

sub find_sequence($$)
  {
    my $descriptor = shift;
    my $sequence = shift;
    my $found = undef;

    if (exists($buffer{$descriptor}))
      {
        my $position;

        $position = index($buffer{$descriptor},"\n");
        while (($position >= 0) and not defined($found))
          {
            my $string;

            $string = substr($buffer{$descriptor}, 0, $position);
            $buffer{$descriptor} = substr($buffer{$descriptor}, $position+1);
            if ($string =~ m/($sequence)/)
              {
                $found = $1;
              }
            else
              {
                $position = index($buffer{$descriptor},"\n");
              }
          }
      }
    return defined($found)?$found:"";
  }

sub flush_buffer($)
  {
    my $descriptor = shift;

    if (exists($buffer{$descriptor}))
      {
        my $result = $buffer{$descriptor};
        delete($buffer{$descriptor});
        return $result;
      }
    else
      {
        return "";
      }
  }

our $control_channel_read;
our $control_channel_write;

if ($ENV{TAKTUK_CONTROL_READ})
  {
    open($control_channel_read, "<&=", $ENV{TAKTUK_CONTROL_READ})
        or print("Error opening taktuk control channel : $!\n");
    binmode($control_channel_read);
  }
if ($ENV{TAKTUK_CONTROL_WRITE})
  {
    open($control_channel_write, ">&=", $ENV{TAKTUK_CONTROL_WRITE})
        or print("Error opening taktuk control channel : $!\n");
    no_flush($control_channel_write);
  }

use constant ESWRIT=>1;
use constant EFCLSD=>2;
use constant ESREAD=>3;
use constant EARGTO=>4;
use constant EARGBD=>5;
use constant ETMOUT=>6;
use constant EINVST=>7;
use constant EINVAL=>8;
use constant ENOERR=>9;

our @taktuk_errors = (
  '"taktuk::syswrite failed, system message : $!"',
  '"TakTuk engine closed the communication channel"',
  '"sysread error, system message : $!"',
  '"field \"to\" not defined"',
  '"field \"body\" not defined"',
  '"timeouted"',
  '"invalid destination set specification"',
  '"invalid field required in get"',
  '"no error"');
  

sub error_msg($)
  {
    my $error = shift;

    $error--;
    if ($error <= $#taktuk_errors)
      {
        return eval($taktuk_errors[$error]);
      }
    else
      {
        return "Unknown error";
      }
  }

sub send(%)
  {
    my %argument = @_;
    my $from = $ENV{TAKTUK_RANK};

    if (not exists($argument{to}))
      {
        $error=EARGTO;
        return undef;
      }
    my $to = $argument{to};
    if (not exists($argument{body}))
      {
        $error=EARGBD;
        return undef;
      }
    my $body = $argument{body};

    my @send_set = taktuk::decode_set($to);
    if (scalar(@send_set))
      {
        $to = taktuk::encode_set(@send_set);
        my $full_message = taktuk::encode($taktuk::send_to,
                                      taktuk::pack($to).
                                      taktuk::encode($taktuk::message,
                                      taktuk::pack($to).
                                      taktuk::pack($from).
                                      $body));
        my $result = taktuk::syswrite($control_channel_write,
                                      taktuk::pack($full_message));
        $error=ESWRIT if not $result;
        return $result?$result:undef;
      }
    else
      {
        $error=EINVST;
        return undef;
      }
  }

sub recv(%)
  {
    my %argument = @_;
    my $result;
    my $message;

    # Notification of the recv to the server
    # Necessary in all cases as a timer should not be created if a message is
    # already there (we have to count waiters internally)
    if (exists($argument{timeout}))
      {
        $message = taktuk::encode($taktuk::wait_message, $argument{timeout});
      }
    else
      {
        $message = $taktuk::wait_message;
      }
    $result = taktuk::syswrite($control_channel_write,taktuk::pack($message));
    if (not $result)
      {
        $error=ESWRIT;
        return ();
      }

    # Now we actually get the message
    my $message_code;
    ($message_code,$message) = wait_message($taktuk::timeout,$taktuk::message);

    if (defined($message_code))
      {
        my ($to, $from);
        if ($message_code eq $taktuk::timeout)
          {
            $error=ETMOUT;
            return ();
          }
        ($to, $message) = taktuk::unpack($message);
        ($from, $message) = taktuk::unpack($message);

        return ($to, $from, $message);
      }
    else
      {
        return ();
      }
  }

our @messages;

sub wait_message(@)
  {
    my @codes = @_;
    my ($code, $body);
    my $result = 1;
    my $message;

    for (my $i=0; $i<$#messages; $i+=2)
      {
        foreach my $message_code (@codes)
          {
            if ($messages[$i] eq $message_code)
              {
                ($code, $body) = ($messages[$i], $messages[$i+1]);
                splice @messages, $i, 2;
                return ($code, $body);
              }
          }
      }
    while ($result)
      {
        $message = get_message($control_channel_read);
        while ($message)
          {
            ($code, $body) = taktuk::decode($message);
            foreach my $message_code (@codes)
              {
                return ($code, $body) if ($message_code eq $code);
              }
            push @messages, $code, $body;
            $message = get_message($control_channel_read);
          }
        $result = read_data($control_channel_read);
      }
    if (defined($result))
      {
        $error=EFCLSD;
      }
    else
      {
        $error=ESREAD;
      }
    return ();
  }

sub get($)
  {
    my $result;
    my $message;

    # Infos query
    $message = taktuk::encode($taktuk::get_info, shift);
    $result = taktuk::syswrite($control_channel_write,taktuk::pack($message));
    if (not $result)
      {
        $error=ESWRIT;
        return -1;
      }

    # Now we actually get the message
    my $message_code;
    ($message_code,$message) = wait_message($taktuk::info,$taktuk::invalid);

    if (defined($message_code))
      {
        if ($message_code eq $taktuk::invalid)
          {
            $error=EINVAL;
            return -1;
          }
        else
          {
            $error=ENOERR;
          }
        return $message;
      }
    else
      {
        return -1;
      }
  }

sub decode_set($)
  {
    my $string = shift;
    my @result = ();
    my $interval;
    my $i;

    ($interval,$string) = split /\//,$string,2;
    while (length($interval))
      {
        my ($min, $max) = split /-/,$interval,2;
        my @replacement = ();
        my $state = 0;
        my $length = 0;

        $i = $#result;
        $max = $min if not $max;
        return () if ($min !~ /^\d+$/o) or ($max !~ /^\d+$/o) or
                     ($min > $max);
        while (($i > -1) and ($max < $result[$i]))
          {
            $i--;
            $state = 1 - $state;
          }
        unshift(@replacement, $max) if not $state;
        while (($i > -1) and ($min <= $result[$i]))
          {
            $i--;
            $state = 1 - $state;
            $length++;
          }
        unshift(@replacement, $min) if not $state;
        $i++;
        #print "splice @result, $i, $length, @replacement;\n";
        splice @result, $i, $length, @replacement;

        if (defined($string))
          {
            ($interval,$string) = split /\//,$string,2;
          }
        else
          {
            $interval = "";
          }
      }
    for ($i=1; $i<$#result; $i+=2)
      {
        splice @result, $i, 2
            while (($i<$#result) and ($result[$i]+1 == $result[$i+1]));
      }
    return @result;
  }

sub encode_set(@)
  {
    my @set = @_;
    my $result = "";

    while (scalar(@set))
      {
        my $min = shift @set;
        my $max = shift @set;

        $result .= "/" if ($result);
        if ($min == $max)
          {
            $result .= "$min";
          }
        else
          {
            $result .= "$min-$max";
          }
      }
    return $result;
  }

###############################################
### COMMAND                                 ###
###############################################

1;
