view release on metacpan or  search on metacpan

lib/App/Greple/Grep.pm  view on Meta::CPAN

package App::Greple::Grep::Block {
    use strict;
    sub min    :lvalue { $_[0]->[0] }
    sub max    :lvalue { $_[0]->[1] }
    sub number :lvalue { $_[0]->[2] }
}

package App::Greple::Grep::Match {
    use strict;
    sub min      :lvalue { $_[0]->[0] }
    sub max      :lvalue { $_[0]->[1] }
    sub index    :lvalue { $_[0]->[2] }
    sub callback :lvalue { $_[0]->[3] }
}

package App::Greple::Grep::Result {
    use strict;
    sub block   { $_[0]->[0] }
    sub matched { $_[0]->@[ 1 .. $#{$_[0]} ] }
    sub min     { $_[0]->block->min }
    sub max     { $_[0]->block->max }
    sub number  { $_[0]->block->number }
}

##
## Match multiple patterns in parallel using child processes.
## Each child scans $_ (shared by copy-on-write) with a single
## pattern and returns [from, to, index] triplets in packed binary
## format.  Returns a list indexed by pattern number; undef elements
## mean the pattern was not processed and should be matched in the
## calling process.
##
use constant NO_INDEX => ~0;

our $default_threshold = 1024 * 1024;

sub parallel_match {
    my $self = shift;
    my $patlist = shift;
    my $max = $self->{parallel} // 0;
    return () if $max < 2;
    my $threshold = $ENV{GREPLE_PARALLEL_THRESHOLD} // $default_threshold;
    return () if length() < $threshold;
    my @eligible = grep { not $patlist->[$_]->is_function } keys @$patlist;
    return () if @eligible < 2;
    warn sprintf("parallel_match: %d patterns, max %d processes\n",
		 scalar @eligible, $max) if $debug{m};
    my @result;
    while (my @chunk = splice @eligible, 0, $max) {
	my @child;
	for my $i (@chunk) {
	    my $pat = $patlist->[$i];
	    pipe(my $r, my $w) or last;
	    my $pid = fork;
	    if (not defined $pid) {	# fall back to sequential
		close $r; close $w;
		last;
	    }
	    if ($pid == 0) {
		close $r;
		binmode $w;
		my @p = match_regions(pattern => $pat->regex,
				      group => $self->{group_index},
				      index => $self->{group_index} >= 2);
		my $data = pack 'J*',
		    map { ($_->[0], $_->[1], $_->[2] // NO_INDEX) } @p;
		syswrite $w, $data if length $data;
		close $w;
		POSIX::_exit(0);
	    }
	    close $w;
	    binmode $r;
	    push @child, [ $i, $pid, $r ];
	}
	for (@child) {
	    my($i, $pid, $r) = @$_;
	    my $data = do { local $/; <$r> };
	    close $r;
	    waitpid $pid, 0;
	    next if $? != 0;		# fall back to sequential
	    my @v = unpack 'J*', $data // '';
	    my @p;
	    for (my $j = 0; $j < @v; $j += 3) {
		push @p, $v[$j+2] == NO_INDEX
		    ? [ @v[$j, $j+1] ] : [ @v[$j .. $j+2] ];
	    }
	    $result[$i] = \@p;
	}
    }
    @result;
}

##
## Compute line borders in a child process, overlapping with pattern
## matching in the calling process.  Border positions are independent
## from the match result and required only after the search.
##
sub start_borders {
    my $self = shift;
    return if $self->{BORDERS_CHILD};	# already started
    return if $self->{block}->@*;	# borders not used for --block
    my $threshold = $ENV{GREPLE_PARALLEL_THRESHOLD} // $default_threshold;
    return if length() < $threshold;
    pipe(my $r, my $w) or return;
    my $pid = fork;
    if (not defined $pid) {		# fall back to sequential
	close $r; close $w;
	return;
    }
    if ($pid == 0) {
	close $r;
	binmode $w;
	my $data = pack 'J*', match_borders $self->{border};
	syswrite $w, $data if length $data;
	close $w;
	POSIX::_exit(0);
    }
    close $w;
    binmode $r;
    $self->{BORDERS_CHILD} = [ $pid, $r ];
    warn "started borders child process\n" if $debug{m};
}

sub read_borders {
    my $self = shift;
    my($pid, $r) = @{delete $self->{BORDERS_CHILD}};
    my $data = do { local $/; <$r> };
    close $r;
    waitpid $pid, 0;
    return if $? != 0;			# fall back to sequential
    unpack 'J*', $data // '';
}

sub discard_borders {
    my $self = shift;
    if (my $child = delete $self->{BORDERS_CHILD}) {
	my($pid, $r) = @$child;
	kill 'TERM', $pid;
	close $r;
	waitpid $pid, 0;
    }
    $self;
}

sub prepare {
    my $self = shift;

    local *_ = $self->{text};
    my $pat_holder = $self->{pattern};
    my @blocks;

    $self->{RESULT} = [];
    $self->{BLOCKS} = [];

    ##
    ## build match result list
    ##
    my @result;
    my $positive_count = 0;
    my $group_index_offset = 0;
    my @patlist = $pat_holder->patterns;
    my @parallel = $self->parallel_match(\@patlist);
    while (my($i, $pat) = each @patlist) {
	my($func, @args) = do {
	    if ($pat->is_function) {
		$pat->function;
	    } else {
		Getopt::EX::Func->new(\&match_regions,
				      pattern => $pat->regex,
				      group => $self->{group_index},
				      index => $self->{group_index} >= 2,
		    );
	    }
	};
	my @p = $parallel[$i]
	    ? @{$parallel[$i]}
	    : $func->call(@args, &FILELABEL => $self->{filename});
	if (@p == 0) {
	    ##