<commit_before># Licensed to the Apache Software Foundation (ASF) under one or more
# contributor license agreements.  See the NOTICE file distributed with
# this work for additional information regarding copyright ownership.
# The ASF licenses this file to You under the Apache License, Version 2.0
# (the "License"); you may not use this file except in compliance with
# the License.  You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

use strict;
use warnings;

package LucyX::Remote::ClusterSearcher;
BEGIN { our @ISA = qw( Lucy::Search::Searcher ) }
use Carp;
use Storable qw( nfreeze thaw );
use Scalar::Util qw( reftype );

# Inside-out member vars.
our %shards;
our %num_shards;
our %password;
our %socks;
our %starts;
our %doc_max;

use IO::Socket::INET;

sub new {
    my ( $either, %args ) = @_;
    my $shards   = delete $args{shards};
    my $password = delete $args{password};
    my $self     = $either->SUPER::new(%args);
    confess("'shards' must be an arrayref")
        unless reftype($shards) eq 'ARRAY';
    $shards{$$self}     = $shards;
    $num_shards{$$self} = scalar @$shards;
    $password{$$self}   = $password;

    # Establish connections.
    my $socks = $socks{$$self} = [];
    for my $shard (@$shards) {
        my $sock = IO::Socket::INET->new(
            PeerAddr => $shard,
            Proto    => 'tcp',
        );
        confess("No socket: $!") unless $sock;
        push @$socks, $sock;
    }

    # Handshake with servers.
    my %handshake_args = ( password => <PASSWORD>, _action => 'handshake' );
    my $responses = $self->_multi_rpc( \%handshake_args );
    for my $response (@$responses) {
        confess unless $response;
    }

    # Derive doc_max and relative start offsets.
    my $doc_max_responses = $self->_multi_rpc( { _action => 'doc_max' } );
    my $doc_max = 0;
    my @starts;
    for my $shard_doc_max (@$doc_max_responses) {
        push @starts, $doc_max;
        $doc_max += $shard_doc_max;
    }
    $starts{$$self} = Lucy::Object::I32Array->new( ints => \@starts );
    $doc_max{$$self} = $doc_max;

    return $self;
}

sub DESTROY {
    my $self = shift;
    $self->close if defined $socks{$$self};
    delete $shards{$$self};
    delete $num_shards{$$self};
    delete $password{$$self};
    delete $socks{$$self};
    delete $starts{$$self};
    delete $doc_max{$$self};
    $self->SUPER::DESTROY;
}

# Send a remote procedure call to all shards.
sub _multi_rpc {
    my ( $self, $args ) = @_;
    my $num_shards = $num_shards{$$self};
    my $request    = $self->_serialize_request($args);
    for ( my $i = 0; $i < $num_shards; $i++ ) {
        $self->_send_request_to_shard( $i, $request );
    }

    # Bail out if we're either closing or shutting down the server remotely.
    return if $args->{_action} eq 'done';
    return if $args->{_action} eq 'terminate';

    my @responses;
    for ( my $i = 0; $i < $num_shards; $i++ ) {
        my $response = $self->_retrieve_response_from_shard($i);
        push @responses, $response->{retval};
    }
    return \@responses;
}

# Send a remote procedure call to one shard.
sub _single_rpc {
    my ( $self, $args, $shard_num ) = @_;
    my $request = $self->_serialize_request($args);
    $self->_send_request_to_shard( $shard_num, $request );
    my $response = $self->_retrieve_response_from_shard($shard_num);
    return $response->{retval};
}

# Serialize a method name and hash-style parameters using the conventions
# understood by SearchServer.
sub _serialize_request {
    my ( $self, $args ) = @_;
    my $serialized = nfreeze($args);
    my $packed_len = pack( 'N', length($serialized) );
    my $request    = "$packed_len$serialized";
    return \$request;
}

# Send a serialized request to one shard.
sub _send_request_to_shard {
    my ( $self, $shard_num, $request ) = @_;
    my $sock      = $socks{$$self}[$shard_num];
    my $check_val = $sock->syswrite($$request);
    confess $! unless $check_val == length($$request);
}
<commit_msg>Schedule using select() loop in ClusterSearcher.<commit_after># Licensed to the Apache Software Foundation (ASF) under one or more
# contributor license agreements.  See the NOTICE file distributed with
# this work for additional information regarding copyright ownership.
# The ASF licenses this file to You under the Apache License, Version 2.0
# (the "License"); you may not use this file except in compliance with
# the License.  You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

use strict;
use warnings;

package LucyX::Remote::ClusterSearcher;
BEGIN { our @ISA = qw( Lucy::Search::Searcher ) }
use Carp;
use Storable qw( nfreeze thaw );
use Scalar::Util qw( reftype );

# Inside-out member vars.
our %shards;
our %num_shards;
our %password;
our %socks;
our %starts;
our %doc_max;
our %select;
our %sock_map;

use IO::Socket::INET;
use IO::Select;

sub new {
    my ( $either, %args ) = @_;
    my $shards   = delete $args{shards};
    my $password = delete $args{password};
    my $self     = $either->SUPER::new(%args);
    confess("'shards' must be an arrayref")
        unless reftype($shards) eq 'ARRAY';
    $shards{$$self}     = $shards;
    $num_shards{$$self} = scalar @$shards;
    $password{$$self}   = $password;

    # Establish connections.
    my $socks = $socks{$$self} = [];
    for my $shard (@$shards) {
        my $sock = IO::Socket::INET->new(
            PeerAddr => $shard,
            Proto    => 'tcp',
        );
        confess("No socket: $!") unless $sock;
        push @$socks, $sock;
    }

    # Create an IO::Select object and a mapping from socket handles to shard
    # numbers.
    $select{$$self} = IO::Select->new($socks);
    my $num = 0;
    my %sock_to_shard_num_map = map { ( "$_" => $num++ ) } @$socks;
    $sock_map{$$self} = \%sock_to_shard_num_map;

    # Handshake with servers.
    my %handshake_args = ( password => <PASSWORD>, _action => 'handshake' );
    my $responses = $self->_multi_rpc( \%handshake_args );
    for my $response (@$responses) {
        confess unless $response;
    }

    # Derive doc_max and relative start offsets.
    my $doc_max_responses = $self->_multi_rpc( { _action => 'doc_max' } );
    my $doc_max = 0;
    my @starts;
    for my $shard_doc_max (@$doc_max_responses) {
        push @starts, $doc_max;
        $doc_max += $shard_doc_max;
    }
    $starts{$$self} = Lucy::Object::I32Array->new( ints => \@starts );
    $doc_max{$$self} = $doc_max;

    return $self;
}

sub DESTROY {
    my $self = shift;
    $self->close if defined $socks{$$self};
    delete $shards{$$self};
    delete $num_shards{$$self};
    delete $password{$$self};
    delete $socks{$$self};
    delete $starts{$$self};
    delete $doc_max{$$self};
    delete $select{$$self};
    delete $sock_map{$$self};
    $self->SUPER::DESTROY;
}

# Send a remote procedure call to all shards.
sub _multi_rpc {
    my ( $self, $args ) = @_;
    my $num_shards = $num_shards{$$self};
    my $request    = $self->_serialize_request($args);
    for ( my $i = 0; $i < $num_shards; $i++ ) {
        $self->_send_request_to_shard( $i, $request );
    }

    # Bail out if we're either closing or shutting down the server remotely.
    return if $args->{_action} eq 'done';
    return if $args->{_action} eq 'terminate';

    my @responses;
    my $remaining = $num_shards;
    my $select    = $select{$$self};
    my $sock_map  = $sock_map{$$self};
    while ($remaining) {
        my @ready = $select->can_read;
        for my $sock ( @{ $ready[0] } ) {
            my $shard_num = $sock_map->{"$sock"};
            my $response  = $self->_retrieve_response_from_shard($shard_num);
            $responses[$shard_num] = $response->{retval};
            $remaining--;
        }
    }
    return \@responses;
}

# Send a remote procedure call to one shard.
sub _single_rpc {
    my ( $self, $args, $shard_num ) = @_;
    my $request = $self->_serialize_request($args);
    $self->_send_request_to_shard( $shard_num, $request );
    my $response = $self->_retrieve_response_from_shard($shard_num);
    return $response->{retval};
}

# Serialize a method name and hash-style parameters using the conventions
# understood by SearchServer.
sub _serialize_request {
    my ( $self, $args ) = @_;
    my $serialized = nfreeze($args);
    my $packed_len = pack( 'N', length($serialized) );
    my $request    = "$packed_len$serialized";
    return \$request;
}

# Send a serialized request to one shard.
sub _send_request_to_shard {
    my ( $self, $shard_num, $request ) = @_;
    my $sock      = $socks{$$self}[$shard_num];
    my $check_val = $sock->syswrite($$request);
    confess $! unless $check_val == length($$request);
}

{'fraction_non_alphanumeric': 0.14308595835557927, 'fraction_numerical': 0.0020288307528029898, 'mean_word_length': 3.251475261007717, 'pattern_counts': {'":': 0, '<': 8, '<?xml version=': 0, '>': 57, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '1873199', 'n_tokens_mistral': 3176, 'n_tokens_neox': 3126, 'n_words': 912}