File Coverage

blib/lib/GRID/Cluster.pm

Criterion	Covered	Total	%
statement	24	233	10.3
branch	0	42	0.0
condition	0	27	0.0
subroutine	8	22	36.3
pod	6	14	42.8
total	38	338	11.2

package GRID::Cluster;

use strict;
use warnings;

use GRID::Machine;
use GRID::Cluster::Result;
use GRID::Cluster::Handle;
use IO::Select;
use List::Util qw(sum);

# Size of the buffer which is used to read from
# remote processes
use constant BUFFERSIZE => 2048;

our $VERSION = '0.04';

# Constructor

sub new {
  my $class = shift;
  my %opts = @_;

  my $self;
  if (exists($opts{config}) && -r $opts{config}) {
    %$self = do $opts{config};
  }
  else {
    $self = {
      debug => $opts{debug},
      max_num_np => $opts{max_num_np},
      hosts => {},
    };
  }

  my @proposed_names = keys %{$self->{max_num_np}};
  $self->{host_names} = [];



  my $logic_id = 0;

  for (@proposed_names) {
    eval{
      $self->{hosts}{$_} = GRID::Machine->new (
                             host        => $_,
                             debug       => $self->{debug}{$_},
                             logic_id    => $logic_id,
                           );
    };

    if ($@) {
      warn "Warning: Host $_ has not been initialized: $@";
      delete $self->{debug}{$_};
      delete $self->{max_num_np}{$_};
      delete $self->{hosts}{$_};
    }
    else {
      push @{$self->{host_names}}, $_;
      $logic_id++;
    }
  }

  bless $self, $class;
  return $self if (@{$self->{host_names}});
  return undef;
}

# Getters

sub get_host {
  my ($self, $host_name) = @_;
  return $self->{hosts}{$host_name} if defined ($self->{hosts}{$host_name});
}

sub get_max_np {
  my $self = shift;
  return sum values %{$self->{max_num_np}};
}

sub get_num_machines {
  my $self = shift;
  my $num = @{$self->{host_names}};
  return $num;
}

sub get_machine_names {
  my $self = shift;
  return keys %{$self->{max_num_np}};
}

# Methods

sub copyandmake {
  my $self = shift;
  my %opts = @_;

  my $r = GRID::Cluster::Result->new();

  for (@{$self->{host_names}}) {
    my $machine_result = $self->{hosts}{$_}->copyandmake(%opts);
    $r->add(host_name => $_, machine_result => $machine_result);
  }

  return $r;
}

sub chdir {
  my ($self, $dir) = @_;

  my $r = GRID::Cluster::Result->new();

  for (@{$self->{host_names}}) {
    my $machine_result = $self->{hosts}{$_}->chdir($dir);
    $r->add(host_name => $_, machine_result => $machine_result);
  }

  return $r;
}

sub modput {
  my $self = shift;
  my @modules = @_;

  my $r = GRID::Cluster::Result->new();

  for (@{$self->{host_names}}) {
    my $machine_result = $self->{hosts}{$_}->modput(@modules);
    $r->add(host_name => $_, machine_result => $machine_result);
  }

  return $r;
}

sub eval {
  my $self = shift;
  my ($code, @args) = @_;

  my $r = GRID::Cluster::Result->new();

  # First round: Send eval operations to each machine
  for (@{$self->{host_names}}) {
    $self->{hosts}{$_}->send_operation("GRID::Machine::EVAL", $code, \@args);
  }

  # Second round: Receive different results from each machine
  for (@{$self->{host_names}}) {
    my $machine_result = $self->{hosts}{$_}->_get_result();
    $r->add(host_name => $_, machine_result => $machine_result);
  }

  return $r;
}

sub qx {
  my $self = shift;
  my @commands = map { "$_ | " } @_;

  my @proc;
  my @pid;
  my %map_id_machine;
  my %id;

  my $counter = 0;

  my $np = @commands;
  my $lp = $np - 1;
  my $readset = IO::Select->new();

  for (@{$self->{host_names}}) {
    for my $actual_proc (0 .. $self->{max_num_np}{$_} - 1) {
      my $m = $self->get_host($_);
      ($proc[$counter], $pid[$counter]) = $m->open(shift @commands);
      $proc[$counter]->blocking(1);

      $map_id_machine{$counter} = $_;
      $readset->add($proc[$counter]);
      my $address = 0 + $proc[$counter];
      $id{$address} = $counter;

      $counter++;

      # See if all workers are busy, if so wait for one to finish
      last if (($counter > $lp) || ($counter >= $self->get_max_np()));
    }
    last if (($counter > $lp) || ($counter >= $self->get_max_np()));
  }

  my $count = 0;
  my @ready;
  my @result;

  do {
    push @ready, $readset->can_read unless @ready;

    my $handle = shift @ready;

    my $me = $id{0 + $handle};

    my ($aux, $bytes, $r);

    while ((!defined($bytes)) || ($bytes))  {
      $bytes = sysread($handle, $aux, BUFFERSIZE);
      $r .= $aux if ((defined($bytes)) && ($bytes));
    }

    $result[$me] = $r;

    $readset->remove($handle) if eof($handle);

    close $handle;

    if (@commands) {
      my $m = $self->get_host($map_id_machine{$me});
      ($proc[$counter], $pid[$counter]) = $m->open(shift @commands);
      $proc[$counter]->blocking(1);

      $map_id_machine{$counter} = $map_id_machine{$me};
      $readset->add($proc[$counter]);
      my $address = 0 + $proc[$counter];
      $id{$address} = $counter;

      $counter++;
    }

  } until (++$count == $np);

  my @results;
  my $i = 0;

  if (wantarray) {
    foreach (@result) {
      $results[$i] = [ split /\n/, $_ ];
      $i++;
    }
    return @results;
  }
  return \@result;
}

sub open {
  my ($self, @command) = @_;

  my @proc;
  my @pid;
  my %map_id_machine;
  my %id;

  my $counter = 0;
  my $np = @command;
  my $lp = $np - 1;
  my $readset = IO::Select->new();

  for (@{$self->{host_names}}) {
    for my $actual_proc (0 .. $self->{max_num_np}{$_} - 1) {
      my $m = $self->get_host($_);
      ($proc[$counter], $pid[$counter]) = $m->open($command[$counter]);
      $proc[$counter]->blocking(1);

      $map_id_machine{$counter} = $_;
      $readset->add($proc[$counter]);
      my $address = 0 + $proc[$counter];
      $id{$address} = $counter;

      last if (++$counter > $lp);
    }
    last if ($counter > $lp);
  }

  return GRID::Cluster::Handle->new (
    readset        => $readset,
    proc           => \@proc,
    pid            => \@pid,
    id             => \%id,
    map_id_machine => \%map_id_machine
  );
}

sub open2 {
  my ($self, @command) = @_;

  my @rproc;
  my @wproc;
  my @pid;
  my %map_id_machine;
  my %id;

  my $counter = 0;
  my $np = @command;
  my $lp = $np - 1;
  my $readset = IO::Select->new();

  for (@{$self->{host_names}}) {
    for my $actual_proc (0 .. $self->{max_num_np}{$_} - 1) {
      my $m = $self->get_host($_);
      $wproc[$counter] = IO::Handle->new();
      $rproc[$counter] = IO::Handle->new();
      $pid[$counter] = $m->open2($rproc[$counter], $wproc[$counter], $command[$counter]);

      $map_id_machine{$counter} = $_;
      $readset->add($rproc[$counter]);
      my $address = 0 + $rproc[$counter];
      $id{$address} = $counter;

      last if (++$counter > $lp);
    }
    last if ($counter > $lp);
  }

  return GRID::Cluster::Handle->new (
    readset        => $readset,
    rproc          => \@rproc,
    wproc          => \@wproc,
    pid            => \@pid,
    id             => \%id,
    map_id_machine => \%map_id_machine
  );
}

sub close {
  my ($self, $cluster_handle) = @_;

  my $readset = $cluster_handle->get_readset();
  my %id = %{$cluster_handle->get_id()};
  my $np = values(%id);

  my $count = 0;
  my @ready;
  my @result;

  do {
    push @ready, $readset->can_read unless @ready;

    my $handle = shift @ready;

    my $me = $id{0 + $handle};

    my ($aux, $bytes, $r);

    while ((!defined($bytes)) || ($bytes))  {
      $bytes = sysread($handle, $aux, BUFFERSIZE);
      $r .= $aux if ((defined($bytes)) && ($bytes));
    }

    $result[$me] = $r;

    $readset->remove($handle) if eof($handle);

    close $handle;

  } until (++$count == $np);

  return \@result;
}

sub close2 {
  my ($self, $cluster_handle) = @_;

  my $readset = $cluster_handle->get_readset();
  my @wproc = @{$cluster_handle->get_wproc()};
  my %id = %{$cluster_handle->get_id()};
  my $np = values(%id);

  my $count = 0;
  my @ready;
  my @result;

  do {
    push @ready, $readset->can_read unless @ready;

    my $handle = shift @ready;

    my $me = $id{0 + $handle};

    my ($aux, $bytes, $r);

    while ((!defined($bytes)) || ($bytes))  {
      $bytes = sysread($handle, $aux, BUFFERSIZE);
      $r .= $aux if ((defined($bytes)) && ($bytes));
    }

    $result[$me] = $r;

    $readset->remove($handle) if eof($handle);

    $handle->close();
    $wproc[$count]->close();

  } until (++$count == $np);

  return \@result;
}

1;
__END__