view release on metacpan or  search on metacpan

lib/App/Sv.pm  view on Meta::CPAN

		$warn->("Failed to fork '$svc->{name}': $!");
		$self->_restart_svc($svc);
		return;
	}
	
	if ($pid == 0) {
		# child
		# set egid/euid
		if ($svc->{group}) {
			$svc->{gid} = getgrnam($svc->{group});
			$) = $svc->{gid};
		}
		if ($svc->{user}) {
			$svc->{uid} = getpwnam($svc->{user});
			$> = $svc->{uid};
		}
		# set process umask
		umask oct($svc->{umask}) if $svc->{umask};
		# change working directory
		if ($svc->{cwd}) {
			chdir $svc->{cwd} 
				or $warn->("Failed cwd for '$svc->{name}': $!");
		}
		# set environment
		%ENV = %{$svc->{env}} if $svc->{env} && ref $svc->{env} eq 'HASH';
		# set session id
		if ($svc->{setsid}) {
			$svc->{pgrp} = POSIX::setsid()
				or $warn->("Failed setsid for '$svc->{name}': $!");
		};
		# start process
		if ($svc->{cmd} && !ref $svc->{cmd}) {
			$debug->("Executing command '$svc->{name}'");
			exec($svc->{cmd});
		}
		elsif ($svc->{cmd} && ref $svc->{cmd} eq 'ARRAY') {
			$debug->("Executing command '$svc->{name}'");
			exec(@{$svc->{cmd}});
		}
		elsif ($svc->{code} && ref $svc->{code} eq 'CODE') {
			$debug->("Executing code '$svc->{name}'");
			$svc->{code}->();
		}
		elsif ($svc->{code} && ref $svc->{code} eq 'ARRAY') {
			my $code = shift @{$svc->{code}};
			if (ref $code eq 'CODE') {
				$debug->("Executing code '$svc->{name}'");
				$code->(@{$svc->{code}});
			}
		}
		POSIX::_exit(1);
	}
	else {
		# parent
		$debug->("Watching pid $pid for '$svc->{name}'");
		$svc->{pid} = $pid;
		$svc->{watcher} = AE::child $pid, sub {
			$self->_child_exited($svc, @_);
		};
		$svc->{start_ts} = time;
		my $t; $t = AE::timer $svc->{start_wait}, 0, sub {
			$self->_check_svc_up($svc);
			undef $t;
		};
	}
	
	return $pid;
}

sub _child_exited {
	my ($self, $svc, undef, $status) = @_;
	
	my $debug = $self->{log}->logger(8);
	$debug->("Child $svc->{pid} exited, status $status: '$svc->{name}'");
	delete $svc->{watcher};
	delete $svc->{pid};
	$svc->{last_status} = $status >> 8; 
	if ($svc->{state} eq 'stop') {
		delete $svc->{start_count};
		$svc->{state} = 'down';
	}
	elsif ($svc->{once}) {
		delete $svc->{start_count};
		$svc->{state} = 'fatal';
	}
	else {
		$self->_restart_svc($svc);
	}
}

sub _restart_svc {
	my ($self, $svc) = @_;

	if ($svc->{start_retries}) {
		if ($svc->{start_count} &&
			($svc->{start_count} >= $svc->{start_retries})) {
			$svc->{state} = 'fatal';
			return;
		}
	}
	else {
		$svc->{state} = 'fatal';
		return;
	}
	my $debug = $self->{log}->logger(8);
	$svc->{state} = 'restart';
	$debug->("Restarting '$svc->{name}' in $svc->{restart_delay} seconds");
	my $t; $t = AE::timer $svc->{restart_delay}, 0, sub {
		$self->_start_svc($svc);
		undef $t;
	};
}

sub _check_svc_up {
	my ($self, $svc) = @_;
	
	return unless $svc->{state} eq 'start';
	if (!$svc->{pid}) {
		$svc->{state} = 'fail';
		return;
	}
	delete $svc->{start_count};
	$svc->{state} = 'up';
}

sub _stop_svc {
	my ($self, $svc) = @_;
	
	$svc->{state} = 'stop';
	my $st = $self->_signal_svc($svc, 'TERM');
	if ($svc->{stop_wait} && $svc->{stop_wait} > 0) {
		my $t; $t = AE::timer $svc->{stop_wait}, 0, sub {
			$self->_check_svc_down($svc);
			undef $t;
		};
	}
	
	return $st;
}

sub _check_svc_down {
	my ($self, $svc) = @_;
	
	return unless $svc->{state} eq 'stop';
	if ($svc->{pid}) {
		my $st = $self->_signal_svc($svc, 'KILL');
	}
}

sub _signal_svc {
	my ($self, $svc, $sig) = @_;
	
	return unless ($svc->{pid} && $sig);
	my $debug = $self->{log}->logger(8);
	$debug->("Sent signal $sig to pid $svc->{pid}");
	my $st = kill($sig, $svc->{pid});
	
	return $st;
}

sub _signal_all_svc {
	my ($self, $sig, $cv) = @_;
	
	my $debug = $self->{log}->logger(8);
	$debug->("Received signal $sig");
	my $is_any_alive = 0;
	foreach my $key (keys %{ $self->{run} }) {
		my $svc = $self->{run}->{$key};
		next unless my $pid = $svc->{pid};
		$debug->("... sent signal $sig to pid $pid");
		$is_any_alive++;
		kill($sig, $pid);
	}

	return if $cv and $is_any_alive;

	$debug->('Exiting...');
	$cv->send if $cv;
}

# Contolling socket
sub _listener {
	my $self = shift;
	
	my $debug = $self->{log}->logger(8);
	my ($host, $port) = parse_hostport($self->{conf}->{listen});
	croak "Socket \'$port\' already in use" if ($host eq 'unix/' && -e $port);
	
	$self->{server} = tcp_server $host, $port,
	sub { $self->_client_conn(@_) },
	sub {
		my ($fh, $host, $port) = @_;