Sorting a memory leak in datastore/journalstore
This commit is contained in:
parent
bf7ba9049c
commit
cb1ea3ceff
12 changed files with 200 additions and 180 deletions
|
@ -152,6 +152,9 @@ int ipfs_node_free(struct IpfsNode* node) {
|
|||
if (node->mode == MODE_ONLINE) {
|
||||
ipfs_routing_online_free(node->routing);
|
||||
}
|
||||
if (node->mode == MODE_OFFLINE) {
|
||||
ipfs_routing_offline_free(node->routing);
|
||||
}
|
||||
if (node->blockstore != NULL) {
|
||||
ipfs_blockstore_free(node->blockstore);
|
||||
}
|
||||
|
|
|
@ -50,7 +50,7 @@ int lmdb_journalstore_cursor_close(struct lmdb_trans_cursor *cursor);
|
|||
|
||||
int journal_record_free(struct JournalRecord* rec);
|
||||
|
||||
int lmdb_journalstore_journal_add(MDB_txn* mdb_txn, struct JournalRecord* record);
|
||||
int lmdb_journalstore_journal_add(struct lmdb_trans_cursor *journalstore_cursor, struct JournalRecord *journalstore_record);
|
||||
|
||||
/***
|
||||
* Attempt to get a specific record identified by its timestamp and bytes
|
||||
|
@ -61,3 +61,11 @@ int lmdb_journalstore_journal_add(MDB_txn* mdb_txn, struct JournalRecord* record
|
|||
*/
|
||||
int lmdb_journalstore_get_record(void* handle, struct lmdb_trans_cursor *journalstore_cursor, struct JournalRecord **journalstore_record);
|
||||
|
||||
/***
|
||||
* Convert the JournalRec struct into a lmdb key and lmdb value
|
||||
* @param journal_record the record to convert
|
||||
* @param db_key where to store the key information
|
||||
* @param db_value where to store the value information
|
||||
*/
|
||||
int lmdb_journalstore_build_key_value_pair(const struct JournalRecord* journal_record, struct MDB_val* db_key, struct MDB_val *db_value);
|
||||
|
||||
|
|
|
@ -3,6 +3,8 @@
|
|||
#include "lmdb.h"
|
||||
|
||||
struct lmdb_trans_cursor {
|
||||
MDB_env* environment;
|
||||
MDB_txn* parent_transaction;
|
||||
MDB_txn* transaction;
|
||||
MDB_dbi* database;
|
||||
MDB_cursor* cursor;
|
||||
|
|
|
@ -82,6 +82,7 @@ ipfs_routing* ipfs_routing_new_offline (struct IpfsNode* local_node, struct RsaP
|
|||
// online using secio, should probably be deprecated
|
||||
ipfs_routing* ipfs_routing_new_online (struct IpfsNode* local_node, struct RsaPrivateKey* private_key);
|
||||
int ipfs_routing_online_free(ipfs_routing*);
|
||||
int ipfs_routing_offline_free(ipfs_routing* incoming);
|
||||
// online using DHT/kademlia, the recommended router
|
||||
ipfs_routing* ipfs_routing_new_kademlia(struct IpfsNode* local_node, struct RsaPrivateKey* private_key);
|
||||
// generic routines
|
||||
|
|
|
@ -61,7 +61,7 @@ struct Libp2pVector* ipfs_journal_get_last(struct Datastore* database, int n) {
|
|||
struct Libp2pVector* vector = libp2p_utils_vector_new(1);
|
||||
if (vector != NULL) {
|
||||
struct lmdb_trans_cursor *cursor = NULL;
|
||||
if (!lmdb_journalstore_cursor_open(database->handle, &cursor)) {
|
||||
if (!lmdb_journalstore_cursor_open(database->datastore_handle, &cursor)) {
|
||||
libp2p_logger_error("journal", "Unable to open a cursor for the journalstore.\n");
|
||||
return NULL;
|
||||
}
|
||||
|
|
|
@ -9,7 +9,9 @@
|
|||
struct lmdb_trans_cursor* lmdb_trans_cursor_new() {
|
||||
struct lmdb_trans_cursor* out = (struct lmdb_trans_cursor*) malloc(sizeof(struct lmdb_trans_cursor));
|
||||
if (out != NULL) {
|
||||
out->environment = NULL;
|
||||
out->cursor = NULL;
|
||||
out->parent_transaction = NULL;
|
||||
out->transaction = NULL;
|
||||
out->database = NULL;
|
||||
}
|
||||
|
@ -22,6 +24,11 @@ struct lmdb_trans_cursor* lmdb_trans_cursor_new() {
|
|||
* @returns true(1)
|
||||
*/
|
||||
int lmdb_trans_cursor_free(struct lmdb_trans_cursor* in) {
|
||||
free(in);
|
||||
if (in != NULL) {
|
||||
if (in->database != NULL) {
|
||||
free(in->database);
|
||||
}
|
||||
free(in);
|
||||
}
|
||||
return 1;
|
||||
}
|
||||
|
|
|
@ -81,6 +81,44 @@ int repo_fsrepo_lmdb_build_record(MDB_val *key, MDB_val *value, struct Datastore
|
|||
return 1;
|
||||
}
|
||||
|
||||
/***
|
||||
* retrieve a record from the database and put in a pre-sized buffer
|
||||
* using an already allocated transaction, and with an already opened
|
||||
* database
|
||||
* @param key the key to look for
|
||||
* @param key_size the length of the key
|
||||
* @param record where to put the results
|
||||
* @param datastore where to look for the data
|
||||
* @param mdb_txn the already opened db transaction
|
||||
* @param datastore_table the reference to the already opened datastore table (database)
|
||||
* @returns true(1) on success
|
||||
*/
|
||||
int repo_fsrepo_lmdb_get_with_transaction(const unsigned char* key, size_t key_size, struct DatastoreRecord** record, MDB_txn *mdb_txn, MDB_dbi *datastore_table) {
|
||||
struct MDB_val db_key;
|
||||
struct MDB_val db_value;
|
||||
|
||||
// check parameters passed in
|
||||
if (mdb_txn == NULL || datastore_table == NULL) {
|
||||
libp2p_logger_error("lmdb_datastore", "get_w_tx: invalid transaction or table reference.\n");
|
||||
return 0;
|
||||
}
|
||||
|
||||
// prepare data
|
||||
db_key.mv_size = key_size;
|
||||
db_key.mv_data = (char*)key;
|
||||
|
||||
if (mdb_get(mdb_txn, *datastore_table, &db_key, &db_value) != 0) {
|
||||
return 0;
|
||||
}
|
||||
|
||||
if (!repo_fsrepo_lmdb_build_record(&db_key, &db_value, record)) {
|
||||
return 0;
|
||||
}
|
||||
|
||||
return 1;
|
||||
|
||||
}
|
||||
|
||||
/***
|
||||
* retrieve a record from the database and put in a pre-sized buffer
|
||||
* @param key the key to look for
|
||||
|
@ -92,20 +130,11 @@ int repo_fsrepo_lmdb_build_record(MDB_val *key, MDB_val *value, struct Datastore
|
|||
int repo_fsrepo_lmdb_get(const unsigned char* key, size_t key_size, struct DatastoreRecord **record, const struct Datastore* datastore) {
|
||||
MDB_txn* mdb_txn;
|
||||
MDB_dbi mdb_dbi;
|
||||
struct MDB_val db_key;
|
||||
struct MDB_val db_value;
|
||||
|
||||
MDB_env* mdb_env = (MDB_env*)datastore->handle;
|
||||
MDB_env* mdb_env = (MDB_env*)datastore->datastore_handle;
|
||||
if (mdb_env == NULL)
|
||||
return 0;
|
||||
|
||||
// debug
|
||||
size_t b58size = 100;
|
||||
uint8_t *b58key = (uint8_t *) malloc(b58size);
|
||||
libp2p_crypto_encoding_base58_encode(key, key_size, &b58key, &b58size);
|
||||
libp2p_logger_debug("lmdb_datastore", "Looking for key %s in datastore.\n", b58key);
|
||||
free(b58key);
|
||||
|
||||
// open transaction
|
||||
if (mdb_txn_begin(mdb_env, NULL, 0, &mdb_txn) != 0)
|
||||
return 0;
|
||||
|
@ -115,28 +144,33 @@ int repo_fsrepo_lmdb_get(const unsigned char* key, size_t key_size, struct Datas
|
|||
return 0;
|
||||
}
|
||||
|
||||
// prepare data
|
||||
db_key.mv_size = key_size;
|
||||
db_key.mv_data = (char*)key;
|
||||
int retVal = repo_fsrepo_lmdb_get_with_transaction(key, key_size, record, mdb_txn, &mdb_dbi);
|
||||
|
||||
if (mdb_get(mdb_txn, mdb_dbi, &db_key, &db_value) != 0) {
|
||||
mdb_txn_commit(mdb_txn);
|
||||
return 0;
|
||||
}
|
||||
|
||||
if (!repo_fsrepo_lmdb_build_record(&db_key, &db_value, record)) {
|
||||
mdb_txn_commit(mdb_txn);
|
||||
return 0;
|
||||
}
|
||||
|
||||
// clean up
|
||||
mdb_txn_commit(mdb_txn);
|
||||
|
||||
return retVal;
|
||||
}
|
||||
|
||||
/**
|
||||
* Open the database and create a new transaction
|
||||
* @param mdb_env the database handle
|
||||
* @param mdb_dbi the table handle to be created
|
||||
* @param mdb_txn the transaction to be created
|
||||
* @returns true(1) on success, false(0) otherwise
|
||||
*/
|
||||
int lmdb_datastore_create_transaction(MDB_env *mdb_env, MDB_dbi *mdb_dbi, MDB_txn **mdb_txn) {
|
||||
// open transaction
|
||||
if (mdb_txn_begin(mdb_env, NULL, 0, mdb_txn) != 0)
|
||||
return 0;
|
||||
if (mdb_dbi_open(*mdb_txn, "DATASTORE", MDB_DUPSORT | MDB_CREATE, mdb_dbi) != 0) {
|
||||
mdb_txn_commit(*mdb_txn);
|
||||
return 0;
|
||||
}
|
||||
return 1;
|
||||
}
|
||||
|
||||
/**
|
||||
* Write data to the datastore with the specified key
|
||||
* Write (or update) data in the datastore with the specified key
|
||||
* @param key the key
|
||||
* @param key_size the length of the key
|
||||
* @param data the data to be written
|
||||
|
@ -146,29 +180,36 @@ int repo_fsrepo_lmdb_get(const unsigned char* key, size_t key_size, struct Datas
|
|||
*/
|
||||
int repo_fsrepo_lmdb_put(unsigned const char* key, size_t key_size, unsigned char* data, size_t data_size, const struct Datastore* datastore) {
|
||||
int retVal;
|
||||
MDB_txn* mdb_txn;
|
||||
MDB_dbi mdb_dbi;
|
||||
struct MDB_val db_key;
|
||||
struct MDB_val db_value;
|
||||
unsigned long long old_timestamp = 0;
|
||||
MDB_txn *datastore_txn;
|
||||
MDB_dbi datastore_table;
|
||||
struct MDB_val datastore_key;
|
||||
struct MDB_val datastore_value;
|
||||
struct DatastoreRecord *datastore_record = NULL;
|
||||
struct JournalRecord *journalstore_record = NULL;
|
||||
struct lmdb_trans_cursor *journalstore_cursor = NULL;
|
||||
|
||||
MDB_env* mdb_env = (MDB_env*)datastore->handle;
|
||||
if (mdb_env == NULL)
|
||||
MDB_env* mdb_env = (MDB_env*)datastore->datastore_handle;
|
||||
if (mdb_env == NULL) {
|
||||
libp2p_logger_error("lmdb_datastore", "put: invalid datastore handle.\n");
|
||||
return 0;
|
||||
}
|
||||
|
||||
// open transaction
|
||||
retVal = mdb_txn_begin(mdb_env, NULL, 0, &mdb_txn);
|
||||
if (retVal != 0)
|
||||
return 0;
|
||||
retVal = mdb_dbi_open(mdb_txn, "DATASTORE", MDB_DUPSORT | MDB_CREATE, &mdb_dbi);
|
||||
if (retVal != 0)
|
||||
// open a transaction to the databases
|
||||
if (!lmdb_datastore_create_transaction(mdb_env, &datastore_table, &datastore_txn)) {
|
||||
return 0;
|
||||
}
|
||||
|
||||
// check the datastore to see if it is already there. If it is there, use its timestamp if it is older.
|
||||
repo_fsrepo_lmdb_get(key, key_size, &datastore_record, datastore);
|
||||
// build the journalstore connectivity stuff
|
||||
journalstore_cursor = lmdb_trans_cursor_new();
|
||||
if (journalstore_cursor == NULL) {
|
||||
libp2p_logger_error("lmdb_datastore", "put: Unable to allocate memory for journalstore cursor.\n");
|
||||
return 0;
|
||||
}
|
||||
journalstore_cursor->environment = mdb_env;
|
||||
journalstore_cursor->parent_transaction = datastore_txn;
|
||||
|
||||
// see if what we want is already in the datastore
|
||||
repo_fsrepo_lmdb_get_with_transaction(key, key_size, &datastore_record, datastore_txn, &datastore_table);
|
||||
if (datastore_record != NULL) {
|
||||
// build the journalstore_record with the search criteria
|
||||
journalstore_record = lmdb_journal_record_new();
|
||||
|
@ -177,13 +218,14 @@ int repo_fsrepo_lmdb_put(unsigned const char* key, size_t key_size, unsigned cha
|
|||
memcpy(journalstore_record->hash, key, key_size);
|
||||
journalstore_record->timestamp = datastore_record->timestamp;
|
||||
// look up the corresponding journalstore record for possible updating
|
||||
journalstore_cursor = lmdb_trans_cursor_new();
|
||||
journalstore_cursor->transaction = mdb_txn;
|
||||
lmdb_journalstore_get_record((void*)mdb_env, journalstore_cursor, &journalstore_record);
|
||||
lmdb_journalstore_get_record(datastore->datastore_handle, journalstore_cursor, &journalstore_record);
|
||||
lmdb_journalstore_cursor_close(journalstore_cursor);
|
||||
} else { // it wasn't previously in the database
|
||||
datastore_record = libp2p_datastore_record_new();
|
||||
if (datastore_record == NULL) {
|
||||
libp2p_logger_error("lmdb_datastore", "put: Unable to allocate memory for DatastoreRecord.\n");
|
||||
lmdb_trans_cursor_free(journalstore_cursor);
|
||||
mdb_txn_commit(datastore_txn);
|
||||
return 0;
|
||||
}
|
||||
}
|
||||
|
@ -192,14 +234,16 @@ int repo_fsrepo_lmdb_put(unsigned const char* key, size_t key_size, unsigned cha
|
|||
unsigned long long now = os_utils_gmtime();
|
||||
if (datastore_record->timestamp == 0 || datastore_record->timestamp > now) {
|
||||
//we need to update the timestamp. Be sure to update the journal too. (done further down)
|
||||
old_timestamp = datastore_record->timestamp;
|
||||
//old_timestamp = datastore_record->timestamp;
|
||||
datastore_record->timestamp = now;
|
||||
}
|
||||
// fill in the other fields
|
||||
datastore_record->key_size = key_size;
|
||||
datastore_record->key = (uint8_t*)key;
|
||||
datastore_record->key = (uint8_t*) malloc(key_size);
|
||||
memcpy(datastore_record->key, key, key_size);
|
||||
datastore_record->value_size = data_size;
|
||||
datastore_record->value = data;
|
||||
datastore_record->value = (uint8_t *) malloc(data_size);
|
||||
memcpy(datastore_record->value, data, data_size);
|
||||
|
||||
// convert it into a byte array
|
||||
|
||||
|
@ -208,35 +252,38 @@ int repo_fsrepo_lmdb_put(unsigned const char* key, size_t key_size, unsigned cha
|
|||
repo_fsrepo_lmdb_encode_record(datastore_record, &record, &record_size);
|
||||
|
||||
// prepare data
|
||||
db_key.mv_size = key_size;
|
||||
db_key.mv_data = (char*)key;
|
||||
datastore_key.mv_size = key_size;
|
||||
datastore_key.mv_data = (char*)key;
|
||||
|
||||
// write
|
||||
db_value.mv_size = record_size;
|
||||
db_value.mv_data = record;
|
||||
datastore_value.mv_size = record_size;
|
||||
datastore_value.mv_data = record;
|
||||
|
||||
retVal = mdb_put(mdb_txn, mdb_dbi, &db_key, &db_value, MDB_NODUPDATA);
|
||||
retVal = mdb_put(datastore_txn, datastore_table, &datastore_key, &datastore_value, MDB_NODUPDATA);
|
||||
|
||||
if (retVal == 0) {
|
||||
// added the datastore record, now work with the journalstore
|
||||
// Successfully added the datastore record. Now work with the journalstore.
|
||||
if (journalstore_record != NULL) {
|
||||
if (journalstore_record->timestamp != datastore_record->timestamp) {
|
||||
// we need to update
|
||||
journalstore_record->timestamp = datastore_record->timestamp;
|
||||
lmdb_journalstore_cursor_put(journalstore_cursor, journalstore_record);
|
||||
lmdb_journalstore_cursor_close(journalstore_cursor);
|
||||
lmdb_journal_record_free(journalstore_record);
|
||||
}
|
||||
} else {
|
||||
// add it to the journalstore
|
||||
journalstore_record = lmdb_journal_record_new();
|
||||
journalstore_record->hash = (uint8_t*)key;
|
||||
journalstore_record->hash = (uint8_t*) malloc(key_size);
|
||||
memcpy(journalstore_record->hash, key, key_size);
|
||||
journalstore_record->hash_size = key_size;
|
||||
journalstore_record->timestamp = datastore_record->timestamp;
|
||||
journalstore_record->pending = 1; // TODO: Calculate this correctly
|
||||
journalstore_record->pin = 1;
|
||||
if (!lmdb_journalstore_journal_add(mdb_txn, journalstore_record)) {
|
||||
if (!lmdb_journalstore_journal_add(journalstore_cursor, journalstore_record)) {
|
||||
libp2p_logger_error("lmdb_datastore", "Datastore record was added, but problem adding Journalstore record. Continuing.\n");
|
||||
}
|
||||
lmdb_journalstore_cursor_close(journalstore_cursor);
|
||||
lmdb_journal_record_free(journalstore_record);
|
||||
retVal = 1;
|
||||
}
|
||||
|
@ -247,8 +294,10 @@ int repo_fsrepo_lmdb_put(unsigned const char* key, size_t key_size, unsigned cha
|
|||
}
|
||||
|
||||
// cleanup
|
||||
mdb_txn_commit(datastore_txn);
|
||||
free(record);
|
||||
mdb_txn_commit(mdb_txn);
|
||||
lmdb_trans_cursor_free(journalstore_cursor);
|
||||
libp2p_datastore_record_free(datastore_record);
|
||||
return retVal;
|
||||
}
|
||||
|
||||
|
@ -279,7 +328,7 @@ int repo_fsrepro_lmdb_open(int argc, char** argv, struct Datastore* datastore) {
|
|||
return 0;
|
||||
}
|
||||
|
||||
datastore->handle = (void*)mdb_env;
|
||||
datastore->datastore_handle = (void*)mdb_env;
|
||||
return 1;
|
||||
}
|
||||
|
||||
|
@ -291,99 +340,11 @@ int repo_fsrepro_lmdb_open(int argc, char** argv, struct Datastore* datastore) {
|
|||
* @param datastore the datastore struct that contains information about the opened database
|
||||
*/
|
||||
int repo_fsrepo_lmdb_close(struct Datastore* datastore) {
|
||||
struct MDB_env* mdb_env = (struct MDB_env*)datastore->handle;
|
||||
struct MDB_env* mdb_env = (struct MDB_env*)datastore->datastore_handle;
|
||||
mdb_env_close(mdb_env);
|
||||
return 1;
|
||||
}
|
||||
|
||||
/***
|
||||
* Create a new cursor on the datastore database
|
||||
* @param datastore the place to store the cursor
|
||||
* @returns true(1) on success, false(0) otherwise
|
||||
*/
|
||||
int repo_fsrepo_lmdb_cursor_open(struct Datastore* datastore) {
|
||||
if (datastore->handle != NULL) {
|
||||
MDB_env* mdb_env = (MDB_env*)datastore->handle;
|
||||
MDB_dbi mdb_dbi;
|
||||
if (datastore->cursor == NULL ) {
|
||||
datastore->cursor = malloc(sizeof(struct lmdb_trans_cursor));
|
||||
struct lmdb_trans_cursor* cursor = (struct lmdb_trans_cursor*)datastore->cursor;
|
||||
// open transaction
|
||||
if (mdb_txn_begin(mdb_env, NULL, 0, &cursor->transaction) != 0)
|
||||
return 0;
|
||||
MDB_txn* mdb_txn = (MDB_txn*)cursor->transaction;
|
||||
if (mdb_dbi_open(mdb_txn, "DATASTORE", MDB_DUPSORT | MDB_CREATE, &mdb_dbi) != 0) {
|
||||
mdb_txn_commit(mdb_txn);
|
||||
return 0;
|
||||
}
|
||||
// open cursor
|
||||
if (mdb_cursor_open(mdb_txn, mdb_dbi, &cursor->cursor) != 0) {
|
||||
mdb_txn_commit(mdb_txn);
|
||||
return 0;
|
||||
}
|
||||
return 1;
|
||||
}
|
||||
}
|
||||
return 0;
|
||||
}
|
||||
|
||||
/***
|
||||
* Get a record using a cursor
|
||||
* @param key the key from the record
|
||||
* @param key_length the length of the key
|
||||
* @param value the value of the record
|
||||
* @param value_length the length of the value
|
||||
* @param CURSOR_FIRST or CURSOR_NEXT
|
||||
* @param datastore holds the reference to the opened cursor
|
||||
* @returns true(1) on success, false(0) otherwise
|
||||
*/
|
||||
int repo_fsrepo_lmdb_cursor_get(unsigned char** key, int* key_length,
|
||||
unsigned char** value, int* value_length,
|
||||
enum DatastoreCursorOp op, struct Datastore* datastore)
|
||||
{
|
||||
if (datastore->cursor != NULL) {
|
||||
struct lmdb_trans_cursor* tc = (struct lmdb_trans_cursor*)datastore->cursor;
|
||||
MDB_val mdb_key;
|
||||
MDB_val mdb_value;
|
||||
MDB_cursor_op co = MDB_FIRST;
|
||||
if (op == CURSOR_FIRST)
|
||||
co = MDB_FIRST;
|
||||
else if (op == CURSOR_NEXT)
|
||||
co = MDB_NEXT;
|
||||
if (mdb_cursor_get(tc->cursor, &mdb_key, &mdb_value, co) != 0) {
|
||||
return 0;
|
||||
}
|
||||
*key = (unsigned char*)malloc(mdb_key.mv_size);
|
||||
memcpy(*key, mdb_key.mv_data, mdb_key.mv_size);
|
||||
*key_length = mdb_key.mv_size;
|
||||
if (value != NULL) { // don't do this if a null is passed in, time saver
|
||||
*value = (unsigned char*)malloc(mdb_value.mv_size);
|
||||
memcpy(*value, mdb_value.mv_data, mdb_value.mv_size);
|
||||
*value_length = mdb_value.mv_size;
|
||||
}
|
||||
return 1;
|
||||
}
|
||||
return 0;
|
||||
}
|
||||
/**
|
||||
* Close an existing cursor
|
||||
* @param datastore the context
|
||||
* @returns true(1) on success
|
||||
*/
|
||||
int repo_fsrepo_lmdb_cursor_close(struct Datastore* datastore) {
|
||||
if (datastore->cursor != NULL) {
|
||||
struct lmdb_trans_cursor* cursor = (struct lmdb_trans_cursor*)datastore->cursor;
|
||||
if (cursor->cursor != NULL) {
|
||||
mdb_cursor_close(cursor->cursor);
|
||||
mdb_txn_commit(cursor->transaction);
|
||||
free(cursor);
|
||||
return 1;
|
||||
}
|
||||
free(cursor);
|
||||
}
|
||||
return 0;
|
||||
}
|
||||
|
||||
/***
|
||||
* Places the LMDB methods into the datastore's function pointers
|
||||
* @param datastore the datastore to fill
|
||||
|
@ -394,10 +355,6 @@ int repo_fsrepo_lmdb_cast(struct Datastore* datastore) {
|
|||
datastore->datastore_close = &repo_fsrepo_lmdb_close;
|
||||
datastore->datastore_put = &repo_fsrepo_lmdb_put;
|
||||
datastore->datastore_get = &repo_fsrepo_lmdb_get;
|
||||
datastore->datastore_cursor_open = &repo_fsrepo_lmdb_cursor_open;
|
||||
datastore->datastore_cursor_get = &repo_fsrepo_lmdb_cursor_get;
|
||||
datastore->datastore_cursor_close = &repo_fsrepo_lmdb_cursor_close;
|
||||
datastore->cursor = NULL;
|
||||
return 1;
|
||||
}
|
||||
|
||||
|
|
|
@ -39,6 +39,7 @@ int lmdb_journalstore_generate_key(const struct JournalRecord* journal_record, s
|
|||
db_key->mv_data = time_varint;
|
||||
return 1;
|
||||
}
|
||||
|
||||
/***
|
||||
* Convert the JournalRec struct into a lmdb key and lmdb value
|
||||
* @param journal_record the record to convert
|
||||
|
@ -113,23 +114,31 @@ int lmdb_journalstore_build_record(const struct MDB_val* db_key, const struct MD
|
|||
* @param hash_size the size of the hash
|
||||
* @returns true(1) on success, false(0) otherwise
|
||||
*/
|
||||
int lmdb_journalstore_journal_add(MDB_txn* mdb_txn, struct JournalRecord* journal_record) {
|
||||
MDB_dbi mdb_dbi;
|
||||
struct MDB_val db_key;
|
||||
struct MDB_val db_value;
|
||||
int lmdb_journalstore_journal_add(struct lmdb_trans_cursor *journalstore_cursor, struct JournalRecord *journalstore_record) {
|
||||
|
||||
if (!lmdb_journalstore_build_key_value_pair(journal_record, &db_key, &db_value)) {
|
||||
libp2p_logger_error("lmdb_journalstore", "Unable to generate key value pair for journal_add.\n");
|
||||
MDB_val journalstore_key;
|
||||
MDB_val journalstore_value;
|
||||
|
||||
if (!lmdb_journalstore_build_key_value_pair(journalstore_record, &journalstore_key, &journalstore_value)) {
|
||||
libp2p_logger_error("lmdbd_journalstore", "add: Unable to convert journalstore record to key/value.\n");
|
||||
return 0;
|
||||
}
|
||||
|
||||
// open the journal table
|
||||
if (mdb_dbi_open(mdb_txn, "JOURNALSTORE", MDB_DUPSORT | MDB_CREATE, &mdb_dbi) != 0) {
|
||||
libp2p_logger_error("lmdb_journalstore", "Unable to open JOURNALSTORE database.\n");
|
||||
return 0;
|
||||
// create transaction if necessary
|
||||
if (journalstore_cursor->transaction == NULL) {
|
||||
mdb_txn_begin(journalstore_cursor->environment, journalstore_cursor->parent_transaction, 0, &journalstore_cursor->transaction);
|
||||
}
|
||||
|
||||
if (mdb_put(mdb_txn, mdb_dbi, &db_key, &db_value, 0) == 0) {
|
||||
if (journalstore_cursor->database == NULL) {
|
||||
// open the journal table
|
||||
journalstore_cursor->database = (MDB_dbi*) malloc(sizeof(MDB_dbi));
|
||||
if (mdb_dbi_open(journalstore_cursor->transaction, "JOURNALSTORE", MDB_DUPSORT | MDB_CREATE, journalstore_cursor->database) != 0) {
|
||||
libp2p_logger_error("lmdb_journalstore", "Unable to open JOURNALSTORE database.\n");
|
||||
return 0;
|
||||
}
|
||||
}
|
||||
|
||||
if (mdb_put(journalstore_cursor->transaction, *journalstore_cursor->database, &journalstore_key, &journalstore_value, 0) != 0) {
|
||||
libp2p_logger_error("lmdb_journalstore", "Unable to add to JOURNALSTORE database.\n");
|
||||
return 0;
|
||||
}
|
||||
|
@ -146,10 +155,34 @@ int lmdb_journalstore_journal_add(MDB_txn* mdb_txn, struct JournalRecord* journa
|
|||
*/
|
||||
int lmdb_journalstore_get_record(void* handle, struct lmdb_trans_cursor *journalstore_cursor, struct JournalRecord **journalstore_record)
|
||||
{
|
||||
if (journalstore_cursor == NULL || journalstore_cursor->transaction == NULL) {
|
||||
libp2p_logger_error("lmdb_journalstore", "get_record: journalstore cursor not initialized properly.\n");
|
||||
|
||||
if (handle == NULL) {
|
||||
libp2p_logger_error("lmdb_journalstore", "get_record: database environment not set up.\n");
|
||||
return 0;
|
||||
}
|
||||
struct MDB_env *mdb_env = (struct MDB_env*)handle;
|
||||
|
||||
// create a new transaction if necessary
|
||||
if (journalstore_cursor->transaction == NULL) {
|
||||
if (mdb_txn_begin(mdb_env, NULL, 0, &journalstore_cursor->transaction) != 0) {
|
||||
libp2p_logger_error("lmdb_journanstore", "get_record: Attempt to begin transaction failed.\n");
|
||||
return 0;
|
||||
}
|
||||
}
|
||||
|
||||
if (journalstore_cursor->database == NULL) {
|
||||
// open the journal table
|
||||
journalstore_cursor->database = (MDB_dbi*) malloc(sizeof(MDB_dbi));
|
||||
if (journalstore_cursor->database == NULL) {
|
||||
libp2p_logger_error("lmdb_journalstore", "get_record: Unable to allocate memory for journalstore database.\n");
|
||||
return 0;
|
||||
}
|
||||
if (mdb_dbi_open(journalstore_cursor->transaction, "JOURNALSTORE", MDB_DUPSORT | MDB_CREATE, journalstore_cursor->database) != 0) {
|
||||
libp2p_logger_error("lmdb_journalstore", "Unable to open JOURNALSTORE database.\n");
|
||||
return 0;
|
||||
}
|
||||
}
|
||||
|
||||
if (journalstore_cursor->cursor == NULL) {
|
||||
if (!lmdb_journalstore_cursor_open(handle, &journalstore_cursor)) {
|
||||
libp2p_logger_error("lmdb_journalstore", "Unable to open cursor in get_record.\n");
|
||||
|
@ -158,12 +191,11 @@ int lmdb_journalstore_get_record(void* handle, struct lmdb_trans_cursor *journal
|
|||
}
|
||||
// search for the timestamp
|
||||
if (!lmdb_journalstore_cursor_get(journalstore_cursor, CURSOR_FIRST, journalstore_record)) {
|
||||
libp2p_logger_error("lmdb_journalstore", "Unable to find any records in table.\n");
|
||||
libp2p_logger_debug("lmdb_journalstore", "Unable to find any records in table.\n");
|
||||
return 0;
|
||||
}
|
||||
// now look for the hash key
|
||||
|
||||
return 0;
|
||||
return 1;
|
||||
}
|
||||
|
||||
/**
|
||||
|
@ -332,17 +364,21 @@ int lmdb_journalstore_cursor_put(struct lmdb_trans_cursor *crsr, struct JournalR
|
|||
}
|
||||
|
||||
/**
|
||||
* Close the cursor
|
||||
* Close the cursor, but does not free the struct. It simply closes the cursor
|
||||
* and commits the transaction.
|
||||
* @param crsr a lmdb_trans_cursor pointer
|
||||
* @returns true(1)
|
||||
*/
|
||||
int lmdb_journalstore_cursor_close(struct lmdb_trans_cursor *cursor) {
|
||||
if (cursor->cursor != NULL) {
|
||||
mdb_cursor_close(cursor->cursor);
|
||||
mdb_txn_commit(cursor->transaction);
|
||||
free(cursor);
|
||||
return 1;
|
||||
} else {
|
||||
free(cursor);
|
||||
if (cursor != NULL) {
|
||||
if (cursor->cursor != NULL) {
|
||||
mdb_cursor_close(cursor->cursor);
|
||||
cursor->cursor = NULL;
|
||||
}
|
||||
if (cursor->transaction != NULL) {
|
||||
mdb_txn_commit(cursor->transaction);
|
||||
cursor->transaction = NULL;
|
||||
}
|
||||
}
|
||||
return 0;
|
||||
return 1;
|
||||
}
|
||||
|
|
|
@ -115,3 +115,8 @@ struct IpfsRouting* ipfs_routing_new_offline (struct IpfsNode* local_node, struc
|
|||
|
||||
return offlineRouting;
|
||||
}
|
||||
|
||||
int ipfs_routing_offline_free(ipfs_routing* incoming) {
|
||||
free(incoming);
|
||||
return 1;
|
||||
}
|
||||
|
|
|
@ -152,7 +152,8 @@ int test_import_small_file() {
|
|||
|
||||
// get the repo
|
||||
drop_and_build_repository(repo_path, 4001, NULL, NULL);
|
||||
ipfs_node_online_new(repo_path, &local_node);
|
||||
|
||||
ipfs_node_offline_new(repo_path, &local_node);
|
||||
|
||||
// write to ipfs
|
||||
struct HashtableNode* write_node;
|
||||
|
|
|
@ -85,7 +85,7 @@ int test_datastore_list_journal() {
|
|||
}
|
||||
// open cursor
|
||||
struct lmdb_trans_cursor *crsr = NULL;
|
||||
if (!lmdb_journalstore_cursor_open(fs_repo->config->datastore->handle, &crsr)) {
|
||||
if (!lmdb_journalstore_cursor_open(fs_repo->config->datastore->datastore_handle, &crsr)) {
|
||||
ipfs_repo_fsrepo_free(fs_repo);
|
||||
return 0;
|
||||
}
|
||||
|
|
|
@ -205,7 +205,7 @@ int drop_build_open_repo(const char* path, struct FSRepo** fs_repo, const char*
|
|||
|
||||
if (config_filename_to_copy != NULL) {
|
||||
// attach config filename to path
|
||||
char *config = (char*) malloc(strlen(path) + 7);
|
||||
char *config = (char*) malloc(strlen(path) + 8);
|
||||
strcpy(config, path);
|
||||
// erase slash if there is one
|
||||
if (config[strlen(path)-1] == '/')
|
||||
|
|
Loading…
Reference in a new issue