Ticket #266: set_flush_threshold.patch

tests/api_transdb.cc

@@ -126 +126 @@
 
     return true;
 }
+
+/// Test set_flush_threshold inside a simple transaction.
+DEFINE_TESTCASE(minmemtransaction1, transactions) {
+    Xapian::WritableDatabase db(get_writable_database("apitest_simpledata"));
+
+    Xapian::doccount docs = db.get_doccount();
+    db.begin_transaction();
+    Xapian::Document doc;
+    doc.set_data("testing");
+    doc.add_term("befuddlement");
+    db.add_document(doc);
+    TEST_EXCEPTION(Xapian::InvalidOperationError, db.begin_transaction());
+    db.set_flush_threshold(0);
+    TEST_EQUAL(db.get_doccount(), docs + 1);
+    TEST_EQUAL(db.get_termfreq("befuddlement"), 1);
+    db.commit_transaction();
+    TEST_EQUAL(db.get_doccount(), docs + 1);
+    TEST_EQUAL(db.get_termfreq("befuddlement"), 1);
+
+    return true;
+}
+
+/// Test cancelling a simple transaction after calling set_flush_threshold(0).
+DEFINE_TESTCASE(minmemcanceltransaction1, transactions) {
+    Xapian::WritableDatabase db(get_writable_database("apitest_simpledata"));
+
+    Xapian::doccount docs = db.get_doccount();
+    db.begin_transaction();
+    Xapian::Document doc;
+    doc.set_data("testing");
+    doc.add_term("befuddlement");
+    db.add_document(doc);
+    TEST_EXCEPTION(Xapian::InvalidOperationError, db.begin_transaction());
+    TEST_EQUAL(db.get_doccount(), docs + 1);
+    TEST_EQUAL(db.get_termfreq("befuddlement"), 1);
+    db.set_flush_threshold(0);
+    db.cancel_transaction();
+    TEST_EQUAL(db.get_doccount(), docs);
+    TEST_EQUAL(db.get_termfreq("befuddlement"), 0);
+
+    return true;
+}

include/xapian/database.h

@@ -5 +5 @@
  * Copyright 2002 Ananova Ltd
  * Copyright 2002,2003,2004,2005,2006,2007,2008,2009,2011 Olly Betts
  * Copyright 2006,2008 Lemur Consulting Ltd
+ * Copyright 2011 Richard Boulton
  *
  * This program is free software; you can redistribute it and/or
  * modify it under the terms of the GNU General Public License as
@@ -565 +566 @@
          */
         void flush() { commit(); }
 
+        /** Set the threshold at which changes are flushed automatically.
+         *
+         *  This may be used to exert some control on how often buffered
+         *  changes are flushed from memory during indexing.  A flush will
+         *  happen after the specified number of changes have been made since
+         *  the previous flush or commit.
+         *
+         *  The initial value for the flush threshold will be read from the
+         *  "XAPIAN_FLUSH_THRESHOLD" environment variable.
+         *
+         *  This may be called whether a transaction is currently in progress
+         *  or not, and takes effect both during transactions and outside
+         *  transactions.
+         *
+         *  If the new threshold is less than or equal to the number of changes
+         *  that have occurred since the previous flush or commit, an immediate
+         *  flush will be performed.
+         *
+         *  A threshold of 0 disables automatic flushing, but causes an
+         *  immediate flush.  Be aware that various operations will still cause
+         *  an implicit flush even if the threshold is set to 0, so this is not
+         *  an effective way to control IO.
+         *
+         *  The old value of the flush threshold will be returned, so a call to
+         *  set_flush_threshold(set_flush_threshold(0)) causes an immediate
+         *  flush, but leaves the threshold unchanged.
+         *
+         *  Not all backends support a flush threshold - such backends will
+         *  always return 0 from this method.
+         *
+         *  @exception Xapian::DatabaseError will be thrown if a problem occurs
+         *             while modifying the database.
+         *
+         *  @exception Xapian::DatabaseCorruptError will be thrown if the
+         *             database is in a corrupt state.
+         */
+        doccount set_flush_threshold(doccount threshold);
+
+        /** Set the threshold at which changes are committed automatically.
+         *
+         *  This may be used to control how often changes are committed during
+         *  indexing, when not within a transaction.  Such a commit will happen
+         *  after the specified number of changes have been made since the
+         *  previous commit.
+         *
+         *  The initial value for the commit threshold will be read from the
+         *  "XAPIAN_FLUSH_THRESHOLD" environment variable.
+         *
+         *  If the new threshold is less than or equal to the number of changes
+         *  that have occurred since the previous commit, an immediate
+         *  commit will be performed (unless a transaction is in progress).
+         *
+         *  A commit automatically performs a flush first.
+         *
+         *  A threshold of 0 disables automatic committing, but causes an
+         *  immediate commit (assuming no transaction is in progress).
+         *
+         *  Not all backends support a commit threshold - such backends will
+         *  always return 0 from this method.
+         *
+         *  @exception Xapian::DatabaseError will be thrown if a problem occurs
+         *             while modifying the database.
+         *
+         *  @exception Xapian::DatabaseCorruptError will be thrown if the
+         *             database is in a corrupt state.
+         */
+        doccount set_commit_threshold(doccount threshold);
+
         /** Begin a transaction.
          *
          *  In Xapian a transaction is a group of modifications to the database

common/database.h

@@ -412 +412 @@
         /** Cancel pending modifications to the database. */
         virtual void cancel();
 
+        /** Set the threshold at which changes are flushed.
+         *
+         *  See WritableDatabase::set_flush_threshold() for more information.
+         */
+        virtual doccount set_flush_threshold(doccount);
+
+        /** Set the threshold at which changes are committed..
+         *
+         *  See WritableDatabase::set_commit_threshold() for more information.
+         */
+        virtual doccount set_commit_threshold(doccount);
+
         /** Begin a transaction.
          *
          *  See WritableDatabase::begin_transaction() for more information.

api/omdatabase.cc

@@ -784 +784 @@
     internal[0]->commit();
 }
 
+doccount
+WritableDatabase::set_flush_threshold(doccount threshold)
+{
+    LOGCALL_VOID(API, "WritableDatabase::set_flush_threshold", threshold);
+    if (internal.size() != 1) only_one_subdatabase_allowed();
+    RETURN(internal[0]->set_flush_threshold(threshold));
+}
+
+doccount
+WritableDatabase::set_commit_threshold(doccount threshold)
+{
+    LOGCALL_VOID(API, "WritableDatabase::set_commit_threshold", threshold);
+    if (internal.size() != 1) only_one_subdatabase_allowed();
+    RETURN(internal[0]->set_commit_threshold(threshold));
+}
+
 void
 WritableDatabase::begin_transaction(bool flushed)
 {

api/error.cc

@@ -38 +38 @@
 
 Xapian::Error::Error(const std::string &msg_, const std::string &context_,
                      const char * type_, const char * error_string_)
-    : msg(msg_), context(context_), error_string(), type(type_),
-      my_errno(0), already_handled(false)
+    : msg(msg_), context(context_), type(type_), my_errno(0),
+      error_string(), already_handled(false)
 {
     if (error_string_) error_string.assign(error_string_);
 }

backends/database.cc

@@ -122 +122 @@
     Assert(false);
 }
 
+doccount
+Database::Internal::set_flush_threshold(doccount)
+{
+    // Writable databases may override this method.
+    return 0;
+}
+
+doccount
+Database::Internal::set_commit_threshold(doccount)
+{
+    // Writable databases may override this method.
+    return 0;
+}
+
 void
 Database::Internal::begin_transaction(bool flushed)
 {

backends/chert/chert_database.cc

@@ -1023 +1023 @@
           freq_deltas(),
           doclens(),
           mod_plists(),
-          change_count(0),
+          changes_since_flush(0),
+          changes_since_commit(0),
           flush_threshold(0),
+          commit_threshold(0),
           modify_shortcut_document(NULL),
           modify_shortcut_docid(0)
 {
@@ -1033 +1035 @@
     const char *p = getenv("XAPIAN_FLUSH_THRESHOLD");
     if (p)
         flush_threshold = atoi(p);
-    if (flush_threshold == 0)
+    else
         flush_threshold = 10000;
+    commit_threshold = flush_threshold;
 }
 
 ChertWritableDatabase::~ChertWritableDatabase()
@@ -1048 +1051 @@
 {
     if (transaction_active())
         throw Xapian::InvalidOperationError("Can't commit during a transaction");
-    if (change_count) flush_postlist_changes();
+    if (changes_since_flush)
+        flush_postlist_changes();
     apply();
 }
 
@@ -1061 +1065 @@
     freq_deltas.clear();
     doclens.clear();
     mod_plists.clear();
-    change_count = 0;
+    changes_since_flush = 0;
 }
 
 void
@@ -1080 +1084 @@
 {
     value_manager.set_value_stats(value_stats);
     ChertDatabase::apply();
+    changes_since_commit = 0;
 }
 
 void
+ChertWritableDatabase::inc_change_counts()
+{
+    // FIXME: this should be done by checking memory usage, not the number of
+    // changes.
+    // We could also look at:
+    // * mod_plists.size()
+    // * doclens.size()
+    // * freq_deltas.size()
+    //
+    // cout << "+++ mod_plists.size() " << mod_plists.size() <<
+    //     ", doclens.size() " << doclens.size() <<
+    //     ", freq_deltas.size() " << freq_deltas.size() << endl;
+
+    ++changes_since_commit;
+    ++changes_since_flush;
+
+    if (commit_threshold && changes_since_commit >= commit_threshold) {
+        flush_postlist_changes();
+        if (!transaction_active()) apply();
+    } else if (flush_threshold && changes_since_flush >= flush_threshold) {
+        flush_postlist_changes();
+    }
+}
+
+void
 ChertWritableDatabase::add_freq_delta(const string & tname,
                                       Xapian::termcount_diff tf_delta,
                                       Xapian::termcount_diff cf_delta)
@@ -1207 +1237 @@
         throw;
     }
 
-    // FIXME: this should be done by checking memory usage, not the number of
-    // changes.
-    // We could also look at:
-    // * mod_plists.size()
-    // * doclens.size()
-    // * freq_deltas.size()
-    //
-    // cout << "+++ mod_plists.size() " << mod_plists.size() <<
-    //     ", doclens.size() " << doclens.size() <<
-    //     ", freq_deltas.size() " << freq_deltas.size() << endl;
-    if (++change_count >= flush_threshold) {
-        flush_postlist_changes();
-        if (!transaction_active()) apply();
-    }
+    inc_change_counts();
 
     RETURN(did);
 }
@@ -1283 +1300 @@
         throw;
     }
 
-    if (++change_count >= flush_threshold) {
-        flush_postlist_changes();
-        if (!transaction_active()) apply();
-    }
+    inc_change_counts();
 }
 
 void
@@ -1451 +1465 @@
         throw;
     }
 
-    if (++change_count >= flush_threshold) {
-        flush_postlist_changes();
-        if (!transaction_active()) apply();
-    }
+    inc_change_counts();
 }
 
 Xapian::Document::Internal *
@@ -1579 +1590 @@
     // If there are changes, we don't have code to iterate the modified value
     // list so we need to flush (but don't commit - there may be a transaction
     // in progress).
-    if (change_count) value_manager.merge_changes();
+    if (changes_since_flush) value_manager.merge_changes();
     RETURN(ChertDatabase::open_value_list(slot));
 }
 
@@ -1590 +1601 @@
     // If there are changes, terms may have been added or removed, and so we
     // need to flush (but don't commit - there may be a transaction in
     // progress).
-    if (change_count) flush_postlist_changes();
+    if (changes_since_flush) flush_postlist_changes();
     RETURN(ChertDatabase::open_allterms(prefix));
 }
 
@@ -1603 +1614 @@
     doclens.clear();
     mod_plists.clear();
     value_stats.clear();
-    change_count = 0;
+    changes_since_flush = 0;
+    changes_since_commit = 0;
 }
 
+doccount 
+ChertWritableDatabase::set_flush_threshold(doccount new_flush_threshold)
+{
+    LOGCALL(DB, doccount, "ChertWritableDatabase::set_flush_threshold",
+            new_flush_threshold);
+    doccount old_flush_threshold = flush_threshold;
+    flush_threshold = new_flush_threshold;
+
+    if (changes_since_flush >= flush_threshold) {
+        flush_postlist_changes();
+    }
+
+    RETURN(old_flush_threshold);
+}
+
+doccount 
+ChertWritableDatabase::set_commit_threshold(doccount new_commit_threshold)
+{
+    LOGCALL(DB, doccount, "ChertWritableDatabase::set_commit_threshold",
+            new_commit_threshold);
+    doccount old_commit_threshold = commit_threshold;
+    commit_threshold = new_commit_threshold;
+
+    if (changes_since_commit >= commit_threshold) {
+        flush_postlist_changes();
+        if (!transaction_active()) apply();
+    }
+
+    RETURN(old_commit_threshold);
+}
+
 void
 ChertWritableDatabase::add_spelling(const string & word,
                                     Xapian::termcount freqinc) const

backends/chert/chert_database.h

@@ -312 +312 @@
         /** The number of documents added, deleted, or replaced since the last
          *  flush.
          */
-        mutable Xapian::doccount change_count;
+        mutable Xapian::doccount changes_since_flush;
 
-        /// If change_count reaches this threshold we automatically flush.
+        /** The number of documents added, deleted, or replaced since the last
+         *  commit.
+         */
+        mutable Xapian::doccount changes_since_commit;
+
+        /** If changes_since_flush reaches this threshold we automatically
+         *  flush.
+         */
         Xapian::doccount flush_threshold;
 
+        /** If changes_since_commit reaches this threshold we automatically
+         *  commit.
+         */
+        Xapian::doccount commit_threshold;
+
         /** A pointer to the last document which was returned by
          *  open_document(), or NULL if there is no such valid document.  This
          *  is used purely for comparing with a supplied document to help with
@@ -338 +350 @@
         /// Apply changes.
         void apply();
 
+        /// Increment the change counts, and flush/commit if appropriate.
+        void inc_change_counts();
+
         /** Add or modify an entry in freq_deltas.
          *
          *  @param tname The term to modify the entry for.
@@ -383 +398 @@
         /** Cancel pending modifications to the database. */
         void cancel();
 
+        /// Set the threshold at which changes are flushed automatically.
+        Xapian::doccount set_flush_threshold(Xapian::doccount);
+
+        /// Set the threshold at which changes are committed automatically.
+        Xapian::doccount set_commit_threshold(Xapian::doccount);
+
         Xapian::docid add_document(const Xapian::Document & document);
         Xapian::docid add_document_(Xapian::docid did, const Xapian::Document & document);
         // Stop the default implementation of delete_document(term) and

backends/brass/brass_database.cc

@@ -1037 +1037 @@
 BrassWritableDatabase::BrassWritableDatabase(const string &dir, int action,
                                                int block_size)
         : BrassDatabase(dir, action, block_size),
-          change_count(0),
+          changes_since_flush(0),
+          changes_since_commit(0),
           flush_threshold(0),
+          commit_threshold(0),
           modify_shortcut_document(NULL),
           modify_shortcut_docid(0)
 {
@@ -1047 +1049 @@
     const char *p = getenv("XAPIAN_FLUSH_THRESHOLD");
     if (p)
         flush_threshold = atoi(p);
-    if (flush_threshold == 0)
+    else
         flush_threshold = 10000;
+    commit_threshold = flush_threshold;
 }
 
 BrassWritableDatabase::~BrassWritableDatabase()
@@ -1062 +1065 @@
 {
     if (transaction_active())
         throw Xapian::InvalidOperationError("Can't commit during a transaction");
-    if (change_count) flush_postlist_changes();
+    if (changes_since_flush)
+        flush_postlist_changes();
     apply();
 }
 
@@ -1072 +1076 @@
     stats.write(postlist_table);
     inverter.flush(postlist_table);
 
-    change_count = 0;
+    changes_since_flush = 0;
 }
 
 void
@@ -1091 +1095 @@
 {
     value_manager.set_value_stats(value_stats);
     BrassDatabase::apply();
+    changes_since_commit = 0;
 }
 
+void
+BrassWritableDatabase::inc_change_counts()
+{
+    // FIXME: this should be done by checking memory usage, not the number of
+    // changes.
+    // We could also look at:
+    // * mod_plists.size()
+    // * doclens.size()
+    // * freq_deltas.size()
+    //
+    // cout << "+++ mod_plists.size() " << mod_plists.size() <<
+    //     ", doclens.size() " << doclens.size() <<
+    //     ", freq_deltas.size() " << freq_deltas.size() << endl;
+
+    ++changes_since_commit;
+    ++changes_since_flush;
+
+    if (commit_threshold && changes_since_commit >= commit_threshold) {
+        flush_postlist_changes();
+        if (!transaction_active()) apply();
+    } else if (flush_threshold && changes_since_flush >= flush_threshold) {
+        flush_postlist_changes();
+    }
+}
+
 Xapian::docid
 BrassWritableDatabase::add_document(const Xapian::Document & document)
 {
@@ -1159 +1189 @@
         throw;
     }
 
-    // FIXME: this should be done by checking memory usage, not the number of
-    // changes.  We could also look at the amount of data the inverter object
-    // currently holds.
-    if (++change_count >= flush_threshold) {
-        flush_postlist_changes();
-        if (!transaction_active()) apply();
-    }
+    inc_change_counts();
 
     RETURN(did);
 }
@@ -1226 +1250 @@
         throw;
     }
 
-    if (++change_count >= flush_threshold) {
-        flush_postlist_changes();
-        if (!transaction_active()) apply();
-    }
+    inc_change_counts();
 }
 
 void
@@ -1391 +1412 @@
         throw;
     }
 
-    if (++change_count >= flush_threshold) {
-        flush_postlist_changes();
-        if (!transaction_active()) apply();
-    }
+    inc_change_counts();
 }
 
 Xapian::Document::Internal *
@@ -1497 +1515 @@
     // If there are changes, we don't have code to iterate the modified value
     // list so we need to flush (but don't commit - there may be a transaction
     // in progress).
-    if (change_count) value_manager.merge_changes();
+    if (changes_since_flush) value_manager.merge_changes();
     RETURN(BrassDatabase::open_value_list(slot));
 }
 
@@ -1505 +1523 @@
 BrassWritableDatabase::open_allterms(const string & prefix) const
 {
     LOGCALL(DB, TermList *, "BrassWritableDatabase::open_allterms", NO_ARGS);
-    if (change_count) {
+    if (changes_since_flush) {
         // There are changes, and terms may have been added or removed, and so
         // we need to flush changes for terms with the specified prefix (but
         // don't commit - there may be a transaction in progress).
         inverter.flush_post_lists(postlist_table, prefix);
         if (prefix.empty()) {
             // We've flushed all the posting list changes, but the document
-            // length and stats haven't been written, so set change_count to 1.
+            // length and stats haven't been written, so set
+            // changes_since_flush to 1.
             // FIXME: Can we handle this better?
-            change_count = 1;
+            changes_since_flush = 1;
         }
     }
     RETURN(BrassDatabase::open_allterms(prefix));
@@ -1528 +1547 @@
 
     inverter.clear();
     value_stats.clear();
-    change_count = 0;
+    changes_since_flush = 0;
+    changes_since_commit = 0;
 }
 
+doccount 
+BrassWritableDatabase::set_flush_threshold(doccount new_flush_threshold)
+{
+    LOGCALL(DB, doccount, "BrassWritableDatabase::set_flush_threshold",
+            new_flush_threshold);
+    doccount old_flush_threshold = flush_threshold;
+    flush_threshold = new_flush_threshold;
+
+    if (changes_since_flush >= flush_threshold) {
+        flush_postlist_changes();
+    }
+
+    RETURN(old_flush_threshold);
+}
+
+doccount 
+BrassWritableDatabase::set_commit_threshold(doccount new_commit_threshold)
+{
+    LOGCALL(DB, doccount, "BrassWritableDatabase::set_commit_threshold",
+            new_commit_threshold);
+    doccount old_commit_threshold = commit_threshold;
+    commit_threshold = new_commit_threshold;
+
+    if (changes_since_commit >= commit_threshold) {
+        flush_postlist_changes();
+        if (!transaction_active()) apply();
+    }
+
+    RETURN(old_commit_threshold);
+}
+
 void
 BrassWritableDatabase::add_spelling(const string & word,
                                     Xapian::termcount freqinc) const

backends/brass/brass_database.h

@@ -305 +305 @@
         /** The number of documents added, deleted, or replaced since the last
          *  flush.
          */
-        mutable Xapian::doccount change_count;
+        mutable Xapian::doccount changes_since_flush;
 
-        /// If change_count reaches this threshold we automatically flush.
+        /** The number of documents added, deleted, or replaced since the last
+         *  commit.
+         */
+        mutable Xapian::doccount changes_since_commit;
+
+        /** If changes_since_flush reaches this threshold we automatically
+         *  flush.
+         */
         Xapian::doccount flush_threshold;
 
+        /** If changes_since_commit reaches this threshold we automatically
+         *  commit.
+         */
+        Xapian::doccount commit_threshold;
+
         /** A pointer to the last document which was returned by
          *  open_document(), or NULL if there is no such valid document.  This
          *  is used purely for comparing with a supplied document to help with
@@ -331 +343 @@
         /// Apply changes.
         void apply();
 
+        /// Increment the change counts, and flush/commit if appropriate.
+        void inc_change_counts();
+
         //@{
         /** Implementation of virtual methods: see Database::Internal for
          *  details.
@@ -340 +355 @@
         /** Cancel pending modifications to the database. */
         void cancel();
 
+        /// Set the threshold at which changes are flushed automatically.
+        Xapian::doccount set_flush_threshold(Xapian::doccount);
+
+        /// Set the threshold at which changes are committed automatically.
+        Xapian::doccount set_commit_threshold(Xapian::doccount);
+
         Xapian::docid add_document(const Xapian::Document & document);
         Xapian::docid add_document_(Xapian::docid did, const Xapian::Document & document);
         // Stop the default implementation of delete_document(term) and