File Coverage

blib/lib/GearmanX/Starter.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 GearmanX::Starter;

use strict;
use warnings;

use Gearman::XS qw(:constants);
use Gearman::XS::Worker;
use Perl::Unsafe::Signals;
use POSIX;

our $VERSION = '0.05';

our $WORKER;

our $QUIT;

sub new {
  my $class = shift;

  bless {}, $class;
}

sub start {
  my $self = shift;

  my $args = shift;

  my $worker_name = $args->{name} || die "Need name for worker";

  my $init_func = $args->{init_func};

  my $logger = $args->{logger};

  my $servers = $args->{servers} || [[]];

  my $sigterm = $args->{sigterm} || [ 'TERM' ];
  my $sleep = $args->{sleep_and_retry} || 0;

  $logger->info("Forking daemon for $worker_name") if $logger;

  _Init() and return 1;

  my $critical;
  for my $sig (@$sigterm) {
    $SIG{$sig} = sub {
      $QUIT++;
      die "GearmanXQuitLoop\n" if !$critical
    };
  }
  $init_func->() if $init_func;

  $logger->info("Creating $worker_name worker") if $logger;
  $0 = $worker_name;

  $WORKER = $args->{worker} || Gearman::XS::Worker->new;

  for my $server (@$servers) {
    if ($WORKER->add_server(@$server) != GEARMAN_SUCCESS) {
      $logger->logdie("Unable to add job server [@$server] to worker $worker_name: " . $WORKER->error)
        if $logger;
    }
  }

  # Wrap each function in another function that flags
  # that we're in a critical section of code
  $logger->info("Adding functions to $worker_name worker") if $logger;
  my $func_list = $args->{func_list} || [];
  if ( !exists($args->{dereg_func}) || $args->{dereg_func} ) {
    my $dereg_fn_name = $args->{dereg_func} || "dereg:$$";
    $dereg_fn_name =~ s/%PID%/$$/;
    push @$func_list, [ $dereg_fn_name, \&_unregister ];
  }
  for my $fun (@$func_list) {
    my ($name, $f, $dont_wrap, $options) = @$fun;
    my $wrapper = $dont_wrap ? $f : sub { $critical = 1; goto $f };
    my $ret2 = $WORKER->add_function($name, 0, $wrapper, $options);
    if ($ret2 != GEARMAN_SUCCESS) {
      $logger->logdie("Failed to register callback function ($name) for worker $worker_name:" . $WORKER->error)
        if $logger;
    }
  }

  $logger->info("Starting $worker_name loop") if $logger;
  my $error_method = $sleep ? 'logwarn' : 'logdie';
  while (1) {
    my $res = eval {
      $critical = 0;
      my $ret;
      UNSAFE_SIGNALS { $ret = $WORKER->work };
      if ($ret != GEARMAN_SUCCESS) {
        $logger->$error_method('Failed to initiate waiting for a job: '. $WORKER->error)
          if $logger;
        sleep $sleep;
      }
      1;
    };
    if ( !$res && $@ !~ /GearmanXQuitLoop/ ) {
      $logger->logdie("Error running loop for worker $worker_name [$@]:".$WORKER->error)
        if $logger;
    }
    last if $QUIT;
  }
  $logger->info("Exiting $worker_name")
    if $logger;
  exit 0;
}

# Daemon code co-opted from Proc::Daemon
sub _Fork {
  my $self = shift;

  my $pid;
  if (defined($pid = fork)) {
    return $pid;
  } else {
    die "Can't fork: $!";
  }
}

sub _OpenMax {
    my $openmax = POSIX::sysconf( &POSIX::_SC_OPEN_MAX );
    (!defined($openmax) || $openmax < 0) ? 64 : $openmax;
}

# Daemonize this process
sub _Init {
  my $sess_id;

  _Fork() and return 1;

  die "Unable to detach from controlling terminal"
    unless $sess_id = POSIX::setsid();

  $SIG{'HUP'} = 'IGNORE';

  _Fork() and exit 0;

  ## Change working directory
  chdir "/";
  ## Clear file creation mask
  umask 0;
  ## Close open file descriptors
  foreach my $i (0 .. _OpenMax()) { POSIX::close($i); }

  ## Reopen stderr, stdout, stdin to /dev/null
  open(STDIN,  "+>/dev/null");
  open(STDOUT, "+>&STDIN");
  open(STDERR, "+>&STDIN");

  return 0;
}

sub _unregister {
  my $job = shift;

  $WORKER->unregister($job->workload);

  return "1";
}

1;

__END__