view release on metacpan or  search on metacpan

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

# --- 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;
	my $t0 = time;
	my $i  = 0;
	while ( $i < $n ) {
		my $end = $i + $batch - 1;
		$end = $n - 1 if $end > $n - 1;
		rt( $s, { rows => [ @{$rows}[ $i .. $end ] ], mode => $mode } );
		$i = $end + 1;
	}
	return time - $t0;
} ## end sub pump

sub report {
	my ( $label, $n, $elapsed ) = @_;
	printf "  %-34s  %10.0f pts/s  (%d rows in %.2fs)\n", $label, $n / $elapsed, $n, $elapsed;
	return;
}

print "=" x 70, "\n";
print " streamd end-to-end benchmarks (JSON lines over a Unix socket)\n";
print "=" x 70, "\n";
printf " %d trees, window %d, eta %d, %d features; JSON backend: %s\n",
	$N_TREES, $WINDOW, $ETA, $NF, JSON::MaybeXS::JSON();
print " (points/second wall-clock at the client; higher is faster)\n";

# Warm both the daemon and the baseline model past the window size so
# every section measures steady-state, full-window work.
srand(42);
my $warm = make_data( 3000, $NF );
srand(43);
my $feed = make_data( 60000, $NF );

my $c = connect_daemon();
pump( $c, $warm, 500, 'learn' );

# -----------------------------------------------------------------------
# 1. In-process baseline: the same stream loop without the wire
# -----------------------------------------------------------------------
print "\n--- 1. in-process score_learn baseline (no socket, no JSON) ---\n";
my $base = Algorithm::Classifier::IsolationForest::Online->new(
	n_trees          => $N_TREES,
	window_size      => $WINDOW,
	max_leaf_samples => $ETA,
	seed             => 42,
);
$base->learn($warm);
{
	my @rows    = @{$feed}[ 0 .. 9999 ];
	my $t0      = time;
	my $s       = $base->score_learn( \@rows );
	my $elapsed = time - $t0;
	report( 'score_learn, one 10k-row call', scalar @rows, $elapsed );
}