#include #include "uthash.h" #include "store_mdb.h" /** * Number of DBs defined. */ #define N_DB 12 /** * Memory map size. */ #if (defined DEBUG || defined TESTING) #define DEFAULT_MAPSIZE 1<<24 // 16Mb (limit for Valgrind) #elif !(defined __LP64__ || defined __LLP64__) || \ defined _WIN32 && !defined _WIN64 #define DEFAULT_MAPSIZE 1<<31 // 2Gb (limit for 32-bit systems) #else #define DEFAULT_MAPSIZE 1UL<<40 // 1Tb #endif #define ENV_DIR_MODE 0750 #define ENV_FILE_MODE 0640 /* * Data types. */ typedef char DbLabel[8]; // TODO Most of these are no longer used. Clean up. typedef enum { LSSTORE_INIT = 1, // Is the store environment set up on disk? LSSTORE_OPEN = 3, // Is the environment open? Assumes init is set. LSSTORE_DIRTY_TXN = 4, // Main txn was opened in a subroutine. } StoreState; typedef enum { OP_ADD, OP_REMOVE, } StoreOp; typedef struct mdbstore_t { MDB_env * env; // Environment handle. MDB_txn * txn; // Current transaction. MDB_dbi dbi[N_DB]; // DB handles. Refer to DbIdx enum. LSUP_Buffer * default_ctx; // Default ctx as a serialized URI. StoreState state; // Store state. } MDBStore; /** @brief Iterator operation. * * Function executed for each iteration of a #MDBIterator. It assumes that a * result triple has already been found and is ready to be composed and * yielded. * * Upon call, the rc value of the iterator structure is set to the MDB_* rc * value for the next result. It is up to the caller to evaluate this value * and decide whether to call the function again. */ typedef void (*iter_op_fn_t)(LSUP_MDBIterator *it); /** @brief Triple iterator. */ typedef struct mdbstore_iter_t { MDBStore * store; // MDB store pointer. MDB_txn * txn; // MDB transaction. MDB_cursor * cur; // MDB cursor. MDB_cursor * ctx_cur; // MDB c:spo index cursor. MDB_val key, data; // Internal data handlers. LSUP_TripleKey spok; // Triple to be populated with match. LSUP_Key ck; // Ctx key to filter by. May be NULL_TRP. iter_op_fn_t iter_op_fn; // Function used to look up next match. const uint8_t * term_order; // Term order used in 1-2bound look-ups. LSUP_Key luk[3]; // 0÷3 lookup keys. size_t i; // Internal counter for paged lookups. int rc; // MDB_* return code for the next result. } MDBIterator; // Set of single keys. typedef struct key_set_t { LSUP_Key key; UT_hash_handle hh; } KeySet; // Set of triple keys. typedef struct triple_set_t { LSUP_TripleKey spok; UT_hash_handle hh; } TripleSet; // Map of context to triple set. typedef struct ctx_triple_map_t { LSUP_Key ck; TripleSet * spok; UT_hash_handle hh; } CtxTripleMap; /* * Static variables. */ /* * TODO At the moment up to 64-bit key / hash values are allowed. Later on, * 128-bit keys should be allowed by compile options, and that will no longer * be compatible with integer keys and data. When 128-bit keys are supported, * integer keys should remain available for code compiled with 64-bit keys. */ #define DUPSORT_MASK MDB_DUPSORT #define DUPFIXED_MASK MDB_DUPSORT | MDB_DUPFIXED /** * Main DBs. These are the master information containers. * * Data columns are: identifier prefix, DB label, flags. */ #define MAIN_TABLE \ ENTRY( T_ST, "t:st", 0 ) /* Key to ser. term */ \ ENTRY( SPO_C, "spo:c", DUPFIXED_MASK ) /* Triple to context */ \ ENTRY( C_, "c:", 0 ) /* Track empty ctx */ \ ENTRY( PFX_NS, "pfx:ns", 0 ) /* Prefix to NS */ \ /** * Lookup DBs. These are indices and may be destroyed and rebuilt. */ #define LOOKUP_TABLE \ ENTRY( S_PO, "s:po", DUPFIXED_MASK ) /* 1-bound lookup */ \ ENTRY( P_SO, "p:so", DUPFIXED_MASK ) /* 1-bound lookup */ \ ENTRY( O_SP, "o:sp", DUPFIXED_MASK ) /* 1-bound lookup */ \ ENTRY( PO_S, "po:s", DUPFIXED_MASK ) /* 2-bound lookup */ \ ENTRY( SO_P, "so:p", DUPFIXED_MASK ) /* 2-bound lookup */ \ ENTRY( SP_O, "sp:o", DUPFIXED_MASK ) /* 2-bound lookup */ \ ENTRY( C_SPO, "c:spo", DUPFIXED_MASK ) /* Context lookup */ \ ENTRY( NS_PFX, "ns:pfx", DUPSORT_MASK ) /* NS to prefix */ \ /** * DB labels. They are prefixed with DB_ */ #define ENTRY(a, b, c) static const DbLabel DB_##a = b; MAIN_TABLE LOOKUP_TABLE #undef ENTRY /* * Numeric index of each DB. Prefixed with IDX_ * * These index numbers are referred to in all the arrays defeined below. They * are independent from the LMDB dbi values which are considered opaque here. */ typedef enum { #define ENTRY(a, b, c) IDX_##a, MAIN_TABLE LOOKUP_TABLE #undef ENTRY } DBIdx; /** * DB labels. */ static const char *db_labels[N_DB] = { #define ENTRY(a, b, c) DB_##a, MAIN_TABLE LOOKUP_TABLE #undef ENTRY }; /* * DB flags. These are aligned with the dbi_labels index. */ static const unsigned int db_flags[N_DB] = { #define ENTRY(a, b, c) c, MAIN_TABLE LOOKUP_TABLE #undef ENTRY }; /* * 1-bound and 2-bound lookup indices. * * N.B. Only the first 6 (1-bound and 2-bound term lookup) are used. * The others are added just because they belong logically to the lookup table. */ static DBIdx lookup_indices[9] = { #define ENTRY(a, b, c) IDX_##a, LOOKUP_TABLE #undef ENTRY }; static const uint8_t lookup_ordering_1bound[3][3] = { {0, 1, 2}, // s:po {1, 0, 2}, // p:so {2, 0, 1}, // o:sp }; static const uint8_t lookup_ordering_2bound[3][3] = { {1, 2, 0}, // po:s {0, 2, 1}, // so:p {0, 1, 2}, // sp:o }; /* * Static prototypes. */ static int index_triple( LSUP_MDBStore *store, StoreOp op, LSUP_TripleKey spok, LSUP_Key ck); inline static LSUP_rc lookup_0bound (MDBIterator *it, size_t *ct); inline static LSUP_rc lookup_1bound (uint8_t idx0, MDBIterator *it, size_t *ct); inline static LSUP_rc lookup_2bound ( uint8_t idx0, uint8_t idx1, MDBIterator *it, size_t *ct); inline static LSUP_rc lookup_3bound(MDBIterator *it, size_t *ct); /** * API. */ LSUP_rc LSUP_mdbstore_setup (const char *path, bool clear) { int rc; // Set environment path. if (!path) return LSUP_ERROR; // TODO Verify that a writable directory exists or can be created. //struct stat path_stat; if (clear) rm_r (path); if (mkdir_p (path, ENV_DIR_MODE) != 0) return LSUP_IO_ERR; // Open a temporary environment and txn to create the DBs. MDB_env *env; mdb_env_create (&env); mdb_env_set_maxdbs (env, N_DB); RCCK (mdb_env_open (env, path, 0, ENV_FILE_MODE)); log_debug ("Environment opened at %s.", path); MDB_txn *txn; mdb_txn_begin (env, NULL, 0, &txn); for (int i = 0; i < N_DB; i++) { log_trace ("Creating DB %s", db_labels[i]); MDB_dbi dbi; rc = mdb_dbi_open (txn, db_labels[i], db_flags[i] | MDB_CREATE, &dbi); if (rc != MDB_SUCCESS) return rc; } mdb_txn_commit (txn); mdb_env_close (env); return rc; } MDBStore * LSUP_mdbstore_new (const char *path, const LSUP_Buffer *default_ctx) { int db_rc; LSUP_MDBStore *store; MALLOC_GUARD (store, NULL); db_rc = mdb_env_create (&store->env); log_trace ("create rc: %d", db_rc); store->default_ctx = ( default_ctx ? LSUP_buffer_new (default_ctx->size, default_ctx->addr) : NULL); // Set map size. size_t mapsize; char *env_mapsize = getenv ("LSUP_MDB_MAPSIZE"); if (env_mapsize == NULL) mapsize = DEFAULT_MAPSIZE; else sscanf (env_mapsize, "%lu", &mapsize); log_info ( "Setting environment map size at %s to %lu bytes.", path, mapsize); db_rc = mdb_env_set_mapsize (store->env, mapsize); db_rc = mdb_env_set_maxdbs (store->env, N_DB); if (UNLIKELY (db_rc != MDB_SUCCESS)) return NULL; db_rc = mdb_env_open (store->env, path, 0, ENV_FILE_MODE); if (UNLIKELY (db_rc != MDB_SUCCESS)) return NULL; // Assign DB handles to store->dbi. MDB_txn *txn; mdb_txn_begin (store->env, NULL, 0, &txn); for (int i = 0; i < N_DB; i++) { db_rc = mdb_dbi_open (txn, db_labels[i], db_flags[i], store->dbi + i); if (UNLIKELY (db_rc != MDB_SUCCESS)) { mdb_txn_abort (txn); return NULL; } } mdb_txn_commit (txn); store->state |= LSSTORE_OPEN; store->txn = NULL; return store; } void LSUP_mdbstore_free (LSUP_MDBStore *store) { if (store->state & LSSTORE_OPEN) { const char *path; mdb_env_get_path (store->env, &path); log_info ("Closing MDB env at %s.", path); mdb_env_close (store->env); } if (store->default_ctx) { LSUP_buffer_done (store->default_ctx); free (store->default_ctx); } free (store); } LSUP_rc LSUP_mdbstore_stat (LSUP_MDBStore *store, MDB_stat *stat) { if (!(store->state & LSSTORE_INIT)) return 0; MDB_txn *txn; mdb_txn_begin (store->env, NULL, MDB_RDONLY, &txn); if (mdb_stat (txn, store->dbi[IDX_SPO_C], stat) != MDB_SUCCESS) return LSUP_DB_ERR; mdb_txn_abort (txn); return LSUP_OK; } size_t LSUP_mdbstore_size (LSUP_MDBStore *store) { // Size is calculated outside of any pending write txn. MDB_stat stat; if (LSUP_mdbstore_stat (store, &stat) != LSUP_OK) return 0; return stat.ms_entries; } MDBIterator * LSUP_mdbstore_add_init (LSUP_MDBStore *store, const LSUP_Buffer *sc) { /* An iterator is used here. Some members are a bit misused but it does * its job without having to define a very similar struct. */ MDBIterator *it; MALLOC_GUARD (it, NULL); it->store = store; it->i = 0; // No other write transaction may be open. mdb_txn_begin (store->env, NULL, 0, &it->store->txn); // Take care of context first. // Serialize and hash. it->ck = NULL_KEY; if (store->default_ctx != NULL) { if (sc == NULL) sc = store->default_ctx; it->ck = LSUP_buffer_hash (sc); // Insert t:st for context. //log_debug ("Adding context: %s", sc); it->key.mv_data = &it->ck; it->key.mv_size = KLEN; it->data.mv_data = sc->addr; it->data.mv_size = sc->size; if (mdb_put( it->store->txn, it->store->dbi[IDX_T_ST], &it->key, &it->data, MDB_NOOVERWRITE) != MDB_SUCCESS) return NULL; } return it; } LSUP_rc LSUP_mdbstore_add_iter (MDBIterator *it, const LSUP_SerTriple *sspo) { int db_rc; LSUP_TripleKey spok = NULL_TRP; // Add triple. for (int i = 0; i < 3; i++) { LSUP_Buffer *st = LSUP_striple_pos (sspo, i); spok[i] = LSUP_buffer_hash (st); it->key.mv_data = spok + i; it->key.mv_size = KLEN; it->data.mv_data = st->addr; it->data.mv_size = st->size; db_rc = mdb_put( it->store->txn, it->store->dbi[IDX_T_ST], &it->key, &it->data, MDB_NOOVERWRITE); if (db_rc != MDB_SUCCESS && db_rc != MDB_KEYEXIST) { log_error ( "MDB error while inserting term: %s", LSUP_strerror(db_rc)); return LSUP_DB_ERR; } } log_trace ("Inserting spok: {%lx, %lx, %lx}", spok[0], spok[1], spok[2]); log_trace ("Into context: %lx", it->ck); // Insert spo:c. it->key.mv_data = spok; it->key.mv_size = TRP_KLEN; // In triple mode, data is empty (= NULL_KEY). it->data.mv_data = &it->ck; it->data.mv_size = it->ck == NULL_KEY ? 0 : KLEN; db_rc = mdb_put( it->store->txn, it->store->dbi[IDX_SPO_C], &it->key, &it->data, MDB_NODUPDATA); if (db_rc == MDB_KEYEXIST) return LSUP_NOACTION; if (db_rc != MDB_SUCCESS) { log_error ( "MDB error while inserting triple: %s", LSUP_strerror(db_rc)); return LSUP_DB_ERR; } // Index. LSUP_rc rc = index_triple (it->store, OP_ADD, spok, it->ck); if (rc == LSUP_OK) it->i++; return rc; } LSUP_rc LSUP_mdbstore_add_done (MDBIterator *it) { LSUP_rc rc = LSUP_OK; if (mdb_txn_commit (it->store->txn) != MDB_SUCCESS) { mdb_txn_abort (it->store->txn); rc = LSUP_DB_ERR; } it->store->txn = NULL; free (it); return rc; } void LSUP_mdbstore_add_abort (MDBIterator *it) { mdb_txn_abort (it->store->txn); it->store->txn = NULL; free (it); } LSUP_rc LSUP_mdbstore_add ( LSUP_MDBStore *store, const LSUP_Buffer *sc, const LSUP_SerTriple strp[], const size_t ct, size_t *inserted) { MDBIterator *it = LSUP_mdbstore_add_init (store, sc); if (UNLIKELY (!it)) return LSUP_DB_ERR; for (size_t i = 0; i < ct; i++) { LSUP_rc rc = LSUP_mdbstore_add_iter (it, strp + i); if (UNLIKELY (rc < 0)) { LSUP_mdbstore_add_abort (it); return rc; } } *inserted = it->i; return LSUP_mdbstore_add_done (it); } /* static LSUP_Key sterm_to_key ( LSUP_MDBStore *store, const LSUP_Buffer *sterm) { // TODO this will be replaced by a lookup when 128-bit hash is introduced. return LSUP_buffer_hash (sterm); } */ static LSUP_rc key_to_sterm (LSUP_MDBIterator *it, const LSUP_Key key, LSUP_Buffer *sterm) { LSUP_rc rc = LSUP_NORESULT; int db_rc; MDB_val key_v, data_v; key_v.mv_data = (void*)&key; key_v.mv_size = KLEN; db_rc = mdb_get (it->txn, it->store->dbi[IDX_T_ST], &key_v, &data_v); if (db_rc == MDB_SUCCESS) { free (sterm->addr); sterm->addr = data_v.mv_data; sterm->size = data_v.mv_size; rc = LSUP_OK; } else if (db_rc == MDB_NOTFOUND) { free (sterm->addr); sterm->addr = NULL; sterm->size = 0; } else rc = LSUP_ERROR; return rc; } MDBIterator * LSUP_mdbstore_lookup( LSUP_MDBStore *store, const LSUP_Buffer *ss, const LSUP_Buffer *sp, const LSUP_Buffer *so, const LSUP_Buffer *sc, size_t *ct) { LSUP_TripleKey spok = { LSUP_buffer_hash (ss), LSUP_buffer_hash (sp), LSUP_buffer_hash (so), }; LSUP_MDBIterator *it; CALLOC_GUARD (it, NULL); it->store = store; it->ck = store->default_ctx ? LSUP_buffer_hash (sc) : NULL_KEY; log_debug ("Lookup context: %lx", it->ck); if (ct) *ct = 0; uint8_t idx0, idx1; // Start RO transaction if not in a write txn already. if (it->store->txn) it->txn = it->store->txn; else { it->rc = mdb_txn_begin (it->store->env, NULL, MDB_RDONLY, &it->txn); if (it->rc != MDB_SUCCESS) { log_error ("Database error: %s", LSUP_strerror (it->rc)); return NULL; } } // Context index loop. if (UNLIKELY (mdb_cursor_open ( it->txn, it->store->dbi[IDX_SPO_C], &it->ctx_cur) != MDB_SUCCESS)) return NULL; /* * Lookup decision tree. */ // s p o (all terms bound) if (spok[0] != NULL_KEY && spok[1] != NULL_KEY && spok[2] != NULL_KEY) { it->luk[0] = spok[0]; it->luk[1] = spok[1]; it->luk[2] = spok[2]; RCNL (lookup_3bound (it, ct)); } else if (spok[0] != NULL_KEY) { it->luk[0] = spok[0]; idx0 = 0; // s p ? if (spok[1] != NULL_KEY) { it->luk[1] = spok[1]; idx1 = 1; RCNL (lookup_2bound (idx0, idx1, it, ct)); // s ? o } else if (spok[2] != NULL_KEY) { it->luk[1] = spok[2]; idx1 = 2; RCNL (lookup_2bound (idx0, idx1, it, ct)); // s ? ? } else RCNL (lookup_1bound (idx0, it, ct)); } else if (spok[1] != NULL_KEY) { it->luk[0] = spok[1]; idx0 = 1; // ? p o if (spok[2] != NULL_KEY) { it->luk[1] = spok[2]; idx1 = 2; RCNL (lookup_2bound (idx0, idx1, it, ct)); // ? p ? } else RCNL (lookup_1bound (idx0, it, ct)); // ? ? o } else if (spok[2] != NULL_KEY) { it->luk[0] = spok[2]; idx0 = 2; RCNL (lookup_1bound (idx0, it, ct)); // ? ? ? (all terms unbound) } else RCNL (lookup_0bound (it, ct)); return it; } /** @brief Get next iterator key. * * The ck pointer is filled with an array of contexts that the triple appears * in, if not NULL. */ inline static LSUP_rc mdbiter_next_key (LSUP_MDBIterator *it, KeySet **ck_p) { if (UNLIKELY (!it)) return LSUP_VALUE_ERR; // Only advance if the previous it->rc wasn't already at the end. if (it->rc == MDB_NOTFOUND) return LSUP_END; if (UNLIKELY (it->rc != MDB_SUCCESS)) { log_error ("Database error: %s", LSUP_strerror (it->rc)); return LSUP_DB_ERR; } LSUP_rc rc; /* Retrieve current value and advance cursor to the next result. * it->rc is set to the result of the next iteration. */ it->iter_op_fn (it); log_trace ( "Found spok: {%lx, %lx, %lx}", it->spok[0], it->spok[1], it->spok[2]); MDB_val key, data; int db_rc; key.mv_size = TRP_KLEN; data.mv_data = &it->ck; data.mv_size = KLEN; if (it->ck) { rc = LSUP_NORESULT; // Intermediary value, will never be returned. while (rc == LSUP_NORESULT) { //log_debug ("begin ctx loop."); // If ctx is specified, look if the matching triple is associated // with it. If not, move on to the next triple. // The loop normally exits when a triple with matching ctx is found // (LSUP_OK), if there are no more triples (LSUP_END), or if there // is an error (LSUP_DB_ERR). key.mv_data = it->spok; db_rc = mdb_cursor_get (it->ctx_cur, &key, &data, MDB_GET_BOTH); if (db_rc == MDB_SUCCESS) { rc = LSUP_OK; log_trace ("Triple found for context."); } else if (db_rc == MDB_NOTFOUND) { log_trace ("No triples found for context."); if (it->rc == MDB_NOTFOUND) rc = LSUP_END; else it->iter_op_fn (it); } else { log_error ("Database error: %s", LSUP_strerror (db_rc)); rc = LSUP_DB_ERR; } } } else rc = LSUP_OK; // Get all contexts for a triple if requested. Add up to previous // iterations if the same pointer is passed. if (ck_p) { key.mv_data = it->spok; db_rc = mdb_cursor_get (it->ctx_cur, &key, &data, MDB_SET_KEY); if (db_rc == MDB_SUCCESS) { do { KeySet *entry; HASH_FIND (hh, *ck_p, data.mv_data, KLEN, entry); if (!entry) { MALLOC_GUARD (entry, LSUP_MEM_ERR); entry->key = *(LSUP_Key *) data.mv_data; HASH_ADD (hh, *ck_p, key, KLEN, entry); } } while ( mdb_cursor_get (it->ctx_cur, &key, &data, MDB_NEXT_DUP) == MDB_SUCCESS); } } return rc; } LSUP_rc LSUP_mdbiter_next ( LSUP_MDBIterator *it, LSUP_SerTriple *sspo, LSUP_Buffer **ctx_p) { LSUP_rc rc; KeySet *ck = NULL; rc = (ctx_p) ? mdbiter_next_key (it, &ck) : mdbiter_next_key (it, NULL); if (rc == LSUP_OK) { if (sspo) { key_to_sterm (it, it->spok[0], sspo->s); key_to_sterm (it, it->spok[1], sspo->p); key_to_sterm (it, it->spok[2], sspo->o); // TODO error handling. } // One-shot contexts for current triple. if (ctx_p) { KeySet *ccur; LSUP_Buffer *ctx = malloc (HASH_COUNT (ck) * sizeof (*ctx)); size_t i = 0; for (ccur = ck; ccur != NULL; ccur = ccur->hh.next) key_to_sterm (it, ccur->key, ctx + (i++)); // TODO error handling. } } return rc; } size_t LSUP_mdbiter_cur (LSUP_MDBIterator *it) { return it->i; } void LSUP_mdbiter_free (MDBIterator *it) { if (it) { if (it->cur) mdb_cursor_close (it->cur); if (it->ctx_cur) mdb_cursor_close (it->ctx_cur); if (it->store->txn != it->txn) mdb_txn_abort (it->txn); free (it); } } LSUP_rc LSUP_mdbstore_remove( LSUP_MDBStore *store, const LSUP_Buffer *ss, const LSUP_Buffer *sp, const LSUP_Buffer *so, const LSUP_Buffer *sc, size_t *ct) { LSUP_rc rc = LSUP_NOACTION; LSUP_Key ck = NULL_KEY; if (store->default_ctx != NULL) { if (sc == NULL) sc = store->default_ctx; ck = LSUP_buffer_hash (sc); } MDB_txn *txn; mdb_txn_begin (store->env, NULL, 0, &txn); MDB_cursor *dcur, *icur; mdb_cursor_open (txn, store->dbi[IDX_SPO_C], &dcur); mdb_cursor_open (txn, store->dbi[IDX_C_SPO], &icur); MDB_val spok_v, ck_v; spok_v.mv_size = TRP_KLEN; ck_v.mv_size = KLEN; ck_v.mv_data = &ck; LSUP_MDBIterator *it = LSUP_mdbstore_lookup (store, ss, sp, so, sc, ct); if (UNLIKELY (!it)) return LSUP_DB_ERR; if (ct) log_debug ("Found %lu triples to remove.", *ct); while (mdbiter_next_key (it, NULL) == LSUP_OK) { spok_v.mv_data = it->spok; rc = mdb_cursor_get (dcur, &spok_v, &ck_v, MDB_GET_BOTH); if (rc == MDB_NOTFOUND) continue; if (UNLIKELY (rc != MDB_SUCCESS)) goto _remove_abort; log_trace ( "Removing {%lx, %lx, %lx}", it->spok[0], it->spok[1], it->spok[2]); // Delete spo:c entry. rc = mdb_cursor_del (dcur, 0); if (UNLIKELY (rc != MDB_SUCCESS)) goto _remove_abort; // Restore ck address after each delete. spok_v.mv_data = it->spok; ck_v.mv_data = &ck; // Delete c:spo entry. rc = mdb_cursor_get (icur, &ck_v, &spok_v, MDB_GET_BOTH); if (rc == MDB_NOTFOUND) continue; if (UNLIKELY (rc != MDB_SUCCESS)) goto _remove_abort; rc = mdb_cursor_del (icur, 0); if (UNLIKELY (rc != MDB_SUCCESS)) goto _remove_abort; spok_v.mv_data = it->spok; ck_v.mv_data = &ck; // If there are no more contexts associated with this triple, // remove from indices. rc = mdb_cursor_get (dcur, &spok_v, NULL, MDB_SET); if (rc == MDB_SUCCESS) continue; if (UNLIKELY (rc != MDB_NOTFOUND)) goto _remove_abort; index_triple (store, OP_REMOVE, it->spok, ck); } LSUP_mdbiter_free (it); if (UNLIKELY (mdb_txn_commit (txn) != MDB_SUCCESS)) { rc = LSUP_TXN_ERR; goto _remove_abort; } return rc; _remove_abort: mdb_txn_abort (txn); log_error ("Database error: %s", LSUP_strerror (rc)); return rc; } LSUP_Buffer ** LSUP_mdbstore_lookup_contexts ( LSUP_MDBStore *store, const LSUP_Buffer *ss, const LSUP_Buffer *sp, const LSUP_Buffer *so) { LSUP_MDBIterator *it = LSUP_mdbstore_lookup ( store, ss, sp, so, NULL, NULL); LSUP_rc rc = LSUP_NORESULT; KeySet *ckey, *ckeys = NULL, *tmp; while (rc != LSUP_END) rc = mdbiter_next_key (it, &ckeys); size_t i = 0; LSUP_Buffer **ctx_a = calloc (HASH_COUNT (ckeys) + 1, sizeof (*ctx_a)); if (UNLIKELY (!ctx_a)) return NULL; HASH_ITER (hh, ckeys, ckey, tmp) { ctx_a[i] = BUF_DUMMY; key_to_sterm (it, ckey->key, ctx_a[i++]); HASH_DEL (ckeys, ckey); free (ckey); } LSUP_mdbiter_free (it); return ctx_a; } LSUP_rc LSUP_mdbstore_nsm_get (LSUP_MDBStore *store, LSUP_NSMap **nsm_p) { LSUP_rc rc = LSUP_NORESULT; LSUP_NSMap *nsm = LSUP_nsmap_new(); if (UNLIKELY (!nsm)) return LSUP_MEM_ERR; *nsm_p = nsm; MDB_txn *txn; mdb_txn_begin (store->env, NULL, MDB_RDONLY, &txn); MDB_cursor *cur; if (mdb_cursor_open (txn, store->dbi[IDX_PFX_NS], &cur) != MDB_SUCCESS) { mdb_txn_abort (txn); return LSUP_DB_ERR; } MDB_val ns_v, pfx_v; if (mdb_cursor_get (cur, &ns_v, &pfx_v, MDB_FIRST) != MDB_SUCCESS) goto finally; do { ns_pfx pfx; char *ns = malloc (ns_v.mv_size); strncpy (pfx, pfx_v.mv_data, pfx_v.mv_size); strncpy (ns, ns_v.mv_data, ns_v.mv_size); LSUP_nsmap_add (nsm, pfx, ns); free (ns); } while (mdb_cursor_get ( cur, &ns_v, &pfx_v, MDB_NEXT_NODUP) == MDB_SUCCESS); finally: mdb_cursor_close (cur); mdb_txn_abort (txn); return rc; } LSUP_rc LSUP_mdbstore_nsm_store (LSUP_MDBStore *store, const LSUP_NSMap *nsm) { MDB_txn *txn; if (!store->txn) { RCCK (mdb_txn_begin (store->env, NULL, 0, &txn)); } else txn = store->txn; LSUP_rc rc = LSUP_NOACTION; int db_rc; MDB_cursor *dcur = NULL, *icur = NULL; if ( mdb_cursor_open (txn, store->dbi[IDX_PFX_NS], &dcur) != MDB_SUCCESS || mdb_cursor_open (txn, store->dbi[IDX_NS_PFX], &icur) != MDB_SUCCESS ) { rc = LSUP_DB_ERR; goto finally; } MDB_val pfx_v, ns_v; const char ***nsm_data = LSUP_nsmap_dump (nsm); for (size_t i = 0; nsm_data[i] != NULL; i++) { // At least 1 action. If not OK, it will change during the iteration. if (i == 0) rc = LSUP_OK; pfx_v.mv_data = (void *)nsm_data[i][0]; pfx_v.mv_size = strlen (nsm_data[i][0]) + 1; ns_v.mv_data = (void *)nsm_data[i][1]; ns_v.mv_size = strlen (nsm_data[i][1]) + 1; // If either ns or pfx exist, quit. if ( mdb_cursor_get (dcur, &pfx_v, &ns_v, MDB_SET) != MDB_NOTFOUND || mdb_cursor_get (icur, &ns_v, &pfx_v, MDB_SET) != MDB_NOTFOUND ) { rc = LSUP_CONFLICT; goto finally; } db_rc = mdb_cursor_put (dcur, &pfx_v, &ns_v, 0); if (db_rc != MDB_SUCCESS) { log_error ("DB error: %s", LSUP_strerror (db_rc)); rc = LSUP_DB_ERR; goto finally; } } finally: if (icur) mdb_cursor_close (icur); if (dcur) mdb_cursor_close (dcur); free (nsm_data); if (txn != store->txn) mdb_txn_commit (txn); return rc; } /* * * Static functions. * * */ /** @brief Index an added or removed triple. * * @param store[in] MDB store to index. * @param op[in] Store operation. One of OP_ADD or OP_REMOVE. * @param spok[in] Triple key to index. * @param ck[in] Context to index, may be NULL. */ static LSUP_rc index_triple( LSUP_MDBStore *store, StoreOp op, LSUP_TripleKey spok, LSUP_Key ck) { int db_rc; LSUP_rc rc = LSUP_NOACTION; MDB_val v1, v2; log_trace ("Indexing triple: %lx %lx %lx", spok[0], spok[1], spok[2]); // Index c:spo. if (op == OP_REMOVE) { log_trace ("Indexing op: REMOVE"); if (ck != NULL_KEY) { MDB_cursor *cur; v1.mv_data = &ck; v1.mv_size = KLEN; v2.mv_data = spok; v2.mv_size = TRP_KLEN; mdb_cursor_open (store->txn, store->dbi[IDX_C_SPO], &cur); if (mdb_cursor_get (cur, &v1, &v2, MDB_GET_BOTH) == MDB_SUCCESS) { db_rc = mdb_cursor_del (cur, 0); if (db_rc != MDB_SUCCESS) return LSUP_DB_ERR; rc = LSUP_OK; } mdb_cursor_close (cur); } } else if (op == OP_ADD) { log_trace ("Indexing op: ADD"); if (ck != NULL_KEY) { v1.mv_data = &ck; v1.mv_size = KLEN; v2.mv_data = spok; v2.mv_size = TRP_KLEN; db_rc = mdb_put( store->txn, store->dbi[IDX_C_SPO], &v1, &v2, MDB_NODUPDATA); if (db_rc != MDB_SUCCESS) return LSUP_DB_ERR; if (db_rc != MDB_KEYEXIST) rc = LSUP_OK; } } else return LSUP_VALUE_ERR; LSUP_DoubleKey dbl_keys[3] = { {spok[1], spok[2]}, // po {spok[0], spok[2]}, // so {spok[0], spok[1]}, // sp }; // Add terms to index. v1.mv_size = KLEN; v2.mv_size = DBL_KLEN; for (int i = 0; i < 3; i++) { MDB_dbi db1 = store->dbi[lookup_indices[i]]; // s:po, p:so, o:sp MDB_dbi db2 = store->dbi[lookup_indices[i + 3]]; // po:s, so:p, sp:o v1.mv_data = spok + i; v2.mv_data = dbl_keys[i]; if (op == OP_REMOVE) { MDB_cursor *cur1, *cur2; mdb_cursor_open( store->txn, store->dbi[lookup_indices[i]], &cur1); db_rc = mdb_cursor_get (cur1, &v1, &v2, MDB_GET_BOTH); if (db_rc == MDB_SUCCESS) mdb_cursor_del (cur1, 0); mdb_cursor_close (cur1); // Restore pointers invalidated after delete. v1.mv_data = spok + i; v2.mv_data = dbl_keys[i]; mdb_cursor_open( store->txn, store->dbi[lookup_indices[i + 3]], &cur2); db_rc = mdb_cursor_get (cur2, &v2, &v1, MDB_GET_BOTH); if (db_rc == MDB_SUCCESS) mdb_cursor_del (cur2, 0); // TODO error handling. rc = LSUP_OK; mdb_cursor_close (cur2); } else { // OP_ADD is guaranteed. // 1-bound index. log_trace ("Indexing in %s: ", db_labels[lookup_indices[i]]); log_trace ( "%lx: %lx %lx", *(size_t*)(v1.mv_data), *(size_t*)(v2.mv_data), *(size_t*)(v2.mv_data) + 1); db_rc = mdb_put (store->txn, db1, &v1, &v2, MDB_NODUPDATA); if (db_rc == MDB_SUCCESS) rc = LSUP_OK; else if (db_rc != MDB_KEYEXIST) return LSUP_DB_ERR; // 2-bound index. log_trace ("Indexing in %s: ", db_labels[lookup_indices[i + 3]]); log_trace ( "%lx %lx: %lx", *(size_t*)(v2.mv_data), *(size_t*)(v2.mv_data) + 1, *(size_t*)(v1.mv_data)); db_rc = mdb_put (store->txn, db2, &v2, &v1, MDB_NODUPDATA); if (db_rc == MDB_SUCCESS) rc = LSUP_OK; else if (db_rc != MDB_KEYEXIST) return LSUP_DB_ERR; } } return rc; } /* * * Term-specific iterators. * * */ /** @brief Advance 0-bound iterator. * * Cursor: spo:c */ inline static void it_next_0bound (MDBIterator *it) { memcpy (it->spok, it->key.mv_data, sizeof (LSUP_TripleKey)); it->rc = mdb_cursor_get (it->cur, &it->key, &it->data, MDB_NEXT); } /** @brief Advance 1-bound iterator. * * Uses paged data in a nested loop. * * Cursor: s:po, p:so, or o:sp. */ inline static void it_next_1bound (MDBIterator *it) { LSUP_DoubleKey *lu_dset = it->data.mv_data; it->spok[it->term_order[0]] = it->luk[0]; it->spok[it->term_order[1]] = lu_dset[it->i][0]; it->spok[it->term_order[2]] = lu_dset[it->i][1]; log_trace ( "Composed triple: {%lx %lx %lx}", it->spok[0], it->spok[1], it->spok[2]); // Ensure next block within the same page is not beyond the last. if (it->i < it->data.mv_size / DBL_KLEN - 1) { it->i ++; //log_debug ("Increasing page cursor to %lu.", it->i); //log_debug ("it->rc: %d", it->rc); } else { // If the last block in the page is being yielded, // move cursor to beginning of next page. it->i = 0; //log_debug ("Reset page cursor to %lu.", it->i); it->rc = mdb_cursor_get ( it->cur, &it->key, &it->data, MDB_NEXT_MULTIPLE); } } /** @brief Advance 2-bound iterator. * * Uses paged data in a nested loop. * * Cursor: po:s, so:p, or sp:o. */ inline static void it_next_2bound (MDBIterator *it) { LSUP_Key *lu_dset = it->data.mv_data; it->spok[it->term_order[0]] = it->luk[0]; it->spok[it->term_order[1]] = it->luk[1]; it->spok[it->term_order[2]] = lu_dset[it->i]; // Ensure next block within the same page is not beyond the last. if (it->i < it->data.mv_size / KLEN - 1) it->i ++; else { // If the last block in the page is being yielded, // move cursor to beginning of next page. it->i = 0; it->rc = mdb_cursor_get (it->cur, &it->key, &it->data, MDB_NEXT_MULTIPLE); } } /** @brief Advance 3-bound iterator. * * This is a special case of 0÷1 results; either there was one matching triple, * which was already set in the first result, or there was none, i.e. it->rc is * already MDB_NOTFOUND and this function will not be called. */ inline static void it_next_3bound (MDBIterator *it) { it->rc = MDB_NOTFOUND; } /* * * Term-specific lookups. * * */ inline static LSUP_rc lookup_0bound (MDBIterator *it, size_t *ct) { log_debug ("Looking up 0 bound terms."); if (ct) { if (it->ck != NULL_KEY) { // Look up by given context. it->rc = mdb_cursor_open ( it->txn, it->store->dbi[IDX_C_SPO], &it->cur); it->key.mv_data = &it->ck; it->key.mv_size = KLEN; it->rc = mdb_cursor_get (it->cur, &it->key, &it->data, MDB_SET); if (it->rc == MDB_SUCCESS) mdb_cursor_count (it->cur, ct); mdb_cursor_close (it->cur); it->cur = NULL; } else { // Look up all contexts. MDB_stat stat; mdb_stat (it->txn, it->store->dbi[IDX_S_PO], &stat); *ct = stat.ms_entries; } log_debug ("Found %lu keys.", *ct); } it->rc = mdb_cursor_open (it->txn, it->store->dbi[IDX_SPO_C], &it->cur); if (it->rc != MDB_SUCCESS) { log_error ("Database error: %s", LSUP_strerror (it->rc)); return LSUP_DB_ERR; } it->rc = mdb_cursor_get (it->cur, &it->key, &it->data, MDB_FIRST); /* mdb_cursor_close (it->cur); it->cur = NULL; */ it->iter_op_fn = it_next_0bound; if (it->rc != MDB_SUCCESS && it->rc != MDB_NOTFOUND) { log_error ("Database error: %s", LSUP_strerror (it->rc)); return LSUP_DB_ERR; } return LSUP_OK; } inline static LSUP_rc lookup_1bound (uint8_t idx0, MDBIterator *it, size_t *ct) { it->term_order = (const uint8_t*)lookup_ordering_1bound[idx0]; log_debug ("Looking up 1 bound term: %lx", it->luk[0]); mdb_cursor_open (it->txn, it->store->dbi[lookup_indices[idx0]], &it->cur); it->key.mv_data = it->luk; it->key.mv_size = KLEN; if (ct) { // If a context is specified, the only way to count triples matching // the context is to loop over them. if (it->ck != NULL_KEY) { log_debug ("Counting in context: %lx", it->ck); MDBIterator *ct_it = NULL; MALLOC_GUARD (ct_it, LSUP_MEM_ERR); ct_it->luk[0] = it->luk[0]; ct_it->ck = it->ck; ct_it->store = it->store; ct_it->txn = it->txn; ct_it->key = it->key; ct_it->data = it->data; ct_it->i = 0; ct_it->ctx_cur = it->ctx_cur; LSUP_rc rc = lookup_1bound (idx0, ct_it, NULL); if (rc < 0) return rc; while (LSUP_mdbiter_next (ct_it, NULL, NULL) != LSUP_END) { (*ct)++; log_trace ("Counter increased to %lu.", *ct); } // Free the counter iterator without freeing the shared txn. if (ct_it->cur) mdb_cursor_close (ct_it->cur); free (ct_it); } else { it->rc = mdb_cursor_get (it->cur, &it->key, &it->data, MDB_SET); if (it->rc == MDB_SUCCESS) mdb_cursor_count (it->cur, ct); } } it->i = 0; it->iter_op_fn = it_next_1bound; it->rc = mdb_cursor_get (it->cur, &it->key, &it->data, MDB_SET); if (it->rc == MDB_SUCCESS) it->rc = mdb_cursor_get (it->cur, &it->key, &it->data, MDB_GET_MULTIPLE); if (it->rc != MDB_SUCCESS && it->rc != MDB_NOTFOUND) { log_error ("Database error: %s", LSUP_strerror (it->rc)); return LSUP_DB_ERR; } return LSUP_OK; } inline static LSUP_rc lookup_2bound(uint8_t idx0, uint8_t idx1, MDBIterator *it, size_t *ct) { uint8_t luk1_offset, luk2_offset; MDB_dbi dbi = 0; // Establish lookup ordering with some awkward offset math. for (int i = 0; i < 3; i++) { if ( ( idx0 == lookup_ordering_2bound[i][0] && idx1 == lookup_ordering_2bound[i][1] ) || ( idx0 == lookup_ordering_2bound[i][1] && idx1 == lookup_ordering_2bound[i][0] ) ) { it->term_order = (const uint8_t*)lookup_ordering_2bound[i]; if (it->term_order[0] == idx0) { luk1_offset = 0; luk2_offset = 1; } else { luk1_offset = 1; luk2_offset = 0; } dbi = it->store->dbi[lookup_indices[i + 3]]; log_debug ( "Looking up 2 bound in %s", db_labels[lookup_indices[i + 3]]); break; } } if (dbi == 0) { log_error ( "Values %d and %d not found in lookup keys.", idx0, idx1); return LSUP_VALUE_ERR; } // Compose term keys in lookup key. LSUP_DoubleKey luk; luk[luk1_offset] = it->luk[0]; luk[luk2_offset] = it->luk[1]; it->key.mv_data = luk; it->key.mv_size = DBL_KLEN; mdb_cursor_open (it->txn, dbi, &it->cur); it->rc = mdb_cursor_get (it->cur, &it->key, &it->data, MDB_SET); if (ct) { // If a context is specified, the only way to count triples matching // the context is to loop over them. if (it->ck != NULL_KEY) { MDBIterator *ct_it; MALLOC_GUARD (ct_it, LSUP_MEM_ERR); ct_it->luk[0] = it->luk[0]; ct_it->luk[1] = it->luk[1]; ct_it->ck = it->ck; ct_it->store = it->store; ct_it->txn = it->txn; ct_it->ctx_cur = it->ctx_cur; lookup_2bound (idx0, idx1, ct_it, NULL); while (LSUP_mdbiter_next (ct_it, NULL, NULL) != LSUP_END) { ct[0] ++; } // Free the counter iterator without freeing the shared txn. if (ct_it->cur) mdb_cursor_close (ct_it->cur); free (ct_it); } else { it->rc = mdb_cursor_get (it->cur, &it->key, &it->data, MDB_SET); if (it->rc == MDB_SUCCESS) mdb_cursor_count (it->cur, ct); } } it->i = 0; it->iter_op_fn = it_next_2bound; it->rc = mdb_cursor_get (it->cur, &it->key, &it->data, MDB_SET); if (it->rc == MDB_SUCCESS) it->rc = mdb_cursor_get (it->cur, &it->key, &it->data, MDB_GET_MULTIPLE); if (it->rc != MDB_SUCCESS && it->rc != MDB_NOTFOUND) { log_error ("Database error: %s", LSUP_strerror (it->rc)); return LSUP_DB_ERR; } return LSUP_OK; } inline static LSUP_rc lookup_3bound (MDBIterator *it, size_t *ct) { log_debug ( "Looking up 3 bound: {%lx, %lx, %lx}", it->luk[0], it->luk[1], it->luk[2]); it->key.mv_data = it->luk; if (it->ck != NULL_KEY) { it->rc = mdb_cursor_open ( it->txn, it->store->dbi[IDX_SPO_C], &it->cur); it->key.mv_size = TRP_KLEN; it->data.mv_data = &it->ck; it->data.mv_size = KLEN; } else { it->rc = mdb_cursor_open (it->txn, it->store->dbi[IDX_S_PO], &it->cur); it->key.mv_size = KLEN; it->data.mv_data = it->luk + 1; it->data.mv_size = DBL_KLEN; } it->rc = mdb_cursor_get (it->cur, &it->key, &it->data, MDB_GET_BOTH); if (it->rc != MDB_SUCCESS && it->rc != MDB_NOTFOUND) { log_error ("Database error: %s", LSUP_strerror (it->rc)); return LSUP_DB_ERR; } mdb_cursor_close (it->cur); it->cur = NULL; if (ct && it->rc == MDB_SUCCESS) *ct = 1; it->iter_op_fn = it_next_3bound; memcpy (it->spok, it->luk, sizeof (LSUP_TripleKey)); return LSUP_OK; }