view release on metacpan or  search on metacpan

lib/GRID/Cluster.pm  view on Meta::CPAN

  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) = @_;