#!/usr/bin/env perl
# SPDX-FileCopyrightText: 2026 GPForum Contributors
# SPDX-License-Identifier: BSD-3-Clause

package main;

use v5.40;

use FindBin qw($Bin);
use lib "$Bin/../lib";

# The event loop GPForum requires (kqueue or epoll) is chosen before anything
# loads EV, for the load generator's own loop.
use GPForum::OS::Reactor;

use Carp qw(croak);
use Const::Fast;
use DBI;
use EV;
use English qw(-no_match_vars);
use Errno   qw(EAGAIN EINPROGRESS EINTR);
use Fcntl   qw(F_GETFL F_SETFL O_NONBLOCK);
use File::Spec;
use File::Temp    qw(tempdir tempfile);
use Getopt::Long  qw(GetOptionsFromArray);
use JSON::MaybeXS qw(decode_json encode_json);
use List::Util    qw(sum0);
use Mojo::UserAgent;
use Mojo::Util qw(url_escape);
use POSIX      qw(WNOHANG _exit setsid);
use Socket     qw(
  AF_INET AF_UNIX IPPROTO_TCP SOCK_STREAM TCP_NODELAY inet_aton
  pack_sockaddr_in pack_sockaddr_un
);
use Time::HiRes qw(clock_gettime CLOCK_MONOTONIC sleep);

use GPForum::Benchmark::Measure qw(percentile);
use GPForum::Benchmark::Process qw(free_port);

our $VERSION = '0.001';

# What the hot pages cost through the listener the service runs, under
# concurrent clients. Each side is a checkout: its own database, migrated
# and seeded with the medium profile by its own bin/gpforum, and its own
# server started as the service starts it -- `bin/gpforum start --service
# --foreground`, production settings, the workers asked for -- listening on
# a Unix socket or on TCP loopback. Members register through /register,
# each from its own forwarded address, are activated in the database and
# sign in through /login, so their sessions are real server-side ones.
#
# Each measurement is one page at one concurrency: that many keep-alive
# connections, as a proxy's upstream connections are, each sending its next
# request when the last is answered, with the headers the shipped nginx site
# adds. The sides alternate, their order reversed every round, and the
# report is each cell's median over the rounds. No proxy runs: the clients
# speak to the listener directly, so the time is the server's.
#
# --connections counts the connections each worker holds in the middle of a
# cell; --burst N opens N more a quarter of the way into it, all at once,
# and reports what they waited on their own, as a proxy's new upstream
# connections wait when the load rises.
const my %DEFAULT => (
    duration => 8,
    warmup   => 2,
    rounds   => 3,
    workers  => 4,
    members  => 400,
    procs    => 2,
);
const my @CONCURRENCY  => ( 1, 8, 32, 128 );
const my $MILLISECONDS => 1_000;
const my $PERCENT      => 100;
const my $P50          => 50;
const my $P95          => 95;
const my $P99          => 99;
const my $USAGE_ERROR  => 2;
const my $READ_LIMIT   => 100_000_000;
const my $READ_MEMBERS => 8;
const my $BUFFER       => 262_144;
const my $READY        => 60;
const my $POLL         => 0.2;
const my $START_DELAY  => 0.2;
const my $RECONNECT    => 0.01;
const my $GRACE        => 30;
const my $STOP_WAIT    => 15;
const my $WRITE_WARMUP => 0.5;
const my $BURST_SHARE  => 0.75;
const my $SECRET_BYTES => 32;
const my $BYTE         => 256;
const my $OCTET        => 256;
const my $HTTP_ERROR   => 400;
const my $HTTP_OK      => 200;
const my $CRLF2        => "\r\n\r\n";
const my %ANSWERED     => map { $_ => 1 } qw(200 302 303);
const my $HOST         => 'forum.bench.test';
const my $PASSWORD     => 'socket bench correct horse staple';
const my $THREAD_ID    => '018f1004-0001-7000-8000-000000000001';
const my $CATEGORY_ID  => '018f1001-0001-7000-8000-000000000001';
const my @REPLY_THREAD => ( 101 .. 116 );
const my $REPLY_BODY => 'body_source=A+socket+bench+reply+{n}%2C+with+a'
  . '+sentence+or+two+of+ordinary+text+in+it.';
const my %HEADERS => (
    'Accept'            => 'text/html',
    'Accept-Encoding'   => 'gzip, deflate, br',
    'X-Forwarded-Proto' => 'https',
);
const my %PAGE => (
    thread_member => { path => "/t/$THREAD_ID",         who => 'member' },
    thread_guest  => { path => "/t/$THREAD_ID",         who => 'guest' },
    home          => { path => q{/},                    who => 'member' },
    category      => { path => "/c/$CATEGORY_ID",       who => 'member' },
    search        => { path => '/search?q=performance', who => 'member' },

    # The form a browser without the page's script posts: answered with a
    # redirect to the new post.
    reply => {
        method => 'POST',
        who    => 'writer',
        body   => "csrf_token={csrf}&command_id={uuid}&$REPLY_BODY",
    },

    # What the page's script posts: the composer's draft fields as well,
    # answered with the new post and a fresh composer.
    reply_htmx => {
        method  => 'POST',
        who     => 'writer',
        headers => { 'HX-Request' => 'true' },
        body    => 'csrf_token={csrf}&command_id={uuid}&draft_id={uuid}'
          . "&draft_revision=0&$REPLY_BODY",
    },
);
const my @PAGES => qw(
  thread_member thread_guest home category search reply reply_htmx
);

exit main(@ARGV);

sub main (@arguments) {
    my $option = _options(@arguments);
    return _usage( $option->{usage} ) if defined $option->{usage};

    my $admin_dsn = $ENV{GPFORUM_DATABASE_DSN}
      // croak 'set GPFORUM_DATABASE_DSN to a server the user may CREATE in';
    my @sides = map { _side( $_, $admin_dsn, $option ) } @{ $option->{side} };
    my ( $rows, $error );
    try {
        for my $side (@sides) {
            _prepare($side);
        }
        $rows = _measure( \@sides, $option );
    }
    catch ($caught) {
        $error = $caught;
    };
    _finish( \@sides, $option );
    croak $error if defined $error;

    print _report( \@sides, $rows, $option ) or croak 'cannot write the report';

    return 0;
}

sub _options (@arguments) {
    my %option = ( %DEFAULT, side => [], page => [], concurrency => [] );
    if (
        !GetOptionsFromArray(
            \@arguments,      \%option,     'side=s@',  'page=s@',
            'concurrency=i@', 'duration=f', 'warmup=f', 'rounds=i',
            'workers=i',      'members=i',  'procs=i',  'json=s',
            'connections',    'keep',       'help',     'burst=i',
        )
      )
    {
        return { usage => $USAGE_ERROR };
    }
    return { usage => 0 } if $option{help};

    if ( !@{ $option{page} } ) {
        @{ $option{page} } = @PAGES;
    }
    if ( !@{ $option{concurrency} } ) {
        @{ $option{concurrency} } = @CONCURRENCY;
    }
    my @unknown = grep { !$PAGE{$_} } @{ $option{page} };
    if (   @arguments
        || @unknown
        || !@{ $option{side} }
        || $option{rounds} < 1
        || $option{members} < $READ_MEMBERS )
    {
        return { usage => $USAGE_ERROR };
    }

    return \%option;
}

sub _usage ($status) {
    my $pages = join q{ }, @PAGES;
    print <<"USAGE" or croak 'cannot write the usage';
Usage: script/bench-socket --side LABEL=CHECKOUT[,tcp][,poll]...
       [--page NAME]... [--concurrency N]... [--duration S] [--warmup S]
       [--rounds N] [--workers N] [--members N] [--procs N] [--connections]
       [--burst N] [--json FILE] [--keep]
       Pages: $pages
       A side listens on a Unix socket and reaches PostgreSQL through its
       socket; tcp makes both TCP loopback (a checkout from before
       2026-10-16), poll runs Mojo::Reactor::Poll (one from before EV was
       required). GPFORUM_DATABASE_DSN names a server, by its socket
       directory, the user may CREATE DATABASE on; for a tcp side it must
       listen on 127.0.0.1 too.
USAGE

    return $status;
}

# A side as --side gives it: a label, a checkout, and how it is reached.
sub _side ( $spelling, $admin_dsn, $option ) {
    my ( $label, $rest ) = split /=/msx, $spelling, 2;
    croak "--side $spelling: LABEL=CHECKOUT[,tcp][,poll]"
      if !defined $rest || !length $label;
    my ( $checkout, @flags ) = split /,/msx, $rest;
    croak "--side $spelling: $checkout has no bin/gpforum"
      if !-x "$checkout/bin/gpforum";
    my %flag = map { $_ => 1 } @flags;
    my $name = "gpforum_bench_socket_${label}_$PROCESS_ID";
    my $dsn  = _side_dsn( $admin_dsn, $name, $flag{tcp} );
    my $run  = tempdir( 'gpforum-bench-socket-XXXXXX', TMPDIR => 1 );

    return {
        label    => $label,
        checkout => File::Spec->rel2abs($checkout),
        tcp      => $flag{tcp}  ? 1 : 0,
        poll     => $flag{poll} ? 1 : 0,
        database => $name,
        dsn      => $dsn,
        admin    => $admin_dsn,
        run      => $run,
        workers  => $option->{workers},
        members  => $option->{members},
        target   => $flag{tcp}
        ? 'tcp:127.0.0.1:' . free_port()
        : "unix:$run/gpforum.sock",
    };
}

# The side's database on the same server: through the socket the admin DSN
# names, or for a tcp side through 127.0.0.1 on the same port.
sub _side_dsn ( $admin_dsn, $name, $tcp ) {
    ( my $dsn = $admin_dsn ) =~ s/dbname=[^;]+/dbname=$name/msx;
    if ($tcp) {
        $dsn =~ s/host=[^;]+/host=127.0.0.1/msx;
    }

    return $dsn;
}

sub _connect ($dsn) {
    return DBI->connect(
        $dsn,
        $ENV{GPFORUM_DATABASE_USER},
        $ENV{GPFORUM_DATABASE_PASSWORD},
        { AutoCommit => 1, PrintError => 0, RaiseError => 1 },
    );
}

# The database, migrated and seeded by the side's own commands; the server;
# its members.
sub _prepare ($side) {
    my $admin = _connect( $side->{admin} );
    $admin->do(
        'CREATE DATABASE ' . $admin->quote_identifier( $side->{database} ) );
    $side->{created} = 1;
    $admin->disconnect;

    my %database = (
        GPFORUM_DATABASE_DSN => $side->{dsn},
        LC_ALL               => 'en_US.UTF-8',
    );
    for my $command ( [ 'migrate', '--quiet' ],
        [ 'performance-seed', '--profile', 'medium' ] )
    {
        _run_quietly( $side, \%database, @{$command} );
    }
    _start($side);
    $side->{members_signed_in} = _sign_in($side);

    return;
}

sub _run_quietly ( $side, $environment, @command ) {
    my $pid = fork // croak "cannot fork: $OS_ERROR";
    if ( !$pid ) {
        local %ENV = ( %ENV, %{$environment} );
        chdir $side->{checkout} or _exit(1);
        open STDOUT, '>>', "$side->{run}/setup.log" or _exit(1);
        open STDERR, '>&', \*STDOUT                 or _exit(1);
        exec {"$side->{checkout}/bin/gpforum"} "$side->{checkout}/bin/gpforum",
          @command
          or _exit(1);
    }
    waitpid $pid, 0;
    croak "$side->{label}: gpforum @command failed; see $side->{run}/setup.log"
      if $CHILD_ERROR;

    return;
}

# The service as its unit starts it, with production settings a laptop can
# give: mail through sendmail (the outbox worker that would send it does not
# run), no antivirus (nothing is uploaded), the read limit raised.
sub _service_environment ($side) {
    my %environment = (
        LC_ALL                        => 'en_US.UTF-8',
        GPFORUM_ENV                   => 'production',
        GPFORUM_WARMUP_ENABLED        => 1,
        GPFORUM_PUBLIC_BASE_URL       => "https://$HOST",
        GPFORUM_SESSION_SECRET        => _random_hex(),
        GPFORUM_METRICS_TOKEN         => _random_hex(),
        GPFORUM_DATABASE_DSN          => $side->{dsn},
        GPFORUM_MAIL_TRANSPORT        => 'sendmail',
        GPFORUM_MAIL_FROM             => "forum\@$HOST",
        GPFORUM_ANTIVIRUS             => 'none',
        GPFORUM_LOG_PATH              => "$side->{run}/gpforum.log",
        GPFORUM_WEB_PROCESSES         => $side->{workers},
        GPFORUM_RUNTIME_WORKER_POLICY => 'configured',
        GPFORUM_RUNTIME_PID_FILE      => "$side->{run}/hypnotoad.pid",
        GPFORUM_RUNTIME_LISTEN        => _listen($side),
        GPFORUM_FORUM_READ_RATE_LIMIT => $READ_LIMIT,
    );
    if ( $side->{poll} ) {
        $environment{MOJO_REACTOR} = 'Mojo::Reactor::Poll';
    }

    return \%environment;
}

sub _listen ($side) {
    my ( $kind, $where ) = split /:/msx, $side->{target}, 2;
    return "http://$where" if $kind eq 'tcp';

    return 'http+unix://' . url_escape($where);
}

sub _random_hex {
    return join q{}, map { sprintf '%02x', int rand $BYTE } 1 .. $SECRET_BYTES;
}

sub _start ($side) {
    my $environment = _service_environment($side);
    my $pid         = fork // croak "cannot fork: $OS_ERROR";
    if ( !$pid ) {
        setsid();
        local %ENV = ( %ENV, %{$environment} );
        if ( !$side->{poll} ) {
            delete $ENV{MOJO_REACTOR};
        }
        chdir $side->{checkout} or _exit(1);
        open STDOUT, '>',  "$side->{run}/service.log" or _exit(1);
        open STDERR, '>&', \*STDOUT                   or _exit(1);
        exec {"$side->{checkout}/bin/gpforum"} "$side->{checkout}/bin/gpforum",
          qw(start --service --foreground)
          or _exit(1);
    }
    $side->{pid} = $pid;
    _await_ready($side);

    return;
}

sub _await_ready ($side) {
    my $deadline = clock_gettime(CLOCK_MONOTONIC) + $READY;
    while ( clock_gettime(CLOCK_MONOTONIC) < $deadline ) {
        croak "$side->{label}: the service stopped; see $side->{run}"
          if waitpid( $side->{pid}, WNOHANG ) == $side->{pid};
        my $tx = _agent()->get( _base($side) . '/health/live' => _proxy() );
        return if ( $tx->res->code // 0 ) == $HTTP_OK;
        sleep $POLL;
    }
    croak "$side->{label}: the service did not answer in $READY s";
}

sub _agent {
    state $ua = do {
        my $agent = Mojo::UserAgent->new(
            max_redirects   => 0,
            request_timeout => $GRACE,
        );
        $agent->cookie_jar->ignore( sub { return 1; } );
        $agent;
    };

    return $ua;
}

sub _base ($side) {
    my ( $kind, $where ) = split /:/msx, $side->{target}, 2;
    return "http://$where" if $kind eq 'tcp';

    return 'http+unix://' . url_escape($where);
}

sub _proxy ( $address = '10.9.0.1' ) {
    return {
        %HEADERS,
        Host              => $HOST,
        'X-Forwarded-For' => $address,
        'X-Real-IP'       => $address,
    };
}

# The members: registered, activated in the database,
# signed in, each with the session's CSRF token read from a page.
sub _sign_in ($side) {
    my $dbh = _connect( $side->{dsn} );
    my @members;
    for my $number ( 1 .. $side->{members} ) {
        push @members, _member( $side, $dbh, $number );
    }
    $dbh->disconnect;

    return \@members;
}

sub _member ( $side, $dbh, $number ) {
    my %member = (
        name    => "socket_bench_$number",
        address => sprintf( '10.%d.%d.%d',
            1 + int( $number / ( $OCTET * $OCTET ) ),
            int( $number / $OCTET ) % $OCTET,
            $number % $OCTET ),
        jar => {},
    );
    my $form = _visit( $side, \%member, GET => '/register' );
    _visit(
        $side,
        \%member,
        POST => '/register',
        {
            command_id   => _field( $form, 'command_id' ),
            csrf_token   => _field( $form, 'csrf_token' ),
            username     => $member{name},
            display_name => "Socket Bench $number",
            email        => "$member{name}\@example.test",
            password     => $PASSWORD,
        }
    );
    $dbh->do(
        q{UPDATE users SET status = 'active', email_verified_at = now()}
          . ' WHERE username = ?',
        undef, $member{name}
    );
    my $login = _visit( $side, \%member, GET => '/login' );
    _visit(
        $side,
        \%member,
        POST => '/login',
        {
            command_id => _field( $login, 'command_id' ),
            csrf_token => _field( $login, 'csrf_token' ),
            identifier => $member{name},
            password   => $PASSWORD,
        }
    );
    my $home = _visit( $side, \%member, GET => q{/} );
    croak "$side->{label}: $member{name} could not sign in"
      if !$home->res->dom->at('form[action="/logout"]');

    return {
        address => $member{address},
        csrf    => _field( $home, 'csrf_token' ),
        cookie  => join q{; },
        map { "$_=$member{jar}{$_}" } sort keys %{ $member{jar} },
    };
}

sub _visit ( $side, $member, $method, $path, $form = undef ) {
    my %headers = %{ _proxy( $member->{address} ) };
    if ( %{ $member->{jar} } ) {
        $headers{Cookie} = join q{; },
          map { "$_=$member->{jar}{$_}" } sort keys %{ $member->{jar} };
    }
    my $url = _base($side) . $path;
    my $tx =
      $form
      ? _agent()->build_tx( $method => $url => \%headers => form => $form )
      : _agent()->build_tx( $method => $url => \%headers );
    $tx = _agent()->start($tx);
    my $code = $tx->res->code // 0;
    croak "$side->{label}: $method $path answered $code"
      if !$code || $code >= $HTTP_ERROR;
    for my $cookie ( @{ $tx->res->cookies } ) {
        $member->{jar}{ $cookie->name } = $cookie->value;
    }

    return $tx;
}

sub _field ( $tx, $name ) {
    my $input = $tx->res->dom->at(qq{input[name="$name"]})
      // croak "no $name field on " . $tx->req->url;

    return $input->attr('value');
}

# Every cell on every side, the sides' order reversed each round.
sub _measure ( $sides, $option ) {
    my @rows;
    for my $round ( 1 .. $option->{rounds} ) {
        my @order = $round % 2 ? @{$sides} : reverse @{$sides};
        for my $page ( @{ $option->{page} } ) {
            for my $concurrency ( @{ $option->{concurrency} } ) {
                for my $side (@order) {
                    my $row = _cell( $side, $page, $concurrency, $option );
                    $row->{round} = $round;
                    push @rows, $row;
                    _log_row( $row, $option );
                }
            }
        }
    }

    return \@rows;
}

sub _log_row ( $row, $option ) {
    printf {*STDERR} 'r%d %-14s c=%-4d %-10s %8.1f req/s p50 %8.2f p95 %8.2f'
      . " p99 %8.2f%s%s\n",
      @{$row}{qw(round page concurrency side rps p50_ms p95_ms p99_ms)},
      ( $row->{connections} ? " connections $row->{connections}" : q{} ),
      (
        $row->{burst}
        ? sprintf( q{ burst p50 %.2f p95 %.2f p99 %.2f},
            @{ $row->{burst} }{qw(p50_ms p95_ms p99_ms)} )
        : q{}
      ) or croak q{cannot write the progress};
    return if !defined $option->{json};

    open my $handle, '>>', $option->{json}
      or croak "cannot append to $option->{json}: $OS_ERROR";
    print {$handle} encode_json($row) . "\n"
      or croak "cannot write $option->{json}";
    close $handle or croak "cannot close $option->{json}";

    return;
}

sub _cell ( $side, $page, $concurrency, $option ) {
    my $definition = { %{ $PAGE{$page} } };
    my $writer     = $definition->{who} eq 'writer';
    if ($writer) {

        # The bench's own counters: each member may write 20 times a minute.
        my $dbh = _connect( $side->{dsn} );
        $dbh->do('DELETE FROM rate_limit_buckets');
        $dbh->disconnect;
    }
    my %load = (
        target      => $side->{target},
        request     => _request_template( $side, $definition ),
        concurrency => $concurrency,
        duration    => $writer ? $option->{duration} / 2 : $option->{duration},
        warmup      => $writer ? $WRITE_WARMUP           : $option->{warmup},
        procs       => $concurrency >= $option->{procs} ? $option->{procs} : 1,
        burst       => $writer                          ? 0 : $option->{burst},
    );
    my $census = $option->{connections} ? _census_start($side) : undef;
    my $result = _load( \%load, $census );

    return {
        %{$result},
        side        => $side->{label},
        page        => $page,
        concurrency => $concurrency,
    };
}

sub _request_template ( $side, $definition ) {
    my @members = @{ $side->{members_signed_in} };
    my $who     = $definition->{who};
    my $paths =
      $definition->{path}
      ? [ $definition->{path} ]
      : [ map { sprintf '/t/018f1004-%04x-7000-8000-%012x/replies', $_, $_ }
          @REPLY_THREAD ];
    my %headers = ( %HEADERS, %{ $definition->{headers} // {} } );

    return {
        method => $definition->{method} // 'GET',
        paths  => $paths,
        head   => join( q{},
            "Host: $HOST\r\n",
            map { "$_: $headers{$_}\r\n" } sort keys %headers ),
        body => $definition->{body},
        members => $who eq 'guest' ? [ { address => '10.9.0.1' } ]
        : $who eq 'member' ? [ @members[ 0 .. $READ_MEMBERS - 1 ] ]
        :                    \@members,
    };
}

# The load: its processes each run their share of the connections and write
# what they measured to a file, merged here.
sub _load ( $load, $census ) {
    my $procs = $load->{procs};
    my $start = clock_gettime(CLOCK_MONOTONIC) + $START_DELAY;
    my @children;
    for my $index ( 0 .. $procs - 1 ) {
        my $share = int( $load->{concurrency} / $procs ) +
          ( $index < $load->{concurrency} % $procs ? 1 : 0 );
        push @children,
          _spawn_client( $load,
            { index => $index, share => $share, start => $start } );
    }
    my $burst = _spawn_burst( $load, $start );
    my $connections;
    if ($census) {
        sleep $START_DELAY + $load->{warmup} + $load->{duration} / 2;
        $connections = _census_count($census);
    }
    my $merged = _merge( $load, \@children, $connections );
    if ($burst) {
        $merged->{burst} = _merge( $burst->{load}, [ $burst->{child} ], undef );
    }

    return $merged;
}

# --burst N: N more connections opened at once a quarter of the way into
# the measured time, next to the busy ones, and measured on their own until
# the end: a proxy opening upstream connections when the load rises.
sub _spawn_burst ( $load, $start ) {
    return undef if !$load->{burst};

    my %burst = (
        %{$load},
        warmup   => 0,
        duration => $load->{duration} * $BURST_SHARE,
    );
    my $opens = $start + $load->{warmup} + $load->{duration} - $burst{duration};

    return {
        load  => \%burst,
        child => _spawn_client(
            \%burst,
            {
                index => $load->{procs},
                share => $load->{burst},
                start => $opens,
            }
        ),
    };
}

sub _spawn_client ( $load, $part ) {
    my ( $handle, $file ) =
      tempfile( 'gpforum-bench-socket-XXXXXX', TMPDIR => 1 );
    close $handle or croak "cannot close $file";
    my $pid = fork // croak "cannot fork: $OS_ERROR";
    if ( !$pid ) {
        my $measured = _client( $load, $part );
        open my $out, '>', $file or _exit(1);
        print {$out} encode_json($measured) or _exit(1);
        close $out                          or _exit(1);
        _exit(0);
    }

    return { pid => $pid, file => $file };
}

sub _merge ( $load, $children, $connections ) {
    my ( @latency, %status );
    my ( $errors,  $bytes ) = ( 0, 0 );
    for my $child ( @{$children} ) {
        waitpid $child->{pid}, 0;
        my $measured = decode_json( _slurp( $child->{file} ) );
        unlink $child->{file};
        push @latency, @{ $measured->{latency} };
        for my $code ( keys %{ $measured->{status} } ) {
            $status{$code} += $measured->{status}{$code};
        }
        $errors += $measured->{errors};
        $bytes  += $measured->{bytes};
    }
    @latency = sort { $a <=> $b } @latency;

    return {
        rps      => @latency / $load->{duration},
        p50_ms   => _ms( percentile( \@latency, $P50 ) ),
        p95_ms   => _ms( percentile( \@latency, $P95 ) ),
        p99_ms   => _ms( percentile( \@latency, $P99 ) ),
        requests => scalar @latency,
        errors   => $errors,
        status   => \%status,
        bytes    => @latency ? int( $bytes / @latency ) : 0,
        ( defined $connections ? ( connections => $connections ) : () ),
    };
}

sub _slurp ($file) {
    open my $handle, '<', $file or croak "a load process failed: $file";
    local $INPUT_RECORD_SEPARATOR = undef;
    my $text = <$handle>;
    close $handle or croak "cannot close $file";

    return $text;
}

sub _ms ($seconds) {
    return 0 + sprintf '%.2f', $seconds * $MILLISECONDS;
}

# One load process: its connections, each sending its next request when
# the last is answered, from the start for warm-up plus duration seconds;
# what was answered after the warm-up is measured.
sub _client ( $load, $part ) {

    # The kernel queue the sign-in's requests opened in the parent is not
    # this process's: libev makes its own again before the loop runs. Its
    # clock is the one it read when the parent's loop last ran, minutes ago
    # by now, and the timers below would count from then.
    EV::default_loop()->loop_fork;
    EV::now_update();
    srand( $PROCESS_ID ^ time );
    my $state = {
        load     => $load,
        index    => $part->{index},
        member   => $part->{index},
        sequence => 0,
        measured => $part->{start} + $load->{warmup},
        ends     => $part->{start} + $load->{warmup} + $load->{duration},
        latency  => [],
        status   => {},
        errors   => 0,
        bytes    => 0,
        open     => $part->{share},
    };
    my @connections = map { {} } 1 .. $part->{share};
    my $go          = EV::timer(
        _until( $part->{start} ),
        0,
        sub {
            for my $connection (@connections) {
                _send( $state, $connection );
            }
        }
    );
    my $guard =
      EV::timer( _until( $state->{ends} ) + $GRACE, 0, sub { EV::break(); } );
    EV::run();

    return {
        latency => $state->{latency},
        status  => $state->{status},
        errors  => $state->{errors},
        bytes   => $state->{bytes},
    };
}

sub _until ($when) {
    my $delay = $when - clock_gettime(CLOCK_MONOTONIC);
    return $delay > 0 ? $delay : 0;
}

sub _send ( $state, $connection ) {
    my $now = clock_gettime(CLOCK_MONOTONIC);
    if ( $now >= $state->{ends} ) {
        _close($connection);
        if ( !--$state->{open} ) {
            EV::break();
        }
        return;
    }
    if ( !$connection->{handle} && !_open( $state, $connection ) ) {
        $connection->{retry} =
          EV::timer( $RECONNECT, 0, sub { _send( $state, $connection ); } );
        return;
    }
    $connection->{out}     = _next_request($state);
    $connection->{started} = $now;
    $connection->{in}      = q{};
    $connection->{watcher} = EV::io( $connection->{handle}, EV::WRITE,
        sub { _write( $state, $connection ); } );

    return;
}

sub _open ( $state, $connection ) {
    my ( $kind, $where ) = split /:/msx, $state->{load}{target}, 2;
    my $handle;
    if ( $kind eq 'unix' ) {
        socket $handle, AF_UNIX, SOCK_STREAM, 0 or croak "socket: $OS_ERROR";
    }
    else {
        socket $handle, AF_INET, SOCK_STREAM, 0 or croak "socket: $OS_ERROR";
        setsockopt $handle, IPPROTO_TCP, TCP_NODELAY, 1
          or croak "setsockopt: $OS_ERROR";
    }
    my $flags = fcntl $handle, F_GETFL, 0 or croak "fcntl: $OS_ERROR";
    fcntl $handle, F_SETFL, $flags | O_NONBLOCK or croak "fcntl: $OS_ERROR";
    my $address =
      $kind eq 'unix'
      ? pack_sockaddr_un($where)
      : pack_sockaddr_in( ( split /:/msx, $where )[1], inet_aton('127.0.0.1') );
    if ( !connect $handle, $address ) {
        if ( $OS_ERROR != EINPROGRESS && $OS_ERROR != EAGAIN ) {
            ++$state->{errors};
            return 0;
        }
    }
    $connection->{handle} = $handle;

    return 1;
}

sub _next_request ($state) {
    my $template = $state->{load}{request};
    my $members  = $template->{members};
    my $member   = $members->[ $state->{member} % @{$members} ];
    $state->{member} += $state->{load}{procs};
    my $sequence = ++$state->{sequence};
    my $paths    = $template->{paths};
    my $path = $paths->[ ( $sequence * $state->{load}{procs} + $state->{index} )
      % @{$paths} ];
    my $address = $member->{address};

    my $request =
        "$template->{method} $path HTTP/1.1\r\n$template->{head}"
      . "X-Forwarded-For: $address\r\nX-Real-IP: $address\r\n"
      . ( defined $member->{cookie} ? "Cookie: $member->{cookie}\r\n" : q{} );
    return "$request\r\n" if !defined $template->{body};

    my $body = _body( $template->{body}, $member, "$PROCESS_ID-$sequence" );
    return
        $request
      . "Content-Type: application/x-www-form-urlencoded\r\n"
      . 'Content-Length: '
      . length($body)
      . "\r\n\r\n$body";
}

sub _body ( $template, $member, $number ) {
    my $body  = $template;
    my $token = url_escape( $member->{csrf} // q{} );
    $body =~ s/[{]csrf[}]/$token/gmsx;
    $body =~ s/[{]uuid[}]/_uuid()/egmsx;
    $body =~ s/[{]n[}]/$number/gmsx;

    return $body;
}

sub _uuid {
    my $hex = join q{}, map { sprintf '%02x', int rand $BYTE } 1 .. 16;    ## no critic (ValuesAndExpressions::ProhibitMagicNumbers) -- a uuid's sixteen bytes
    substr $hex, 12, 1, '4';                                               ## no critic (ValuesAndExpressions::ProhibitMagicNumbers) -- the version digit
    return join q{-}, unpack 'A8 A4 A4 A4 A12', $hex;
}

sub _write ( $state, $connection ) {
    my $written = syswrite $connection->{handle}, $connection->{out};
    if ( !defined $written ) {
        return if $OS_ERROR == EAGAIN || $OS_ERROR == EINTR;
        return _failed( $state, $connection );
    }
    substr $connection->{out}, 0, $written, q{};
    return if length $connection->{out};

    $connection->{watcher} = EV::io( $connection->{handle}, EV::READ,
        sub { _read( $state, $connection ); } );

    return;
}

sub _failed ( $state, $connection ) {
    ++$state->{errors};
    _close($connection);
    _send( $state, $connection );

    return;
}

sub _close ($connection) {
    undef $connection->{watcher};
    if ( $connection->{handle} ) {
        close $connection->{handle} or undef;
    }
    undef $connection->{handle};

    return;
}

sub _read ( $state, $connection ) {
    my $read = sysread $connection->{handle}, $connection->{in}, $BUFFER,
      length $connection->{in};
    if ( !defined $read ) {
        return if $OS_ERROR == EAGAIN || $OS_ERROR == EINTR;
        return _failed( $state, $connection );
    }
    return _failed( $state, $connection ) if !$read;

    my $response = _response( $connection->{in} ) // return;
    my $now      = clock_gettime(CLOCK_MONOTONIC);
    if (   $connection->{started} >= $state->{measured}
        && $now <= $state->{ends} )
    {
        push @{ $state->{latency} }, $now - $connection->{started};
        ++$state->{status}{ $response->{code} };
        $state->{bytes} += $response->{body};
    }
    if ( $response->{close} ) {
        _close($connection);
    }
    _send( $state, $connection );

    return;
}

# The response in a buffer, once it is whole: its status, its body's length,
# and whether the server closes the connection after it.
sub _response ($buffer) {
    my $end = index $buffer, $CRLF2;
    return undef if $end < 0;

    my $head    = substr $buffer, 0, $end;
    my ($code)  = $head =~ m{\A HTTP/1[.][01] [ ] (\d{3})}msx;
    my $closes  = $head =~ /^Connection: \s* close/imsx ? 1 : 0;
    my $body_at = $end + length $CRLF2;
    if ( $head =~ /^Content-Length: \s* (\d+)/imsx ) {
        return undef if length $buffer < $body_at + $1;
        return { code => $code, body => $1, close => $closes };
    }
    if ( $head =~ /^Transfer-Encoding: \s* chunked/imsx ) {
        return undef if $buffer !~ /\r\n0\r\n\r\n\z/msx;
        return {
            code  => $code,
            body  => length($buffer) - $body_at,
            close => $closes
        };
    }

    return { code => $code, body => length($buffer) - $body_at, close => 1 };
}

# The connections each worker holds, before a cell and in its middle: the
# difference is the cell's, by worker. It needs pgrep and lsof.
sub _census_start ($side) {
    return { side => $side, before => _worker_sockets($side) };
}

sub _census_count ($census) {
    my $now    = _worker_sockets( $census->{side} );
    my $before = $census->{before};

    return join q{/}, reverse sort { $a <=> $b }
      map { $now->{$_} - ( $before->{$_} // 0 ) } keys %{$now};
}

sub _worker_sockets ($side) {
    open my $file, '<', "$side->{run}/hypnotoad.pid"
      or croak "no pid file in $side->{run}";
    my $manager = <$file>;
    close $file or croak 'cannot close the pid file';
    chomp $manager;
    my %sockets;
    for my $worker ( _command_lines( 'pgrep', '-P', $manager ) ) {
        my @open =
          $side->{tcp}
          ? _command_lines( 'lsof', '-nP', '-a', '-p', $worker, '-iTCP',
            '-sTCP:ESTABLISHED' )
          : _command_lines( 'lsof', '-nP', '-a', '-p', $worker, '-U' );
        $sockets{$worker} = @open ? @open - 1 : 0;
    }

    return \%sockets;
}

sub _command_lines (@command) {
    open my $output, q{-|}, @command
      or croak "cannot run $command[0]: $OS_ERROR";
    my @lines = <$output>;
    chomp @lines;
    close $output or undef;

    return @lines;
}

sub _finish ( $sides, $option ) {
    for my $side ( @{$sides} ) {
        _stop($side);
        next if !$side->{created} || $option->{keep};

        my $admin = _connect( $side->{admin} );
        $admin->do( 'DROP DATABASE IF EXISTS '
              . $admin->quote_identifier( $side->{database} )
              . ' WITH (FORCE)' );
        $admin->disconnect;
    }

    return;
}

# The manager in the foreground stops its workers on TERM; anything still
# running after that is killed with its session.
sub _stop ($side) {
    return if !$side->{pid};

    kill 'TERM', $side->{pid};
    my $deadline = clock_gettime(CLOCK_MONOTONIC) + $STOP_WAIT;
    while ( clock_gettime(CLOCK_MONOTONIC) < $deadline ) {
        return if waitpid( $side->{pid}, WNOHANG ) == $side->{pid};
        sleep $POLL;
    }
    kill 'KILL', -$side->{pid};
    waitpid $side->{pid}, 0;

    return;
}

# Each cell's median over the rounds, and the change from the first side.
sub _report ( $sides, $rows, $option ) {
    my %cell;
    for my $row ( @{$rows} ) {
        push @{ $cell{ $row->{page} }{ $row->{concurrency} }{ $row->{side} } },
          $row;
    }
    my $first = $sides->[0]{label};
    my $text =
      sprintf q{socket bench: %d workers, %d rounds of %s s after %s s}
      . " warm-up, medians\n",
      @{$option}{qw(workers rounds duration warmup)};
    $text .= sprintf "%-14s %5s %-10s %9s %8s %8s %8s %7s %7s\n", 'page', 'c',
      'side', 'req/s', 'p50 ms', 'p95 ms', 'p99 ms', 'req/s', 'errors';
    for my $page ( @{ $option->{page} } ) {
        for my $concurrency ( @{ $option->{concurrency} } ) {
            my $base = _median_row( $cell{$page}{$concurrency}{$first} );
            for my $side ( @{$sides} ) {
                my $rows   = $cell{$page}{$concurrency}{ $side->{label} };
                my $median = _median_row($rows);
                $text .=
                  sprintf "%-14s %5d %-10s %9.1f %8.2f %8.2f %8.2f %7s %7d\n",
                  $page, $concurrency, $side->{label},
                  @{$median}{qw(rps p50_ms p95_ms p99_ms)},
                  _change( $base->{rps}, $median->{rps} ), $median->{errors};
                next if !$option->{burst};

                my $burst =
                  _median_row( [ map { $_->{burst} // () } @{ $rows // [] } ] );
                $text .= sprintf "%-14s %5s %-10s %9.1f %8.2f %8.2f %8.2f\n",
                  q{}, "+$option->{burst}", $side->{label},
                  @{$burst}{qw(rps p50_ms p95_ms p99_ms)};
            }
        }
    }

    return $text;
}

sub _median_row ($rows) {
    my %median;
    for my $key (qw(rps p50_ms p95_ms p99_ms)) {
        my @values = sort { $a <=> $b } map { $_->{$key} } @{ $rows // [] };
        $median{$key} = @values ? percentile( \@values, $P50 ) : 0;
    }
    $median{errors} = sum0 map { $_->{errors} + _refused($_) } @{ $rows // [] };

    return \%median;
}

# Answers that are neither a page nor a redirect, as the errors' column
# counts them.
sub _refused ($row) {
    return sum0 map { $row->{status}{$_} }
      grep { !exists $ANSWERED{$_} } keys %{ $row->{status} };
}

sub _change ( $base, $value ) {
    return q{} if !$base;

    return sprintf '%+.0f%%', $PERCENT * ( $value / $base - 1 );
}