Compare commits

...

3 commits

Author SHA1 Message Date
LDA
cf1b78b224 Merge branch 'lmdbwerk' of https://git.telodendria.io/lda/Cytoplasm into lmdbwerk 2024-08-11 15:29:16 +02:00
LDA
e133eebef3 [MOD/WIP] Use one shared DBI
Yeah, I was stupid for not doing that earlier.
2024-08-11 15:28:21 +02:00
LDA
3df1e4ab7b [ADD/WIP] Start rolling in intents 2024-08-11 12:46:49 +02:00
5 changed files with 122 additions and 102 deletions

View file

@ -48,6 +48,17 @@
*/ */
typedef struct Db Db; 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 * When an object is locked, a reference is returned. This reference
* is owned by the current thread, and the database is inaccessible to * 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,...); 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. * Immediately and permanently remove an object from the database.
* This function assumes the object is not locked, otherwise undefined * This function assumes the object is not locked, otherwise undefined

View file

@ -327,7 +327,30 @@ DbLock(Db * db, size_t nArgs,...)
return NULL; 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); ArrayFree(args);

View file

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

View file

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

View file

@ -17,6 +17,7 @@ typedef struct LMDB {
Db base; /* The base implementation required to pass */ Db base; /* The base implementation required to pass */
MDB_env *environ; MDB_env *environ;
MDB_dbi dbi;
} LMDB; } LMDB;
typedef struct LMDBRef { typedef struct LMDBRef {
DbRef base; DbRef base;
@ -135,14 +136,13 @@ LMDBKeyHead(MDB_val key)
} }
static DbRef * static DbRef *
LMDBLock(Db *d, Array *k) LMDBLock(Db *d, DbHint hint, Array *k)
{ {
LMDB *db = (LMDB *) d; LMDB *db = (LMDB *) d;
MDB_txn *transaction; MDB_txn *transaction;
LMDBRef *ret = NULL; LMDBRef *ret = NULL;
MDB_val key, empty_json; MDB_val key, json_val;
MDB_dbi dbi; int code, flags;
int code;
if (!d || !k) if (!d || !k)
{ {
return NULL; return NULL;
@ -151,8 +151,11 @@ LMDBLock(Db *d, Array *k)
pthread_mutex_lock(&d->lock); pthread_mutex_lock(&d->lock);
key = LMDBTranslateKey(k); key = LMDBTranslateKey(k);
/* create a txn */ /* Create a transaction, honoring hints. */
if ((code = mdb_txn_begin(db->environ, NULL, 0, &transaction)) != 0) /* TODO: Do we want to create a "main" transaction that everyone inherits
* from? */
flags = hint == DB_HINT_READONLY ? MDB_RDONLY : 0;
if ((code = mdb_txn_begin(db->environ, NULL, flags, &transaction)) != 0)
{ {
/* Very bad! */ /* Very bad! */
Log(LOG_ERR, Log(LOG_ERR,
@ -161,25 +164,13 @@ LMDBLock(Db *d, Array *k)
); );
goto end; goto end;
} }
/* apparently you need to give it a dbi */
if ((code = mdb_dbi_open(transaction, "db", MDB_CREATE, &dbi)) != 0)
{
Log(LOG_ERR,
"%s: could not get transaction dbi: %s",
__func__, mdb_strerror(code)
);
goto end;
}
empty_json.mv_size = 0; json_val.mv_size = 0;
empty_json.mv_data = NULL; json_val.mv_data = NULL;
/* get data from it */ code = mdb_get(transaction, db->dbi, &key, &json_val);
code = mdb_get(transaction, dbi, &key, &empty_json);
if (code == MDB_NOTFOUND) if (code == MDB_NOTFOUND)
{ {
/* No use logging it, as that is just locking behaviour */
mdb_txn_abort(transaction); mdb_txn_abort(transaction);
mdb_dbi_close(db->environ, dbi);
goto end; goto end;
} }
else if (code != 0) else if (code != 0)
@ -189,7 +180,6 @@ LMDBLock(Db *d, Array *k)
__func__, mdb_strerror(code) __func__, mdb_strerror(code)
); );
mdb_txn_abort(transaction); mdb_txn_abort(transaction);
mdb_dbi_close(db->environ, dbi);
goto end; goto end;
} }
@ -205,11 +195,20 @@ LMDBLock(Db *d, Array *k)
StringArrayAppend(ret->base.name, ent); StringArrayAppend(ret->base.name, ent);
} }
} }
ret->base.json = LMDBDecode(empty_json); ret->base.json = LMDBDecode(json_val);
ret->transaction = transaction; ret->base.hint = hint;
ret->dbi = dbi; ret->transaction = NULL;
if (hint == DB_HINT_WRITE)
{
ret->transaction = transaction;
}
else
{
mdb_txn_abort(transaction);
}
end: end:
if (!ret) if (!ret || hint == DB_HINT_READONLY)
{ {
pthread_mutex_unlock(&d->lock); pthread_mutex_unlock(&d->lock);
} }
@ -222,7 +221,6 @@ LMDBExists(Db *d, Array *k)
MDB_val key, empty; MDB_val key, empty;
LMDB *db = (LMDB *) d; LMDB *db = (LMDB *) d;
MDB_txn *transaction; MDB_txn *transaction;
MDB_dbi dbi;
int code; int code;
bool ret = false; bool ret = false;
if (!d || !k) if (!d || !k)
@ -243,21 +241,9 @@ LMDBExists(Db *d, Array *k)
); );
goto end; goto end;
} }
/* apparently you need to give it a dbi */
if ((code = mdb_dbi_open(transaction, "db", MDB_CREATE, &dbi)) != 0)
{
Log(LOG_ERR,
"%s: could not get transaction dbi: %s",
__func__, mdb_strerror(code)
);
goto end;
} ret = mdb_get(transaction, db->dbi, &key, &empty) == 0;
ret = mdb_get(transaction, dbi, &key, &empty) == 0;
mdb_txn_abort(transaction); mdb_txn_abort(transaction);
mdb_dbi_close(db->environ, dbi);
end: end:
LMDBKillKey(key); LMDBKillKey(key);
pthread_mutex_unlock(&d->lock); pthread_mutex_unlock(&d->lock);
@ -269,7 +255,6 @@ LMDBDelete(Db *d, Array *k)
MDB_val key, empty; MDB_val key, empty;
LMDB *db = (LMDB *) d; LMDB *db = (LMDB *) d;
MDB_txn *transaction; MDB_txn *transaction;
MDB_dbi dbi;
int code; int code;
bool ret = false; bool ret = false;
if (!d || !k) if (!d || !k)
@ -290,21 +275,9 @@ LMDBDelete(Db *d, Array *k)
); );
goto end; goto end;
} }
/* apparently you need to give it a dbi */
if ((code = mdb_dbi_open(transaction, "db", MDB_CREATE, &dbi)) != 0)
{
Log(LOG_ERR,
"%s: could not get transaction dbi: %s",
__func__, mdb_strerror(code)
);
goto end;
} ret = mdb_del(transaction, db->dbi, &key, &empty) == 0;
ret = mdb_del(transaction, dbi, &key, &empty) == 0;
mdb_txn_commit(transaction); mdb_txn_commit(transaction);
mdb_dbi_close(db->environ, dbi);
end: end:
LMDBKillKey(key); LMDBKillKey(key);
pthread_mutex_unlock(&d->lock); pthread_mutex_unlock(&d->lock);
@ -319,7 +292,7 @@ LMDBUnlock(Db *d, DbRef *r)
FILE *fakestream; FILE *fakestream;
Stream *stream; Stream *stream;
MDB_val key, val; MDB_val key, val;
bool ret; bool ret = true;
if (!d || !r) if (!d || !r)
{ {
@ -328,28 +301,30 @@ LMDBUnlock(Db *d, DbRef *r)
val.mv_data = NULL; val.mv_data = NULL;
val.mv_size = 0; 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); fakestream = open_memstream((char **) &val.mv_data, &val.mv_size);
stream = StreamFile(fakestream); stream = StreamFile(fakestream);
JsonEncode(r->json, stream, JSON_DEFAULT); JsonEncode(r->json, stream, JSON_DEFAULT);
StreamFlush(stream); StreamFlush(stream);
StreamClose(stream); StreamClose(stream);
ret = mdb_put(ref->transaction, ref->dbi, &key, &val, 0) == 0; ret = mdb_put(ref->transaction, db->dbi, &key, &val, 0) == 0;
mdb_txn_commit(ref->transaction); mdb_txn_commit(ref->transaction);
mdb_dbi_close(db->environ, ref->dbi); LMDBKillKey(key);
}
StringArrayFree(ref->base.name); StringArrayFree(ref->base.name);
JsonFree(ref->base.json); JsonFree(ref->base.json);
Free(ref); Free(ref);
LMDBKillKey(key);
if (val.mv_data) if (val.mv_data)
{ {
free(val.mv_data); free(val.mv_data);
} }
if (ret) if (ret && r->hint == DB_HINT_WRITE)
{ {
pthread_mutex_unlock(&d->lock); pthread_mutex_unlock(&d->lock);
} }
@ -362,7 +337,6 @@ LMDBCreate(Db *d, Array *k)
MDB_txn *transaction; MDB_txn *transaction;
LMDBRef *ret = NULL; LMDBRef *ret = NULL;
MDB_val key, empty_json; MDB_val key, empty_json;
MDB_dbi dbi;
int code; int code;
if (!d || !k) if (!d || !k)
{ {
@ -382,38 +356,25 @@ LMDBCreate(Db *d, Array *k)
); );
goto end; goto end;
} }
/* apparently you need to give it a dbi */
if ((code = mdb_dbi_open(transaction, "db", MDB_CREATE, &dbi)) != 0)
{
Log(LOG_ERR,
"%s: could not get transaction dbi: %s",
__func__, mdb_strerror(code)
);
goto end;
}
empty_json.mv_size = 2; empty_json.mv_size = 2;
empty_json.mv_data = "{}"; empty_json.mv_data = "{}";
/* put data in it */ /* put data in it */
code = mdb_put(transaction, dbi, &key, &empty_json, MDB_NOOVERWRITE); code = mdb_put(transaction, db->dbi, &key, &empty_json, MDB_NOOVERWRITE);
if (code == MDB_KEYEXIST) if (code == MDB_KEYEXIST)
{ {
mdb_dbi_close(db->environ, dbi);
mdb_txn_abort(transaction); mdb_txn_abort(transaction);
goto end; goto end;
} }
else if (code == MDB_MAP_FULL) else if (code == MDB_MAP_FULL)
{ {
Log(LOG_ERR, "%s: db is full", __func__); Log(LOG_ERR, "%s: db is full", __func__);
mdb_dbi_close(db->environ, dbi);
mdb_txn_abort(transaction); mdb_txn_abort(transaction);
goto end; goto end;
} }
else if (code != 0) else if (code != 0)
{ {
Log(LOG_ERR, "%s: mdb_put failure: %s", __func__, mdb_strerror(code)); Log(LOG_ERR, "%s: mdb_put failure: %s", __func__, mdb_strerror(code));
mdb_dbi_close(db->environ, dbi);
mdb_txn_abort(transaction); mdb_txn_abort(transaction);
goto end; goto end;
} }
@ -430,9 +391,9 @@ LMDBCreate(Db *d, Array *k)
StringArrayAppend(ret->base.name, ent); StringArrayAppend(ret->base.name, ent);
} }
} }
ret->base.hint = DB_HINT_WRITE;
ret->base.json = HashMapCreate(); ret->base.json = HashMapCreate();
ret->transaction = transaction; ret->transaction = transaction;
ret->dbi = dbi;
end: end:
if (!ret) if (!ret)
{ {
@ -454,7 +415,6 @@ LMDBList(Db *d, Array *k)
MDB_cursor *cursor; MDB_cursor *cursor;
MDB_cursor_op op = MDB_SET_RANGE; MDB_cursor_op op = MDB_SET_RANGE;
MDB_txn *txn; MDB_txn *txn;
MDB_dbi dbi;
int code; int code;
if (!d || !k) if (!d || !k)
@ -475,24 +435,13 @@ LMDBList(Db *d, Array *k)
); );
goto end; goto end;
} }
/* apparently you need to give it a dbi */ if ((code = mdb_cursor_open(txn, db->dbi, &cursor)) != 0)
if ((code = mdb_dbi_open(txn, "db", MDB_CREATE, &dbi)) != 0)
{
Log(LOG_ERR,
"%s: could not get transaction dbi: %s",
__func__, mdb_strerror(code)
);
mdb_txn_abort(txn);
goto end;
}
if ((code = mdb_cursor_open(txn, dbi, &cursor)) != 0)
{ {
Log(LOG_ERR, Log(LOG_ERR,
"%s: could not get cursor: %s", "%s: could not get cursor: %s",
__func__, mdb_strerror(code) __func__, mdb_strerror(code)
); );
mdb_txn_abort(txn); mdb_txn_abort(txn);
mdb_dbi_close(db->environ, dbi);
goto end; goto end;
} }
@ -520,7 +469,6 @@ LMDBList(Db *d, Array *k)
mdb_cursor_close(cursor); mdb_cursor_close(cursor);
mdb_txn_abort(txn); mdb_txn_abort(txn);
mdb_dbi_close(db->environ, dbi);
end: end:
LMDBKillKey(key); LMDBKillKey(key);
@ -538,6 +486,7 @@ LMDBClose(Db *d)
return; return;
} }
mdb_dbi_close(db->environ, db->dbi);
mdb_env_close(db->environ); mdb_env_close(db->environ);
} }
@ -546,6 +495,8 @@ DbOpenLMDB(char *dir, size_t size)
{ {
/* TODO */ /* TODO */
MDB_env *env = NULL; MDB_env *env = NULL;
MDB_txn *txn;
MDB_dbi dbi;
int code; int code;
LMDB *db; LMDB *db;
if (!dir || !size) if (!dir || !size)
@ -583,10 +534,35 @@ DbOpenLMDB(char *dir, size_t size)
return NULL; return NULL;
} }
/* Initialise a DBI */
{
if ((code = mdb_txn_begin(env, NULL, 0, &txn)) != 0)
{
Log(LOG_ERR,
"%s: could not begin transaction: %s",
__func__, mdb_strerror(code)
);
mdb_env_close(env);
return NULL;
}
if ((code = mdb_dbi_open(txn, "db", MDB_CREATE, &dbi)) != 0)
{
Log(LOG_ERR,
"%s: could not get transaction dbi: %s",
__func__, mdb_strerror(code)
);
mdb_txn_abort(txn);
mdb_env_close(env);
return NULL;
}
mdb_txn_commit(txn);
}
db = Malloc(sizeof(*db)); db = Malloc(sizeof(*db));
DbInit((Db *) db); DbInit((Db *) db);
db->environ = env; db->environ = env;
db->dbi = dbi;
db->base.lockFunc = LMDBLock; db->base.lockFunc = LMDBLock;
db->base.create = LMDBCreate; db->base.create = LMDBCreate;