view release on metacpan or  search on metacpan

lib/Algorithm/Classifier/IsolationForest/App/Command/streamd.pm  view on Meta::CPAN

	# Keep the persisted default cutoff tracking the stream, like the
	# stream command does before its save.
	if ( defined $OIF->{contamination} && $OIF->window_count ) {
		$OIF->relearn_threshold;
	}

	my $base = 'oiforest-' . strftime( '%Y%m%d-%H%M%S', localtime );
	my $name = $base . '.json';
	my $n    = 0;
	while ( -e File::Spec->catfile( $OPT{'model_dir'}, $name ) ) {
		$n++;
		$name = $base . '-' . $n . '.json';
	}
	write_file( File::Spec->catfile( $OPT{'model_dir'}, $name ), { 'atomic' => 1 }, $OIF->to_json );

	my $tmp = File::Spec->catfile( $OPT{'model_dir'}, '.latest.tmp.' . $$ );
	unlink $tmp;
	symlink( $name, $tmp )
		or _log( 'WARNING: symlink for latest.json failed: ' . $! );
	rename( $tmp, File::Spec->catfile( $OPT{'model_dir'}, 'latest.json' ) )
		or _log( 'WARNING: renaming latest.json symlink failed: ' . $! );

	$DIRTY = 0;
	_log( 'saved ' . $name . ' (' . $why . ', seen=' . $OIF->seen . ')' );
	_prune_models() if defined $OPT{'keep'};
	return $name;
} ## end sub _save_model

# Keep only the newest --keep timestamped models, deleting the rest
# oldest-first.  Ordered by mtime rather than by the name's timestamp so
# a clock stepping backwards cannot strand a file forever.  latest.json is
# a symlink, not a match for the model pattern, so it is never a
# candidate.
#
# Args: none.  Reads $OPT{'model_dir'} and $OPT{'keep'}, and is only
# called when --keep is set.
#
# Returns: 1, or an empty return when the model directory cannot be
# opened -- pruning is housekeeping, so it never takes the daemon down.
#
# Example:
#   _prune_models() if defined $OPT{'keep'};
sub _prune_models {
	opendir( my $dh, $OPT{'model_dir'} ) or return;
	my @models = sort { ( stat($a) )[9] <=> ( stat($b) )[9] }
		map { File::Spec->catfile( $OPT{'model_dir'}, $_ ) }
		grep { /\Aoiforest-.*\.json\z/ } readdir($dh);
	closedir $dh;
	while ( scalar @models > $OPT{'keep'} ) {
		my $old = shift @models;
		unlink $old and _log( 'pruned ' . $old );
	}
	return 1;
} ## end sub _prune_models

#-------------------------------------------------------------------------------
# connection handling
#-------------------------------------------------------------------------------

# Close one client connection and forget everything about it.  The single
# teardown path, so a connection can never be left in a selector after its
# socket is gone.
#
# Args:
#   $s :: the client socket.
#   $rsel :: the read IO::Select set it may be registered in.
#   $wsel :: the write IO::Select set it may be registered in.
#
# Returns: 1.  Callers return its value directly, which is why the read
# and write loops read as "return _drop(...)".
#
# Example:
#   return _drop( $s, $rsel, $wsel ) if $got == 0;   # client closed
sub _drop {
	my ( $s, $rsel, $wsel ) = @_;
	$rsel->remove($s);
	$wsel->remove($s);
	delete $CONN{ fileno($s) };
	close $s;
	return 1;
}

# Read whatever is available from one client, dispatch every complete
# line in it, and push the replies back out.  Partial lines stay in the
# connection's input buffer for the next readable event, which is what
# lets one non-blocking loop serve many clients.
#
# A client that sends an unterminated line past MAX_INBUF, or backs up
# past MAX_OUTBUF of unread replies, is dropped: neither is recoverable,
# and both would otherwise let one client exhaust the daemon's memory.
#
# Args:
#   $s :: the readable client socket.
#   $rsel :: the read IO::Select set.
#   $wsel :: the write IO::Select set, which gains $s when a reply cannot
#            be written in full.
#
# Returns: 1 after servicing the client, or the result of _drop when the
# connection ended or misbehaved.  An empty return on EAGAIN/EINTR, or for
# a socket with no connection record, leaves the client untouched.
#
# Example:
#   _read_from( $_, $rsel, $wsel ) for $rsel->can_read($timeout);
sub _read_from {
	my ( $s, $rsel, $wsel ) = @_;
	my $c = $CONN{ fileno($s) } or return;

	my $got = sysread( $s, my $chunk, 65536 );
	if ( !defined $got ) {
		return if $!{EAGAIN} || $!{EWOULDBLOCK} || $!{EINTR};
		return _drop( $s, $rsel, $wsel );
	}
	return _drop( $s, $rsel, $wsel ) if $got == 0;    # client closed

	$c->{inbuf} .= $chunk;
	if ( length( $c->{inbuf} ) > MAX_INBUF && $c->{inbuf} !~ /\n/ ) {
		_log( 'dropping client: unterminated line exceeded ' . MAX_INBUF . ' bytes' );
		return _drop( $s, $rsel, $wsel );
	}

	while ( $c->{inbuf} =~ s/\A([^\n]*)\n// ) {