Cantella-Worker

 view release on metacpan or  search on metacpan

lib/Cantella/Worker/Manager/Prefork.pm  view on Meta::CPAN

        shutdown => '_shutdown',
        spawn_workers => '_spawn_workers',
        retire_worker => '_retire_worker',
        worker_process_error => '_worker_process_error',
        worker_process_close => '_worker_process_close',
        worker_process_stdout => '_worker_process_stdout',
        worker_process_stderr => '_worker_process_stderr',
        worker_process_sig_chld => '_worker_process_sig_chld',
      } ],
    inline_states => {
      _keep_alive => sub {
        $_[KERNEL]->delay(_keep_alive => 1000);
      },
      sig_term => sub {
        my ($kernel) = $_[KERNEL];
        $kernel->yield('shutdown');
        $kernel->sig_handled;
      },
      sig_int => sub {
        my ($kernel) = $_[KERNEL];
        $kernel->yield('shutdown');
        $kernel->sig_handled;

lib/Cantella/Worker/Manager/Prefork.pm  view on Meta::CPAN

sub _start {
  my ($self, $kernel, $heap) = @_[OBJECT, KERNEL, HEAP];
  $self->logger->info("starting worker manager");
  $kernel->alias_set($self->alias);
  $kernel->sig(INT => 'sig_int');
  $kernel->sig(TERM => 'sig_term');
  $kernel->sig(USR1 => 'sig_usr1');
  $kernel->sig(USR2 => 'sig_usr2');
  $heap->{spawn_workers} = 1;

  $kernel->delay(_keep_alive => 1000);
  $kernel->yield('spawn_workers');
}

sub _pause {
  my ($self, $kernel, $heap) = @_[OBJECT, KERNEL, HEAP];
  $self->logger->info("pausing worker manager");
  $heap->{spawn_workers} = 0;
  $self->signal_workers('USR1');
}

lib/Cantella/Worker/Manager/Prefork.pm  view on Meta::CPAN

Defaults to an empty hash ref.

=head2 workers

=over 4

=item B<workers> - accessor

=back

Read-write integer. The maximum number of subprocesses to have alive at any one
time. Defaults to 5.

=head2 max_worker_age

=over 4

=item B<max_worker_age> - accessor

=item B<has_max_worker_age> - predicate

lib/Cantella/Worker/Manager/Prefork.pm  view on Meta::CPAN

=head2 current_worker_count

=over 4

=item B<arguments:> none

=item B<return value:> C<$number_of_live_children>

=back

Returns the number of children workers that are still alive.

=head2 worker_wheels

=over 4

=item B<arguments:> none

=item <return value:> C<@worker_poe_wheel_run_objects>

=back

lib/Cantella/Worker/Manager/Prefork.pm  view on Meta::CPAN

=over 4

=item B<handles event:> C<retire_worker>

=item B<arguments:> C<$wheel_id>

=item B<return value:> none

=back

If the worker process working under $wheel_id is still alive, send a sig TERM

=head2 _worker_process_close

=over 4

=item B<handles event:> C<worker_process_close>

=item B<arguments:> none

=item B<return value:> none

lib/Cantella/Worker/Manager/Prefork.pm  view on Meta::CPAN

=item B<handles event:> C<worker_process_sig_chld>

=item B<arguments:> C<$pid, $exit_status_code>

=item B<return value:> none

=back

Once a sig CHLD comes back from a worker, this event will schedule a
C<spawn_workers> event to make sure there's enough workers alive.

=head2 _worker_process_stdout

=over 4

=item B<handles event:> C<worker_process_stdout>

=item B<arguments:> C<$line>

=item B<return value:> none

lib/Cantella/Worker/Manager/Prefork.pm  view on Meta::CPAN

=back

Remove all alarms, signal the workers to shutdown and wait for the session to die
The shutdown event can be triggered sending signal C<INT> or C<TERM>

=head1 OTHER EVENTS

=over 4

=item B<_keep_alive> - Used to keep the session alive while paused. Does nothing
other than schedule the next keep alive_1000 seconds away.

=item B<sig_int> - mark sig INT as handled and yield to C<shutdown>

=item B<sig_term> - mark sig TERM as handled and yield to C<shutdown>

=item B<sig_usr1> - mark sig USR1 as handled and yield to C<_pause>

=item B<sig_usr2> - mark sig USR2 as handled and yield to C<_resume>

=back

lib/Cantella/Worker/Role/Worker.pm  view on Meta::CPAN

    object_states => [
      $self, {
        poll => '_poll',
        work => '_work',
        _start   => '_start',
        _pause   => '_pause',
        _resume  => '_resume',
        shutdown => '_shutdown',
      } ],
    inline_states => {
      _keep_alive => sub {
        $_[KERNEL]->delay(_keep_alive => 1000);
      },
      sig_term => sub {
        my ($kernel) = $_[KERNEL];
        $kernel->yield('shutdown');
        $kernel->sig_handled;
      },
      sig_int => sub {
        my ($kernel) = $_[KERNEL];
        $kernel->yield('shutdown');
        $kernel->sig_handled;

lib/Cantella/Worker/Role/Worker.pm  view on Meta::CPAN

  $heap->{poll} = 1;
  $kernel->alias_set($self->alias);
  $kernel->yield('poll');
}

sub _pause {
  my ($self, $kernel, $heap) = @_[OBJECT, KERNEL, HEAP];
  $self->logger->info("pausing worker ${$}");
  $heap->{poll} = 0;
  $kernel->delay('poll'); #clear any alarm, if it exists
  $kernel->delay(_keep_alive => 1000); #keep the session alive
}

sub _resume {
  my ($self, $kernel, $heap) = @_[OBJECT, KERNEL, HEAP];
  $self->logger->info("resuming worker ${$}");
  $heap->{poll} = 1;
  $kernel->delay('poll'); #clear any alarm, if it exists
  $kernel->yield('poll');
}

lib/Cantella/Worker/Role/Worker.pm  view on Meta::CPAN

=item B<return value:> none

=back

Remove all alarms and wait for the session to die

=head1 OTHER EVENTS

=over 4

=item B<_keep_alive> - Used to keep the session alive while paused. Does nothing
other than schedule the next keep alive 1000 seconds away.

=item B<sig_int> - mark sig INT as handled and yield to C<shutdown>

=item B<sig_term> - mark sig TERM as handled and yield to C<shutdown>

=item B<sig_usr1> - mark sig USR1 as handled and yield to C<_pause>

=item B<sig_usr2> - mark sig USR2 as handled and yield to C<_resume>

=back