File Coverage

blib/lib/MapReduce/Framework/Simple.pm

Criterion	Covered	Total	%
statement	69	217	31.8
branch	11	60	18.3
condition	2	13	15.3
subroutine	12	20	60.0
pod	4	4	100.0
total	98	314	31.2

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);

our $VERSION = "0.07";

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);

# 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)];
	}
    }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 $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);
    $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 = _perl_to_msgpack(
		{
		    data => $data->[$k]->[0],
		    code => $stringified_code
		   }
	       );
	    my $result_chil_from_remote = _post_content(
		$data->[$k]->[1],
		'application/x-msgpack; charset=x-user-defined',
		$payload,
		$self->verify_hostname
	       );
	    my $result_with_id;
	    if($result_chil_from_remote->{is_success}){
		my $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 = _perl_to_msgpack(
		{
		    data => $failed_data->[$k]->[0],
		    code => $stringified_code
		   }
	       );
	    my $rand_remote = $succeeded_remotes_list[int(rand($#succeeded_remotes_list))];
	    my $result_chil_from_remote = _post_content(
		$rand_remote,
		'application/x-msgpack; charset=x-user-defined',
		$payload,
		$self->verify_hostname
	       );
	    my $result_with_id;
	    if($result_chil_from_remote->{is_success}){
		my $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;
    unless(defined($worker)){
	$worker = 4;
    }
    unless(defined($port)){
	$port = 5000;
    }
    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{
	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})]];
	    }
	   };
	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 _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__