Gearman-Driver

 view release on metacpan or  search on metacpan

README  view on Meta::CPAN

    Gearman::Driver::Console

    *   default: 47300

    *   isa: "Int"

    Set this to 0 to disable management console at all.

  interval
    Each n seconds Net::Telnet::Gearman is used in Gearman::Driver::Observer
    to check status of free/running/busy workers on gearmand. This is used
    to fork more workers depending on the queue size and the
    MinProcesses/MaxProcesses attribute of the job method. See also:
    Gearman::Driver::Worker

    *   default: 5

    *   isa: "Int"

  max_idle_time
    Whenever Gearman::Driver::Observer notices that there are more processes

README  view on Meta::CPAN

            namespaces           => [qw(My::Workers)],
            unknown_job_callback => sub {
                my ( $driver, $status ) = @_;
                # notify nagios here for example
            }
        );

    $status might look like:

        $VAR1 = {
            'busy'    => 0,
            'free'    => 0,
            'name'    => 'GDExamples::Convert::unknown_job',
            'queue'   => 6,
            'running' => 0
        };

  worker_options
    You can pass runtime options to the worker module, these will merged
    with 'GLOBAL' and pass to the worker constructor. ( worker options
    override globals )

lib/Gearman/Driver.pm  view on Meta::CPAN

    default       => 47300,
    documentation => 'Port of management console (default: 47300)',
    is            => 'rw',
    isa           => 'Int',
    required      => 1,
);

=head2 interval

Each n seconds L<Net::Telnet::Gearman> is used in
L<Gearman::Driver::Observer> to check status of free/running/busy
workers on gearmand. This is used to fork more workers depending
on the queue size and the MinProcesses/MaxProcesses
L<attribute|Gearman::Driver::Worker/METHODATTRIBUTES> of the
job method. See also: L<Gearman::Driver::Worker>

=over 4

=item * default: C<5>

=item * isa: C<Int>

lib/Gearman/Driver.pm  view on Meta::CPAN

        namespaces           => [qw(My::Workers)],
        unknown_job_callback => sub {
            my ( $driver, $status ) = @_;
            # notify nagios here for example
        }
    );

C<$status> might look like:

    $VAR1 = {
        'busy'    => 0,
        'free'    => 0,
        'name'    => 'GDExamples::Convert::unknown_job',
        'queue'   => 6,
        'running' => 0
    };

=cut

has 'unknown_job_callback' => (
    default => sub {

lib/Gearman/Driver.pm  view on Meta::CPAN

    # When $job->add_process is called and ProcessGroup is used
    # this may end up in a race condition and more processes than
    # wanted are started. To fix that we remember what kind of
    # processes we need to start in each single run of this callback.
    my %to_start = ();

    my $status = $response->{data};
    foreach my $row (@$status) {
        if ( my $job = $self->_find_job( $row->{name} ) ) {
            $to_start{$job->name} ||= 0;
            if ( $job->count_processes <= $row->{busy} && $row->{queue} ) {
                my $diff = $row->{queue} - $row->{busy};
                my $free = $job->max_processes - $job->count_processes;
                if ($free) {
                    my $start = $diff > $free ? $free : $diff;
                    $to_start{$job->name} += $start;
                }
            }

            elsif ( $job->count_processes && $job->count_processes > $job->min_processes && $row->{queue} == 0 ) {
                my $idle = time - $job->lastrun;
                if ( $job->lastrun && ($idle >= $self->max_idle_time) ) {

lib/Gearman/Driver/Observer.pm  view on Meta::CPAN

use Net::Telnet::Gearman;
use POE;

=head1 NAME

Gearman::Driver::Observer - Observes Gearman status interface

=head1 DESCRIPTION

Each n seconds L<Net::Telnet::Gearman> is used to fetch status of
free/running/busy workers from the Gearman server. L<Gearman::Driver>
decides to fork more workers depending on the queue size and the
MinProcesses/MaxProcesses attribute of the job methods.

Currently there's no public interface.

=cut

has 'callback' => (
    is       => 'rw',
    isa      => 'CodeRef',

lib/Gearman/Driver/Observer.pm  view on Meta::CPAN

    my %data  = ();
    my @error = ();

    foreach my $telnet ( $_[OBJECT]->telnet ) {
        eval {
            my $status = $telnet->status;

            foreach my $row (@$status) {
                $data{ $row->name } ||= {
                    name    => $row->name,
                    busy    => 0,
                    free    => 0,
                    queue   => 0,
                    running => 0,
                };
                $data{ $row->name }{busy}    += $row->busy;
                $data{ $row->name }{free}    += $row->free;
                $data{ $row->name }{queue}   += $row->queue;
                $data{ $row->name }{running} += $row->running;
            }
        };

        # Try to re-open the telnet connection
        if ($@) {
            push @error, $@ if $@;
            eval { $telnet->open };