but I think performance would be similar to a queue,

The problem with a queue is that all locking is applied to the entire shared array, thus every lock blocks all contenders, even if they are after different elements of the array.

If you look carefully at my example, you'll see that the @buffers array itself isn't (explicitly) shared and is never locked; only the per-thread scalar elements are.

And as only the reader thread and 1 worker thread per buffer are competing for any given lock, all workers threads are free to continue independantly of each other.

And finally, the lock on any given buffer is only held for the brief time it takes to copy its contents to a local buffer, thus the reader thread can be repopulating it with the next record whilst the worker thread is processing the previous one.

The upshot is that in my use of this technique, it beats Thread::Queue by a wide margin for applications where the processing of a record takes 3x or more time, than that required to read it.

Test it. You might be pleasantly surprised.

You lose sequence.

Sorry, I hadn't realised that was a requirement. I'll think about that.

Can you do the same with hashes?

Yes. But be aware that if you don't lock the hash and insertions and deletions are going on, things can get a little strange. (Eg. Iterating the hash keys might return a key, but by the time you try to access the associate value, the key might have been deleted.

If the hash is essentially static -- only the values changing -- then only sharing the value scalars and applying locking to them individually can be much more efficient than locking the entire hash for every update.

Actually it looks like if the hash itself is not shared, any changes to keys will not affect other running threads. This includes deleting keys (the key/value will still be there in other threads). Only changes to the shared values will affect other running threads (including undefining them).

use strict;
use threads;
use threads::shared;

my %hash = (
    'a' => 'foo',
    'b' => 'foo',
    'c' => 'foo',
    'd' => 'foo'
);
share($hash{$_}) foreach('a'..'f');
my $tid = 0;
foreach my $key('a'..'f') {
    if(exists $hash{$key}) {
        if(defined $hash{$key}) {
            print "$key => $hash{$key} in thread $tid\n";
        }
        else {
            print "$key => undef in thread $tid\n";
        }
    }
    else {
        print "$key does not exist in thread $tid\n";
    }
}

my $th1 = async {
    $tid = 1;
    print "thread $tid started\n";
    sleep 2;
    foreach my $key('a'..'f') {
        if(exists $hash{$key}) {
            if(defined $hash{$key}) {
                print "$key => $hash{$key} in thread $tid\n";
            }
            else {
                print "$key => undef in thread $tid\n";
            }
        }
        else {
            print "$key does not exist in thread $tid\n";
        }
    }
    print "delete b from thread $tid\n";
    delete $hash{'b'};
    print "undefine d from thread $tid\n";
    undef $hash{'d'};
    print "setting f => foo in thread $tid\n";
    $hash{'f'} = 'foo';
};
my $th2 = async {
    $tid = 2;
    print "thread $tid started\n";
    sleep 3;
    foreach my $key('a'..'f') {
        if(exists $hash{$key}) {
            if(defined $hash{$key}) {
                print "$key => $hash{$key} in thread $tid\n";
            }
            else {
                print "$key => undef in thread $tid\n";
            }
        }
        else {
            print "$key does not exist in thread $tid\n";
        }
    }
};
sleep 1;
print "delete a from thread $tid\n";
delete $hash{'a'};
print "undefine c from thread $tid\n";
undef $hash{'c'};
print "setting e => foo in thread $tid\n";
$hash{'e'} = 'foo';
$th1->join();
$th2->join();
foreach my $key('a'..'f') {
    if(exists $hash{$key}) {
        if(defined $hash{$key}) {
            print "$key => $hash{$key} in thread $tid\n";
        }
        else {
            print "$key => undef in thread $tid\n";
        }
    }
    else {
        print "$key does not exist in thread $tid\n";
    }
}
[download]

a => foo in thread 0
b => foo in thread 0
c => foo in thread 0
d => foo in thread 0
e => undef in thread 0
f => undef in thread 0
thread 1 started
thread 2 started
delete a from thread 0
undefine c from thread 0
setting e => foo in thread 0
a => foo in thread 1
b => foo in thread 1
c => undef in thread 1
d => foo in thread 1
e => foo in thread 1
f => undef in thread 1
delete b from thread 1
undefine d from thread 1
setting f => foo in thread 1
a => foo in thread 2
b => foo in thread 2
c => undef in thread 2
d => undef in thread 2
e => foo in thread 2
f => foo in thread 2
a does not exist in thread 0
b => foo in thread 0
c => undef in thread 0
d => undef in thread 0
e => foo in thread 0
f => foo in thread 0
[download]