Algorithm-Classifier-IsolationForest
view release on metacpan or search on metacpan
benchmarking/bench-streamd.pl view on Meta::CPAN
#!/usr/bin/perl
# benchmarking/bench-streamd.pl
#
# Benchmarks `iforest streamd` end to end: the script spawns its own
# daemon on a temp Unix socket and pumps rows through the JSON-lines
# protocol, measuring points/second wall-clock at the client.
#
# Sections:
# 1. in-process baseline -- score_learn on an identical model in this
# process; the ceiling everything else is measured against. The
# gap between it and the socket numbers is protocol + JSON + IPC
# overhead, not model work.
# 2. batch-size sweep -- prequential rows per {"rows": [...]}
# message; directly informs streamc's --batch choice.
# 3. modes -- prequential vs score vs learn at a fixed
benchmarking/bench-streamd.pl view on Meta::CPAN
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",
benchmarking/bench-streamd.pl view on Meta::CPAN
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);
lib/Algorithm/Classifier/IsolationForest.pm view on Meta::CPAN
$iforest->save('model.json');
my $reloaded = Algorithm::Classifier::IsolationForest->load('model.json');
# Extended Isolation Forest (oblique hyperplane splits)
my $eif = Algorithm::Classifier::IsolationForest->new(
mode => 'extended',
seed => 42,
);
$eif->fit(\@data);
# Parallel training (fork-based, Unix-like platforms): build the
# n_trees across several worker processes.
my $iforest = Algorithm::Classifier::IsolationForest->new(
n_trees => 200,
sample_size => 256,
seed => 42,
parallel_fit => 4, # 4 forked workers
);
$iforest->fit(\@data);
# Pre-pack a dataset to skip the per-call input-walk cost when the
lib/Algorithm/Classifier/IsolationForest.pm view on Meta::CPAN
# my $vec = $plan->apply_named( { method => 'GET', host => 'h' } );
# }
sub _plan {
my ($self) = @_;
return undef unless $self->{mungers};
$self->{_munger_plan} //= _compile_mungers( $self->{feature_names}, $self->{mungers} );
return $self->{_munger_plan};
}
# Memoised "does this perl have a real fork()?". False on Windows
# without Cygwin; true on every Unix-like platform. fit() consults it
# before honouring parallel_fit, which is how that option degrades to a
# serial fit instead of failing.
#
# Args: none.
#
# Returns: 1 when Config's d_fork is defined, 0 otherwise. Config is
# loaded on the first call only and the answer cached for the process.
#
# Example:
# $self->_fit_trees_parallel(...) if $workers > 1 && _fork_supported();
lib/Algorithm/Classifier/IsolationForest/App/Command/streamc.pm view on Meta::CPAN
sub opt_spec {
return (
[
'set=s',
'Named streamd instance to talk to; the socket becomes <set>.sock under the run dir, '
. 'exactly as streamd resolves it. Must match /\A[A-Za-z0-9+\-@_]+\z/.'
],
[
'socket=s',
'Unix domain socket streamd listens on; default /var/run/iforest_streamd/streamd.sock. With '
. '--set this is instead the base run dir (default /var/run/iforest_streamd) holding <set>.sock.',
{ 'completion' => 'files' }
],
[ 'timeout=i', 'Seconds to wait for each reply from the daemon.', { 'default' => 30 } ],
# stream mode
[
'i=s',
'Input to stream through the daemon, one row per line; - reads stdin.',
{ 'completion' => 'files' }
lib/Algorithm/Classifier/IsolationForest/App/Command/streamc.pm view on Meta::CPAN
[ 'stats', 'Print the daemon stats (seen, window, threshold, connections, set, ...).' ],
[ 'save', 'Ask the daemon to save the model now; prints the file name.' ],
[ 'relearn-threshold', 'Ask the daemon to relearn the contamination decision threshold.' ],
[ 'json', 'Command mode: print the raw JSON reply instead of the text rendering.' ],
);
} ## end sub opt_spec
sub abstract { 'Client for iforest streamd: stream rows through it or send it commands' }
sub description {
'Talks to a running `iforest streamd` daemon over its Unix socket,
speaking the same one-JSON-document-per-line protocol.
Stream mode (-i) feeds rows through the daemon and prints one result
per row, in order. Input is CSV by default (positional rows, matching
`iforest stream`; fields are sent as numbers when they look like
numbers and as raw strings otherwise, so munged columns pass through
untouched) and the output is `$score,$label` lines, with -d prepending
the input columns. With --jsonl each input line is instead a JSON row
-- an array for positional data, an object for a tagged row through
the full munger plan -- and the output is the daemon\'s reply JSON
lib/Algorithm/Classifier/IsolationForest/App/Command/streamc.pm view on Meta::CPAN
if ( defined $opt->{'set'} ) {
my $run = defined $opt->{'socket'} ? $opt->{'socket'} : '/var/run/iforest_streamd';
$socket = File::Spec->catfile( $run, $opt->{'set'} . '.sock' );
} else {
$socket = defined $opt->{'socket'} ? $opt->{'socket'} : '/var/run/iforest_streamd/streamd.sock';
}
die( '--socket, "'
. $socket
. '", is '
. length($socket)
. ' bytes; Unix socket paths are limited to ~104 bytes -- use a shorter path'
. "\n" )
if length($socket) > 100;
$SOCK = IO::Socket::UNIX->new( Peer => $socket )
or die( 'failed to connect to "'
. $socket . '": '
. $!
. ' -- is streamd running'
. ( defined $opt->{'set'} ? ' with --set ' . $opt->{'set'} : '' ) . '?'
. "\n" );
lib/Algorithm/Classifier/IsolationForest/App/Command/streamd.pm view on Meta::CPAN
sub opt_spec {
return (
[
'set=s',
'Named instance. Appended to --model-dir, and the socket/pid become <set>.sock / <set>.pid '
. 'under the run dir, so several daemons run side by side with no other flags. '
. 'Must match /\A[A-Za-z0-9+\-@_]+\z/.'
],
[
'socket=s',
'Unix domain socket to listen on; default /var/run/iforest_streamd/streamd.sock. With --set '
. 'this is instead the base run dir (default /var/run/iforest_streamd) the <set>.sock is created in.',
{ 'completion' => 'files' }
],
[
'pid=s',
'Where to write the daemon pid; default /var/run/iforest_streamd/streamd.pid. With --set '
. 'this is instead the base run dir (default /var/run/iforest_streamd) the <set>.pid is created in.',
{ 'completion' => 'files' }
],
[
lib/Algorithm/Classifier/IsolationForest/App/Command/streamd.pm view on Meta::CPAN
],
[
'prototype=s',
'JSON prototype file to create the model from (new models only). May not be combined '
. 'with -t or --mungers. See PROTOTYPES in the module POD.',
{ 'completion' => 'files' }
],
);
} ## end sub opt_spec
sub abstract { 'Run an Online Isolation Forest scoring daemon on a Unix socket, speaking JSON lines' }
sub description {
'Runs a prequential scoring daemon around an Online Isolation Forest
model (Algorithm::Classifier::IsolationForest::Online): clients connect
to the Unix domain socket and exchange one JSON document per line.
At startup the daemon resumes from <model-dir>/latest.json when it
exists; otherwise it creates a new model from the creation knobs (-n,
--window, --eta, --growth, --subsample, -s, -c, -t, --mungers,
--prototype -- the same set `iforest stream` takes). The model is saved
to a timestamped file in --model-dir every --save-interval seconds
(only when something was learned), on SIGUSR1, on the save command, and
at shutdown; the symlink latest.json is atomically repointed at every
save, so a restart resumes the stream losing at most one interval.
lib/Algorithm/Classifier/IsolationForest/App/Command/streamd.pm view on Meta::CPAN
if defined $OPT{$path_opt};
}
# sun_path is 104 bytes on the BSDs and 108 on Linux (including the
# NUL); Socket.pm just warns and TRUNCATES an over-long path, which
# binds a socket nobody will ever find. Refuse loudly instead.
die( '--socket, "'
. $OPT{'socket'}
. '", is '
. length( $OPT{'socket'} )
. ' bytes; Unix socket paths are limited to ~104 bytes -- use a shorter path'
. "\n" )
if length( $OPT{'socket'} ) > 100;
# --- directories, before anything forks or binds -----------------------
_ensure_dir( $OPT{'model_dir'}, '--model-dir' );
_ensure_dir( dirname( $OPT{'socket'} ), '--socket' );
_ensure_dir( dirname( $OPT{'pid'} ), '--pid' );
# --- refuse to double-start ---------------------------------------------
if ( -e $OPT{'socket'} ) {
lib/Algorithm/Classifier/IsolationForest/App/Command/streamd.pm view on Meta::CPAN
if ( $mode eq 'score' ) {
return $OIF->score_samples( [$vec] )->[0];
}
my $score = $OIF->score_learn( [$vec] )->[0];
$DIRTY = 1;
return $score;
} ## end sub _apply_row
=head1 NAME
Algorithm::Classifier::IsolationForest::App::Command::streamd - Run an Online Isolation Forest scoring daemon on a Unix socket, speaking JSON lines
=head1 DESCRIPTION
Wraps the prequential loop of C<iforest stream> in a daemon: it listens
on a Unix domain socket, serves many concurrent connections from one
shared model, and exchanges one JSON document per line. Because values
travel as JSON rather than positionally, raw input headed for mungers may
safely contain commas, newlines or any unicode, and object rows run the
full munger plan -- expanding and combining mungers included, which
positional CSV cannot express.
An optional C<"tag"> on a request is echoed back verbatim. Errors are
always per-message: a bad row gets an C<{"error": ...}> reply and the
connection lives on.
t/91-streamd.t view on Meta::CPAN
#!perl
# 91-streamd.t
#
# Integration test for `iforest streamd`: starts the daemon in the
# foreground in a forked child on a temp Unix socket and drives it over
# the JSON-lines protocol. Covers startup artefacts (socket, pid file),
# the request kinds (row / rows / cmd) and modes, client-tag echo on
# success and error, per-message error isolation, multiple concurrent
# connections, command and interval saves with the latest.json symlink,
# clean SIGTERM shutdown, resume-from-latest, and (when
# Algorithm::ToNumberMunger is installed) raw munged values that CSV
# framing could never carry.
#
# Skipped on Windows (Unix sockets + fork) and without JSON::MaybeXS.
use strict;
use warnings;
use Test::More;
use File::Temp qw(tempdir);
use File::Spec;
use IO::Select ();
use IO::Socket::UNIX ();
my $bin = File::Spec->rel2abs('bin/iforest');
plan skip_all => 'bin/iforest not found' unless -x $bin;
plan skip_all => 'streamd needs Unix sockets and fork()' if $^O eq 'MSWin32';
plan skip_all => 'JSON::MaybeXS is not installed'
unless eval { require JSON::MaybeXS; 1 };
my $JSON = JSON::MaybeXS->new( utf8 => 1 );
my $tmp = tempdir( CLEANUP => 1 );
my $sock = "$tmp/s.sock";
my $pidf = "$tmp/s.pid";
my $mdir = "$tmp/models";
my $logf = "$tmp/streamd.log";
my $latest = "$mdir/latest.json";
# sun_path caps out around 104 bytes; a deep TMPDIR would make the whole
# run fail for reasons that have nothing to do with the daemon.
plan skip_all => 'temp socket path too long for a Unix socket'
if length($sock) > 100;
my %BUF; # per-connection read buffer, for line framing
my @ALL_PIDS; # every daemon spawned, for the END sweep
# Fork+exec a daemon with the given argv and wait for its socket.
sub spawn_daemon {
my ( $wait_sock, @argv ) = @_;
my $pid = fork();
die "fork failed: $!" unless defined $pid;
t/92-streamc.t view on Meta::CPAN
#!perl
# 92-streamc.t
#
# Integration test for `iforest streamc`: spawns a streamd daemon (as
# t/91 does) and drives it through the real streamc CLI in a
# subprocess. Covers command mode (--ping/--stats/--save, exit codes,
# --json), CSV stream mode in all three --mode settings with -d,
# --jsonl tagged rows, client- and daemon-side error attribution by
# input line, --set socket resolution, and connect-failure behaviour.
#
# Skipped on Windows (Unix sockets + fork) and without JSON::MaybeXS.
use strict;
use warnings;
use Test::More;
use File::Temp qw(tempdir);
use File::Spec;
my $bin = File::Spec->rel2abs('bin/iforest');
plan skip_all => 'bin/iforest not found' unless -x $bin;
plan skip_all => 'streamc needs Unix sockets and fork()' if $^O eq 'MSWin32';
plan skip_all => 'JSON::MaybeXS is not installed'
unless eval { require JSON::MaybeXS; 1 };
my $JSON = JSON::MaybeXS->new( utf8 => 1 );
my $tmp = tempdir( CLEANUP => 1 );
my $mdir = "$tmp/models";
my $logf = "$tmp/streamd.log";
plan skip_all => 'temp socket path too long for a Unix socket'
if length("$tmp/alpha.sock") > 100;
my @ALL_PIDS;
END {
for my $pid (@ALL_PIDS) {
kill( 'TERM', $pid ) if kill( 0, $pid );
}
}
( run in 1.397 second using v1.01-cache-2.11-cpan-64ef6c95b5d )