Hm. Even ignoring simple errors ( naming the passed $sth as $db_source, and then doing nothing with it), I'm not at all sure than your sample code makes any sense at all.

There are more fundemental coding issues: like you finish the sth and close the connection as soon as you've started the threads, which inevitably is going to be before they've had a chance to make much use of them. If it was ever going to work, you'd have to do that after you've joined the threads.

What do you expect to happen when you retrieve the results from a statement handle on multiple threads?

Assuming for a moment that DBI and MySQL had no problems with you calling from multiple threads, then there are two possibilities:

• Each thread retrieves its own copy of every row of data.

  This seems unlikely as that would require DBI to track what data had been given to each thread.

• Each thread retrieves the next available row of data. Ie. Each row is only seen by one thread.

  Assuming it worked at all--I don't have MySQL to try--this seems like the most likely scenario.

  But then that doesn't make much sense if each thread is talking to a different server. And you would have no control over which rows which thread would get. Unless all the servers are identical, how could whatever this data is, be usable with whichever server it gets random assigned to?

Finally, if your concern is that populating the queue with all the returned data will consume to much memory, don't do it all at once. Instead, feed the queue slowly:

use DBI;
use threads;
use Thead::Queue;

my $Q = new Thread::Queue;

my $dbh = DBI->connect(...);
my $sth = $dbh->prepare("SELECT ...");
$sth->execute();


$thr1 = threads->new(\&some_sub, $Q );
$thr2 = threads->new(\&some_sub, $Q );

while( my $ref = $sth->fetchrow_hashref() ) {
   $Q->enqueue( join $;, %$ref );
   sleep 1 while $Q->pending > 10;
}

$Q->enqueue( (undef) x 2 );

$thr1->join; $thr2-join;

$sth->finish();
$dbh->disconnect();

sub some_sub {
    my ( $Q ) = @_;
    while( my %row = split $;, $Q->dequeue ) {
        #...
    } 
}
[download]