view release on metacpan or  search on metacpan

lib/AC/MrGamoo/Job.pm  view on Meta::CPAN

# -*- perl -*-

# Copyright (c) 2010 AdCopy
# Author: Jeff Weisberg
# Created: 2010-Jan-13 13:03 (EST)
# Function: m/r jobs we are controlling
#
# $Id: Job.pm,v 1.6 2011/01/18 18:00:23 jaw Exp $

package AC::MrGamoo::Job;
use AC::MrGamoo::Debug 'job';
use AC::MrGamoo::Config;
use AC::MrGamoo::Job::Plan;
use AC::MrGamoo::Job::RePlan;
use AC::MrGamoo::Job::Util;
use AC::MrGamoo::Job::Done;
use AC::MrGamoo::Job::Info;
use AC::MrGamoo::Job::Action;
use AC::MrGamoo::FileList;
use AC::MrGamoo::PeerList;
use AC::MrGamoo::MySelf;
use AC::MrGamoo::EUConsole;
use AC::MrGamoo::Stats;
use AC::DC::IO;
use AC::Misc;
use JSON;
use Time::HiRes 'time';
use strict;

# RSN - config? tune?
our $TASKTIMEOUT = 10;
our $XFERTIMEOUT = 10;
our $TASKSRVRMAX = 4;
our $XFERSRVRMAX = 4;
our $REQMAX      = 10;
our $MAXLOAD     = 0.5;

our %REGISTRY;
our $MSGID = $$;
my $_trying;

our $MAXFILE = `sh -c "ulimit -n"`;
$MAXFILE = 255 if $^O eq 'solaris' && $MAXFILE > 255;

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

# schedule periodic "cronjob"
AC::DC::Sched->new(
    info	=> "job periodic",
    freq	=> 2,
    func	=> \&periodic,
   );

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

sub new {
    my $class = shift;
    # %{ APCMRMJobCreate }

    my $me = bless {
        request		=> { @_ },
        phase_no	=> -1,
        file_info	=> {},
        tmp_file	=> [],
        server_info	=> {},
        task_running	=> {},
        task_pending	=> {},
        xfer_running	=> {},
        xfer_pending	=> {},
        request_running	=> {},
        request_pending	=> {},
        statistics	=> { job_start => time() },
    }, $class;

    if( $REGISTRY{ $me->{request}{jobid} } ){
        verbose("ignoring duplicate request job $me->{request}{jobid}");
        # will cause a 200 OK, so the requestor will not retry
        return $REGISTRY{ $me->{request}{jobid} };
    }

    verbose("new job: $me->{request}{jobid} ($me->{request}{traceinfo})");

    my $cf = $me->{options} = decode_json( $me->{request}{options} ) if $me->{request}{options};

    # open connection  to eu-console
    $me->{euconsole} = AC::MrGamoo::EUConsole->new( $me->{request}{jobid}, $me->{request}{console} );

    # partially compile
    eval {
        $me->{mr} = AC::MrGamoo::Submit::Compile->new( text => $me->{request}{jobsrc} );
    };
    if(my $e = $@){
        problem("cannot compile job: $e");
        return;
    }

    # RSN - get_file_list + Plan may take too long - do in sub-process

    # get file list
    my $files = get_file_list( $cf );
    #print STDERR "files: ", dumper($files), "\n";

    for my $f (@$files){
        $me->{file_info}{ $f->{filename} } = $f;
    }

    # get server list
    my $servers = get_peer_list( $cf );
    #print STDERR "servers: ", dumper($servers), "\n";

    # plan job
    my $plan = AC::MrGamoo::Job::Plan->new( $me, $servers, $files );
    #print STDERR "plan: ", dumper($plan), "\n";

    $me->{plan} = $plan;

    $me->{maxfail} = 5 * ( (keys %{$plan->{taskidx}}) + @{$plan->{copying}});

    $me->{server_info}{$_->{id}} = {} for @$servers;