I didn't see any mentions of files so I wasn't going to jump to the conclusion that "huge dataset" means some single huge file or even any files at all.

"The job is embarrassingly parallelizable, so I was going to simply carve it up into pieces. The problem was getting each of the pieces into the threads. If I split up the data before hand, then all of it still gets copied into the threads."

So the OP already has an idea of how to "carve up" the data and it would seem that reading in the data isn't unacceptably slow as-is, it is just the running out of memory when creating iThreads after that. So I don't see why you jump to the conclusion of wanting to read the data in parallel either.

If you have your heart set on writing some pseudo code for loading the data, then you'll need to await the "more information" that you already asked for. In the mean time, the answer I provided may well be enough for the OP to adjust the way he already knows how to load the data so that much less memory is required.

To amplify what JavaFan mentioned, there is a module, forks.pm that will allow "copy on write" sharing of the loaded data. Exactly how the data is loaded and used might mean that this is an insignificant advantage in the long run but it is also trivial to try (if you aren't running in MS Windows) and might make a huge difference.

Having the parent read in the data and hand off each piece to the appropriate thread(s) (I'm guessing via Thread::Queue might be a good way) is the most general method that springs to my mind. I'd probably do something similar except using processes and simple pipes, as I've often done.

Having the parent read in the data and hand off each piece to the appropriate thread(s) (I'm guessing via Thread::Queue might be a good way) is the most general method that springs to my mind.

Something like this maybe?:

#! perl -slw
use strict;
use threads;
use threads::shared;
use Thread::Queue;

my $sem : shared;

sub worker {
    my $Q = shift;
    while( my $line = $Q->dequeue ) {
         $line =~ s[a][A]g;
         lock $sem;
         print $line;
    }
}

our $T //= 4;
my $n = $T -1;

my @Qs = map Thread::Queue->new, 0 .. $n;

my @threads = map{
    threads->create( \&worker, $Qs[ $_ ] )
} 0 .. $n;

my $i = 0;
while( <> ) {
    $Qs[ $i ]->enqueue( $_ );
    $i = ( $i + 1 ) % $T;
}

$Qs[ $_ ]->enqueue( undef ) for 0 .. $n;

$_->join for @threads;
__END__
junk71 3.5GB >nul
[download]

Problem: That will take ~4 hours to process a 3.5 GB file. And that's with the output redirected to nul so there is no competition for the disk head.

I'd probably do something similar except using processes and simple pipes, as I've often done.

So something like this, but using processes instead of threads perhaps?:

package IOFan;
use strict;
use feature qw[ state ];
use warnings;
use threads;
use threads::shared;
require Exporter;

our @ISA = 'Exporter';
our @EXPORT = 'fan';

sub tprint {
    my $tid = threads->tid;
    state $sem :shared;
    lock $sem;
    print "$tid: ",@_, "\n";
}

sub twarn {
    my $tid = threads->tid;
    state $sem :shared;
    lock $sem;
    print STDERR "$tid: ",@_, "\n";
}

sub worker {
    twarn 'Started';
    sleep 1;
    my( $in, $out, $code ) = @_;
    while( <$in> ) {
        $code->();
        print $out $_;
    }
    close $in;
    close $out;
    twarn 'ended';
}

use constant { RDR =>0, WTR => 1 };

sub fan (&@) {
    my( $code, $nThreads, $in, $out ) = @_;
    my $n = $nThreads -1;
    $out = \*STDOUT unless defined $out;
    $in  = \*STDIN  unless defined $in;

    my( @fanout, @fanin );
    pipe $fanout[ $_ ][RDR], $fanout[ $_ ][WTR] or die $! for 0 .. $n;
    do{ my $std = select $_->[WTR]; $|++; select $std } for @fanout;
    pipe  $fanin[ $_ ][RDR],  $fanin[ $_ ][WTR] or die $! for 0 .. $n;
    do{ my $std = select $_->[WTR]; $|++; select $std } for @fanin;

    my @threads = map{ threads->create(
        \&worker, $fanout[ $_ ][RDR], $fanin[ $_ ][WTR], $code
    ) } 0 .. $n;

    close $fanout[ $_ ][RDR] for 0 .. $n;
    close  $fanin[ $_ ][WTR] for 0 .. $n;

    async {
        twarn 'reader started';
        my $i = 0;
        while( <$in> ) {
            print { $fanout[ $i = ++$i%$nThreads ][WTR] } $_;
        }
        sleep 1;
        close $fanout[ $_ ][WTR] for 0 .. $n;
        twarn 'reader done';
    }->detach;


    twarn 'writer started';
    my $i = 0;

    while( readline( $fanin[ $i = ++$i % $nThreads ][RDR] ) ) {
        print $out $_;
    }
    close $fanin[ $_ ][RDR] for 0 .. $n;
    twarn 'writer done';
}

1;
------------
#! perl -sw
use strict;
use IOFan;

fan{
    s[a][A]g;
} $ARGV[0];
[download]

This fares better and only takes ~15 minutes to process the 3.5GB.

But all that effort is for naught as this:

 C:\test>perl -pe"s[a][A]" phrases.txt >nul
[download]

processes the same 3.5GB in exactly the same way, but in less than 2 minutes.

Now the "processing" in all these examples is pretty light, just a single pass of each record, but it serves to highlight the scale of the overhead involved in distributing the records. And the scale that the processing would need to involve to make either of these distribution mechanisms viable.

The pipes mechanism is more efficient than the shared queues, and should be pretty much the same between processes as it is between threads, so I doubt there is much to be gained by going that route.

Maybe you have some mechanism in mind that will radically alter the overheads equation, but there is an awful lot of in-the-mind's-eye-expertise around here, and having spent a long time trying to find a good way to do this very thing, I'd really like to be educated by someone who's actually made it work.

---

Examine what is said, not who speaks -- Silence betokens consent -- Love the truth but pardon error.

"Science is about questioning the status quo. Questioning authority".

In the absence of evidence, opinion is indistinguishable from prejudice.

I'm not sure where to start. Your numbers seem ridiculous. Part of the reason is that your subconscious appears to be making typos to shift the numbers dramatically. Another part of the reason may be that your system is fubar somehow (or perhaps having more than 2 cores makes things run much, much slower, which I've certainly seen happen, at least in my mind's eye).

The claim that a trivial addition of Thread::Queue causes rather trivial operations to run 120x slower should certainly motivate some investigation. My investigation shows that the penalty is very close to 2x slower (rather, 2x more CPU, which translates to 'takes about the same elapsed time'), which seems reasonable based on my in-the-mind's-eye expertise.

And that's with the output redirected to nul so there is no competition for the disk head.

That seems silly to me. But, more importantly, if you'd not done that, you could've checked your work and found several mistakes, at least one of which likely would drastically change your numbers.

In any case, I didn't redirect to nul so there was all that competition for the disk head (at least in your mind's eye) and yet I see about 2x slow-down not 120x slow-down (or even the 16x..32x slow-down you'll likely see after you fix your stuff).

I don't have time to scale up to 3.5GB which means that one explanation for the off-by-a-factor-of-8x..16x discrepancy might be some extreme non-linear performance of Thread::Queue (in a case where the performance shouldn't be non-linear).

My testing does show some non-linear slow-down of Thead::Queue which can easily be at least partially overcome in this particular case, but the trend doesn't look even close to me getting the 16x over-all slow-down at 3.5GB (assuming the 'pipe' version actually runs at about the same speed as the corrected trivial 'perl -pe' version, which seems likely but I didn't look at the 'pipe' version at all and still spent more time on this than I really should have right now).

So, if I drop the 'g' from 's/a/A/g' like you did, it makes the 'perl -pe' version run about 10x faster for me. If you got the same speed-up, that'd mean it really runs at roughly the same speed as the 'pipe' version (which means you probably used shorter lines than me and got closer to a 5x speed-up).

I doubt I can replicate your problem with Thread::Queue (based on quick tests on three systems) so you should probably investigate that some more.

The non-linear slow-down of Thread::Queue is interesting and might be worth investigating. It is troubling that throwing in a bunch of 'restarts' via exec( $^X, $0 ) actually makes this code run faster (when given huge input).

I wish I had 4 cores to play with. The Thread::Queue solutions take slightly longer than the trivial one-liner by using two threads and slightly more than twice as much CPU. But that is also what happens with:

perl -pe's/a/A/g' 1kLines1.txt > output1.tmp &
perl -pe's/a/A/g' 1kLines2.txt > output2.tmp; wait
[download]

(yes, even if I redirect both to /dev/null) so using both cores has no benefit at all for this example. If I had 4 cores, I could explore that in more interesting ways.

But the level of processing here is rather trivial which seems likely to be very much unlike the original problem so most of this likely has little bearing on it (based on my saner numbers).

- tye        

"I didn't see any mentions of files so I wasn't going to jump to the conclusion that "huge dataset" means some single huge file or even any files at all."

Sorry, the data is in a single tab-delimited text file. I'm doing something similar to a correlation coefficient on terms assigned to genes in a large spread to generate similarity. For this, I'm using a kappa statistic, which involves pairwise comparisons between all genes and all annotations, it is a very large matrix.

Thanks for the suggestions though. I'll try implementing them and get back to the thread