<commit_before>package Mojo::Server::Prefork;
use Mojo::Base 'Mojo::Server::Daemon';

use Fcntl ':flock';
use File::Spec::Functions qw(catfile tmpdir);
use IO::Poll 'POLLIN';
use List::Util 'shuffle';
use Mojo::Util 'steady_time';
use POSIX 'WNOHANG';
use Scalar::Util 'weaken';
use Time::HiRes ();

has accepts         => 1000;
has accept_interval => 0.025;
has [qw(graceful_timeout heartbeat_timeout)] => 20;
has heartbeat_interval => 5;
has lock_file          => sub { catfile tmpdir, 'prefork.lock' };
has lock_timeout       => 1;
has multi_accept       => 50;
has pid_file           => sub { catfile tmpdir, 'prefork.pid' };
has workers            => 4;

sub DESTROY {
  my $self = shift;

  # Worker
  return unless $self->{finished};

  # Manager
  if (my $file = $self->{lock_file}) { unlink $file if -w $file }
  if (my $file = $self->pid_file)    { unlink $file if -w $file }
}

sub check_pid {
  my $file = shift->pid_file;
  return undef unless open my $handle, '<', $file;
  my $pid = <$handle>;
  chomp $pid;

  # Running
  return $pid if $pid && kill 0, $pid;

  # Not running
  unlink $file if -w $file;
  return undef;
}

sub run {
  my $self = shift;

  # No Windows support
  say 'Preforking not available for Windows.' and exit 0 if $^O eq 'MSWin32';

  # Prepare lock file and event loop
  $self->{lock_file} = $self->lock_file . ".$$";
  my $loop = $self->ioloop->max_accepts($self->accepts);
  $loop->$_($self->$_) for qw(accept_interval multi_accept);

  # Pipe for worker communication
  pipe($self->{reader}, $self->{writer}) or die "Can't create pipe: $!";
  $self->{poll} = IO::Poll->new;
  $self->{poll}->mask($self->{reader}, POLLIN);

  # Clean manager environment
  local $SIG{INT} = local $SIG{TERM} = sub { $self->_term };
  local $SIG{CHLD} = sub {
    while ((my $pid = waitpid -1, WNOHANG) > 0) { $self->_reap($pid) }
  };
  local $SIG{QUIT} = sub { $self->_term(1) };
  local $SIG{TTIN} = sub { $self->workers($self->workers + 1) };
  local $SIG{TTOU} = sub {
    $self->workers($self->workers - 1) if $self->workers > 0;
    return unless $self->workers;
    $self->{pool}{shuffle keys %{$self->{pool}}}{graceful} ||= steady_time;
  };

  # Preload application before starting workers
  $self->start->app->log->info("Manager $$ started.");
  $self->{running} = 1;
  $self->_manage while $self->{running};
}

sub _heartbeat {
  my $self = shift;

  # Poll for heartbeats
  my $poll = $self->{poll};
  $poll->poll(1);
  return unless $poll->handles(POLLIN);
  return unless $self->{reader}->sysread(my $chunk, 4194304);

  # Update heartbeats
  my $time = steady_time;
  $self->{pool}{$1} and $self->emit(heartbeat => $1)->{pool}{$1}{time} = $time
    while $chunk =~ /(\d+)\n/g;
}

sub _manage {
  my $self = shift;

  # Spawn more workers and check PID file
  if (!$self->{finished}) {
    $self->_spawn while keys %{$self->{pool}} < $self->workers;
    $self->_pid_file;
  }

  # Shutdown
  elsif (!keys %{$self->{pool}}) { return delete $self->{running} }

  # Manage workers
  $self->emit('wait')->_heartbeat;
  my $log = $self->app->log;
  for my $pid (keys %{$self->{pool}}) {
    my $w = $self->{pool}{$pid};

    # No heartbeat (graceful stop)
    my $interval = $self->heartbeat_interval;
    my $timeout  = $self->heartbeat_timeout;
    my $time     = steady_time;
    if (!$w->{graceful} && ($w->{time} + $interval + $timeout <= $time)) {
      $log->info("Worker $pid has no heartbeat, restarting.");
      $w->{graceful} = $time;
    }

    # Graceful stop with timeout
    $w->{graceful} ||= $time if $self->{graceful};
    if ($w->{graceful}) {
      $log->debug("Trying to stop worker $pid gracefully.");
      kill 'QUIT', $pid;
      $w->{force} = 1 if $w->{graceful} + $self->graceful_timeout <= $time;
    }

    # Normal stop
    if (($self->{finished} && !$self->{graceful}) || $w->{force}) {
      $log->debug("Stopping worker $pid.");
      kill 'KILL', $pid;
    }
  }
}

sub _pid_file {
  my $self = shift;

  # Check if PID file already exists
  return if -e (my $file = $self->pid_file);

  # Create PID file
  $self->app->log->info(qq{Creating process id file "$file".});
  die qq{Can't create process id file "$file": $!}
    unless open my $handle, '>', $file;
  chmod 0644, $handle;
  print $handle $$;
}

sub _reap {
  my ($self, $pid) = @_;

  # Clean up dead worker
  $self->app->log->debug("Worker $pid stopped.")
    if delete $self->emit(reap => $pid)->{pool}{$pid};
}

sub _spawn {
  my $self = shift;

  # Manager
  die "Can't fork: $!" unless defined(my $pid = fork);
  return $self->emit(spawn => $pid)->{pool}{$pid} = {time => steady_time}<commit_msg>make sure worker is still active<commit_after>package Mojo::Server::Prefork;
use Mojo::Base 'Mojo::Server::Daemon';

use Fcntl ':flock';
use File::Spec::Functions qw(catfile tmpdir);
use IO::Poll 'POLLIN';
use List::Util 'shuffle';
use Mojo::Util 'steady_time';
use POSIX 'WNOHANG';
use Scalar::Util 'weaken';
use Time::HiRes ();

has accepts         => 1000;
has accept_interval => 0.025;
has [qw(graceful_timeout heartbeat_timeout)] => 20;
has heartbeat_interval => 5;
has lock_file          => sub { catfile tmpdir, 'prefork.lock' };
has lock_timeout       => 1;
has multi_accept       => 50;
has pid_file           => sub { catfile tmpdir, 'prefork.pid' };
has workers            => 4;

sub DESTROY {
  my $self = shift;

  # Worker
  return unless $self->{finished};

  # Manager
  if (my $file = $self->{lock_file}) { unlink $file if -w $file }
  if (my $file = $self->pid_file)    { unlink $file if -w $file }
}

sub check_pid {
  my $file = shift->pid_file;
  return undef unless open my $handle, '<', $file;
  my $pid = <$handle>;
  chomp $pid;

  # Running
  return $pid if $pid && kill 0, $pid;

  # Not running
  unlink $file if -w $file;
  return undef;
}

sub run {
  my $self = shift;

  # No Windows support
  say 'Preforking not available for Windows.' and exit 0 if $^O eq 'MSWin32';

  # Prepare lock file and event loop
  $self->{lock_file} = $self->lock_file . ".$$";
  my $loop = $self->ioloop->max_accepts($self->accepts);
  $loop->$_($self->$_) for qw(accept_interval multi_accept);

  # Pipe for worker communication
  pipe($self->{reader}, $self->{writer}) or die "Can't create pipe: $!";
  $self->{poll} = IO::Poll->new;
  $self->{poll}->mask($self->{reader}, POLLIN);

  # Clean manager environment
  local $SIG{INT} = local $SIG{TERM} = sub { $self->_term };
  local $SIG{CHLD} = sub {
    while ((my $pid = waitpid -1, WNOHANG) > 0) {
      $self->app->log->debug("Worker $pid stopped.")
        if delete $self->emit(reap => $pid)->{pool}{$pid};
    }
  };
  local $SIG{QUIT} = sub { $self->_term(1) };
  local $SIG{TTIN} = sub { $self->workers($self->workers + 1) };
  local $SIG{TTOU} = sub {
    $self->workers($self->workers - 1) if $self->workers > 0;
    return unless $self->workers;
    $self->{pool}{shuffle keys %{$self->{pool}}}{graceful} ||= steady_time;
  };

  # Preload application before starting workers
  $self->start->app->log->info("Manager $$ started.");
  $self->{running} = 1;
  $self->_manage while $self->{running};
}

sub _heartbeat {
  my $self = shift;

  # Poll for heartbeats
  my $poll = $self->{poll};
  $poll->poll(1);
  return unless $poll->handles(POLLIN);
  return unless $self->{reader}->sysread(my $chunk, 4194304);

  # Update heartbeats
  my $time = steady_time;
  $self->{pool}{$1} and $self->emit(heartbeat => $1)->{pool}{$1}{time} = $time
    while $chunk =~ /(\d+)\n/g;
}

sub _manage {
  my $self = shift;

  # Spawn more workers and check PID file
  if (!$self->{finished}) {
    $self->_spawn while keys %{$self->{pool}} < $self->workers;
    $self->_pid_file;
  }

  # Shutdown
  elsif (!keys %{$self->{pool}}) { return delete $self->{running} }

  # Manage workers
  $self->emit('wait')->_heartbeat;
  my $log = $self->app->log;
  for my $pid (keys %{$self->{pool}}) {
    next unless my $w = $self->{pool}{$pid};

    # No heartbeat (graceful stop)
    my $interval = $self->heartbeat_interval;
    my $timeout  = $self->heartbeat_timeout;
    my $time     = steady_time;
    if (!$w->{graceful} && ($w->{time} + $interval + $timeout <= $time)) {
      $log->info("Worker $pid has no heartbeat, restarting.");
      $w->{graceful} = $time;
    }

    # Graceful stop with timeout
    $w->{graceful} ||= $time if $self->{graceful};
    if ($w->{graceful}) {
      $log->debug("Trying to stop worker $pid gracefully.");
      kill 'QUIT', $pid;
      $w->{force} = 1 if $w->{graceful} + $self->graceful_timeout <= $time;
    }

    # Normal stop
    if (($self->{finished} && !$self->{graceful}) || $w->{force}) {
      $log->debug("Stopping worker $pid.");
      kill 'KILL', $pid;
    }
  }
}

sub _pid_file {
  my $self = shift;

  # Check if PID file already exists
  return if -e (my $file = $self->pid_file);

  # Create PID file
  $self->app->log->info(qq{Creating process id file "$file".});
  die qq{Can't create process id file "$file": $!}
    unless open my $handle, '>', $file;
  chmod 0644, $handle;
  print $handle $$;
}

sub _spawn {
  my $self = shift;

  # Manager
  die "Can't fork: $!" unless defined(my $pid = fork);
  return $self->emit(spawn => $pid)->{pool}{$pid} = {time => steady_time}
{'fraction_non_alphanumeric': 0.19377575363261765, 'fraction_numerical': 0.009108653220559532, 'mean_word_length': 3.3341165413533833, 'pattern_counts': {'":': 2, '<': 13, '<?xml version=': 0, '>': 206, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '3713192', 'n_tokens_mistral': 3493, 'n_tokens_neox': 3356, 'n_words': 888}