File Coverage

blib/lib/Thread/Pipeline.pm

Criterion	Covered	Total	%
statement	16	18	88.8
branch			n/a
condition			n/a
subroutine	6	6	100.0
pod			n/a
total	22	24	91.6

package Thread::Pipeline;
{
  $Thread::Pipeline::VERSION = '0.004';
}

# $Id: Pipeline.pm 16 2012-12-28 09:03:33Z [e-mail] $

# NAME: Thread::Pipeline
# ABSTRACT: multithreaded pipeline manager



use 5.010;
use strict;
use warnings;
use utf8;
use Carp;

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



sub new {
    my ($class, $blocks, %opt) = @_;
    my $self :shared = shared_clone {
        blocks      => {},
        out_queue   => Thread::Queue::Any->new(),
        input_ids   => [],
    };
    bless $self, $class;

    if ( ref $blocks eq 'HASH' ) {
        while ( my ($id, $info) = each %$blocks ) {
            $self->add_block( $id => $info );
        }
    }
    elsif ( ref $blocks eq 'ARRAY' ) {
        for my $i ( 0 .. @$blocks/2 - 1 ) {
            my ( $id, $info, $next_id ) = @$blocks[ $i*2 .. $i*2+2 ];
            my %block = %$info;
            $block{main_input} //= 1  if $i == 0;
            $block{out} //= $next_id // '_out';
            $self->add_block( $id => \%block );
        }
    }

    return $self;
}



sub add_block {
    my ($self, $block_id, $block_info, %opt) = @_;

    my $queue :shared = Thread::Queue::Any->new();
    my $block = shared_clone {
        queue => $queue,
    };

    my $threads_num :shared = $block_info->{num_threads} || 1;
    my $thread_sub = sub {
        while (1) {
            # get incoming data block
            my ($in_data) = $queue->dequeue();

            # process it
            my @out_data;
            if ( defined $in_data || $block_info->{need_finalize} ) {
                eval { @out_data = $block_info->{sub}->( $in_data, $self ); 1 }
                or carp "Worker '$block_id' died in thread tid=" . threads->tid() . ": $@";
            }

            # send results to next block
            if ( $block_info->{out} ) {
                for my $item ( @out_data ) {
                    next if !defined $item;
                    $self->enqueue( $item, block => $block_info->{out} );
                }
            }

            # finish work if incoming data was undefined
            last if !defined $in_data;
        }

        lock $threads_num;
        $threads_num --;

        # send undef to next block
        if ( !$threads_num ) {
            $block_info->{post_sub}->()  if $block_info->{post_sub};
            if ( $block_info->{out} && $block_info->{out} ne '_out' ) {
                $self->no_more_data($block_info->{out});
            }
        }

        return;
    };

    my @threads = map { threads->create($thread_sub) } ( 1 .. $threads_num );
    $block->{threads} = shared_clone \@threads;

    $self->{blocks}->{$block_id} = $block;
    push @{ $self->{input_ids} }, $block_id  if $block_info->{main_input};

    return $self;
}



sub enqueue {
    my ($self, $data, %opt) = @_;

    my $ids = $opt{block} || $self->{input_ids};

    for my $block_id ( @{ ref $ids ? $ids : [$ids]  } ) {
        if ( $block_id eq '_out' ) {
            $self->{out_queue}->enqueue($data);
        }
        else {
            my $block = $self->{blocks}->{$block_id};
            croak "Unknown block id: $block_id"  if !$block;
            $block->{queue}->enqueue( $data );
        }
    }

    return $self;
}



sub no_more_data {
    my ($self, $ids) = @_;
    $ids ||= $self->{input_ids};

    for my $block_id ( @{ ref $ids ? $ids : [$ids]  } ) {
        my $num = $self->get_threads_num($block_id);
        my $block = $self->{blocks}->{$block_id};
        $block->{queue}->enqueue( undef )  for ( 1 .. $num );
    }

    return $self;
}




sub get_results {
    my ($self, %opt) = @_;

    for my $block ( values %{ $self->{blocks} } ) {
        for my $thread ( @{ $block->{threads} } ) {
            $thread->join();
        }
    }

    my @result;
    while ( my @items = $self->{out_queue}->dequeue_dontwait() ) {
        push @result, @items;
    }

    return @result;
}



sub get_threads_num {
    my ($self, $block_id) = @_;

    my $block = $self->{blocks}->{$block_id};
    croak "Unknown block id: $block_id"  if !$block;

    return scalar @{ $block->{threads} };
}



1;

__END__