File Coverage

blib/lib/Feersum/Runner.pm

Criterion	Covered	Total	%
statement	198	334	59.2
branch	66	194	34.0
condition	15	62	24.1
subroutine	28	31	90.3
pod	4	4	100.0
total	311	625	49.7

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

use EV;
use Feersum;
use Socket qw/SOMAXCONN SOL_SOCKET SO_REUSEADDR AF_INET SOCK_STREAM
              inet_aton pack_sockaddr_in/;
BEGIN {
    # IPv6 support (Socket 1.95+, Perl 5.14+)
    eval { Socket->import(qw/AF_INET6 inet_pton pack_sockaddr_in6/); 1 }
        or do {
            *AF_INET6 = sub () { undef };
            *inet_pton = sub { undef };
            *pack_sockaddr_in6 = sub { undef };
        };
}
BEGIN {
    # SO_REUSEPORT may not be available on all systems
    eval { Socket->import('SO_REUSEPORT'); 1 }
        or *SO_REUSEPORT = sub () { undef };
}
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;
use constant MAX_PRE_FORK => $ENV{FEERSUM_MAX_PRE_FORK} || 1000;

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

sub _cleanup {
    my $self = shift;
    return if $self->{_cleaned_up};
    $self->{_cleaned_up} = 1;
    if (my $f = $self->{endjinn}) {
        $f->request_handler(sub{});
        $f->unlisten();
    }
    $self->{_quit} = undef;
    $self->{running} = 0;
    return;
}

sub DESTROY {
    local $@;
    $_[0]->_cleanup();
}

sub _create_socket {
    my ($self, $listen, $use_reuseport) = @_;

    my $sock;
    if ($listen =~ m#^[/\.]+\w#) {
        require IO::Socket::UNIX;
        if (-S $listen) {
            unlink $listen or carp "unlink stale socket '$listen': $!";
        }
        my $saved = umask(0);
        $sock = eval {
            IO::Socket::UNIX->new(
               Local => rel2abs($listen),
               Listen => SOMAXCONN,
            );
        };
        my $err = $@;
        umask($saved);  # Restore umask even if socket creation failed
        die $err if $err;
        croak "couldn't bind to socket" unless $sock;
        $sock->blocking(0) || do { close($sock); croak "couldn't unblock socket: $!"; };
    }
    else {
        require IO::Socket::INET;
        # SO_REUSEPORT must be set BEFORE bind for multiple sockets per port
        if ($use_reuseport && defined SO_REUSEPORT) {
            # Parse listen address - handle IPv6 bracketed notation [host]:port
            my ($host, $port, $is_ipv6);
            if ($listen =~ /^\[([^\]]+)\]:(\d*)$/) {
                # IPv6 with port: [::1]:8080
                ($host, $port, $is_ipv6) = ($1, $2 || 0, 1);
            } elsif ($listen =~ /^\[([^\]]+)\]$/) {
                # IPv6 without port: [::1]
                ($host, $port, $is_ipv6) = ($1, 0, 1);
            } elsif ($listen =~ /:.*:/) {
                # Bare IPv6 - reject ambiguous cases that look like host:port
                if ($listen =~ /:(\d{1,5})$/) {
                    my $maybe_port = $1;
                    # 5 digits = definitely a port; >=1024 = likely a port
                    if ($maybe_port <= 65535 && (length($maybe_port) == 5 || $maybe_port >= 1024)) {
                        croak "ambiguous IPv6 address '$listen': use bracket notation [host]:port " .
                              "(e.g., [::1]:$maybe_port or [2001:db8::1]:$maybe_port)";
                    }
                }
                ($host, $port, $is_ipv6) = ($listen, 0, 1);
            } else {
                # IPv4: host:port
                ($host, $port) = split /:/, $listen, 2;
                $host ||= '0.0.0.0';
                $port ||= 0;
                $is_ipv6 = 0;
            }

            # Validate port range (0-65535)
            if ($port !~ /^\d+$/ || $port > 65535) {
                croak "invalid port '$port': must be 0-65535";
            }

            my ($domain, $sockaddr);
            if ($is_ipv6) {
                defined AF_INET6()
                    or croak "IPv6 not supported on this system";
                my $addr = inet_pton(AF_INET6(), $host)
                    or croak "couldn't resolve IPv6 address '$host'";
                $domain = AF_INET6();
                $sockaddr = pack_sockaddr_in6($port, $addr);
            } else {
                my $addr = inet_aton($host)
                    or croak "couldn't resolve address '$host'";
                $domain = AF_INET();
                $sockaddr = pack_sockaddr_in($port, $addr);
            }

            # Create socket with correct address family
            socket($sock, $domain, SOCK_STREAM(), 0)
                or croak "couldn't create socket: $!";
            setsockopt($sock, SOL_SOCKET, SO_REUSEADDR, pack("i", 1))
                or do { close($sock); croak "setsockopt SO_REUSEADDR failed: $!"; };
            setsockopt($sock, SOL_SOCKET, SO_REUSEPORT, pack("i", 1))
                or do { close($sock); croak "setsockopt SO_REUSEPORT failed: $!"; };
            bind($sock, $sockaddr)
                or do { close($sock); croak "couldn't bind to socket: $!"; };
            listen($sock, SOMAXCONN)
                or do { close($sock); croak "couldn't listen: $!"; };

            # Wrap in IO::Handle for ->blocking() method
            require IO::Handle;
            bless $sock, 'IO::Handle';
            $sock->blocking(0)
                || do { close($sock); croak "couldn't unblock socket: $!"; };
        }
        else {
            # Validate port in listen address for better error messages
            if ($listen =~ /:(\d+)$/) {
                my $port = $1;
                croak "invalid port '$port': must be 0-65535" if $port > 65535;
            } elsif ($listen =~ /:(\S+)$/) {
                my $port = $1;
                croak "invalid port '$port': must be numeric" unless $port =~ /^\d+$/;
            }
            $sock = IO::Socket::INET->new(
                LocalAddr => $listen,
                ReuseAddr => 1,
                Proto => 'tcp',
                Listen => SOMAXCONN,
                Blocking => 0,
            );
            croak "couldn't bind to socket: $!" unless $sock;
        }
    }
    return $sock;
}

sub _prepare {
    my $self = shift;

    # Normalize listen to arrayref (accept scalar for convenience)
    if (defined $self->{listen} && !ref $self->{listen}) {
        $self->{listen} = [ $self->{listen} ];
    }
    $self->{listen} ||=
        [ ($self->{host}||DEFAULT_HOST).':'.($self->{port}||DEFAULT_PORT) ];
    croak "listen must be an array reference"
        if ref $self->{listen} ne 'ARRAY';
    croak "listen array cannot be empty"
        if @{$self->{listen}} == 0;
    $self->{_listen_addrs} = [ @{$self->{listen}} ];

    if (my $opts = $self->{options}) {
        $self->{$_} = delete $opts->{$_} for grep defined($opts->{$_}),
            qw/pre_fork keepalive read_timeout header_timeout write_timeout max_connection_reqs reuseport epoll_exclusive
               read_priority write_priority accept_priority max_accept_per_loop max_connections
               max_read_buf max_body_len max_uri_len wbuf_low_water
               reverse_proxy proxy_protocol psgix_io h2 tls tls_cert_file tls_key_file/;
        # Warn about unknown options (likely typos)
        for my $unknown (keys %$opts) {
            carp "Unknown option '$unknown' ignored";
        }
    }

    # Validate pre_fork early (before socket creation) to fail fast
    if ($self->{pre_fork}) {
        my $n = $self->{pre_fork};
        if ($n !~ /^\d+$/ || $n < 1) {
            croak "pre_fork must be a positive integer";
        }
        if ($n > MAX_PRE_FORK) {
            croak "pre_fork=$n exceeds maximum of " . MAX_PRE_FORK;
        }
    }

    # Enable reuseport automatically in prefork mode if SO_REUSEPORT available
    my $use_reuseport = $self->{reuseport} && $self->{pre_fork} && defined SO_REUSEPORT;
    $self->{_use_reuseport} = $use_reuseport;

    my $f = Feersum->endjinn;

    # EPOLLEXCLUSIVE must be set BEFORE use_socket() so the separate accept epoll
    # is created with EPOLLEXCLUSIVE flag (Linux 4.5+)
    if ($self->{epoll_exclusive} && $self->{pre_fork} && $^O eq 'linux') {
        $f->set_epoll_exclusive(1);
    }

    # Create sockets and attach to server for each listen address
    my @socks;
    for my $listen (@{$self->{_listen_addrs}}) {
        my $sock = $self->_create_socket($listen, $use_reuseport);
        push @socks, $sock;
        $f->use_socket($sock);
    }
    $self->{sock} = $socks[0];   # backward compat: primary socket
    $self->{_socks} = \@socks;   # all sockets

    $f->set_keepalive($_) for grep defined, delete $self->{keepalive};
    $f->set_reverse_proxy($_) for grep defined, delete $self->{reverse_proxy};
    $f->set_proxy_protocol($_) for grep defined, delete $self->{proxy_protocol};
    $f->set_psgix_io($_) for grep defined, delete $self->{psgix_io};
    $f->read_timeout($_) for grep defined, delete $self->{read_timeout};
    $f->header_timeout($_) for grep defined, delete $self->{header_timeout};
    $f->write_timeout($_) for grep defined, delete $self->{write_timeout};
    $f->max_connection_reqs($_) for grep defined, delete $self->{max_connection_reqs};
    # Validate priority values (-2 to +2 per libev)
    for my $prio_name (qw/read_priority write_priority accept_priority/) {
        my $val = $self->{$prio_name};
        if (defined $val) {
            # Must be an integer (not float, not string)
            croak "$prio_name must be an integer" unless $val =~ /^-?\d+$/;
            croak "$prio_name must be between -2 and 2" if $val < -2 || $val > 2;
        }
    }
    $f->read_priority($_) for grep defined, delete $self->{read_priority};
    $f->write_priority($_) for grep defined, delete $self->{write_priority};
    $f->accept_priority($_) for grep defined, delete $self->{accept_priority};
    # Validate max_accept_per_loop (must be positive integer)
    if (defined(my $val = $self->{max_accept_per_loop})) {
        croak "max_accept_per_loop must be a positive integer"
            unless $val =~ /^\d+$/ && $val > 0;
    }
    $f->max_accept_per_loop($_) for grep defined, delete $self->{max_accept_per_loop};
    # Validate max_connections (must be non-negative integer, 0 = unlimited)
    if (defined(my $val = $self->{max_connections})) {
        croak "max_connections must be a non-negative integer"
            unless $val =~ /^\d+$/;
    }
    $f->max_connections($_) for grep defined, delete $self->{max_connections};
    $f->max_read_buf($_) for grep defined, delete $self->{max_read_buf};
    $f->max_body_len($_) for grep defined, delete $self->{max_body_len};
    $f->max_uri_len($_) for grep defined, delete $self->{max_uri_len};
    $f->wbuf_low_water($_) for grep defined, delete $self->{wbuf_low_water};

    # Build tls hash from flat options (for Plack -o tls_cert_file=... -o tls_key_file=...)
    if (!$self->{tls}) {
        if (my $cert = delete $self->{tls_cert_file}) {
            my $key = delete $self->{tls_key_file}
                or croak "tls_cert_file requires tls_key_file";
            $self->{tls} = { cert_file => $cert, key_file => $key };
        } elsif (delete $self->{tls_key_file}) {
            croak "tls_key_file requires tls_cert_file";
        }
    } else {
        # tls hash takes precedence; discard flat options
        delete $self->{tls_cert_file};
        delete $self->{tls_key_file};
    }

    # TLS configuration: apply to all listeners
    if (my $tls = delete $self->{tls}) {
        croak "tls must be a hash reference" unless ref $tls eq 'HASH';
        croak "tls requires cert_file" unless $tls->{cert_file};
        croak "tls requires key_file" unless $tls->{key_file};
        -f $tls->{cert_file} && -r _
            or croak "tls cert_file '$tls->{cert_file}': not found or not readable";
        -f $tls->{key_file} && -r _
            or croak "tls key_file '$tls->{key_file}': not found or not readable";

        # H2 is off by default; only enable if h2 => 1 was passed
        if (delete $self->{h2}) {
            $tls->{h2} = 1;
        }

        if ($f->has_tls()) {
            for my $i (0 .. $#socks) {
                $f->set_tls(listener => $i, %$tls);
            }
            # Save TLS config for reuseport children to re-apply
            $self->{_tls_config} = $tls;
            $self->{quiet} or warn "Feersum [$$]: TLS enabled on "
                . scalar(@socks) . " listener(s)\n";
        } else {
            croak "tls option requires Feersum compiled with TLS support (need picotls submodule + OpenSSL; see Alien::OpenSSL)";
        }
    } else {
        if (delete $self->{h2}) {
            croak "h2 requires TLS (provide tls_cert_file and tls_key_file, or a tls hash)";
        }
    }

    $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->{running} = 1;
    $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;
    }
    croak "app not defined or failed to compile" unless $app;

    $self->assign_request_handler($app);

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

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

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

    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};
        $self->{_n_kids} = 0;

        # With SO_REUSEPORT, each child creates its own sockets
        # This eliminates accept() contention for better scaling
        if ($self->{_use_reuseport}) {
            $self->{endjinn}->unlisten();
            for my $old_sock (@{$self->{_socks} || []}) {
                close($old_sock)
                    or do { warn "close parent socket in child: $!"; POSIX::_exit(1); };
            }
            my @new_socks;
            eval {
                for my $listen (@{$self->{_listen_addrs}}) {
                    my $sock = $self->_create_socket($listen, 1);
                    push @new_socks, $sock;
                    $self->{endjinn}->use_socket($sock);
                }
                1;
            } or do {
                warn "Feersum [$$]: child socket creation failed: $@";
                POSIX::_exit(1);
            };
            $self->{sock} = $new_socks[0];
            $self->{_socks} = \@new_socks;

            # Re-apply TLS config on new listeners
            if (my $tls = $self->{_tls_config}) {
                for my $i (0 .. $#new_socks) {
                    $self->{endjinn}->set_tls(listener => $i, %$tls);
                }
            }
        }

        eval { EV::run; }; ## no critic (RequireCheckingReturnValueOfEval)
        carp $@ if $@;
        POSIX::_exit($@ ? 1 : 0);  # _exit avoids running parent's END blocks
    }

    weaken $self;  # prevent circular ref with watcher callback
    $self->{_n_kids}++;
    $self->{_kids}[$slot] = EV::child $pid, 0, sub {
        my $w = shift;
        return unless $self;  # guard against destruction during shutdown
        $self->{quiet} or warn "Feersum [$$]: child $pid exited ".
            "with rstatus ".$w->rstatus."\n";
        $self->{_n_kids}--;
        if ($self->{_shutdown}) {
            unless ($self->{_n_kids}) {
                $self->{_death} = undef;
                EV::break(EV::BREAK_ALL());
            }
            return;
        }
        # Without SO_REUSEPORT, parent needs to accept during respawn
        unless ($self->{_use_reuseport}) {
            my $feersum = $self->{endjinn};
            my @socks = @{$self->{_socks} || [$self->{sock}]};
            my $all_valid = 1;
            for my $sock (@socks) {
                unless (defined fileno $sock) {
                    $all_valid = 0;
                    last;
                }
            }
            if ($all_valid) {
                for my $sock (@socks) {
                    $feersum->accept_on_fd(fileno $sock);
                }
                $self->_fork_another($slot);
                $feersum->unlisten;
            } else {
                carp "fileno returned undef during respawn, cannot respawn worker";
            }
        }
        else {
            # With SO_REUSEPORT, just spawn new child (it creates its own socket)
            $self->_fork_another($slot);
        }
    };
    return;
}

sub _start_pre_fork {
    my $self = shift;

    # pre_fork value already validated in _prepare()

    POSIX::setsid() or croak "setsid() failed: $!";

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

    # Parent stops accepting - children handle connections
    $self->{endjinn}->unlisten();

    # With SO_REUSEPORT, parent can close its sockets entirely
    # Children have their own sockets
    if ($self->{_use_reuseport}) {
        for my $sock (@{$self->{_socks} || []}) {
            close($sock)
                or warn "close parent socket after fork: $!";
        }
        $self->{sock} = undef;
        $self->{_socks} = [];
    }
    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 process group (including self,
        # but protected by _shutdown flag above)
        kill POSIX::SIGQUIT, -$$;
        $death += DEATH_TIMER_INCR;
    }
    else {
        # in child or solo process
        $self->{endjinn}->graceful_shutdown(sub { POSIX::_exit(0) });
    }

    $self->{_death} = EV::timer $death, 0, sub { POSIX::_exit(1) };
    return;
}

1;
__END__