view release on metacpan or  search on metacpan

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

		} ## end for my $s (@ready)

		# Drain clients whose replies did not fit in one write.
		if ( $wsel->count ) {
			for my $s ( $wsel->can_write(0) ) {
				_flush( $s, $rsel, $wsel );
			}
		}

		if ($REOPEN_LOG) {
			$REOPEN_LOG = 0;
			_open_log();
			_log('log reopened on SIGHUP');
		}
		if ( $SAVE_NOW || time >= $next_save ) {
			_save_model( $SAVE_NOW ? 'signal' : 'interval' ) if $DIRTY || $SAVE_NOW;
			$SAVE_NOW  = 0;
			$next_save = time + $OPT{'save_interval'};
		}
	} ## end while ($RUN)

	# --- shutdown ------------------------------------------------------------
	_log('shutting down');
	_save_model('shutdown') if $DIRTY;
	for my $c ( values %CONN ) {
		close $c->{sock};
	}
	%CONN = ();
	close $listener;
	unlink $OPT{'socket'};
	unlink $OPT{'pid'};
	_log('bye');

	return 1;
} ## end sub execute

#-------------------------------------------------------------------------------
# startup helpers
#-------------------------------------------------------------------------------

sub _ensure_dir {
	my ( $dir, $flag ) = @_;
	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 );