dq-server

#!/usr/bin/perl -w
=head1 NAME

dq-server - run a command on all tasks in an IPC::DirQueue queue

=head1 SYNOPSIS

B<dq-server> --dir I<qdirectory> [--njobs N] I<command arg arg ...>

=head1 DESCRIPTION

B<dq-server> will remove each task in turn from an C<IPC::DirQueue> directory,
running the named command on each one.  Once the queue is empty, it will wait
indefinitely for more tasks to arrive.   A limit on how many jobs to dequeue
can be specified using B<--njobs>.

The command is run as:

    command arg arg ... nameofdatafile

=head1 OPTIONS

=over 4

=item B<--njobs N>  (default: 0)

How many jobs to dequeue from the queue directory.   C<0> means unlimited.

=back

=head1 SEE ALSO

IPC::DirQueue(3)
dq-deque(1)
dq-list(1)
dq-server(1)
dq-submit(1)

=cut

use strict;
use lib 'lib';
use IPC::DirQueue;
use Getopt::Long;

sub usage {
  die "usage: dq-server --dir qdirectory command arg arg...\n";
}

our $dir;
our $njobs = 0;

GetOptions(
  'dir=s'   => \$dir,
  'njobs=i' => \$njobs
) or usage();
$dir or usage();

my $jobsleft = $njobs;
my $dq = IPC::DirQueue->new({ dir => $dir });

while (1) {
  last if ($njobs && $jobsleft-- <= 0);

  my $job = $dq->wait_for_queued_job();
  if (!$job) { next; }

  my @sys = (@ARGV, $job->get_data_path());
  print "running ".join (' ', @sys)."\n";
  system (@sys);
  if ($? >> 8 != 0) {
    warn "command failed: exit=".($? >> 8)."\n";
  }

  $job->finish();
  print "finished\n";
}

exit;