view release on metacpan or  search on metacpan

lib/Developer/Dashboard/CollectorRunner.pm  view on Meta::CPAN

                command      => $job->{command},
                cwd          => $job->{cwd},
                interval     => $job->{interval},
                schedule     => $state_schedule,
                status       => 'error',
                error        => $error,
                heartbeat_at => _now_iso8601(),
            }
        );
        exit 255;
    }
    exit 0;
}

# _reap_finished_loop_workers($active_workers)
# Reaps exited scheduled worker children and removes them from the active set
# so bounded parallel collector modes do not leak zombies.
# Input: hash reference keyed by active worker pid.
# Output: count of reaped worker processes.
sub _reap_finished_loop_workers {
    my ( $self, $active_workers ) = @_;
    $active_workers ||= {};
    my $reaped = 0;
    for my $pid ( keys %{$active_workers} ) {
        my $waited = $self->_waitpid_nonblocking($pid);
        next if $waited != $pid;
        delete $active_workers->{$pid};
        $reaped++;
    }
    return $reaped;
}

# _waitpid_nonblocking($pid)
# Wraps non-blocking waitpid so loop-reap behaviour can be tested without
# relying on process timing races.
# Input: worker pid integer.
# Output: waitpid return value.
sub _waitpid_nonblocking {
    my ( $self, $pid ) = @_;
    return waitpid( $pid, 1 );
}

# _terminate_loop_workers($active_workers)
# Stops and reaps all active scheduled collector workers during loop shutdown.
# Input: hash reference keyed by active worker pid.
# Output: true value.
sub _terminate_loop_workers {
    my ( $self, $active_workers ) = @_;
    $active_workers ||= {};
    for my $pid ( keys %{$active_workers} ) {
        next if !$self->_pid_is_running($pid);
        kill 15, -$pid if !is_windows();
        kill 15, $pid;
    }
    for my $pid ( keys %{$active_workers} ) {
        for ( 1 .. 20 ) {
            last if !$self->_pid_is_running($pid);
            sleep 0.1;
        }
        # Send the group SIGKILL unconditionally: a command child that ignores
        # SIGTERM can still be alive in the worker's process group even after
        # the worker (group leader) has exited, and a running leader is only
        # signalled directly when it is still alive. kill on an empty group is a
        # harmless no-op.
        kill 9, -$pid if !is_windows();
        kill 9, $pid if $self->_pid_is_running($pid);
        $self->_reap_child_process($pid);
        delete $active_workers->{$pid};
    }
    return 1;
}

# _active_worker_pids($active_workers)
# Normalizes one active-worker tracking hash into a stable numeric pid list for
# persisted loop state and lifecycle diagnostics.
# Input: hash reference keyed by worker pid.
# Output: sorted list of numeric worker pids.
sub _active_worker_pids {
    my ( $self, $active_workers ) = @_;
    $active_workers ||= {};
    my @pids;
    for my $pid ( keys %{$active_workers} ) {
        next if !defined $pid;    # uncoverable branch true hash keys are always defined strings
        next if $pid !~ /^\d+$/;
        next if $pid <= 0;
        push @pids, $pid;
    }
    return sort { $a <=> $b } @pids;
}

# _settle_single_tick_workers($active_workers)
# Gives single-tick test loops a bounded chance to observe immediate worker
# completion before returning control to the caller.
# Input: hash reference keyed by active worker pid.
# Output: true value after the bounded settle window.
sub _settle_single_tick_workers {
    my ( $self, $active_workers ) = @_;
    $active_workers ||= {};
    for ( 1 .. 50 ) {
        last if !keys %{$active_workers};
        $self->_reap_finished_loop_workers($active_workers);
        last if !keys %{$active_workers};
        sleep 0.01;
    }
    return 1;
}

# _sleep_until_next_tick(%args)
# Sleeps until the next collector loop tick while periodically reaping any
# finished worker children so zombies do not sit around for an entire interval
# when a CHLD wakeup is missed.
# Input: interval seconds and active_workers hash reference.
# Output: true value after the bounded sleep window completes.
sub _sleep_until_next_tick {
    my ( $self, %args ) = @_;
    my $remaining = defined $args{interval} ? $args{interval} : 0;
    $remaining = 0 if $remaining < 0;
    my $active_workers = $args{active_workers} || {};
    my $slice = $remaining > 0.1 ? 0.1 : $remaining;
    while ( $remaining > 0 ) {
        $slice = $remaining if $remaining < $slice || $slice <= 0;    # uncoverable condition right slice is always positive once remaining > 0
        sleep $slice;
        $remaining -= $slice;

lib/Developer/Dashboard/CollectorRunner.pm  view on Meta::CPAN

# _open_file_descriptors()
# Lists the current process file-descriptor numbers from procfs or /dev/fd so
# detached children can close inherited caller pipes safely.
# Input: none.
# Output: sorted list of descriptor integers.
sub _open_file_descriptors {
    my ($self) = @_;
    my %seen;
    my @fds;
    for my $path ( glob('/proc/self/fd/*'), glob('/dev/fd/*') ) {
        next if $path !~ m{(?:/proc/self/fd|/dev/fd)/(\d+)\z};    # uncoverable branch true the fd globs only ever yield numeric descriptor paths
        my $fd = $1 + 0;
        next if $seen{$fd}++;
        push @fds, $fd;
    }
    return sort { $a <=> $b } @fds;
}

# _descriptor_is_inherited_pipe($fd)
# Returns whether one descriptor currently points at an inherited capture or
# IPC endpoint that a detached collector child should close after stdio has
# been redirected.
# Input: descriptor integer.
# Output: boolean true when the descriptor target is an inherited pipe,
# socketpair, or anonymous kernel handle.
sub _descriptor_is_inherited_pipe {
    my ( $self, $fd, %args ) = @_;
    return 0 if !defined $fd || $fd !~ /^\d+$/;
    my $proc_target = readlink("/proc/self/fd/$fd");
    my $dev_target  = readlink("/dev/fd/$fd");
    my $target = defined $proc_target ? $proc_target : $dev_target;
    return 0 if !defined $target || $target eq '';    # uncoverable condition right a resolved fd symlink target is never the empty string
    return 1 if $target =~ /^pipe:/;
    return 0 if !$args{close_ipc};
    return $target =~ /^(?:socket:|anon_inode:)/ ? 1 : 0;
}

# _reap_child_process($pid)
# Reaps one managed collector child owned by the current process when it has
# already exited.
# Input: process id integer.
# Output: boolean true when waitpid reaped the child.
sub _reap_child_process {
    my ( $self, $pid ) = @_;
    return 0 if !defined $pid || $pid !~ /^\d+$/ || $pid < 1;
    my $waited = waitpid( $pid, 1 );
    return $waited == $pid ? 1 : 0;
}

# _process_exists($pid)
# Checks whether the current process can still signal one process id.
# Input: process id integer.
# Output: boolean true when signal 0 succeeds.
sub _process_exists {
    my ( $self, $pid ) = @_;
    return kill( 0, $pid ) ? 1 : 0;
}

# _pid_is_running($pid)
# Determines whether one collector loop pid is still alive after opportunistic
# child reaping.
# Input: process id integer.
# Output: boolean true when the pid is still running.
sub _pid_is_running {
    my ( $self, $pid ) = @_;
    return 0 if !defined $pid || $pid !~ /^\d+$/ || $pid < 1;
    return 0 if $self->_reap_child_process($pid);
    return 0 if ( $self->_read_process_state($pid) || '' ) eq 'Z';
    return $self->_process_exists($pid) ? 1 : 0;
}

# _detach_process_session()
# Detaches the current collector loop from the parent session when the active
# platform supports POSIX setsid.
# Input: none.
# Output: true value after detaching or after explicitly skipping setsid on
# platforms that do not implement it.
sub _detach_process_session {
    my ($self) = @_;
    return 1 if is_windows();
    setsid();
    return 1;
}

# _scrub_coverage_environment()
# Removes Devel::Cover-specific environment variables from managed collector
# children so daemonized loop processes do not inherit repository test
# instrumentation.
# Input: none.
# Output: none.
sub _scrub_coverage_environment {
    my ($self) = @_;
    return if !$self->_coverage_instrumentation_active;
    delete @ENV{qw(PERL5OPT HARNESS_PERL_SWITCHES)};
    return;
}

# _coverage_instrumentation_active()
# Detects whether the current process environment requests Devel::Cover
# instrumentation.
# Input: none.
# Output: boolean true when PERL5OPT or HARNESS_PERL_SWITCHES mentions
# Devel::Cover.
sub _coverage_instrumentation_active {
    my ($self) = @_;
    my $perl5opt = join ' ', grep { defined && $_ ne '' } @ENV{qw(PERL5OPT HARNESS_PERL_SWITCHES)};
    return $perl5opt =~ /Devel::Cover/ ? 1 : 0;
}

# _job_is_due($job, $name)
# Decides whether the current loop tick should execute the collector job.
# Input: collector job hash reference and collector name string.
# Output: boolean due flag.
sub _job_is_due {
    my ( $self, $job, $name ) = @_;
    my $mode = $job->{schedule} || ( $job->{cron} ? 'cron' : $job->{interval} ? 'interval' : 'manual' );    # uncoverable condition false the schedule fallback ternary always yields a non-empty string
    return 0 if $mode eq 'manual';
    return 1 if $mode eq 'interval';
    return $self->_cron_due( $job->{cron}, $name );
}