1 /** @file remote-database.cc
2 * @brief Remote backend database class
4 /* Copyright (C) 2006,2007,2008,2009,2010,2011,2012,2013,2014,2015,2017 Olly Betts
5 * Copyright (C) 2007,2009,2010 Lemur Consulting Ltd
7 * This program is free software; you can redistribute it and/or
8 * modify it under the terms of the GNU General Public License as
9 * published by the Free Software Foundation; either version 2 of the
10 * License, or (at your option) any later version.
12 * This program is distributed in the hope that it will be useful,
13 * but WITHOUT ANY WARRANTY; without even the implied warranty of
14 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
15 * GNU General Public License for more details.
17 * You should have received a copy of the GNU General Public License
18 * along with this program; if not, write to the Free Software
19 * Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA
24 #include "remote-database.h"
26 #include "safeerrno.h"
29 #include "backends/inmemory/inmemory_positionlist.h"
30 #include "net_postlist.h"
31 #include "net_termlist.h"
33 #include "remote-document.h"
36 #include "net/length.h"
37 #include "net/serialise.h"
38 #include "net/serialise-error.h"
39 #include "serialise-double.h"
41 #include "stringutils.h" // For STRINGIZE().
42 #include "weight/weightinternal.h"
48 #include "xapian/constants.h"
49 #include "xapian/error.h"
50 #include "xapian/matchspy.h"
53 using Xapian::Internal::intrusive_ptr
;
55 XAPIAN_NORETURN(static void throw_bad_message(const string
& context
));
57 throw_bad_message(const string
& context
)
59 throw Xapian::NetworkError("Bad message received", context
);
62 XAPIAN_NORETURN(static void throw_connection_closed_unexpectedly());
64 throw_connection_closed_unexpectedly()
66 throw Xapian::NetworkError("Connection closed unexpectedly");
69 RemoteDatabase::RemoteDatabase(int fd
, double timeout_
,
70 const string
& context_
, bool writable
,
72 : link(fd
, fd
, context_
),
76 mru_slot(Xapian::BAD_VALUENO
),
80 // It's simplest to just ignore SIGPIPE. We'll still know if the
81 // connection dies because we'll get EPIPE back from write().
82 if (signal(SIGPIPE
, SIG_IGN
) == SIG_ERR
) {
83 throw Xapian::NetworkError("Couldn't set SIGPIPE to SIG_IGN", errno
);
88 // Transactions only make sense when writing, so flag them as
89 // "unimplemented" so that our destructor doesn't call dtor_called()
90 // since that might try to call commit() which will cause a message to
91 // be sent to the remote server and probably an InvalidOperationError
92 // exception message to be returned.
93 transaction_state
= TRANSACTION_UNIMPLEMENTED
;
96 update_stats(MSG_MAX
);
99 if (flags
& Xapian::DB_RETRY_LOCK
) {
100 const string
& body
= encode_length(flags
& Xapian::DB_RETRY_LOCK
);
101 update_stats(MSG_WRITEACCESS
, body
);
103 update_stats(MSG_WRITEACCESS
);
109 RemoteDatabase::positionlist_count(Xapian::docid did
,
110 const std::string
& term
) const
112 if (cached_stats_valid
&& !has_positional_info
)
115 send_message(MSG_POSITIONLISTCOUNT
, encode_length(did
) + term
);
118 get_message(message
, REPLY_POSITIONLISTCOUNT
);
119 const char * p
= message
.data();
120 const char * p_end
= p
+ message
.size();
121 Xapian::termcount count
;
122 decode_length(&p
, p_end
, count
);
124 throw Xapian::NetworkError("Bad REPLY_POSITIONLISTCOUNT message received", context
);
130 RemoteDatabase::keep_alive()
132 send_message(MSG_KEEPALIVE
, string());
134 get_message(message
, REPLY_DONE
);
138 RemoteDatabase::open_metadata_keylist(const std::string
&prefix
) const
140 // Ensure that total_length and doccount are up-to-date.
141 if (!cached_stats_valid
) update_stats();
143 send_message(MSG_METADATAKEYLIST
, prefix
);
146 unique_ptr
<NetworkTermList
> tlist(
147 new NetworkTermList(0, doccount
,
148 intrusive_ptr
<const RemoteDatabase
>(this),
150 vector
<NetworkTermListItem
> & items
= tlist
->items
;
152 string term
= prefix
;
154 while ((type
= get_message(message
)) == REPLY_METADATAKEYLIST
) {
155 NetworkTermListItem item
;
156 term
.resize(size_t(static_cast<unsigned char>(message
[0])));
157 term
.append(message
, 1, string::npos
);
159 items
.push_back(item
);
161 if (type
!= REPLY_DONE
)
162 throw_bad_message(context
);
164 tlist
->current_position
= tlist
->items
.begin();
165 return tlist
.release();
169 RemoteDatabase::open_term_list(Xapian::docid did
) const
173 // Ensure that total_length and doccount are up-to-date.
174 if (!cached_stats_valid
) update_stats();
176 send_message(MSG_TERMLIST
, encode_length(did
));
179 get_message(message
, REPLY_DOCLENGTH
);
180 const char * p
= message
.c_str();
181 const char * p_end
= p
+ message
.size();
182 Xapian::termcount doclen
;
183 decode_length(&p
, p_end
, doclen
);
185 throw Xapian::NetworkError("Bad REPLY_DOCLENGTH message received", context
);
188 unique_ptr
<NetworkTermList
> tlist(
189 new NetworkTermList(doclen
, doccount
,
190 intrusive_ptr
<const RemoteDatabase
>(this),
192 vector
<NetworkTermListItem
> & items
= tlist
->items
;
196 while ((type
= get_message(message
)) == REPLY_TERMLIST
) {
197 NetworkTermListItem item
;
199 p_end
= p
+ message
.size();
200 decode_length(&p
, p_end
, item
.wdf
);
201 decode_length(&p
, p_end
, item
.termfreq
);
202 term
.resize(size_t(static_cast<unsigned char>(*p
++)));
203 term
.append(p
, p_end
);
205 items
.push_back(item
);
207 if (type
!= REPLY_DONE
)
208 throw_bad_message(context
);
210 tlist
->current_position
= tlist
->items
.begin();
211 return tlist
.release();
215 RemoteDatabase::open_allterms(const string
& prefix
) const {
216 // Ensure that total_length and doccount are up-to-date.
217 if (!cached_stats_valid
) update_stats();
219 send_message(MSG_ALLTERMS
, prefix
);
221 unique_ptr
<NetworkTermList
> tlist(
222 new NetworkTermList(0, doccount
,
223 intrusive_ptr
<const RemoteDatabase
>(this),
225 vector
<NetworkTermListItem
> & items
= tlist
->items
;
227 string term
= prefix
;
230 while ((type
= get_message(message
)) == REPLY_ALLTERMS
) {
231 NetworkTermListItem item
;
232 const char * p
= message
.data();
233 const char * p_end
= p
+ message
.size();
234 decode_length(&p
, p_end
, item
.termfreq
);
235 term
.resize(size_t(static_cast<unsigned char>(*p
++)));
236 term
.append(p
, p_end
);
238 items
.push_back(item
);
240 if (type
!= REPLY_DONE
)
241 throw_bad_message(context
);
243 tlist
->current_position
= tlist
->items
.begin();
244 return tlist
.release();
248 RemoteDatabase::open_post_list(const string
&term
) const
250 return new NetworkPostList(intrusive_ptr
<const RemoteDatabase
>(this), term
);
254 RemoteDatabase::read_post_list(const string
&term
, NetworkPostList
& pl
) const
256 send_message(MSG_POSTLIST
, term
);
260 get_message(message
, REPLY_POSTLISTSTART
);
262 const char * p
= message
.data();
263 const char * p_end
= p
+ message
.size();
264 Xapian::doccount termfreq
;
265 decode_length(&p
, p_end
, termfreq
);
267 while ((type
= get_message(message
)) == REPLY_POSTLISTITEM
) {
268 pl
.append_posting(message
);
270 if (type
!= REPLY_DONE
)
271 throw_bad_message(context
);
277 RemoteDatabase::open_position_list(Xapian::docid did
, const string
&term
) const
279 send_message(MSG_POSITIONLIST
, encode_length(did
) + term
);
281 vector
<Xapian::termpos
> positions
;
285 Xapian::termpos lastpos
= static_cast<Xapian::termpos
>(-1);
286 while ((type
= get_message(message
)) == REPLY_POSITIONLIST
) {
287 const char * p
= message
.data();
288 const char * p_end
= p
+ message
.size();
290 decode_length(&p
, p_end
, inc
);
292 positions
.push_back(lastpos
);
294 if (type
!= REPLY_DONE
)
295 throw_bad_message(context
);
297 return new InMemoryPositionList(positions
);
301 RemoteDatabase::has_positions() const
303 if (!cached_stats_valid
) update_stats();
304 return has_positional_info
;
308 RemoteDatabase::reopen()
310 mru_slot
= Xapian::BAD_VALUENO
;
311 return update_stats(MSG_REOPEN
);
315 RemoteDatabase::close()
320 // Currently lazy is used when fetching documents from the MSet, and in three
321 // cases in multimatch.cc. One of the latter is when using a MatchDecider,
322 // which we don't support with the remote backend currently. The others are
323 // for the sort key and collapse key which in the remote cases are fetched
324 // during the remote match and passed across with the MSet. So we can safely
325 // ignore "lazy" here for now without any performance penalty during the match
327 Xapian::Document::Internal
*
328 RemoteDatabase::open_document(Xapian::docid did
, bool /*lazy*/) const
332 send_message(MSG_DOCUMENT
, encode_length(did
));
334 map
<Xapian::valueno
, string
> values
;
335 get_message(doc_data
, REPLY_DOCDATA
);
339 while ((type
= get_message(message
)) == REPLY_VALUE
) {
340 const char * p
= message
.data();
341 const char * p_end
= p
+ message
.size();
342 Xapian::valueno slot
;
343 decode_length(&p
, p_end
, slot
);
344 values
.insert(make_pair(slot
, string(p
, p_end
)));
346 if (type
!= REPLY_DONE
)
347 throw_bad_message(context
);
349 return new RemoteDocument(this, did
, doc_data
, values
);
353 RemoteDatabase::update_stats(message_type msg_code
, const string
& body
) const
355 // MSG_MAX signals that we're handling the opening greeting, which isn't in
356 // response to an explicit message.
357 if (msg_code
!= MSG_MAX
)
358 send_message(msg_code
, body
);
361 reply_type type
= get_message(message
);
362 if (type
!= REPLY_UPDATE
|| message
.size() < 3) {
363 if (type
== REPLY_DONE
) {
364 // The database was already open at the latest revision.
367 throw Xapian::NetworkError("Handshake failed - is this a Xapian server?", context
);
370 const char *p
= message
.c_str();
371 const char *p_end
= p
+ message
.size();
373 // The protocol major versions must match. The protocol minor version of
374 // the server must be >= that of the client.
375 int protocol_major
= static_cast<unsigned char>(*p
++);
376 int protocol_minor
= static_cast<unsigned char>(*p
++);
377 if (protocol_major
!= XAPIAN_REMOTE_PROTOCOL_MAJOR_VERSION
||
378 protocol_minor
< XAPIAN_REMOTE_PROTOCOL_MINOR_VERSION
) {
379 string
errmsg("Server supports protocol version");
380 if (protocol_minor
) {
382 errmsg
+= str(protocol_major
);
385 errmsg
+= str(protocol_major
);
387 errmsg
+= str(protocol_minor
);
389 " - client is using "
390 STRINGIZE(XAPIAN_REMOTE_PROTOCOL_MAJOR_VERSION
)
392 STRINGIZE(XAPIAN_REMOTE_PROTOCOL_MINOR_VERSION
);
393 throw Xapian::NetworkError(errmsg
, context
);
396 decode_length(&p
, p_end
, doccount
);
397 decode_length(&p
, p_end
, lastdocid
);
398 lastdocid
+= doccount
;
399 decode_length(&p
, p_end
, doclen_lbound
);
400 decode_length(&p
, p_end
, doclen_ubound
);
401 doclen_ubound
+= doclen_lbound
;
403 throw Xapian::NetworkError("Bad stats update message received", context
);
405 has_positional_info
= (*p
++ == '1');
406 decode_length(&p
, p_end
, total_length
);
407 uuid
.assign(p
, p_end
);
408 cached_stats_valid
= true;
413 RemoteDatabase::get_doccount() const
415 if (!cached_stats_valid
) update_stats();
420 RemoteDatabase::get_lastdocid() const
422 if (!cached_stats_valid
) update_stats();
427 RemoteDatabase::get_total_length() const
429 if (!cached_stats_valid
) update_stats();
434 RemoteDatabase::term_exists(const string
& tname
) const
436 Assert(!tname
.empty());
437 send_message(MSG_TERMEXISTS
, tname
);
439 reply_type type
= get_message(message
);
440 if (type
!= REPLY_TERMEXISTS
&& type
!= REPLY_TERMDOESNTEXIST
)
441 throw_bad_message(context
);
442 return (type
== REPLY_TERMEXISTS
);
446 RemoteDatabase::get_freqs(const string
& term
,
447 Xapian::doccount
* termfreq_ptr
,
448 Xapian::termcount
* collfreq_ptr
) const
450 Assert(!term
.empty());
456 send_message(MSG_FREQS
, term
);
457 get_message(message
, REPLY_FREQS
);
459 send_message(MSG_TERMFREQ
, term
);
460 get_message(message
, REPLY_TERMFREQ
);
463 p_end
= p
+ message
.size();
464 decode_length(&p
, p_end
, *termfreq_ptr
);
465 } else if (collfreq_ptr
) {
466 send_message(MSG_COLLFREQ
, term
);
467 get_message(message
, REPLY_COLLFREQ
);
469 p_end
= p
+ message
.size();
472 decode_length(&p
, p_end
, *collfreq_ptr
);
477 RemoteDatabase::read_value_stats(Xapian::valueno slot
) const
479 if (mru_slot
!= slot
) {
480 send_message(MSG_VALUESTATS
, encode_length(slot
));
482 get_message(message
, REPLY_VALUESTATS
);
483 const char * p
= message
.data();
484 const char * p_end
= p
+ message
.size();
486 decode_length(&p
, p_end
, mru_valstats
.freq
);
488 decode_length_and_check(&p
, p_end
, len
);
489 mru_valstats
.lower_bound
.assign(p
, len
);
491 decode_length_and_check(&p
, p_end
, len
);
492 mru_valstats
.upper_bound
.assign(p
, len
);
495 throw Xapian::NetworkError("Bad REPLY_VALUESTATS message received", context
);
501 RemoteDatabase::get_value_freq(Xapian::valueno slot
) const
503 read_value_stats(slot
);
504 return mru_valstats
.freq
;
508 RemoteDatabase::get_value_lower_bound(Xapian::valueno slot
) const
510 read_value_stats(slot
);
511 return mru_valstats
.lower_bound
;
515 RemoteDatabase::get_value_upper_bound(Xapian::valueno slot
) const
517 read_value_stats(slot
);
518 return mru_valstats
.upper_bound
;
522 RemoteDatabase::get_doclength_lower_bound() const
524 return doclen_lbound
;
528 RemoteDatabase::get_doclength_upper_bound() const
530 return doclen_ubound
;
534 RemoteDatabase::get_wdf_upper_bound(const string
&) const
536 // The default implementation returns get_collection_freq(), but we
537 // don't want the overhead of a remote message and reply per query
538 // term, and we can get called in the middle of a remote exchange
539 // too. FIXME: handle this bound in the stats local/remote code...
540 return doclen_ubound
;
544 RemoteDatabase::get_doclength(Xapian::docid did
) const
547 send_message(MSG_DOCLENGTH
, encode_length(did
));
549 get_message(message
, REPLY_DOCLENGTH
);
550 const char * p
= message
.c_str();
551 const char * p_end
= p
+ message
.size();
552 Xapian::termcount doclen
;
553 decode_length(&p
, p_end
, doclen
);
555 throw Xapian::NetworkError("Bad REPLY_DOCLENGTH message received", context
);
561 RemoteDatabase::get_unique_terms(Xapian::docid did
) const
564 send_message(MSG_UNIQUETERMS
, encode_length(did
));
566 get_message(message
, REPLY_UNIQUETERMS
);
567 const char * p
= message
.c_str();
568 const char * p_end
= p
+ message
.size();
569 Xapian::termcount doclen
;
570 decode_length(&p
, p_end
, doclen
);
572 throw Xapian::NetworkError("Bad REPLY_UNIQUETERMS message received", context
);
578 RemoteDatabase::get_message(string
&result
, reply_type required_type
) const
580 double end_time
= RealTime::end_time(timeout
);
581 int type_int
= link
.get_message(result
, end_time
);
583 throw_connection_closed_unexpectedly();
584 reply_type type
= static_cast<reply_type
>(type_int
);
585 if (type
== REPLY_EXCEPTION
) {
586 unserialise_error(result
, "REMOTE:", context
);
588 if (required_type
!= REPLY_MAX
&& type
!= required_type
) {
589 string
errmsg("Expecting reply type ");
590 errmsg
+= str(int(required_type
));
592 errmsg
+= str(int(type
));
593 throw Xapian::NetworkError(errmsg
);
600 RemoteDatabase::send_message(message_type type
, const string
&message
) const
602 double end_time
= RealTime::end_time(timeout
);
603 link
.send_message(static_cast<unsigned char>(type
), message
, end_time
);
607 RemoteDatabase::do_close()
609 // In the constructor, we set transaction_state to
610 // TRANSACTION_UNIMPLEMENTED if we aren't writable so that we can check
612 bool writable
= (transaction_state
!= TRANSACTION_UNIMPLEMENTED
);
614 // Only call dtor_called() if we're writable.
615 if (writable
) dtor_called();
617 // If we're writable, wait for a confirmation of the close, so we know that
618 // changes have been written and flushed, and the database write lock
619 // released. For the non-writable case, there's no need to wait, so don't
620 // slow down searching by waiting here.
621 link
.do_close(writable
);
625 RemoteDatabase::set_query(const Xapian::Query
& query
,
626 Xapian::termcount qlen
,
627 Xapian::doccount collapse_max
,
628 Xapian::valueno collapse_key
,
629 Xapian::Enquire::docid_order order
,
630 Xapian::valueno sort_key
,
631 Xapian::Enquire::Internal::sort_setting sort_by
,
632 bool sort_value_forward
,
634 int percent_cutoff
, double weight_cutoff
,
635 const Xapian::Weight
*wtscheme
,
636 const Xapian::RSet
&omrset
,
637 const vector
<Xapian::Internal::opt_intrusive_ptr
<Xapian::MatchSpy
>> & matchspies
)
639 string tmp
= query
.serialise();
640 string message
= encode_length(tmp
.size());
643 // Serialise assorted Enquire settings.
644 message
+= encode_length(qlen
);
645 message
+= encode_length(collapse_max
);
646 if (collapse_max
) message
+= encode_length(collapse_key
);
647 message
+= char('0' + order
);
648 message
+= encode_length(sort_key
);
649 message
+= char('0' + sort_by
);
650 message
+= char('0' + sort_value_forward
);
651 message
+= serialise_double(time_limit
);
652 message
+= char(percent_cutoff
);
653 message
+= serialise_double(weight_cutoff
);
655 tmp
= wtscheme
->name();
656 message
+= encode_length(tmp
.size());
659 tmp
= wtscheme
->serialise();
660 message
+= encode_length(tmp
.size());
663 tmp
= serialise_rset(omrset
);
664 message
+= encode_length(tmp
.size());
667 for (auto i
: matchspies
) {
670 throw Xapian::UnimplementedError("MatchSpy subclass not suitable for use with remote searches - name() method returned empty string");
672 message
+= encode_length(tmp
.size());
675 tmp
= i
->serialise();
676 message
+= encode_length(tmp
.size());
680 send_message(MSG_QUERY
, message
);
684 RemoteDatabase::get_remote_stats(bool nowait
, Xapian::Weight::Internal
&out
)
686 if (nowait
&& !link
.ready_to_read()) return false;
689 get_message(message
, REPLY_STATS
);
690 unserialise_stats(message
, out
);
696 RemoteDatabase::send_global_stats(Xapian::doccount first
,
697 Xapian::doccount maxitems
,
698 Xapian::doccount check_at_least
,
699 const Xapian::Weight::Internal
&stats
)
701 string message
= encode_length(first
);
702 message
+= encode_length(maxitems
);
703 message
+= encode_length(check_at_least
);
704 message
+= serialise_stats(stats
);
705 send_message(MSG_GETMSET
, message
);
709 RemoteDatabase::get_mset(Xapian::MSet
&mset
,
710 const vector
<Xapian::Internal::opt_intrusive_ptr
<Xapian::MatchSpy
>> & matchspies
)
713 get_message(message
, REPLY_RESULTS
);
714 const char * p
= message
.data();
715 const char * p_end
= p
+ message
.size();
717 for (auto i
: matchspies
) {
719 throw Xapian::NetworkError("Expected serialised matchspy");
721 decode_length_and_check(&p
, p_end
, len
);
722 string
spyresults(p
, len
);
724 i
->merge_results(spyresults
);
726 mset
= unserialise_mset(p
, p_end
);
730 RemoteDatabase::commit()
732 send_message(MSG_COMMIT
, string());
734 // We need to wait for a response to ensure documents have been committed.
736 get_message(message
, REPLY_DONE
);
740 RemoteDatabase::cancel()
742 cached_stats_valid
= false;
743 mru_slot
= Xapian::BAD_VALUENO
;
745 send_message(MSG_CANCEL
, string());
749 RemoteDatabase::add_document(const Xapian::Document
& doc
)
751 cached_stats_valid
= false;
752 mru_slot
= Xapian::BAD_VALUENO
;
754 send_message(MSG_ADDDOCUMENT
, serialise_document(doc
));
757 get_message(message
, REPLY_ADDDOCUMENT
);
759 const char * p
= message
.data();
760 const char * p_end
= p
+ message
.size();
762 decode_length(&p
, p_end
, did
);
767 RemoteDatabase::delete_document(Xapian::docid did
)
769 cached_stats_valid
= false;
770 mru_slot
= Xapian::BAD_VALUENO
;
772 send_message(MSG_DELETEDOCUMENT
, encode_length(did
));
774 get_message(dummy
, REPLY_DONE
);
778 RemoteDatabase::delete_document(const std::string
& unique_term
)
780 cached_stats_valid
= false;
781 mru_slot
= Xapian::BAD_VALUENO
;
783 send_message(MSG_DELETEDOCUMENTTERM
, unique_term
);
787 RemoteDatabase::replace_document(Xapian::docid did
,
788 const Xapian::Document
& doc
)
790 cached_stats_valid
= false;
791 mru_slot
= Xapian::BAD_VALUENO
;
793 string message
= encode_length(did
);
794 message
+= serialise_document(doc
);
796 send_message(MSG_REPLACEDOCUMENT
, message
);
800 RemoteDatabase::replace_document(const std::string
& unique_term
,
801 const Xapian::Document
& doc
)
803 cached_stats_valid
= false;
804 mru_slot
= Xapian::BAD_VALUENO
;
806 string message
= encode_length(unique_term
.size());
807 message
+= unique_term
;
808 message
+= serialise_document(doc
);
810 send_message(MSG_REPLACEDOCUMENTTERM
, message
);
812 get_message(message
, REPLY_ADDDOCUMENT
);
814 const char * p
= message
.data();
815 const char * p_end
= p
+ message
.size();
817 decode_length(&p
, p_end
, did
);
822 RemoteDatabase::get_uuid() const
828 RemoteDatabase::get_metadata(const string
& key
) const
830 send_message(MSG_GETMETADATA
, key
);
832 get_message(metadata
, REPLY_METADATA
);
837 RemoteDatabase::set_metadata(const string
& key
, const string
& value
)
839 string data
= encode_length(key
.size());
842 send_message(MSG_SETMETADATA
, data
);
846 RemoteDatabase::add_spelling(const string
& word
,
847 Xapian::termcount freqinc
) const
849 string data
= encode_length(freqinc
);
851 send_message(MSG_ADDSPELLING
, data
);
855 RemoteDatabase::remove_spelling(const string
& word
,
856 Xapian::termcount freqdec
) const
858 string data
= encode_length(freqdec
);
860 send_message(MSG_REMOVESPELLING
, data
);
864 RemoteDatabase::locked() const
866 throw Xapian::UnimplementedError("Database::locked() not implemented for remote backend");