stress test concurrent connections

server.pl

#!/usr/bin/env perl

use strict;
use warnings;
use feature ':5.10';

use POE;
use HTTP::Engine;

my $body = "0"x1_000_000;

my $mod = 'POE'; #'ServerSimple';

my $total = 0;

HTTP::Engine->new(
    interface => {
        module => $mod,
        args   => {
            host => 'localhost',
            port =>  3000,
        },
        request_handler => sub {
            my $req = shift;
            $| = 1;
            $total++;
            print {*STDERR} "($total)[". $req->uri->query. ']';
            my $res = HTTP::Engine::Response->new;
            $res->body($body);
            return $res;
        },
    },
)->run;

POE::Kernel->run;

slow-wget.pl

#!/usr/bin/env perl

use strict;
use warnings;
use feature ':5.10';

use String::TT qw/tt strip/;
use AnyEvent;
use AnyEvent::Socket;
use URI;
use AnyEvent::Util;

use EV;
use Set::Object;
use Event::Join;

my $url = shift || 'http://localhost:3000/';
my $id_pool = 0;
my $req = Set::Object->new;
my $res = Set::Object->new;
$| = 1;

my $connections = 1000;

my $all_start_time = AnyEvent->now;
my $all_done = AnyEvent->condvar;
my $join = Event::Join->new(
    events => [1..$connections],
    on_completion => sub { $all_done->send },
);

for my $id (1..$connections){
    my $start_time = AnyEvent->now;
    my $c = get(URI->new($url), 20_000); # bytes per second per connection
    $c->cb( sub {
        my $c = shift;
        my $data = $c->recv;
        $req->delete( $c );
        $res->insert( {
            id   => $id,
            data => $data,
            time => AnyEvent->now - $start_time,
        });
        $join->send_event($id);
    });
    $req->insert($c);
}

print "Waiting...\n";
$all_done->wait;
print "done waiting.\n";

my $all_end_time = AnyEvent->now;

my $ok = 0;
my $error = 0;
my $bytes = 0;
my $seconds = 0;
for my $member ($res->members){
    $ok++ if defined $member;
    $error++ if !defined $member;
    if(defined $member){
        $seconds += $member->{time};
        $bytes   += length ($member->{data});
    }
}

print "ok: $ok, errors: $error\n";
print "$bytes bytes in $seconds seconds: ". $bytes/$seconds. " bps\n";
print $all_end_time - $all_start_time, " wall clock seconds\n";
print '('. $bytes/($all_end_time - $all_start_time). ' bps with concurrency)';
print "\n";

sub get {
    my $url = shift;
    my $rate = shift;
    my $id = $id_pool++;

    my $done = AnyEvent->condvar;
    my $response = '';

    my $handle_input; $handle_input = sub {
        my $w  = shift;
        my $fh = shift;
        die $fh, @_;
    };

    my $connected = AnyEvent->condvar;
    $connected->cb(sub {
        my $connected = shift;
        my $fh = $connected->recv;

        my $start_time = AnyEvent->now;
        my $rw;
        my $tw; $tw = AnyEvent->timer( after => 0, interval => 0.1, cb => sub {
            return if $rw;
            $rw = AnyEvent->io( fh => $fh, poll => 'r', cb => sub {

                my $target_bytes = $rate * (AnyEvent->now - $start_time);
                my $have_bytes = length $response;
                my $need = int ($target_bytes - $have_bytes);
                # warn "target: $target_bytes, length: $have_bytes, need: $need";
                return unless $need > 0;

                my $bytes = sysread($fh, my $buf, $need);
                if(!defined $bytes){
                    undef $rw; # try again later
                    print "[-$id]";
                }
                elsif($bytes == 0){
                    undef $rw;
                    undef $tw;
                    $done->send($response);
                    print "[+$id]";
                }
                else {
                    $response .= $buf;
                    print "[$id]";
                    #$buf =~ s/0+/.../g;
                    #print $buf;
                }
                $rw = 0;
            });
        });


        my $ww; $ww = AnyEvent->io( fh => $fh, poll => 'w', cb => sub {
            my $u = $url; # String::TT bug
            my $i = $id;
            my $data = tt strip q{
                GET [% u.path %]?[% i %] HTTP/1.1
                Host: [% u.host %]


            };
            $data =~ s/\n/\r\n/g;
            my $written = syswrite($fh, $data);
            undef $ww;
        });


    });

    tcp_connect $url->host, $url->port, sub {
        my ($fh) = @_
          or die "connect failed: $!";
        print '!';
        fh_nonblocking $fh, 1;
        $connected->send($fh);
    };

    return $done;
}