File Coverage

blib/lib/POE/Component/JobQueue.pm

Criterion	Covered	Total	%
statement	109	123	88.6
branch	31	58	53.4
condition	19	44	43.1
subroutine	14	16	87.5
pod	0	9	0.0
total	173	250	69.2

# $Id: JobQueue.pm 29 2009-07-28 06:33:37Z rcaputo $
# License and documentation are after __END__.

package POE::Component::JobQueue;

use strict;

use vars qw($VERSION);
$VERSION = '0.571';

use Carp qw (croak);

use POE::Session;

sub DEBUG () { 0 };

# Spawn a new PoCo::JobQueue session.  This basically is a
# constructor, but it isn't named "new" because it doesn't create a
# usable object.  Instead, it spawns the object off as a session.

sub spawn {
  my $type = shift;

  croak "$type requires an even number of parameters" if @_ % 2;

  my %params = @_;

  ### Parameters that are common to both types of job queue.

  my $alias = delete $params{Alias};
  $alias = 'queuer' unless defined $alias and length $alias;

  my $worker = delete $params{Worker};
  croak "$type requires a coderef Worker parameter"
    unless defined $worker and ref($worker) eq 'CODE';

  my $worker_limit = delete $params{WorkerLimit};
  $worker_limit = 8 unless defined $worker_limit and $worker_limit > 0;

  croak "$type requires either an Active or a Passive parameter block"
    unless defined($params{Active}) xor defined($params{Passive});

  ### Parameters and states that are common to both types of queue.

  my @args    = ( $alias, $worker_limit, $worker );
  my %states  =
    ( _child  => \&poco_jobqueue_both_child,
      stop    => \&poco_jobqueue_both_stop,
      _stop   => sub {},
    );

  ### Modal parameters and states go here.

  # Set up for an active queue.
  if (exists $params{Active}) {
    my $active = delete $params{Active};

    my $poll_interval = delete $active->{PollInterval};
    $poll_interval = undef
      unless defined $poll_interval and $poll_interval > 0;

    my $ack_alias = delete $active->{AckAlias};
    $ack_alias = undef unless defined $ack_alias and length $ack_alias;

    my $ack_state = delete $active->{AckState};
    $ack_state = undef unless defined $ack_state and length $ack_state;

    croak "$type must have neither or both AckAlias and AckState"
      if defined($ack_alias) xor defined($ack_state);

    $states{_start}  = \&poco_jobqueue_active_start;
    $states{dequeue} = \&poco_jobqueue_active_dequeue;

    push @args, $poll_interval, $ack_alias, $ack_state;
  }

  # Set up for a passive queue.
  elsif (exists $params{Passive}) {
    my $passive = delete $params{Passive};

    my $prioritizer = delete $passive->{Prioritizer};
    $prioritizer = sub { 1 } unless defined $prioritizer;

    croak( "$type doesn't know these Passive parameters: ",
           join(', ', sort keys %$passive)
         ) if scalar keys %$passive;

    $states{_start}  = \&poco_jobqueue_passive_start;
    $states{dequeue} = \&poco_jobqueue_passive_dequeue;
    $states{enqueue} = \&poco_jobqueue_passive_enqueue;

    push @args, $prioritizer;
  }

  croak( "$type doesn't know these parameters: ",
         join(', ', sort keys %params)
       ) if scalar keys %params;

  # Spawn whichever queue we've built.
  POE::Session->create
    ( inline_states => \%states,
      args          => \@args,
    );

  undef;
}

# Helper function for active job queues.

sub poco_jobqueue_active_meta_postback {
  die "unimplemented bit";
}

# Start an active job queue.  This type of queue polls for new jobs.

sub poco_jobqueue_active_start {
  my ( $kernel, $heap,
       $alias, $worker_limit, $worker_ref,
       $poll_interval, $ack_alias, $ack_state
     ) = @_[KERNEL, HEAP, ARG0..ARG5];

  # Common parameters.
  $heap->{alias}         = $alias;
  $heap->{worker_limit}  = $worker_limit;
  $heap->{worker_ref}    = $worker_ref;

  # Active queue parameters.
  $heap->{poll_interval} = $poll_interval;
  $heap->{meta_postback} =
    sub {
      my @job = @_;
      my $session = $kernel->alias_resolve( $ack_alias );
      return $session->postback( $ack_state, @job ) if defined $session;
      return sub { 1 };
    };

  # State variables.  Pending polls starts at 1 because we're going to
  # fake an initial poll to get things started.
  $heap->{worker_count}  = 0;
  $heap->{pending_polls} = 0;
  $heap->{latest_worker} = 0;

  # Register an alias.
  $kernel->alias_set($alias);

  # Start an initial set of workers.
  $kernel->yield( 'dequeue' );
}

# Start a passive job queue.  This type of queue waits for something
# else to enqueue jobs.

sub poco_jobqueue_passive_start {
  my ( $kernel, $heap,
       $alias, $worker_limit, $worker_ref,
       $prioritizer
     ) = @_[KERNEL, HEAP, ARG0..ARG3];

  # Common parameters.
  $heap->{alias}         = $alias;
  $heap->{worker_limit}  = $worker_limit;
  $heap->{worker_ref}    = $worker_ref;

  # Active queue parameters.
  $heap->{prioritizer}   = $prioritizer;

  # State variables.
  $heap->{worker_count}  = 0;
  $heap->{job_queue}     = [ ];

  # Register an alias.
  $kernel->alias_set($alias);
}

# A worker either has come or gone.  Track the number of running
# workers, and spawn new ones if appropriate.

sub poco_jobqueue_both_child {
  my ($kernel, $heap, $operation) = @_[KERNEL, HEAP, ARG0];

  # A worker has begun its job.  Count it so we know how many exist.

  if ($operation eq 'gain' or $operation eq 'create') {
    DEBUG and warn "JQ: job queue $heap->{alias} got a new worker";
    $heap->{worker_count}++;
  }

  # A worker has finished.  Decrement our worker count, and try to
  # start another worker to take its place.

  else {
    DEBUG and warn "JQ: job queue $heap->{alias} lost a worker";
    warn( "worker count ($heap->{worker_count}) exceeded the limit (",
          $heap->{worker_limit}, ")"
        ) if $heap->{worker_count} > $heap->{worker_limit};
    $heap->{worker_count}--;
    $kernel->yield('dequeue') unless (
      $heap->{latest_worker} or $heap->{stopped}
    );
  }
}

# Remove the alias, stop active polling and delete outstanding job queue

sub poco_jobqueue_both_stop {
  my ($kernel, $heap) = @_[KERNEL, HEAP];

  $kernel->alias_remove($heap->{alias});
  $kernel->alarm_remove_all();

  delete $heap->{pollinterval} if ($heap->{pollinterval});
  delete $heap->{job_queue} if ($heap->{job_queue});

  $heap->{stopped} = 1;
}

# Attempt to fill empty worker slots.

# This is a token for ARG0 that signifies this was a timed event.
sub TIMED () { 31415 }

sub poco_jobqueue_active_dequeue {
  my ($kernel, $heap, $is_timed) = @_[KERNEL, HEAP, ARG0];

  # If this is a poll from a timed event, then decrement the pending
  # polls count.  The pending polls count is just to ensure that
  # redundant delays are not set, because each redundant delay would
  # force the existing one forward in time.  They would delay polling
  # past the hard polling interval, which would probably be bad (and
  # could stave off polling indefinitely in some instances).  I think
  # this is a bit of a hack, and something better should replace it.

  if (defined $is_timed and $is_timed == TIMED) {
    # Decrement the number of pending polls.  There can be only one,
    # so throw in a die for assertion testing.
    die "pending polls should now be zero (not $heap->{pending_polls})"
      if --$heap->{pending_polls};
  }

  # Attempt to fill the empty worker slots.
  while (
    not $heap->{stopped}
    and $heap->{worker_count} < $heap->{worker_limit}
  ) {

    # Call the worker to fetch a new job and spawn a session.
    my $previous_worker_count = $heap->{worker_count};
    $heap->{worker_ref}->( $heap->{meta_postback} );

    # If the worker count hasn't changed, then we've run out of jobs.
    # Begin polling, if applicable, and exit the spawn loop.
    if ($heap->{worker_count} == $previous_worker_count) {
      if (defined $heap->{poll_interval} and !$heap->{pending_polls}) {
        $heap->{pending_polls}++;
        $kernel->delay( dequeue => $heap->{poll_interval} => TIMED );
      }
      $heap->{latest_worker}++ unless defined $heap->{poll_interval};
      last;
    }
  }
}

# Attempt to fill empty worker slots.

sub poco_jobqueue_passive_dequeue {
  my ($kernel, $heap) = @_[KERNEL, HEAP];

  # Attempt to fill the empty worker slots.
  while (
    not $heap->{stopped}
    and $heap->{worker_count} < $heap->{worker_limit}
  ) {

    # Try to fetch another job from the queue.
    my $next_job = shift @{ $heap->{job_queue} };
    last unless defined $next_job;

    DEBUG and
      warn "JQ: job queue $heap->{alias} is starting a new worker";

    # Start a new session with the job.
    $heap->{worker_ref}->( @$next_job );
  }

  # Avoid accidentally returning something.
  undef;
}

# Enqueue a job in a passive queue.

sub poco_jobqueue_passive_enqueue {
  my ($kernel, $sender, $heap, $return_state, @job) =
    @_[KERNEL, SENDER, HEAP, ARG0..$#_];

  if ($heap->{stopped}) {
    DEBUG and warn(
      "JQ: $heap->{alias} can not enqueue new jobs after 'stop'\n"
    );

    return;
  }

  DEBUG and warn "JQ: job queue $heap->{alias} enqueuing a new job";

  my $postback;
  if (defined $return_state) {
    $postback = $sender->postback( $return_state, @job );
  }

  # Add the job to the queue.  Use the prioritizer to find the right
  # place to put it.

  my $queue_index = @{ $heap->{job_queue} };
  while ($queue_index--) {
    last if
      $heap->{prioritizer}->( $heap->{job_queue}->[$queue_index],
                              \@job,
                            ) >= 0;
  }

  # Place the new job after the index we found.
  splice( @{$heap->{job_queue}}, $queue_index+1, 0, [ $postback, @job ] );

  # Dequeue a new event.
  $kernel->yield( 'dequeue' );
}

1;

__END__