libtaktuk-perl 3.7.5-1 / usr / share / perl5 / TakTuk.pm

###############################################################################
#                                                                             #
#  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                                           #
#           Laboratoire LIG - ENSIMAG - Antenne de Montbonnot                 #
#           51 avenue Jean Kuntzmann                                          #
#           38330 Montbonnot Saint Martin                                     #
#           FRANCE                                                            #
#                                                                             #
###############################################################################

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($);
 
BEGIN {
    die "FATAL : Perl interpreter too old ($]), Taktuk require Perl >= 5.6.1\n"
        if ($] < 5.006001);
}

our $VERSION = "3.7.5";
our $VERSION_NETPROTO = 1;
our $RELEASE = sprintf "%d", q$Rev: 569 $ =~ /(\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 $wait="B";
our $id="C";
our $eof="D";
our $taktuk_perl="E";
our $gateway="F";
our $get="G";
our $invalid="H";
our $info="I";
our $command_send_to="J";
our $command_message="K";
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';
our $reduce_wait = 'w';

# State events
use constant TAKTUK_READY => 0;
use constant TAKTUK_NUMBERED => 1;
use constant TAKTUK_TERMINATED => 2;
use constant CONNECTION_FAILED => 3;
use constant CONNECTION_INITIALIZED => 4;
use constant CONNECTION_LOST => 5;
use constant COMMAND_STARTED => 6;
use constant COMMAND_FAILED => 7;
use constant COMMAND_TERMINATED => 8;
use constant UPDATE_FAILED => 9;
use constant PIPE_STARTED => 10;
use constant PIPE_FAILED => 11;
use constant PIPE_TERMINATED => 12;
use constant FILE_RECEPTION_STARTED =>13;
use constant FILE_RECEPTION_FAILED =>14;
use constant FILE_RECEPTION_TERMINATED =>15;
use constant FILE_SEND_FAILED =>16;
use constant INVALID_TARGET => 17;
use constant NO_TARGET => 18;
use constant MESSAGE_DELIVERED => 19;
use constant INVALID_DESTINATION => 20;
use constant UNAVAILABLE_DESTINATION => 21;
use constant WAIT_COMPLETE => 22;
use constant WAIT_REDUCE_COMPLETE => 23;

###############################################
### 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};
    my $target;

    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};
    if (not exists($argument{target})) {
        $target = "any";
    } else {
        $target = $argument{target};
    }

    my $full_message = TakTuk::encode($TakTuk::send_to,
                                  TakTuk::pack($to).
                                  TakTuk::encode($TakTuk::message,
                                  TakTuk::pack($target).
                                  TakTuk::pack($from).
                                  $body));
    my $result = TakTuk::syswrite($control_channel_write,
                                  TakTuk::pack($full_message));
    $error=ESWRIT if not $result;
    return $result?$result: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 $from;
        if ($message_code eq $TakTuk::timeout) {
            $error=ETMOUT;
            return ();
        }
        ($from, $message) = TakTuk::unpack($message);

        return ($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;
    }
}

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

1;
###############################################################################
#                                                                             #
#  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                                           #
#           Laboratoire LIG - ENSIMAG - Antenne de Montbonnot                 #
#           51 avenue Jean Kuntzmann                                          #
#           38330 Montbonnot Saint Martin                                     #
#           FRANCE                                                            #
#                                                                             #
###############################################################################

=pod TakTuk communication layer interface documentation (Perl interface)

=begin html

<center><h1>USER MANUAL</h1></center>

=end html

=head1 NAME

TakTuk - Perl module that provides an interface to C<taktuk(1)> communication
facilities

=head1 SYNOPSIS

  use TakTuk;
  
  my $rank = TakTuk::get('rank');
  my $count = TakTuk::get('count');
  
  print "I'm process $rank among $count\n";
  
  if ($rank > 1) {
      my ($from, $message) = TakTuk::recv();
      if (not defined($message)) {
          print "Trying to recv: ",
                TakTuk::error_msg($TakTuk::error), "\n";
      } else {
          print "$rank received $message from $from\n";
      }
  }
  
  sleep 1;
  my $next = $rank+1;
  $next = 1 if ($next > $count);
  if (not TakTuk::send(to=>$next, body=>"[Salut numero $rank]")) {
      print "Trying to send to $next: ",
            TakTuk::error_msg($TakTuk::error), "\n";
  }
  
  if ($rank == 1) {
      my ($from, $message) = TakTuk::recv(timeout=>5);
      if (not defined($message)) {
          print "Trying to recv :",
                TakTuk::error_msg($TakTuk::error), "\n";
      } else {
          print "$rank received $message from $from\n";
      }
  }

=head1 DESCRIPTION

The B<TakTuk> communication layer Perl interface provides a way for programs
executed using the C<taktuk(1)> command to exchange data. It is based on a
simple send/receive model using multicast-like sends and optionally timeouted
receives.  This is only designed to be a control facility, in particular this
is not a high performance communication library.

The Perl communication interface for B<TakTuk> is made of functions that can be
called by scripts executed using the C<taktuk_perl> command of the B<TakTuk>
engine (preferred way, less installation requirements on remote machines) or
using the B<TakTuk> Perl module provided with the B<TakTuk> distribution.
These functions are:

=over

=item B<TakTuk::get($)>

gets some information from B<TakTuk>. Currently available information includes
'target', 'rank', 'count', 'father', 'child_min' and 'child_max'. This is a
better way to get this information than environment variables as its takes
into account renumbering that might occur after process spawn.

=item B<TakTuk::send(%)>

sends a scalar to a single peer or a set specification (see C<taktuk(1)> for
information about set specifications).  The two mandatory fields in the
arguments are C<to> (with a set specification) and C<body>. Optionally, a field
C<target> might be given. Returns an undefined value upon error.

=item B<TakTuk::recv(%)>

blocks until the reception of a message. Returns a list of two elements:
the logical number of the source of the message and the message itself.
Accepts an optional C<timeout> argument with a numeric value.
Returns an empty list upon error.

=back

When an error occur, all these functions set the variable C<$TakTuk::error>
to the numeric code of the error that occured. A textual description of the
error is provided by the function C<TakTuk::error_msg($)> that takes the error
code as an argument.

Error codes are the following :

=over

=item TakTuk::ESWRIT

a call to C<TakTuk::syswrite> failed. This is due to a C<syswrite> error
different than C<EAGAIN>. The code should be accessible using C<$!>.

=item TakTuk::EFCLSD

the communication channel to the B<TakTuk> engine has been closed. This
typically occur when shutting down the logical network (using Ctrl-C on root
node for instance).

=item TakTuk::ESREAD (C<TakTuk::recv> only)

a call to C<sysread> failed (the code should be accessible using C<$!>).

=item TakTuk::EARGTO (C<TakTuk::send> only)

C<to> field missing in the arguments.

=item TakTuk::EARGBD (C<TakTuk::send> only)

C<body> field missing in the arguments.

=item TakTuk::ETMOUT (C<TakTuk::recv> only)

The call to C<TakTuk::recv> timeouted. This only occur when giving a C<timeout>
field as C<TakTuk::recv> argument.

=begin comment

=item TakTuk::EINVST (C<TakTuk::send> only)

The set specification given as a destination to the C<TakTuk::send> function is
not correct.

=end comment

=back

Finally, the B<TakTuk> Perl module defines some constants which value match the
different states reported by the stream C<state> (see C<taktuk(1)> for details
about this stream). These constant are the following:

  TakTuk::TAKTUK_READY
  TakTuk::TAKTUK_NUMBERED
  TakTuk::TAKTUK_TERMINATED
  TakTuk::CONNECTION_FAILED
  TakTuk::CONNECTION_INITIALIZED
  TakTuk::CONNECTION_LOST
  TakTuk::COMMAND_STARTED
  TakTuk::COMMAND_FAILED
  TakTuk::COMMAND_TERMINATED
  TakTuk::UPDATE_FAILED
  TakTuk::PIPE_STARTED
  TakTuk::PIPE_FAILED
  TakTuk::PIPE_TERMINATED
  TakTuk::FILE_RECEPTION_STARTED
  TakTuk::FILE_RECEPTION_FAILED
  TakTuk::FILE_RECEPTION_TERMINATED
  TakTuk::FILE_SEND_FAILED
  TakTuk::INVALID_TARGET
  TakTuk::NO_TARGET
  TakTuk::MESSAGE_DELIVERED
  TakTuk::INVALID_DESTINATION
  TakTuk::UNAVAILABLE_DESTINATION

=head1 SEE ALSO

C<tatkuk(1)>, C<taktukcomm(3)>, C<TakTuk::Pilot(3)>

=head1 AUTHOR

The original concept of B<TakTuk> has been proposed by Cyrille Martin in his PhD thesis. People involved in this work include Jacques Briat, Olivier Richard, Thierry Gautier and Guillaume Huard.

The author of the version 3 (perl version) and current maintainer of the package is Guillaume Huard.

=head1 COPYRIGHT

The C<TakTuk> communication interface library is provided under the terms
of the GNU General Public License version 2 or later.

=cut