A per-instance shared queue among forked child processes?

[EvanK]

I'm writing a script that will fork off a series of child processes (let's say 5 or 10 as an example), and then all of the child processes will be adding and removing items for processing to/from a shared queue. The shared queue among the child processes is turning out to be kind of difficult to implement.

This queue really just needs to be a FIFO, where I can push items to the end of it, and pop items from the beginning of it. And it could get quite large, so I can't store it in-memory, because then I'll likely run afoul of the OOM killer. The real catch here is that each instance of the script will need its *OWN* shared queue, available only to its own child processes, because I'll likely end up running multiple instances of it at once.

I had thought of making a db connection in the parent process before forking, then cloning the db handle for each child and using a mysql temp table for the "queue", but it seems cloning actually makes a brand new user session with mysql, so one child can't use (or even see) another child's temp tables (as mysql by design restricts access of a temp table to the session in which it was created). I can't easily use normal (permanent) tables because multiple instances of the script will step all over each other.

So, how could I cleanly and efficiently share what could potentially be a LARGE amount of data between multiple child processes, yet keep it confined to only that instance of the script?

Update:

I ended up using chrestomanci's solution, with just one table and using the pid of the parent as a key for each instance. This seemed like the best of both worlds (easy to implement, easy to understand), and since it's a mysql instance on the same server, it's producing some good throughput.

I might look into SQLite as salva suggested, though I'm using a named lock for popping from the queue (since I have to do a select, followed immediately by a delete) and I'm foggy on how (or if) i can exclusively lock a sqlite db in the same fashion.

Update 2:

It looks like BEGIN EXCLUSIVE TRANSACTION should do essentially the same thing as I was using a named lock for, explicitly preventing other processes from reading/writing until the transaction is ended.

[BrowserUk]

For large volumes of data--I'm assuming large individual items--I'd use a subdirectory named for the parent pid, and files named by monotonic numbers for the items; and have the parent manage the item numbers and coordinate the pushes & pops via a socket.

If the parent already has a role to play beyond waiting for its kids, then it just requires one more fork for the socket manager.

When a child pushes an item, it sends a push message to the parent socket, and gets back a new item number, creates and writes the file.

When a child pops an item, it sends a pop message to the parent socket and gets back an item number which forms the name of the file it must read and delete.

If the items can persist across runs, or the parent crashes, you supply the name of the original pid directory that the new instance will take over, and it renames it after itself; scans the directory to discover the lowest/highest item numbers; opens its port and kicks off its kids with the port number.

Since any substantial volume of data sent to a DB is going to end up in the file system anyway, why funnel it all through a pipe or socket when the entire "data management" part of the equation consists of nothing more than coordinating the increments of two monotonically increasing numbers?

[wind]

I can't easily use normal (permanent) tables because multiple instances of the script will step all over each other.

Why not just use a permanent table that is associated with each child's process id?

my $pid = fork();

if (not defined $pid) {
    warn "resources not avilable.\n";

} elsif ($pid == 0) {
    print "Child: My PID is $$\n";
    exit(0);
}

print "Parent: I talk to child by adding to QUEUE associated with $pid
+\n";

[EvanK]

That's a great idea that I can't believe I didn't think of. I can create the table in the parent:

$queue_table = 'queue_'.$$;
$dbh->do('CREATE TABLE '.$queue_table.' (payload TEXT NOT NULL)');

And then access it from the children after the fork, via $queue_table.

[wind]

I personally would not create a new table for each instance, but would instead have a static table that contains the key PARENTID or CHILDID depending on if you wanted the queues to be associated to the parent or child.

my $sth = $dbh->prepare(q{INSERT INTO TABLE QUEUE SET PARENTID=?, PAYL
+OAD=?});
$sth->execute($parentid, $payload) or die $dbh->errstr;

Before each new instance of a parent is run, it should simply scan the table to see if there are any old records left over from a previous process with the same pid.

[chrestomanci]

An alternative to keeping the queue in a database would be to keep it in a plain text file, and manage access to the file with standard file locking flock etc.

In general though, I think that if you have a suitable database available and you are familiar with it, then that would be the more scalable and reliable solution.

I don't think you should create a separate table for each parent PID though as that will pollute the database with lots of tables that have the same purpose. If I where you I would share the same table between all workers, and just have a row that records the PID of the parent. Then in each worker the select call can filter on that PID and only get jobs that pertain to it.

[salva]

use SQLite!

[BrowserUk]

Doesn't SQLite get horribly slow when you have multiple contending writers?

[salva]

Well, it seems that no, as far as journaling is disabled.

#!/usr/bin/perl

use strict;
use warnings;

use POSIX qw(_exit);
use DBI;
use File::Temp;

$| = 1;
my $dbfile = "/tmp/db-$$";

my $dbh = DBI->connect("dbi:SQLite:$dbfile", "", "", {RaiseError => 1}
+);

$dbh->do("create table queue (child, ix)");

for my $id (0..3) {
    fork or do {
        print "<$id enter>";
        my $dbh = DBI->connect("dbi:SQLite:$dbfile", "", "", {RaiseErr
+or => 1});
        $dbh->do("PRAGMA journal_mode = OFF");
        my $sth;
        while (not defined $sth) {
            $sth = eval { $dbh->prepare("insert into queue values (?, 
+?)") };
        }
        print "<$id with sth>";
        for (1..1000) {
            eval {
                $sth->execute($id, $_);
                print $id;
            };
            $@ and print "<$id error: $@>"
        };
        print "<$id exit>";
        _exit(0);
    }
}

1 while (wait > 0);

my $sth = $dbh->prepare("select count(*) from queue");
$sth->execute;
my $row = $sth->fetchrow_arrayref;
print "\n@$row\n";

On my Linux box it does 4000 insertions in 35 seconds and having several concurrent processes accessing the same database does not affect its performance noticeably.

update: It seems that even better than disabling the journal, it is to disable synchronous writes (= ensuring that written data actually gets into the disk surface):

$dbh->do("PRAGMA synchronous = OFF");

That makes 4000 inserts in 2.5 seconds in my computer, and is safer as the database would not get corrupted unless the OS failed to flush cache data to disk.

[sundialsvc4]

I would be very reluctant to use SQLite in such a situation, out of fear that “this is not really the sort of thing that its designers (probably) had in mind.”   I am categorically nervous about going too far outside the envelope.

If you have “4,000 inserts to do in 2.5 seconds,” I wonder if you could get away with somehow batching the queue-entries.   If there is any sort of queue-buildup at all, maybe you could hold ’em until you’ve got, say, 10 entries accumulated and then drop them into the queue all at once.   (They would be dequeued and processed in similar groups.)   This would reduce queue overhead by 10% and maybe those microseconds are otherwise adding-up to a useful-to-avoid amount of time.   Sometimes they do ...

[sundialsvc4]

When you start talking about “a very large number of records, processes, etc.,” I freely admit that I find myself asking, “well, why not use a database table for this?”

Sure, there’s going to be a little bit of overhead, but in the grand scheme of things you probably don’t care.   A database table could hold any number of rows, and it could, through appropriate columns, also support any number of “queues.”   You could indicate that the record had been processed by, say, posting a non-NULL value into the row, and... that would also be a great place to put any other information (status, anyone?) that you might see fit to place there.   You also know that the values will be persistent.   If you wanted to do statistics and so-forth, having a persistent record of the event just might be the bee’s knees.

In other words, my instincts tell me that using a database table in this instance (a) won’t be “egregiously painful,” and (b) just might be useful.   “It’s not contra- indicated, and it just might be ... indicated.”

[believer]

Have you considered using threads instead of forks? Thread::Queue is pretty nice. If memory usage of the queue really is an issue, you can refer to files / db rows / ...

[EvanK]

I was using threads in my original version of the script, but ran into several issues:
• Starting new threads seemed to be much heavier in resource usage than starting new forks (and I'm forking on demand several times within a main event loop)
• Worker threads were getting seemingly random SIGALRM signals, which ended up bombing out the entire process with an "Alarm Clock" message
• I couldn't find a definitive answer on whether many of the modules I'm using are threadsafe.
• Quite frankly, I understand forking much better on a conceptual level than ithreads
So I ended up rewriting it with forks and it seems much more stable this way.

[salva]

It looks like BEGIN EXCLUSIVE TRANSACTION should do essentially the same thing as I was using a named lock for, explicitly preventing other processes from reading/writing until the transaction is ended.

As SQLite has support for transactions, when writting the script inside Re^4: A per-instance shared queue among forked child processes? I use them but then I found that too many of them failed due to collisions so I change the code to use an additional atomic reserve step.

Now, it...

1. looks for some element from the queue,
2. tries to atomically reserve it and
3. deletes it