File Coverage

blib/lib/POE/Component/ProcTerminator.pm

Criterion	Covered	Total	%
statement	92	122	75.4
branch	6	22	27.2
condition	4	13	30.7
subroutine	23	24	95.8
pod	1	1	100.0
total	126	182	69.2

package POE::Component::ProcTerminator::Batch;
use strict;
use warnings;
use Proc::Terminator;
use Moo;

extends 'Proc::Terminator::Batch';
has timer_id => (is => 'rw');
has failure_callback => (
    is => 'ro',
    isa => sub { ref $_[0] eq 'CODE'
        or die "Failure callback should be a CODE reference" }
);
has cleanup_flags => ( is => 'rw', default => 0 );

package POE::Component::ProcTerminator;
use warnings;
use strict;
use POE::Sugar::Attributes;
use POE;
use POE::Session;
use Proc::Terminator;
use base qw(POE::Sugar::Attributes Exporter);
use Class::Struct;
use Time::HiRes qw(time);
use POSIX qw(:signal_h);

our @EXPORT;
our $DefaultSession = "proc_terminator";

use Constant::Generate [qw(
    SIGKILL_LAST
    SIGKILL_ONLY
    CLEANUP_BLOCK
)], -type => 'bit', -export => 1, -prefix => 'PROCTERMf_';

struct
    'POE::Component::ProcTerminator::SessionInfo' =>
    [
        # hash of 'batches', indexed by their timer IDs
        batches => '*%',

        # default options
        defaults => '*%'

    ];

our $VERSION = 0.03;


sub spawn {
    my ($cls,%options) = @_;
    my $alias = delete $options{Alias};
    $alias ||= $DefaultSession;

    my $inline_states =
        POE::Sugar::Attributes->inline_states(__PACKAGE__, $alias);

    my $sessinfo = POE::Component::ProcTerminator::SessionInfo->new();
    $sessinfo->defaults(\%options);
    POE::Session->create(
        inline_states => $inline_states,
        heap => $sessinfo
    );
}

sub _do_terminate :Event(terminate)
{
    my ($sessinfo,$pids,$options) = @_[HEAP, ARG0..ARG2];
    $options ||= {};
    $options = { %{ $sessinfo->defaults }, %$options };

    $options->{cleanup_flags} ||= 0;

    if (delete $options->{max_wait}) {
        warn "max_wait does not make sense in POE::Component::ProcTerminator";
    }
    my $batch = POE::Component::ProcTerminator::Batch->with_pids(
        $pids,
        max_wait => 0,
        %$options
    );
    $batch->begin_time(time);
    my $ret = $batch->loop_once;
    if ($ret) {
        my $new_aid =$_[KERNEL]->delay_set(
            _grace_next => $batch->grace_period,
            $batch
        );
        $batch->timer_id($new_aid);
    } elsif ((!defined $ret) && $batch->failure_callback) {
        $batch->failure_callback->($batch);
    }
}

sub _grace_next :Event {
    my ($sessinfo,$batch) = @_[HEAP,ARG0];
    my $ret = $batch->loop_once();
    if ($ret) {
        my $new_aid = $_[KERNEL]->delay_set(
            $_[STATE],
            $batch->grace_period,
            $batch,
        );
        $batch->timer_id($new_aid);

    } else {
        if ( (!defined $ret) && $batch->failure_callback ) {
            $batch->failure_callback->($batch);
        }
        delete $sessinfo->batches->{$batch->timer_id};
    }
}


sub _unknown :Event(_default)
{
    my ($event,$args) = @_[ARG0,ARG1];
    warn "Unhandled Event '$event'";
}


sub _clean_all :Stop {
    my $sessinfo = $_[HEAP];

    # iterate through all the batches. Figure out which ones should
    # be killed immediately
    my $remaining_batch = Proc::Terminator::Batch->with_pids([]);
    my $counter = 0;

    while (my ($aid,$batch) = each %{$sessinfo->batches}) {
        if ($batch->cleanup_flags & PROCTERMf_SIGKILL_ONLY) {
            $batch->cleanup_flags(0);
            kill (SIGKILL, $batch->pid);
            $_[KERNEL]->alarm_remove($batch->timer_id);
        } elsif ($batch->cleanup_flags & PROCTERMf_SIGKILL_LAST) {
            while (my ($pid,$ctx) = each %{$batch->procs}) {
                if ($ctx->siglist->[$#{$ctx->siglist}] != SIGKILL) {
                    push @{$ctx->siglist}, SIGKILL;
                }
            }
        }

        delete $sessinfo->batches->{$aid};

        if (($batch->cleanup_flags & PROCTERMf_CLEANUP_BLOCK) == 0) {
            $batch->loop_once();
        } else {
            # cleanup..
            @{$remaining_batch->procs}{keys %{$batch->procs}} =
                values %{$batch->procs};
        }
    }

    if (%{$remaining_batch->procs}) {
        $remaining_batch->max_wait(5); # configurable maybe?
        $remaining_batch->begin_time(time);
        while ($remaining_batch->loop_once) {
            sleep(0.25);
        }
    }

    # ensure the final signal is indeed sent..
    while (my ($pid,$ctx) = each %{$remaining_batch->procs}) {
        my $sig = pop @{$ctx->siglist};
        if ($sig) {
            kill ($sig, $ctx->pid);
        }
    }
}

1;

__END__