threadingqueue.pl

#!/usr/bin/env perl

### mockup perl code to run a large number of items without filling the queue

use strict;
use warnings;
use 5.014;

use threads;
use threads::shared;
use Thread::Queue;

use String::Random;
use Array::Iterator;

my $parallel = 20; # number of threads
my $total = 100;   # number of mockup items to process
my @log;           # shared results
share(@log);

my $queue = Thread::Queue->new();

# random work items
sub genString {
    my $s = new String::Random;
    return $s->randpattern("CCcccCccCccnnn");
}

# run the queue of items
sub runQueue {
    say "Creating runqueue";

    # create bogus work items
    my @all;
    for ( 0..$total ) {
        push @all, genString;
    }
    my $iter = Array::Iterator->new(\@all);

    # create a limited queue
    my $i = 0;
    while ( my $item = $iter->next ) {
        last if $i++ >= $parallel*2; # limit queue size
        $queue->enqueue( $item );
    }
    say $queue->pending()." items pending";
    # create a number of threads
    say "Creating threads";
    threads->create( {}, 'processItem' ) for ( 1 .. $parallel );

    # wait for all the queues to finish
    my $item;
    while( threads->list(threads::running) != 0 ) {
        for ( $queue->pending()..$parallel ) {
            $queue->enqueue( $iter->next ) if $iter->has_next;
        }
        say "running ".scalar threads->list(threads::running);
        say "qpend ".$queue->pending();
        sleep 2; # always sleep less then min runtime of processItem
    }
}

sub processItem {
    while ( defined ( my $item = $queue->dequeue_nb ) ) {
        my $l = "processing $item with ".threads->tid();
        say $l;
        sleep 3; # remove
        lock @log; #always lock and store at the end
        push @log, $l;
    }
}

runQueue;

map { say } @log;