File Coverage

blib/lib/Thread/Task/Concurrent.pm

Criterion	Covered	Total	%
statement	85	97	87.6
branch	12	26	46.1
condition	4	9	44.4
subroutine	14	15	93.3
pod	6	6	100.0
total	121	153	79.0

package Thread::Task::Concurrent;

use warnings;
use strict;
use 5.010;

my $can_use_threads = $threads::threads;

use threads::shared;

use Thread::Queue;
use Thread::Task::Concurrent::Util qw/unshared_clone/;

my ($tmsg_sub);

if ($can_use_threads) {
    $tmsg_sub = sub { say STDERR '[' . ( $_[1] // threads->tid ) . '] ' . $_[0]; };
} else {
    warn "threads module not loaded, working in serial mode";
    $tmsg_sub = sub { say STDERR '[' . ( $_[1] // "main" ) . '] ' . $_[0]; }
}

use Mouse;

use Mouse::Exporter;

Mouse::Exporter->setup_import_methods( as_is => ['tmsg'] );

our $VERSION = 0.01_05;

has queue         => ( is => 'rw' );
has task          => ( is => 'rw', required => 1 );
has arg           => ( is => 'rw' );
has max_instances => ( is => 'rw', default => 4 );
has threads       => ( is => 'rw' );
has verbose       => ( is => 'rw' );
has result_queue  => ( is => 'rw' );
has finished      => ( is => 'rw' );
has _start_time   => ( is => 'rw' );
has _real_task    => ( is => 'rw' );

sub BUILD {
    my ($self) = @_;

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

    $self->result_queue( Thread::Queue->new() );
}

{
    my $enqueue_finished : shared;
    my $wait : shared;
    my $tasks_running : shared;

    sub start {
        my ($self) = @_;

        tmsg( "starting", "main" ) if ( $self->verbose );
        $self->_start_time(time);

        my $q  = $self->queue;
        my $rq = $self->result_queue;

        my $task = $self->task;
        my $arg  = $self->arg;

        $tasks_running = 0;

        my $real_task = sub {
        ELEMENT:
            while (1) {
                {
                    lock($enqueue_finished);
                    if ( $enqueue_finished && $q->pending == 0 ) {
                        #broadcast as much as possible, so no thread gets stuck
                        lock($wait);
                        cond_broadcast($wait);
                        last ELEMENT;
                    }
                }

                my $i = $q->dequeue_nb;
                unless ( defined($i) ) {
                    lock($wait);
                    cond_wait($wait);
                    next ELEMENT;
                }
                {
                    lock($tasks_running);
                    $tasks_running++;
                }

                tmsg("running task ...") if ( $self->verbose );
                my @result = $task->( $i, $arg );
                $rq->enqueue(@result) if ( @result && @result > 0 );

                {
                    lock($tasks_running);
                    $tasks_running--;

                    tmsg( "task done, tasks running: " . $tasks_running . ", pending: " . $q->pending )
                        if ( $self->verbose );
                }
            }
        };

        $self->_real_task($real_task);
        my @threads;
        for ( my $i = 0; $i < $self->max_instances; $i++ ) {
            # early exit if no threads are loaded/supported
            last unless ($can_use_threads);
            push @threads, threads->create($real_task);

        }
        $self->threads( \@threads );

        return $self;
    }

    sub join {
        my ($self) = @_;

        {
            lock($enqueue_finished);
            $enqueue_finished = 1;
        }

        my $threads = $self->threads;
        for my $t (@$threads) {

            #broadcast as much as possible, so no thread gets stuck
            {
                lock($wait);
                cond_broadcast($wait);
            }
            tmsg( "waiting for thread " . $t->tid, 'main' ) if ( $self->verbose );
            $t->join;
            tmsg( "thread " . $t->tid . " joined successfully", 'main' ) if ( $self->verbose );
        }

        #execute task if we have no thread support (@$threads is empty)
        $self->_real_task->() unless ($can_use_threads);

        $self->finished(1);
        tmsg( "time: " . sprintf( "%dd %dh %dm %ds", ( gmtime( time - $self->_start_time ) )[ 7, 2, 1, 0 ] ),
            'main' )
            if ( $self->verbose );
        $self;
    }

    sub enqueue {
        my ($self) = shift;

        $self->queue->enqueue(@_);
        lock($wait);
        cond_broadcast($wait);
        return $self;
    }

    sub result {
        my ($self) = @_;

        $self->join
            unless ( $self->finished );
        my $rq = $self->result_queue;
        my @results;
        while ( defined( my $item = $rq->dequeue_nb ) ) {
            push @results, unshared_clone($item);
        }

        if ( @results && @results > 0 ) {
            return \@results;
        } else {
            return;
        }
    }
}

sub tmsg { return $tmsg_sub->(@_); }

__PACKAGE__->meta->make_immutable;

1;
__END__