lsup_rdf/src/graph.c

#include "graph.h"

// Initial size of lookup graph. It will double each time capacity is reached.
#define LOOKUP_GR_INIT_SIZE 64

// Expand hash table memory by this factor to keep a good load factor.
#define PREALLOC_FACTOR 1.4

// Assume VERY coarsly that the number of unique terms will be in general
// 1.7 times the number of triples. This is conservative to maintain load
// factor low.
#define IDX_SIZE_RATIO 1.7

// RAMdisk path for MDB volatile store.
#define MDB_RAMDISK_PATH TMPDIR "/lsup_mem_graph"

typedef enum KSetFlag {
    LSUP_KS_NONE        = 0,
    LSUP_KS_CHECK_CAP   = 1 << 0,
    LSUP_KS_CHECK_DUP   = 1 << 1,
} KSetFlag;


/**
 * Static handles.
 */
static const char *default_ctx_label = "urn:lsup:default";
static LSUP_Buffer *default_ctx = NULL;
static LSUP_MDBStore *default_store = NULL, *default_tmp_store = NULL;


typedef struct Graph {
    LSUP_store_type         store_type;     // In-memory or MDB-backed
    LSUP_Term               *uri;                 // Graph "name" (URI)
    union {
        LSUP_HTStore *      ht_store;
        LSUP_MDBStore *     mdb_store;
    };
} Graph;


typedef struct GraphIterator {
    const Graph *           graph;      // Parent graph.
    union {                             // Internal store iterator.
        LSUP_HTIterator *   ht_iter;
        LSUP_MDBIterator *  mdb_iter;
    };
    size_t                  ct;         // Total matches.
} GraphIterator;


/**
 * Extern inline functions.
 */
size_t LSUP_graph_size (const LSUP_Graph *gr);


/* * * Static prototypes. * * */

static inline LSUP_rc mdbstore_init();


/* * * Post-lookup callback prototypes * * */

int match_add_fn(
        LSUP_Graph *src, LSUP_Graph *dest, const LSUP_TripleKey *spok,
        void *ctx);


int match_rm_fn(
        LSUP_Graph *src, LSUP_Graph *dest, const LSUP_TripleKey *spok,
        void *ctx);

static LSUP_rc
graph_iter_next_buffer (GraphIterator *it, LSUP_SerTriple *sspo);


/* Atexit functions. */
void ctx_cleanup()
{
/*@ @brief Close LMDB environment.
 *
 * Run at exit.
 */
    LSUP_mdbstore_free (default_store);
    LSUP_mdbstore_free (default_tmp_store);
    LSUP_buffer_free (default_ctx);
}


static inline bool is_null_trp (const LSUP_TripleKey *trp)
{
    return (
            *trp[0] == NULL_KEY &&
            *trp[1] == NULL_KEY &&
            *trp[2] == NULL_KEY);
}

#define ENTRY(a, b) (be) == (LSUP_STORE_##a) ||
static inline bool
check_backend (LSUP_store_type be)
{ return (BACKEND_TBL false); }
#undef ENTRY


/* * * GRAPH * * */

Graph *
LSUP_graph_new (const LSUP_store_type store_type)
{
    if (!check_backend (store_type)) return NULL;

    LSUP_Graph *gr = malloc (sizeof (LSUP_Graph));
    if (UNLIKELY (!gr)) return NULL;

    gr->uri = LSUP_uri_new (NULL);
    gr->store_type = store_type;

    if (UNLIKELY (mdbstore_init() != LSUP_OK)) return NULL;

    if (gr->store_type == LSUP_STORE_MEM) {
        gr->ht_store = LSUP_htstore_new();
        if (UNLIKELY (!gr->ht_store)) return NULL;

    } else if (gr->store_type == LSUP_STORE_MDB) {
        gr->mdb_store = default_store;
        if (UNLIKELY (!gr->mdb_store)) return NULL;

    } else { // LSUP_STORE_MDB_TMP
        gr->mdb_store = default_tmp_store;
    }

    return gr;
}


/**
 * Copy triples from a source graph into a destination one.
 *
 * The destination graph is not initialized here, so the copy is cumulative.
 */
static LSUP_rc
graph_copy_contents (const LSUP_Graph *src, LSUP_Graph *dest)
{
    LSUP_rc rc = LSUP_NOACTION;
    const LSUP_Triple trp = {NULL, NULL, NULL};

    GraphIterator *it = LSUP_graph_lookup (src, &trp, NULL);

    LSUP_SerTriple sspo;

    while (graph_iter_next_buffer (it, &sspo) != LSUP_END) {
        TRACE ("Inserting triple #%lu\n", LSUP_graph_iter_cur (it));
        LSUP_rc add_rc = LSUP_graph_add (dest, NULL, 0, &sspo, 1, NULL);
        if (LIKELY (add_rc == LSUP_OK)) rc = LSUP_OK;
        else if (add_rc < 0) return add_rc;
    }

    return rc;
}


LSUP_Graph *
LSUP_graph_copy (const Graph *src)
{
    LSUP_Graph *dest = LSUP_graph_new (src->store_type);
    if (UNLIKELY (!dest)) return NULL;

    LSUP_rc rc = graph_copy_contents (src, dest);
    if (UNLIKELY (rc != LSUP_OK)) return NULL;

    return dest;
}


// TODO support boolean ops between any types of graphs.
Graph *
LSUP_graph_bool_op(
        const LSUP_bool_op op, const Graph *gr1, const Graph *gr2)
{
    if (UNLIKELY (gr1->store_type != LSUP_STORE_MEM)) {
        fprintf(
                stderr,
                "First operand %s is not an in-memory graph. "
                "Cannot perform boolean operation.",
                gr1->uri->data);
        return NULL;
    }
    if (UNLIKELY (gr2->store_type != LSUP_STORE_MEM)) {
        fprintf(
                stderr,
                "Second operand %s is not an in-memory graph. "
                "Cannot perform boolean operation.",
                gr2->uri->data);
        return NULL;
    }

    LSUP_Graph *res = LSUP_graph_new (LSUP_STORE_MEM);
    res->ht_store = LSUP_htstore_bool_op (op, gr1->ht_store, gr2->ht_store);

    return res;
}


void
LSUP_graph_free (LSUP_Graph *gr)
{
    if (LIKELY (gr != NULL)) {
        LSUP_term_free (gr->uri);

        if (gr->store_type == LSUP_STORE_MEM)
            LSUP_htstore_free (gr->ht_store);
        /*
        // NOTE This is not required bacause MDB store is only one for now,
        // cleared at exit.
        else
            LSUP_mdbstore_free (gr->mdb_store);
        */

        free (gr);
    }
}


LSUP_Term *
LSUP_graph_uri (const LSUP_Graph *gr) { return gr->uri; }


LSUP_rc
LSUP_graph_set_uri (LSUP_Graph *gr, const char *uri)
{ return LSUP_uri_init (gr->uri, uri); }


size_t
LSUP_graph_size (const Graph *gr)
{
    TRACE ("Store type: %d\n", gr->store_type);
    if (gr->store_type == LSUP_STORE_MEM)
        LSUP_htstore_size (gr->ht_store);

    return LSUP_mdbstore_size (gr->mdb_store);
}


bool
LSUP_graph_equals (const Graph *gr1, const Graph *gr2)
{
    LSUP_Graph *res = LSUP_graph_bool_op (LSUP_BOOL_XOR, gr1, gr2);

    return (LSUP_graph_size (res) == 0);
}


GraphIterator *
LSUP_graph_add_init (LSUP_Graph *gr)
{
    GraphIterator *it = malloc (sizeof (*it));
    if (UNLIKELY (!it)) return NULL;

    if (gr->store_type == LSUP_STORE_MEM) {
        it->ht_iter = LSUP_htstore_add_init (gr->ht_store);
    } else {
        LSUP_Buffer *sc = LSUP_buffer_new_from_term (gr->uri);
        it->mdb_iter = LSUP_mdbstore_add_init (gr->mdb_store, sc);
        LSUP_buffer_free (sc);
    }

    it->graph = gr;

    return it;
}


LSUP_rc
LSUP_graph_add_iter (LSUP_GraphIterator *it, const LSUP_SerTriple *sspo)
{
    if (it->graph->store_type == LSUP_STORE_MEM) {
        return LSUP_htstore_add_iter (it->ht_iter, sspo);
    } else {
        return LSUP_mdbstore_add_iter (it->mdb_iter, sspo);
    }
}


void
LSUP_graph_add_done (LSUP_GraphIterator *it)
{
    if (it->graph->store_type == LSUP_STORE_MEM) {
        LSUP_htstore_add_done (it->ht_iter);
    } else {
        LSUP_mdbstore_add_done (it->mdb_iter);
    }
}


LSUP_rc
LSUP_graph_add(
        LSUP_Graph *gr,
        const LSUP_Triple trp[], size_t trp_ct,
        const LSUP_SerTriple strp[], size_t strp_ct, size_t *inserted)
{
    /*
     * NOTE It is possible to pass both sets of RDF triples and buffer triples.
     */

    LSUP_rc rc = LSUP_NOACTION;

    // Initialize iterator.
    LSUP_GraphIterator *it = LSUP_graph_add_init (gr);

    // Serialize and insert RDF triples.
    LSUP_SerTriple *sspo = STRP_DUMMY;
    for (size_t i = 0; i < trp_ct; i++) {
        TRACE ("Inserting triple #%lu\n", i);

        LSUP_triple_serialize (trp + i, sspo);
        LSUP_rc db_rc = LSUP_graph_add_iter (it, sspo);

        if (db_rc == LSUP_OK) rc = LSUP_OK;
        if (UNLIKELY (db_rc < 0)) return db_rc;
    }
    LSUP_striple_free (sspo);

    // Insert serialized triples.
    for (size_t i = 0; i < strp_ct; i++) {
        TRACE ("Inserting serialized triple #%lu\n", i);
        LSUP_rc db_rc = LSUP_graph_add_iter (it, strp + i);

        if (db_rc == LSUP_OK) rc = LSUP_OK;
        if (UNLIKELY (db_rc < 0)) return db_rc;
    }

    LSUP_graph_add_done (it);

    if (inserted) {
        if (gr->store_type == LSUP_STORE_MEM) {
            *inserted = LSUP_htiter_cur (it->ht_iter);
        } else {
            *inserted = LSUP_mdbiter_cur (it->mdb_iter);
        }
    }

    return rc;
}


LSUP_rc
LSUP_graph_remove (Graph *gr, const LSUP_Triple *spo, size_t *ct)
{
    LSUP_rc rc;

    LSUP_SerTriple *sspo = LSUP_striple_new_from_triple (spo);
    LSUP_Buffer *sc = LSUP_buffer_new_from_term (gr->uri);

    if (gr->store_type == LSUP_STORE_MEM)
        rc = 0;// TODO uncomment LSUP_htstore_remove (gr->ht_store, &sspo, ct);
    else
        rc = LSUP_mdbstore_remove (gr->mdb_store, sspo, sc, ct);

    LSUP_striple_free (sspo);
    LSUP_buffer_free (sc);

    return rc;
}


GraphIterator *
LSUP_graph_lookup (const Graph *gr, const LSUP_Triple *spo, size_t *ct)
{
    GraphIterator *it = malloc (sizeof (GraphIterator));
    if (UNLIKELY (!it)) return NULL;

    it->graph = gr;

    LSUP_SerTriple *sspo = LSUP_striple_new_from_triple (spo);
    LSUP_Buffer *sc = LSUP_buffer_new_from_term (gr->uri);

    if (gr->store_type == LSUP_STORE_MEM) {
        // TODO uncomment it->ht_iter = LSUP_htstore_lookup (gr->ht_store, sspo, &it->ct);

    } else {
        it->mdb_iter = LSUP_mdbstore_lookup (gr->mdb_store, sspo, sc, ct);
    }

    LSUP_striple_free (sspo);
    LSUP_buffer_free (sc);

    return it;
}


/** @brief Advance iterator and return serialized triple.
 *
 * This is an internal function to pass raw buffers between higher-level
 * functions without serializing and deserializing triples.
 */
inline static LSUP_rc
graph_iter_next_buffer (GraphIterator *it, LSUP_SerTriple *sspo)
{
    LSUP_rc rc;

    if (it->graph->store_type == LSUP_STORE_MEM)
        rc = 0;// TODO uncomment LSUP_htiter_next (it->ht_iter, sspo);
    else rc = LSUP_mdbiter_next (it->mdb_iter, sspo);

    return rc;
}


LSUP_rc
LSUP_graph_iter_next (GraphIterator *it, LSUP_Triple *spo)
{
    LSUP_SerTriple *sspo = NULL;
    if (spo) sspo = STRP_DUMMY;

    LSUP_rc rc = graph_iter_next_buffer (it, sspo);

    if (rc == LSUP_OK) {
        if (spo) {
            LSUP_term_deserialize (sspo->s, spo->s);
            LSUP_term_deserialize (sspo->p, spo->p);
            LSUP_term_deserialize (sspo->o, spo->o);
        }
    }

    // Addresses of triple buffers are owned by the DB. Do not free.
    if (sspo) {
        free (sspo->s);
        free (sspo->p);
        free (sspo->o);
        free (sspo);
    }

    return rc;
}


void
LSUP_graph_iter_free (GraphIterator *it)
{
    if (it->graph->store_type == LSUP_STORE_MEM)
        NULL;// TODO uncomment LSUP_htiter_free (it->ht_iter);
    else
        LSUP_mdbiter_free (it->mdb_iter);

    free (it);
}


size_t
LSUP_graph_iter_cur (GraphIterator *it)
{ return LSUP_mdbiter_cur (it->mdb_iter); }


bool
LSUP_graph_contains (const LSUP_Graph *gr, const LSUP_Triple *spo)
{
    GraphIterator *it = LSUP_graph_lookup (gr, spo, NULL);
    bool rc = LSUP_graph_iter_next (it, NULL) != LSUP_NORESULT;

    LSUP_graph_iter_free (it);

    return rc;
}


/* * * Static functions * * */

/** @brief Initialize default context and MDB environments.
 *
 * This is done only once per process.
 *
 * The ramdisk store persists after the application is closed, but will be
 * wiped clean the next time this function is called.
 */
static inline LSUP_rc
mdbstore_init()
{
    char *path;

    // RAM disk store.
    if (UNLIKELY (!default_tmp_store)) {
        printf ("Initializing RAM disk back end.\n");
        path = MDB_RAMDISK_PATH;
        if (LSUP_mdbstore_setup (path, true) != LSUP_OK) return LSUP_DB_ERR;
        default_tmp_store = LSUP_mdbstore_new (path, default_ctx);
        if (UNLIKELY (!default_tmp_store)) return LSUP_DB_ERR;
    }

    // Permanent store.
    if (UNLIKELY (!default_store)) {
        printf ("Initializing persistent back end.\n");
        // NOTE This method will only allow one persistent disk back end per
        // application. TODO maybe later allow multiple backends if useful.
        path = getenv ("LSUP_MDB_STORE_PATH");
        if (!path) {
            path = DEFAULT_ENV_PATH;
            fprintf(
                stderr,
                "WARNING: `LSUP_MDB_STORE_PATH' environment variable is not "
                "set. The default location %s will be used as the graph "
                "store.\n", path
            );
        }
        if (LSUP_mdbstore_setup (path, false) != LSUP_OK) return LSUP_DB_ERR;
        default_store = LSUP_mdbstore_new (path, default_ctx);
        if (UNLIKELY (!default_store)) return LSUP_DB_ERR;
    }

    if (UNLIKELY (!default_ctx)) {
        LSUP_Term *default_ctx_uri = LSUP_uri_new (default_ctx_label);
        default_ctx = LSUP_buffer_new_from_term (default_ctx_uri);
        LSUP_term_free (default_ctx_uri);
        atexit (ctx_cleanup);
    }

    return LSUP_OK;
}