Ticket #711: remote_protocol_msg_readaccess_select.diff

xapian-core/backends/dbfactory_remote.cc

@@ -33 +33 @@
 
 Database
 Remote::open(const string &host, unsigned int port, useconds_t timeout_,
-             useconds_t connect_timeout)
+             useconds_t connect_timeout, int flags, const string &dir)
 {
-    LOGCALL_STATIC(API, Database, "Remote::open", host | port | timeout_ | connect_timeout);
+    LOGCALL_STATIC(API, Database, "Remote::open", host | port | timeout_ | connect_timeout | flags | dir);
     RETURN(Database(new RemoteTcpClient(host, port, timeout_ * 1e-3,
-                                        connect_timeout * 1e-3, false, 0)));
+                                        connect_timeout * 1e-3, false, flags, dir)));
 }
 
 WritableDatabase
 Remote::open_writable(const string &host, unsigned int port,
                       useconds_t timeout_, useconds_t connect_timeout,
-                      int flags)
+                      int flags, const string &dir)
 {
-    LOGCALL_STATIC(API, WritableDatabase, "Remote::open_writable", host | port | timeout_ | connect_timeout | flags);
+    LOGCALL_STATIC(API, WritableDatabase, "Remote::open_writable", host | port | timeout_ | connect_timeout | flags | dir);
     RETURN(WritableDatabase(new RemoteTcpClient(host, port, timeout_ * 1e-3,
                                                 connect_timeout * 1e-3, true,
-                                                flags)));
+                                                flags, dir)));
 }
 
 Database
 Remote::open(const string &program, const string &args,
-             useconds_t timeout_)
+             useconds_t timeout_, int flags, const string &dir)
 {
-    LOGCALL_STATIC(API, Database, "Remote::open", program | args | timeout_);
-    RETURN(Database(new ProgClient(program, args, timeout_ * 1e-3, false, 0)));
+    LOGCALL_STATIC(API, Database, "Remote::open", program | args | timeout_ | flags | dir);
+    RETURN(Database(new ProgClient(program, args, timeout_ * 1e-3, false, flags, dir)));
 }
 
 WritableDatabase
 Remote::open_writable(const string &program, const string &args,
-                      useconds_t timeout_, int flags)
+                      useconds_t timeout_, int flags, const string &dir)
 {
-    LOGCALL_STATIC(API, WritableDatabase, "Remote::open_writable", program | args | timeout_ | flags);
+    LOGCALL_STATIC(API, WritableDatabase, "Remote::open_writable", program | args | timeout_ | flags | dir);
     RETURN(WritableDatabase(new ProgClient(program, args,
-                                           timeout_ * 1e-3, true, flags)));
+                                           timeout_ * 1e-3, true, flags, dir)));
 }
 
 }

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

@@ -68 +68 @@
 }
 
 RemoteDatabase::RemoteDatabase(int fd, double timeout_,
-                               const string & context_, bool writable,
-                               int flags)
-        : link(fd, fd, context_),
+                               const string & context_, bool writable_,
+                               int flags_, const string & dir)
+        : db_dir(dir),
+          writable(writable_),
+          flags(flags_),
+          link(fd, fd, context_),
           context(context_),
           cached_stats_valid(),
           mru_valstats(),
@@ -95 +98 @@
     }
 
     update_stats(MSG_MAX);
-
-    if (writable) {
-        if (flags & Xapian::DB_RETRY_LOCK) {
-            const string & body = encode_length(flags & Xapian::DB_RETRY_LOCK);
-            update_stats(MSG_WRITEACCESS, body);
-        } else {
-            update_stats(MSG_WRITEACCESS);
-        }
-    }
 }
 
 void
@@ -339 +333 @@
 
     string message;
     reply_type type = get_message(message);
-    if (type != REPLY_UPDATE || message.size() < 3) {
+    if (type != REPLY_UPDATE || message.size() < 2) {
         if (type == REPLY_DONE) {
             // The database was already open at the latest revision.
             return false;
@@ -373 +367 @@
         throw Xapian::NetworkError(errmsg, context);
     }
 
+    if (p == p_end) {
+        message = encode_length(flags);
+        message += encode_length(db_dir.size());
+        message += db_dir;
+
+        if (writable) {
+            send_message(MSG_WRITEACCESS, message);
+        } else {
+            send_message(MSG_READACCESS, message);
+        }
+
+        get_message(message, REPLY_UPDATE);
+        if (message.size() < 3) {
+            throw Xapian::NetworkError("Database was not selected", context);
+        }
+
+        p = message.c_str();
+        p_end = p + message.size();
+
+        // The protocol versions where already checked.
+        p += 2;
+    } else if (writable) {
+        if (flags) {
+            send_message(MSG_WRITEACCESS, encode_length(flags));
+        } else {
+            send_message(MSG_WRITEACCESS, string());
+        }
+
+        get_message(message, REPLY_UPDATE);
+        if (message.size() < 3) {
+            throw Xapian::NetworkError("Database was not selected", context);
+        }
+
+        p = message.c_str();
+        p_end = p + message.size();
+
+        // The protocol versions where already checked.
+        p += 2;
+    }
+
     decode_length(&p, p_end, doccount);
     decode_length(&p, p_end, lastdocid);
     lastdocid += doccount;
@@ -597 +631 @@
     // In the constructor, we set transaction_state to
     // TRANSACTION_UNIMPLEMENTED if we aren't writable so that we can check
     // it here.
-    bool writable = (transaction_state != TRANSACTION_UNIMPLEMENTED);
+    bool writable_ = (transaction_state != TRANSACTION_UNIMPLEMENTED);
 
     // Only call dtor_called() if we're writable.
-    if (writable) dtor_called();
+    if (writable_) dtor_called();
 
     // If we're writable, wait for a confirmation of the close, so we know that
     // changes have been written and flushed, and the database write lock
     // released.  For the non-writable case, there's no need to wait, so don't
     // slow down searching by waiting here.
-    link.do_close(writable);
+    link.do_close(writable_);
 }
 
 void

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

@@ -49 +49 @@
     /// Don't allow copying.
     RemoteDatabase(const RemoteDatabase &);
 
+    // Directory to store databases in.
+    const std::string db_dir;
+
+    // The database is writable.
+    const bool writable;
+
+    // Bitwise-or of Xapian::DB_* flags.
+    const int flags;
+
     /// The object which does the I/O.
     mutable RemoteConnection link;
 
@@ -103 +112 @@
      *                  operations will never timeout.
      *  @param context_ The context to return with any error messages.
      *  @param writable Is this a WritableDatabase?
-     *  @param flags    Xapian::DB_RETRY_LOCK or 0.
+     *  @param flags    Bitwise-or of Xapian::DB_* constants.
+     *  @param dir      Database directory index to open.
      */
     RemoteDatabase(int fd, double timeout_, const string & context_,
-                   bool writable, int flags);
+                   bool writable, int flags, const string & dir);
 
     /// Receive a message from the server.
     reply_type get_message(string & message, reply_type required_type = REPLY_MAX) const;

xapian-core/bin/xapian-tcpsrv.cc

@@ -72 +72 @@
 };
 
 static void show_usage() {
-    cout << "Usage: " PROG_NAME " [OPTIONS] DATABASE_DIRECTORY...\n\n"
+    cout << "Usage: " PROG_NAME " [OPTIONS] [DATABASE_DIRECTORY] ...\n\n"
 "Options:\n"
 "  --port PORTNUM          listen on port PORTNUM for connections (no default)\n"
 "  --interface ADDRESS     listen on the interface associated with name or\n"
@@ -143 +143 @@
         }
     }
 
-    if (syntax_error || argv[optind] == NULL) {
+    if (syntax_error) {
         show_usage();
         exit(1);
     }

xapian-core/common/remoteprotocol.h

@@ -89 +89 @@
     MSG_METADATAKEYLIST,        // Iterator for metadata keys
     MSG_FREQS,                  // Get termfreq and collfreq
     MSG_UNIQUETERMS,            // Get number of unique terms in doc
+    MSG_READACCESS,             // Select currenty active read access database
     MSG_MAX
 };
 

xapian-core/include/xapian/dbfactory.h

@@ -171 +171 @@
  *                              Xapian::NetworkTimeoutError is thrown.  A
  *                              timeout of 0 means don't timeout.  (Default is
  *                              10000ms, which is 10 seconds).
- */
+ * @param flags         bitwise-or of Xapian::DB_* constants.
+ * @param dir           database directory index to open.
+*/
 XAPIAN_VISIBILITY_DEFAULT
-Database open(const std::string &host, unsigned int port, useconds_t timeout = 10000, useconds_t connect_timeout = 10000);
+Database open(const std::string &host, unsigned int port, useconds_t timeout = 10000, useconds_t connect_timeout = 10000, int flags = 0, const std::string &dir = std::string());
 
 /** Construct a WritableDatabase object for update access to a remote database
  *  accessed via a TCP connection.
@@ -192 +194 @@
  *                              Xapian::NetworkTimeoutError is thrown.  A
  *                              timeout of 0 means don't timeout.  (Default is
  *                              10000ms, which is 10 seconds).
+ * @param flags         bitwise-or of Xapian::DB_* constants.
+ * @param dir           database directory index to open.
  */
 XAPIAN_VISIBILITY_DEFAULT
-WritableDatabase open_writable(const std::string &host, unsigned int port, useconds_t timeout = 0, useconds_t connect_timeout = 10000, int flags = 0);
+WritableDatabase open_writable(const std::string &host, unsigned int port, useconds_t timeout = 0, useconds_t connect_timeout = 10000, int flags = 0, const std::string &dir = std::string());
 
 /** Construct a Database object for read-only access to a remote database
  *  accessed via a program.
@@ -209 +213 @@
  *                      then Xapian::NetworkTimeoutError is thrown.  A timeout
  *                      of 0 means don't timeout.  (Default is 10000ms, which
  *                      is 10 seconds).
- * @param flags         Xapian::DB_RETRY_LOCK or 0.
+ * @param flags         bitwise-or of Xapian::DB_* constants.
+ * @param dir           database directory index to open.
  */
 XAPIAN_VISIBILITY_DEFAULT
-Database open(const std::string &program, const std::string &args, useconds_t timeout = 10000);
+Database open(const std::string &program, const std::string &args, useconds_t timeout = 10000, int flags = 0, const std::string &dir = std::string());
 
 /** Construct a WritableDatabase object for update access to a remote database
  *  accessed via a program.
@@ -226 +231 @@
  *                      for any individual operation on the remote database
  *                      then Xapian::NetworkTimeoutError is thrown.  (Default
  *                      is 0, which means don't timeout).
- * @param flags         Xapian::DB_RETRY_LOCK or 0.
+ * @param flags         bitwise-or of Xapian::DB_* constants.
+ * @param dir           database directory index to open.
  */
 XAPIAN_VISIBILITY_DEFAULT
-WritableDatabase open_writable(const std::string &program, const std::string &args, useconds_t timeout = 0, int flags = 0);
+WritableDatabase open_writable(const std::string &program, const std::string &args, useconds_t timeout = 0, int flags = 0, const std::string &dir = std::string());
 
 }
 #endif

xapian-core/net/progclient.cc

@@ -64 +64 @@
 #endif
 
 ProgClient::ProgClient(const string &progname, const string &args,
-                       double timeout_, bool writable, int flags)
+                       double timeout_, bool writable, int flags, const string & dir)
         : RemoteDatabase(run_program(progname, args
 #ifndef __WIN32__
                                                    , pid
 #endif
         ),
                          timeout_, get_progcontext(progname, args), writable,
-                         flags)
+                         flags, dir)
 {
-    LOGCALL_CTOR(DB, "ProgClient", progname | args | timeout_ | writable | flags);
+    LOGCALL_CTOR(DB, "ProgClient", progname | args | timeout_ | writable | flags | dir);
 }
 
 string

xapian-core/net/progclient.h

@@ -88 +88 @@
                const std::string &arg,
                double msecs_timeout,
                bool writable,
-               int flags);
+               int flags,
+               const std::string & dir);
 
     /** Destructor. */
     ~ProgClient();

xapian-core/net/remoteserver.cc

@@ -54 +54 @@
     throw Xapian::InvalidOperationError("Server is read-only");
 }
 
+XAPIAN_NORETURN(static void throw_no_db());
+static void
+throw_no_db()
+{
+    throw Xapian::InvalidOperationError("Server has no open database");
+}
+
 /// Class to throw when we receive the connection closing message.
 struct ConnectionClosed { };
 
-RemoteServer::RemoteServer(const std::vector<std::string> &dbpaths,
+RemoteServer::RemoteServer(const std::vector<std::string> &dbpaths_,
                            int fdin_, int fdout_,
                            double active_timeout_, double idle_timeout_,
                            bool writable_)
@@ -67 +74 @@
 {
     // Catch errors opening the database and propagate them to the client.
     try {
-        Assert(!dbpaths.empty());
-        // We always open the database read-only to start with.  If we're
-        // writable, the client can ask to be upgraded to write access once
-        // connected if it wants it.
-        db = new Xapian::Database(dbpaths[0]);
-        // Build a better description than Database::get_description() gives
-        // in the variable context.  FIXME: improve Database::get_description()
-        // and then just use that instead.
-        context = dbpaths[0];
-
-        if (!writable) {
-            vector<std::string>::const_iterator i(dbpaths.begin());
-            for (++i; i != dbpaths.end(); ++i) {
-                db->add_database(Xapian::Database(*i));
-                context += ' ';
-                context += *i;
-            }
-        } else {
-            AssertEq(dbpaths.size(), 1); // Expecting exactly one database.
+        if (!dbpaths_.empty()) {
+            // We always open the database read-only to start with.  If we're
+            // writable, the client can ask to be upgraded to write access once
+            // connected if it wants it.
+            select_db(dbpaths_, false, Xapian::DB_OPEN);
         }
     } catch (const Xapian::Error &err) {
         // Propagate the exception to the client.
@@ -192 +185 @@
                 &RemoteServer::msg_openmetadatakeylist,
                 &RemoteServer::msg_freqs,
                 &RemoteServer::msg_uniqueterms,
+                &RemoteServer::msg_readaccess,
             };
 
             string message;
@@ -239 +233 @@
 void
 RemoteServer::msg_allterms(const string &message)
 {
+    if (!db)
+        throw_no_db();
+
     string prev = message;
     string reply;
 
@@ -262 +259 @@
 void
 RemoteServer::msg_termlist(const string &message)
 {
+    if (!db)
+        throw_no_db();
+
     const char *p = message.data();
     const char *p_end = p + message.size();
     Xapian::docid did;
@@ -289 +289 @@
 void
 RemoteServer::msg_positionlist(const string &message)
 {
+    if (!db)
+        throw_no_db();
+
     const char *p = message.data();
     const char *p_end = p + message.size();
     Xapian::docid did;
@@ -310 +313 @@
 void
 RemoteServer::msg_postlist(const string &message)
 {
+    if (!db)
+        throw_no_db();
+
     const string & term = message;
 
     Xapian::doccount termfreq = db->get_termfreq(term);
@@ -333 +339 @@
 }
 
 void
-RemoteServer::msg_writeaccess(const string & msg)
+RemoteServer::msg_readaccess(const string &message)
 {
-    if (!writable) 
+    int flags = Xapian::DB_OPEN;
+    const char *p = message.c_str();
+    const char *p_end = p + message.size();
+    if (p != p_end) {
+        unsigned flag_bits;
+        decode_length(&p, p_end, flag_bits);
+        flags |= flag_bits &~ Xapian::DB_ACTION_MASK_;
+    }
+
+    if (p != p_end) {
+        std::vector<string> dbpaths_;
+        while (p != p_end) {
+            size_t len;
+            decode_length_and_check(&p, p_end, len);
+            string dbpath(p, len);
+            dbpaths_.push_back(dbpath);
+            p += len;
+        }
+        select_db(dbpaths_, false, flags);
+    } else {
+        select_db(dbpaths, false, flags);
+    }
+
+    msg_update(message);
+}
+
+
+void
+RemoteServer::msg_writeaccess(const string & message)
+{
+    if (!writable)
         throw_read_only();
 
     int flags = Xapian::DB_OPEN;
-    const char *p = msg.c_str();
-    const char *p_end = p + msg.size();
+    const char *p = message.c_str();
+    const char *p_end = p + message.size();
     if (p != p_end) {
         unsigned flag_bits;
         decode_length(&p, p_end, flag_bits);
         flags |= flag_bits &~ Xapian::DB_ACTION_MASK_;
+    }
+
+    if (p != p_end) {
+        std::vector<string> dbpaths_;
+        size_t len;
+        decode_length_and_check(&p, p_end, len);
+        string dbpath(p, len);
+        dbpaths_.push_back(dbpath);
+        p += len;
         if (p != p_end) {
-            throw Xapian::NetworkError("Junk at end of MSG_WRITEACCESS");
+            throw Xapian::NetworkError("only one database directory allowed on writable databases");
         }
+        select_db(dbpaths_, true, flags);
+    } else {
+        select_db(dbpaths, true, flags);
     }
 
-    wdb = new Xapian::WritableDatabase(context, flags);
-    delete db;
-    db = wdb;
-    msg_update(msg);
+    msg_update(message);
 }
 
 void
 RemoteServer::msg_reopen(const string & msg)
 {
+    if (!db)
+        throw_no_db();
+
     if (!db->reopen()) {
         send_message(REPLY_DONE, string());
         return;
@@ -373 +421 @@
         char(XAPIAN_REMOTE_PROTOCOL_MAJOR_VERSION),
         char(XAPIAN_REMOTE_PROTOCOL_MINOR_VERSION)
     };
+
     string message(protocol, 2);
-    Xapian::doccount num_docs = db->get_doccount();
-    message += encode_length(num_docs);
-    message += encode_length(db->get_lastdocid() - num_docs);
-    Xapian::termcount doclen_lb = db->get_doclength_lower_bound();
-    message += encode_length(doclen_lb);
-    message += encode_length(db->get_doclength_upper_bound() - doclen_lb);
-    message += (db->has_positions() ? '1' : '0');
-    // FIXME: clumsy to reverse calculate total_len like this:
-    totlen_t total_len = totlen_t(db->get_avlength() * db->get_doccount() + .5);
-    message += encode_length(total_len);
-    //message += encode_length(db->get_total_length());
-    string uuid = db->get_uuid();
-    message += uuid;
+
+    if (db) {
+        Xapian::doccount num_docs = db->get_doccount();
+        message += encode_length(num_docs);
+        message += encode_length(db->get_lastdocid() - num_docs);
+        Xapian::termcount doclen_lb = db->get_doclength_lower_bound();
+        message += encode_length(doclen_lb);
+        message += encode_length(db->get_doclength_upper_bound() - doclen_lb);
+        message += (db->has_positions() ? '1' : '0');
+        // FIXME: clumsy to reverse calculate total_len like this:
+        totlen_t total_len = totlen_t(db->get_avlength() * db->get_doccount() + .5);
+        message += encode_length(total_len);
+        //message += encode_length(db->get_total_length());
+        string uuid = db->get_uuid();
+        message += uuid;
+    }
+
     send_message(REPLY_UPDATE, message);
 }
 
@@ -485 +538 @@
         p += len;
     }
 
+    if (!db)
+        throw_no_db();
+
     Xapian::Weight::Internal local_stats;
     MultiMatch match(*db, query, qlen, &rset, collapse_max, collapse_key,
                      percent_cutoff, weight_cutoff, order,
@@ -528 +584 @@
 void
 RemoteServer::msg_document(const string &message)
 {
+    if (!db)
+        throw_no_db();
+
     const char *p = message.data();
     const char *p_end = p + message.size();
     Xapian::docid did;
@@ -549 +608 @@
 void
 RemoteServer::msg_keepalive(const string &)
 {
+    if (!db)
+        throw_no_db();
+
     // Ensure *our* database stays alive, as it may contain remote databases!
     db->keep_alive();
     send_message(REPLY_DONE, string());
@@ -557 +619 @@
 void
 RemoteServer::msg_termexists(const string &term)
 {
+    if (!db)
+        throw_no_db();
+
     send_message((db->term_exists(term) ? REPLY_TERMEXISTS : REPLY_TERMDOESNTEXIST), string());
 }
 
 void
 RemoteServer::msg_collfreq(const string &term)
 {
+    if (!db)
+        throw_no_db();
+
     send_message(REPLY_COLLFREQ, encode_length(db->get_collection_freq(term)));
 }
 
 void
 RemoteServer::msg_termfreq(const string &term)
 {
+    if (!db)
+        throw_no_db();
+
     send_message(REPLY_TERMFREQ, encode_length(db->get_termfreq(term)));
 }
 
 void
 RemoteServer::msg_freqs(const string &term)
 {
+    if (!db)
+        throw_no_db();
+
     string msg = encode_length(db->get_termfreq(term));
     msg += encode_length(db->get_collection_freq(term));
     send_message(REPLY_FREQS, msg);
@@ -583 +657 @@
 void
 RemoteServer::msg_valuestats(const string & message)
 {
+    if (!db)
+        throw_no_db();
+
     const char *p = message.data();
     const char *p_end = p + message.size();
     while (p != p_end) {
@@ -604 +681 @@
 void
 RemoteServer::msg_doclength(const string &message)
 {
+    if (!db)
+        throw_no_db();
+
     const char *p = message.data();
     const char *p_end = p + message.size();
     Xapian::docid did;
@@ -614 +694 @@
 void
 RemoteServer::msg_uniqueterms(const string &message)
 {
+    if (!db)
+        throw_no_db();
+
     const char *p = message.data();
     const char *p_end = p + message.size();
     Xapian::docid did;
@@ -715 +798 @@
 void
 RemoteServer::msg_getmetadata(const string & message)
 {
+    if (!db)
+        throw_no_db();
+
     send_message(REPLY_METADATA, db->get_metadata(message));
 }
 
 void
 RemoteServer::msg_openmetadatakeylist(const string & message)
 {
+    if (!db)
+        throw_no_db();
+
     string prev = message;
     string reply;
 
@@ -778 +867 @@
     decode_length(&p, p_end, freqdec);
     wdb->remove_spelling(string(p, p_end - p), freqdec);
 }
+
+void
+RemoteServer::select_db(const std::vector<std::string> &dbpaths_, bool writable_, int flags) {
+    if (writable_) {
+        AssertEq(dbpaths_.size(), 1); // Expecting exactly one database.
+        Xapian::WritableDatabase * wdb_ = new Xapian::WritableDatabase(dbpaths_[0], flags);
+        context = dbpaths_[0];
+        delete db;
+        db = wdb_;
+        wdb = wdb_;
+    } else {
+        Assert(!dbpaths_.empty());  // Expecting at least one database.
+        Xapian::Database * db_;
+        if (dbpaths_.size() == 1) {
+            db_ = new Xapian::Database(dbpaths_[0], flags);
+            context = dbpaths_[0];
+        } else {
+            db_ = new Xapian::Database();
+            // Build a better description than Database::get_description() gives
+            // in the variable context.  FIXME: improve Database::get_description()
+            // and then just use that instead.
+            context = "";
+            for (auto& path : dbpaths_) {
+                db->add_database(Xapian::Database(path, flags));
+                context += ' ';
+                context += path;
+            }
+        }
+        delete db;
+        db = db_;
+        wdb = NULL;
+    }
+    dbpaths = dbpaths_;
+}

xapian-core/net/remoteserver.h

@@ -69 +69 @@
     /// The registry, which allows unserialisation of user subclasses.
     Xapian::Registry reg;
 
+    std::vector<std::string> dbpaths;
+
     /// Accept a message from the client.
     message_type get_message(double timeout, std::string & result,
                              message_type required_type = MSG_MAX);
@@ -170 +172 @@
     // get number of unique terms
     void msg_uniqueterms(const std::string & message);
 
+    // select an active database for read access
+    void msg_readaccess(const std::string & message);
+
+    void select_db(const std::vector<std::string> &dbpaths_, bool writable_, int flags);
+
   public:
     /** Construct a RemoteServer.
      *

xapian-core/net/remotetcpclient.h

@@ -75 +75 @@
      */
     RemoteTcpClient(const std::string & hostname, int port,
                     double timeout_, double timeout_connect, bool writable,
-                    int flags)
+                    int flags, const std::string & dir)
         : RemoteDatabase(open_socket(hostname, port, timeout_connect),
                          timeout_, get_tcpcontext(hostname, port),
-                         writable, flags) { }
+                         writable, flags, dir) { }
 
     /** Destructor. */
     ~RemoteTcpClient();