view release on metacpan or  search on metacpan

lib/Feersum/Runner.pm  view on Meta::CPAN

package Feersum::Runner;
use warnings;
use strict;

use EV;
use Feersum;
use Socket qw/SOMAXCONN/;
use POSIX ();
use Scalar::Util qw/weaken/;
use Carp qw/carp croak/;
use File::Spec::Functions 'rel2abs';

use constant DEATH_TIMER => 5.0; # seconds
use constant DEATH_TIMER_INCR => 2.0; # seconds
use constant DEFAULT_HOST => 'localhost';
use constant DEFAULT_PORT => 5000;

our $INSTANCE;
sub new { ## no critic (RequireArgUnpacking)
    my $c = shift;
    croak "Only one Feersum::Runner instance can be active at a time"
        if $INSTANCE && $INSTANCE->{running};
    $INSTANCE = bless {quiet=>1, @_, running=>0}, $c;
    return $INSTANCE;
}

sub DESTROY {
    local $@;
    my $self = shift;
    if (my $f = $self->{endjinn}) {
        $f->request_handler(sub{});
        $f->unlisten();
    }
    $self->{_quit} = undef;
    return;
}

sub _prepare {
    my $self = shift;

    $self->{listen} ||=
        [ ($self->{host}||DEFAULT_HOST).':'.($self->{port}||DEFAULT_PORT) ];
    croak "Feersum doesn't support multiple 'listen' directives yet"
        if @{$self->{listen}} > 1;
    my $listen = shift @{$self->{listen}};

    my $sock;
    if ($listen =~ m#^[/\.]+\w#) {
        require IO::Socket::UNIX;
        unlink $listen if -S $listen;
        my $saved = umask(0);
        $sock = IO::Socket::UNIX->new(
           Local => rel2abs($listen),
           Listen => SOMAXCONN,
        );
        umask($saved);
        croak "couldn't bind to socket: $!" unless $sock;
        $sock->blocking(0) || croak "couldn't unblock socket: $!";
    }
    else {
        require IO::Socket::INET;
        $sock = IO::Socket::INET->new(
            LocalAddr => $listen,
            ReuseAddr => 1,
            Proto => 'tcp',
            Listen => SOMAXCONN,
            Blocking => 0,
        );
        croak "couldn't bind to socket: $!" unless $sock;
    }
    $self->{sock} = $sock;
    my $f = Feersum->endjinn;
    $f->use_socket($sock);

    if (my $opts = $self->{options}) {
        $self->{$_} = delete $opts->{$_} for grep defined($opts->{$_}),
            qw/pre_fork keepalive read_timeout max_connection_reqs/;
    }
    $f->set_keepalive($_) for grep defined, delete $self->{keepalive};
    $f->read_timeout($_) for grep $_, delete $self->{read_timeout};
    $f->max_connection_reqs($_) for grep $_, delete $self->{max_connection_reqs};

    $self->{endjinn} = $f;
    return;
}

# for overriding:
sub assign_request_handler { ## no critic (RequireArgUnpacking)
    return $_[0]->{endjinn}->request_handler($_[1]);
}

sub run {
    my $self = shift;
    weaken $self;

    $self->{quiet} or warn "Feersum [$$]: starting...\n";
    $self->_prepare();

    my $app = shift || delete $self->{app};

    if (!$app && $self->{app_file}) {
        local ($@, $!);
        $app = do(rel2abs($self->{app_file}));
        warn "couldn't parse $self->{app_file}: $@" if $@;
        warn "couldn't do $self->{app_file}: $!" if ($! && !defined $app);
        warn "couldn't run $self->{app_file}: didn't return anything"
            unless $app;
    }
    die "app not defined or failed to compile" unless $app;

    $self->assign_request_handler($app);
    undef $app;

    $self->{_quit} = EV::signal 'QUIT', sub { $self->quit };

    $self->_start_pre_fork if $self->{pre_fork};
    EV::run;
    $self->{quiet} or warn "Feersum [$$]: done\n";
    $self->DESTROY();
    return;
}

sub _fork_another {
    my ($self, $slot) = @_;
    weaken $self;

    my $pid = fork;
    croak "failed to fork: $!" unless defined $pid;
    unless ($pid) {
        EV::default_loop()->loop_fork;
        $self->{quiet} or warn "Feersum [$$]: starting\n";
        delete $self->{_kids};
        delete $self->{pre_fork};
        eval { EV::run; }; ## no critic (RequireCheckingReturnValueOfEval)
        carp $@ if $@;
        POSIX::exit($@ ? -1 : 0); ## no critic (ProhibitMagicNumbers)
    }

    $self->{_n_kids}++;
    $self->{_kids}[$slot] = EV::child $pid, 0, sub {
        my $w = shift;
        $self->{quiet} or warn "Feersum [$$]: child $pid exited ".
            "with rstatus ".$w->rstatus."\n";
        $self->{_n_kids}--;
        if ($self->{_shutdown}) {
            EV::break(EV::BREAK_ALL()) unless $self->{_n_kids};
            return;
        }
        my $feersum = $self->{endjinn};
        $feersum->accept_on_fd(fileno $self->{sock});
        $self->_fork_another($slot);
        $feersum->unlisten;
    };
    return;
}

sub _start_pre_fork {
    my $self = shift;

    POSIX::setsid();

    $self->{_kids} = [];
    $self->{_n_kids} = 0;
    $self->_fork_another($_) for (1 .. $self->{pre_fork});

    $self->{endjinn}->unlisten();
    return;
}

sub quit {
    my $self = shift;
    return if $self->{_shutdown};

    $self->{_shutdown} = 1;
    $self->{quiet} or warn "Feersum [$$]: shutting down...\n";
    my $death = DEATH_TIMER;

    if ($self->{_n_kids}) {
        # in parent, broadcast SIGQUIT to the group (not self)
        kill 3, -$$; ## no critic (ProhibitMagicNumbers)
        $death += DEATH_TIMER_INCR;
    }
    else {
        # in child or solo process
        $self->{endjinn}->graceful_shutdown(sub { POSIX::exit(0) });