view release on metacpan or  search on metacpan

lib/AC/MrGamoo/Task/Running.pm  view on Meta::CPAN

    my $me = shift;

    debug("setup console: $me->{request}{jobid}, $me->{request}{console}");
    $me->{euconsole} = AC::MrGamoo::EUConsole->new( $me->{request}{jobid}, $me->{request}{console} );
}

sub _send_eumsg {
    my $me   = shift;
    my $type = shift;
    my $msg  = shift;

    return unless $me->{euconsole};
    $me->{euconsole}->send_msg($type, $msg);
}

sub _update_status {
    my $phase = shift;
    my $amt   = shift;

    # send status to parent process
    debug("sending status @ $^T / $phase/$amt");
    print STATUS "$phase $amt\n";
}

sub _maybe_update_status {
    my $me = shift;

    $^T = time();

    return if $^T < ($me->{status_time} + $STATUSTIME);
    $me->{status_time} = $^T;
    _update_status( @_ );
}

sub _setup_outfiles {
    my $me = shift;
    my @out;

    my $gz = $me->attr(undef, 'compress');
    for my $file ( @{$me->{request}{outfile}} ){
        my $f = conf_value('basedir') . '/' . $file;
        my($dir) = $f =~ m|^(.+)/[^/]+$|;

        eval{ mkpath($dir, undef, 0777) };
        push @out, AC::MrGamoo::OutFile->new( $f, $gz );
    }

    $me->{outfd} = \@out;
}

sub _close_outfiles {
    my $me = shift;

    for my $io ( @{$me->{outfd}} ){
        $io->close();
    }
    delete $me->{outfd};
}

sub _output_partition {
    my ($me, $n, $key, $data) = @_;

    # md5 is twice as fast as sha1.
    # anything written  in perl is 10 times slower
    my $hash = unpack('N', md5( $key ));
    my $p    = $hash % $n;
    my $io   = $me->{outfd}[$p];
    $io->output( encode_json( [ $key, $data ] ), "\n" );
}


# end-user's 'print' come here
sub eu_print_stdout {
    my $me = shift;

    _send_eumsg($me, 'stdout', "@_");
}

sub eu_print_stderr {
    my $me = shift;

    _send_eumsg($me, 'stderr', "@_");
}

################################################################

sub _do_map {
    my $me = shift;
    my $mr = $me->{mr};

    debug("doing map");
    my $n        = @{$me->{request}{outfile}};
    my $h_filter = $mr->get_code('filefilter');
    my $h_read   = $mr->get_code('readinput') || { code => \&readinput };
    my $h_map    = $mr->get_code('map');
    my $f_filter = $h_filter ? $h_filter->{code} : undef;
    my $f_read   = $h_read->{code};
    my $f_map    = $h_map->{code};

    my $linen = 0;

    my $maxrun = $me->attr($h_map, 'maxrun');
    alarm( $maxrun ) if $maxrun;

    for my $file (@{$me->{request}{infile}}){
        _maybe_update_status( $me, 'RUNNING', $linen );
        $me->{R}{config}{current_file} = $file;	# in case user wants for debugging

        # filter file list
        if( $f_filter ){
            next unless $f_filter->( $file );
        }

        debug("map file: $file");

        my $f = conf_value('basedir') . '/' . $file;

        open(my $fd, $f) || die "cannot open file '$f': $!\n";

        while(1){
            _maybe_update_status( $me, 'RUNNING', $linen++ );

            # read input
            my($d, $eof) = $f_read->( $fd );
            last if $eof;
            next unless defined $d;

            # map
            my($key, $data) = $f_map->( $d );
            next unless defined $key;
            _output_partition( $me, $n, $key, $data );
        }
    }

    $h_map->{cleanup}->()    if $h_map->{cleanup};
    $h_read->{cleanup}->()   if $h_read->{cleanup};
    $h_filter->{cleanup}->() if $h_filter && $h_filter->{cleanup};
}

sub _do_reduce {
    my $me = shift;
    my $mr = $me->{mr};

    my $n        = @{$me->{request}{outfile}};
    my($stage)   = $me->{request}{phase} =~ m|reduce/(\d+)|;
    my $h_reduce = $mr->get_code('reduce', $stage);
    my $f_reduce = $h_reduce->{code};
    my $rown = 0;

    my $maxrun = $me->attr($h_reduce, 'maxrun');
    alarm( $maxrun ) if $maxrun;

    debug( "doing reduce step $stage" );

    # sort
    my @cmd = _sort_cmd( $me, $h_reduce );
    open(SORT, '-|', @cmd) || die "cannot open sort pipe: $!\n";
    _sort_underway( $me, \*SORT );
    my $iter = AC::MrGamoo::Iter::File->new( \*SORT, sub{ _maybe_update_status($me, 'RUNNING', $rown++) } );

    # reduce
    while( defined(my $k = $iter->key()) ){
        _maybe_update_status( $me, 'RUNNING', $rown++ );
        my($key, $data) = $f_reduce->( $k, $iter );
        _output_partition( $me, $n, $key, $data ) if defined $key;
    }

    $h_reduce->{cleanup}() if $h_reduce->{cleanup};
}

sub _do_final {
    my $me = shift;
    my $mr = $me->{mr};

    my $h_final  = $mr->get_code('final');
    my $linen    = 0;

    if( $h_final ){
        debug("doing final");
        my $maxrun = $me->attr($h_final, 'maxrun');
        alarm( $maxrun ) if $maxrun;

        my $f_final = $h_final->{code};

        # sort
        my @cmd = _sort_cmd( $me, $h_final );
        open(SORT, '-|', @cmd) || die "cannot open sort pipe: $!\n";
        _sort_underway( $me, \*SORT );

        while(<SORT>){
            chomp;
            _maybe_update_status( $me, 'RUNNING', $linen++ );
            my $x = decode_json($_);
            $f_final->( $x->[0], $x->[1] );
        }

        $h_final->{cleanup}() if $h_final->{cleanup};
    }
}

################################################################

sub _sort_cmd {
    my $me = shift;
    my $hc = shift;

    my $gz   = $me->attr(undef, 'compress');
    my $sort = $me->attr($hc,'sortprog') || conf_value('sortprog') || $SORTPROG;
    my @file = map { conf_value('basedir') . '/' . $_ } @{$me->{request}{infile}};

    if( $gz ){
        my $zcat = $me->attr($hc,'gzprog') || conf_value('gzprog') || $GZPROG;
        my $cmd  = $zcat . ' ' . join(' ', @file) . ' | ' . $sort;
        debug("running cmd: $cmd");
        return $cmd;
    }else{
        my @cmd = ($sort, @file);
        debug("running cmd: @cmd");
        return @cmd;
    }
}

sub _sort_underway {
    my $me = shift;
    my $fd = shift;