perl echo servers

poll.pl

#!/usr/bin/env perl

use strict;
use warnings;
use FindBin;
use lib "$FindBin::Bin/../lib";
use Getopt::Long;
use IO::Socket::INET;
use IO::Poll;
use Log::Minimal;
use Pod::Usage;
use Socket;

our $VERSION = '0.01';
our $BUFFER_SIZE = 1024;
local $Log::Minimal::LOG_LEVEL = 'DEBUG';
local $Log::Minimal::AUTODUMP  = 1;

my $opts = +{};
GetOptions(
    $opts,
    'listen|l=s',
    'timeout|t=s',
    'help|h', 'man',
);

pod2usage(1) if ($opts->{help});
pod2usage(-verbose => 2) if ($opts->{man});

sub handle_connection {
    my ( $p, $conn, $opts ) = @_;

    $conn->recv(my $buf, $BUFFER_SIZE);
    unless ( defined $buf && length $buf > 0 ) {
        warnf('connection closed');
        $p->remove(0);
        $conn->close;
        return 0;
    }

    warnf('buffer: %s', $buf);
    unless ( $conn->send($buf) ) {
        warnf('write error');
        $p->remove($conn);
        $conn->close;
        return 0;
    }

    return 1;
}

sub main {
    my $opts = shift;

    warnf('pid: %d', $$);
    my $p = IO::Poll->new;
    my $listen = IO::Socket::INET->new(
        Blocking  => 0,
        LocalHost => 'localhost',
        LocalPort => $opts->{listen},
        Listen    => 1,
        Proto     => 'tcp',
        ReuseAddr => 1,
        Timeout   => $opts->{timeout},
        Type      => SOCK_STREAM,
    );

    $p->mask( $listen => POLLIN|POLLHUP|POLLERR );

    my $clients = 0;
    my $term_recieved = 0;

    $SIG{TERM} = sub {
        $term_recieved++;
        warnf('term signal recieved');
    };

    while (1) {
        if ( $term_recieved ) {
            warnf("shutfown");
            exit 0;
        }

        $p->poll($opts->{timeout});

        if ( my @ready = $p->handles(POLLIN|POLLHUP|POLLERR) ) {
            warnf('ready handles (%s)', \@ready);
            for my $sock (@ready) {
                warnf('sock: %s, events: %d', fileno($sock), $p->events($sock));

                if ( $sock == $listen ) {
                    $clients++;
                    warnf('New connection (clients: %d)', $clients);
                    my $conn = $listen->accept;
                    $conn->blocking(0);
                    $p->mask( $conn => POLLIN|POLLHUP|POLLERR );
                }
                else {
                    unless ( handle_connection($p, $sock, $opts) ) {
                        $clients--;
                    }
                }
            }
        }
    }
}

%$opts = (
    listen => 50000,
    timeout => 10,
    %$opts,
);

main($opts);

prefork.pl

#!/usr/bin/env perl

use strict;
use warnings;
use FindBin;
use lib "$FindBin::Bin/../lib";
use Getopt::Long;
use Socket qw(IPPROTO_TCP TCP_NODELAY);
use IO::Socket::INET;
use Log::Minimal;
use Parallel::Prefork;
use Pod::Usage;

our $VERSION = '0.01';
our $BUFFER_SIZE = 1024;
local $Log::Minimal::LOG_LEVEL = 'DEBUG';

my $opts = +{};
GetOptions(
    $opts,
    'listen|l=s',
    'max-clients|c=s',
    'max-requests-per-child|r=s',
    'timeout|t=s',
    'help|h', 'man',
);

pod2usage(1) if ($opts->{help});
pod2usage(-verbose => 2) if ($opts->{man});

sub handle_connection {
    my ($conn, $opts) = @_;

    while (1) {
        $conn->recv(my $buf, $BUFFER_SIZE);
        if ( length $buf == 0 ) {
            warnf('connection closed by peer (pid: %d)', $$);
            last;
        }

        warnf('buffer: %s (pid: %d)', $buf, $$);

        unless ( $conn->send( $buf ) == length $buf ) {
            warnf('write error (pid: %d)', $$);
        }
    }

    $conn->close;
    warnf('close (pid: %d)', $$);

    return 1;
}

sub accept_loop {
    my ($listen, $opts) = @_;

    my $term_recieved = 0;
    $SIG{TERM} = sub {
        $term_recieved++;
        warnf('signal recieved (pid: %d, term_recieved: %d)', $$, $term_recieved);
        exit 0 if ( $term_recieved > 1 );
    };

    my $requests = 0;

    while ( $requests++ <= $opts->{'max-requests-per-child'} ) {
        if ( $term_recieved ) {
            warnf('exit (pid: %d)', $$);
            exit 0;
        }

        if ( my $conn = $listen->accept ) {
            warnf('accept (pid: %d)', $$);
            handle_connection( $conn, $opts );
        }
    }

    warnf('requests %d (pid: %d)', $requests, $$);
    exit 0;
}


sub main {
    my $opts = shift;

    warnf('pid: %d', $$);

    my $listen = IO::Socket::INET->new(
        Blocking  => 1,
        LocalHost => 'localhost',
        LocalPort => $opts->{listen},
        Listen    => SOMAXCONN,
        Proto     => 'tcp',
        ReuseAddr => 1,
        Timeout   => $opts->{timeout},
        Type      => SOCK_STREAM,
    );

    my $pm = Parallel::Prefork->new({
        max_workers => $opts->{'max-clients'},
        trap_signals => {
            TERM => 'TERM',
            HUP  => 'TERM',
        },
    });

    while ( $pm->signal_received ne 'TERM' ) {
        $pm->start(
            sub {
                accept_loop($listen, $opts);
            }
        );
    }

    $pm->wait_all_children;

    warnf('shutdown server (pid: %d)', $$);
    exit 0;
}

%$opts = (
    listen => 50000,
    timeout => 10,
    'max-clients' => 10,
    'max-requests-per-child' => 100,
    %$opts,
);

main($opts);

select.pl

#!/usr/bin/env perl

use strict;
use warnings;
use FindBin;
use lib "$FindBin::Bin/../lib";
use IO::Socket::INET;
use IO::Select;
use Log::Minimal;
use Getopt::Long;
use Pod::Usage;
use Socket;

our $VERSION = '0.01';
our $BUFFER_SIZE = 1024;
local $Log::Minimal::LOG_LEVEL = 'DEBUG';
local $Log::Minimal::AUTODUMP  = 1;

my $opts = +{};
GetOptions(
    $opts,
    'listen|l=s',
    'timeout|t=s',
    'help|h', 'man',
);

pod2usage(1) if ($opts->{help});
pod2usage(-verbose => 2) if ($opts->{man});

sub handle_connection {
    my ( $s, $conn, $opts ) = @_;

    $conn->recv(my $buf, $BUFFER_SIZE);
    if ( length $buf == 0 ) {
        warnf('connection closed');
        $s->remove($conn);
        $conn->close;
        return 0;
    }

    warnf('buffer: %s', $buf);
    unless ( $conn->send($buf) == length $buf ) {
        warnf('write error');
        $s->remove($conn);
        $conn->close;
        return 0;
    }

    return 1;
}

sub main {
    my $opts = shift;

    warnf('pid: %d', $$);

    my $s = IO::Select->new;
    my $listen = IO::Socket::INET->new(
        Blocking  => 0,
        LocalHost => 'localhost',
        LocalPort => $opts->{listen},
        Listen    => 1,
        Proto     => 'tcp',
        ReuseAddr => 1,
        Timeout   => $opts->{timeout},
        Type      => SOCK_STREAM,
    );
    $listen->blocking(0);

    $s->add($listen);

    my $clients = 0;
    my $term_recieved = 0;

    $SIG{TERM} = sub {
        $term_recieved++;
        warnf('term signal recieved');
    };

    while (1) {
        if ( $term_recieved ) {
            warnf 'shutdown';
            exit 0;
        }

        if ( my @ready = $s->can_read($opts->{timeout}) ) {
            warnf('ready handles (%s)', \@ready);
            for my $sock (@ready) {
                if ( $sock == $listen ) {
                    $clients++;
                    warnf('new connection (clients: %d)', $clients);
                    my $conn = $listen->accept;
                    $conn->blocking(0);
                    $s->add( $conn );
                }
                else {
                    unless ( handle_connection($s, $sock, $opts) ) {
                        $clients--;
                    }
                }
            }
        }

        warnf($s->as_string);
    }
}

%$opts = (
    listen => 50000,
    timeout => 10,
    %$opts,
);

main($opts);