Added RO mode to permdb

7 files changed, 135 insertions, 15 deletions

diff --git a/c_src/filebuffer.c b/c_src/filebuffer.c
index 921ba37..40b79ad 100644
--- a/c_src/filebuffer.c
+++ b/c_src/filebuffer.c
@@ -231,6 +231,18 @@ bf_open(const char *path, int flags, const char *name, int lock)
 }
 
 void
+bf_reload(buffered_file *file)
+{
+        off_t datafile_filesize = lseek(file->fd, 0, SEEK_END);
+        if (datafile_filesize < 0) {
+                err(1, "lseek %s", file->name);
+        }
+        file->filesize = (uint64_t) datafile_filesize;
+        file->datasize = file->filesize;
+        file->lastcommit = file->datasize;
+}
+
+void
 bf_close(buffered_file *file)
 {
         bf_flush(file);
diff --git a/c_src/filebuffer.h b/c_src/filebuffer.h
index f24d86a..b2fb7ef 100644
--- a/c_src/filebuffer.h
+++ b/c_src/filebuffer.h
@@ -15,6 +15,9 @@ bf_close(buffered_file *file);
 void
 bf_truncate(buffered_file *file);
 
+void
+bf_reload(buffered_file *file);
+
 
 void
 bf_add(buffered_file *file, const void *data, uint64_t length);
diff --git a/c_src/permdb.c b/c_src/permdb.c
index a33c43a..f92958d 100644
--- a/c_src/permdb.c
+++ b/c_src/permdb.c
@@ -37,6 +37,7 @@ struct permdb_object {
         buffered_file *datafile;
         buffered_file *indexfile;
         char *error;
+        int write_enabled;
 };
 
 static const node_object nullnode = {{0, 0, 0, 0}};
@@ -447,8 +448,40 @@ rebuild_index_file(permdb_object *state)
         return committree(state);
 }
 
+unsigned int
+try_reload_database(permdb_object *state) {
+        if (state->write_enabled) {
+                errx(1, "try_reload_database called on write enabled database");
+        }
+
+        uint64_t oldindexsize = bf_total_length(state->indexfile);
+        
+        fprintf(stderr, "reloading database: datafile %llu indexfile %llu\n",
+                (unsigned long long) bf_total_length(state->datafile), 
+                (unsigned long long) bf_total_length(state->indexfile));
+        bf_reload(state->datafile);
+        bf_reload(state->indexfile);
+        fprintf(stderr, "reloaded database: datafile %llu indexfile %llu\n",
+                (unsigned long long) bf_total_length(state->datafile), 
+                (unsigned long long) bf_total_length(state->indexfile));
+
+        if (bf_total_length(state->indexfile) == oldindexsize) {
+                return 0;
+        } else {
+                if (datafile_verify_file(state->datafile) < 0) {
+                        warnx("data file verification failed");
+                }
+                if (indexfile_verify_file(state->indexfile) < 0) {
+                        warnx("cannot rebuild in readonly mode");
+                }
+
+                delete_all_nodes_in_cache(state);
+                return 1;
+        }
+}
+
 permdb_object *
-permdb_alloc(const char *dbpath)
+permdb_alloc(const char *dbpath, int lock)
 {
         char *idxpath = NULL;
         if (asprintf(&idxpath, "%s.idx", dbpath) == -1) {
@@ -464,14 +497,22 @@ permdb_alloc(const char *dbpath)
         state->datafile = NULL;
         state->indexfile = NULL;
 
-        state->datafile = bf_open(dbpath, O_RDWR|O_CREAT, "datafile", 1);
+        state->write_enabled = lock;
+
+        int mode = O_RDONLY;
+
+        if (state->write_enabled) {
+                mode = O_RDWR|O_CREAT;
+        }
+
+        state->datafile = bf_open(dbpath, mode, "datafile", state->write_enabled);
         if (state->datafile == NULL) {
                 permdb_free(state);
                 free(idxpath);
                 return NULL;
         }
 
-        state->indexfile = bf_open(idxpath, O_RDWR|O_CREAT, "indexfile", 1);
+        state->indexfile = bf_open(idxpath, mode, "indexfile", state->write_enabled);
         if (state->indexfile == NULL) {
                 permdb_free(state);
                 free(idxpath);
@@ -482,12 +523,22 @@ permdb_alloc(const char *dbpath)
 
         if (bf_total_length(state->datafile) == 0
             && bf_total_length(state->indexfile) == 0) {
+                if (!state->write_enabled) {
+                        warnx("data and index files are empty: %s", dbpath);
+                        permdb_free(state);
+                        return NULL;
+                }
                 dprintf(WRITE, (stderr, "writing header\n"));
                 indexfile_add_header(state->indexfile);
                 datafile_add_header(state->datafile);
                 initial_commit(state);
         } else if (bf_total_length(state->datafile) > 0
                    && bf_total_length(state->indexfile) == 0) {
+                if (!state->write_enabled) {
+                        warnx("cannot rebuild in readonly mode: %s", dbpath);
+                        permdb_free(state);
+                        return NULL;
+                }
                 if (rebuild_index_file(state) < 0) {
                         warnx("index file rebuilding failed: %s", dbpath);
                         permdb_free(state);
@@ -500,6 +551,11 @@ permdb_alloc(const char *dbpath)
                 return NULL;
         }
         if (indexfile_verify_file(state->indexfile) < 0) {
+                if (!state->write_enabled) {
+                        warnx("cannot rebuild in readonly mode: %s", dbpath);
+                        permdb_free(state);
+                        return NULL;
+                }
                 warnx("index file verification failed, rebuilding: %s", dbpath);
 
                 if (rebuild_index_file(state) < 0) {
@@ -933,6 +989,10 @@ addvalue(permdb_object *state, const unsigned char *key, unsigned int keylength,
 
         node_entry lastentry = getpath(state, key, nodes);
 
+        if (!state->write_enabled) {
+                return -1;
+        }
+
         if (lastentry == NODE_ENTRY_ERROR_NODE) {
                 utarray_free(nodes);
                 return -1;
@@ -1003,8 +1063,8 @@ addvalue(permdb_object *state, const unsigned char *key, unsigned int keylength,
 }
 
 unsigned char *
-getvalue(permdb_object *state, const unsigned char *key, size_t keylength,
-         size_t *datalen)
+getvalue_try(permdb_object *state, const unsigned char *key, size_t keylength,
+             size_t *datalen)
 {
         node_entry entry = getlastnode(state, key);
         if (entry == 0) {
@@ -1028,8 +1088,24 @@ getvalue(permdb_object *state, const unsigned char *key, size_t keylength,
         return readdata(state, olddataoffset, *datalen);
 }
 
+unsigned char *
+getvalue(permdb_object *state, const unsigned char *key, size_t keylength,
+             size_t *datalen)
+{
+        unsigned char *result = getvalue_try(state, key, keylength, datalen);
+
+        if (result == NULL && !state->write_enabled) {
+                int reloaded = try_reload_database(state);
+                if (reloaded) {
+                        return getvalue_try(state, key, keylength, datalen);
+                }
+        }
+        
+        return result;
+}
+
 unsigned int
-keyexists(permdb_object *state, const unsigned char *key, size_t keylength)
+keyexists_try(permdb_object *state, const unsigned char *key, size_t keylength)
 {
         node_entry entry = getlastnode(state, key);
         if (entry == 0) {
@@ -1053,6 +1129,21 @@ keyexists(permdb_object *state, const unsigned char *key, size_t keylength)
         return 1;
 }
 
+unsigned int
+keyexists(permdb_object *state, const unsigned char *key, size_t keylength)
+{
+        unsigned int result = keyexists_try(state, key, keylength);
+
+        if (result == 0 && !state->write_enabled) {
+                int reloaded = try_reload_database(state);
+                if (reloaded) {
+                        return keyexists_try(state, key, keylength);
+                }
+        }
+        
+        return result;
+}
+
 static int
 string_length_comparison(struct nodecache *a, struct nodecache *b) {
         size_t a_len = a->key->length;
@@ -1081,6 +1172,10 @@ string_length_comparison(struct nodecache *a, struct nodecache *b) {
 int
 committree(permdb_object *state)
 {
+        if (!state->write_enabled) {
+                return -1;
+        }
+        
         if (state->dirtynodes == NULL) {
                 return 0;
         }
diff --git a/c_src/permdb.h b/c_src/permdb.h
index f6bee7f..4625a7b 100644
--- a/c_src/permdb.h
+++ b/c_src/permdb.h
@@ -46,7 +46,7 @@ void
 portloop(permdb_object *state);
 
 permdb_object *
-permdb_alloc(const char *dbpath);
+permdb_alloc(const char *dbpath, int lock);
 
 void
 permdb_free(permdb_object *state);
diff --git a/c_src/permdbport.c b/c_src/permdbport.c
index f5f151d..1db6df7 100644
--- a/c_src/permdbport.c
+++ b/c_src/permdbport.c
@@ -17,18 +17,27 @@
 static void __attribute__((noreturn))
 usage()
 {
-        errx(1, "usage: permdbport <path>");
+        errx(1, "usage: permdbport <path> [nolock]");
 }
 
 int
 main(int argc, char *argv[])
 {
-        if (argc != 2) {
+        if (argc < 2) {
                 usage();
         }
         const char *store = argv[1];
+        int lock = 1;
+        for (int i = 2; i < argc; i++) {
+                const char *arg = argv[i];
+                if (strcmp(arg, "nolock") == 0) {
+                        lock = 0;
+                } else {
+                        usage();
+                }
+        }
 
-        permdb_object *state = permdb_alloc(store);
+        permdb_object *state = permdb_alloc(store, lock);
 
         if (state == NULL) {
                 fprintf(stderr, "permdbport failed to start\n");
diff --git a/c_src/permdbpy.c b/c_src/permdbpy.c
index 9bd2e8a..491f759 100644
--- a/c_src/permdbpy.c
+++ b/c_src/permdbpy.c
@@ -43,11 +43,11 @@ PyTypeObject permdb_type = {
 };
 
 permdb_object_py *
-permdb_alloc_py(const char *dbpath)
+permdb_alloc_py(const char *dbpath, int write_enable)
 {
         struct permdb_object *permdb;
 
-        permdb = permdb_alloc(dbpath);
+        permdb = permdb_alloc(dbpath, write_enable);
 
         if (permdb == NULL) {
                 PyErr_SetString(PyExc_RuntimeError, "Cannot allocate permdb object");
@@ -84,12 +84,13 @@ static PyObject *
 permdb_alloc_wrapper(PyObject *self, PyObject *args)
 {
         const char *dbpath = NULL;
+        int write_enable = 1;
 
-        if (!PyArg_ParseTuple(args, "s", &dbpath)) {
+        if (!PyArg_ParseTuple(args, "s|i", &dbpath, &write_enable)) {
                 return NULL;
         }
 
-        return (PyObject*)permdb_alloc_py(dbpath);
+        return (PyObject*)permdb_alloc_py(dbpath, write_enable);
 }
 
 static PyObject *
diff --git a/c_src/permdbtest.c b/c_src/permdbtest.c
index 2ab6510..04e6cfc 100644
--- a/c_src/permdbtest.c
+++ b/c_src/permdbtest.c
@@ -65,7 +65,7 @@ main(int argc, char *argv[])
         int datasize = atoi(argv[3]);
         int nfsync = atoi(argv[4]);
 
-        permdb_object *state = permdb_alloc(store);
+        permdb_object *state = permdb_alloc(store, 1);
 
         if (state == NULL) {
                 errx(1, "permdb object creation failed\n");