log: Replace ovsdb_log_get_offset() by a more abstract mechanism.

Upcoming support for clustered databases will need to provide a more
abstract way to determine when a given file should be compacted, so this
changes the standalone database support to use this mechanism in advance.

Signed-off-by: Ben Pfaff <blp@ovn.org>
Acked-by: Justin Pettit <jpettit@ovn.org>

3 files changed, 62 insertions, 25 deletions

diff --git a/ovsdb/file.c b/ovsdb/file.c
index 983759e35..90c2b9d20 100644
--- a/ovsdb/file.c
+++ b/ovsdb/file.c
@@ -73,7 +73,6 @@ static struct ovsdb_error *ovsdb_file_create(struct ovsdb *,
                                              struct ovsdb_log *,
                                              const char *file_name,
                                              unsigned int n_transactions,
-                                             off_t snapshot_size,
                                              struct ovsdb_file **filep);
 
 /* Opens database 'file_name' and stores a pointer to the new database in
@@ -209,7 +208,6 @@ ovsdb_file_open__(const char *file_name,
      *
      * The schema precedes the snapshot in the log; we could compensate for its
      * size, but it's just not that important. */
-    off_t snapshot_size = 0;
     unsigned int n_transactions = 0;
     while ((error = ovsdb_log_read(log, &json)) == NULL && json) {
         struct ovsdb_txn *txn;
@@ -230,7 +228,7 @@ ovsdb_file_open__(const char *file_name,
         }
 
         if (n_transactions == 1) {
-            snapshot_size = ovsdb_log_get_offset(log);
+            ovsdb_log_mark_base(log);
         }
     }
     if (error) {
@@ -245,8 +243,7 @@ ovsdb_file_open__(const char *file_name,
     if (!read_only) {
         struct ovsdb_file *file;
 
-        error = ovsdb_file_create(db, log, file_name, n_transactions,
-                                  snapshot_size, &file);
+        error = ovsdb_file_create(db, log, file_name, n_transactions, &file);
         if (error) {
             goto error;
         }
@@ -513,7 +510,6 @@ struct ovsdb_file {
     long long int last_compact;
     long long int next_compact;
     unsigned int n_transactions;
-    off_t snapshot_size;
 };
 
 static const struct ovsdb_replica_class ovsdb_file_class;
@@ -521,8 +517,7 @@ static const struct ovsdb_replica_class ovsdb_file_class;
 static struct ovsdb_error *
 ovsdb_file_create(struct ovsdb *db, struct ovsdb_log *log,
                   const char *file_name,
-                  unsigned int n_transactions, off_t snapshot_size,
-                  struct ovsdb_file **filep)
+                  unsigned int n_transactions, struct ovsdb_file **filep)
 {
     struct ovsdb_file *file;
     char *deref_name;
@@ -546,7 +541,6 @@ ovsdb_file_create(struct ovsdb *db, struct ovsdb_log *log,
     file->file_name = abs_name;
     file->last_compact = time_msec();
     file->next_compact = file->last_compact + COMPACT_MIN_MSEC;
-    file->snapshot_size = snapshot_size;
     file->n_transactions = n_transactions;
     ovsdb_add_replica(db, &file->replica);
 
@@ -612,12 +606,9 @@ ovsdb_file_commit(struct ovsdb_replica *replica,
      * tried), and if there are at least 100 transactions in the database, and
      * if the database is at least 10 MB, and the database is at least 4x the
      * size of the previous snapshot, then compact the database. */
-    off_t log_size = ovsdb_log_get_offset(file->log);
     if (time_msec() >= file->next_compact
         && file->n_transactions >= 100
-        && log_size >= 10 * 1024 * 1024
-        && log_size / 4 >= file->snapshot_size)
-    {
+        && ovsdb_log_grew_lots(file->log)) {
         error = ovsdb_file_compact(file);
         if (error) {
             char *s = ovsdb_error_to_string_free(error);
@@ -663,10 +654,9 @@ ovsdb_file_compact(struct ovsdb_file *file)
     int retval;
 
     comment = xasprintf("compacting database online "
-                        "(%.3f seconds old, %u transactions, %llu bytes)",
+                        "(%.3f seconds old, %u transactions)",
                         (time_wall_msec() - file->last_compact) / 1000.0,
-                        file->n_transactions,
-                        (unsigned long long) ovsdb_log_get_offset(file->log));
+                        file->n_transactions);
     VLOG_INFO("%s: %s", file->file_name, comment);
 
     /* Commit the old version, so that we can be assured that we'll eventually
@@ -696,6 +686,7 @@ ovsdb_file_compact(struct ovsdb_file *file)
     if (error) {
         goto exit;
     }
+    ovsdb_log_mark_base(new_log);
 
     /* Replace original file by the temporary file.
      *
@@ -750,6 +741,7 @@ ovsdb_file_compact(struct ovsdb_file *file)
             ovs_fatal(0, "error reading database");
         }
         json_destroy(json);
+        ovsdb_log_mark_base(file->log);
     }
 
     /* Success! */
diff --git a/ovsdb/log.c b/ovsdb/log.c
index a5243ce52..5c15e53db 100644
--- a/ovsdb/log.c
+++ b/ovsdb/log.c
@@ -76,6 +76,7 @@ struct ovsdb_log {
     char *magic;
     struct lockfile *lockfile;
     FILE *stream;
+    off_t base;
 };
 
 /* Whether the OS supports renaming open files.
@@ -105,6 +106,12 @@ static bool is_magic_ok(const char *needle, const char *haystack);
  * Whether the file will be locked using lockfile_lock() depends on 'locking':
  * use true to lock it, false not to lock it, or -1 to lock it only if
  * 'open_mode' is a mode that allows writing.
+ *
+ * A log consists of a series of records.  After opening or creating a log with
+ * this function, the client may use ovsdb_log_read() to read any existing
+ * records, one by one.  The client may also use ovsdb_log_write() to write new
+ * records (if some records have not yet been read at this point, then the
+ * first write truncates them).
  */
 struct ovsdb_error *
 ovsdb_log_open(const char *name, const char *magic,
@@ -245,6 +252,7 @@ ovsdb_log_open(const char *name, const char *magic,
     file->stream = stream;
     file->prev_offset = 0;
     file->offset = 0;
+    file->base = 0;
     *filep = file;
     return NULL;
 
@@ -519,7 +527,14 @@ ovsdb_log_compose_record(const struct json *json,
 }
 
 /* Writes log record 'json' to 'file'.  Returns NULL if successful or an error
- * (which the caller must eventually destroy) on failure. */
+ * (which the caller must eventually destroy) on failure.
+ *
+ * If the log contains some records that have not yet been read, then calling
+ * this function truncates them.
+ *
+ * Log writes are atomic.  A client may use ovsdb_log_commit() to ensure that
+ * they are durable.
+ */
 struct ovsdb_error *
 ovsdb_log_write(struct ovsdb_log *file, const struct json *json)
 {
@@ -599,13 +614,21 @@ ovsdb_log_commit(struct ovsdb_log *file)
     return NULL;
 }
 
-/* Returns the current offset into the file backing 'log', in bytes.  This
- * reflects the number of bytes that have been read or written in the file.  If
- * the whole file has been read, this is the file size. */
-off_t
-ovsdb_log_get_offset(const struct ovsdb_log *log)
+/* Sets the current position in 'log' as the "base", that is, the initial size
+ * of the log that ovsdb_log_grew_lots() uses to determine whether the log has
+ * grown enough to make compacting worthwhile. */
+void
+ovsdb_log_mark_base(struct ovsdb_log *log)
 {
-    return log->offset;
+    log->base = log->offset;
+}
+
+/* Returns true if 'log' has grown enough above the base that it's worthwhile
+ * to compact it, false otherwise. */
+bool
+ovsdb_log_grew_lots(const struct ovsdb_log *log)
+{
+    return log->offset > 10 * 1024 * 1024 && log->offset / 4 > log->base;
 }
 
 /* Attempts to atomically replace the contents of 'log', on disk, by the 'n'
@@ -631,6 +654,7 @@ ovsdb_log_replace(struct ovsdb_log *log, struct json **entries, size_t n)
             return error;
         }
     }
+    ovsdb_log_mark_base(new);
 
     return ovsdb_log_replace_commit(log, new);
 }
@@ -765,7 +789,8 @@ ovsdb_log_replace_commit(struct ovsdb_log *old, struct ovsdb_log *new)
     old->magic = new->magic;
     new->magic = NULL;
     /* Keep old->lockfile. */
-    /* stream was already replaced above. */
+    old->base = new->base;
+
     /* Free 'new'. */
     ovsdb_log_close(new);
 
diff --git a/ovsdb/log.h b/ovsdb/log.h
index 6bb2aaa21..18900fa50 100644
--- a/ovsdb/log.h
+++ b/ovsdb/log.h
@@ -16,6 +16,25 @@
 #ifndef OVSDB_LOG_H
 #define OVSDB_LOG_H 1
 
+/* OVSDB log.
+ *
+ * A log consists of a series of records.  After opening or creating a log with
+ * ovsdb_log_open(), the client may use ovsdb_log_read() to read any existing
+ * records, one by one.  The client may also use ovsdb_log_write() to write new
+ * records (if some records have not yet been read at this point, then the
+ * first write truncates them).
+ *
+ * Log writes are atomic.  A client may use ovsdb_log_commit() to ensure that
+ * they are durable.
+ *
+ * Logs provide a mechansim to allow the client to tell when they have grown
+ * enough that compacting may be warranted.  After reading existing log
+ * contents, the client uses ovsdb_log_mark_base() to mark the "base" to be
+ * considered as the initial size of the log.  Thereafter, a client may call
+ * ovsdb_log_grew_lots() to get an indication whether the log has grown enough
+ * that compacting is advised.
+ */
+
 #include <sys/types.h>
 #include "compiler.h"
 
@@ -54,7 +73,8 @@ struct ovsdb_error *ovsdb_log_write(struct ovsdb_log *, const struct json *)
 struct ovsdb_error *ovsdb_log_commit(struct ovsdb_log *)
     OVS_WARN_UNUSED_RESULT;
 
-off_t ovsdb_log_get_offset(const struct ovsdb_log *);
+void ovsdb_log_mark_base(struct ovsdb_log *);
+bool ovsdb_log_grew_lots(const struct ovsdb_log *);
 
 struct ovsdb_error *ovsdb_log_replace(struct ovsdb_log *,
                                       struct json **entries, size_t n)