File Coverage

blib/lib/Mojo/Run.pm

Criterion	Covered	Total	%
statement	53	285	18.6
branch	0	124	0.0
condition	0	72	0.0
subroutine	18	40	45.0
pod	1	11	9.0
total	72	532	13.5

package Mojo::Run;

use Mojo::Base -base;

use bytes;
use Carp;
use Errno;
use Socket;
use Time::HiRes qw(time gettimeofday);
use Scalar::Util qw(blessed);
use Storable qw(thaw nfreeze);
use POSIX ":sys_wait_h";
use Mojo::Log;
use Mojo::IOLoop;
use Data::Dumper;
use Mojo::Reactor;
has 'num_forks'  => sub { 0 };
has 'max_forks'  => sub { 0 };
has 'log'        => sub { Mojo::Log   ->new };
has 'ioloop'     => sub { Mojo::IOLoop->new };
has [qw/reactor error is_child/];

our $VERSION = '0.3';

my $_obj  = undef;

BEGIN {
	*portable_pipe = sub () { my ($r, $w);
		pipe $r, $w or return;

		($r, $w);
	};
	*portable_socketpair = sub () {
		socketpair my $fh1, my $fh2, Socket::AF_UNIX(), Socket::SOCK_STREAM(), PF_UNSPEC
			or return;
		$fh1->autoflush(1);
		$fh2->autoflush(1);

		($fh1, $fh2)
	};
}

sub new { __PACKAGE__->singleton }

sub singleton {
	return $_obj if defined $_obj;
	return $_obj = __PACKAGE__->_constructor;
}

sub _constructor {
	my $proto = shift;
	my $class = ref($proto) || $proto;
	my $self  = $class->SUPER::new;

	bless $self => $class;

	# install SIGCHLD handler
	$SIG{'CHLD'} = sub { _sig_chld($self, @_) };

	return $self;
}

sub log_level {
	my ($self, $level) = @_;

	$self->log->level($level) if defined $level;

	return $self->log->level;
}

sub spawn {
	my ($self, %opt) = @_;

	unless (defined $self && blessed($self) && $self->isa(__PACKAGE__)) {
		my $obj = __PACKAGE__->new;
		return $obj->spawn(%opt);
	}

	$self->error('');

	if ($self->max_forks > 0 && $self->num_forks >= $self->max_forks) {
		$self->error("Unable to spawn another subprocess: "
			."Limit of " . $self->max_forks . " concurrently spawned process(es) is reached."
		);
		return 0;
	}

	# normalize and validate run parameters...
	my $proc = $self->_getRunStruct(\%opt);
	return 0 unless $self->_validateRunStruct($proc);

	$self->log->debug("Spawning command "
		."timeout: "
		.($proc->{exec_timeout} > 0 ? sprintf("%-.3f seconds]", $proc->{exec_timeout}) : "none")
		." : [$proc->{cmd}]"
	);
	my ($stdout_p, $stdout_c) = portable_socketpair;
	my ($stderr_p, $stderr_c) = portable_socketpair;
	my ($stdres_p, $stdres_c) = portable_socketpair;

	$proc->{time_started} = time;
	$proc->{running     } = 1;
	$proc->{hdr_stdout  } = $stdout_c;
	$proc->{hdr_stderr  } = $stderr_c;
	$proc->{hdr_stdres  } = $stdres_c;

	my $pid = fork;

	if ($pid) {
		# parent
		$self->num_forks($self->num_forks + 1);

		$self->log->debug("Subprocess spawned as pid $pid.");

		$proc->{pid} = $pid;

		# exec timeout
		if (defined $proc->{exec_timeout} && $proc->{exec_timeout} > 0) {
			$self->log->debug(
				"[process $pid]: Setting execution timeout to " .
				sprintf("%-.3f seconds.", $proc->{exec_timeout})
			);
			my $timer = $self->ioloop->timer(
				$proc->{exec_timeout},
				sub { _timeout_cb($self, $pid) }
			);

			# save timer
			$proc->{id_timeout} = $timer;
		}

		$self->{_data}->{$pid} = $proc;

		close $stdout_p;
		close $stderr_p;
		close $stdres_p;

		$self->watch('stdout', $pid);
		$self->watch('stderr', $pid);
		$self->watch('stdres', $pid);
	} else {
		# child

		$self->is_child(1);

		close $stdout_c;
		close $stderr_c;
		close $stdres_c;

		# Stdio should not be tied.
		if (tied *STDOUT) {
			carp "Cannot redirect into tied STDOUT.  Untying it";
			untie *STDOUT;
		}
		if (tied *STDERR) {
			carp "Cannot redirect into tied STDERR.  Untying it";
			untie *STDERR;
		}

		# Redirect STDOUT
		open STDOUT, ">&" . fileno($stdout_p)
			or croak "can't redirect stdout in child pid $$: $!";
		# Redirect STDERR
		open STDERR, ">&" . fileno($stderr_p)
			or croak "can't redirect stderr in child pid $$: $!";

		select STDERR; $| = 1;
		select STDOUT; $| = 1;

		if (ref $proc->{cmd} eq 'CODE') {
			my @rv = eval { $proc->{cmd}->($$, $proc->{param}); };

			if ($@) {
				carp "exec of coderef failed: $@\n";
				exit 255;
			}

			print $stdres_p nfreeze(\ @rv);

		} else {
			exec(ref $proc->{cmd} eq 'ARRAY' ? @{ $proc->{cmd} } : $proc->{cmd}) or do {
				carp "exec failed";
				exit 255;
			};
		}

		close $stdout_p;
		close $stderr_p;
		close $stdres_p;

		exit 1;
	}

	return $pid;
}

sub start { shift->ioloop->start }

sub watch {
	my $self = shift;
	my $io   = lc(shift || '');
	my $pid  = shift;

	my $proc = $self->get_proc($pid);

	$self->log->error('Cant start IO watcher off NULL process'           ) and return unless $proc;
	$self->log->error("[process $proc->{pid}]: IO ($io) is unsupported"  ) and return unless $io ~~ [qw/stdout stderr stdres/];
	$self->log->error("[process $proc->{pid}]: IO handler ($io) is EMPTY") and return unless $proc->{"hdr_$io"};

	my $id = fileno $proc->{"hdr_$io"};

	$self->ioloop->reactor->io($proc->{"hdr_$io"}, sub {
		my $chunk = undef;
		my $len   = sysread $proc->{"hdr_$io"}, $chunk, 65536;

		return unless defined $len or $! != Errno::EINTR;

		if (!$len) {
			$self->drop_handle($pid, $io);
			return;
		}

		if (defined $proc->{"$io\_cb"}) {
			$self->log->debug("[process $proc->{pid}]: (handle: $id) Invoking ".uc($io)." callback.");

			eval { $proc->{"$io\_cb"}->($proc->{pid}, $chunk) };

			if ($@) {
				$self->log->error("[process $proc->{pid}]: (handle: $id) Exception in $io\_cb: $@");
			}
		} else {
			# append to buffer
			$self->log->debug("[process $proc->{pid}]: (handle: $id) Appending $len bytes to ".uc($io)." buffer.");
			$proc->{"buf_$io"} .= $chunk;
		}
	})->watch($proc->{"hdr_$io"}, 1, 0);
}

sub drop_handle {
	my $self = shift;
	my $pid  = shift;
	my $io   = lc(shift || '');

	my $proc = $self->get_proc($pid);
	return unless $proc;

	$self->log->debug("[process $pid]: Got HUP for unmanaged handle ".$proc->{"hdr_$io"}."; ignoring.") and return
		unless $proc->{"hdr_$io"};


	$self->ioloop->remove( $proc->{"hdr_$io"} );
	undef $proc->{"hdr_$io"};

	$self->log->debug("[process $pid]: ".uc($io)." closed.");

	$self->complete($pid);
}

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

	no warnings;
	my $err = "[process $pid]: Unable to get process data structure: ";

	unless (defined $pid) {
		$self->error($err . "Undefined pid.");
		return undef;
	}

	unless (
		exists $self->{_data}->{$pid}
		&& defined $self->{_data}->{$pid}
	) {
		$self->error($err . "Non-managed process pid: $pid");
		return undef;
	}

	return $self->{_data}->{$pid};
}

sub cleanup {
	my ($self, $pid, $exit_val, $signum, $core) = @_;

	my $proc = $self->get_proc($pid);
	unless (defined $proc) {
		no warnings;
		$self->log->warn("Untracked process pid $pid exited with exit status $exit_val by signal $signum, core: $core.");
		return 0;
	}
	return 0 if $proc->{cleanup};

	$proc->{cleanup} = 1;

	$self->log->debug("[process $pid]: Got SIGCHLD, "
		. "exited with exit status: $exit_val by signal $signum"
		. (($core) ? "with core dump." : ".")
	);

	if (defined $proc->{id_timeout}) {
		$self->ioloop->remove($proc->{id_timeout});
		$proc->{id_timeout} = undef;
	}
	if ($proc->{hard_kill}) {
		for (qw/stderr stdout stdres/) {
			$self->drop_handle($pid, $_) if $proc->{"hdr_$_"};
		}
	}
	$proc->{exit_status} = $exit_val;
	$proc->{exit_core  } = $core;
	$proc->{exit_signal} = $signum;

	# command timings...
	my $te = time;
	$proc->{time_stopped      } = $te;
	$proc->{time_duration_exec} = $te - $proc->{time_started};

	# this process is no longer running
	$proc->{running} = 0;

	$self->complete($pid);
}

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

	my $proc = $self->get_proc($pid);

	return 0 if !$force
		&& (
			$proc->{running}
			|| defined $proc->{hdr_stdout}
			|| defined $proc->{hdr_stdres}
			|| defined $proc->{hdr_stderr}
		);


	if ($proc && %$proc) {
		$self->log->debug("[process $pid]: All streams closed, process execution complete.");

		$proc->{time_duration_total} = time - $proc->{time_started};

		# fire exit callback!
		if (defined $proc->{exit_cb} && ref $proc->{exit_cb} eq 'CODE') {
			my $result = eval { $proc->{buf_stdres} ? thaw($proc->{buf_stdres}) : undef};

			if ($@) {
				croak "Error de-serializing subprocess data: $@";
			}

			# prepare callback structure
			my $cb_d = {
				cmd => ref $proc->{cmd} eq 'CODE'  ? 'CODE'                     :
					   ref $proc->{cmd} eq 'ARRAY' ? join(' ', @{$proc->{cmd}}) :
					   $proc->{cmd}
				,
				param               => $proc->{param},
				exit_status         => $proc->{exit_status},
				exit_signal         => $proc->{exit_signal},
				exit_core           => $proc->{exit_core},
				stdout              => $proc->{buf_stdout},
				stderr              => ($proc->{buf_stderr} ? $proc->{buf_stderr} : '').($proc->{stderr} ? $proc->{stderr} : ''),
				result              => $result,
				time_started        => $proc->{time_started},
				time_stopped        => $proc->{time_stopped},
				time_duration_exec  => $proc->{time_duration_exec},
				time_duration_total => $proc->{time_duration_total},
			};

			# safely invoke callback
			$self->log->debug("[process $pid]: invoking exit_cb callback.");
			eval { $proc->{exit_cb}->($pid, $cb_d); };

			$self->log->error("[process $pid]: Error running exit_cb: $@") if $@;
		} else {
			$self->log->error("[process $pid]: No exit_cb callback!");
		}
	}

	delete $self->{_data}->{$pid};
	$self->num_forks($self->num_forks - 1);
}

sub _sig_chld {
	my ($self) = @_;

	no strict 'subs';

	my $i = 0;
	while ((my $pid = waitpid(-1, WNOHANG)) > 0) {
		$i++;
		my $exit_val = $? >> 8;
		my $signum   = $? & 127;
		my $core     = $? & 128;

		# do process cleanup
		$self->cleanup($pid, $exit_val, $signum, $core);
	}

	$self->log->debug("SIGCHLD handler cleaned up after $i process(es).")
	  if $i > 0;
}

sub _getRunStruct {
	my ($self, $opt) = @_;

	my $s = {
		pid          => 0,
		cmd          => undef,
		param        => undef,
		error        => undef,
		stdout_cb    => undef,
		stderr_cb    => undef,
		exit_cb      => undef,
		exec_timeout => 0,
		buf_stdout   => '',
		buf_stderr   => '',
		buf_stdres   => '',
		hdr_stdout   => undef,
		hdr_stderr   => undef,
		hdr_stdres   => undef,
	};

	# apply user defined vars...
	$s->{$_} = $opt->{$_}
		for grep { exists $s->{$_} } keys %$opt;

	return $s;
}

sub _validateRunStruct {
	my ($self, $s) = @_;

	# command?
	$self->error('Undefined command.') and return
		unless defined $s->{cmd};

	# check command...
	my $cmd_ref = ref $s->{cmd};
	$self->error('Zero-length command.') and return
		if $cmd_ref eq '' && length $s->{cmd} == 0;

	$self->error('Command can be pure scalar, arrayref or coderef.') and return
		if $cmd_ref ne '' && not $cmd_ref ~~ ['CODE', 'ARRAY'];

	# callbacks...
	$self->error("STDOUT callback defined, but is not code reference.") and return
		if defined $s->{stdout_cb} && ref $s->{stdout_cb} ne 'CODE';

	$self->error("STDERR callback defined, but is not code reference.") and return
		if defined $s->{stderr_cb} && ref $s->{stderr_cb} ne 'CODE';

	$self->error("Process exit_cb callback defined, but is not code reference.") and return
		if defined $s->{exit_cb} && ref($s->{exit_cb}) ne 'CODE';

	# exec timeout
	{ no warnings; $s->{exec_timeout} += 0; }

	return 1;
}

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

	my $proc = $self->get_proc($pid);
	return 0 unless $proc;

	# drop timer (can't hurt...)
	if (defined $proc->{id_timeout}) {
		$self->ioloop->remove($proc->{id_timeout});
		$proc->{id_timeout} = undef;
	}

	# is process still alive?
	return 0 unless kill 0, $pid;

	$self->log->debug("[process $pid]: Execution timeout ("
		.sprintf("%-.3f seconds).", $proc->{exec_timeout})
		." Killing process.");

	$proc->{stderr} .= ";Execution timeout.";

	# kill the motherfucker!

	unless (CORE::kill(9, $pid)) {
		$self->log->warn("[process $pid]: Unable to kill process: $!");
	}
	$proc->{hard_kill} = 1;
	$self->cleanup($pid, 0, 9, 0);

	return 1;
}

sub kill {
	my ($self, $pid, $signal) = @_;
	$signal = 15 unless defined $signal;

	my $proc = $self->get_proc($pid);
	return 0 unless $proc;

	# kill the process...
	unless (kill($signal, $pid)) {
		$self->error("Unable to send signal $signal to process $pid: $!");
		return 0;
	}

	return 1;
}

sub DESTROY {
	my ($self) = @_;

	# perform cleanup...
	unless ($self->is_child) {
		foreach my $pid (keys %{$self->{_data}}) {
			my $proc = $self->{_data}->{$pid};

			$self->log->debug("Killing subprocess $pid with SIGKILL") if $self->log;
			# kill process (HARD!)
			$self->kill($pid, 9);

			next unless defined $self->ioloop;

			# drop fds
			$self->drop_handle($pid, $_) for grep {$proc->{"hdr_$_"}} qw/stdout stderr stdres/;

			# fire exit callbacks (if any)
			$self->complete($pid, 1);
		}
	}

	# disable sigchld hander
	$SIG{'CHLD'} = 'IGNORE';
}

1;

=pod

=head1 NAME

Mojo::Run - asynchronous external command execution for Mojo

=head1 VERSION

version 0.3

=head1 SYNOPSIS

	use Mojo::Run;
	use Mojo::Log;

	my $run = Mojo::Run->new;
	$run->max_forks(10);
	$run->log(Mojo::Log->new(
	    level => 'error',
	    path  => 'log/mojo_run.log',
	));

	$run->spawn(
	    cmd => sub {
	        my $pid   = shift;
	        my $param = shift; # {a => 1, b => 2}

	        my $data = {};
	        ... do something
	        return $data;
	    },
	    param => {a => 1, b => 2},
	    exec_timeout => 120, # sec
	    stdout_cb => sub {
	        my ($pid, $chunk) = @_;
	    },
	    stderr_cb => sub {
	        my ($pid, $chunk) = @_;
	    },
	    exit_cb => sub {
	        my $pid = shift;
	        my $res = shift;
	        warn $res->{result}->[0];
	    },
	);
	$run->spawn(
	    cmd => 'ps aux',
	    exit_cb => sub {
	        my $pid = shift;
	        my $res = shift;
	    },
	);
	$run->spawn(
	    cmd => ['perl', '-v'],
	    exit_cb => sub {
	        my $pid = shift;
	        my $res = shift;
	    },
	);

	$run->start;

=head1 Result

Result in B< exit_cb > is a HASH with following keys:

=over

=item B< cmd >

=item B< param >

=item B< exit_status >

=item B< exit_signal >

=item B< exit_core >

=item B< stdout >

=item B< stderr >

=item B< result >

=item B< time_started >

=item B< time_stopped >

=item B< time_duration_exec >

=item B< time_duration_total >

=back

=head1 SOURCE REPOSITORY

L

=head1 AUTHOR

Alexey Likhatskiy,

=head1 LICENSE AND COPYRIGHT

Copyright (C) 2012-2013 "Alexey Likhatskiy"

This is free software; you can redistribute it and/or modify it under the same terms as the Perl 5 programming language system itself.