POE & JobQueue performance issues

poe list <[email protected]> Thu, 4 Sep 2008 18:53:58 -0400

Newsgroups	gmane.comp.lang.perl.poe
Message-ID	<[email protected]>

Hi,

I have developed a POE/JobQueue script that listens for messages off a message queue 
and then executes one of two programs according to the message along with a 
re-encapsulation of the message.  The programs that are run are relatively long
running.  Each can take about 5-10 seconds to run.  When I run either of the programs
in the worker directly, then the performance is really poor.  When I enable the debug
in JobQueue, I see that workers are created relatively slowly.  On the other hand,
if I execute the programs (system) w/ a & request ( system ( CMD & ), then the 
workers are created much faster and the message queue is drained very quickly.

The performance difference is approximately 2 minutes to handle 100 messages when 
no spawn is requested in the system call.  On the other hand, it runs 100 messages
in about 20 seconds when I spawn off the programs.

I am attaching the main script.  To effect the behavior change, I simply change the 
system execution in the CALL subroutine.  Any ideas why the performance is so poor
when the script doesn't spawn off the calls?  I have tested it with 10, 50 and 100
workers.  It makes no difference really.

Thanks,

John

pagerQ.pl (text/plain, 6.6 KB)

#!/usr/local/bin/perl -w

use strict;

sub POE::Kernel::ASSERT_DEFAULT () { 1 }
use POE;
use POE::Component::JobQueue;
use lib '/opt/MsgSys/lib';
use WheelMsgQ;
use Getopt::Std;
use Sys::Syslog qw (setlogsock openlog syslog closelog);

use vars qw ( $opt_q $opt_w $opt_d $opt_h );


$| = 1;
my $MSGQID = 100;
my $MAXWORKERS = 10;
my $DEBUGLEVEL = 0;

# Paging scripts
my $PAGEHOME = '/opt/MsgSys';
my $MESSENGER = "$PAGEHOME/message.pl";
my $DIRECT = "$PAGEHOME/sendDirect.pl";


# Defined for standard support in Getopts::Std
sub VERSION_MESSAGE {}
sub HELP_MESSAGE {
  print STDERR "$0 [ -q QUEUEID ] [ -w WORKERS ] [ -d DEBUGLEVEL ] | -h\n";
  print STDERR "Where: -q QUEUEID indicates the Message Queue ID to use\n";
  print STDERR "       -w WORKERS indicates the maximum workers to spawn\n";
  print STDERR "       -d DEBUGLEVEL identifies the debug level\n";
  print STDERR "       -h displays this usage\n";
  exit;
}

if ( ! getopts('d:w:q:h') || defined $opt_h) {
  HELP_MESSAGE();
}
$MSGQID = $opt_q if (defined $opt_q);
$MAXWORKERS = $opt_w if (defined $opt_w);
$DEBUGLEVEL = $opt_d if (defined $opt_d);

setlogsock ('unix');
openlog ('pagerQ', 'pid', 'daemon');

sub DEBUG {
  my ($debug, @args) = @_;
  return if ($debug > $DEBUGLEVEL);

  syslog ('notice', "%s", join (' ', @args))
}

DEBUG (0, "MSQID: $MSGQID");
DEBUG (0, "MAXWORKERS: $MAXWORKERS");
DEBUG (0, "DEBUGLEVEL: $DEBUGLEVEL");

sub CALL {
  DEBUG (3, @_);
  system (join (' ', @_, '&'));
#  system (@_);
}

POE::Session->create
  ( inline_states =>
      { _start => sub { $_[HEAP]->{wheel} =
              WheelMsgQ->new (
                MSGQID   => $_[ARG0],
                InputEvent => 'got_line',
                ErrorEvent => 'got_error',
              );
            $_[HEAP]->{first} = 0;
        },
        got_line => sub { 
            $_[KERNEL]->post( passive => enqueue => response => $_[ARG0] );
        },
        got_error => sub { DEBUG (0, "got_error: $_[ARG0]\n"); },
      },
    args => [$MSGQID],
  );
#------------------------------------------------------------------------------

sub worker_start {
  my ($kernel, $heap, $postback, $task) =
    @_[KERNEL, HEAP, ARG0..ARG1];

  my $id = \$_[HEAP];
  $id =~ s/REF\(0x//o;
  $id =~ s/\)//o;

  $heap->{task} = $task;
  $heap->{postback} = $postback;

  DEBUG ( 3, "$id: task started ($task)\n");

# New style specifying version first
  my ($type, $ts, $list, $client, $message, $host, $group, $severity);
  my ($service, $color);

  if ( $task =~ /^v(\d+)/io) {
    my $ver = $1;
    $task =~ s/^v\d+\<\|\>//oi;

    if ($ver == 1) {
      my @fields;
      @fields = split (/\<\|\>/o, $task);
      foreach (@fields) {
        my ($param, $value);
        ($param, $value) = /^(.[^-]*)-(.*$)/o;
        if ("timestamp" =~ /^$param/) {
          $ts = $value;
        } elsif ("type" =~ /^$param/) {
          $type = $value;
        } elsif ("list" =~ /^$param/) {
          $list = $value;
        } elsif ("client" =~ /^$param/) {
          $client = $value;
        } elsif ("message" =~ /^$param/) {
          $message = $value;
        } elsif ("host" =~ /^$param/) {
          $host = $value;
        } elsif ("service" =~ /^$param/) {
          $service = $value;
        } elsif ("group" =~ /^$param/) {
          $group = $value;
        } elsif ("severity" =~ /^$param/) {
          $severity = $value;
        } else {
          DEBUG (0, "Unknown param: $param - $task");
        }
      }
# Message type is direct page
      if ($type eq "dp") {
        if (!defined $ts || !defined $list || 
            !defined $client || !defined $message) {
          DEBUG (0, "Missing parameters for direct page: $task");
        } else {
          CALL ("$DIRECT $client $list $ts \"$message\" --id $id --debug 2");
        }
# Standard Message type
      } elsif ($type eq "msg") {
        if (!defined $client || !defined $severity || !defined $host ||
            !defined $service || !defined $ts || !defined $message) {
          DEBUG (0, "Missing parameters for message: $task");
        } else {
          if (defined $group) {
#            $group = "--group $group";
            $group = "";
          }
          CALL ("$MESSENGER --id $id --client $client --color $severity --host $host --service $service $group --time $ts --message \"$message\" --debug 1");
        }
      } else {
        DEBUG (0, "Unknown type: $type - $task");
      }
    } else {
      DEBUG (0, "Unknown version: $ver: $task");
    }
# Message to be sent direct - no rules applied
  } else {
# Remove leading timestamp if still there
    if ($task =~ /^\d+:/o) {
      $task =~ s/^\d+:\s*//o;
      DEBUG (0, "Removed unneeded timestamp");
    }
    if ( $task =~ /^\s*SENDTO\s/o) {
      ($ts, $list, $client, $message) = $task =~ 
         /^\s*SENDTO\s+(\d+)\s*:\s*([^:\s]+)\s*:\s*([^:\s]+)\s*:\s*(.*)$/o;
    
      if (!defined $message) {
        DEBUG (0,"Badly Formatted message: $task");
        next;
      }
      CALL ("$DIRECT $client $list $ts \"$message\" --id $id --debug 2");
    }
    else {

# Message format:  
# CLIENT HOST SERVICE COLOR TIMESTAMP MESSAGE

      ($client, $host, $service, $color, $ts, $message) = $task =~ 
         /^\s*([^\s]+)\s*([^\s]+)\s+([^\s]+)\s+([^\s]+)\s+(\d+)\s+(.*)/o;
      if (!defined $message) {
        DEBUG (0,"Badly Formatted message: $task");
        next;
      }

      CALL ("$MESSENGER --id $id --client $client --color $color --host $host --service $service --time $ts --message \"$message\" --debug 2");
    }
  }
}

sub worker_done {
  my $heap = $_[HEAP];

  DEBUG (3, "test finished ($heap->{task})\n");

  my $postback = delete $heap->{postback};
  if (ref $postback eq 'ARRAY') {
    my $session = $_[KERNEL]->alias_resolve($postback->[0]);
    if (defined $session) {
      $postback = $session->postback( $postback->[1], $heap->{task} );
    }
    else {
      $postback = sub { 1 };
    }
  }

  # Causes some evil recursion somewhere. :(
  # $postback->( $heap->{test}, $heap->{task} );
}

#------------------------------------------------------------------------------

sub spawn_worker {
  my ($outer_postback, $outer_task) = @_;

  POE::Session->create
    ( inline_states =>
      { _start => \&worker_start,
        done   => \&worker_done,
        _stop => sub {},
      },
      args => [ $outer_postback, $outer_task ]
    );
}

#------------------------------------------------------------------------------

POE::Component::JobQueue->spawn
  ( Alias       => 'passive',
    WorkerLimit => $MAXWORKERS,
    Worker      =>
    sub {
      my ($postback, $task) = @_;
      &spawn_worker($postback, $task ) if defined $task;
    },

    Passive => { },
  );


$poe_kernel->run();