| # 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 ) } |
| our $VERSION = '0.004001'; |
| $VERSION = eval $VERSION; |
| use Carp; |
| use Storable qw( nfreeze thaw ); |
| use Scalar::Util qw( reftype ); |
| |
| # Inside-out member vars. |
| our %shards; |
| our %num_shards; |
| our %starts; |
| our %doc_max; |
| |
| use IO::Socket::INET; |
| |
| sub new { |
| my ( $either, %args ) = @_; |
| my $addrs = delete $args{shards}; |
| my $self = $either->SUPER::new(%args); |
| confess("'shards' must be an arrayref") |
| unless reftype($addrs) eq 'ARRAY'; |
| $num_shards{$$self} = scalar @$addrs; |
| |
| # Establish connections. |
| my @shards; |
| for my $addr (@$addrs) { |
| my $sock = IO::Socket::INET->new( |
| PeerAddr => $addr, |
| Proto => 'tcp', |
| Blocking => 0, |
| ); |
| confess("No socket: $!") unless $sock; |
| push @shards, |
| { |
| addr => $addr, |
| sock => $sock, |
| }; |
| } |
| $shards{$$self} = \@shards; |
| |
| # Handshake with servers. |
| my %handshake_args = ( _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 $shards{$$self}; |
| delete $shards{$$self}; |
| delete $num_shards{$$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 ) = @_; |
| return $self->_rpc( $args, $shards{$$self} ); |
| } |
| |
| # Send a remote procedure call to one shard. |
| sub _single_rpc { |
| my ( $self, $args, $shard_num ) = @_; |
| my $shard = $shards{$$self}[$shard_num]; |
| my $responses = $self->_rpc( $args, [$shard] ); |
| return $responses->[0]; |
| } |
| |
| sub _rpc { |
| my ( $self, $args, $shards ) = @_; |
| |
| my $request = $self->_serialize_request($args); |
| my $timeout = 5; |
| my $shutdown = $args->{_action} eq 'done' |
| || $args->{_action} eq 'terminate'; |
| |
| my ( $rin, $win, $ein ) = ( '', '', '' ); |
| |
| # Initialize shards to send the request |
| for my $shard (@$shards) { |
| my $fileno = $shard->{sock}->fileno; |
| vec( $win, $fileno, 1 ) = 1; |
| $shard->{response} = undef; |
| $shard->{error} = undef; |
| $shard->{buf} = $request; |
| $shard->{sent} = 0; |
| $shard->{callback} = \&_cb_send; |
| $shard->{shutdown} = $shutdown; |
| } |
| |
| my $remaining = @$shards; |
| |
| # Event loop |
| while ( $remaining > 0 ) { |
| my ( $rout, $wout, $eout ); |
| |
| my $n = select( $rout = $rin, $wout = $win, $eout = $ein, $timeout ); |
| |
| confess("select: $!") if $n == -1; |
| confess("I/O timeout") if $n == 0; |
| |
| for my $shard (@$shards) { |
| next if !$shard->{callback}; |
| my $fileno = $shard->{sock}->fileno; |
| next if !vec( $rout, $fileno, 1 ) && !vec( $wout, $fileno, 1 ); |
| # Dispatch event |
| $shard->{callback}->( $shard, \$rin, \$win, \$ein ); |
| --$remaining if !$shard->{callback}; |
| } |
| } |
| |
| # Collect responses and cleanup |
| my @responses; |
| my @errors; |
| for my $shard (@$shards) { |
| if ( defined $shard->{error} ) { |
| push( @errors, $shard->{error} . ' @ ' . $shard->{addr} ); |
| } |
| else { |
| push( @responses, $shard->{response}{retval} ); |
| } |
| $shard->{response} = undef; |
| $shard->{error} = undef; |
| $shard->{buf} = undef; |
| } |
| confess( 'RPC error: ' . join( ', ', @errors ) ) if @errors; |
| return \@responses; |
| } |
| |
| # 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 (partial) request to a shard |
| sub _cb_send { |
| my ( $shard, $rin, $win, $ein ) = @_; |
| |
| my $msg = substr( ${ $shard->{buf} }, $shard->{sent} ); |
| my $sent = $shard->{sock}->send($msg); |
| |
| if ( !defined($sent) ) { |
| $shard->{error} = $!; |
| $shard->{callback} = undef; |
| vec( $$win, $shard->{sock}->fileno, 1 ) = 0; |
| return; |
| } |
| |
| $shard->{sent} += $sent; |
| |
| if ( $sent >= length($msg) ) { |
| # Complete |
| my $fileno = $shard->{sock}->fileno; |
| vec( $$win, $fileno, 1 ) = 0; |
| if ( $shard->{shutdown} ) { |
| # Bail out if we're either closing or shutting down the server |
| # remotely. |
| $shard->{callback} = undef; |
| } |
| else { |
| # Setup shard to read response length |
| $shard->{buf} = ''; |
| $shard->{callback} = \&_cb_recv_length; |
| vec( $$rin, $fileno, 1 ) = 1; |
| } |
| } |
| } |
| |
| # Receive a (partial) response length from a shard |
| sub _cb_recv_length { |
| my ( $shard, $rin, $win, $ein ) = @_; |
| |
| my $data; |
| my $r = $shard->{sock}->recv( $data, 4 - length( $shard->{buf} ) ); |
| |
| if ( !defined($r) || length($data) == 0 ) { |
| $shard->{error} = !defined($r) ? $! : 'Remote shutdown'; |
| $shard->{callback} = undef; |
| vec( $$rin, $shard->{sock}->fileno, 1 ) = 0; |
| return; |
| } |
| |
| $shard->{buf} .= $data; |
| |
| if ( length( $shard->{buf} ) >= 4 ) { |
| # Complete, setup shard to receive response |
| $shard->{response_size} = unpack( 'N', $shard->{buf} ); |
| $shard->{buf} = ''; |
| $shard->{callback} = \&_cb_recv_response; |
| } |
| } |
| |
| # Receive a (partial) response from a shard |
| sub _cb_recv_response { |
| my ( $shard, $rin, $win, $ein ) = @_; |
| |
| my $data; |
| my $remaining = $shard->{response_size} - length( $shard->{buf} ); |
| my $r = $shard->{sock}->recv( $data, $remaining ); |
| |
| if ( !defined($r) || length($data) == 0 ) { |
| $shard->{error} = !defined($r) ? $! : 'Remote shutdown'; |
| $shard->{callback} = undef; |
| vec( $$rin, $shard->{sock}->fileno, 1 ) = 0; |
| return; |
| } |
| |
| $shard->{buf} .= $data; |
| |
| if ( length( $shard->{buf} ) >= $shard->{response_size} ) { |
| # Finished |
| $shard->{response} = thaw( $shard->{buf} ); |
| $shard->{callback} = undef; |
| } |
| } |
| |
| sub top_docs { |
| my ( $self, %args ) = @_; |
| my $starts = $starts{$$self}; |
| my $num_shards = $num_shards{$$self}; |
| my $query = $args{query}; |
| my $num_wanted = $args{num_wanted}; |
| my $sort_spec = $args{sort_spec}; |
| |
| # Weight query if necessary. |
| my $compiler |
| = $query->isa("Lucy::Search::Compiler") |
| ? $query |
| : $query->make_compiler( |
| searcher => $self, |
| boost => $query->get_boost, |
| ); |
| |
| # Create HitQueue. |
| my $hit_q; |
| if ($sort_spec) { |
| $hit_q = Lucy::Search::HitQueue->new( |
| schema => $self->get_schema, |
| sort_spec => $sort_spec, |
| wanted => $num_wanted, |
| ); |
| } |
| else { |
| $hit_q = Lucy::Search::HitQueue->new( wanted => $num_wanted, ); |
| } |
| |
| # Gather remote responses and aggregate. |
| $args{_action} = 'top_docs'; |
| my $responses = $self->_multi_rpc( \%args ); |
| my $total_hits = 0; |
| for ( my $i = 0; $i < $num_shards; $i++ ) { |
| my $base = $starts->get($i); |
| my $sub_top_docs = $responses->[$i]; |
| my @sub_match_docs = sort { $a->get_doc_id <=> $b->get_doc_id } |
| @{ $sub_top_docs->get_match_docs }; |
| for my $match_doc (@sub_match_docs) { |
| $match_doc->set_doc_id( $match_doc->get_doc_id + $base ); |
| $hit_q->insert($match_doc); |
| } |
| $total_hits += $sub_top_docs->get_total_hits; |
| } |
| |
| # Return a TopDocs object with the best of the best. |
| my $best_match_docs = $hit_q->pop_all; |
| return Lucy::Search::TopDocs->new( |
| total_hits => $total_hits, |
| match_docs => $best_match_docs, |
| ); |
| } |
| |
| sub terminate { |
| my $self = shift; |
| $self->_multi_rpc( { _action => 'terminate' } ); |
| return; |
| } |
| |
| sub fetch_doc { |
| my ( $self, $doc_id ) = @_; |
| my $starts = $starts{$$self}; |
| my $tick = Lucy::Index::PolyReader::sub_tick( $starts, $doc_id ); |
| my $start = $starts->get($tick); |
| my %args = ( doc_id => $doc_id - $start, _action => 'fetch_doc' ); |
| my $hit_doc = $self->_single_rpc( \%args, $tick ); |
| $hit_doc->set_doc_id($doc_id); |
| return $hit_doc; |
| } |
| |
| sub fetch_doc_vec { |
| my ( $self, $doc_id ) = @_; |
| my $starts = $starts{$$self}; |
| my $tick = Lucy::Index::PolyReader::sub_tick( $starts, $doc_id ); |
| my $start = $starts->get($tick); |
| my %args = ( doc_id => $doc_id - $start, _action => 'fetch_doc_vec' ); |
| return $self->_single_rpc( \%args, $tick ); |
| } |
| |
| sub doc_max { |
| my $self = shift; |
| return $doc_max{$$self}; |
| } |
| |
| sub doc_freq { |
| my $self = shift; |
| my %args = ( @_, _action => 'doc_freq' ); |
| my $responses = $self->_multi_rpc( \%args ); |
| my $doc_freq = 0; |
| $doc_freq += $_ for @$responses; |
| return $doc_freq; |
| } |
| |
| sub close { |
| my $self = shift; |
| return unless $shards{$$self}; |
| $self->_multi_rpc( { _action => 'done' } ); |
| for my $shard ( @{ $shards{$$self} } ) { |
| close $shard->{sock} or confess("Error when closing socket: $!"); |
| } |
| delete $shards{$$self}; |
| } |
| |
| 1; |
| |
| __END__ |
| |
| =head1 NAME |
| |
| LucyX::Remote::ClusterSearcher - Search multiple remote indexes. |
| |
| =head1 SYNOPSIS |
| |
| my $searcher = eval { |
| LucyX::Remote::ClusterSearcher->new( |
| schema => MySchema->new, |
| shards => [ 'search1:7890', 'search2:7890', 'search3:7890' ], |
| ); |
| }; |
| ... |
| my $hits = eval { $searcher->hits( query => $query ) }; |
| |
| =head1 DESCRIPTION |
| |
| ClusterSearcher is a subclass of L<Lucy::Search::Searcher> which can be used |
| to search a composite index made up of multiple shards, where each shard is |
| represented by a host:port pair running L<LucyX::Remote::SearchServer>. |
| |
| =head1 METHODS |
| |
| =head2 new |
| |
| Constructor. Takes hash-style params. |
| |
| =over |
| |
| =item * |
| |
| B<schema> - A Schema, which must match the Schema used by each remote node. |
| |
| =item * |
| |
| B<shards> - An array of host:port pairs running LucyX::Remote::SearchServer |
| instances, which identifying the shards that make up the composite index. |
| |
| =back |
| |
| =cut |
