#!/usr/bin/env perl
use strict;
use warnings;

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

use Inline C => Config => CCFLAGSEX => '-O2 -fomit-frame-pointer';
use Inline C => <<'END_OF_C_CODE';

#include <stdint.h>

void num_steps_c( SV* _n, SV* _s )
{
    uint64_t n, input;
    int steps = 0;

    n = input = SvUV(_n);

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

    sv_setuv(_n, n);
    sv_setiv(_s, steps);

    return;
}

END_OF_C_CODE

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 @seqs;

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

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

        $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));
        push @seqs, [ $input, $steps + 1 ] if $steps > 400;
    }
}

sub _collatz {
    my ( $input ) = @_;

    num_steps_c( my $n = $input, my $steps );
    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      => MCE::Candy::out_iter_array(\@seqs),
);

mce_flow_s sub {
    my ( $mce, $chunk_ref, $chunk_id ) = @_;
    collatz_seq($chunk_id, @{ $chunk_ref });
    @seqs > 20
        ? MCE->gather($chunk_id, ( sort { $b->[1] <=> $a->[1] } @seqs )[ 0..19 ])
        : MCE->gather($chunk_id, @seqs);
    @seqs = ();
}, 2, $size;

MCE::Flow->finish; unmap $cache;
@seqs = ( sort { $b->[1] <=> $a->[1]} @seqs )[ 0..19 ];

printf "Collatz(%5d) has sequence length of %3d steps\n", @$_
    for @seqs;