<commit_before>package RUM::Cluster::SGE;

use strict;
use warnings;

use RUM::Logging;

use Carp;

our $log = RUM::Logging->get_logger();

sub new {
    my ($class, $config) = @_;
    my $self = {};
    $self->{config} = $config;

    my $dir = $config->output_dir;

    $self->{cmd} = {};
    $self->{cmd}{preproc}  =  "perl $0 --child --output $dir --preprocess";
    $self->{cmd}{proc}     =  "perl $0 --child --output $dir --chunk \$SGE_TASK_ID";
    $self->{cmd}{postproc} =  "perl $0 --child --output $dir --postprocess";

    $self->{jids}{preproc} = [];
    $self->{jids}{proc} = [];
    $self->{jids}{postproc} = [];

    return bless $self, $class;
}

sub config {
    $_[0]->{config};
}

sub preproc_jids  { $_[0]->{jids}{preproc} };
sub proc_jids     { $_[0]->{jids}{proc} };
sub postproc_jids { $_[0]->{jids}{postproc} };

sub _write_shell_script {
    my ($self, $phase) = @_;
    my $filename = $self->config->in_output_dir(
        $self->config->name . "_$phase" . ".sh");
    open my $out, ">", $filename or croak "Can't open $filename for writing: $!";
    my $cmd = $self->{cmd}{$phase} or croak "Don't have command for phase $phase";

    print $out 'RUM_CHUNK=$SGE_TASK_ID\n';
    print $out $self->{cmd}{$phase};
    close $out;
    return $filename;
}

sub submit_preproc {
    my ($self) = @_;
    $log->info("Submitting preprocessing job");
    my $sh = $self->_write_shell_script("preproc");
    my $jid = $self->qsub("sh", $sh);
    push @{ $self->preproc_jid }, $jid;
}

sub submit_proc {
    my ($self, $c) = @_;
    my $sh = $self->_write_shell_script("proc");
    my $n = $self->config->num_chunks;
    my @args = ("-t", "1:$n");
    my @prereqs = @{ $self->preproc_jids };
    push @args, "-hold_jid", join(",", @prereqs) if @prereqs;
    my $jid = $self->qsub(@args, "sh", $sh);
    push @{ self->proc_jids }, $jid;
}

sub submit_postproc {
    my ($self, $c) = @_;
    my $sh = $self->_write_shell_script("postproc");
    my @args;
    my @prereqs = $self->proc_jids;
    push @args, "-hold_jid", join(",", @prereqs) if @prereqs;
    push @{ $self->postproc_jids }, $self->qsub(@args, "sh", $sh);
}

sub parse_qsub_out {
    my $self = shift;
    local $_ = shift;
    /Your.* job(-array)? (\d+)/ and return $2;
}
<commit_msg>Fix bug in SGE submission<commit_after>package RUM::Cluster::SGE;

use strict;
use warnings;

use RUM::Logging;

use Carp;

our $log = RUM::Logging->get_logger();

sub new {
    my ($class, $config) = @_;
    my $self = {};
    $self->{config} = $config;

    my $dir = $config->output_dir;

    $self->{cmd} = {};
    $self->{cmd}{preproc}  =  "perl $0 --child --output $dir --preprocess";
    $self->{cmd}{proc}     =  "perl $0 --child --output $dir --chunk \$SGE_TASK_ID";
    $self->{cmd}{postproc} =  "perl $0 --child --output $dir --postprocess";

    $self->{jids}{preproc} = [];
    $self->{jids}{proc} = [];
    $self->{jids}{postproc} = [];

    return bless $self, $class;
}

sub config {
    $_[0]->{config};
}

sub preproc_jids  { $_[0]->{jids}{preproc} };
sub proc_jids     { $_[0]->{jids}{proc} };
sub postproc_jids { $_[0]->{jids}{postproc} };

sub _write_shell_script {
    my ($self, $phase) = @_;
    my $filename = $self->config->in_output_dir(
        $self->config->name . "_$phase" . ".sh");
    open my $out, ">", $filename or croak "Can't open $filename for writing: $!";
    my $cmd = $self->{cmd}{$phase} or croak "Don't have command for phase $phase";

    print $out 'RUM_CHUNK=$SGE_TASK_ID\n';
    print $out $self->{cmd}{$phase};
    close $out;
    return $filename;
}

sub submit_preproc {
    my ($self) = @_;
    $log->info("Submitting preprocessing job");
    my $sh = $self->_write_shell_script("preproc");
    my $jid = $self->qsub("sh", $sh);
    push @{ $self->preproc_jids }, $jid;
}

sub submit_proc {
    my ($self, $c) = @_;
    my $sh = $self->_write_shell_script("proc");
    my $n = $self->config->num_chunks;
    my @args = ("-t", "1:$n");
    my @prereqs = @{ $self->preproc_jids };
    push @args, "-hold_jid", join(",", @prereqs) if @prereqs;
    my $jid = $self->qsub(@args, "sh", $sh);
    push @{ $self->proc_jids }, $jid;
}

sub submit_postproc {
    my ($self, $c) = @_;
    my $sh = $self->_write_shell_script("postproc");
    my @args;
    my @prereqs = @{ $self->proc_jids };
    push @args, "-hold_jid", join(",", @prereqs) if @prereqs;
    my $jid = $self->qsub(@args, "sh", $sh);
    push @{ $self->postproc_jids }, $jid;
}

sub parse_qsub_out {
    my $self = shift;
    local $_ = shift;
    /Your.* job(-array)? (\d+)/ and return $2;
}

{'fraction_non_alphanumeric': 0.22309883848345388, 'fraction_numerical': 0.0039447731755424065, 'mean_word_length': 3.130316742081448, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 73, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '3705339', 'n_tokens_mistral': 1834, 'n_tokens_neox': 1811, 'n_words': 403}