view release on metacpan or  search on metacpan

benchmarking/bench-streamd.pl  view on Meta::CPAN

# Reference numbers (2026-07-08, 8-core dev box, C backend,
# Cpanel::JSON::XS, 100 trees, window 2048, eta 32, 5 features):
# in-process score_learn ~2,770 pts/s; over the socket ~2,700 pts/s at
# any batch >= 16 (~2% overhead) and ~2,500 pts/s even at batch 1;
# score mode ~29,000 pts/s (no learning -- the tree walk is cheap, the
# learn is what costs); tagged objects ~21,000 vs positional ~30,000 in
# score mode; throughput is flat across 1/2/4 concurrent clients (one
# shared model, one loop -- by design); ping ~29,000 round trips/s;
# save ~140 ms.  The wire is nowhere near the bottleneck -- the model
# is.
#
# Run with:
#   perl -Ilib benchmarking/bench-streamd.pl

use strict;
use warnings;
use FindBin;
use lib "$FindBin::Bin/../lib";
use lib "$FindBin::Bin";
use BenchAccel                                     qw(wall_rate wall_time_median);
use Time::HiRes                                    qw(time);
use File::Temp                                     qw(tempdir);
use IO::Select                                     ();
use IO::Socket::UNIX                               ();
use POSIX                                          ();
use Algorithm::Classifier::IsolationForest::Online ();

eval { require JSON::MaybeXS; 1 }
	or die "this benchmark needs JSON::MaybeXS (streamd's wire protocol): $@";
my $JSON = JSON::MaybeXS->new( utf8 => 1 );

use constant PI => 3.14159265358979;

sub gaussian {
	my ( $mu, $sigma ) = @_;
	return $mu + $sigma * sqrt( -2 * log( rand() || 1e-12 ) ) * cos( 2 * PI * rand() );
}

sub make_data {
	my ( $n, $nf ) = @_;
	return [
		map {
			[ map { gaussian( 0, 1 ) } 1 .. $nf ]
		} 1 .. $n
	];
}

# --- model shape (matches bench-online-score-accel.pl) -------------------
my $N_TREES = 100;
my $WINDOW  = 2048;
my $ETA     = 32;
my $NF      = 5;
my @TAGS    = map { "f$_" } 0 .. $NF - 1;

# --- spawn the daemon -----------------------------------------------------
my $bin = "$FindBin::Bin/../src_bin/iforest";
die "cannot find $bin\n" unless -f $bin;

my $tmp  = tempdir( CLEANUP => 1 );
my $sock = "$tmp/b.sock";
die "temp socket path too long for a Unix socket ($sock)\n" if length($sock) > 100;

my $daemon = fork();
die "fork failed: $!" unless defined $daemon;
if ( !$daemon ) {
	open( STDOUT, '>>', "$tmp/streamd.log" ) or die $!;
	open( STDERR, '>>', "$tmp/streamd.log" ) or die $!;
	exec(
		$^X, "-I$FindBin::Bin/../lib", $bin, 'streamd', '-f',
		'--socket'        => $sock,
		'--pid'           => "$tmp/b.pid",
		'--model-dir'     => "$tmp/models",
		'--save-interval' => 3600,            # no interval saves mid-benchmark
		'-n'              => $N_TREES,
		'--window'        => $WINDOW,
		'--eta'           => $ETA,
		'-s'              => 42,
		( map { ( '-t' => $_ ) } @TAGS ),
	) or die "exec failed: $!";
} ## end if ( !$daemon )
for ( 1 .. 100 ) {
	last if -S $sock;
	select( undef, undef, undef, 0.1 );    ## no critic (ProhibitSleepViaSelect)
}
die "daemon never came up; see $tmp/streamd.log\n" unless -S $sock;

END {
	kill( 'TERM', $daemon ) if $daemon && kill( 0, $daemon );
}

# --- wire helpers ----------------------------------------------------------
my %BUF;

sub connect_daemon {
	my $s = IO::Socket::UNIX->new( Peer => $sock ) or die "connect failed: $!";
	$s->autoflush(1);
	$BUF{ fileno($s) } = '';
	return $s;
}

sub rt {
	my ( $s, $msg ) = @_;
	print {$s} $JSON->encode($msg) . "\n";
	my $buf = \$BUF{ fileno($s) };
	my $sel = IO::Select->new($s);
	while ( $$buf !~ /\n/ ) {
		die "no reply from the daemon\n" unless $sel->can_read(30);
		my $got = sysread( $s, my $chunk, 262144 );
		die "daemon closed the connection\n" unless $got;
		$$buf .= $chunk;
	}
	$$buf =~ s/\A([^\n]*)\n//;
	my $reply = $JSON->decode($1);
	die "daemon error: $reply->{error}\n" if defined $reply->{error};
	return $reply;
} ## end sub rt

# Pump rows through in lockstep batches; returns elapsed seconds.
sub pump {
	my ( $s, $rows, $batch, $mode ) = @_;
	my $n  = scalar @$rows;