File Coverage

blib/lib/Gearman/Glutch.pm

Criterion	Covered	Total	%
statement	63	95	66.3
branch	2	16	12.5
condition	1	6	16.6
subroutine	17	21	80.9
pod	3	6	50.0
total	86	144	59.7

package Gearman::Glutch;
use strict;
use warnings;
use utf8;
use parent qw/Class::Accessor::Fast/;

use 5.010001;
our $VERSION = '0.02';

use POSIX ":sys_wait_h";

use Gearman::Worker;
use Gearman::Server;
use Danga::Socket;

__PACKAGE__->mk_accessors(qw/server port max_workers max_reqs_per_child on_spawn_child on_complete/);

sub new {
    my $class = shift;
    my %args = @_==1 ? %{$_[0]} : @_;
    my $port = $args{port} || Carp::croak "Missing mandatory parameter: port";
    return bless {
        server => Gearman::Server->new(),
        functions => [],
        max_workers => 1,
        %args
    }, $class;
}

sub register_function {
    my $self = shift;
    push @{$self->{functions}}, \@_;
}

sub run {
    my $self = shift;

    local $SIG{PIPE} = 'IGNORE'; # handled manually

    my $ssock  = $self->server->create_listening_sock($self->port);
    $self->{ssock} = $ssock;

    # great graceful shutdown support
    $Gearmand::graceful_shutdown = 0;
    local *Gearmand::shutdown_graceful = sub {
        return if $Gearmand::graceful_shutdown;
        my $ofds = Danga::Socket->OtherFds;
        delete $ofds->{fileno($ssock)};
        $ssock->close;
        $Gearmand::graceful_shutdown = 1;
        exit 0 unless $self->server->jobs_outstanding;
    };

    # spawn workers
    for my $i (1..$self->max_workers) {
        $self->spawn_worker();
    }

    local $SIG{CHLD} = sub {
        my $kid;
        do {
            $kid = waitpid(-1, WNOHANG);
            if ($kid) {
                delete $self->{pids}->{$kid};
            }
        } while $kid > 0;

        while (0+(keys %{$self->{pids}}) < $self->max_workers) {
            $self->spawn_worker();
        }
    };

    $SIG{INT} = $SIG{TERM} = sub {
        $self->shutdown();
        exit 0;
    };

    # run main loop
    Danga::Socket->EventLoop();
}

sub spawn_worker {
    my $self = shift;

    my $pid = $self->server->start_worker(sub {
        close $self->{ssock};
        $self->{child}++;

        my $worker = Gearman::Worker->new();
        for my $func (@{$self->{functions}}) {
            $worker->register_function(@$func);
        }
        my $i=0;
        $worker->work(
            stop_if => sub {
                if (defined($self->max_reqs_per_child) && $i >= $self->max_reqs_per_child) {
                    1;
                } else {
                    0
                }
            },
            on_complete => sub {
                $self->on_complete->(@_) if $self->on_complete;
                ++$i;
            },
        );
        exit 0;
    });
    $self->on_spawn_child->($pid) if $self->on_spawn_child;
    $self->{pids}->{$pid}++;
    return $pid;
}

sub DESTROY {
    my $self = shift;
    return if $self->{child};
    $self->shutdown();
}

sub shutdown :method {
    my $self = shift;

    $self->signal_all_children('TERM');
    $self->_wait_all_children();
}

sub signal_all_children {
    my ( $self, $sig ) = @_;
    for my $pid ( sort keys %{ $self->{pids} } ) {
        kill $sig, $pid;
    }
}

sub _wait_all_children {
    my $self = shift;

    my $kid;
    do {
        $kid = waitpid(-1, WNOHANG);
    } while $kid > 0;
}

1;

__END__