Getting optional LMDB support into Cytoplasm #43

Merged
jordan merged 14 commits from lda/Cytoplasm:lmdbwerk into master 2024-08-23 20:19:59 +00:00
5 changed files with 86 additions and 24 deletions
Showing only changes of commit 3df1e4ab7b - Show all commits

View file

@ -48,6 +48,17 @@
*/
typedef struct Db Db;
/**
* Some "hints" for the database backend for operations.
* Hints are a way for the program to declare what to except of it
* (and the program MUST adhere to these hints, but the backend
* MAY adhere).
*/
typedef enum DbHint {
DB_HINT_READONLY, /* The database reference is treated as read-only */
DB_HINT_WRITE /* The database reference is treated as read/write */
} DbHint;
/**
* When an object is locked, a reference is returned. This reference
* is owned by the current thread, and the database is inaccessible to
@ -118,6 +129,15 @@ extern DbRef * DbCreate(Db *, size_t,...);
*/
extern DbRef * DbLock(Db *, size_t,...);
/**
* Behaves like
* .Fn DbLock ,
* but with hints on the reference itself, as
* .Fn DbLock
* itself is read/write.
*/
extern DbRef * DbLockIntent(Db *, DbHint, size_t,...);
/**
* Immediately and permanently remove an object from the database.
* This function assumes the object is not locked, otherwise undefined

View file

@ -327,7 +327,30 @@ DbLock(Db * db, size_t nArgs,...)
return NULL;
}
ret = db->lockFunc(db, args);
ret = db->lockFunc(db, DB_HINT_WRITE, args);
ArrayFree(args);
return ret;
}
DbRef *
DbLockIntent(Db * db, DbHint hint, size_t nArgs,...)
{
va_list ap;
Array *args;
DbRef *ret;
va_start(ap, nArgs);
args = ArrayFromVarArgs(nArgs, ap);
va_end(ap);
if (!args || !db->lockFunc)
{
return NULL;
}
ret = db->lockFunc(db, hint, args);
ArrayFree(args);

View file

@ -101,7 +101,7 @@ DbFileName(FlatDb * db, Array * args)
}
static DbRef *
FlatLock(Db *d, Array *dir)
FlatLock(Db *d, DbHint hint, Array *dir)
{
FlatDb *db = (FlatDb *) d;
FlatDbRef *ref = NULL;
@ -146,6 +146,8 @@ FlatLock(Db *d, Array *dir)
ref = Malloc(sizeof(*ref));
DbRefInit(d, (DbRef *) ref);
/* TODO: Hints */
ref->base.hint = hint;
ref->base.ts = UtilLastModified(path);
ref->base.json = JsonDecode(stream);
ref->stream = stream;
@ -248,7 +250,7 @@ FlatCreate(Db *d, Array *dir)
/* FlatLock() will lock again for us */
pthread_mutex_unlock(&d->lock);
ret = FlatLock(d, dir);
ret = FlatLock(d, DB_HINT_WRITE, dir);
return ret;
}

View file

@ -66,7 +66,7 @@ struct Db
/* Functions for implementation-specific operations
* (opening a ref, closing a db, removing an entry, ...) */
DbRef * (*lockFunc)(Db *, Array *);
DbRef * (*lockFunc)(Db *, DbHint, Array *);
DbRef * (*create)(Db *, Array *);
Array * (*list)(Db *, Array *);
bool (*unlock)(Db *, DbRef *);
@ -89,8 +89,7 @@ struct DbRef
DbRef *prev;
DbRef *next;
/* TODO: Functions for implementation-specific operations */
DbHint hint;
/* Implementation-specific constructs */
};

View file

@ -135,14 +135,14 @@ LMDBKeyHead(MDB_val key)
}
static DbRef *
LMDBLock(Db *d, Array *k)
LMDBLock(Db *d, DbHint hint, Array *k)
{
LMDB *db = (LMDB *) d;
MDB_txn *transaction;
LMDBRef *ret = NULL;
MDB_val key, empty_json;
MDB_dbi dbi;
int code;
int code, flags;
if (!d || !k)
{
return NULL;
@ -152,7 +152,8 @@ LMDBLock(Db *d, Array *k)
key = LMDBTranslateKey(k);
/* create a txn */
if ((code = mdb_txn_begin(db->environ, NULL, 0, &transaction)) != 0)
flags = hint == DB_HINT_READONLY ? MDB_RDONLY : 0;
if ((code = mdb_txn_begin(db->environ, NULL, flags, &transaction)) != 0)
{
/* Very bad! */
Log(LOG_ERR,
@ -206,10 +207,21 @@ LMDBLock(Db *d, Array *k)
}
}
ret->base.json = LMDBDecode(empty_json);
ret->transaction = transaction;
ret->dbi = dbi;
ret->base.hint = hint;
ret->transaction = NULL;
if (hint == DB_HINT_WRITE)
{
ret->transaction = transaction;
ret->dbi = dbi;
}
else
{
mdb_txn_abort(transaction);
mdb_dbi_close(db->environ, dbi);
}
end:
if (!ret)
if (!ret || hint == DB_HINT_READONLY)
{
pthread_mutex_unlock(&d->lock);
}
@ -319,7 +331,7 @@ LMDBUnlock(Db *d, DbRef *r)
FILE *fakestream;
Stream *stream;
MDB_val key, val;
bool ret;
bool ret = true;
if (!d || !r)
{
@ -328,28 +340,31 @@ LMDBUnlock(Db *d, DbRef *r)
val.mv_data = NULL;
val.mv_size = 0;
key = LMDBTranslateKey(r->name);
if (ref->transaction && r->hint == DB_HINT_WRITE)
{
key = LMDBTranslateKey(r->name);
fakestream = open_memstream((char **) &val.mv_data, &val.mv_size);
stream = StreamFile(fakestream);
JsonEncode(r->json, stream, JSON_DEFAULT);
StreamFlush(stream);
StreamClose(stream);
fakestream = open_memstream((char **) &val.mv_data, &val.mv_size);
stream = StreamFile(fakestream);
JsonEncode(r->json, stream, JSON_DEFAULT);
StreamFlush(stream);
StreamClose(stream);
ret = mdb_put(ref->transaction, ref->dbi, &key, &val, 0) == 0;
ret = mdb_put(ref->transaction, ref->dbi, &key, &val, 0) == 0;
mdb_txn_commit(ref->transaction);
mdb_dbi_close(db->environ, ref->dbi);
mdb_txn_commit(ref->transaction);
mdb_dbi_close(db->environ, ref->dbi);
LMDBKillKey(key);
}
StringArrayFree(ref->base.name);
JsonFree(ref->base.json);
Free(ref);
LMDBKillKey(key);
if (val.mv_data)
{
free(val.mv_data);
}
if (ret)
if (ret && r->hint == DB_HINT_WRITE)
{
pthread_mutex_unlock(&d->lock);
}
@ -430,6 +445,7 @@ LMDBCreate(Db *d, Array *k)
StringArrayAppend(ret->base.name, ent);
}
}
ret->base.hint = DB_HINT_WRITE;
ret->base.json = HashMapCreate();
ret->transaction = transaction;
ret->dbi = dbi;
@ -464,6 +480,8 @@ LMDBList(Db *d, Array *k)
pthread_mutex_lock(&d->lock);
/* Marked as read-only, as we just don't need to write anything
* when listing */
if ((code = mdb_txn_begin(db->environ, NULL, 0, &txn)) != 0)
{
/* Very bad! */