File Coverage

blib/lib/MapReduce/Framework/Simple.pm

Criterion	Covered	Total	%
statement	85	335	25.3
branch	13	92	14.1
condition	3	20	15.0
subroutine	16	25	64.0
pod	4	5	80.0
total	121	477	25.3

package MapReduce::Framework::Simple;
use 5.010001;
use strict;
use warnings;
use B::Deparse;
use Mouse;
use Data::MessagePack;
use Parallel::ForkManager;
use Plack::Request;
use WWW::Mechanize;
use List::Util qw(shuffle);
use Data::Serializer;
use Storable;
no warnings qw(once);
$Storable::Deparse = 1;
$Storable::Eval = 1;
use warnings;

our $VERSION = "0.09";

has 'verify_hostname' => (is => 'rw', isa => 'Int', default => 1);
has 'skip_undef_result' => (is => 'rw', isa => 'Int', default => 1);
has 'warn_discarded_data' => (is => 'rw', isa => 'Int', default => 1);
has 'die_discarded_data' => (is => 'rw', isa => 'Int', default => 0);
has 'worker_log' => (is => 'rw', isa => 'Int', default => 0);
has 'force_plackup' => (is => 'rw', isa => 'Int', default => 0);
has 'server_spec' => (is => 'rw', isa => 'HashRef', default => sub{{cores => 1, clock => 2000, mem => 1}});
has 'worker_num' => (is => 'rw', isa => 'Int', default => 1);
has 'port' => (is => 'rw', isa => 'Int', default => 5000);
has 'path' => (is => 'rw', isa => 'Str', default => '/eval');


# To make load balanced data.
sub create_assigned_data {
    my $self = shift;
    my $data = shift;
    my $servers = shift;
    my $options = shift;
    my $chunk_num = 10;
    my $method = 'volume_uniform';
    if(defined($options)){
	if(defined($options->{chunk_num})){
	    $chunk_num = $options->{chunk_num};
	}
	if(defined($options->{method})){
	    $method = $options->{method};
	}
    }
    my $output;
    if($method eq 'element_shuffle'){

	@$data = shuffle(@$data);
	for(0 .. $#$data){
	    push(@{$output->[$_ % $chunk_num]->[0]},$data->[$_]);
	}
	for(0 .. $#$output){
	    $output->[$_]->[1] = $servers->[$_ % scalar(@$servers)];
	}
    }elsif($method eq 'element_sequential'){
	for(0 .. $#$data){
	    push(@{$output->[$_ % $chunk_num]->[0]},$data->[$_]);
	}
	for(0 .. $#$output){
	    $output->[$_]->[1] = $servers->[$_ % scalar(@$servers)];
	}
    }elsif($method eq 'element_server_cores'){
	@$data = shuffle(@$data);
	my $server_spec = $self->check_server_spec($servers);
	my $tmp_server_list;
	foreach my $key (keys %$server_spec){
	    push(@$tmp_server_list,$key);
	    for(2 .. $server_spec->{$key}->{server_spec}->{cores}){
		push(@$tmp_server_list,$key);
	    }
	}
	for(0 .. $#$data){
	    push(@{$output->[$_ % scalar(@$tmp_server_list)]->[0]},$data->[$_]);
	}
	for(0 .. $#$output){
	    $output->[$_]->[1] = $tmp_server_list->[$_ % scalar(@$tmp_server_list)];
	}
    }elsif($method eq 'element_server_workers'){
	@$data = shuffle(@$data);
	my $server_spec = $self->check_server_spec($servers);
	my $tmp_server_list;
	foreach my $key (keys %$server_spec){
	    push(@$tmp_server_list,$key);
	    for(2 .. $server_spec->{$key}->{server_spec}->{worker_num}){
		push(@$tmp_server_list,$key);
	    }
	}
	for(0 .. $#$data){
	    push(@{$output->[$_ % scalar(@$tmp_server_list)]->[0]},$data->[$_]);
	}
	for(0 .. $#$output){
	    $output->[$_]->[1] = $tmp_server_list->[$_ % scalar(@$tmp_server_list)];
	}
    }elsif($method eq 'element_server_core_clock'){
	@$data = shuffle(@$data);
	my $server_spec = $self->check_server_spec($servers);
	my $tmp_server_list;

	my $tmp_max_cc = 0;
	my $tmp_min_cc = 1_000_000_000_000;
	foreach my $key (keys %$server_spec){
	    my $tmp_cc = ($server_spec->{$key}->{server_spec}->{cores} + 0.0001)
		* ($server_spec->{$key}->{server_spec}->{clock} + 0.0001);
	    if($tmp_max_cc < $tmp_cc){
		$tmp_max_cc = $tmp_cc;
	    }
	    if($tmp_min_cc > $tmp_cc){
		$tmp_min_cc = $tmp_cc;
	    }
	}

	my $pre_ratio = ((log($tmp_max_cc) / log(10))+(log($tmp_max_cc) / log(10)));
	my $ratio = 1/ 10 ** int($pre_ratio / 2);
	foreach my $key (keys %$server_spec){
	    push(@$tmp_server_list,$key);
	    my $tmp_cc = $server_spec->{$key}->{server_spec}->{cores}
		* $server_spec->{$key}->{server_spec}->{clock};
	    my $add_num = int($tmp_cc * $ratio);
	    for(2 .. $add_num){
		push(@$tmp_server_list,$key);
	    }
	}
	for(0 .. $#$data){
	    push(@{$output->[$_ % scalar(@$tmp_server_list)]->[0]},$data->[$_]);
	}
	for(0 .. $#$output){
	    $output->[$_]->[1] = $tmp_server_list->[$_ % scalar(@$tmp_server_list)];
	}
    }else{
	my $mp = Data::MessagePack->new();
	@$data = map $_->[0],
	    sort {$a->[1] <=> $b->[1]} map [$_, bytes::length $mp->pack($_)],
	    @$data;
	for(0 .. $#$data){
	    push(@{$output->[$_ % $chunk_num]->[0]},$data->[$_]);
	}
	for(0 .. $#$output){
	    $output->[$_]->[1] = $servers->[$_ % scalar(@$servers)];
	}
    }
    return($output);
}

# MapReduce client(Master)
sub map_reduce {
    my $self = shift;
    my $data = shift;
    my $mapper_ref = shift;
    my $reducer_ref = shift;
    my $max_proc = shift;
    my $options = shift;
    my $remote_flg = 1;
    if(defined($options) and defined($options->{remote})){
	$remote_flg = $options->{remote};
    }
    my $storable_flg = 0;
    if(defined($options) and defined($options->{storable})){
	$storable_flg = $options->{storable};
    }
    my $stringified_code = B::Deparse->new->coderef2text($mapper_ref);
    my $result;
    my $succeeded_remotes;
    my $failed_remotes;
    my $failed_data;
    my $discarded_data;
    my $pm = Parallel::ForkManager->new($max_proc);

    my $ds;
    if($storable_flg == 1){
	$ds = Data::Serializer->new(
	    serializer => 'Storable',
	    compress => 1
	   );
    }

    $pm->run_on_finish(
	sub {
	    my ($pid, $exit_code, $ident, $exit_signal, $core_dump, $data_structure) = @_;
	    if (defined $data_structure) {
		if($data_structure->{is_success} == 1){
		    $succeeded_remotes->{$data_structure->{remote}} = 1;
		    $result->[$data_structure->{id}] = $data_structure->{result};
		}else{
		    $failed_remotes->{$data_structure->{remote}} = 1;
		    push(@$failed_data,$data_structure->{failed_data});
		    $result->[$data_structure->{id}] = undef;
		}
	    }
	}
       );
    if($remote_flg == 1){
	for(my $k=0; $k <= $#$data; $k++){
	    $pm->start and next;
	    my $payload;
	    if($storable_flg == 1){
		$payload = $ds->serialize(
		    {
			data => $data->[$k]->[0],
			code => $stringified_code
		       }
		   );
	    }else{
		$payload = _perl_to_msgpack(
		    {
			data => $data->[$k]->[0],
			code => $stringified_code
		       }
		   );
	    }
	    my $path_add = '';
	    if($storable_flg == 1){ $path_add = '/STORABLE'; };
	    my $content_type_name = 'application/x-msgpack; charset=x-user-defined';
	    if($storable_flg == 1){ $content_type_name = 'application/octet-stream; charset=x-user-defined'; };
	    my $result_chil_from_remote = _post_content(
		$data->[$k]->[1].$path_add,
		$content_type_name,
		$payload,
		$self->verify_hostname
	       );
	    my $result_with_id;
	    if($result_chil_from_remote->{is_success}){
		my $result_chil;
		if($storable_flg == 1){
		    $result_chil = $ds->deserialize($result_chil_from_remote->{res});
		}else{
		    $result_chil = _msgpack_to_perl($result_chil_from_remote->{res});
		}
		$result_with_id = {id => $k, result => $result_chil->{result}, remote => $data->[$k]->[1], is_success => 1};
	    }else{
		$result_with_id = {id => $k, remote => $data->[$k]->[0], is_success => 0, failed_data => $data->[$k]};
	    }
	    $pm->finish(0,$result_with_id);
	}
    }else{
	for(my $k=0; $k <= $#$data; $k++){
	    $pm->start and next;
	    my $result_chil = $mapper_ref->($data->[$k]);
	    my $result_with_id = {id => $k, result => $result_chil, is_success => 1, remote => 'LOCAL'};
	    $pm->finish(0,$result_with_id);
	}
    }
    $pm->wait_all_children;
    my $result_failover;
    if($remote_flg == 1 and $#$failed_data >= 0){
	my @succeeded_remotes_list;
	foreach my $key (keys %$succeeded_remotes){
	    push(@succeeded_remotes_list,$key);
	}
	my $pm2 = Parallel::ForkManager->new($max_proc);
	$pm2->run_on_finish(
	    sub {
		my ($pid, $exit_code, $ident, $exit_signal, $core_dump, $data_structure) = @_;
		if (defined $data_structure) {
		    if($data_structure->{is_success} == 1){
			$succeeded_remotes->{$data_structure->{remote}} = 1;
			$result_failover->[$data_structure->{id}] = $data_structure->{result};
		    }else{
			$failed_remotes->{$data_structure->{remote}} = 1;
			push(@$discarded_data,$data_structure->{failed_data});
			$result_failover->[$data_structure->{id}] = undef;
		    }
		}
	    }
	   );

	for(my $k=0; $k <= $#$failed_data; $k++){
	    $pm2->start and next;
	    my $payload;
	    if($storable_flg == 1){
		$payload = $ds->serialize(
		{
		    data => $failed_data->[$k]->[0],
		    code => $stringified_code
		   }
	       );
	    }else{
		$payload = _perl_to_msgpack(
		    {
			data => $failed_data->[$k]->[0],
			code => $stringified_code
		       }
		   );
	    }

	    my $path_add = '';
	    if($storable_flg == 1){ $path_add = '/STORABLE' };
	    my $content_type_name = 'application/x-msgpack; charset=x-user-defined';
	    if($storable_flg == 1){ $content_type_name = 'application/octet-stream; charset=x-user-defined'; };
	    my $rand_remote = $succeeded_remotes_list[int(rand($#succeeded_remotes_list))];
	    my $result_chil_from_remote = _post_content(
		$rand_remote.$path_add,
		$content_type_name,
		$payload,
		$self->verify_hostname
	       );
	    my $result_with_id;
	    if($result_chil_from_remote->{is_success}){
		my $result_chil;
		if($storable_flg == 1){
		    $result_chil = $ds->deserialize($result_chil_from_remote->{res});
		}else{
		    $result_chil = _msgpack_to_perl($result_chil_from_remote->{res});
		}
		$result_with_id = {id => $#$data + $k, result => $result_chil->{result}, remote => $rand_remote, is_success => 1};
	    }else{
		$result_with_id = {id => $#$data + $k, remote => $rand_remote, is_success => 0, failed_data => $failed_data->[$k]};
	    }
	    $pm2->finish(0,$result_with_id);
	}
	$pm2->wait_all_children;
    }
    my $result_merged;
    push(@$result_merged,@$result);
    if($#$result_failover >= 0){
	push(@$result_merged,@$result_failover);
    }
    if($#$discarded_data >= 0){
	if($self->die_discarded_data == 1){
	    die "Fatal: Discarded data exist due to remote server couldn't process requested data.\n";
	}elsif($self->warn_discarded_data == 1){
	    warn "Warning: Discarded data exist.\n";
	}
    }
    if($self->skip_undef_result == 1){
	my $result_skip;
	for(0 .. $#$result_merged){
	    if(defined($result_merged->[$_])){
		push(@$result_skip,$result_merged->[$_]);
	    }
	}
	return($reducer_ref->($result_skip));
    }else{
	return($reducer_ref->($result_merged));
    }
}

sub worker {
    my $self = shift;
    my $path = shift;
    my $worker = shift;
    my $port = shift;

    $self->{path} = $path;

    unless(defined($worker)){
	$worker = 4;
    }
    $self->{worker_num} = $worker;

    unless(defined($port)){
	$port = 5000;
    }
    $self->{port} = $port;

    my $rc = eval{
	require Plack::Handler::Starlet;
	1;
    };
    if($rc and $self->force_plackup == 0){
	print "Starting MapReduce Framework Worker by Starlet\n";
	print "Path: $path\nPort: $port\n";
	my $app = $self->load_worker_plack_app($path);
	my $handler = Plack::Handler::Starlet->new(
	    max_workers => $worker,
	    port => $port
	   );
	$handler->run($app);
    }else{
	$self->{worker_num} = 1;
	require Plack::Runner;
	my $runner = Plack::Runner->new;
	print "Starting MapReduce Framework Worker by plackup. The number of workers will be ignored.\n";
	print "Path: $path\nPort: $port\n";
	my $app = $self->load_worker_plack_app($path);
	$runner->parse_options('--port',$port);
	$runner->run($app);
    }
}

sub load_worker_plack_app {
    my $self = shift;
    my $path = shift;
    my $app = sub {
	my $env = shift;
	my $req = Plack::Request->new($env);
	if($self->worker_log == 1){
	    print "REQ,$$,".$req->address.',';
	    my @tar = localtime(time());
	    printf(
		"%04d-%02d-%02d %02d:%02d:%02d",
		$tar[5]+1900,$tar[4]+1,$tar[3],$tar[2],$tar[1],$tar[0]
	       );
	    print "\n";
	}
	my $response = {
	    $path => sub {
		my $msg_req = $req->content //
		    return [400,['Content-Type' => 'text/plain'],['Content body required.']];
		my $perl_req = _msgpack_to_perl($msg_req) //
		    return [400,['Content-Type' => 'text/plain'],['Valid MessagePack required']];
		my $data = $perl_req->{data};
		my $code_text = $perl_req->{code};
		my $code_ref;
		eval('$code_ref = sub '.$code_text.';');
		my $result = $code_ref->($data);
		return [200,['Content-Type' => 'application/x-msgpack; charset=x-user-defined'],[_perl_to_msgpack({result => $result})]];
	    },
	    $path.'/server_spec' => sub {
		my $server_spec = $self->{server_spec};
		$server_spec->{worker_num} = $self->{worker_num};
		$server_spec->{port} = $self->{port};
		$server_spec->{path} = $self->{path};

		return [200,['Content-Type' => 'application/x-msgpack; charset=x-user-defined'],[_perl_to_msgpack({server_spec => $server_spec})]];
	    },
	    $path.'/STORABLE' => sub {
		my $ds = Data::Serializer->new(
		    serializer => 'Storable',
		    compress => 1
		   );
		my $msg_req = $req->content //
		    return [400,['Content-Type' => 'text/plain'],['Content body required.']];
		my $perl_req = $ds->deserialize($msg_req) //
		    return [400,['Content-Type' => 'text/plain'],['Valid MessagePack required']];
		my $data = $perl_req->{data};
		my $code_text = $perl_req->{code};
		my $code_ref;
		eval('$code_ref = sub '.$code_text.';');
		my $result_pre = $code_ref->($data);
		my $result = $ds->serialize({result => $result_pre});
		return [200,['Content-Type' => 'application/octet-stream; charset=x-user-defined'],[$result]];
	    },
	   };
	if($self->worker_log == 1){
	    print "END,$$,".$req->address.',';
	    my @tar = localtime(time());
	    printf(
		"%04d-%02d-%02d %02d:%02d:%02d",
		$tar[5]+1900,$tar[4]+1,$tar[3],$tar[2],$tar[1],$tar[0]
	       );
	    print "\n";
	}
	if(defined($response->{$env->{PATH_INFO}})){
	    return $response->{$env->{PATH_INFO}}->();
	}else{
	    return [404,['Content-Type' => 'text/plain'],['Not Found']];
	}

    };
    return($app);
}

sub check_server_spec {
    my $self = shift;
    my $server_list = shift;
    my $output;
    for(my $k=0; $k <= $#$server_list; $k++){
	my $result_chil_from_remote = _post_content(
	    $server_list->[$k] . '/server_spec',
	    'application/x-msgpack; charset=x-user-defined',
	    '',
	    $self->verify_hostname
	   );
	if($result_chil_from_remote->{is_success}){
	    my $tmp_spec = _msgpack_to_perl($result_chil_from_remote->{res});
	    $output->{$server_list->[$k]} = $tmp_spec;
	}
    }
    return $output;
}

sub _post_content {
    my $url = shift;
    my $content_type = shift;
    my $data = shift;
    my $ssl_opt = shift;
    my $ua = WWW::Mechanize->new(
	ssl_opts => {
	    verify_hostname => $ssl_opt
	   }
       );
    my $is_success = 1;
    eval{
	$ua->post($url,'Content-Type' => $content_type, Content => $data);
    };
    if($@){
	$is_success = 0;
    }
    my $res = $ua->content();
    return {res => $res, is_success => $is_success};
}

sub _perl_to_msgpack {
    my $data = shift;
    my $msgpack = Data::MessagePack->new();
    my $packed = $msgpack->pack($data);
    return($packed);
}

sub _msgpack_to_perl {
    my $msg_text = shift;
    my $msgpack = Data::MessagePack->new();
    my $unpacked = $msgpack->unpack($msg_text);
    return($unpacked);
}



__PACKAGE__->meta->make_immutable();

1;
__END__