Your design means that you are (attempting) to store a huge amount of data, thousands of files, into shared memory. This design doesn't make any sense. Shared memory is comparatively slow, because of locking considerations and has high memory usage in order that you can access it from multiple threads. And that is all entirely unnecessary,

Rather than loading the contents of the files in your directory traversal thread and then sharing them with all the other threads (so that just one of them can process each of them), the smarter and more efficient design would be to share the names of the files, and have each processing thread: read a path from the queue; slurp the file; process; and then discard it. This will save huge amounts of shared memory contention and speed your throughput, without exhausting memory.

Here's a quick example that gets the paths of all the .pl files in the sub-tree below the CWD, and processes through them using 4 threads counting the mys they find, before finally totalling the sub-counts and printing out the final tally:

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

sub worker {
    my $Q = shift;
    my $count = 0;
    while( my $path = $Q->dequeue ) {
        chomp $path;
        my $file = do{ local( @ARGV, $/ ) = $path; <> };
        $count += () = $file =~ m[my]g;
    }
    return $count;
}


our $THREADS //= 4;

my $Q = new Thread::Queue;

my @workers = map{
    threads->create( \&worker, $Q );
} 1 .. $THREADS;

open DIR, '-|', q[ dir /s /b *.pl ] or die $!;
$Q->enqueue( <DIR> );
close DIR;

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

my $count = 0;

$count += $_->join for @workers;
print "Found $count 'my's";

__END__
C:\test>921522-2.pl
Found 20825 'my's
[download]

 while ( my $bl = $$q->dequeue_nb()){
[download]

There is almost no reason to ever use ->dequeue_nb() in a worker thread. Enqueue undef to tell a worker thread to quit, and enqueue a value (or an arrayref of values for bulk processing) when the worker should do something.

sub slurp_directory_as_bin{
...
[download]

Personally, if you have that many files, I would consider to either store all the file names in a database, or to spawn an external process (like ls -lR or dir /b /s) to read in all the filenames. With the database you get more flexibility to slice and dice your dataset, and if it doesn't change (that often), having convenient access makes all the other things more convenient too. With the external process, you can start processing the first files in the "directory reader thread" while it still reads in more files.

I'd never seen any documentation mentioned that enqueue undef did that, thanks for highlighting it.

There is no documentation mentioning that because you have to do all of this yourself. You will need to make the receiving threads quit once they read a undef from their input queue.

 > 3. previous attempts have hit major stability and
 > time snags, even at the prototyping stage due to the
 > sheer volume of files that make up a comprehensive sample

I notice (based on the "F:/" pathname) that you're on Win32.

You have a File::Find::find like recursive file processing part in your code. This is always going to be slower than necesary on Win32 when coded in Perl.

Consider using/writing some C/XS that generates the file list and avoids all the unnecesary stat (-d !) calls by using FindNextFile().

Also consider using forks over threads. They're easier on Win32 than you might think.

Take a look at qfind.c and peg in my CPAN directory for ideas:

Try comparing the time taken for qfind to generate a file list compared to a pure Perl solution eg.

c:\> perl -e "${^WIN32_SLOPPY_STAT}=1; use Time::HiRes; $start = Time::HiRes::time; open Q, 'qfind.exe |'; while (<Q>) {}; close Q; print 'Took ', (Time::HiRes::time - $start)"

c:\> perl -e "${^WIN32_SLOPPY_STAT}=1; use Time::HiRes; use File::Find; $start = Time::HiRes::time; File::Find::find(sub { }, '.'); print 'Took ', (Time::HiRes::time - $start)"

On my Perl source directory of ~10_000 files this is <0.3 sec vs 1.7 sec. I suspect on your 1.2 million files this gives a *considerable* speed up.

Oh, and make sure you  BEGIN { ${^WIN32_SLOPPY_STAT} = 1 }; at the top of your code!

Good luck.

qfind is interesting and quite fast, but given that the OP is talking about slurping the contents of millions of image files, the time taken to produce the list of those files is likely to be completely insignificant.

---

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 think it's probably moot though since you're almost certainly going to be IO bound, not cpu bound.

Make sure you're solving the right problem before adding infrastructure :)

I am just wrapping up a project *very* similar to this right now. Except that the environment is Ubuntu Linux.

I used forks and forks::shared which mimics the threads interface.

The whole thing ran in the Amazon cloud, copying files from one S3 bucket to another S3 bucket. This meant that I was also able to scale this thing out across multiple machines, copying files together.

I wanted to be able to stop/restart/reset the thing at will, so I created a database with a "jobs" table that looked something like this:

create table job_lists (
  job_list_id  bigint unsigned not null primary key auto_increment,
  name         varchar(100) not null,
  unique(name)
) engine=innodb charset=utf8;

create table jobs (
  job_id         bigint unsigned not null primary key auto_increment,
  job_list_id    bigint unsigned not null,
  reference_key  varchar(100) not null,
  data           text,
  is_started     tinyint not null default 0,
  started_on     timestamp null,
  is_completed   tinyint not null default 0,
  completed_on   timestamp null,
  unique(job_list_id, reference_key),
  foreign key fk_jobs_to_job_lists (job_list_id)
    references job_lists (job_list_id) on delete cascade
) engine=innodb charset=utf8;
[download]

The first step was to get a list of all the files to be copied, and them as records in the "jobs" table. I call this the "Enqueue" step.

The next step was to create a webservice running under Plack which has 2 functions: get_jobs (returns a list of new job data as JSON) and update_jobs - (given a JSON array of job ids, marks them all as "is_completed=1, completed_on=now()").

Finally the "Dequeue" step. A simple perl script which loops through the following:

• Fetch new jobs from the webservice if we don't have enough to distribute to our worker threads.
• Spawn threads until we have enough.
• If there are finished jobs, ask the webservice to mark them as completed.

#!/usr/bin/perl -w

# Basically:
use strict;
use warnings 'all';
use forks;
use forks::shared;
use Time::HiRes 'gettimeofday';

my $max_workers = 10;
my $IS_RUNNING : shared = 1;
my @new_jobs : shared = ( );
my @finished_jobs : shared = ( );
my $finished_count : shared = 0;
my $job_chunk_size = 10;
my $start_time;

# Handle signals:
$SIG{INT} = $SIG{TERM} = sub {
  SCOPE: {
    lock($IS_RUNNING);
    $IS_RUNNING = 0;
  };
};

# Main loop:
while( $IS_RUNNING )
{
  # Create some workers if we don't have enough:
  my $running_procs = scalar( eval { $_->is_running } threads->list );
  for( $running_procs..$max_workers )
  {
    threads->create( \&worker );
  }# end for()

  # Get some jobs if we don't have enough:
  if( scalar(@new_jobs) < $max_workers * $job_chunk_size )
  {
    if( my @unfiled_jobs = get_next_chunk_of_jobs() )
    {
      SCOPE: {
        lock(@new_jobs);
        push @new_jobs, @unfiled_jobs;
      };
    }# end if()
  }# end if()

  $start_time ||= gettimeofday();

  # Mark-as-completed any jobs which are finished:
  if( @finished_jobs )
  {
    my @out = ( );
    SCOPE: {
      lock(@finished_jobs);
      @out = splice(@finished_jobs, 0, scalar(@finished_jobs));
    };
    mark_as_completed( @out );
  }# end if()

  # Print the rate at which we are finishing work:
  my $diff = gettimeofday() - $start_time;
  my $rate = $finished_count / $diff;
  warn "[Main] $finished_count jobs done in $diff sec: $rate/sec\n";

  # And wait a moment...:
  sleep(5);
}# end while()

# We are shutting down...:
warn "\nShutting down...\n";
map { eval { $_->join } } threads->list;


# This is what the worker processes do:
sub worker
{
  my $tid = threads->tid;
  CHUNK: while( $IS_RUNNING )
  {
    my @jobs = (sub{
      SCOPE: {
        lock(@new_jobs);
        return splice(@new_jobs, 0, $job_chunk_size);
      };
    })->();

    # Did we get any jobs?:
    unless( @jobs )
    {
      sleep(4);
      next CHUNK;
    }# end unless()
  
    # Process the jobs:
    map { process_job( $_ ) } @jobs;

    # Finished with this chunk of jobs:
    SCOPE: {
      lock(@finished_jobs);
      push @finished_jobs, map { $_->{job_id} } @jobs;
      lock($finished_count);
      $finished_count += scalar(@jobs);
    };
  }# end while()
}# end worker()


# Your code here:
sub process_job
{
  my ($job) = @_;

  # Do your thing here...
}# end process_job()
[download]

Over the weekend, I processed some 4Million images (several terabytes) in about 4 hours using a script very similar to this.

That is a work of thunderous beauty and I thank you tremendously for sharing it. Admittedly I only understand about 2/3rds of it right now but it's about time I move on from begginer level syntax, as I believe you've saved me several weekends of effort. Thanks again.

package MTD;
use strict;
use warnings;
our $VERSION = "0.1";

use threads;
use Thread::Queue;

sub weave_dir{ #  MTD::weave_dir($directory,$limit,$queueobjectref,\&f
+unction,$functionvar);
    my ($directory,$limit,$q,$function,$var) = @_;
    
    unless(-d $directory){die("$directory is not a directory; weave_di
+r gave up on [$!]");}
    
    #what happens if this process is called by another process which h
+as it's own collection of threads ?
    if(scalar(threads->list()) != 0 ){ print("WARNING: weave_dir is st
+arting with threads already in existance, no guarantee what will happ
+en\n");}

    #unless we have an existing queue variable being passed to us, mak
+e one. 
    unless($q){
        $q = Thread::Queue->new();    #create shared queue
    }
        
    my $lister = threads->create(\&enqueue_directorylist,$directory,\$
+q,1); # create directory listing thread.
    
    while( scalar(threads->list()) < $limit){threads->create(\&qnanny,
+\$q,$function,$var);} # create a queue watching process so long as we
+'re under limit for threads. 
    
    #close the listing process and assign it's opportunity cost thread
+ to another queue processing function.
    $lister->join();
    undef $lister;
    threads->create(\&qnanny,\$q,$function,$var);
            
    my @threads = threads->list(); #get listing of thread objects curr
+ently in existance (potential hazard when other threads from calling 
+process are running ?) 

    #dump undef into the queue for each thread running,nanny thread st
+ops when it recieves undef.
    foreach my $t (@threads){
        $q->enqueue(undef);
    }
    
    #actually join all the weave_dir threads. 
    foreach my $z (@threads){
        #print("joining $z\n");
        
        $z->join();
    }
    
    #print("done");    
}

sub qnanny{ # qnanny($queueobject,$functiontouse,$singlescalarvariable
+topasstothefunctiontouse);
    my ($q,$func,$var) = @_;
    #while there objects in this queue and if it's an actual value as 
+opposed to undef
    while (my $r = $$q->dequeue()){ 
        #pass a reference to the object in the queue to the function s
+pecified, along with a single scalar variable.
        $func->(\$r,$var);
    }
    #print(threads->tid() . " is terminating\n");
    
    #we're done
    threads->exit();
}

sub enqueue_directorylist{ # enqueue_directorylist($director,$queueref
+erence,$recursivemode);
    my ($dir,$queue,$r) = @_;
    
    my @subdirs = ();# array for list of subdirectories found in the d
+irectory, only used in [r]ecursive mode
    opendir(DIR,$dir) or die "could not open $dir: $!"; # create the d
+irectory sequence 

    while (my $file = readdir(DIR)){    #while we have a file to evalu
+ate
        
        next if ($file eq '.'|| $file eq '..' ); #skip self and parent
+ dir
        next if ($file eq 'RECYCLER' || $file eq 'System Volume Inform
+ation'); #skip problem directories on windows root paths

        #if the current specimen is a directory
        if (-d $dir.$file){
            #if in recursive mode
            if ($r){
                #put the reference to the subdirectory in storage for 
+later
                push (@subdirs, $dir.$file.'/');    
            } # otherwise
        next; #dont try to process (yet)
        }
    #if we got this far, add the full file path to the directory
    $$queue->enqueue("$dir$file");
    }
    #repeat the above until.... 
    foreach my $sdir (@subdirs){
        enqueue_directorylist($sdir,$queue,$r);
    }
}


return 1;
[download]

use MTD;
use strict;
use warnings;

use threads;
use threads::shared;
use Thread::Semaphore;

main();
sub main{
    my $directory = q#F:/#;
    my    $limit = 4;
    my $queueobjectref = undef;
    my %fvar :shared;
    my @ar :shared;
    %fvar = (
        t_sem => Thread::Semaphore->new(),
        array => \@ar,
    );
        
    MTD::weave_dir($directory,$limit,$queueobjectref,\&push_fname,\%fv
+ar);
    sleep(1);
    foreach my $i (@{$fvar{array}}){
        print("$i\n");
    }
}

sub push_fname{
    my($filename,$var) = @_;
    $$var{t_sem}->down();    
    push(@{$$var{array}}, $$filename); # reference to an array within 
+a hash reference as an array
    $$var{t_sem}->up();    
}
[download]