view release on metacpan or  search on metacpan

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

        server	=> $server,
        created => time(),
    };

    $job->{task_pending}{$id} = $me;

    debug("  => pending task $info->{id} => $id on $server");

    return $me;
}

sub start {
    my $me  = shift;
    my $job = shift;

    my $server = $me->{server};
    debug("starting task $job->{request}{jobid}/$me->{info}{id}/$me->{id} on $server");

    # send request to server
    my $ti = $me->{info};

    my $x = $job->_send_request( $server, "task $me->{id}", {
        type		=> 'mrgamoo_taskcreate',
        msgidno		=> $^T,
        want_reply	=> 1,
    }, {
        jobid		=> $job->{request}{jobid},
        taskid		=> $me->{id},
        jobsrc		=> $job->{request}{jobsrc},
        options		=> $job->{request}{options},
        initres		=> $job->{request}{initres},
        console		=> ($job->{request}{console} || ''),
        phase		=> $ti->{phase},
        infile		=> $ti->{infile},
        outfile		=> [ map { $_->{filename} } @{$ti->{outfile}} ],
        master		=> my_server_id(),
    } );

    unless( $x ){
        verbose("cannot start task");
        $me->failed($job);
        return;
    }

    # no success cb here. we will either timeout, or get a TaskStatus msg.
    $x->set_callback('on_failure', \&_cb_start_task_fail, $me, $job );

    $me->started($job, 'task');
    $x->start();
}

sub _cb_start_task_fail {
    my $io  = shift;
    my $evt = shift;
    my $me  = shift;
    my $job = shift;

    $me->failed($job, 'network');
}

sub update_status {
    my $me  = shift;
    my $job = shift;
    my $phase = shift;
    my $progress = shift;

    $me->{status_time}  = $^T;
    $me->{status_phase} = $phase;
    $me->{status_amt}   = $progress;
    $me->{status_fail}  = 1 if $phase eq 'FAILED';

    debug("task is $phase $progress");

    if( $phase eq 'FINISHED' ){
        if( $me->{status_fail} ){
            $me->failed( $job, "status fail" );
        }else{
            $me->finished( $job );
        }
    }

    return 1;
}

sub failed {
    my $me   = shift;
    my $job  = shift;
    my $why  = shift;

    debug("task failed: $why $me->{status_time}");

    return if $job->something_failed();
    $me->SUPER::failed($job, 'task');
    $me->{info}->failed( $me, $job );
    if( $why eq 'timeout' ){
        $me->abort($job)
    }else{
        # $job->_try_to_do_something();
    }
}

sub finished {
    my $me   = shift;
    my $job  = shift;
    my $why  = shift;

    debug('task finish');

    $me->SUPER::finished($job, 'task');
    $me->{info}->finished( $me, $job );

    $job->_try_to_do_something();
}

sub abort {
    my $me  = shift;
    my $job = shift;

    debug("aborting task $me->{id}");

    AC::MrGamoo::Job::Request->new( $job,