From 2c6f954461721bdf07534d5ccd06465a9d60eb35 Mon Sep 17 00:00:00 2001 From: Shane Bryldt Date: Tue, 17 Jan 2017 18:20:43 +0000 Subject: [PATCH] FS-9775: Committing start of very basic datastore concept, will start tieing in the physical transport layer for replication under new ticket/branch --- libs/libblade/src/blade_datastore.c | 115 ++++++++++++++++++++ libs/libblade/src/blade_stack.c | 27 +++++ libs/libblade/src/include/blade_datastore.h | 7 ++ libs/libblade/src/include/blade_stack.h | 7 ++ libs/libblade/src/include/blade_types.h | 2 + libs/libblade/test/bladec.c | 38 +++++++ 6 files changed, 196 insertions(+) diff --git a/libs/libblade/src/blade_datastore.c b/libs/libblade/src/blade_datastore.c index 57982215e9..d97cd9840d 100644 --- a/libs/libblade/src/blade_datastore.c +++ b/libs/libblade/src/blade_datastore.c @@ -42,8 +42,18 @@ typedef enum { struct blade_datastore_s { bdspvt_flag_t flags; ks_pool_t *pool; + unqlite *db; }; +struct blade_datastore_fetch_userdata_s +{ + blade_datastore_t *bds; + blade_datastore_fetch_callback_t callback; + void *userdata; +}; +typedef struct blade_datastore_fetch_userdata_s blade_datastore_fetch_userdata_t; + + KS_DECLARE(ks_status_t) blade_datastore_destroy(blade_datastore_t **bdsP) { @@ -61,6 +71,11 @@ KS_DECLARE(ks_status_t) blade_datastore_destroy(blade_datastore_t **bdsP) flags = bds->flags; pool = bds->pool; + if (bds->db) { + unqlite_close(bds->db); + bds->db = NULL; + } + ks_pool_free(bds->pool, &bds); if (pool && (flags & BDS_MYPOOL)) ks_pool_close(&pool); @@ -84,6 +99,17 @@ KS_DECLARE(ks_status_t) blade_datastore_create(blade_datastore_t **bdsP, ks_pool bds->pool = pool; *bdsP = bds; + if (unqlite_open(&bds->db, NULL, UNQLITE_OPEN_IN_MEMORY) != UNQLITE_OK) { + const char *errbuf = NULL; + blade_datastore_error(bds, &errbuf, NULL); + ks_log(KS_LOG_ERROR, "BDS Error: %s\n", errbuf); + return KS_STATUS_FAIL; + } + + // @todo unqlite_lib_config(UNQLITE_LIB_CONFIG_MEM_ERR_CALLBACK) + + // @todo VM init if document store is used (and output consumer callback) + return KS_STATUS_SUCCESS; } @@ -93,6 +119,95 @@ KS_DECLARE(void) blade_datastore_pulse(blade_datastore_t *bds, int32_t timeout) ks_assert(timeout >= 0); } +KS_DECLARE(void) blade_datastore_error(blade_datastore_t *bds, const char **buffer, int32_t *buffer_length) +{ + ks_assert(bds); + ks_assert(bds->db); + ks_assert(buffer); + + unqlite_config(bds->db, UNQLITE_CONFIG_ERR_LOG, buffer, buffer_length); +} + +KS_DECLARE(ks_status_t) blade_datastore_store(blade_datastore_t *bds, const void *key, int32_t key_length, const void *data, int64_t data_length) +{ + int32_t rc; + ks_status_t ret = KS_STATUS_SUCCESS; + + ks_assert(bds); + ks_assert(bds->db); + ks_assert(key); + ks_assert(key_length > 0); + ks_assert(data); + ks_assert(data_length > 0); + + rc = unqlite_begin(bds->db); + + if (rc != UNQLITE_OK) { + if (rc == UNQLITE_BUSY) ret = KS_STATUS_TIMEOUT; + else { + const char *errbuf; + blade_datastore_error(bds, &errbuf, NULL); + ks_log(KS_LOG_ERROR, "BDS Error: %s\n", errbuf); + + ret = KS_STATUS_FAIL; + } + } else if (unqlite_kv_store(bds->db, key, key_length, data, data_length) == UNQLITE_OK) unqlite_commit(bds->db); + else unqlite_rollback(bds->db); + + return ret; +} + +int blade_datastore_fetch_callback(const void *data, unsigned int data_length, void *userdata) +{ + int rc = UNQLITE_OK; + blade_datastore_fetch_userdata_t *ud = NULL; + + ks_assert(data); + ks_assert(data_length > 0); + ks_assert(userdata); + + ud = (blade_datastore_fetch_userdata_t *)userdata; + if (!ud->callback(ud->bds, data, data_length, ud->userdata)) rc = UNQLITE_ABORT; + + return rc; +} + +KS_DECLARE(ks_status_t) blade_datastore_fetch(blade_datastore_t *bds, + blade_datastore_fetch_callback_t callback, + const void *key, + int32_t key_length, + void *userdata) +{ + int32_t rc; + ks_status_t ret = KS_STATUS_SUCCESS; + blade_datastore_fetch_userdata_t ud; + + ks_assert(bds); + ks_assert(bds->db); + ks_assert(callback); + ks_assert(key); + ks_assert(key_length > 0); + + ud.bds = bds; + ud.callback = callback; + ud.userdata = userdata; + + rc = unqlite_kv_fetch_callback(bds->db, key, key_length, blade_datastore_fetch_callback, &ud); + + if (rc != UNQLITE_OK) { + if (rc == UNQLITE_BUSY) ret = KS_STATUS_TIMEOUT; + else { + const char *errbuf; + blade_datastore_error(bds, &errbuf, NULL); + ks_log(KS_LOG_ERROR, "BDS Error: %s\n", errbuf); + + ret = KS_STATUS_FAIL; + } + } + + return ret; +} + /* For Emacs: * Local Variables: * mode:c diff --git a/libs/libblade/src/blade_stack.c b/libs/libblade/src/blade_stack.c index 311d368a3a..d7cf76235c 100644 --- a/libs/libblade/src/blade_stack.c +++ b/libs/libblade/src/blade_stack.c @@ -162,6 +162,33 @@ KS_DECLARE(void) blade_handle_datastore_start(blade_handle_t *bh) blade_datastore_create(&bh->datastore, bh->pool); } +KS_DECLARE(ks_status_t) blade_handle_datastore_store(blade_handle_t *bh, const void *key, int32_t key_length, const void *data, int64_t data_length) +{ + ks_assert(bh); + ks_assert(bh->datastore); + ks_assert(key); + ks_assert(key_length > 0); + ks_assert(data); + ks_assert(data_length > 0); + + return blade_datastore_store(bh->datastore, key, key_length, data, data_length); +} + +KS_DECLARE(ks_status_t) blade_handle_datastore_fetch(blade_handle_t *bh, + blade_datastore_fetch_callback_t callback, + const void *key, + int32_t key_length, + void *userdata) +{ + ks_assert(bh); + ks_assert(bh->datastore); + ks_assert(callback); + ks_assert(key); + ks_assert(key_length > 0); + + return blade_datastore_fetch(bh->datastore, callback, key, key_length, userdata); +} + /* For Emacs: * Local Variables: diff --git a/libs/libblade/src/include/blade_datastore.h b/libs/libblade/src/include/blade_datastore.h index f51864cb8a..62918a3a41 100644 --- a/libs/libblade/src/include/blade_datastore.h +++ b/libs/libblade/src/include/blade_datastore.h @@ -39,6 +39,13 @@ KS_BEGIN_EXTERN_C KS_DECLARE(ks_status_t) blade_datastore_create(blade_datastore_t **bdsP, ks_pool_t *pool); KS_DECLARE(ks_status_t) blade_datastore_destroy(blade_datastore_t **bdsP); KS_DECLARE(void) blade_datastore_pulse(blade_datastore_t *bds, int32_t timeout); +KS_DECLARE(void) blade_datastore_error(blade_datastore_t *bds, const char **buffer, int32_t *buffer_length); +KS_DECLARE(ks_status_t) blade_datastore_store(blade_datastore_t *bds, const void *key, int32_t key_length, const void *data, int64_t data_length); +KS_DECLARE(ks_status_t) blade_datastore_fetch(blade_datastore_t *bds, + blade_datastore_fetch_callback_t callback, + const void *key, + int32_t key_length, + void *userdata); KS_END_EXTERN_C #endif diff --git a/libs/libblade/src/include/blade_stack.h b/libs/libblade/src/include/blade_stack.h index a5744e459f..81411fabdc 100644 --- a/libs/libblade/src/include/blade_stack.h +++ b/libs/libblade/src/include/blade_stack.h @@ -47,6 +47,13 @@ KS_DECLARE(void) blade_handle_myid(blade_handle_t *bh, char *buffer); KS_DECLARE(void) blade_handle_autoroute(blade_handle_t *bh, ks_bool_t autoroute, ks_port_t port); KS_DECLARE(ks_status_t) blade_handle_bind(blade_handle_t *bh, const char *ip, ks_port_t port, ks_dht_endpoint_t **endpoint); KS_DECLARE(void) blade_handle_pulse(blade_handle_t *bh, int32_t timeout); +KS_DECLARE(void) blade_handle_datastore_start(blade_handle_t *bh); +KS_DECLARE(ks_status_t) blade_handle_datastore_store(blade_handle_t *bh, const void *key, int32_t key_length, const void *data, int64_t data_length); +KS_DECLARE(ks_status_t) blade_handle_datastore_fetch(blade_handle_t *bh, + blade_datastore_fetch_callback_t callback, + const void *key, + int32_t key_length, + void *userdata); KS_END_EXTERN_C #endif diff --git a/libs/libblade/src/include/blade_types.h b/libs/libblade/src/include/blade_types.h index 5367bbbf38..0bb43a4c07 100644 --- a/libs/libblade/src/include/blade_types.h +++ b/libs/libblade/src/include/blade_types.h @@ -41,6 +41,8 @@ typedef struct blade_handle_s blade_handle_t; typedef struct blade_peer_s blade_peer_t; typedef struct blade_datastore_s blade_datastore_t; +typedef ks_bool_t (*blade_datastore_fetch_callback_t)(blade_datastore_t *bds, const void *data, uint32_t data_length, void *userdata); + KS_END_EXTERN_C #endif diff --git a/libs/libblade/test/bladec.c b/libs/libblade/test/bladec.c index f03fe52de4..478b614bc0 100644 --- a/libs/libblade/test/bladec.c +++ b/libs/libblade/test/bladec.c @@ -30,12 +30,16 @@ void command_test(blade_handle_t *bh, char *args); void command_quit(blade_handle_t *bh, char *args); void command_myid(blade_handle_t *bh, char *args); void command_bind(blade_handle_t *bh, char *args); +void command_store(blade_handle_t *bh, char *args); +void command_fetch(blade_handle_t *bh, char *args); static const struct command_def_s command_defs[] = { { "test", command_test }, { "quit", command_quit }, { "myid", command_myid }, { "bind", command_bind }, + { "store", command_store }, + { "fetch", command_fetch }, { NULL, NULL } }; @@ -202,3 +206,37 @@ void command_bind(blade_handle_t *bh, char *args) blade_handle_bind(bh, ip, p, NULL); } + +void command_store(blade_handle_t *bh, char *args) +{ + char *key; + char *data; + + ks_assert(args); + + blade_handle_datastore_start(bh); + + parse_argument(&args, &key, ' '); + parse_argument(&args, &data, ' '); + + blade_handle_datastore_store(bh, key, strlen(key), data, strlen(data) + 1); +} + +ks_bool_t blade_datastore_fetch_callback(blade_datastore_t *bds, const void *data, uint32_t data_length, void *userdata) +{ + ks_log(KS_LOG_INFO, "%s\n", data); + return KS_TRUE; +} + +void command_fetch(blade_handle_t *bh, char *args) +{ + char *key; + + ks_assert(args); + + blade_handle_datastore_start(bh); + + parse_argument(&args, &key, ' '); + + blade_handle_datastore_fetch(bh, blade_datastore_fetch_callback, key, strlen(key), bh); +}