view release on metacpan or  search on metacpan

lib/Acme/Ghost/Prefork.pm  view on Meta::CPAN

sub again {
    my $self = shift;
    my %args = @_;

    # Prefork management subsystem
    $self->{pool}               = {}; # pid => {...}
    $self->{running}            = 0; # 0 - not running; 1 - running
    $self->{finished}           = 0; # 1 - marker for spirits and manager stopping
    $self->{gracefully_stop}    = 0; # 1 - marker for gracefully stopping
    $self->{reader}             = undef; # Readable pipe to get messages from spirits
    $self->{writer}             = undef; # Writable pipe to send messages to manager
    $self->{spare}              = $args{spare} || SPARE;
    $self->{spirits}            = $args{spirits} || $args{workers} || SPIRITS;
    $self->{heartbeat_interval} = $args{heartbeat_interval} || HEARTBEAT_INTERVAL;
    $self->{heartbeat_timeout}  = $args{heartbeat_timeout} || HEARTBEAT_TIMEOUT;
    $self->{graceful_timeout}   = $args{graceful_timeout} || GRACEFUL_TIMEOUT;
    $self->{spirit_cb}          = $args{spirit};

    return $self;
}
sub startup {
    my $self = shift;

    # Pipe for spirit communication
    pipe($self->{reader}, $self->{writer}) or croak("Can't create pipe: $!\n");

    # Set manager signals
    local $SIG{INT}  = local $SIG{TERM} = sub { $self->_stop };
    local $SIG{QUIT} = sub { $self->_stop(1) };
    local $SIG{CHLD} = sub { while ((my $pid = waitpid -1, WNOHANG) > 0) { $self->_stopped($pid) } };
    local $SIG{TTIN} = sub { $self->_increase };
    local $SIG{TTOU} = sub { $self->_decrease };

    # Starting
    $self->log->info("Manager $$ started");
    $self->{running} = 1;
    $self->_manage while $self->{running};
    $self->log->info("Manager $$ stopped");
}
sub healthy {
    return scalar grep { $_->{healthy} } values %{shift->{pool}};
}
sub tick { # Spirit level
    my $self = shift;
    my $finished = shift || 0; # 0 - no finished; 1 - finished
    $self->_heartbeat($finished);
    return $self->ok;
}

# User hooks
sub finish { }      # Emitted when the server shuts down
sub heartbeat { }   # Emitted when a heartbeat message has been received from a spirit
sub reap { }        # Emitted when a child process exited
sub spawn { }       # Emitted when a spirit process is spawned
sub waitup { }      # Emitted when the manager starts waiting for new heartbeat messages
sub spirit {
    my $self = shift;
    my $cb = $self->{spirit_cb};
    return unless $cb;
    return $self->$cb if ref($cb) eq 'CODE';
    $self->log->error("Callback `spirit` is incorrect");
    $self->tick(1);
}

# Internal methods
sub _increase { # Manager level
    my $self = shift;
    $self->log->debug(sprintf("> Increase spirit pool by one")) if DEBUG;
    $self->{spirits} = $self->{spirits} + 1;
}
sub _decrease { # Manager level
    my $self = shift;
    $self->log->debug(sprintf("> Decrease spirit pool by one")) if DEBUG;
    return unless $self->{spirits} > 0;
    $self->{spirits} = $self->{spirits} - 1;

    # Set graceful time for first found unfinished pid (spirit)
    for my $w (values %{$self->{pool}}) {
        unless ($w->{graceful}) {
            $w->{graceful} = Time::HiRes::time;
            last;
        }
    }
}
sub _stop { # Manager level
    my ($self, $graceful) = @_;
    $self->log->debug(sprintf("> Received stop signal/command: %s",
        $graceful ? 'graceful shutdown' : 'forced shutdown')) if DEBUG;
    $self->finish($graceful);
    $self->{finished} = 1;
    $self->{gracefully_stop} = $graceful ? 1 : 0;
}
sub _stopped { # Manager level (Calls when a child process exited)
    my $self = shift;
    my $pid = shift;
    $self->log->debug(sprintf("> Reap %s", $pid)) if DEBUG;
    $self->reap($pid);

    return unless my $w = delete $self->{pool}{$pid};
    $self->log->info("Spirit $pid stopped");
    unless ($w->{healthy}) {
        $self->log->error("Spirit $pid stopped too early, shutting down");
        $self->_stop;
    }
}
sub _manage { # Manager level
    my $self = shift;

    # Spawn more spirits if necessary
    if (!$self->{finished}) { # No finished
        my $graceful = grep { $_->{graceful} } values %{$self->{pool}}; # Number gracefuled spirits
        my $spare = $self->{spare};
           $spare = $graceful # Check gracefuls
                ? $graceful > $spare # Check difference between graceful numbers and spare numbers
                    ? $spare # graceful numbers greater than spare numbers - use original spare value
                    : $graceful # graceful numbers less or equal to spare numbers - set spare to graceful
                : 0; # No gracefuls - no spares - set spare to 0 ('spare = 0')
        my $required = ($self->{spirits} - keys %{$self->{pool}}) + $spare; # How many spirits are required?
        $self->log->debug(sprintf("> graceful=%d; spare=%d; need=%d", $graceful, $spare, $required))
            if DEBUG && $required;
        $self->_spawn while $required-- > 0; # Spawn required spirits
    } elsif (!keys %{$self->{pool}}) { # No PIDs found, shutdown!
        return delete $self->{running}; # Return from the manager and exit immediately
    }

    # Wait for heartbeats
    $self->_wait;

    # Stops
    my $interval = $self->{heartbeat_interval};
    my $hb_to    = $self->{heartbeat_timeout};
    my $gf_to    = $self->{graceful_timeout};
    my $now      = Time::HiRes::time;
    my $log      = $self->log;
    for my $pid (keys %{$self->{pool}}) {
        next unless my $w = $self->{pool}{$pid}; # Get spirit struct

        # No heartbeat (graceful stop)
        if (!$w->{graceful} && ($w->{time} + $interval + $hb_to <= $now)) {
            $log->error("Spirit $pid has no heartbeat ($hb_to seconds), restarting");
            $w->{graceful} = $now;
        }

        # Graceful stop with timeout
        my $graceful = $w->{graceful} ||= $self->{gracefully_stop} ? $now : undef;
        if ($graceful && !$w->{attempt}) {
            $w->{attempt}++;
            $log->info("Stopping spirit $pid gracefully ($gf_to seconds)");
            kill 'QUIT', $pid or $self->_stopped($pid);
        }
        $w->{force} = 1 if $graceful && $graceful + $gf_to <= $now; # The conditions for a graceful stop by timeout were violated

        # Normal stop
        if ($w->{force} || ($self->{finished} && !$graceful)) {
            $log->warn("Stopping spirit $pid immediately");
            kill 'KILL', $pid or $self->_stopped($pid);
        }
    }
}
sub _spawn { # Manager level (Spawn a spirit and transferring control to it)
    my $self = shift;

    # Manager
    croak("Can't fork: $!\n") unless defined(my $pid = fork);
    if ($pid) { # Parent (manager)
        $self->spawn($pid);
        return $self->{pool}{$pid} = {time => Time::HiRes::time};
    }
    $self->{spirited} = 1; # Inspiration! (disables cleanup)

    weaken $self;

    # Clean spirit signals
    $SIG{$_} = 'DEFAULT' for qw/CHLD INT TERM TTIN TTOU/;

    # Set QUIT signal
    $SIG{QUIT} = sub {
        $self->log->warn("Spirit $$ received QUIT signal") if DEBUG;
        $self->_heartbeat(1); # Send finish command to manager
    };

    # Close reader pipe
    delete $self->{reader};

    # Reset the random number seed for spirit
    srand;

    $self->log->info("Spirit $$ started");

    # Start spirit
    $self->spirit;

    exit 0; # EXIT FROM APPLICATION
}
sub _wait { # Manager level
    my $self = shift;

    # Call waitup hook
    $self->waitup;