#!/usr/bin/env perl
use strict;
use warnings;
use feature qw/say/;

use File::Map qw/map_anonymous unmap/;
use Time::HiRes qw/usleep/;
use MCE::Flow;

my $size = shift || 1e6;

$size = 1e6 if $size < 1e6;  # minimum
$size = 1e9 if $size > 1e9;  # maximum

map_anonymous my $cache, $size * 2 + 2, 'shared';

# fill cache with zeroes
substr($cache, 0, $size * 2 + 2, pack('s', 0) x ( $size + 1 ));

# local to workers and the manager process
my ( $length, $number ) = ( 0, 0 );

sub collatz_longest {
    my ( $chunk_id, $seq_beg, $seq_end ) = @_;
    my ( $n, $steps, $tmp );

    for my $input ( $seq_beg .. $seq_end ) {
        $n = $input, $steps = 0;

        $n % 2 ? ( $steps += 2, $n = (3 * $n + 1) >> 1 )
               : ( $steps += 1, $n = $n >> 1 )
            while $n != 1 && $n >= $input;

        $tmp = unpack('s', substr($cache, $n * 2, 2));

        # another worker with a lesser chunk_id is not yet
        # completed processing $n, so pause a little

        # if ( $tmp == 0 && $chunk_id > 1 ) {
        #     do {
        #         usleep 100;
        #         $tmp = unpack('s', substr($cache, $n * 2, 2));
        #     } while ( $tmp == 0 );
        # }

        # do this instead (faster): compute $n if cache miss
        $tmp = _collatz($n) if $tmp == 0 && $chunk_id > 1;

        substr($cache, $input * 2, 2, pack('s', $steps += $tmp));
        $length = $steps, $number = $input if $steps > $length;
    }
}

sub _collatz {
    my ( $input ) = @_;
    my ( $n, $steps ) = ( $input, 0 );

    $n % 2 ? ( $steps += 2, $n = (3 * $n + 1) >> 1 )
           : ( $steps += 1, $n = $n >> 1 )
        while $n != 1 && $n >= $input;

    my $tmp = unpack('s', substr($cache, $n * 2, 2));

    $tmp = _collatz($n) if $tmp == 0;
    substr($cache, $input * 2, 2, pack('s', $steps += $tmp));

    return $steps
}

my $chunk_size;

$chunk_size  = int( $size / MCE::Util::get_ncpu() / 80 + 1 );
$chunk_size += 1 if $chunk_size % 2;

MCE::Flow->init(
    max_workers => MCE::Util::get_ncpu(),
    chunk_size  => $chunk_size,  # specify 2500 if pausing above
    bounds_only => 1,
    gather      => sub {
        $length = $_[0], $number = $_[1] if $_[0] > $length;
    },
    user_end    => sub {
        MCE->gather($length, $number);
    },
);

mce_flow_s sub {
    my ( $mce, $seq_ref, $chunk_id ) = @_;
    collatz_longest($chunk_id, @{ $seq_ref });
}, 2, $size;

MCE::Flow->finish; unmap $cache;

say "Longest Collatz (index and value)";
say "$number $length";