#! perl -slw
use strict;
use threads;
use threads::shared;
use Time::HiRes qw[ time sleep ];
use Thread::Conveyor;

our $O //= 'memory';
our $N //= 1e4;
our $T //= 4; ++$T; $T &= ~1;
our $SIZE //= 50;

my $Q1_n = Thread::Conveyor->new( { maxboxes => $SIZE, optimize => $O });
my $Qn_n = Thread::Conveyor->new( { maxboxes => $SIZE, optimize => $O });
my $Qn_1 = Thread::Conveyor->new( { maxboxes => $SIZE, optimize => $O });

my $done1 :shared = 0;
my $done2 :shared = 0;

my @t1 = map async( sub{ $Qn_n->put( $_ ) while defined( $_ = $Q1_n->take ); lock $done1; ++$done1; } ), 1 .. $T/2;
my @t2 = map async( sub{ $Qn_1->put( $_ ) while defined( $_ = $Qn_n->take ); lock $done2; ++$done2; } ), 1 .. $T/2;

my $bits :shared = chr(0); $bits x= $N/ 8 + 1;
my $t = async{
    while( defined( $_ = $Qn_1->take ) ) {
        warn "value duplicated" if vec( $bits, $_, 1 );
        vec( $bits, $_, 1 ) = 1;
    }
};

my $start = time;

$Q1_n->put( $_ ) for 1 .. $N;
$Q1_n->put( (undef) x ($T/2) ); sleep 0.01 while $done1 < @t1;
$Qn_n->put( (undef) x ($T/2) );  sleep 0.01 while $done2 < @t2;
$Qn_1->put( undef );             sleep 0.01 until $t->is_joinable;

my $stop = time;

my $b = unpack '%32b*', $bits;
die "NOK: $b : \n" unless $b == $N;

printf "$N items by $T threads via three Qs size $SIZE in %.6f seconds\n",
    $stop - $start;

sleep 10;

__END__
C:\test>perl async\Q.pm -N=1e4 -T=2 -SIZE=40
1e4 items by 2 threads via three Qs size 40 in 5.458000 seconds

C:\test>t-TCcrap -N=100 -T=2 -SIZE=40
100 items by 2 threads via three Qs size 40 in 0.148909 seconds

C:\test>t-TCcrap -N=1e4 -T=2 -SIZE=40
1e4 items by 2 threads via three Qs size 40 in 11.035008 seconds

C:\test>t-TCcrap -N=1e5 -T=2 -SIZE=40
value duplicated at C:\test\t-TCcrap.pl line 26.
value duplicated at C:\test\t-TCcrap.pl line 26.
Thread 1 terminated abnormally: Can't use an undefined value as an ARRAY reference at C:/Perl64/site/lib/Thread/Tie/Array.pm (loaded on demand from offset 1939 for 176 bytes) line 75.
Terminating on signal SIGINT(2)

C:\test>t-TCcrap -N=100 -T=2 -SIZE=4
100 items by 2 threads via three Qs size 4 in 0.148934 seconds
Terminating on signal SIGINT(2)

C:\test>t-TCcrap -N=100 -T=4 -SIZE=4
Terminating on signal SIGINT(2)

C:\test>t-TCcrap -N=100 -T=4 -SIZE=4
Terminating on signal SIGINT(2)

C:\test>t-TCcrap -N=100 -T=3 -SIZE=4
Terminating on signal SIGINT(2)

C:\test>t-TCcrap -N=100 -T=3 -SIZE=40
Thread 1 terminated abnormally: Can't use an undefined value as an ARRAY reference at C:/Perl64/site/lib/Thread/Tie/Array.pm (loaded on demand from offset 1939 for 176 bytes) line 75.
Terminating on signal SIGINT(2)

C:\test>t-TCcrap -N=100 -T=3 -SIZE=40
value duplicated at C:\test\t-TCcrap.pl line 26.
value duplicated at C:\test\t-TCcrap.pl line 26.
Terminating on signal SIGINT(2)

C:\test>t-TCcrap -N=1e5 -T=2 -SIZE=40
value duplicated at C:\test\t-TCcrap.pl line 26.
value duplicated at C:\test\t-TCcrap.pl line 26.
Thread 1 terminated abnormally: Can't use an undefined value as an ARRAY reference at C:/Perl64/site/lib/Thread/Tie/Array.pm (loaded on demand from offset 1939 for 176 bytes) line 75.
Terminating on signal SIGINT(2)