#!/usr/bin/perl

use warnings;
use strict;
use threads;
use Thread::Queue;

my $DataQueue = Thread::Queue->new();
my $Queue1 = $DataQueue;
my $Queue2;
my @threads;

while (<>) {
    $DataQueue->enqueue($_);
    $Queue2 = Thread::Queue->new();
    print "Starting thread.\n";
    my $thr = threads->new(\&bubble,$Queue1,$Queue2);
    $Queue1 = $Queue2;
    push @threads, $thr;
}

$DataQueue->enqueue(undef);

for (@threads) {
    print $_->join, " ";
}
print "\n";

sub bubble {
    my ($upstream, $downstream) = @_;
    my $num1 = $upstream->dequeue;
    chomp $num1;
    while (my $num2 = $upstream->dequeue) {
	chomp $num2;
	# Keep the highest value
	print "Comparing $num1 and $num2.\n";
	if ($num1 < $num2) {
	    $downstream->enqueue($num1);
	    $num1 = $num2;
	} else {
	    $downstream->enqueue($num2);
	}
    }
    # Return the highest value seen.
    $downstream->enqueue(undef);
    return $num1;
}