Basic example on how to use redis for job queueing in Perl

README.pod

Redis jobqueue client

Redis has a feature called lists. The lists can be treated as a stack where elements can be pushed onto or popped from. There are several calls which help you to manage such lists. Once nice feature is the blpop command which does a blocking call on the queue. Therefore it's a nice and elegant way to implement a queue systems for tasks.

One client (worker) pushes the clients on the queue and other clients (worker) pop from this queue. This will help to decrease the amount of code and you are not running in loops and check if new data is available on the list.

This is a fully working example which I use to manage a build system for building Perl modules on the win32 platform. You can read the full blog article about this at my blog at the following url jobqueues.

Author

Ulrich Habel [email protected]

queuerunner.pl

#!/usr/bin/env perl

use v5.10;
use strict;
use warnings;
use Redis::Client;

my $queue    = 'queue:jobs';    # Name of the queue in redis
my $fail     = 'queue:fail';    # Queue with all the failed jobs
my $timeout  = 60;               # Timeout for the single job run time
my $wait     = 5;                # Timer for the running tasks
my $runnable = 1;                # Variable for job looping

# We might loose a job when this job is terminated with Ctrl-C.
# So let's catch the signals and shutdown the jobrunner after
# a full run is completed.

sub INT_handler {

  # Setting the value for the while loop to false == 0
  # so the while loop will terminate after the run
  warn "Shutdown requested - please wait until job is completed.\n";
  $runnable = 0;
}

$SIG{'INT'} = 'INT_handler';

# Instantiate a Redis::Client object
my $client = Redis::Client->new(host => 'localhost', port => 6379);


# This loop will run forever
while ($runnable) {

  say "Jobrunner is active";

  # Get an element from the workqueue or wait until there
  # is one available
  # (You might want to use a timeout to see if your worker
  # is still runnning = $wait );
  #
  # Please note: We are getting an element, if we fail we put
  # it back in the fail queue
  my ($list, $job) = $client->blpop($queue, $wait);
  $job // next;

  eval {
    local $SIG{ALRM} = sub {
      die "Timeout for job ($job)\n";
    };
    alarm $timeout;

    # do your job workload here! - probably call something,
    # do some video encoding, whatever you like
    # Please remember: Your job will get killed after $wait!
    say "I have a job";

    alarm 0;
  };

  # Check if we ran into an error
  if ($@) {

    say "Job fail ($job)";

    # Let's catch some errors - add more if you know more fail
    # reasons for your job runner.
    given ($@) {
      when (/^Timeout/) { warn "Job got a timeout\n"; }
      default           { warn "Job failed. Reason($@)\n"; }
    }

    # Let's push the failed job to a fail queue
    $client->rpush($fail => $job);

  }
  else {

    # The job succeeded - nothing to do here
    say "Job success ($job)";
  }

}    # end of while