[PATCH] Advanced tagging and filtering
[email protected] ("Malte S. Stretz") Fri, 29 Dec 2006 21:51:34 +0100
| Newsgroups | perl.ipc.dirqueue |
|---|---|
| Message-ID | <[email protected]> |
--Boundary-00=_Y/XlFmuHD0z+qHi
Content-Type: text/plain;
charset="iso-8859-1"
Content-Transfer-Encoding: 7bit
Content-Disposition: inline
Hi Justin,
I was looking for a nice IPC module on CPAN and what did I find? Yet
another Perl Module created by you :)
Unfortunately was IPC::DirQueue really slow in one of the cases I tried to
use it for. Thats why I wrote the attached patch. But first: What do I
need?
I have a server which has a tcpdump running, listening for UDP packets, and
where you can connect via TCP to request notifications about some kind of
packets. The notifications are based on ids which are inside the UDP
packets. (This is actually used to do some UDP hole punching as a
University project.)
So we've got two parallel queues, one for the incoming requests and one for
the outgoing replies. On the one side we can have n processes pushing
requests into the incoming queue and waiting for replies on the outgoing
one. On the other side is just a single consumer/producer, which parses
the tcpdump output, looks wether there's a request for the id in the
incoming queue and if so, puts a reply into the outgoing queue which is
then sent via TCP to the user.
That worked pretty well but at some point I got an almost-deadlock: If one
of the clients pushes loads of requests into the incoming queue, it can
overwhelm the consumer which can happen to look at all those new packets
first before it sees the request it was actually looking for. Additionally
is there quite some overhead because each time the metadata has to be read
first before I can identify the owner.
Actually the following simple code can have the effect I described above:
use IPC::DirQueue;
my $dq = IPC::DirQueue->new({ dir => 'dq' });
$dq->enqueue_string("foo!", {id => 'foo'});
$dq->enqueue_string("bar!", {id => 'bar'});
while (1) {
my $job = $dq->wait_for_queued_job(0, 0.1);
my $id = $job->{metadata}->{id};
#print $id;
if ($id eq 'foo') {
my $data = $job->get_data();
$job->finish();
print $data;
}
else {
$job->return_to_queue();
}
}
If started often enough, at some point we will have loads of bars which
block the foos. Just uncomment the print $id and the app will scream bar
all the time but never foo. Hmmm... I guess there's actually another bug
hidden because sometimes I see no foo at all; maybe the dir iterator is
reset or something.
Whatever, I actually went to increase the speed for applications like this.
The attached patch does the following:
a) Adds a parameter file_mode to the constructor; doesn't really increase
speed but I guess in most cases one wants to change both modes and this way
its less code :)
b) Makes the TAG part of the filename configurable. Ah, yeah, TAG was
called HASH before :) I added the parameters tag (static tag), tag_sub
(callback to generate a dynamic tag) and two more (tag_max_length and
tag_warn, cf. inline doc).
c) Adds $filter parameter to wait_for_queued_job() and pickup_queued_job().
These accept a regular expression which is matched against the queue file
name. I don't really like the position of the $filter parameter in the
wait method, I'd prefer to have it as the first parameter and experimented
with a 'ref $_[1] eq Regexp' to keep it backwards compatible but then just
went the simple way and put it to the end :)
d) The actual filtering stuff is done in the iterators. I guess I did
quite some refactoring/rewriting there. The most important thing is the
$iter->{filter} code, that I made the iterator more or less independent
from the $self object and thus could use it to rewrite the fanout stuff to
use a recursive (unordered) iterator, thus reducing code duplication.
The tests all work (again) and I attached another one for the stuff I did.
So, with the patch, change the code above into the following and its fast
as... dunno, but something really fast :)
use IPC::DirQueue;
my $dq = IPC::DirQueue->new({ dir => 'dq', tag_sub => sub {
return $_[0]->{metadata}->{id};
}});
$dq->enqueue_string("foo!", {id => 'foo'});
$dq->enqueue_string("bar!", {id => 'bar'});
while (1) {
my $job = $dq->wait_for_queued_job(0, 0.1, qr/\.foo(\..*|)$/);
my $id = $job->{metadata}->{id};
#print $id;
my $data = $job->get_data();
print $data;
$job->finish();
}
Cheers,
Malte
--Boundary-00=_Y/XlFmuHD0z+qHi
Content-Type: text/x-diff;
charset="iso-8859-1";
name="IPC-DirQueue-0.08-head.patch"
Content-Transfer-Encoding: 7bit
Content-Disposition: attachment;
filename="IPC-DirQueue-0.08-head.patch"
Index: lib/IPC/DirQueue.pm
===================================================================
--- lib/IPC/DirQueue.pm (revision 8491)
+++ lib/IPC/DirQueue.pm (working copy)
@@ -90,23 +90,40 @@
Name the directory where the queue files are stored. This is required.
-=item data_file_mode => $mode (default: 0666)
+=item file_mode => $mode (default: 0666)
-The C<chmod>-style file mode for data files. This should be specified
-as a string with a leading 0. It will be affected by the current
-process C<umask>.
+The C<chmod>-style file mode for data and queue control files. This
+should be specified as a string with a leading 0. It will be affected
+by the current process C<umask>.
-=item queue_file_mode => $mode (default: 0666)
+=item data_file_mode => $mode (default: C<file_mode>)
-The C<chmod>-style file mode for queue control files. This should be
-specified as a string with a leading 0. It will be affected by the
-current process C<umask>.
+Override the file mode for data files. See C<file_mode>.
+=item queue_file_mode => $mode (default: C<file_mode>)
+
+Override the file mode for queue control files. See C<file_mode>.
+
=item ordered => { 0 | 1 } (default: 1)
Whether the jobs should be processed in order of submission, or
in no particular order.
+=item tag => $string (default: some hash)
+
+Sets the TAG part of the file name. The default is a semi-random
+hash based on the hostname and the process id.
+
+=item tag_sub => $subref (default: returns C<tag>)
+
+The TAG can be generated dynamically by routine referenced by this
+parameter. See also 'QUEUE DIRECTORY STRUCTURE' below.
+
+=item tag_max_length => int (default: 128)
+
+The tag is part of the file name, so it should not be too long. The default
+limit is already quite generous, there shouldn't be any reason to increase it.
+
=item queue_fanout => { 0 | 1 } (default: 0)
Whether the queue directory should be 'fanned out'. This allows better
@@ -142,9 +159,10 @@
bless ($self, $class);
die "no 'dir' specified" unless $self->{dir};
- $self->{data_file_mode} ||= '0666';
+ $self->{file_mode} ||= '0666';
+ $self->{data_file_mode} ||= $self->{file_mode};
$self->{data_file_mode} = oct ($self->{data_file_mode});
- $self->{queue_file_mode} ||= '0666';
+ $self->{queue_file_mode} ||= $self->{file_mode};
$self->{queue_file_mode} = oct ($self->{queue_file_mode});
if ($self->{queue_fanout}) {
@@ -154,6 +172,13 @@
elsif (!defined $self->{ordered}) {
$self->{ordered} = 1;
}
+
+ $self->{tag} ||= hash_string_to_filename($self->gethostname().$$);
+ $self->{tag_sub} ||= sub { return $self->{tag}; };
+ $self->{tag_max_length} ||= 128;
+ if (!defined $self->{tag_warn}) {
+ $self->{tag_warn} = 1;
+ }
$self->{buf_size} ||= 65536;
$self->{active_file_lifetime} ||= 600;
@@ -427,10 +452,14 @@
###########################################################################
-=item $job = $dq->pickup_queued_job();
+=item $job = $dq->pickup_queued_job([ $filter ]);
Pick up the next job in the queue, so that it can be processed.
+The parameter C<$filter> can be used to specify a regular expression which
+is matched against the queued filename. All files which don't match will be
+skipped.
+
If no job is available for processing, either because the queue is
empty or because other worker processes are already working on
them, C<undef> is returned; otherwise, a new instance of C<IPC::DirQueue::Job>
@@ -442,13 +471,13 @@
=cut
sub pickup_queued_job {
- my ($self) = @_;
+ my ($self, $filter) = @_;
my $pathqueuedir = $self->q_subdir('queue');
my $pathactivedir = $self->q_subdir('active');
$self->ensure_dir_exists ($pathactivedir);
- my $iter = $self->queue_iter_start($pathqueuedir);
+ my $iter = $self->queue_iter_start($pathqueuedir, $filter);
while (1) {
my $nextfile = $self->queue_iter_next($iter);
@@ -596,7 +625,7 @@
###########################################################################
-=item $job = $dq->wait_for_queued_job ([ $timeout [, $pollinterval] ]);
+=item $job = $dq->wait_for_queued_job ([ $timeout [, $pollinterval [, $filter ]] ]);
Wait for a job to be queued within the next C<$timeout> seconds.
@@ -619,10 +648,12 @@
the nearest round multiple of C<$pollinterval> greater than C<$timeout>
will be used instead. Also note that C<$timeout> is used as an integer.
+The job can be filtered with C<$filter> as in C<pickup_queued_job()>.
+
=cut
sub wait_for_queued_job {
- my ($self, $timeout, $pollintvl) = @_;
+ my ($self, $timeout, $pollintvl, $filter) = @_;
my $finishtime;
if ($timeout && $timeout > 0) {
@@ -651,7 +682,7 @@
my @stat = stat ($pathqueuedir);
my $qdirlaststat = $stat[9];
- my $job = $self->pickup_queued_job();
+ my $job = $self->pickup_queued_job($filter);
if ($job) { return $job; }
# there's another semi-race condition here, brought about by a lack of
@@ -825,20 +856,6 @@
###########################################################################
-sub get_dir_filelist_sorted {
- my ($self, $dir) = @_;
-
- if (!opendir (DIR, $dir)) {
- return []; # no dir? nothing queued
- }
- # have to read the lot, to sort them.
- my @files = sort grep { /^\d/ } readdir(DIR);
- closedir DIR;
- return \@files;
-}
-
-###########################################################################
-
sub copy_in_to_out_fh {
my ($self, $fhin, $callbackin, $fhout, $outfname) = @_;
@@ -1101,28 +1118,41 @@
my @gmt = gmtime ($job->{time_submitted_secs});
- # NN.20040718140300MMMM.hash(hostname.$$)[.rand]
+ # NN.20040718140300MMMM.tag[.rand]
#
# NN = priority, default 50
# MMMM = microseconds from Time::HiRes::gettimeofday()
+ # tag = some base64-ish string, default hash(hostname.$$)
# hostname = current hostname
- my $buf = sprintf ("%02d.%04d%02d%02d%02d%02d%02d%06d.%s",
+ my $base = sprintf ("%02d.%04d%02d%02d%02d%02d%02d%06d.",
$job->{pri},
$gmt[5]+1900, $gmt[4]+1, $gmt[3], $gmt[2], $gmt[1], $gmt[0],
- $job->{time_submitted_msecs},
- hash_string_to_filename ($self->gethostname().$$));
+ $job->{time_submitted_msecs});
# normally, this isn't used. but if there's a collision,
# all retries after that will do this; in this case, the
# extra anti-collision stuff is useful
- if ($addextra) {
- $buf .= ".".$$.".".$self->get_random_int();
- }
+ my $extra = $addextra ? ".".$$.".".$self->get_random_int() : "";
- return $buf;
+ return $base.$self->get_q_filename_tag($job, $base, $extra).$extra;
}
+sub get_q_filename_tag {
+ my($self, $job, $base, $extra) = @_;
+ # create a (new?) tag
+ my $str = $self->{tag_sub}->($job, $base, $extra) || '';
+ # weed out all dangerous chars
+ my $tag = filter_unsafe_chars($str);
+ # limit the length
+ $tag = substr($tag, 0, $self->{tag_max_length});
+ # warn the user if it was filtered
+ if ($self->{tag_warn} && $tag ne $str) {
+ warn "IPC::DirQueue: the tag was filtered\n";
+ }
+ return $tag;
+}
+
sub hash_string_to_filename {
my ($str) = @_;
# get a 16-bit checksum of the input, then uuencode that string
@@ -1130,6 +1160,12 @@
# transcode from uuencode-space into safe, base64-ish space
$str =~ y/ -_/A-Za-z0-9+_/;
# and remove the stuff that wasn't in that "safe" range
+ return filter_unsafe_chars($str);
+}
+
+sub filter_unsafe_chars {
+ my ($str) = @_;
+ # remove any chars which aren't in the "safe" base64-ish range
$str =~ y/A-Za-z0-9+_//cd;
return $str;
}
@@ -1205,62 +1241,93 @@
###########################################################################
sub queue_iter_start {
- my ($self, $pathqueuedir) = @_;
-
- if ($self->{indexclient}) {
- dbg ("queue iter: getting list for $pathqueuedir");
- my @files = sort grep { /^\d/ } $self->{indexclient}->ls($pathqueuedir);
-
- if (scalar @files <= 0) {
- return if $self->queuedir_is_bad($pathqueuedir);
- }
-
- return { files => \@files };
+ my ($self, $pathqueuedir, $filter, $type) = @_;
+
+ $filter ||= qr/^/;
+ dbg ("queue iter: filter $filter in $pathqueuedir");
+ unless (ref $filter eq 'CODE') {
+ # we need to copy $filter here else the closure will get annoyed
+ my $f = $filter;
+ $filter = sub {
+ if (wantarray) { # grep is picky about list context
+ return grep { /^\d/ && /$f/ } @_;
+ }
+ else {
+ $_ = shift;
+ return unless defined;
+ return unless /^\d/;
+ return unless /$f/;
+ return $_;
+ }
+ };
}
- elsif ($self->{ordered}) {
- dbg ("queue iter: opening $pathqueuedir (ordered)");
- my $files = $self->get_dir_filelist_sorted($pathqueuedir);
- if (scalar @$files <= 0) {
- return if $self->queuedir_is_bad($pathqueuedir);
- }
+
+ # iterator always has:
+ # dir = directory processed
+ # filter = filter routine applied to the files
+ # iterator can have:
+ # files = cached list of files (indexclient, ordered)
+ # sub = pointer to sub-iterator (fanout)
+ # fh = directory handle (unordered)
+ my $iter = {
+ dir => $pathqueuedir,
+ filter => $filter,
+ };
- return { files => $files };
+ unless ($type) {
+ $type = $self->{indexclient} ? 'indexclient'
+ : $self->{ordered} ? 'ordered'
+ : $self->{queue_fanout} ? 'fanout'
+ : 'unordered';
}
- elsif ($self->{queue_fanout}) {
- return $self->queue_iter_fanout_start($pathqueuedir);
+ if ($type eq 'ordered') {
+ $iter = $self->queue_iter_ordered_start($iter);
}
- else {
+ elsif ($type eq 'unordered') {
my $dirfh;
- dbg ("queue iter: opening $pathqueuedir");
- if (!opendir ($dirfh, $pathqueuedir)) {
- return if $self->queuedir_is_bad($pathqueuedir);
- if (!opendir ($dirfh, $pathqueuedir)) {
+ dbg ("unordered: opening $iter->{dir}");
+ if (!opendir ($dirfh, $iter->{dir})) {
+ return if $self->queuedir_is_bad($iter->{dir});
+ if (!opendir ($dirfh, $iter->{dir})) {
warn "oops? pathqueuedir bad";
return;
}
}
- return { fh => $dirfh };
+ $iter->{fh} = $dirfh;
}
+ elsif ($type eq 'fanout') {
+ $iter = $self->queue_iter_fanout_start($iter);
+ }
+ elsif ($type eq 'indexclient') {
+ dbg ("indexclient: getting list for $iter->{dir}");
+ my @files = sort $iter->{filter}->($self->{indexclient}->ls($iter->{dir}));
- die "cannot get here";
+ if (scalar @files <= 0) {
+ return if $self->queuedir_is_bad($iter->{dir});
+ }
+
+ $iter->{files} = \@files;
+ }
+ else {
+ die "unknown iterator type $type";
+ }
+
+ return $iter;
}
sub queue_iter_next {
my ($self, $iter) = @_;
- if ($self->{indexclient}) {
+ if ($iter->{files}) {
return shift @{$iter->{files}};
}
- elsif ($self->{ordered}) {
- return shift @{$iter->{files}};
+ elsif ($iter->{fh}) {
+ return $self->queue_iter_unordered_next($iter);
}
- elsif ($self->{queue_fanout}) {
+ elsif ($iter->{fanoutlist}) {
return $self->queue_iter_fanout_next($iter);
}
- else {
- return readdir($iter->{fh});
- }
return;
}
@@ -1269,12 +1336,58 @@
my ($self, $iter) = @_;
return unless $iter;
- if (defined $iter->{fanfh}) { closedir($iter->{fanfh}); }
+ $self->queue_iter_stop($iter->{sub});
+ dbg ("queue iter: closing $iter->{dir}");
if (defined $iter->{fh}) { closedir($iter->{fh}); }
+ delete $iter->{fh};
+ delete $iter->{files};
+ delete $iter->{fanoutlist};
+
+ return undef;
}
+###########################################################################
+sub queue_iter_ordered_start {
+ my ($self, $iter) = @_;
+
+ my @files = ();
+
+ dbg ("ordered: opening $iter->{dir} (ordered)");
+ if (opendir (DIR, $iter->{dir})) {
+ # have to read the lot, to sort them.
+ @files = sort $iter->{filter}->(readdir(DIR));
+ closedir DIR;
+ }
+
+ if (scalar @files <= 0) {
+ return if $self->queuedir_is_bad($iter->{dir});
+ }
+
+ $iter->{type} = 'files';
+ $iter->{files} = \@files;
+ return $iter;
+}
+
###########################################################################
+sub queue_iter_unordered_next {
+ my ($self, $iter) = @_;
+ my $file = undef;
+
+ while (1) {
+ $file = readdir($iter->{fh});
+ last unless defined $file;
+ #dbg("unordered: candidate $file");
+ $file = $iter->{filter}->($file);
+ last if defined $file;
+ #dbg("unordered: candidate filtered");
+ }
+
+ return $file;
+}
+
+###########################################################################
+
sub queue_dir_fanout_create {
my ($self, $pathqueuedir) = @_;
@@ -1323,26 +1436,23 @@
}
sub queue_iter_fanout_start {
- my ($self, $pathqueuedir) = @_;
- my $iter = { };
+ my ($self, $iter) = @_;
{
my @fanouts;
- dbg ("queue iter: opening $pathqueuedir");
- if (!opendir (DIR, $pathqueuedir)) {
+ dbg ("fanout: opening $iter->{dir}");
+ if (!opendir (DIR, $iter->{dir})) {
@fanouts = (); # no dir? nothing queued
}
else {
my %map = map {
- $_ => (-M $pathqueuedir.SLASH.$_)
+ $_ => (-M $iter->{dir}.SLASH.$_)
} grep { /^[a-z0-9]$/ } readdir(DIR);
@fanouts = sort { $map{$a} <=> $map{$b} } keys %map;
- dbg ("fanout: $pathqueuedir, order is ".join ' ', @fanouts);
+ dbg ("fanout: $iter->{dir}, order is ".join ' ', @fanouts);
}
closedir DIR;
$iter->{fanoutlist} = \@fanouts;
- $iter->{pathqueuedir} = $pathqueuedir;
-
}
return $iter;
}
@@ -1350,40 +1460,34 @@
sub queue_iter_fanout_next {
my ($self, $iter) = @_;
- # dir handles are:
- # /path/to/queue = $iter->{fh}
- # /f = $iter->{fanfh}
-
next_fanout:
- # open the {fanfh} handle, if it isn't already going
- if (!defined $iter->{fanfh}) {
+ # start the sub iterator, if it isn't already going
+ if (!defined $iter->{sub}) {
my $nextfanout = shift @{$iter->{fanoutlist}};
if (!defined $nextfanout) {
dbg ("fanout: end of list");
return;
}
- my $dirfh;
+ my $subiter;
dbg ("fanout: opening next dir: $nextfanout");
- if (!opendir ($dirfh, $iter->{pathqueuedir}.SLASH.$nextfanout)) {
- warn "opendir failed $iter->{pathqueuedir}/$nextfanout: $!";
- return;
- }
+ $subiter = $self->queue_iter_start($iter->{dir}.SLASH.$nextfanout,
+ $iter->{filter}, 'unordered');
+ return unless $subiter;
+ $iter->{sub} = $subiter;
$iter->{fanstr} = $nextfanout;
- $iter->{fanfh} = $dirfh;
}
- my $fname = readdir($iter->{fanfh});
+ my $fname = $self->queue_iter_next($iter->{sub});
if (defined $fname) {
return $iter->{fanstr}.SLASH.$fname; # best-case scenario
}
dbg ("fanout: finished this dir, trying next one");
- closedir($iter->{fanfh});
+ $iter->{sub} = $self->queue_iter_stop($iter->{sub});
$iter->{fanstr} = undef;
- $iter->{fanfh} = undef;
goto next_fanout;
}
@@ -1451,15 +1555,32 @@
The filename format is as follows:
- 50.20040909232529941258.HASH[.PID.RAND]
+ 50.20040909232529941258.TAG[.PID.RAND]
+ | base |tag | extra |
The first two digits (C<50>) are the priority of the job. Lower priority
numbers are run first. C<20040909232529> is the current date and time when the
enqueueing process was run, in C<YYYYMMDDHHMMSS> format. C<941258> is the time in
-microseconds, as returned by C<gettimeofday()>. And finally, C<HASH> is a
-variable-length hash of some semi-random data, used to increase the chance of
-uniqueness.
+microseconds, as returned by C<gettimeofday()>.
+The C<TAG> is a short string-sequence. Per default it is a variable-length hash
+of some semi-random data, used to increase the chance of uniqueness. But it is also
+possible to set this manually with the C<tag> parameter of the constructor, or even
+to have it generated dynamically by the routine set with C<tag_sub>.
+
+The sub routine is called with the parameters C<$job>, C<$base> and C<$extra>. The
+first is the job currently enqueued and thus allows access to the metadata. The
+latter two parameters correspond to the already known parts of the filename as
+shown above (including the dots).
+
+Note that only characters from the set [A-Za-z0-9+_] are allowed and the length
+of the string is limited to 128 characters. If any of these restrictions are
+violated, a warning will be thrown.
+
+In combination with the C<$filter> parameter the retriving methods offer, the
+C<TAG> can be used to filter the queued data more efficiently as if the metadata
+was used.
+
If there is a collision, the timestamps are regenerated after a 250 msec sleep,
and further randomness will be added at the end of the string (namely, the
current process ID and a random integer value). Up to 10 retries will be
--Boundary-00=_Y/XlFmuHD0z+qHi
Content-Type: application/x-perl;
name="55filtered.t"
Content-Transfer-Encoding: 7bit
Content-Disposition: attachment;
filename="55filtered.t"
#!/usr/bin/perl -w
BEGIN { $COUNT = 10; }
use Test; BEGIN { plan tests => 1 + $COUNT * (5 + 1) + $COUNT * 3; }
use lib '../lib'; if (-d 't') { chdir 't'; }
use IPC::DirQueue;
mkdir ("log");
mkdir ("log/qdir");
my $bq = IPC::DirQueue->new({ dir => 'log/qdir', tag_sub => sub {
my($job, $base, $extra) = @_;
ok($job);
ok(defined $base);
ok($base =~ /^\d{2}\.\d{20}\.$/);
ok(defined $extra);
my $id = $job->{metadata}->{id};
ok($id =~ /^\d+$/);
print "queueing $id into $base$id$extra\n";
return $id;
}});
ok ($bq);
start_writer();
start_worker();
exit;
sub start_writer {
for my $j (1 .. $COUNT) {
ok ($bq->enqueue_string ("hello $j! $$", { id => $j }));
}
}
sub start_worker {
my @k = (1 .. $COUNT);
while (@k) {
my $k = splice(@k, int(rand(scalar @k)), 1);
print "looking for $k...\n";
my $job;
$job = $bq->pickup_queued_job(qr/^$k$/);
ok(!$job);
$job = $bq->wait_for_queued_job(0, 1, qr/^\d+\.\d+\.$k(\..*|)$/);
ok($job);
my $data = $job->get_data();
ok ($data =~ /^hello $k! \d+$/)
or warn "got: [$data]";
$job->finish();
print "finished $k\n";
}
}
--Boundary-00=_Y/XlFmuHD0z+qHi--