view release on metacpan or  search on metacpan

t/91-streamd.t  view on Meta::CPAN

sub stop_daemon {
	my ($pid) = @_;
	kill( 'TERM', $pid );
	waitpid( $pid, 0 );
	return $? >> 8;
}

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

sub read_reply {
	my ( $s, $timeout ) = @_;
	$timeout //= 10;
	my $deadline = time + $timeout;
	my $sel      = IO::Select->new($s);
	my $buf      = \$BUF{ fileno($s) };
	while ( $$buf !~ /\n/ ) {
		my $left = $deadline - time;
		return undef if $left <= 0 || !$sel->can_read($left);
		my $got = sysread( $s, my $chunk, 65536 );
		return undef unless $got;
		$$buf .= $chunk;
	}
	$$buf =~ s/\A([^\n]*)\n//;
	return $JSON->decode($1);
} ## end sub read_reply

# One request in, one decoded reply out.
sub rt {
	my ( $s, $msg ) = @_;
	print {$s} $JSON->encode($msg) . "\n";
	return read_reply($s);
}

my $daemon = start_daemon(
	'--save-interval' => 1,
	'-n'              => 20,
	'--window'        => 64,
	'--eta'           => 8,
	'-s'              => 42,
	'-t'              => 'cpu',
	'-t'              => 'mem'
);

END {
	for my $pid (@ALL_PIDS) {
		kill( 'TERM', $pid ) if kill( 0, $pid );
	}
}

subtest 'startup artefacts' => sub {
	ok( -S $sock, 'socket exists' ) or diag( scalar `cat $logf` );
	ok( -f $pidf, 'pid file exists' );
	my $recorded = do { local ( @ARGV, $/ ) = ($pidf); <> };
	chomp $recorded;
	is( $recorded, $daemon, 'pid file records the daemon pid (foreground: the child)' );
};

my $c = connect_client();

subtest 'ping and client-tag echo' => sub {
	is_deeply( rt( $c, { cmd => 'ping' } ), { ok => 'pong' }, 'ping pongs, no tag when none sent' );
	is_deeply(
		rt( $c, { cmd => 'ping', tag => 'req-1' } ),
		{ ok => 'pong', tag => 'req-1' },
		'string tag echoed back'
	);
	is_deeply(
		rt( $c, { cmd => 'ping', tag => { batch => 7, src => [ 'a', 'b' ] } } ),
		{ ok => 'pong', tag => { batch => 7, src => [ 'a', 'b' ] } },
		'structured tag echoed back verbatim'
	);
}; ## end 'ping and client-tag echo' => sub

subtest 'learn, prequential, score modes' => sub {
	srand(5);
	my @warm = map { [ rand, rand ] } 1 .. 60;
	is_deeply(
		rt( $c, { rows => \@warm, mode => 'learn' } ),
		{ ok => { learned => 60 } },
		'rows batch in learn mode learns them all'
	);

	my $r = rt( $c, { row => [ 0.5, 0.5 ], tag => 'p1' } );
	is( $r->{tag}, 'p1', 'row reply carries the tag' );
	like( $r->{score}, qr/\A[\d.eE+-]+\z/, 'prequential row returns a numeric score' );
	ok( $r->{label} == 0 || $r->{label} == 1, 'and a 0/1 label' );

	my $stats = rt( $c, { cmd => 'stats' } );
	is( $stats->{ok}{seen},       61, 'prequential row was learned (seen=61)' );
	is( $stats->{ok}{n_features}, 2,  'stats reports n_features' );

	rt( $c, { row => [ 0.5, 0.5 ], mode => 'score' } );
	is( rt( $c, { cmd => 'stats' } )->{ok}{seen}, 61, 'score mode does not advance the model' );

	my $batch = rt( $c, { rows => [ [ 0.4, 0.4 ], [ 0.6, 0.6 ] ], mode => 'score' } );
	is( scalar @{ $batch->{scores} },    2, 'rows batch returns one [score,label] pair per row' );
	is( scalar @{ $batch->{scores}[0] }, 2, 'pairs have two elements' );

	my $tagged     = rt( $c, { row => { cpu => 0.5, mem => 0.5 }, mode => 'score' } );
	my $positional = rt( $c, { row => [ 0.5, 0.5 ], mode => 'score' } );
	cmp_ok( abs( $tagged->{score} - $positional->{score} ),
		'<', 1e-12, 'tagged (object) row form scores like the positional form' );
}; ## end 'learn, prequential, score modes' => sub

subtest 'errors are per-message and carry the tag' => sub {
	print {$c} "this is not json\n";
	like( read_reply($c)->{error}, qr/not a JSON object/, 'unparseable line gets an error reply' );

	my $r = rt( $c, { row => [ 1, 2, 3 ], tag => 'w1' } );
	like( $r->{error}, qr/3 features but model expects 2/, 'wrong-width row errors' );
	is( $r->{tag}, 'w1', 'error reply carries the tag' );

	like(