LMPX.COM :: Malte S. Stretz :: [PATCH] Advanced tagging and filtering

From: Malte S. Stretz Date: Fri Dec 29 13:51:34 2006

Subject: [PATCH] Advanced tagging and filtering
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

Attachment: [text/x-diff] IPC-DirQueue-0.08-head.patch

Attachment: [application/x-perl] 55filtered.t
Navigate in group perl.ipc.dirqueue at sever nntp.perl.org
Previous	Next

© No Copyright You are free to use Anything	Site Maintained by PHP Developer Powered By PHP Consultants
From:	Malte S. Stretz	Date:	Fri Dec 29 13:51:34 2006
Subject:	[PATCH] Advanced tagging and filtering