view release on metacpan or  search on metacpan

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

	if ( !-d $dir ) {
		my $err;
		make_path( $dir, { mode => oct('0755'), error => \$err } );
		die(      'could not create "'
				. $dir
				. '" (needed for '
				. $flag
				. '); create it, fix permissions, or point '
				. $flag
				. ' somewhere writable'
				. "\n" )
			if !-d $dir;
	} ## end if ( !-d $dir )
	die( '"' . $dir . '" (needed for ' . $flag . ') is not writable; fix permissions or override ' . $flag . "\n" )
		unless -w $dir;
	return 1;
} ## end sub _ensure_dir

# Classic double-fork daemonization.  The parents leave via POSIX::_exit
# so no END blocks (Inline's, App::Cmd's) run twice.
sub _daemonize {
	defined( my $pid = fork() ) or die( 'fork failed: ' . $! . "\n" );
	POSIX::_exit(0) if $pid;
	setsid()                 or die( 'setsid failed: ' . $! . "\n" );
	defined( $pid = fork() ) or die( 'second fork failed: ' . $! . "\n" );
	POSIX::_exit(0) if $pid;
	chdir '/'                       or die( 'chdir / failed: ' . $! . "\n" );
	open( STDIN, '<', '/dev/null' ) or die( 'reopen STDIN failed: ' . $! . "\n" );
	return 1;
} ## end sub _daemonize

sub _open_log {
	if ( defined $OPT{'log'} ) {
		open( my $fh, '>>', $OPT{'log'} ) or die( 'failed to open log "' . $OPT{'log'} . '": ' . $! . "\n" );
		$fh->autoflush(1);
		$LOG_FH = $fh;
		if ( !$OPT{'f'} ) {
			open( STDOUT, '>>', $OPT{'log'} ) or die( 'reopen STDOUT failed: ' . $! . "\n" );
			open( STDERR, '>>', $OPT{'log'} ) or die( 'reopen STDERR failed: ' . $! . "\n" );
			STDOUT->autoflush(1);
			STDERR->autoflush(1);
		}
	} else {
		$LOG_FH = \*STDERR;
	}
	return 1;
} ## end sub _open_log

sub _log {
	my ($msg) = @_;
	print {$LOG_FH} strftime( '%Y-%m-%dT%H:%M:%S', localtime ) . ' [' . $$ . '] ' . $msg . "\n";
	return 1;
}

#-------------------------------------------------------------------------------
# model persistence
#-------------------------------------------------------------------------------

# Timestamped save + atomic symlink flip.  Returns the file name saved
# to (relative to model-dir, which is also what the symlink stores so
# the directory stays relocatable).
sub _save_model {
	my ($why) = @_;

	# 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

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
#-------------------------------------------------------------------------------

sub _drop {
	my ( $s, $rsel, $wsel ) = @_;
	$rsel->remove($s);
	$wsel->remove($s);
	delete $CONN{ fileno($s) };
	close $s;
	return 1;
}

sub _read_from {
	my ( $s, $rsel, $wsel ) = @_;
	my $c = $CONN{ fileno($s) } or return;