File Coverage

blib/lib/POE/Component/CPANPLUS/YACSmoke.pm

Criterion	Covered	Total	%
statement	7	9	77.7
branch			n/a
condition			n/a
subroutine	3	3	100.0
pod			n/a
total	10	12	83.3

package POE::Component::CPANPLUS::YACSmoke;

use strict;
use warnings;
use POE qw(Wheel::Run);
use Storable;
use Digest::MD5 qw(md5_hex);
use vars qw($VERSION);

$VERSION = '1.62';

my $GOT_KILLFAM;
my $GOT_PTY;

BEGIN {
	$GOT_KILLFAM = 0;
	eval {
		require Proc::ProcessTable;
		$GOT_KILLFAM = 1;
	};
	$GOT_PTY = 0;
	eval {
		require IO::Pty;
		$GOT_PTY = 1;
	};
}

sub spawn {
  my $package = shift;
  my %opts = @_;
  $opts{lc $_} = delete $opts{$_} for keys %opts;
  my $options = delete $opts{options};

  if ( $^O eq 'MSWin32' ) {
    eval    { require Win32; };
    if ($@) { die "Win32 but failed to load:\n$@" }
    eval    { require Win32::Job; };
    if ($@) { die "Win32::Job but failed to load:\n$@" }
    eval    { require Win32::Process; };
    if ($@) { die "Win32::Process but failed to load:\n$@" }
  }

  my $self = bless \%opts, $package;
  $self->{session_id} = POE::Session->create(
	object_states => [
	   $self => { shutdown  => '_shutdown',
		      submit    => '_command',
		      push      => '_command',
		      unshift   => '_command',
		      recent    => '_command',
		      check     => '_command',
		      indices   => '_command',
		      author    => '_command',
		      flush	=> '_command',
		      'package' => '_command',
	   },
	   $self => [ qw(_start _spawn_wheel _wheel_error _wheel_closed _wheel_stdout _wheel_stderr _wheel_idle _wheel_kill _sig_child _sig_handle) ],
	],
	heap => $self,
	( ref($options) eq 'HASH' ? ( options => $options ) : () ),
  )->ID();
  return $self;
}

sub session_id {
  return $_[0]->{session_id};
}

sub pending_jobs {
  return @{ $_[0]->{job_queue} };
}

sub current_job {
  my $self = shift;
  return unless $self->{_current_job};
  my $item = Storable::dclone( $self->{_current_job} );
  return $item;
}

sub current_log {
  my $self = shift;
  return unless $self->{_wheel_log};
  my $item = Storable::dclone( $self->{_wheel_log} );
  return $item;
}

sub pause_queue {
  my $self = shift;
  $self->{paused} = 1;
}

sub resume_queue {
  my $self = shift;
  my $pause = delete $self->{paused};
  $poe_kernel->post( $self->{session_id}, '_spawn_wheel' ) if $pause;
}

sub paused {
  return $_[0]->{paused};
}

sub statistics {
  my $self = shift;
  my @stats;
  push @stats, $self->{stats}->{$_} for qw(started totaljobs avg_run min_run max_run);
  return @stats if wantarray;
  return \@stats;
}

sub shutdown {
  my $self = shift;
  $poe_kernel->post( $self->{session_id}, 'shutdown' );
}

sub _start {
  my ($kernel,$self) = @_[KERNEL,OBJECT];
  $kernel->sig( 'HUP', '_sig_handle' );
  $self->{session_id} = $_[SESSION]->ID();
  if ( $self->{alias} ) {
	$kernel->alias_set( $self->{alias} );
  } else {
	$kernel->refcount_increment( $self->{session_id} => __PACKAGE__ );
  }
  $self->{job_queue} = [ ];
  $self->{idle} = 600 unless $self->{idle};
  $self->{timeout} = 3600 unless $self->{timeout};
  $self->{stats} = {
	started => time(),
	totaljobs => 0,
	avg_run => 0,
	min_run => 0,
	max_run => 0,
	_sum => 0,
  };
  $ENV{APPDATA} = $self->{appdata} if $self->{appdata};
  undef;
}

sub _sig_handle {
  $poe_kernel->sig_handled();
}

sub _shutdown {
  my ($kernel,$self) = @_[KERNEL,OBJECT];
  $kernel->sig( 'HUP' );
  $kernel->sig( 'KILL' );
  $kernel->alias_remove( $_ ) for $kernel->alias_list();
  $kernel->refcount_decrement( $self->{session_id} => __PACKAGE__ ) unless $self->{alias};
  $kernel->refcount_decrement( $_->{session}, __PACKAGE__ ) for @{ $self->{job_queue} };
  $self->{_shutdown} = 1;
  undef;
}

sub _command {
  my ($kernel,$self,$state,$sender) = @_[KERNEL,OBJECT,STATE,SENDER];
  return if $self->{_shutdown};
  my $args;
  if ( ref( $_[ARG0] ) eq 'HASH' ) {
	$args = { %{ $_[ARG0] } };
  } else {
	$args = { @_[ARG0..$#_] };
  }

  $state = 'push' if $state eq 'submit';

  $args->{lc $_} = delete $args->{$_} for grep { $_ !~ /^_/ } keys %{ $args };

  my $ref = $args->{session} ? $kernel->alias_resolve( $args->{session} ) : $sender;
  $args->{session} = $ref->ID();

  if ( !$args->{module} and $state !~ /^(recent|check|indices|package|author|flush)$/i ) {
	warn "No 'module' specified for $state";
	return;
  }

  unless ( $args->{event} ) {
	warn "No 'event' specified for $state";
	return;
  }

  if ( $state =~ /^(package|author)$/ and !$args->{search} ) {
	warn "No 'search' criteria specified for $state";
	return;
  }

  $args->{submitted} = time();

  if ( $state eq 'recent' ) {
    if ( $^O eq 'MSWin32' ) {
	$args->{program} = \&_recent_modules;
	$args->{program_args} = [ $args->{perl} || $self->{perl} || $^X ];
    }
    else {
	my $perl = $args->{perl} || $self->{perl} || $^X;
	my $code = 'my $smoke = CPANPLUS::YACSmoke->new(); print "$_\n" for $smoke->_download_list();';
	$args->{program} = [ $perl, '-MCPANPLUS::YACSmoke', '-e', $code ];
    }
  }
  elsif ( $state eq 'check' ) {
    if ( $^O eq 'MSWin32' ) {
	$args->{program} = \&_check_yacsmoke;
	$args->{program_args} = [ $args->{perl} || $self->{perl} || $^X ];
    }
    else {
	my $perl = $args->{perl} || $self->{perl} || $^X;
	$args->{program} = [ $perl, '-MCPANPLUS::YACSmoke', '-e', 1 ];
    }
    $args->{debug} = 1;
  }
  elsif ( $state eq 'indices' ) {
    $args->{prioritise} = 0 unless $args->{prioritise};
    if ( $^O eq 'MSWin32' ) {
	$args->{program} = \&_reload_indices;
	$args->{program_args} = [ $args->{perl} || $self->{perl} || $^X ];
    }
    else {
	my $perl = $args->{perl} || $self->{perl} || $^X;
	my $code = 'CPANPLUS::Backend->new()->reload_indices( update_source => 1 );';
	$args->{program} = [ $perl, '-MCPANPLUS::Backend', '-e', $code ];
    }
  }
  elsif ( $state eq 'author' ) {
    if ( $^O eq 'MSWin32' ) {
	$args->{program} = \&_author_search;
	$args->{program_args} = [ $args->{perl} || $self->{perl} || $^X, $args->{type} || 'cpanid', $args->{search} ];
    }
    else {
	my $perl = $args->{perl} || $self->{perl} || $^X;
	my $code = 'my $type = shift; my $search = shift; my $cb = CPANPLUS::Backend->new(); my %mods = map { $_->package() => 1 } map { $_->modules() } $cb->search( type => $type, allow => [ qr/$search/ ], verbose => 0 ); print qq{$_\n} for sort keys %mods;';
	$args->{program} = [ $perl, '-MCPANPLUS::Backend', '-e', $code, $args->{type} || 'cpanid', $args->{search} ];
    }
  }
  elsif ( $state eq 'package' ) {
    if ( $^O eq 'MSWin32' ) {
	$args->{program} = \&_package_search;
	$args->{program_args} = [ $args->{perl} || $self->{perl} || $^X, $args->{type} || 'package', $args->{search} ];
    }
    else {
	my $perl = $args->{perl} || $self->{perl} || $^X;
	my $code = 'my $type = shift; my $search = shift; my $cb = CPANPLUS::Backend->new(); my %mods = map { $_->package() => 1 } $cb->search( type => $type, allow => [ qr/$search/ ], verbose => 0 ); print qq{$_\n} for sort keys %mods;';
	$args->{program} = [ $perl, '-MCPANPLUS::Backend', '-e', $code, $args->{type} || 'package', $args->{search} ];
    }
  }
  elsif ( $state eq 'flush' ) {
    $args->{prioritise} = 0 unless $args->{prioritise};
    if ( $^O eq 'MSWin32' ) {
	$args->{program} = \&_flush;
	$args->{program_args} = [ $args->{perl} || $self->{perl} || $^X, ( $args->{type} and $args->{type} eq 'all' ? 'all' : 'old' ) ];
    }
    else {
	my $perl = $args->{perl} || $self->{perl} || $^X;
	my $code = 'my $type = shift; my $smoke = CPANPLUS::YACSmoke->new(); $smoke->flush($type) if $smoke->can("flush");';
	$args->{program} = [ $perl, '-MCPANPLUS::YACSmoke', '-e', $code, ( $args->{type} and $args->{type} eq 'all' ? 'all' : 'old' ) ];
    }
  }
  else {
    if ( $^O eq 'MSWin32' ) {
	$args->{program} = \&_test_module;
	$args->{program_args} = [ $args->{perl} || $self->{perl} || $^X, $args->{module} ];
    }
    else {
	my $perl = $args->{perl} || $self->{perl} || $^X;
	my $code = 'my $module = shift; my $smoke = CPANPLUS::YACSmoke->new(); $smoke->test($module);';
	$args->{program} = [ $perl, '-MCPANPLUS::YACSmoke', '-e', $code, $args->{module} ];
    }
  }

  $kernel->refcount_increment( $args->{session}, __PACKAGE__ );

  $args->{cmd} = $state;

  if ( $state eq 'unshift' or $state eq 'recent' or $args->{prioritise} ) {
    unshift @{ $self->{job_queue} }, $args;
  }
  else {
    push @{ $self->{job_queue} }, $args;
  }

  $kernel->yield( '_spawn_wheel' );

  undef;
}

sub _sig_child {
  my ($kernel,$self,$thing,$pid,$status) = @_[KERNEL,OBJECT,ARG0..ARG2];
  push @{ $self->{_wheel_log} }, "$thing $pid $status";
  warn "$thing $pid $status\n" if $self->{debug};
  $kernel->delay( '_wheel_idle' );
  delete $self->{_digests};
  delete $self->{_loop_detect};
  my $job = delete $self->{_current_job};
  $job->{status} = $status;
  my $log = delete $self->{_wheel_log};
  if ( $job->{cmd} eq 'recent' ) {
    pop @{ $log };
    s/\x0D$// for @{ $log };
    $job->{recent} = $log;
  }
  elsif ( $job->{cmd} =~ /^(package|author)$/ ) {
    pop @{ $log };
    s/\x0D$// for @{ $log };
    @{ $job->{results} } = grep { $_ !~ /^\[/ } @{ $log };
  }
  else {
    $job->{log} = $log;
  }
  $job->{end_time} = time();
  unless ( $self->{debug} ) {
    delete $job->{program};
    delete $job->{program_args};
  }
  # Stats
  my $run_time = $job->{end_time} - $job->{start_time};
  $self->{stats}->{max_run} = $run_time if $run_time > $self->{stats}->{max_run};
  $self->{stats}->{min_run} = $run_time if $self->{stats}->{min_run} == 0;
  $self->{stats}->{min_run} = $run_time if $run_time < $self->{stats}->{min_run};
  $self->{stats}->{_sum} += $run_time;
  $self->{stats}->{totaljobs}++;
  $self->{stats}->{avg_run} = $self->{stats}->{_sum} / $self->{stats}->{totaljobs};
  $self->{debug} = delete $job->{global_debug};
  #$ENV{APPDATA} = delete $job->{backup_env} if $job->{appdata};
  $kernel->post( $job->{session}, $job->{event}, $job );
  $kernel->refcount_decrement( $job->{session}, __PACKAGE__ );
  $kernel->yield( '_spawn_wheel' );
  $kernel->sig_handled();
}

sub _spawn_wheel {
  my ($kernel,$self) = @_[KERNEL,OBJECT];
  return if $self->{wheel};
  return if $self->{_shutdown};
  return if $self->{paused};
  my $job = shift @{ $self->{job_queue} };
  return unless $job;
  my $backup_env;
  if ( $job->{appdata} ) {
    $backup_env = $ENV{APPDATA};
    $ENV{APPDATA} = $job->{appdata};
  }
  $self->{wheel} = POE::Wheel::Run->new(
    Program     => $job->{program},
    ProgramArgs => $job->{program_args},
    StdoutEvent => '_wheel_stdout',
    StderrEvent => '_wheel_stderr',
    ErrorEvent  => '_wheel_error',
    CloseEvent  => '_wheel_close',
    ( $GOT_PTY and !$self->{no_pty} ? ( Conduit => 'pty-pipe' ) : () ),
  );
  if ( $job->{appdata} ) {
    delete $ENV{APPDATA};
    $ENV{APPDATA} = $backup_env if $backup_env;
  }
  unless ( $self->{wheel} ) {
	warn "Couldn\'t spawn a wheel for $job->{module}\n";
	$kernel->refcount_decrement( $job->{session}, __PACKAGE__ );
	return;
  }
  if ( defined $job->{debug} ) {
	$job->{global_debug} = delete $self->{debug};
	$self->{debug} = $job->{debug};
  }
  $self->{_wheel_log} = [ ];
  $self->{_digests} = { };
  $self->{_loop_detect} = 0;
  $self->{_current_job} = $job;
  $job->{PID} = $self->{wheel}->PID();
  $job->{start_time} = time();
  $kernel->sig_child( $job->{PID}, '_sig_child' );
  $kernel->delay( '_wheel_idle', 60 ) unless $job->{cmd} eq 'indices';
  undef;
}

sub _wheel_error {
  $poe_kernel->delay( '_wheel_idle' );
  delete $_[OBJECT]->{wheel};
  undef;
}

sub _wheel_closed {
  $poe_kernel->delay( '_wheel_idle' );
  delete $_[OBJECT]->{wheel};
  undef;
}

sub _wheel_stdout {
  my ($self, $input, $wheel_id) = @_[OBJECT, ARG0, ARG1];
  $self->{_wheel_time} = time();
  push @{ $self->{_wheel_log} }, $input;
  warn $input, "\n" if $self->{debug};
  if ( $self->_detect_loop( $input ) ) {
    $self->{_current_job}->{excess_kill} = 1;
    $poe_kernel->yield( '_wheel_kill', 'Killing current run CPAN::Shell loop detected' );
    return;
  }
  undef;
}

sub _wheel_stderr {
  my ($self, $input, $wheel_id) = @_[OBJECT, ARG0, ARG1];
  $self->{_wheel_time} = time();
  if ( $^O eq 'MSWin32' and !$self->{_current_job}->{GRP_PID} and my ($pid) = $input =~ /(\d+)/ ) {
     $self->{_current_job}->{GRP_PID} = $pid;
     warn "Grp PID: $pid\n" if $self->{debug};
     return;
  }
  push @{ $self->{_wheel_log} }, $input unless $self->{_current_job}->{cmd} eq 'recent';
  warn $input, "\n" if $self->{debug};
  if ( $self->_detect_loop( $input ) ) {
    $self->{_current_job}->{excess_kill} = 1;
    $poe_kernel->yield( '_wheel_kill', 'Killing current run CPAN::Shell loop detected' );
    return;
  }
  undef;
}

sub _detect_loop {
  my $self = shift;
  my $input = shift || return;
  return if $self->{_loop_detect};
  my $digest = md5_hex( $input );
  $self->{_digests}->{ $digest }++;
  return unless ++$self->{_digests}->{ $digest } > 300;
  return $self->{_loop_detect} = 1;
}

sub _wheel_idle {
  my ($kernel,$self) = @_[KERNEL,OBJECT];
  my $now = time();
  if ( $now - $self->{_wheel_time} >= $self->{idle} ) {
    $self->{_current_job}->{idle_kill} = 1;
    $kernel->yield( '_wheel_kill', 'Killing current run due to excessive idle' );
    return;
  }
  if ( $now - $self->{_current_job}->{start_time} >= $self->{timeout} ) {
    $self->{_current_job}->{excess_kill} = 1;
    $kernel->yield( '_wheel_kill', 'Killing current run due to excessive run-time' );
    return;
  }
  $kernel->delay( '_wheel_idle', 60 );
  return;
}

sub _wheel_kill {
  my ($kernel,$self,$reason) = @_[KERNEL,OBJECT,ARG0];
  push @{ $self->{_wheel_log} }, $reason;
  warn $reason, "\n" if $self->{debug};
  if ( $^O eq 'MSWin32' and $self->{wheel} ) {
    my $grp_pid = $self->{_current_job}->{GRP_PID};
    return unless $grp_pid;
    warn Win32::FormatMessage( Win32::GetLastError() )
	unless Win32::Process::KillProcess( $grp_pid, 0 );
  }
  else {
    if ( !$self->{no_grp_kill} ) {
      if ( $^O eq 'solaris' ) {
         kill( 9, '-' . $self->{wheel}->PID() ) if $self->{wheel};
      }
      else {
         $self->{wheel}->kill(-9) if $self->{wheel};
      }
    }
    elsif ( $GOT_KILLFAM ) {
      _kill_family( 9, $self->{wheel}->PID() ) if $self->{wheel};
    }
    else {
      $self->{wheel}->kill(9) if $self->{wheel};
    }
  }
  return;
}

sub _check_yacsmoke {
  my $perl = shift;
  my $cmdline = $perl . q{ -MCPANPLUS::YACSmoke -e 1};
  my $job = Win32::Job->new()
    or die Win32::FormatMessage( Win32::GetLastError() );
  my $pid = $job->spawn( $perl, $cmdline )
    or die Win32::FormatMessage( Win32::GetLastError() );
  warn $pid, "\n";
  my $ok = $job->watch( sub { 0 }, 60 );
  my $hashref = $job->status();
  return $hashref->{$pid}->{exitcode};
}

sub _test_module {
  my $perl = shift;
  my $module = shift;
  my $cmdline = $perl . ' -MCPANPLUS::YACSmoke -e "my $module = shift; my $smoke = CPANPLUS::YACSmoke->new(); $smoke->test($module);" ' . $module;
  my $job = Win32::Job->new()
    or die Win32::FormatMessage( Win32::GetLastError() );
  my $pid = $job->spawn( $perl, $cmdline )
    or die Win32::FormatMessage( Win32::GetLastError() );
  warn $pid, "\n";
  my $ok = $job->watch( sub { 0 }, 60 );
  my $hashref = $job->status();
  return $hashref->{$pid}->{exitcode};
}

sub _flush {
  my $perl = shift;
  my $type = shift;
  my $cmdline = $perl . ' -MCPANPLUS::YACSmoke -e "my $type = shift; my $smoke = CPANPLUS::YACSmoke->new(); $smoke->flush($type) if $smoke->can(q{flush});" ' . $type;
  my $job = Win32::Job->new()
    or die Win32::FormatMessage( Win32::GetLastError() );
  my $pid = $job->spawn( $perl, $cmdline )
    or die Win32::FormatMessage( Win32::GetLastError() );
  warn $pid, "\n";
  my $ok = $job->watch( sub { 0 }, 60 );
  my $hashref = $job->status();
  return $hashref->{$pid}->{exitcode};
}
sub _recent_modules {
  my $perl = shift;
  my $cmdline = $perl . ' -MCPANPLUS::YACSmoke -e "my $smoke = CPANPLUS::YACSmoke->new();print qq{$_\n} for $smoke->{plugin}->download_list();"';
  my $job = Win32::Job->new()
    or die Win32::FormatMessage( Win32::GetLastError() );
  my $pid = $job->spawn( $perl, $cmdline )
    or die Win32::FormatMessage( Win32::GetLastError() );
  warn $pid, "\n";
  my $ok = $job->watch( sub { 0 }, 60 );
  my $hashref = $job->status();
  return $hashref->{$pid}->{exitcode};
}

sub _reload_indices {
  my $perl = shift;
  my $cmdline = $perl . ' -MCPANPLUS::Backend -e "CPANPLUS::Backend->new()->reload_indices( update_source => 1 );"';
  my $job = Win32::Job->new()
    or die Win32::FormatMessage( Win32::GetLastError() );
  my $pid = $job->spawn( $perl, $cmdline )
    or die Win32::FormatMessage( Win32::GetLastError() );
  warn $pid, "\n";
  my $ok = $job->watch( sub { 0 }, 60 );
  my $hashref = $job->status();
  return $hashref->{$pid}->{exitcode};
}

sub _author_search {
  my $perl = shift;
  my $type = shift;
  my $search = shift;
  my $cmdline = $perl . ' -MCPANPLUS::YACSmoke -e "my $type = shift; my $search = shift; my $cb = CPANPLUS::Backend->new(); my %mods = map { $_->package() => 1 } map { $_->modules() } $cb->search( type => $type, allow => [ qr/$search/ ], [ verbose => 0 ] ); print qq{$_\n} for sort keys %mods;" ' . $type . " " . $search;
  my $job = Win32::Job->new()
    or die Win32::FormatMessage( Win32::GetLastError() );
  my $pid = $job->spawn( $perl, $cmdline )
    or die Win32::FormatMessage( Win32::GetLastError() );
  warn $pid, "\n";
  my $ok = $job->watch( sub { 0 }, 60 );
  my $hashref = $job->status();
  return $hashref->{$pid}->{exitcode};
}

sub _package_search {
  my $perl = shift;
  my $type = shift;
  my $search = shift;
  my $cmdline = $perl . ' -MCPANPLUS::YACSmoke -e "my $type = shift; my $search = shift; my $cb = CPANPLUS::Backend->new(); my %mods = map { $_->package() => 1 } $cb->search( type => $type, allow => [ qr/$search/ ], [ verbose => 0 ] ); print qq{$_\n} for sort keys %mods;" ' . $type . " " . $search;
  my $job = Win32::Job->new()
    or die Win32::FormatMessage( Win32::GetLastError() );
  my $pid = $job->spawn( $perl, $cmdline )
    or die Win32::FormatMessage( Win32::GetLastError() );
  warn $pid, "\n";
  my $ok = $job->watch( sub { 0 }, 60 );
  my $hashref = $job->status();
  return $hashref->{$pid}->{exitcode};
}

sub _kill_family {
  my ($signal, @pids) = @_;
  my $pt = Proc::ProcessTable->new;
  my (@procs) =  @{$pt->table};
  my (@kids) = _get_pids( \@procs, @pids );
  @pids = (@pids, @kids);
  kill $signal, reverse @pids;
}

sub _get_pids {
  my($procs, @kids) = @_;
  my @pids;
  foreach my $kid (@kids) {
    foreach my $proc (@$procs) {
      if ($proc->ppid == $kid) {
	my $pid = $proc->pid;
	push @pids, $pid, _get_pids( $procs, $pid );
      }
    }
  }
  @pids;
}

1;
__END__