xapian-core/backends/remote/remote-database.cc

   1 /** @file remote-database.cc
   2  *  @brief Remote backend database class
   3  */
   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
   6  *
   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.
  11  *
  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.
  16  *
  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
  20  */
  21
  22 #include <config.h>
  23
  24 #include "remote-database.h"
  25
  26 #include "safeerrno.h"
  27 #include <signal.h>
  28
  29 #include "backends/inmemory/inmemory_positionlist.h"
  30 #include "net_postlist.h"
  31 #include "net_termlist.h"
  32 #include "noreturn.h"
  33 #include "remote-document.h"
  34 #include "omassert.h"
  35 #include "realtime.h"
  36 #include "net/length.h"
  37 #include "net/serialise.h"
  38 #include "net/serialise-error.h"
  39 #include "serialise-double.h"
  40 #include "str.h"
  41 #include "stringutils.h" // For STRINGIZE().
  42 #include "weight/weightinternal.h"
  43
  44 #include <memory>
  45 #include <string>
  46 #include <vector>
  47
  48 #include "xapian/constants.h"
  49 #include "xapian/error.h"
  50 #include "xapian/matchspy.h"
  51
  52 using namespace std;
  53 using Xapian::Internal::intrusive_ptr;
  54
  55 XAPIAN_NORETURN(static void throw_bad_message(const string & context));
  56 static void
  57 throw_bad_message(const string & context)
  58 {
  59     throw Xapian::NetworkError("Bad message received", context);
  60 }
  61
  62 XAPIAN_NORETURN(static void throw_connection_closed_unexpectedly());
  63 static void
  64 throw_connection_closed_unexpectedly()
  65 {
  66     throw Xapian::NetworkError("Connection closed unexpectedly");
  67 }
  68
  69 RemoteDatabase::RemoteDatabase(int fd, double timeout_,
  70                                const string & context_, bool writable,
  71                                int flags)
  72         : link(fd, fd, context_),
  73           context(context_),
  74           cached_stats_valid(),
  75           mru_valstats(),
  76           mru_slot(Xapian::BAD_VALUENO),
  77           timeout(timeout_)
  78 {
  79 #ifndef __WIN32__
  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);
  84     }
  85 #endif
  86
  87     if (!writable) {
  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;
  94     }
  95
  96     update_stats(MSG_MAX);
  97
  98     if (writable) {
  99         if (flags & Xapian::DB_RETRY_LOCK) {
 100             const string & body = encode_length(flags & Xapian::DB_RETRY_LOCK);
 101             update_stats(MSG_WRITEACCESS, body);
 102         } else {
 103             update_stats(MSG_WRITEACCESS);
 104         }
 105     }
 106 }
 107
 108 Xapian::termcount
 109 RemoteDatabase::positionlist_count(Xapian::docid did,
 110                                    const std::string& term) const
 111 {
 112     if (cached_stats_valid && !has_positional_info)
 113         return 0;
 114
 115     send_message(MSG_POSITIONLISTCOUNT, encode_length(did) + term);
 116
 117     string message;
 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);
 123     if (p != p_end) {
 124         throw Xapian::NetworkError("Bad REPLY_POSITIONLISTCOUNT message received", context);
 125     }
 126     return count;
 127 }
 128
 129 void
 130 RemoteDatabase::keep_alive()
 131 {
 132     send_message(MSG_KEEPALIVE, string());
 133     string message;
 134     get_message(message, REPLY_DONE);
 135 }
 136
 137 TermList *
 138 RemoteDatabase::open_metadata_keylist(const std::string &prefix) const
 139 {
 140     // Ensure that total_length and doccount are up-to-date.
 141     if (!cached_stats_valid) update_stats();
 142
 143     send_message(MSG_METADATAKEYLIST, prefix);
 144
 145     string message;
 146     unique_ptr<NetworkTermList> tlist(
 147         new NetworkTermList(0, doccount,
 148                             intrusive_ptr<const RemoteDatabase>(this),
 149                             0));
 150     vector<NetworkTermListItem> & items = tlist->items;
 151
 152     string term = prefix;
 153     char type;
 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);
 158         item.tname = term;
 159         items.push_back(item);
 160     }
 161     if (type != REPLY_DONE)
 162         throw_bad_message(context);
 163
 164     tlist->current_position = tlist->items.begin();
 165     return tlist.release();
 166 }
 167
 168 TermList *
 169 RemoteDatabase::open_term_list(Xapian::docid did) const
 170 {
 171     Assert(did);
 172
 173     // Ensure that total_length and doccount are up-to-date.
 174     if (!cached_stats_valid) update_stats();
 175
 176     send_message(MSG_TERMLIST, encode_length(did));
 177
 178     string message;
 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);
 184     if (p != p_end) {
 185         throw Xapian::NetworkError("Bad REPLY_DOCLENGTH message received", context);
 186     }
 187
 188     unique_ptr<NetworkTermList> tlist(
 189         new NetworkTermList(doclen, doccount,
 190                             intrusive_ptr<const RemoteDatabase>(this),
 191                             did));
 192     vector<NetworkTermListItem> & items = tlist->items;
 193
 194     string term;
 195     char type;
 196     while ((type = get_message(message)) == REPLY_TERMLIST) {
 197         NetworkTermListItem item;
 198         p = message.data();
 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);
 204         item.tname = term;
 205         items.push_back(item);
 206     }
 207     if (type != REPLY_DONE)
 208         throw_bad_message(context);
 209
 210     tlist->current_position = tlist->items.begin();
 211     return tlist.release();
 212 }
 213
 214 TermList *
 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();
 218
 219     send_message(MSG_ALLTERMS, prefix);
 220
 221     unique_ptr<NetworkTermList> tlist(
 222         new NetworkTermList(0, doccount,
 223                             intrusive_ptr<const RemoteDatabase>(this),
 224                             0));
 225     vector<NetworkTermListItem> & items = tlist->items;
 226
 227     string term = prefix;
 228     string message;
 229     char type;
 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);
 237         item.tname = term;
 238         items.push_back(item);
 239     }
 240     if (type != REPLY_DONE)
 241         throw_bad_message(context);
 242
 243     tlist->current_position = tlist->items.begin();
 244     return tlist.release();
 245 }
 246
 247 LeafPostList *
 248 RemoteDatabase::open_post_list(const string &term) const
 249 {
 250     return new NetworkPostList(intrusive_ptr<const RemoteDatabase>(this), term);
 251 }
 252
 253 Xapian::doccount
 254 RemoteDatabase::read_post_list(const string &term, NetworkPostList & pl) const
 255 {
 256     send_message(MSG_POSTLIST, term);
 257
 258     string message;
 259     char type;
 260     get_message(message, REPLY_POSTLISTSTART);
 261
 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);
 266
 267     while ((type = get_message(message)) == REPLY_POSTLISTITEM) {
 268         pl.append_posting(message);
 269     }
 270     if (type != REPLY_DONE)
 271         throw_bad_message(context);
 272
 273     return termfreq;
 274 }
 275
 276 PositionList *
 277 RemoteDatabase::open_position_list(Xapian::docid did, const string &term) const
 278 {
 279     send_message(MSG_POSITIONLIST, encode_length(did) + term);
 280
 281     vector<Xapian::termpos> positions;
 282
 283     string message;
 284     char type;
 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();
 289         Xapian::termpos inc;
 290         decode_length(&p, p_end, inc);
 291         lastpos += inc + 1;
 292         positions.push_back(lastpos);
 293     }
 294     if (type != REPLY_DONE)
 295         throw_bad_message(context);
 296
 297     return new InMemoryPositionList(positions);
 298 }
 299
 300 bool
 301 RemoteDatabase::has_positions() const
 302 {
 303     if (!cached_stats_valid) update_stats();
 304     return has_positional_info;
 305 }
 306
 307 bool
 308 RemoteDatabase::reopen()
 309 {
 310     mru_slot = Xapian::BAD_VALUENO;
 311     return update_stats(MSG_REOPEN);
 312 }
 313
 314 void
 315 RemoteDatabase::close()
 316 {
 317     do_close();
 318 }
 319
 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
 326 // process.
 327 Xapian::Document::Internal *
 328 RemoteDatabase::open_document(Xapian::docid did, bool /*lazy*/) const
 329 {
 330     Assert(did);
 331
 332     send_message(MSG_DOCUMENT, encode_length(did));
 333     string doc_data;
 334     map<Xapian::valueno, string> values;
 335     get_message(doc_data, REPLY_DOCDATA);
 336
 337     reply_type type;
 338     string message;
 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)));
 345     }
 346     if (type != REPLY_DONE)
 347         throw_bad_message(context);
 348
 349     return new RemoteDocument(this, did, doc_data, values);
 350 }
 351
 352 bool
 353 RemoteDatabase::update_stats(message_type msg_code, const string & body) const
 354 {
 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);
 359
 360     string message;
 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.
 365             return false;
 366         }
 367         throw Xapian::NetworkError("Handshake failed - is this a Xapian server?", context);
 368     }
 369
 370     const char *p = message.c_str();
 371     const char *p_end = p + message.size();
 372
 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) {
 381             errmsg += "s ";
 382             errmsg += str(protocol_major);
 383             errmsg += ".0 to ";
 384         }
 385         errmsg += str(protocol_major);
 386         errmsg += '.';
 387         errmsg += str(protocol_minor);
 388         errmsg +=
 389             " - client is using "
 390             STRINGIZE(XAPIAN_REMOTE_PROTOCOL_MAJOR_VERSION)
 391             "."
 392             STRINGIZE(XAPIAN_REMOTE_PROTOCOL_MINOR_VERSION);
 393         throw Xapian::NetworkError(errmsg, context);
 394     }
 395
 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;
 402     if (p == p_end) {
 403         throw Xapian::NetworkError("Bad stats update message received", context);
 404     }
 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;
 409     return true;
 410 }
 411
 412 Xapian::doccount
 413 RemoteDatabase::get_doccount() const
 414 {
 415     if (!cached_stats_valid) update_stats();
 416     return doccount;
 417 }
 418
 419 Xapian::docid
 420 RemoteDatabase::get_lastdocid() const
 421 {
 422     if (!cached_stats_valid) update_stats();
 423     return lastdocid;
 424 }
 425
 426 Xapian::totallength
 427 RemoteDatabase::get_total_length() const
 428 {
 429     if (!cached_stats_valid) update_stats();
 430     return total_length;
 431 }
 432
 433 bool
 434 RemoteDatabase::term_exists(const string & tname) const
 435 {
 436     Assert(!tname.empty());
 437     send_message(MSG_TERMEXISTS, tname);
 438     string message;
 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);
 443 }
 444
 445 void
 446 RemoteDatabase::get_freqs(const string & term,
 447                           Xapian::doccount * termfreq_ptr,
 448                           Xapian::termcount * collfreq_ptr) const
 449 {
 450     Assert(!term.empty());
 451     string message;
 452     const char * p;
 453     const char * p_end;
 454     if (termfreq_ptr) {
 455         if (collfreq_ptr) {
 456             send_message(MSG_FREQS, term);
 457             get_message(message, REPLY_FREQS);
 458         } else {
 459             send_message(MSG_TERMFREQ, term);
 460             get_message(message, REPLY_TERMFREQ);
 461         }
 462         p = message.data();
 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);
 468         p = message.data();
 469         p_end = p + message.size();
 470     }
 471     if (collfreq_ptr) {
 472         decode_length(&p, p_end, *collfreq_ptr);
 473     }
 474 }
 475
 476 void
 477 RemoteDatabase::read_value_stats(Xapian::valueno slot) const
 478 {
 479     if (mru_slot != slot) {
 480         send_message(MSG_VALUESTATS, encode_length(slot));
 481         string message;
 482         get_message(message, REPLY_VALUESTATS);
 483         const char * p = message.data();
 484         const char * p_end = p + message.size();
 485         mru_slot = slot;
 486         decode_length(&p, p_end, mru_valstats.freq);
 487         size_t len;
 488         decode_length_and_check(&p, p_end, len);
 489         mru_valstats.lower_bound.assign(p, len);
 490         p += len;
 491         decode_length_and_check(&p, p_end, len);
 492         mru_valstats.upper_bound.assign(p, len);
 493         p += len;
 494         if (p != p_end) {
 495             throw Xapian::NetworkError("Bad REPLY_VALUESTATS message received", context);
 496         }
 497     }
 498 }
 499
 500 Xapian::doccount
 501 RemoteDatabase::get_value_freq(Xapian::valueno slot) const
 502 {
 503     read_value_stats(slot);
 504     return mru_valstats.freq;
 505 }
 506
 507 std::string
 508 RemoteDatabase::get_value_lower_bound(Xapian::valueno slot) const
 509 {
 510     read_value_stats(slot);
 511     return mru_valstats.lower_bound;
 512 }
 513
 514 std::string
 515 RemoteDatabase::get_value_upper_bound(Xapian::valueno slot) const
 516 {
 517     read_value_stats(slot);
 518     return mru_valstats.upper_bound;
 519 }
 520
 521 Xapian::termcount
 522 RemoteDatabase::get_doclength_lower_bound() const
 523 {
 524     return doclen_lbound;
 525 }
 526
 527 Xapian::termcount
 528 RemoteDatabase::get_doclength_upper_bound() const
 529 {
 530     return doclen_ubound;
 531 }
 532
 533 Xapian::termcount
 534 RemoteDatabase::get_wdf_upper_bound(const string &) const
 535 {
 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;
 541 }
 542
 543 Xapian::termcount
 544 RemoteDatabase::get_doclength(Xapian::docid did) const
 545 {
 546     Assert(did != 0);
 547     send_message(MSG_DOCLENGTH, encode_length(did));
 548     string message;
 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);
 554     if (p != p_end) {
 555         throw Xapian::NetworkError("Bad REPLY_DOCLENGTH message received", context);
 556     }
 557     return doclen;
 558 }
 559
 560 Xapian::termcount
 561 RemoteDatabase::get_unique_terms(Xapian::docid did) const
 562 {
 563     Assert(did != 0);
 564     send_message(MSG_UNIQUETERMS, encode_length(did));
 565     string message;
 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);
 571     if (p != p_end) {
 572         throw Xapian::NetworkError("Bad REPLY_UNIQUETERMS message received", context);
 573     }
 574     return doclen;
 575 }
 576
 577 reply_type
 578 RemoteDatabase::get_message(string &result, reply_type required_type) const
 579 {
 580     double end_time = RealTime::end_time(timeout);
 581     int type_int = link.get_message(result, end_time);
 582     if (type_int < 0)
 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);
 587     }
 588     if (required_type != REPLY_MAX && type != required_type) {
 589         string errmsg("Expecting reply type ");
 590         errmsg += str(int(required_type));
 591         errmsg += ", got ";
 592         errmsg += str(int(type));
 593         throw Xapian::NetworkError(errmsg);
 594     }
 595
 596     return type;
 597 }
 598
 599 void
 600 RemoteDatabase::send_message(message_type type, const string &message) const
 601 {
 602     double end_time = RealTime::end_time(timeout);
 603     link.send_message(static_cast<unsigned char>(type), message, end_time);
 604 }
 605
 606 void
 607 RemoteDatabase::do_close()
 608 {
 609     // In the constructor, we set transaction_state to
 610     // TRANSACTION_UNIMPLEMENTED if we aren't writable so that we can check
 611     // it here.
 612     bool writable = (transaction_state != TRANSACTION_UNIMPLEMENTED);
 613
 614     // Only call dtor_called() if we're writable.
 615     if (writable) dtor_called();
 616
 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);
 622 }
 623
 624 void
 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,
 633                          double time_limit,
 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)
 638 {
 639     string tmp = query.serialise();
 640     string message = encode_length(tmp.size());
 641     message += tmp;
 642
 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);
 654
 655     tmp = wtscheme->name();
 656     message += encode_length(tmp.size());
 657     message += tmp;
 658
 659     tmp = wtscheme->serialise();
 660     message += encode_length(tmp.size());
 661     message += tmp;
 662
 663     tmp = serialise_rset(omrset);
 664     message += encode_length(tmp.size());
 665     message += tmp;
 666
 667     for (auto i : matchspies) {
 668         tmp = i->name();
 669         if (tmp.empty()) {
 670             throw Xapian::UnimplementedError("MatchSpy subclass not suitable for use with remote searches - name() method returned empty string");
 671         }
 672         message += encode_length(tmp.size());
 673         message += tmp;
 674
 675         tmp = i->serialise();
 676         message += encode_length(tmp.size());
 677         message += tmp;
 678     }
 679
 680     send_message(MSG_QUERY, message);
 681 }
 682
 683 bool
 684 RemoteDatabase::get_remote_stats(bool nowait, Xapian::Weight::Internal &out)
 685 {
 686     if (nowait && !link.ready_to_read()) return false;
 687
 688     string message;
 689     get_message(message, REPLY_STATS);
 690     unserialise_stats(message, out);
 691
 692     return true;
 693 }
 694
 695 void
 696 RemoteDatabase::send_global_stats(Xapian::doccount first,
 697                                   Xapian::doccount maxitems,
 698                                   Xapian::doccount check_at_least,
 699                                   const Xapian::Weight::Internal &stats)
 700 {
 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);
 706 }
 707
 708 void
 709 RemoteDatabase::get_mset(Xapian::MSet &mset,
 710                          const vector<Xapian::Internal::opt_intrusive_ptr<Xapian::MatchSpy>> & matchspies)
 711 {
 712     string message;
 713     get_message(message, REPLY_RESULTS);
 714     const char * p = message.data();
 715     const char * p_end = p + message.size();
 716
 717     for (auto i : matchspies) {
 718         if (p == p_end)
 719             throw Xapian::NetworkError("Expected serialised matchspy");
 720         size_t len;
 721         decode_length_and_check(&p, p_end, len);
 722         string spyresults(p, len);
 723         p += len;
 724         i->merge_results(spyresults);
 725     }
 726     mset = unserialise_mset(p, p_end);
 727 }
 728
 729 void
 730 RemoteDatabase::commit()
 731 {
 732     send_message(MSG_COMMIT, string());
 733
 734     // We need to wait for a response to ensure documents have been committed.
 735     string message;
 736     get_message(message, REPLY_DONE);
 737 }
 738
 739 void
 740 RemoteDatabase::cancel()
 741 {
 742     cached_stats_valid = false;
 743     mru_slot = Xapian::BAD_VALUENO;
 744
 745     send_message(MSG_CANCEL, string());
 746 }
 747
 748 Xapian::docid
 749 RemoteDatabase::add_document(const Xapian::Document & doc)
 750 {
 751     cached_stats_valid = false;
 752     mru_slot = Xapian::BAD_VALUENO;
 753
 754     send_message(MSG_ADDDOCUMENT, serialise_document(doc));
 755
 756     string message;
 757     get_message(message, REPLY_ADDDOCUMENT);
 758
 759     const char * p = message.data();
 760     const char * p_end = p + message.size();
 761     Xapian::docid did;
 762     decode_length(&p, p_end, did);
 763     return did;
 764 }
 765
 766 void
 767 RemoteDatabase::delete_document(Xapian::docid did)
 768 {
 769     cached_stats_valid = false;
 770     mru_slot = Xapian::BAD_VALUENO;
 771
 772     send_message(MSG_DELETEDOCUMENT, encode_length(did));
 773     string dummy;
 774     get_message(dummy, REPLY_DONE);
 775 }
 776
 777 void
 778 RemoteDatabase::delete_document(const std::string & unique_term)
 779 {
 780     cached_stats_valid = false;
 781     mru_slot = Xapian::BAD_VALUENO;
 782
 783     send_message(MSG_DELETEDOCUMENTTERM, unique_term);
 784 }
 785
 786 void
 787 RemoteDatabase::replace_document(Xapian::docid did,
 788                                  const Xapian::Document & doc)
 789 {
 790     cached_stats_valid = false;
 791     mru_slot = Xapian::BAD_VALUENO;
 792
 793     string message = encode_length(did);
 794     message += serialise_document(doc);
 795
 796     send_message(MSG_REPLACEDOCUMENT, message);
 797 }
 798
 799 Xapian::docid
 800 RemoteDatabase::replace_document(const std::string & unique_term,
 801                                  const Xapian::Document & doc)
 802 {
 803     cached_stats_valid = false;
 804     mru_slot = Xapian::BAD_VALUENO;
 805
 806     string message = encode_length(unique_term.size());
 807     message += unique_term;
 808     message += serialise_document(doc);
 809
 810     send_message(MSG_REPLACEDOCUMENTTERM, message);
 811
 812     get_message(message, REPLY_ADDDOCUMENT);
 813
 814     const char * p = message.data();
 815     const char * p_end = p + message.size();
 816     Xapian::docid did;
 817     decode_length(&p, p_end, did);
 818     return did;
 819 }
 820
 821 string
 822 RemoteDatabase::get_uuid() const
 823 {
 824     return uuid;
 825 }
 826
 827 string
 828 RemoteDatabase::get_metadata(const string & key) const
 829 {
 830     send_message(MSG_GETMETADATA, key);
 831     string metadata;
 832     get_message(metadata, REPLY_METADATA);
 833     return metadata;
 834 }
 835
 836 void
 837 RemoteDatabase::set_metadata(const string & key, const string & value)
 838 {
 839     string data = encode_length(key.size());
 840     data += key;
 841     data += value;
 842     send_message(MSG_SETMETADATA, data);
 843 }
 844
 845 void
 846 RemoteDatabase::add_spelling(const string & word,
 847                              Xapian::termcount freqinc) const
 848 {
 849     string data = encode_length(freqinc);
 850     data += word;
 851     send_message(MSG_ADDSPELLING, data);
 852 }
 853
 854 void
 855 RemoteDatabase::remove_spelling(const string & word,
 856                                 Xapian::termcount freqdec) const
 857 {
 858     string data = encode_length(freqdec);
 859     data += word;
 860     send_message(MSG_REMOVESPELLING, data);
 861 }
 862
 863 bool
 864 RemoteDatabase::locked() const
 865 {
 866     throw Xapian::UnimplementedError("Database::locked() not implemented for remote backend");
 867 }