use XML::Simple;
my @xmlOptsO = (KeepRoot=>1, KeyAttr=>[]);

use tests;
my ($tests, @workers) = tests->new();

$tests->enqueue(XMLout("<system><job><step name='cleanup'><nextjob>0:1
+</nextjob></job></system>", @xmlOptsO)); #start
my $system = $test->dequeue(); #complete
[download]

package tests;
#constructor
sub new {
    my $class = shift;

    #queues
    my @testStartQueue :shared;
    my $testStartLock :shared;
    my @testDoneQueue :shared;
    my $testDoneLock :shared;

    my %self :shared = (
        'testStartQueue' => \@testStartQueue,
        'testStartLock' => \$testStartLock,
        'testDoneQueue' => \@testDoneQueue,
        'testDoneLock' => \$testDoneLock,
    );
    bless(\%self, $class);

    my @workers;
    for (my $x = 0; $x < 5; $x++) {
        push(@workers, threads->create("executetest", \%self)); #testi
+ng workers
    }

    return (\%self, @workers);
}

#TEST START-----------------------------------------------------------
+--
sub enqueue {
    my $self = shift;
    return $self->_enqueue('testStartLock', 'testStartQueue', @_);
}

sub _tsdequeue {
    my $self = shift;
    return $self->_dequeue('testStartLock', 'testStartQueue', @_);
}
#TEST START-----------------------------------------------------------
+--

#TEST DONE------------------------------------------------------------
+-
sub _tdenqueue {
    my $self = shift;
    return $self->_enqueue('testDoneLock', 'testDoneQueue', @_);
}

sub dequeue {
    my $self = shift;
    return $self->_dequeue('testDoneLock', 'testDoneQueue', @_);
}
#TEST DONE------------------------------------------------------------
+-

#Methods
sub end {
    my $self = shift;
    lock ${$self->{'testStartLock'}};
    lock(${$self->{'testDoneLock'}});
    # No more data is coming
    $$self{'ENDED'} = 1;
    # Try to release at least one blocked thread
    cond_signal(${$self->{'testStartLock'}});
    cond_signal(${$self->{'testDoneLock'}});
    return;
}

sub _enqueue {
    my $self = shift;
    my $lock = shift;
    my $queue = shift;

    unless ($$self{'ENDED'}) {
        lock(${$self->{$lock}});
        push(@{$self->{$queue}}, map { shared_clone($_) } @_) and cond
+_signal(${$self->{$lock}});
    }
    return
}

sub _dequeue {
    my $self = shift;
    my $lock = shift;
    my $queue = shift;

    lock(${$self->{$lock}});
    my $count = @_ ? $self->_validate_count(shift) : 1;

    cond_wait(${$self->{$lock}}) while ((@{$self->{$queue}} < $count) 
+&& ! $$self{'ENDED'});
    cond_signal(${$self->{$lock}}) if ((@{$self->{$queue}} > $count) |
+| $$self{'ENDED'});

    # Return single item
    return shift(@{$self->{$queue}}) if ($count == 1);

    # Return multiple items
    my @items;
    for (1..$count) {
        last if (! @{$self->{$queue}}); #this should only happen in th
+e event there are either enough values or end has been called.
        push(@items, shift(@{$self->{$queue}}));
    }
    return @items;
}

#THREAD
sub executetest {
    local *__ANON__ = 'executetest';
    my $self = shift;
    print "SELF: $self\n";

    logLine("Thread executetest [".threads->self->tid()."] start");
    while (defined(my $system = $self->_tsdequeue())) { #test start qu
+eue
        logLine("processing executetest");
        my $test = (split(":",$system->{job}->[0]->{nextjob}->[0]))[0]
+; #test number
        my $testname = $system->{job}->[0]->{step}->[$test]->{name}; #
+test name

        logLine("Starting test [$testname]");
        $system = $self->{&{\&{$testname}}}->($system); #test

        logLine("Completed test [$testname]");
        $self->_tdenqueue($system); #test done queue
    }
    croak("Thread executetest [".threads->self->tid(). "] end\n");
    return "executetest";
}

sub logLine {
    print $_[0]."\n";
}

#example test
sub cleanup {
    $self = shift; #fails
    print "SELF: $self\n";
    my $system = $_[0];

    $self->logLine("MESSAGE"); #here there are actually semaphore call
+s which fail b/c '$self' is "".

    return $system;
}
[download]