Added RO mode to permdb

9 files changed, 172 insertions, 31 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");
diff --git a/src/permdb.erl b/src/permdb.erl
index 461b8b3..ba9fd72 100644
--- a/src/permdb.erl
+++ b/src/permdb.erl
@@ -5,7 +5,7 @@
 
 -behaviour(gen_server).
 
--export([start_link/2, stop/1, init_module/0]).
+-export([start_link/2, start_link/3, stop/1, init_module/0]).
 -export([getvalue/2, addvalue/3, commit/1, commit/2, keyexists/2]).
 
 %% gen_server callbacks.
@@ -40,12 +40,18 @@ commit(Name) ->
 commit(Name, Timeout) ->
     gen_server:call(Name, {commit}, Timeout).
 
-init([Name, Filename]) ->
+init([Name, Filename, WriteFlag]) ->
     Cachename = list_to_atom(atom_to_list(Name) ++ "_cache"),
     ets:new(Cachename, [set, public, named_table]),
     process_flag(trap_exit, true),
+    WriteFlagArg = case WriteFlag of
+                       write ->
+                           [];
+                       _ ->
+                           ["nolock"]
+                   end,
     Port = open_port({spawn_executable, code:priv_dir(plop) ++ "/permdbport"},
-                     [{packet, 4}, {args, [Filename]}, binary]),
+                     [{packet, 4}, {args, [Filename | WriteFlagArg]}, binary]),
     {ok, #state{cachename = Cachename,
                 name = Name,
                 port = Port,
@@ -56,8 +62,11 @@ init_module() ->
     ok.
 
 start_link(Name, Filename) ->
+    start_link(Name, Filename, write).
+
+start_link(Name, Filename, WriteFlag) ->
     gen_server:start_link({local, Name}, ?MODULE,
-                          [Name, Filename], []).
+                          [Name, Filename, WriteFlag], []).
 
 stop(Name) ->
     gen_server:call(Name, stop).
diff --git a/test/permdbtest.erl b/test/permdbtest.erl
index 1c43861..5d0453b 100755
--- a/test/permdbtest.erl
+++ b/test/permdbtest.erl
@@ -14,7 +14,7 @@ timeprint(Time) ->
     io_lib:format("~.2fs", [Time/1000000]).
 
 testinit(Filename) ->
-    permdb:start_link(testdb, Filename).
+    permdb:start_link(testdb, Filename, write).
 
 teststop() ->
     permdb:stop(testdb).
@@ -24,42 +24,48 @@ constructdata(VSeed, Size) ->
     B = binary:part(VSeed, 0, Size rem 32),
     <<A/binary, B/binary>>.
 
-getvalue_loop([], _Port, _Datasize) ->
+getvalue_loop([], _Port, _Datasize, _DB) ->
     none;
-getvalue_loop([{K, VSeed}|Rest], Port, Datasize) ->
+getvalue_loop([{K, VSeed}|Rest], Port, Datasize, DB) ->
     V = case VSeed of
             noentry ->
                 noentry;
             _ ->
                 constructdata(VSeed, Datasize)
         end,
-    case permdb:getvalue(testdb, K) of
+    case permdb:getvalue(DB, K) of
         V ->
-            getvalue_loop(Rest, Port, Datasize);
+            getvalue_loop(Rest, Port, Datasize, DB);
         VOther ->
             io:format("expected: ~p got: ~p~nkey: ~p~n", [V, VOther, K]),
             exit(mismatch)
     end.
 
-addvalue_loop([], _Port, _Datasize) ->
+addvalue_loop([], _Port, _Datasize, _DB) ->
     none;
-addvalue_loop([{K, VSeed}|Rest], Port, Datasize) ->
+addvalue_loop([{K, VSeed}|Rest], Port, Datasize, DB) ->
     V = constructdata(VSeed, Datasize),
-    case permdb:addvalue(testdb, K, V) of
+    case permdb:addvalue(DB, K, V) of
         ok ->
-            addvalue_loop(Rest, Port, Datasize);
+            addvalue_loop(Rest, Port, Datasize, DB);
         Other ->
             io:format("expected: 0 or 1 got: ~p~n", [Other]),
             exit(mismatch)
     end.
 
 testget(_Filename, TestData, Datasize) ->
-    getvalue_loop(TestData, none, Datasize),
+    testget(_Filename, TestData, Datasize, testdb).
+
+testget(_Filename, TestData, Datasize, DB) ->
+    getvalue_loop(TestData, none, Datasize, DB),
     ok.
 
 testadd(_Filename, TestData, Datasize) ->
-    addvalue_loop(TestData, none, Datasize),
-    case permdb:commit(testdb) of
+    testadd(_Filename, TestData, Datasize, testdb).
+
+testadd(_Filename, TestData, Datasize, DB) ->
+    addvalue_loop(TestData, none, Datasize, DB),
+    case permdb:commit(DB) of
         <<0>> ->
             ok;
         Other ->
@@ -138,7 +144,13 @@ main([]) ->
     stop(),
 
     testinit(Filename),
+
+    permdb:start_link(testdb_ro, Filename, read),
+
     testget(Filename, gentestdata(1+2+3+4), 99),
+    testadd(Filename, gentestdata(1+2+3+4+5), 99),
+    testget(Filename, gentestdata(1+2+3+4+5), 99, testdb_ro),
+    permdb:stop(testdb_ro),
     stop(),
 
     ok.