ben.carrasco/freeswitch
1
0 Fork 0
mirror of https://github.com/signalwire/freeswitch.git synced 2026-11-15 05:48:33 +00:00
Code Issues Packages Projects Releases Wiki Activity

FS-10167: Revisited ks_pool to embed the ks_pool_t pointer into the prefix of each allocation, updated libks and libblade to remove unneccessary pool pointers in structures, pool-allocated memory can now be freed without needing to pass the original pool

Browse Source
This commit is contained in:
Shane Bryldt committed 2017-08-03 20:26:07 -06:00
1 parent 077fde224b
commit 53690120ac
47 files changed
+500 -567
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

libs/libblade/src/blade_connection.c
+7 -24
View File
@@ -35,7 +35,6 @@
struct blade_connection_s {
blade_handle_t *handle;
ks_pool_t *pool;
void *transport_data;
blade_transport_callbacks_t *transport_callbacks;
@@ -59,7 +58,7 @@ ks_status_t blade_connection_onstate_shutdown(blade_connection_t *bc);
ks_status_t blade_connection_onstate_run(blade_connection_t *bc);
static void blade_connection_cleanup(ks_pool_t *pool, void *ptr, void *arg, ks_pool_cleanup_action_t action, ks_pool_cleanup_type_t type)
static void blade_connection_cleanup(void *ptr, void *arg, ks_pool_cleanup_action_t action, ks_pool_cleanup_type_t type)
{
blade_connection_t *bc = (blade_connection_t *)ptr;
@@ -72,12 +71,6 @@ static void blade_connection_cleanup(ks_pool_t *pool, void *ptr, void *arg, ks_p
blade_connection_shutdown(bc);
break;
case KS_MPCL_DESTROY:
// @todo remove this, it's just for posterity in debugging
bc->sending = NULL;
bc->lock = NULL;
//ks_pool_free(bc->pool, &bc->id);
bc->id = NULL;
break;
}
}
@@ -96,7 +89,6 @@ KS_DECLARE(ks_status_t) blade_connection_create(blade_connection_t **bcP, blade_
bc = ks_pool_alloc(pool, sizeof(blade_connection_t));
bc->handle = bh;
bc->pool = pool;
ks_cond_create(&bc->cond, pool);
ks_assert(bc->cond);
@@ -111,7 +103,7 @@ KS_DECLARE(ks_status_t) blade_connection_create(blade_connection_t **bcP, blade_
ks_q_create(&bc->sending, pool, 0);
ks_assert(bc->sending);
ks_pool_set_cleanup(pool, bc, NULL, blade_connection_cleanup);
ks_pool_set_cleanup(bc, NULL, blade_connection_cleanup);
ks_log(KS_LOG_DEBUG, "Created\n");
@@ -129,13 +121,11 @@ KS_DECLARE(ks_status_t) blade_connection_destroy(blade_connection_t **bcP)
ks_assert(*bcP);
bc = *bcP;
pool = bc->pool;
//ks_pool_free(bc->pool, bcP);
ks_pool_close(&pool);
*bcP = NULL;
pool = ks_pool_get(bc);
ks_pool_close(&pool);
return KS_STATUS_SUCCESS;
}
@@ -182,13 +172,6 @@ KS_DECLARE(blade_handle_t *) blade_connection_handle_get(blade_connection_t *bc)
return bc->handle;
}
KS_DECLARE(ks_pool_t *) blade_connection_pool_get(blade_connection_t *bc)
{
ks_assert(bc);
return bc->pool;
}
KS_DECLARE(const char *) blade_connection_id_get(blade_connection_t *bc)
{
ks_assert(bc);
@@ -343,8 +326,8 @@ KS_DECLARE(void) blade_connection_session_set(blade_connection_t *bc, const char
{
ks_assert(bc);
if (bc->session) ks_pool_free(bc->pool, &bc->session);
bc->session = ks_pstrdup(bc->pool, id);
if (bc->session) ks_pool_free(&bc->session);
bc->session = ks_pstrdup(ks_pool_get(bc), id);
}
void *blade_connection_state_thread(ks_thread_t *thread, void *data)
libs/libblade/src/blade_connectionmgr.c
+5 -10
View File
@@ -35,13 +35,12 @@
struct blade_connectionmgr_s {
blade_handle_t *handle;
ks_pool_t *pool;
ks_hash_t *connections; // id, blade_connection_t*
};
static void blade_connectionmgr_cleanup(ks_pool_t *pool, void *ptr, void *arg, ks_pool_cleanup_action_t action, ks_pool_cleanup_type_t type)
static void blade_connectionmgr_cleanup(void *ptr, void *arg, ks_pool_cleanup_action_t action, ks_pool_cleanup_type_t type)
{
//blade_connectionmgr_t *bcmgr = (blade_connectionmgr_t *)ptr;
@@ -69,12 +68,11 @@ KS_DECLARE(ks_status_t) blade_connectionmgr_create(blade_connectionmgr_t **bcmgr
bcmgr = ks_pool_alloc(pool, sizeof(blade_connectionmgr_t));
bcmgr->handle = bh;
bcmgr->pool = pool;
ks_hash_create(&bcmgr->connections, KS_HASH_MODE_CASE_INSENSITIVE, KS_HASH_FLAG_RWLOCK | KS_HASH_FLAG_DUP_CHECK | KS_HASH_FLAG_FREE_KEY, bcmgr->pool);
ks_hash_create(&bcmgr->connections, KS_HASH_MODE_CASE_INSENSITIVE, KS_HASH_FLAG_RWLOCK | KS_HASH_FLAG_DUP_CHECK | KS_HASH_FLAG_FREE_KEY, pool);
ks_assert(bcmgr->connections);
ks_pool_set_cleanup(pool, bcmgr, NULL, blade_connectionmgr_cleanup);
ks_pool_set_cleanup(bcmgr, NULL, blade_connectionmgr_cleanup);
*bcmgrP = bcmgr;
@@ -92,10 +90,7 @@ KS_DECLARE(ks_status_t) blade_connectionmgr_destroy(blade_connectionmgr_t **bcmg
bcmgr = *bcmgrP;
*bcmgrP = NULL;
ks_assert(bcmgr);
pool = bcmgr->pool;
pool = ks_pool_get(bcmgr);
ks_pool_close(&pool);
return KS_STATUS_SUCCESS;
@@ -149,7 +144,7 @@ KS_DECLARE(ks_status_t) blade_connectionmgr_connection_add(blade_connectionmgr_t
ks_assert(bcmgr);
ks_assert(bc);
key = ks_pstrdup(bcmgr->pool, blade_connection_id_get(bc));
key = ks_pstrdup(ks_pool_get(bcmgr), blade_connection_id_get(bc));
ks_hash_insert(bcmgr->connections, (void *)key, bc);
ks_log(KS_LOG_DEBUG, "Connection Added: %s\n", key);
libs/libblade/src/blade_identity.c
+13 -13
View File
@@ -34,8 +34,6 @@
#include "blade.h"
struct blade_identity_s {
ks_pool_t *pool;
const char *uri;
const char *components;
@@ -46,7 +44,7 @@ struct blade_identity_s {
};
// @todo missed a structure to use cleanup callbacks
static void blade_identity_cleanup(ks_pool_t *pool, void *ptr, void *arg, ks_pool_cleanup_action_t action, ks_pool_cleanup_type_t type)
static void blade_identity_cleanup(void *ptr, void *arg, ks_pool_cleanup_action_t action, ks_pool_cleanup_type_t type)
{
blade_identity_t *bi = (blade_identity_t *)ptr;
@@ -56,8 +54,8 @@ static void blade_identity_cleanup(ks_pool_t *pool, void *ptr, void *arg, ks_poo
case KS_MPCL_ANNOUNCE:
break;
case KS_MPCL_TEARDOWN:
if (bi->uri) ks_pool_free(bi->pool, &bi->uri);
if (bi->components) ks_pool_free(bi->pool, &bi->components);
if (bi->uri) ks_pool_free(&bi->uri);
if (bi->components) ks_pool_free(&bi->components);
if (bi->parameters) ks_hash_destroy(&bi->parameters);
break;
case KS_MPCL_DESTROY:
@@ -73,9 +71,8 @@ KS_DECLARE(ks_status_t) blade_identity_create(blade_identity_t **biP, ks_pool_t
ks_assert(pool);
bi = ks_pool_alloc(pool, sizeof(blade_identity_t));
bi->pool = pool;
ks_pool_set_cleanup(pool, bi, NULL, blade_identity_cleanup);
ks_pool_set_cleanup(bi, NULL, blade_identity_cleanup);
*biP = bi;
@@ -91,7 +88,7 @@ KS_DECLARE(ks_status_t) blade_identity_destroy(blade_identity_t **biP)
bi = *biP;
ks_pool_free(bi->pool, biP);
ks_pool_free(biP);
return KS_STATUS_SUCCESS;
}
@@ -100,18 +97,21 @@ KS_DECLARE(ks_status_t) blade_identity_parse(blade_identity_t *bi, const char *u
{
char *tmp = NULL;
char *tmp2 = NULL;
ks_pool_t *pool = NULL;
ks_assert(bi);
ks_assert(uri);
ks_log(KS_LOG_DEBUG, "Parsing URI: %s\n", uri);
pool = ks_pool_get(bi);
if (bi->uri) {
ks_pool_free(bi->pool, &bi->uri);
ks_pool_free(bi->pool, &bi->components);
ks_pool_free(&bi->uri);
ks_pool_free(&bi->components);
}
bi->uri = ks_pstrdup(bi->pool, uri);
bi->components = tmp = ks_pstrdup(bi->pool, uri);
bi->uri = ks_pstrdup(pool, uri);
bi->components = tmp = ks_pstrdup(pool, uri);
bi->name = tmp;
if (!(tmp = strchr(tmp, '@'))) return KS_STATUS_FAIL;
@@ -137,7 +137,7 @@ KS_DECLARE(ks_status_t) blade_identity_parse(blade_identity_t *bi, const char *u
}
if (!bi->parameters) {
ks_hash_create(&bi->parameters, KS_HASH_MODE_CASE_INSENSITIVE, KS_HASH_FLAG_NOLOCK | KS_HASH_FLAG_DUP_CHECK, bi->pool);
ks_hash_create(&bi->parameters, KS_HASH_MODE_CASE_INSENSITIVE, KS_HASH_FLAG_NOLOCK | KS_HASH_FLAG_DUP_CHECK, pool);
ks_assert(bi->parameters);
}
ks_hash_insert(bi->parameters, key, val);
libs/libblade/src/blade_mastermgr.c
+26 -22
View File
@@ -35,14 +35,13 @@
struct blade_mastermgr_s {
blade_handle_t *handle;
ks_pool_t *pool;
// @todo how does "exclusive" play into the controllers, does "exclusive" mean only one provider can exist for a given protocol and realm? what does non exclusive mean?
ks_hash_t *protocols; // protocols that have been published with blade.publish, and the details to locate a protocol controller with blade.locate
};
static void blade_mastermgr_cleanup(ks_pool_t *pool, void *ptr, void *arg, ks_pool_cleanup_action_t action, ks_pool_cleanup_type_t type)
static void blade_mastermgr_cleanup(void *ptr, void *arg, ks_pool_cleanup_action_t action, ks_pool_cleanup_type_t type)
{
//blade_mastermgr_t *bmmgr = (blade_mastermgr_t *)ptr;
@@ -70,12 +69,11 @@ KS_DECLARE(ks_status_t) blade_mastermgr_create(blade_mastermgr_t **bmmgrP, blade
bmmgr = ks_pool_alloc(pool, sizeof(blade_mastermgr_t));
bmmgr->handle = bh;
bmmgr->pool = pool;
ks_hash_create(&bmmgr->protocols, KS_HASH_MODE_CASE_INSENSITIVE, KS_HASH_FLAG_RWLOCK | KS_HASH_FLAG_DUP_CHECK | KS_HASH_FLAG_FREE_KEY | KS_HASH_FLAG_FREE_VALUE, bmmgr->pool);
ks_hash_create(&bmmgr->protocols, KS_HASH_MODE_CASE_INSENSITIVE, KS_HASH_FLAG_RWLOCK | KS_HASH_FLAG_DUP_CHECK | KS_HASH_FLAG_FREE_KEY | KS_HASH_FLAG_FREE_VALUE, pool);
ks_assert(bmmgr->protocols);
ks_pool_set_cleanup(pool, bmmgr, NULL, blade_mastermgr_cleanup);
ks_pool_set_cleanup(bmmgr, NULL, blade_mastermgr_cleanup);
*bmmgrP = bmmgr;
@@ -93,9 +91,7 @@ KS_DECLARE(ks_status_t) blade_mastermgr_destroy(blade_mastermgr_t **bmmgrP)
bmmgr = *bmmgrP;
*bmmgrP = NULL;
ks_assert(bmmgr);
pool = bmmgr->pool;
pool = ks_pool_get(bmmgr);
ks_pool_close(&pool);
@@ -111,8 +107,13 @@ KS_DECLARE(blade_handle_t *) blade_mastermgr_handle_get(blade_mastermgr_t *bmmgr
KS_DECLARE(ks_status_t) blade_mastermgr_purge(blade_mastermgr_t *bmmgr, const char *nodeid)
{
ks_pool_t *pool = NULL;
ks_hash_t *cleanup = NULL;
ks_assert(bmmgr);
pool = ks_pool_get(bmmgr);
ks_hash_write_lock(bmmgr->protocols);
for (ks_hash_iterator_t *it = ks_hash_first(bmmgr->protocols, KS_UNLOCKED); it; it = ks_hash_next(&it)) {
const char *key = NULL;
@@ -121,7 +122,7 @@ KS_DECLARE(ks_status_t) blade_mastermgr_purge(blade_mastermgr_t *bmmgr, const ch
ks_hash_this(it, (const void **)&key, NULL, (void **)&bp);
if (blade_protocol_purge(bp, nodeid)) {
if (!cleanup) ks_hash_create(&cleanup, KS_HASH_MODE_CASE_INSENSITIVE, KS_HASH_FLAG_RWLOCK | KS_HASH_FLAG_DUP_CHECK, bmmgr->pool);
if (!cleanup) ks_hash_create(&cleanup, KS_HASH_MODE_CASE_INSENSITIVE, KS_HASH_FLAG_RWLOCK | KS_HASH_FLAG_DUP_CHECK, pool);
ks_hash_insert(cleanup, (void *)key, bp);
}
}
@@ -152,7 +153,7 @@ KS_DECLARE(blade_protocol_t *) blade_mastermgr_protocol_lookup(blade_mastermgr_t
ks_assert(protocol);
ks_assert(realm);
key = ks_psprintf(bmmgr->pool, "%s@%s", protocol, realm);
key = ks_psprintf(ks_pool_get(bmmgr), "%s@%s", protocol, realm);
bp = (blade_protocol_t *)ks_hash_search(bmmgr->protocols, (void *)key, KS_READLOCKED);
// @todo if (bp) blade_protocol_read_lock(bp);
@@ -163,6 +164,7 @@ KS_DECLARE(blade_protocol_t *) blade_mastermgr_protocol_lookup(blade_mastermgr_t
KS_DECLARE(ks_status_t) blade_mastermgr_controller_add(blade_mastermgr_t *bmmgr, const char *protocol, const char *realm, const char *controller)
{
ks_pool_t *pool = NULL;
blade_protocol_t *bp = NULL;
char *key = NULL;
@@ -171,7 +173,9 @@ KS_DECLARE(ks_status_t) blade_mastermgr_controller_add(blade_mastermgr_t *bmmgr,
ks_assert(realm);
ks_assert(controller);
key = ks_psprintf(bmmgr->pool, "%s@%s", protocol, realm);
pool = ks_pool_get(bmmgr);
key = ks_psprintf(pool, "%s@%s", protocol, realm);
ks_hash_write_lock(bmmgr->protocols);
@@ -181,16 +185,16 @@ KS_DECLARE(ks_status_t) blade_mastermgr_controller_add(blade_mastermgr_t *bmmgr,
}
if (!bp) {
blade_protocol_create(&bp, bmmgr->pool, protocol, realm);
blade_protocol_create(&bp, pool, protocol, realm);
ks_assert(bp);
ks_log(KS_LOG_DEBUG, "Protocol Added: %s\n", key);
ks_hash_insert(bmmgr->protocols, (void *)ks_pstrdup(bmmgr->pool, key), bp);
ks_hash_insert(bmmgr->protocols, (void *)ks_pstrdup(pool, key), bp);
}
blade_protocol_controllers_add(bp, controller);
ks_pool_free(bmmgr->pool, &key);
ks_pool_free(&key);
ks_hash_write_unlock(bmmgr->protocols);
@@ -208,7 +212,7 @@ KS_DECLARE(ks_status_t) blade_mastermgr_channel_add(blade_mastermgr_t *bmmgr, co
ks_assert(realm);
ks_assert(channel);
key = ks_psprintf(bmmgr->pool, "%s@%s", protocol, realm);
key = ks_psprintf(ks_pool_get(bmmgr), "%s@%s", protocol, realm);
bp = (blade_protocol_t *)ks_hash_search(bmmgr->protocols, (void *)key, KS_READLOCKED);
if (!bp) {
@@ -219,7 +223,7 @@ KS_DECLARE(ks_status_t) blade_mastermgr_channel_add(blade_mastermgr_t *bmmgr, co
blade_protocol_channel_add(bp, channel);
done:
ks_pool_free(bmmgr->pool, &key);
ks_pool_free(&key);
ks_hash_read_unlock(bmmgr->protocols);
@@ -237,7 +241,7 @@ KS_DECLARE(ks_status_t) blade_mastermgr_channel_remove(blade_mastermgr_t *bmmgr,
ks_assert(realm);
ks_assert(channel);
key = ks_psprintf(bmmgr->pool, "%s@%s", protocol, realm);
key = ks_psprintf(ks_pool_get(bmmgr), "%s@%s", protocol, realm);
bp = (blade_protocol_t *)ks_hash_search(bmmgr->protocols, (void *)key, KS_READLOCKED);
if (!bp) {
@@ -248,7 +252,7 @@ KS_DECLARE(ks_status_t) blade_mastermgr_channel_remove(blade_mastermgr_t *bmmgr,
blade_protocol_channel_remove(bp, channel);
done:
ks_pool_free(bmmgr->pool, &key);
ks_pool_free(&key);
ks_hash_read_unlock(bmmgr->protocols);
@@ -269,7 +273,7 @@ KS_DECLARE(ks_status_t) blade_mastermgr_channel_authorize(blade_mastermgr_t *bmm
ks_assert(controller);
ks_assert(target);
key = ks_psprintf(bmmgr->pool, "%s@%s", protocol, realm);
key = ks_psprintf(ks_pool_get(bmmgr), "%s@%s", protocol, realm);
bp = (blade_protocol_t *)ks_hash_search(bmmgr->protocols, (void *)key, KS_READLOCKED);
if (!bp) {
@@ -280,7 +284,7 @@ KS_DECLARE(ks_status_t) blade_mastermgr_channel_authorize(blade_mastermgr_t *bmm
ret = blade_protocol_channel_authorize(bp, remove, channel, controller, target);
done:
ks_pool_free(bmmgr->pool, &key);
ks_pool_free(&key);
ks_hash_read_unlock(bmmgr->protocols);
@@ -300,7 +304,7 @@ KS_DECLARE(ks_bool_t) blade_mastermgr_channel_verify(blade_mastermgr_t *bmmgr, c
ks_assert(channel);
ks_assert(target);
key = ks_psprintf(bmmgr->pool, "%s@%s", protocol, realm);
key = ks_psprintf(ks_pool_get(bmmgr), "%s@%s", protocol, realm);
bp = (blade_protocol_t *)ks_hash_search(bmmgr->protocols, (void *)key, KS_READLOCKED);
if (!bp) goto done;
@@ -308,7 +312,7 @@ KS_DECLARE(ks_bool_t) blade_mastermgr_channel_verify(blade_mastermgr_t *bmmgr, c
ret = blade_protocol_channel_verify(bp, channel, target);
done:
ks_pool_free(bmmgr->pool, &key);
ks_pool_free(&key);
ks_hash_read_unlock(bmmgr->protocols);
libs/libblade/src/blade_protocol.c
+14 -14
View File
@@ -34,8 +34,6 @@
#include "blade.h"
struct blade_protocol_s {
ks_pool_t *pool;
const char *name;
const char *realm;
ks_hash_t *controllers;
@@ -44,7 +42,7 @@ struct blade_protocol_s {
};
static void blade_protocol_cleanup(ks_pool_t *pool, void *ptr, void *arg, ks_pool_cleanup_action_t action, ks_pool_cleanup_type_t type)
static void blade_protocol_cleanup(void *ptr, void *arg, ks_pool_cleanup_action_t action, ks_pool_cleanup_type_t type)
{
blade_protocol_t *bp = (blade_protocol_t *)ptr;
@@ -54,8 +52,8 @@ static void blade_protocol_cleanup(ks_pool_t *pool, void *ptr, void *arg, ks_poo
case KS_MPCL_ANNOUNCE:
break;
case KS_MPCL_TEARDOWN:
if (bp->name) ks_pool_free(bp->pool, &bp->name);
if (bp->realm) ks_pool_free(bp->pool, &bp->realm);
if (bp->name) ks_pool_free(&bp->name);
if (bp->realm) ks_pool_free(&bp->realm);
if (bp->controllers) ks_hash_destroy(&bp->controllers);
if (bp->channels) ks_hash_destroy(&bp->channels);
break;
@@ -74,17 +72,16 @@ KS_DECLARE(ks_status_t) blade_protocol_create(blade_protocol_t **bpP, ks_pool_t
ks_assert(realm);
bp = ks_pool_alloc(pool, sizeof(blade_protocol_t));
bp->pool = pool;
bp->name = ks_pstrdup(pool, name);
bp->realm = ks_pstrdup(pool, realm);
ks_hash_create(&bp->controllers, KS_HASH_MODE_CASE_INSENSITIVE, KS_HASH_FLAG_RWLOCK | KS_HASH_FLAG_DUP_CHECK | KS_HASH_FLAG_FREE_KEY, bp->pool);
ks_hash_create(&bp->controllers, KS_HASH_MODE_CASE_INSENSITIVE, KS_HASH_FLAG_RWLOCK | KS_HASH_FLAG_DUP_CHECK | KS_HASH_FLAG_FREE_KEY, pool);
ks_assert(bp->controllers);
ks_hash_create(&bp->channels, KS_HASH_MODE_CASE_INSENSITIVE, KS_HASH_FLAG_RWLOCK | KS_HASH_FLAG_DUP_CHECK | KS_HASH_FLAG_FREE_KEY, bp->pool);
ks_hash_create(&bp->channels, KS_HASH_MODE_CASE_INSENSITIVE, KS_HASH_FLAG_RWLOCK | KS_HASH_FLAG_DUP_CHECK | KS_HASH_FLAG_FREE_KEY, pool);
ks_assert(bp->channels);
ks_pool_set_cleanup(pool, bp, NULL, blade_protocol_cleanup);
ks_pool_set_cleanup(bp, NULL, blade_protocol_cleanup);
*bpP = bp;
@@ -100,7 +97,7 @@ KS_DECLARE(ks_status_t) blade_protocol_destroy(blade_protocol_t **bpP)
bp = *bpP;
ks_pool_free(bp->pool, bpP);
ks_pool_free(bpP);
return KS_STATUS_SUCCESS;
}
@@ -161,7 +158,7 @@ KS_DECLARE(ks_status_t) blade_protocol_controllers_add(blade_protocol_t *bp, con
ks_assert(bp);
ks_assert(nodeid);
key = ks_pstrdup(bp->pool, nodeid);
key = ks_pstrdup(ks_pool_get(bp), nodeid);
ks_hash_insert(bp->controllers, (void *)key, (void *)KS_TRUE);
ks_log(KS_LOG_DEBUG, "Protocol Controller Added: %s to %s@%s\n", nodeid, bp->name, bp->realm);
@@ -172,12 +169,15 @@ KS_DECLARE(ks_status_t) blade_protocol_controllers_add(blade_protocol_t *bp, con
KS_DECLARE(ks_status_t) blade_protocol_channel_add(blade_protocol_t *bp, const char *name)
{
ks_status_t ret = KS_STATUS_SUCCESS;
ks_pool_t *pool = NULL;
ks_hash_t *authorized = NULL;
char *key = NULL;
ks_assert(bp);
ks_assert(name);
pool = ks_pool_get(bp);
ks_hash_write_lock(bp->channels);
if (ks_hash_search(bp->channels, (void *)name, KS_UNLOCKED)) {
@@ -185,9 +185,9 @@ KS_DECLARE(ks_status_t) blade_protocol_channel_add(blade_protocol_t *bp, const c
goto done;
}
ks_hash_create(&authorized, KS_HASH_MODE_CASE_INSENSITIVE, KS_HASH_FLAG_RWLOCK | KS_HASH_FLAG_DUP_CHECK | KS_HASH_FLAG_FREE_KEY, bp->pool);
ks_hash_create(&authorized, KS_HASH_MODE_CASE_INSENSITIVE, KS_HASH_FLAG_RWLOCK | KS_HASH_FLAG_DUP_CHECK | KS_HASH_FLAG_FREE_KEY, pool);
key = ks_pstrdup(bp->pool, name);
key = ks_pstrdup(pool, name);
ks_hash_insert(bp->channels, (void *)key, (void *)authorized);
ks_log(KS_LOG_DEBUG, "Protocol Channel Added: %s to %s@%s\n", key, bp->name, bp->realm);
@@ -239,7 +239,7 @@ KS_DECLARE(ks_status_t) blade_protocol_channel_authorize(blade_protocol_t *bp, k
} else ret = KS_STATUS_NOT_FOUND;
}
else {
ks_hash_insert(authorizations, (void *)ks_pstrdup(bp->pool, target), (void *)KS_TRUE);
ks_hash_insert(authorizations, (void *)ks_pstrdup(ks_pool_get(bp), target), (void *)KS_TRUE);
ks_log(KS_LOG_DEBUG, "Protocol Channel Authorization Added: %s to %s@%s/%s\n", target, bp->name, bp->realm, channel);
}
}
libs/libblade/src/blade_routemgr.c
+9 -10
View File
@@ -35,13 +35,12 @@
struct blade_routemgr_s {
blade_handle_t *handle;
ks_pool_t *pool;
ks_hash_t *routes; // id, id
};
static void blade_routemgr_cleanup(ks_pool_t *pool, void *ptr, void *arg, ks_pool_cleanup_action_t action, ks_pool_cleanup_type_t type)
static void blade_routemgr_cleanup(void *ptr, void *arg, ks_pool_cleanup_action_t action, ks_pool_cleanup_type_t type)
{
//blade_routemgr_t *brmgr = (blade_routemgr_t *)ptr;
@@ -69,13 +68,12 @@ KS_DECLARE(ks_status_t) blade_routemgr_create(blade_routemgr_t **brmgrP, blade_h
brmgr = ks_pool_alloc(pool, sizeof(blade_routemgr_t));
brmgr->handle = bh;
brmgr->pool = pool;
// @note can let removes free keys and values for routes, both are strings and allocated from the same pool as the hash itself
ks_hash_create(&brmgr->routes, KS_HASH_MODE_CASE_INSENSITIVE, KS_HASH_FLAG_RWLOCK | KS_HASH_FLAG_DUP_CHECK | KS_HASH_FLAG_FREE_KEY | KS_HASH_FLAG_FREE_VALUE, brmgr->pool);
ks_hash_create(&brmgr->routes, KS_HASH_MODE_CASE_INSENSITIVE, KS_HASH_FLAG_RWLOCK | KS_HASH_FLAG_DUP_CHECK | KS_HASH_FLAG_FREE_KEY | KS_HASH_FLAG_FREE_VALUE, pool);
ks_assert(brmgr->routes);
ks_pool_set_cleanup(pool, brmgr, NULL, blade_routemgr_cleanup);
ks_pool_set_cleanup(brmgr, NULL, blade_routemgr_cleanup);
*brmgrP = brmgr;
@@ -93,9 +91,7 @@ KS_DECLARE(ks_status_t) blade_routemgr_destroy(blade_routemgr_t **brmgrP)
brmgr = *brmgrP;
*brmgrP = NULL;
ks_assert(brmgr);
pool = brmgr->pool;
pool = ks_pool_get(brmgr);
ks_pool_close(&pool);
@@ -126,6 +122,7 @@ KS_DECLARE(blade_session_t *) blade_routemgr_route_lookup(blade_routemgr_t *brmg
KS_DECLARE(ks_status_t) blade_routemgr_route_add(blade_routemgr_t *brmgr, const char *target, const char *router)
{
ks_pool_t *pool = NULL;
char *key = NULL;
char *value = NULL;
@@ -133,8 +130,10 @@ KS_DECLARE(ks_status_t) blade_routemgr_route_add(blade_routemgr_t *brmgr, const
ks_assert(target);
ks_assert(router);
key = ks_pstrdup(brmgr->pool, target);
value = ks_pstrdup(brmgr->pool, router);
pool = ks_pool_get(brmgr);
key = ks_pstrdup(pool, target);
value = ks_pstrdup(pool, router);
ks_hash_insert(brmgr->routes, (void *)key, (void *)value);
libs/libblade/src/blade_rpc.c
+14 -21
View File
@@ -35,7 +35,6 @@
struct blade_rpc_s {
blade_handle_t *handle;
ks_pool_t *pool;
const char *method;
const char *protocol;
@@ -47,7 +46,6 @@ struct blade_rpc_s {
struct blade_rpc_request_s {
blade_handle_t *handle;
ks_pool_t *pool;
const char *session_id;
@@ -60,7 +58,6 @@ struct blade_rpc_request_s {
struct blade_rpc_response_s {
blade_handle_t *handle;
ks_pool_t *pool;
const char *session_id;
@@ -70,7 +67,7 @@ struct blade_rpc_response_s {
};
static void blade_rpc_cleanup(ks_pool_t *pool, void *ptr, void *arg, ks_pool_cleanup_action_t action, ks_pool_cleanup_type_t type)
static void blade_rpc_cleanup(void *ptr, void *arg, ks_pool_cleanup_action_t action, ks_pool_cleanup_type_t type)
{
//blade_rpc_t *brpc = (blade_rpc_t *)ptr;
@@ -102,14 +99,13 @@ KS_DECLARE(ks_status_t) blade_rpc_create(blade_rpc_t **brpcP, blade_handle_t *bh
brpc = ks_pool_alloc(pool, sizeof(blade_rpc_t));
brpc->handle = bh;
brpc->pool = pool;
brpc->method = ks_pstrdup(pool, method);
if (protocol) brpc->protocol = ks_pstrdup(pool, protocol);
if (realm) brpc->realm = ks_pstrdup(pool, realm);
brpc->callback = callback;
brpc->data = data;
ks_pool_set_cleanup(pool, brpc, NULL, blade_rpc_cleanup);
ks_pool_set_cleanup(brpc, NULL, blade_rpc_cleanup);
*brpcP = brpc;
@@ -125,13 +121,12 @@ KS_DECLARE(ks_status_t) blade_rpc_destroy(blade_rpc_t **brpcP)
ks_assert(*brpcP);
brpc = *brpcP;
*brpcP = NULL;
pool = brpc->pool;
pool = ks_pool_get(brpc);
ks_pool_close(&pool);
*brpcP = NULL;
return KS_STATUS_SUCCESS;
}
@@ -178,7 +173,7 @@ KS_DECLARE(void *) blade_rpc_data_get(blade_rpc_t *brpc)
}
static void blade_rpc_request_cleanup(ks_pool_t *pool, void *ptr, void *arg, ks_pool_cleanup_action_t action, ks_pool_cleanup_type_t type)
static void blade_rpc_request_cleanup(void *ptr, void *arg, ks_pool_cleanup_action_t action, ks_pool_cleanup_type_t type)
{
blade_rpc_request_t *brpcreq = (blade_rpc_request_t *)ptr;
@@ -188,7 +183,7 @@ static void blade_rpc_request_cleanup(ks_pool_t *pool, void *ptr, void *arg, ks_
case KS_MPCL_ANNOUNCE:
break;
case KS_MPCL_TEARDOWN:
ks_pool_free(brpcreq->pool, (void **)&brpcreq->session_id);
ks_pool_free((void **)&brpcreq->session_id);
cJSON_Delete(brpcreq->message);
// @todo delete data if present, requires update to ks_pool for self tracking the pool in allocation header
break;
@@ -215,14 +210,13 @@ KS_DECLARE(ks_status_t) blade_rpc_request_create(blade_rpc_request_t **brpcreqP,
brpcreq = ks_pool_alloc(pool, sizeof(blade_rpc_request_t));
brpcreq->handle = bh;
brpcreq->pool = pool;
brpcreq->session_id = ks_pstrdup(pool, session_id);
brpcreq->message = cJSON_Duplicate(json, 1);
brpcreq->message_id = cJSON_GetObjectCstr(brpcreq->message, "id");
brpcreq->callback = callback;
brpcreq->data = data;
ks_pool_set_cleanup(pool, brpcreq, NULL, blade_rpc_request_cleanup);
ks_pool_set_cleanup(brpcreq, NULL, blade_rpc_request_cleanup);
*brpcreqP = brpcreq;
@@ -238,14 +232,14 @@ KS_DECLARE(ks_status_t) blade_rpc_request_destroy(blade_rpc_request_t **brpcreqP
brpcreq = *brpcreqP;
ks_pool_free(brpcreq->pool, brpcreqP);
ks_pool_free(brpcreqP);
return KS_STATUS_SUCCESS;
}
KS_DECLARE(ks_status_t) blade_rpc_request_duplicate(blade_rpc_request_t **brpcreqP, blade_rpc_request_t *brpcreq)
{
return blade_rpc_request_create(brpcreqP, brpcreq->handle, brpcreq->pool, brpcreq->session_id, brpcreq->message, brpcreq->callback, brpcreq->data);
return blade_rpc_request_create(brpcreqP, brpcreq->handle, ks_pool_get(brpcreq), brpcreq->session_id, brpcreq->message, brpcreq->callback, brpcreq->data);
}
KS_DECLARE(blade_handle_t *) blade_rpc_request_handle_get(blade_rpc_request_t *brpcreq)
@@ -302,7 +296,7 @@ KS_DECLARE(ks_status_t) blade_rpc_request_raw_create(ks_pool_t *pool, cJSON **js
ks_uuid(&msgid);
mid = ks_uuid_str(pool, &msgid);
cJSON_AddStringToObject(root, "id", mid);
ks_pool_free(pool, &mid);
ks_pool_free(&mid);
cJSON_AddStringToObject(root, "method", method);
@@ -317,7 +311,7 @@ KS_DECLARE(ks_status_t) blade_rpc_request_raw_create(ks_pool_t *pool, cJSON **js
}
static void blade_rpc_response_cleanup(ks_pool_t *pool, void *ptr, void *arg, ks_pool_cleanup_action_t action, ks_pool_cleanup_type_t type)
static void blade_rpc_response_cleanup(void *ptr, void *arg, ks_pool_cleanup_action_t action, ks_pool_cleanup_type_t type)
{
blade_rpc_response_t *brpcres = (blade_rpc_response_t *)ptr;
@@ -327,7 +321,7 @@ static void blade_rpc_response_cleanup(ks_pool_t *pool, void *ptr, void *arg, ks
case KS_MPCL_ANNOUNCE:
break;
case KS_MPCL_TEARDOWN:
ks_pool_free(brpcres->pool, (void **)&brpcres->session_id);
ks_pool_free((void **)&brpcres->session_id);
blade_rpc_request_destroy(&brpcres->request);
cJSON_Delete(brpcres->message);
break;
@@ -354,12 +348,11 @@ KS_DECLARE(ks_status_t) blade_rpc_response_create(blade_rpc_response_t **brpcres
brpcres = ks_pool_alloc(pool, sizeof(blade_rpc_response_t));
brpcres->handle = bh;
brpcres->pool = pool;
brpcres->session_id = ks_pstrdup(pool, session_id);
brpcres->request = brpcreq;
brpcres->message = cJSON_Duplicate(json, 1);
ks_pool_set_cleanup(pool, brpcres, NULL, blade_rpc_response_cleanup);
ks_pool_set_cleanup(brpcres, NULL, blade_rpc_response_cleanup);
*brpcresP = brpcres;
@@ -375,7 +368,7 @@ KS_DECLARE(ks_status_t) blade_rpc_response_destroy(blade_rpc_response_t **brpcre
brpcres = *brpcresP;
ks_pool_free(brpcres->pool, brpcresP);
ks_pool_free(brpcresP);
return KS_STATUS_SUCCESS;
}
libs/libblade/src/blade_rpcmgr.c
+13 -17
View File
@@ -35,7 +35,6 @@
struct blade_rpcmgr_s {
blade_handle_t *handle;
ks_pool_t *pool;
ks_hash_t *corerpcs; // method, blade_rpc_t*
ks_hash_t *protocolrpcs; // method, blade_rpc_t*
@@ -44,7 +43,7 @@ struct blade_rpcmgr_s {
};
static void blade_rpcmgr_cleanup(ks_pool_t *pool, void *ptr, void *arg, ks_pool_cleanup_action_t action, ks_pool_cleanup_type_t type)
static void blade_rpcmgr_cleanup(void *ptr, void *arg, ks_pool_cleanup_action_t action, ks_pool_cleanup_type_t type)
{
blade_rpcmgr_t *brpcmgr = (blade_rpcmgr_t *)ptr;
ks_hash_iterator_t *it = NULL;
@@ -91,18 +90,17 @@ KS_DECLARE(ks_status_t) blade_rpcmgr_create(blade_rpcmgr_t **brpcmgrP, blade_han
brpcmgr = ks_pool_alloc(pool, sizeof(blade_rpcmgr_t));
brpcmgr->handle = bh;
brpcmgr->pool = pool;
ks_hash_create(&brpcmgr->corerpcs, KS_HASH_MODE_CASE_INSENSITIVE, KS_HASH_FLAG_RWLOCK | KS_HASH_FLAG_DUP_CHECK | KS_HASH_FLAG_FREE_KEY, brpcmgr->pool);
ks_hash_create(&brpcmgr->corerpcs, KS_HASH_MODE_CASE_INSENSITIVE, KS_HASH_FLAG_RWLOCK | KS_HASH_FLAG_DUP_CHECK | KS_HASH_FLAG_FREE_KEY, pool);
ks_assert(brpcmgr->corerpcs);
ks_hash_create(&brpcmgr->protocolrpcs, KS_HASH_MODE_CASE_INSENSITIVE, KS_HASH_FLAG_RWLOCK | KS_HASH_FLAG_DUP_CHECK | KS_HASH_FLAG_FREE_KEY, brpcmgr->pool);
ks_hash_create(&brpcmgr->protocolrpcs, KS_HASH_MODE_CASE_INSENSITIVE, KS_HASH_FLAG_RWLOCK | KS_HASH_FLAG_DUP_CHECK | KS_HASH_FLAG_FREE_KEY, pool);
ks_assert(brpcmgr->protocolrpcs);
ks_hash_create(&brpcmgr->requests, KS_HASH_MODE_CASE_INSENSITIVE, KS_HASH_FLAG_RWLOCK | KS_HASH_FLAG_DUP_CHECK | KS_HASH_FLAG_FREE_KEY, brpcmgr->pool);
ks_hash_create(&brpcmgr->requests, KS_HASH_MODE_CASE_INSENSITIVE, KS_HASH_FLAG_RWLOCK | KS_HASH_FLAG_DUP_CHECK | KS_HASH_FLAG_FREE_KEY, pool);
ks_assert(brpcmgr->requests);
ks_pool_set_cleanup(pool, brpcmgr, NULL, blade_rpcmgr_cleanup);
ks_pool_set_cleanup(brpcmgr, NULL, blade_rpcmgr_cleanup);
*brpcmgrP = brpcmgr;
@@ -120,9 +118,7 @@ KS_DECLARE(ks_status_t) blade_rpcmgr_destroy(blade_rpcmgr_t **brpcmgrP)
brpcmgr = *brpcmgrP;
*brpcmgrP = NULL;
ks_assert(brpcmgr);
pool = brpcmgr->pool;
pool = ks_pool_get(brpcmgr);
ks_pool_close(&pool);
@@ -157,7 +153,7 @@ KS_DECLARE(ks_status_t) blade_rpcmgr_corerpc_add(blade_rpcmgr_t *brpcmgr, blade_
ks_assert(brpcmgr);
ks_assert(brpc);
key = ks_pstrdup(brpcmgr->pool, blade_rpc_method_get(brpc));
key = ks_pstrdup(ks_pool_get(brpcmgr), blade_rpc_method_get(brpc));
ks_hash_insert(brpcmgr->corerpcs, (void *)key, (void *)brpc);
ks_log(KS_LOG_DEBUG, "CoreRPC Added: %s\n", key);
@@ -191,12 +187,12 @@ KS_DECLARE(blade_rpc_t *) blade_rpcmgr_protocolrpc_lookup(blade_rpcmgr_t *brpcmg
ks_assert(protocol);
ks_assert(realm);
key = ks_psprintf(brpcmgr->pool, "%s@%s/%s", protocol, realm, method);
key = ks_psprintf(ks_pool_get(brpcmgr), "%s@%s/%s", protocol, realm, method);
brpc = ks_hash_search(brpcmgr->protocolrpcs, (void *)key, KS_READLOCKED);
// @todo if (brpc) blade_rpc_read_lock(brpc);
ks_hash_read_unlock(brpcmgr->protocolrpcs);
ks_pool_free(brpcmgr->pool, &key);
ks_pool_free(&key);
return brpc;
}
@@ -220,7 +216,7 @@ KS_DECLARE(ks_status_t) blade_rpcmgr_protocolrpc_add(blade_rpcmgr_t *brpcmgr, bl
realm = blade_rpc_realm_get(brpc);
ks_assert(realm);
key = ks_psprintf(brpcmgr->pool, "%s@%s/%s", protocol, realm, method);
key = ks_psprintf(ks_pool_get(brpcmgr), "%s@%s/%s", protocol, realm, method);
ks_assert(key);
ks_hash_insert(brpcmgr->protocolrpcs, (void *)key, (void *)brpc);
@@ -250,14 +246,14 @@ KS_DECLARE(ks_status_t) blade_rpcmgr_protocolrpc_remove(blade_rpcmgr_t *brpcmgr,
realm = blade_rpc_realm_get(brpc);
ks_assert(realm);
key = ks_psprintf(brpcmgr->pool, "%s@%s/%s", protocol, realm, method);
key = ks_psprintf(ks_pool_get(brpcmgr), "%s@%s/%s", protocol, realm, method);
ks_assert(key);
ks_hash_remove(brpcmgr->protocolrpcs, (void *)key);
ks_log(KS_LOG_DEBUG, "ProtocolRPC Removed: %s\n", key);
ks_pool_free(brpcmgr->pool, &key);
ks_pool_free(&key);
return KS_STATUS_SUCCESS;
}
@@ -283,7 +279,7 @@ KS_DECLARE(ks_status_t) blade_rpcmgr_request_add(blade_rpcmgr_t *brpcmgr, blade_
ks_assert(brpcmgr);
ks_assert(brpcreq);
key = ks_pstrdup(brpcmgr->pool, blade_rpc_request_messageid_get(brpcreq));
key = ks_pstrdup(ks_pool_get(brpcmgr), blade_rpc_request_messageid_get(brpcreq));
ks_hash_insert(brpcmgr->requests, (void *)key, (void *)brpcreq);
ks_log(KS_LOG_DEBUG, "Request Added: %s\n", key);
libs/libblade/src/blade_session.c
+16 -19
View File
@@ -35,7 +35,6 @@
struct blade_session_s {
blade_handle_t *handle;
ks_pool_t *pool;
volatile blade_session_state_t state;
@@ -63,7 +62,7 @@ ks_status_t blade_session_onstate_shutdown(blade_session_t *bs);
ks_status_t blade_session_onstate_run(blade_session_t *bs);
ks_status_t blade_session_process(blade_session_t *bs, cJSON *json);
static void blade_session_cleanup(ks_pool_t *pool, void *ptr, void *arg, ks_pool_cleanup_action_t action, ks_pool_cleanup_type_t type)
static void blade_session_cleanup(void *ptr, void *arg, ks_pool_cleanup_action_t action, ks_pool_cleanup_type_t type)
{
blade_session_t *bs = (blade_session_t *)ptr;
@@ -97,7 +96,6 @@ KS_DECLARE(ks_status_t) blade_session_create(blade_session_t **bsP, blade_handle
bs = ks_pool_alloc(pool, sizeof(blade_session_t));
bs->handle = bh;
bs->pool = pool;
if (id) bs->id = ks_pstrdup(pool, id);
else {
@@ -117,10 +115,10 @@ KS_DECLARE(ks_status_t) blade_session_create(blade_session_t **bsP, blade_handle
ks_q_create(&bs->receiving, pool, 0);
ks_assert(bs->receiving);
ks_hash_create(&bs->realms, KS_HASH_MODE_CASE_INSENSITIVE, KS_HASH_FLAG_RWLOCK | KS_HASH_FLAG_DUP_CHECK | KS_HASH_FLAG_FREE_KEY, bs->pool);
ks_hash_create(&bs->realms, KS_HASH_MODE_CASE_INSENSITIVE, KS_HASH_FLAG_RWLOCK | KS_HASH_FLAG_DUP_CHECK | KS_HASH_FLAG_FREE_KEY, pool);
ks_assert(bs->realms);
ks_hash_create(&bs->routes, KS_HASH_MODE_CASE_INSENSITIVE, KS_HASH_FLAG_RWLOCK | KS_HASH_FLAG_DUP_CHECK | KS_HASH_FLAG_FREE_KEY, bs->pool);
ks_hash_create(&bs->routes, KS_HASH_MODE_CASE_INSENSITIVE, KS_HASH_FLAG_RWLOCK | KS_HASH_FLAG_DUP_CHECK | KS_HASH_FLAG_FREE_KEY, pool);
ks_assert(bs->routes);
bs->properties = cJSON_CreateObject();
@@ -128,7 +126,7 @@ KS_DECLARE(ks_status_t) blade_session_create(blade_session_t **bsP, blade_handle
ks_rwl_create(&bs->properties_lock, pool);
ks_assert(bs->properties_lock);
ks_pool_set_cleanup(pool, bs, NULL, blade_session_cleanup);
ks_pool_set_cleanup(bs, NULL, blade_session_cleanup);
ks_log(KS_LOG_DEBUG, "Created\n");
@@ -146,13 +144,12 @@ KS_DECLARE(ks_status_t) blade_session_destroy(blade_session_t **bsP)
ks_assert(*bsP);
bs = *bsP;
pool = bs->pool;
//ks_pool_free(bs->pool, bsP);
ks_pool_close(&pool);
*bsP = NULL;
pool = ks_pool_get(bs);
ks_pool_close(&pool);
return KS_STATUS_SUCCESS;
}
@@ -235,7 +232,7 @@ KS_DECLARE(ks_status_t) blade_session_realm_add(blade_session_t *bs, const char
ks_assert(bs);
ks_assert(realm);
key = ks_pstrdup(bs->pool, realm);
key = ks_pstrdup(ks_pool_get(bs), realm);
ks_hash_insert(bs->realms, (void *)key, (void *)KS_TRUE);
return KS_STATUS_SUCCESS;
@@ -264,7 +261,7 @@ KS_DECLARE(ks_status_t) blade_session_route_add(blade_session_t *bs, const char
ks_assert(bs);
ks_assert(nodeid);
key = ks_pstrdup(bs->pool, nodeid);
key = ks_pstrdup(ks_pool_get(bs), nodeid);
ks_hash_insert(bs->routes, (void *)key, (void *)KS_TRUE);
return KS_STATUS_SUCCESS;
@@ -404,9 +401,9 @@ KS_DECLARE(ks_status_t) blade_session_connection_set(blade_session_t *bs, const
if (id) {
if (bs->connection) {
// @todo best that can be done in this situation is see if the connection is still available, and if so then disconnect it... this really shouldn't happen
ks_pool_free(bs->pool, &bs->connection);
ks_pool_free(&bs->connection);
}
bs->connection = ks_pstrdup(bs->pool, id);
bs->connection = ks_pstrdup(ks_pool_get(bs), id);
ks_assert(bs->connection);
bs->ttl = 0;
@@ -417,7 +414,7 @@ KS_DECLARE(ks_status_t) blade_session_connection_set(blade_session_t *bs, const
} else if (bs->connection) {
ks_log(KS_LOG_DEBUG, "Session (%s) cleared connection (%s)\n", bs->id, bs->connection);
ks_pool_free(bs->pool, &bs->connection);
ks_pool_free(&bs->connection);
bs->ttl = ks_time_now() + (5 * KS_USEC_PER_SEC);
}
@@ -601,7 +598,7 @@ KS_DECLARE(ks_status_t) blade_session_send(blade_session_t *bs, cJSON *json, bla
// 1) Sending a request (client: method caller or consumer)
ks_log(KS_LOG_DEBUG, "Session (%s) sending request (%s) for %s\n", bs->id, id, method);
blade_rpc_request_create(&brpcreq, bs->handle, blade_handle_pool_get(bs->handle), bs->id, json, callback, data);
blade_rpc_request_create(&brpcreq, bs->handle, ks_pool_get(bs->handle), bs->id, json, callback, data);
ks_assert(brpcreq);
// @todo set request TTL and figure out when requests are checked for expiration (separate thread in the handle?)
@@ -727,7 +724,7 @@ ks_status_t blade_session_process(blade_session_t *bs, cJSON *json)
callback = blade_rpc_callback_get(brpc);
ks_assert(callback);
blade_rpc_request_create(&brpcreq, bs->handle, blade_handle_pool_get(bs->handle), bs->id, json, NULL, NULL);
blade_rpc_request_create(&brpcreq, bs->handle, ks_pool_get(bs->handle), bs->id, json, NULL, NULL);
ks_assert(brpcreq);
disconnect = callback(brpcreq, blade_rpc_data_get(brpc));
@@ -787,7 +784,7 @@ ks_status_t blade_session_process(blade_session_t *bs, cJSON *json)
callback = blade_rpc_request_callback_get(brpcreq);
blade_rpc_response_create(&brpcres, bs->handle, bs->pool, bs->id, brpcreq, json);
blade_rpc_response_create(&brpcres, bs->handle, ks_pool_get(bs), bs->id, brpcreq, json);
ks_assert(brpcres);
if (callback) disconnect = callback(brpcres, blade_rpc_request_data_get(brpcreq));
libs/libblade/src/blade_sessionmgr.c
+15 -19
View File
@@ -35,22 +35,19 @@
struct blade_sessionmgr_s {
blade_handle_t *handle;
ks_pool_t *pool;
ks_hash_t *sessions; // id, blade_session_t*
ks_hash_t *callbacks; // id, blade_session_callback_data_t*
};
struct blade_session_callback_data_s {
ks_pool_t *pool;
const char *id;
void *data;
blade_session_callback_t callback;
};
static void blade_sessionmgr_cleanup(ks_pool_t *pool, void *ptr, void *arg, ks_pool_cleanup_action_t action, ks_pool_cleanup_type_t type)
static void blade_sessionmgr_cleanup(void *ptr, void *arg, ks_pool_cleanup_action_t action, ks_pool_cleanup_type_t type)
{
//blade_sessionmgr_t *bsmgr = (blade_sessionmgr_t *)ptr;
@@ -66,7 +63,7 @@ static void blade_sessionmgr_cleanup(ks_pool_t *pool, void *ptr, void *arg, ks_p
}
}
static void blade_session_callback_data_cleanup(ks_pool_t *pool, void *ptr, void *arg, ks_pool_cleanup_action_t action, ks_pool_cleanup_type_t type)
static void blade_session_callback_data_cleanup(void *ptr, void *arg, ks_pool_cleanup_action_t action, ks_pool_cleanup_type_t type)
{
blade_session_callback_data_t *bscd = (blade_session_callback_data_t *)ptr;
@@ -76,7 +73,7 @@ static void blade_session_callback_data_cleanup(ks_pool_t *pool, void *ptr, void
case KS_MPCL_ANNOUNCE:
break;
case KS_MPCL_TEARDOWN:
ks_pool_free(bscd->pool, &bscd->id);
ks_pool_free(&bscd->id);
break;
case KS_MPCL_DESTROY:
break;
@@ -95,15 +92,14 @@ KS_DECLARE(ks_status_t) blade_sessionmgr_create(blade_sessionmgr_t **bsmgrP, bla
bsmgr = ks_pool_alloc(pool, sizeof(blade_sessionmgr_t));
bsmgr->handle = bh;
bsmgr->pool = pool;
ks_hash_create(&bsmgr->sessions, KS_HASH_MODE_CASE_INSENSITIVE, KS_HASH_FLAG_RWLOCK | KS_HASH_FLAG_DUP_CHECK | KS_HASH_FLAG_FREE_KEY, bsmgr->pool);
ks_hash_create(&bsmgr->sessions, KS_HASH_MODE_CASE_INSENSITIVE, KS_HASH_FLAG_RWLOCK | KS_HASH_FLAG_DUP_CHECK | KS_HASH_FLAG_FREE_KEY, pool);
ks_assert(bsmgr->sessions);
ks_hash_create(&bsmgr->callbacks, KS_HASH_MODE_CASE_INSENSITIVE, KS_HASH_FLAG_RWLOCK | KS_HASH_FLAG_DUP_CHECK | KS_HASH_FLAG_FREE_KEY | KS_HASH_FLAG_FREE_VALUE, bsmgr->pool);
ks_hash_create(&bsmgr->callbacks, KS_HASH_MODE_CASE_INSENSITIVE, KS_HASH_FLAG_RWLOCK | KS_HASH_FLAG_DUP_CHECK | KS_HASH_FLAG_FREE_KEY | KS_HASH_FLAG_FREE_VALUE, pool);
ks_assert(bsmgr->callbacks);
ks_pool_set_cleanup(pool, bsmgr, NULL, blade_sessionmgr_cleanup);
ks_pool_set_cleanup(bsmgr, NULL, blade_sessionmgr_cleanup);
*bsmgrP = bsmgr;
@@ -121,9 +117,7 @@ KS_DECLARE(ks_status_t) blade_sessionmgr_destroy(blade_sessionmgr_t **bsmgrP)
bsmgr = *bsmgrP;
*bsmgrP = NULL;
ks_assert(bsmgr);
pool = bsmgr->pool;
pool = ks_pool_get(bsmgr);
ks_pool_close(&pool);
@@ -178,7 +172,7 @@ KS_DECLARE(ks_status_t) blade_sessionmgr_session_add(blade_sessionmgr_t *bsmgr,
ks_assert(bsmgr);
ks_assert(bs);
key = ks_pstrdup(bsmgr->pool, blade_session_id_get(bs));
key = ks_pstrdup(ks_pool_get(bsmgr), blade_session_id_get(bs));
ks_hash_insert(bsmgr->sessions, (void *)key, bs);
ks_log(KS_LOG_DEBUG, "Session Added: %s\n", key);
@@ -213,6 +207,7 @@ KS_DECLARE(ks_status_t) blade_sessionmgr_session_remove(blade_sessionmgr_t *bsmg
KS_DECLARE(ks_status_t) blade_sessionmgr_callback_add(blade_sessionmgr_t *bsmgr, void *data, blade_session_callback_t callback, const char **id)
{
ks_pool_t *pool = NULL;
blade_session_callback_data_t *bscd = NULL;
uuid_t uuid;
@@ -220,17 +215,18 @@ KS_DECLARE(ks_status_t) blade_sessionmgr_callback_add(blade_sessionmgr_t *bsmgr,
ks_assert(callback);
ks_assert(id);
pool = ks_pool_get(bsmgr);
ks_uuid(&uuid);
bscd = ks_pool_alloc(bsmgr->pool, sizeof(blade_session_callback_data_t));
bscd->pool = bsmgr->pool;
bscd->id = ks_uuid_str(bsmgr->pool, &uuid);
bscd = ks_pool_alloc(pool, sizeof(blade_session_callback_data_t));
bscd->id = ks_uuid_str(pool, &uuid);
bscd->data = data;
bscd->callback = callback;
ks_pool_set_cleanup(bsmgr->pool, bscd, NULL, blade_session_callback_data_cleanup);
ks_pool_set_cleanup(bscd, NULL, blade_session_callback_data_cleanup);
ks_hash_insert(bsmgr->callbacks, (void *)ks_pstrdup(bscd->pool, bscd->id), bscd);
ks_hash_insert(bsmgr->callbacks, (void *)ks_pstrdup(pool, bscd->id), bscd);
*id = bscd->id;
libs/libblade/src/blade_stack.c
+29 -42
View File
@@ -34,7 +34,6 @@
#include "blade.h"
struct blade_handle_s {
ks_pool_t *pool;
ks_thread_pool_t *tpool;
blade_transportmgr_t *transportmgr;
@@ -57,7 +56,7 @@ ks_bool_t blade_rpcsubscribe_response_handler(blade_rpc_response_t *brpcres, voi
ks_bool_t blade_rpcbroadcast_request_handler(blade_rpc_request_t *brpcreq, void *data);
static void blade_handle_cleanup(ks_pool_t *pool, void *ptr, void *arg, ks_pool_cleanup_action_t action, ks_pool_cleanup_type_t type)
static void blade_handle_cleanup(void *ptr, void *arg, ks_pool_cleanup_action_t action, ks_pool_cleanup_type_t type)
{
blade_handle_t *bh = (blade_handle_t *)ptr;
@@ -98,7 +97,6 @@ KS_DECLARE(ks_status_t) blade_handle_create(blade_handle_t **bhP)
ks_assert(tpool);
bh = ks_pool_alloc(pool, sizeof(blade_handle_t));
bh->pool = pool;
bh->tpool = tpool;
blade_transportmgr_create(&bh->transportmgr, bh);
@@ -126,7 +124,7 @@ KS_DECLARE(ks_status_t) blade_handle_create(blade_handle_t **bhP)
ks_assert(bh->sessionmgr);
ks_pool_set_cleanup(pool, bh, NULL, blade_handle_cleanup);
ks_pool_set_cleanup(bh, NULL, blade_handle_cleanup);
*bhP = bh;
@@ -141,18 +139,17 @@ KS_DECLARE(ks_status_t) blade_handle_destroy(blade_handle_t **bhP)
ks_pool_t *pool;
ks_assert(bhP);
ks_assert(*bhP);
bh = *bhP;
*bhP = NULL;
ks_assert(bh);
pool = bh->pool;
// shutdown cannot happen inside of the cleanup callback because it'll lock a mutex for the pool during cleanup callbacks which connections and sessions need to finish their cleanup
// and more importantly, memory needs to remain intact until shutdown is completed to avoid various things hitting teardown before shutdown runs
blade_handle_shutdown(bh);
pool = ks_pool_get(bh);
ks_pool_close(&pool);
return KS_STATUS_SUCCESS;
@@ -261,12 +258,6 @@ KS_DECLARE(ks_status_t) blade_handle_shutdown(blade_handle_t *bh)
return KS_STATUS_SUCCESS;
}
KS_DECLARE(ks_pool_t *) blade_handle_pool_get(blade_handle_t *bh)
{
ks_assert(bh);
return bh->pool;
}
KS_DECLARE(ks_thread_pool_t *) blade_handle_tpool_get(blade_handle_t *bh)
{
ks_assert(bh);
@@ -354,7 +345,6 @@ KS_DECLARE(ks_status_t) blade_handle_connect(blade_handle_t *bh, blade_connectio
typedef struct blade_rpcsubscribe_data_s blade_rpcsubscribe_data_t;
struct blade_rpcsubscribe_data_s {
ks_pool_t *pool;
blade_rpc_response_callback_t original_callback;
void *original_data;
blade_rpc_request_callback_t channel_callback;
@@ -381,7 +371,7 @@ KS_DECLARE(ks_status_t) blade_handle_rpcregister(blade_handle_t *bh, const char
goto done;
}
pool = blade_handle_pool_get(bh);
pool = ks_pool_get(bh);
ks_assert(pool);
blade_rpc_request_raw_create(pool, &req, &req_params, NULL, "blade.register");
@@ -485,7 +475,7 @@ KS_DECLARE(ks_status_t) blade_handle_rpcpublish(blade_handle_t *bh, const char *
goto done;
}
pool = blade_handle_pool_get(bh);
pool = ks_pool_get(bh);
ks_assert(pool);
blade_rpc_request_raw_create(pool, &req, &req_params, NULL, "blade.publish");
@@ -498,13 +488,13 @@ KS_DECLARE(ks_status_t) blade_handle_rpcpublish(blade_handle_t *bh, const char *
ks_assert(id);
cJSON_AddStringToObject(req_params, "requester-nodeid", id);
ks_pool_free(pool, &id);
ks_pool_free(&id);
blade_upstreammgr_masterid_copy(bh->upstreammgr, pool, &id);
ks_assert(id);
cJSON_AddStringToObject(req_params, "responder-nodeid", id);
ks_pool_free(pool, &id);
ks_pool_free(&id);
// @todo may want to switch this system to use a blade_rpcpublish_args_t with validation on the contents on this list internally
// and to produce the entire json block internally in case the channel args change to include additional information like an encryption key,
@@ -677,7 +667,7 @@ KS_DECLARE(ks_status_t) blade_handle_rpcauthorize(blade_handle_t *bh, const char
goto done;
}
pool = blade_handle_pool_get(bh);
pool = ks_pool_get(bh);
ks_assert(pool);
blade_rpc_request_raw_create(pool, &req, &req_params, NULL, "blade.authorize");
@@ -692,13 +682,13 @@ KS_DECLARE(ks_status_t) blade_handle_rpcauthorize(blade_handle_t *bh, const char
ks_assert(id);
cJSON_AddStringToObject(req_params, "requester-nodeid", id);
ks_pool_free(pool, &id);
ks_pool_free(&id);
blade_upstreammgr_masterid_copy(bh->upstreammgr, pool, &id);
ks_assert(id);
cJSON_AddStringToObject(req_params, "responder-nodeid", id);
ks_pool_free(pool, &id);
ks_pool_free(&id);
cJSON_AddItemToObject(req_params, "channels", cJSON_Duplicate(channels, 1));
@@ -899,7 +889,7 @@ KS_DECLARE(ks_status_t) blade_handle_rpclocate(blade_handle_t *bh, const char *p
goto done;
}
pool = blade_handle_pool_get(bh);
pool = ks_pool_get(bh);
ks_assert(pool);
blade_rpc_request_raw_create(pool, &req, &req_params, NULL, "blade.locate");
@@ -912,13 +902,13 @@ KS_DECLARE(ks_status_t) blade_handle_rpclocate(blade_handle_t *bh, const char *p
ks_assert(id);
cJSON_AddStringToObject(req_params, "requester-nodeid", id);
ks_pool_free(pool, &id);
ks_pool_free(&id);
blade_upstreammgr_masterid_copy(bh->upstreammgr, pool, &id);
ks_assert(id);
cJSON_AddStringToObject(req_params, "responder-nodeid", id);
ks_pool_free(pool, &id);
ks_pool_free(&id);
ks_log(KS_LOG_DEBUG, "Session (%s) locate request started\n", blade_session_id_get(bs));
@@ -1057,7 +1047,7 @@ KS_DECLARE(ks_status_t) blade_handle_rpcexecute(blade_handle_t *bh, const char *
}
}
pool = blade_handle_pool_get(bh);
pool = ks_pool_get(bh);
ks_assert(pool);
blade_rpc_request_raw_create(pool, &req, &req_params, NULL, "blade.execute");
@@ -1070,7 +1060,7 @@ KS_DECLARE(ks_status_t) blade_handle_rpcexecute(blade_handle_t *bh, const char *
ks_assert(localid);
cJSON_AddStringToObject(req_params, "requester-nodeid", localid);
ks_pool_free(pool, &localid);
ks_pool_free(&localid);
cJSON_AddStringToObject(req_params, "responder-nodeid", nodeid);
@@ -1317,7 +1307,7 @@ KS_DECLARE(void) blade_rpcexecute_response_send(blade_rpc_request_t *brpcreq, cJ
}
static void blade_rpcsubscribe_data_cleanup(ks_pool_t *pool, void *ptr, void *arg, ks_pool_cleanup_action_t action, ks_pool_cleanup_type_t type)
static void blade_rpcsubscribe_data_cleanup(void *ptr, void *arg, ks_pool_cleanup_action_t action, ks_pool_cleanup_type_t type)
{
blade_rpcsubscribe_data_t *brpcsd = (blade_rpcsubscribe_data_t *)ptr;
@@ -1327,7 +1317,7 @@ static void blade_rpcsubscribe_data_cleanup(ks_pool_t *pool, void *ptr, void *ar
case KS_MPCL_ANNOUNCE:
break;
case KS_MPCL_TEARDOWN:
if (brpcsd->relayed_messageid) ks_pool_free(brpcsd->pool, &brpcsd->relayed_messageid);
if (brpcsd->relayed_messageid) ks_pool_free(&brpcsd->relayed_messageid);
break;
case KS_MPCL_DESTROY:
break;
@@ -1348,7 +1338,7 @@ KS_DECLARE(ks_status_t) blade_handle_rpcsubscribe(blade_handle_t *bh, const char
ks_assert(realm);
ks_assert(subscribe_channels || unsubscribe_channels);
pool = blade_handle_pool_get(bh);
pool = ks_pool_get(bh);
ks_assert(pool);
// @note this is always produced by a subscriber, and sent upstream, master will only use the internal raw call
@@ -1357,22 +1347,21 @@ KS_DECLARE(ks_status_t) blade_handle_rpcsubscribe(blade_handle_t *bh, const char
goto done;
}
blade_upstreammgr_localid_copy(bh->upstreammgr, bh->pool, &localid);
blade_upstreammgr_localid_copy(bh->upstreammgr, ks_pool_get(bh), &localid);
ks_assert(localid);
// @note since this is allocated in the handle's pool, if the handle is shutdown during a pending request, then the data
// memory will be cleaned up with the handle, otherwise should be cleaned up in the response callback
temp_data = (blade_rpcsubscribe_data_t *)ks_pool_alloc(pool, sizeof(blade_rpcsubscribe_data_t));
temp_data->pool = pool;
temp_data->original_callback = callback;
temp_data->original_data = data;
temp_data->channel_callback = channel_callback;
temp_data->channel_data = channel_data;
ks_pool_set_cleanup(pool, temp_data, NULL, blade_rpcsubscribe_data_cleanup);
ks_pool_set_cleanup(temp_data, NULL, blade_rpcsubscribe_data_cleanup);
ret = blade_handle_rpcsubscribe_raw(bh, protocol, realm, subscribe_channels, unsubscribe_channels, localid, KS_FALSE, blade_rpcsubscribe_response_handler, temp_data);
ks_pool_free(bh->pool, &localid);
ks_pool_free(&localid);
done:
if (bs) blade_session_read_unlock(bs);
@@ -1410,7 +1399,7 @@ ks_status_t blade_handle_rpcsubscribe_raw(blade_handle_t *bh, const char *protoc
goto done;
}
pool = blade_handle_pool_get(bh);
pool = ks_pool_get(bh);
ks_assert(pool);
if (unsubscribe_channels) {
@@ -1467,7 +1456,7 @@ ks_bool_t blade_rpcsubscribe_request_handler(blade_rpc_request_t *brpcreq, void
bh = blade_rpc_request_handle_get(brpcreq);
ks_assert(bh);
pool = blade_handle_pool_get(bh);
pool = ks_pool_get(bh);
ks_assert(pool);
bs = blade_sessionmgr_session_lookup(blade_handle_sessionmgr_get(bh), blade_rpc_request_sessionid_get(brpcreq));
@@ -1571,9 +1560,8 @@ ks_bool_t blade_rpcsubscribe_request_handler(blade_rpc_request_t *brpcreq, void
blade_session_send(bs, res, NULL, NULL);
} else {
blade_rpcsubscribe_data_t *temp_data = (blade_rpcsubscribe_data_t *)ks_pool_alloc(pool, sizeof(blade_rpcsubscribe_data_t));
temp_data->pool = pool;
temp_data->relayed_messageid = ks_pstrdup(pool, blade_rpc_request_messageid_get(brpcreq));
ks_pool_set_cleanup(pool, temp_data, NULL, blade_rpcsubscribe_data_cleanup);
ks_pool_set_cleanup(temp_data, NULL, blade_rpcsubscribe_data_cleanup);
blade_handle_rpcsubscribe_raw(bh, req_params_protocol, req_params_realm, req_params_subscribe_channels, req_params_unsubscribe_channels, req_params_subscriber_nodeid, downstream, blade_rpcsubscribe_response_handler, temp_data);
}
@@ -1604,7 +1592,6 @@ ks_bool_t blade_rpcsubscribe_response_handler(blade_rpc_response_t *brpcres, voi
cJSON *res_result_failed_channels = NULL;
ks_assert(brpcres);
ks_assert(data);
bh = blade_rpc_response_handle_get(brpcres);
ks_assert(bh);
@@ -1695,7 +1682,7 @@ ks_bool_t blade_rpcsubscribe_response_handler(blade_rpc_response_t *brpcres, voi
}
done:
if (temp_data) ks_pool_free(temp_data->pool, &temp_data);
if (temp_data) ks_pool_free(&temp_data);
blade_session_read_unlock(bs);
return ret;
}
@@ -1798,7 +1785,7 @@ ks_bool_t blade_rpcbroadcast_request_handler(blade_rpc_request_t *brpcreq, void
const char *localid = NULL;
ks_pool_t *pool = NULL;
pool = blade_handle_pool_get(bh);
pool = ks_pool_get(bh);
blade_upstreammgr_localid_copy(bh->upstreammgr, pool, &localid);
ks_assert(localid);
@@ -1807,7 +1794,7 @@ ks_bool_t blade_rpcbroadcast_request_handler(blade_rpc_request_t *brpcreq, void
callback = blade_subscription_callback_get(bsub);
if (callback) ret = callback(brpcreq, blade_subscription_callback_data_get(bsub));
}
ks_pool_free(pool, &localid);
ks_pool_free(&localid);
}
// build the actual response finally
libs/libblade/src/blade_subscription.c
+8 -11
View File
@@ -34,8 +34,6 @@
#include "blade.h"
struct blade_subscription_s {
ks_pool_t *pool;
const char *protocol;
const char *realm;
const char *channel;
@@ -46,7 +44,7 @@ struct blade_subscription_s {
};
static void blade_subscription_cleanup(ks_pool_t *pool, void *ptr, void *arg, ks_pool_cleanup_action_t action, ks_pool_cleanup_type_t type)
static void blade_subscription_cleanup(void *ptr, void *arg, ks_pool_cleanup_action_t action, ks_pool_cleanup_type_t type)
{
blade_subscription_t *bsub = (blade_subscription_t *)ptr;
@@ -56,9 +54,9 @@ static void blade_subscription_cleanup(ks_pool_t *pool, void *ptr, void *arg, ks
case KS_MPCL_ANNOUNCE:
break;
case KS_MPCL_TEARDOWN:
if (bsub->protocol) ks_pool_free(bsub->pool, &bsub->protocol);
if (bsub->realm) ks_pool_free(bsub->pool, &bsub->subscribers);
if (bsub->channel) ks_pool_free(bsub->pool, &bsub->channel);
if (bsub->protocol) ks_pool_free(&bsub->protocol);
if (bsub->realm) ks_pool_free(&bsub->subscribers);
if (bsub->channel) ks_pool_free(&bsub->channel);
if (bsub->subscribers) ks_hash_destroy(&bsub->subscribers);
break;
case KS_MPCL_DESTROY:
@@ -77,15 +75,14 @@ KS_DECLARE(ks_status_t) blade_subscription_create(blade_subscription_t **bsubP,
ks_assert(channel);
bsub = ks_pool_alloc(pool, sizeof(blade_subscription_t));
bsub->pool = pool;
bsub->protocol = ks_pstrdup(pool, protocol);
bsub->realm = ks_pstrdup(pool, realm);
bsub->channel = ks_pstrdup(pool, channel);
ks_hash_create(&bsub->subscribers, KS_HASH_MODE_CASE_INSENSITIVE, KS_HASH_FLAG_NOLOCK | KS_HASH_FLAG_DUP_CHECK | KS_HASH_FLAG_FREE_KEY, bsub->pool);
ks_hash_create(&bsub->subscribers, KS_HASH_MODE_CASE_INSENSITIVE, KS_HASH_FLAG_NOLOCK | KS_HASH_FLAG_DUP_CHECK | KS_HASH_FLAG_FREE_KEY, pool);
ks_assert(bsub->subscribers);
ks_pool_set_cleanup(pool, bsub, NULL, blade_subscription_cleanup);
ks_pool_set_cleanup(bsub, NULL, blade_subscription_cleanup);
*bsubP = bsub;
@@ -101,7 +98,7 @@ KS_DECLARE(ks_status_t) blade_subscription_destroy(blade_subscription_t **bsubP)
bsub = *bsubP;
ks_pool_free(bsub->pool, bsubP);
ks_pool_free(bsubP);
return KS_STATUS_SUCCESS;
}
@@ -145,7 +142,7 @@ KS_DECLARE(ks_status_t) blade_subscription_subscribers_add(blade_subscription_t
ks_assert(bsub);
ks_assert(nodeid);
key = ks_pstrdup(bsub->pool, nodeid);
key = ks_pstrdup(ks_pool_get(bsub), nodeid);
ks_hash_insert(bsub->subscribers, (void *)key, (void *)KS_TRUE);
return KS_STATUS_SUCCESS;
libs/libblade/src/blade_subscriptionmgr.c
+46 -40
View File
@@ -35,7 +35,6 @@
struct blade_subscriptionmgr_s {
blade_handle_t *handle;
ks_pool_t *pool;
ks_hash_t *subscriptions; // key, blade_subscription_t*
ks_hash_t *subscriptions_cleanup; // target, ks_hash_t*
@@ -43,7 +42,7 @@ struct blade_subscriptionmgr_s {
};
static void blade_subscriptionmgr_cleanup(ks_pool_t *pool, void *ptr, void *arg, ks_pool_cleanup_action_t action, ks_pool_cleanup_type_t type)
static void blade_subscriptionmgr_cleanup(void *ptr, void *arg, ks_pool_cleanup_action_t action, ks_pool_cleanup_type_t type)
{
//blade_routemgr_t *brmgr = (blade_routemgr_t *)ptr;
@@ -71,16 +70,15 @@ KS_DECLARE(ks_status_t) blade_subscriptionmgr_create(blade_subscriptionmgr_t **b
bsmgr = ks_pool_alloc(pool, sizeof(blade_subscriptionmgr_t));
bsmgr->handle = bh;
bsmgr->pool = pool;
// @note can let removes free keys and values for subscriptions, both are allocated from the same pool as the hash itself
ks_hash_create(&bsmgr->subscriptions, KS_HASH_MODE_CASE_INSENSITIVE, KS_HASH_FLAG_RWLOCK | KS_HASH_FLAG_DUP_CHECK | KS_HASH_FLAG_FREE_KEY | KS_HASH_FLAG_FREE_VALUE, bsmgr->pool);
ks_hash_create(&bsmgr->subscriptions, KS_HASH_MODE_CASE_INSENSITIVE, KS_HASH_FLAG_RWLOCK | KS_HASH_FLAG_DUP_CHECK | KS_HASH_FLAG_FREE_KEY | KS_HASH_FLAG_FREE_VALUE, pool);
ks_assert(bsmgr->subscriptions);
ks_hash_create(&bsmgr->subscriptions_cleanup, KS_HASH_MODE_CASE_INSENSITIVE, KS_HASH_FLAG_RWLOCK | KS_HASH_FLAG_DUP_CHECK | KS_HASH_FLAG_FREE_KEY | KS_HASH_FLAG_FREE_VALUE, bsmgr->pool);
ks_hash_create(&bsmgr->subscriptions_cleanup, KS_HASH_MODE_CASE_INSENSITIVE, KS_HASH_FLAG_RWLOCK | KS_HASH_FLAG_DUP_CHECK | KS_HASH_FLAG_FREE_KEY | KS_HASH_FLAG_FREE_VALUE, pool);
ks_assert(bsmgr->subscriptions_cleanup);
ks_pool_set_cleanup(pool, bsmgr, NULL, blade_subscriptionmgr_cleanup);
ks_pool_set_cleanup(bsmgr, NULL, blade_subscriptionmgr_cleanup);
*bsmgrP = bsmgr;
@@ -98,9 +96,7 @@ KS_DECLARE(ks_status_t) blade_subscriptionmgr_destroy(blade_subscriptionmgr_t **
bsmgr = *bsmgrP;
*bsmgrP = NULL;
ks_assert(bsmgr);
pool = bsmgr->pool;
pool = ks_pool_get(bsmgr);
ks_pool_close(&pool);
@@ -138,19 +134,20 @@ KS_DECLARE(blade_subscription_t *) blade_subscriptionmgr_subscription_lookup(bla
ks_assert(realm);
ks_assert(channel);
key = ks_psprintf(bsmgr->pool, "%s@%s/%s", protocol, realm, channel);
key = ks_psprintf(ks_pool_get(bsmgr), "%s@%s/%s", protocol, realm, channel);
bsub = (blade_subscription_t *)ks_hash_search(bsmgr->subscriptions, (void *)key, KS_READLOCKED);
// @todo if (bsub) blade_subscription_read_lock(bsub);
ks_hash_read_unlock(bsmgr->subscriptions);
ks_pool_free(bsmgr->pool, &key);
ks_pool_free(&key);
return bsub;
}
KS_DECLARE(ks_bool_t) blade_subscriptionmgr_subscriber_add(blade_subscriptionmgr_t *bsmgr, blade_subscription_t **bsubP, const char *protocol, const char *realm, const char *channel, const char *subscriber)
{
ks_pool_t *pool = NULL;
char *key = NULL;
blade_subscription_t *bsub = NULL;
ks_hash_t *bsub_cleanup = NULL;
@@ -162,29 +159,31 @@ KS_DECLARE(ks_bool_t) blade_subscriptionmgr_subscriber_add(blade_subscriptionmgr
ks_assert(channel);
ks_assert(subscriber);
key = ks_psprintf(bsmgr->pool, "%s@%s/%s", protocol, realm, channel);
pool = ks_pool_get(bsmgr);
key = ks_psprintf(pool, "%s@%s/%s", protocol, realm, channel);
ks_hash_write_lock(bsmgr->subscriptions);
bsub = (blade_subscription_t *)ks_hash_search(bsmgr->subscriptions, (void *)key, KS_UNLOCKED);
if (!bsub) {
blade_subscription_create(&bsub, bsmgr->pool, protocol, realm, channel);
blade_subscription_create(&bsub, pool, protocol, realm, channel);
ks_assert(bsub);
ks_hash_insert(bsmgr->subscriptions, (void *)ks_pstrdup(bsmgr->pool, key), bsub);
ks_hash_insert(bsmgr->subscriptions, (void *)ks_pstrdup(pool, key), bsub);
propagate = KS_TRUE;
}
bsub_cleanup = (ks_hash_t *)ks_hash_search(bsmgr->subscriptions_cleanup, (void *)subscriber, KS_UNLOCKED);
if (!bsub_cleanup) {
ks_hash_create(&bsub_cleanup, KS_HASH_MODE_CASE_INSENSITIVE, KS_HASH_FLAG_RWLOCK | KS_HASH_FLAG_DUP_CHECK | KS_HASH_FLAG_FREE_KEY, bsmgr->pool);
ks_hash_create(&bsub_cleanup, KS_HASH_MODE_CASE_INSENSITIVE, KS_HASH_FLAG_RWLOCK | KS_HASH_FLAG_DUP_CHECK | KS_HASH_FLAG_FREE_KEY, pool);
ks_assert(bsub_cleanup);
ks_log(KS_LOG_DEBUG, "Subscription Added: %s\n", key);
ks_hash_insert(bsmgr->subscriptions_cleanup, (void *)ks_pstrdup(bsmgr->pool, subscriber), (void *)bsub_cleanup);
ks_hash_insert(bsmgr->subscriptions_cleanup, (void *)ks_pstrdup(pool, subscriber), (void *)bsub_cleanup);
}
ks_hash_insert(bsub_cleanup, (void *)ks_pstrdup(bsmgr->pool, key), (void *)KS_TRUE);
ks_hash_insert(bsub_cleanup, (void *)ks_pstrdup(pool, key), (void *)KS_TRUE);
blade_subscription_subscribers_add(bsub, subscriber);
@@ -192,7 +191,7 @@ KS_DECLARE(ks_bool_t) blade_subscriptionmgr_subscriber_add(blade_subscriptionmgr
ks_log(KS_LOG_DEBUG, "Subscriber Added: %s to %s\n", subscriber, key);
ks_pool_free(bsmgr->pool, &key);
ks_pool_free(&key);
if (bsubP) *bsubP = bsub;
@@ -212,7 +211,7 @@ KS_DECLARE(ks_bool_t) blade_subscriptionmgr_subscriber_remove(blade_subscription
ks_assert(channel);
ks_assert(subscriber);
key = ks_psprintf(bsmgr->pool, "%s@%s/%s", protocol, realm, channel);
key = ks_psprintf(ks_pool_get(bsmgr), "%s@%s/%s", protocol, realm, channel);
ks_hash_write_lock(bsmgr->subscriptions);
@@ -220,26 +219,27 @@ KS_DECLARE(ks_bool_t) blade_subscriptionmgr_subscriber_remove(blade_subscription
if (bsub) {
bsub_cleanup = (ks_hash_t *)ks_hash_search(bsmgr->subscriptions_cleanup, (void *)subscriber, KS_UNLOCKED);
ks_assert(bsub_cleanup);
ks_hash_remove(bsub_cleanup, key);
if (bsub_cleanup) {
ks_hash_remove(bsub_cleanup, key);
if (ks_hash_count(bsub_cleanup) == 0) {
ks_hash_remove(bsmgr->subscriptions_cleanup, (void *)subscriber);
}
if (ks_hash_count(bsub_cleanup) == 0) {
ks_hash_remove(bsmgr->subscriptions_cleanup, (void *)subscriber);
}
ks_log(KS_LOG_DEBUG, "Subscriber Removed: %s from %s\n", subscriber, key);
blade_subscription_subscribers_remove(bsub, subscriber);
ks_log(KS_LOG_DEBUG, "Subscriber Removed: %s from %s\n", subscriber, key);
blade_subscription_subscribers_remove(bsub, subscriber);
if (ks_hash_count(blade_subscription_subscribers_get(bsub)) == 0) {
ks_log(KS_LOG_DEBUG, "Subscription Removed: %s\n", key);
ks_hash_remove(bsmgr->subscriptions, (void *)key);
propagate = KS_TRUE;
if (ks_hash_count(blade_subscription_subscribers_get(bsub)) == 0) {
ks_log(KS_LOG_DEBUG, "Subscription Removed: %s\n", key);
ks_hash_remove(bsmgr->subscriptions, (void *)key);
propagate = KS_TRUE;
}
}
}
ks_hash_write_unlock(bsmgr->subscriptions);
ks_pool_free(bsmgr->pool, &key);
ks_pool_free(&key);
if (bsubP) *bsubP = bsub;
@@ -248,11 +248,14 @@ KS_DECLARE(ks_bool_t) blade_subscriptionmgr_subscriber_remove(blade_subscription
KS_DECLARE(void) blade_subscriptionmgr_purge(blade_subscriptionmgr_t *bsmgr, const char *target)
{
ks_pool_t *pool = NULL;
ks_bool_t unsubbed = KS_FALSE;
ks_assert(bsmgr);
ks_assert(target);
pool = ks_pool_get(bsmgr);
while (!unsubbed) {
ks_hash_t *subscriptions = NULL;
const char *protocol = NULL;
@@ -276,24 +279,25 @@ KS_DECLARE(void) blade_subscriptionmgr_purge(blade_subscriptionmgr_t *bsmgr, con
ks_assert(bsub);
// @note allocate these to avoid lifecycle issues when the last subscriber is removed causing the subscription to be removed
protocol = ks_pstrdup(bsmgr->pool, blade_subscription_protocol_get(bsub));
realm = ks_pstrdup(bsmgr->pool, blade_subscription_realm_get(bsub));
channel = ks_pstrdup(bsmgr->pool, blade_subscription_channel_get(bsub));
protocol = ks_pstrdup(pool, blade_subscription_protocol_get(bsub));
realm = ks_pstrdup(pool, blade_subscription_realm_get(bsub));
channel = ks_pstrdup(pool, blade_subscription_channel_get(bsub));
}
ks_hash_read_unlock(bsmgr->subscriptions);
if (!unsubbed) {
blade_subscriptionmgr_subscriber_remove(bsmgr, NULL, protocol, realm, channel, target);
ks_pool_free(bsmgr->pool, &protocol);
ks_pool_free(bsmgr->pool, &realm);
ks_pool_free(bsmgr->pool, &channel);
ks_pool_free(&protocol);
ks_pool_free(&realm);
ks_pool_free(&channel);
}
}
}
KS_DECLARE(ks_status_t) blade_subscriptionmgr_broadcast(blade_subscriptionmgr_t *bsmgr, const char *excluded_nodeid, const char *protocol, const char *realm, const char *channel, const char *event, cJSON *params, blade_rpc_response_callback_t callback, void *data)
{
ks_pool_t *pool = NULL;
const char *bsub_key = NULL;
blade_subscription_t *bsub = NULL;
blade_session_t *bs = NULL;
@@ -305,9 +309,11 @@ KS_DECLARE(ks_status_t) blade_subscriptionmgr_broadcast(blade_subscriptionmgr_t
ks_assert(realm);
ks_assert(channel);
bsub_key = ks_psprintf(bsmgr->pool, "%s@%s/%s", protocol, realm, channel);
pool = ks_pool_get(bsmgr);
blade_rpc_request_raw_create(bsmgr->pool, &req, &req_params, NULL, "blade.broadcast");
bsub_key = ks_psprintf(pool, "%s@%s/%s", protocol, realm, channel);
blade_rpc_request_raw_create(pool, &req, &req_params, NULL, "blade.broadcast");
cJSON_AddStringToObject(req_params, "protocol", protocol);
cJSON_AddStringToObject(req_params, "realm", realm);
cJSON_AddStringToObject(req_params, "channel", channel);
@@ -359,7 +365,7 @@ KS_DECLARE(ks_status_t) blade_subscriptionmgr_broadcast(blade_subscriptionmgr_t
cJSON_Delete(req);
ks_pool_free(bsmgr->pool, &bsub_key);
ks_pool_free(&bsub_key);
return KS_STATUS_SUCCESS;
}
libs/libblade/src/blade_transport.c
+4 -7
View File
@@ -35,7 +35,6 @@
struct blade_transport_s {
blade_handle_t *handle;
ks_pool_t *pool;
const char *name;
void *data;
@@ -43,7 +42,7 @@ struct blade_transport_s {
};
static void blade_transport_cleanup(ks_pool_t *pool, void *ptr, void *arg, ks_pool_cleanup_action_t action, ks_pool_cleanup_type_t type)
static void blade_transport_cleanup(void *ptr, void *arg, ks_pool_cleanup_action_t action, ks_pool_cleanup_type_t type)
{
//blade_transport_t *bt = (blade_transport_t *)ptr;
@@ -71,12 +70,11 @@ KS_DECLARE(ks_status_t) blade_transport_create(blade_transport_t **btP, blade_ha
bt = ks_pool_alloc(pool, sizeof(blade_transport_t));
bt->handle = bh;
bt->pool = pool;
bt->name = ks_pstrdup(pool, name);
bt->data = data;
bt->callbacks = callbacks;
ks_pool_set_cleanup(pool, bt, NULL, blade_transport_cleanup);
ks_pool_set_cleanup(bt, NULL, blade_transport_cleanup);
ks_log(KS_LOG_DEBUG, "Created transport %s\n", name);
@@ -94,13 +92,12 @@ KS_DECLARE(ks_status_t) blade_transport_destroy(blade_transport_t **btP)
ks_assert(*btP);
bt = *btP;
*btP = NULL;
pool = bt->pool;
pool = ks_pool_get(bt);
ks_pool_close(&pool);
*btP = NULL;
return KS_STATUS_SUCCESS;
}
libs/libblade/src/blade_transport_wss.c
+14 -19
View File
@@ -41,7 +41,6 @@ typedef struct blade_transport_wss_link_s blade_transport_wss_link_t;
struct blade_transport_wss_s {
blade_handle_t *handle;
ks_pool_t *pool;
blade_transport_t *transport;
blade_transport_callbacks_t *callbacks;
@@ -59,7 +58,6 @@ struct blade_transport_wss_s {
struct blade_transport_wss_link_s {
blade_transport_wss_t *transport;
ks_pool_t *pool;
const char *session_id;
ks_socket_t sock;
@@ -109,7 +107,7 @@ static blade_transport_callbacks_t g_transport_wss_callbacks =
};
static void blade_transport_wss_cleanup(ks_pool_t *pool, void *ptr, void *arg, ks_pool_cleanup_action_t action, ks_pool_cleanup_type_t type)
static void blade_transport_wss_cleanup(void *ptr, void *arg, ks_pool_cleanup_action_t action, ks_pool_cleanup_type_t type)
{
//blade_transport_wss_t *btwss = (blade_transport_wss_t *)ptr;
@@ -138,12 +136,11 @@ KS_DECLARE(ks_status_t) blade_transport_wss_create(blade_transport_t **btP, blad
btwss = ks_pool_alloc(pool, sizeof(blade_transport_wss_t));
btwss->handle = bh;
btwss->pool = pool;
blade_transport_create(&btwss->transport, bh, pool, BLADE_MODULE_WSS_TRANSPORT_NAME, btwss, &g_transport_wss_callbacks);
btwss->callbacks = &g_transport_wss_callbacks;
ks_pool_set_cleanup(pool, btwss, NULL, blade_transport_wss_cleanup);
ks_pool_set_cleanup(btwss, NULL, blade_transport_wss_cleanup);
ks_log(KS_LOG_DEBUG, "Created\n");
@@ -152,7 +149,7 @@ KS_DECLARE(ks_status_t) blade_transport_wss_create(blade_transport_t **btP, blad
return KS_STATUS_SUCCESS;
}
static void blade_transport_wss_link_cleanup(ks_pool_t *pool, void *ptr, void *arg, ks_pool_cleanup_action_t action, ks_pool_cleanup_type_t type)
static void blade_transport_wss_link_cleanup(void *ptr, void *arg, ks_pool_cleanup_action_t action, ks_pool_cleanup_type_t type)
{
blade_transport_wss_link_t *btwssl = (blade_transport_wss_link_t *)ptr;
@@ -162,7 +159,7 @@ static void blade_transport_wss_link_cleanup(ks_pool_t *pool, void *ptr, void *a
case KS_MPCL_ANNOUNCE:
break;
case KS_MPCL_TEARDOWN:
if (btwssl->session_id) ks_pool_free(btwssl->pool, &btwssl->session_id);
if (btwssl->session_id) ks_pool_free(&btwssl->session_id);
if (btwssl->kws) kws_destroy(&btwssl->kws);
else ks_socket_close(&btwssl->sock);
break;
@@ -182,11 +179,10 @@ ks_status_t blade_transport_wss_link_create(blade_transport_wss_link_t **btwsslP
btwssl = ks_pool_alloc(pool, sizeof(blade_transport_wss_link_t));
btwssl->transport = btwss;
btwssl->pool = pool;
btwssl->sock = sock;
if (session_id) btwssl->session_id = ks_pstrdup(pool, session_id);
ks_pool_set_cleanup(pool, btwssl, NULL, blade_transport_wss_link_cleanup);
ks_pool_set_cleanup(btwssl, NULL, blade_transport_wss_link_cleanup);
ks_log(KS_LOG_DEBUG, "Created\n");
@@ -403,9 +399,8 @@ ks_status_t blade_transport_wss_listen(blade_transport_wss_t *btwss, ks_sockaddr
}
listener_index = btwss->listeners_count++;
btwss->listeners_poll = (struct pollfd *)ks_pool_resize(btwss->pool,
btwss->listeners_poll,
sizeof(struct pollfd) * btwss->listeners_count);
if (!btwss->listeners_poll) btwss->listeners_poll = (struct pollfd *)ks_pool_alloc(ks_pool_get(btwss), sizeof(struct pollfd) * btwss->listeners_count);
else btwss->listeners_poll = (struct pollfd *)ks_pool_resize(btwss->listeners_poll, sizeof(struct pollfd) * btwss->listeners_count);
ks_assert(btwss->listeners_poll);
btwss->listeners_poll[listener_index].fd = listener;
btwss->listeners_poll[listener_index].events = POLLIN; // | POLLERR;
@@ -461,7 +456,7 @@ void *blade_transport_wss_listeners_thread(ks_thread_t *thread, void *data)
blade_connection_create(&bc, btwss->handle);
ks_assert(bc);
blade_transport_wss_link_create(&btwssl, blade_connection_pool_get(bc), btwss, sock, NULL);
blade_transport_wss_link_create(&btwssl, ks_pool_get(bc), btwss, sock, NULL);
ks_assert(btwssl);
blade_connection_transport_set(bc, btwssl, btwss->callbacks);
@@ -561,7 +556,7 @@ ks_status_t blade_transport_wss_onconnect(blade_connection_t **bcP, blade_transp
blade_connection_create(&bc, btwss->handle);
ks_assert(bc);
blade_transport_wss_link_create(&btwssl, blade_connection_pool_get(bc), btwss, sock, session_id);
blade_transport_wss_link_create(&btwssl, ks_pool_get(bc), btwss, sock, session_id);
ks_assert(btwssl);
blade_connection_transport_set(bc, btwssl, btwss->callbacks);
@@ -745,7 +740,7 @@ blade_connection_state_hook_t blade_transport_wss_onstate_startup_inbound(blade_
btwssl = (blade_transport_wss_link_t *)blade_connection_transport_get(bc);
// @todo: SSL init stuffs based on data from config to pass into kws_init
if (kws_init(&btwssl->kws, btwssl->sock, NULL, NULL, KWS_BLOCK, btwssl->pool) != KS_STATUS_SUCCESS) {
if (kws_init(&btwssl->kws, btwssl->sock, NULL, NULL, KWS_BLOCK, ks_pool_get(btwssl)) != KS_STATUS_SUCCESS) {
ks_log(KS_LOG_DEBUG, "Failed websocket init\n");
ret = BLADE_CONNECTION_STATE_HOOK_DISCONNECT;
goto done;
@@ -858,7 +853,7 @@ blade_connection_state_hook_t blade_transport_wss_onstate_startup_inbound(blade_
cJSON_AddStringToObject(json_result, "nodeid", nodeid);
pool = blade_handle_pool_get(bh);
pool = ks_pool_get(bh);
blade_upstreammgr_masterid_copy(blade_handle_upstreammgr_get(bh), pool, &master_nodeid);
if (!master_nodeid) {
ks_log(KS_LOG_DEBUG, "Master nodeid unavailable\n");
@@ -867,7 +862,7 @@ blade_connection_state_hook_t blade_transport_wss_onstate_startup_inbound(blade_
goto done;
}
cJSON_AddStringToObject(json_result, "master-nodeid", master_nodeid);
ks_pool_free(pool, &master_nodeid);
ks_pool_free(&master_nodeid);
// add the list of actual realms the local node will permit the remote node to register or route, this is the same list that the remote side would be adding to the handle with blade_handle_realm_add()
@@ -942,10 +937,10 @@ blade_connection_state_hook_t blade_transport_wss_onstate_startup_outbound(blade
bh = blade_connection_handle_get(bc);
btwssl = (blade_transport_wss_link_t *)blade_connection_transport_get(bc);
pool = blade_handle_pool_get(bh);
pool = ks_pool_get(bh);
// @todo: SSL init stuffs based on data from config to pass into kws_init
if (kws_init(&btwssl->kws, btwssl->sock, NULL, "/blade:blade.invalid:blade", KWS_BLOCK, btwssl->pool) != KS_STATUS_SUCCESS) {
if (kws_init(&btwssl->kws, btwssl->sock, NULL, "/blade:blade.invalid:blade", KWS_BLOCK, ks_pool_get(btwssl)) != KS_STATUS_SUCCESS) {
ks_log(KS_LOG_DEBUG, "Failed websocket init\n");
ret = BLADE_CONNECTION_STATE_HOOK_DISCONNECT;
goto done;
libs/libblade/src/blade_transportmgr.c
+5 -9
View File
@@ -35,14 +35,13 @@
struct blade_transportmgr_s {
blade_handle_t *handle;
ks_pool_t *pool;
ks_hash_t *transports; // name, blade_transport_t*
blade_transport_t *default_transport; // default wss transport
};
static void blade_transportmgr_cleanup(ks_pool_t *pool, void *ptr, void *arg, ks_pool_cleanup_action_t action, ks_pool_cleanup_type_t type)
static void blade_transportmgr_cleanup(void *ptr, void *arg, ks_pool_cleanup_action_t action, ks_pool_cleanup_type_t type)
{
blade_transportmgr_t *btmgr = (blade_transportmgr_t *)ptr;
ks_hash_iterator_t *it = NULL;
@@ -80,12 +79,11 @@ KS_DECLARE(ks_status_t) blade_transportmgr_create(blade_transportmgr_t **btmgrP,
btmgr = ks_pool_alloc(pool, sizeof(blade_transportmgr_t));
btmgr->handle = bh;
btmgr->pool = pool;
ks_hash_create(&btmgr->transports, KS_HASH_MODE_CASE_INSENSITIVE, KS_HASH_FLAG_RWLOCK | KS_HASH_FLAG_DUP_CHECK | KS_HASH_FLAG_FREE_KEY, btmgr->pool);
ks_hash_create(&btmgr->transports, KS_HASH_MODE_CASE_INSENSITIVE, KS_HASH_FLAG_RWLOCK | KS_HASH_FLAG_DUP_CHECK | KS_HASH_FLAG_FREE_KEY, pool);
ks_assert(btmgr->transports);
ks_pool_set_cleanup(pool, btmgr, NULL, blade_transportmgr_cleanup);
ks_pool_set_cleanup(btmgr, NULL, blade_transportmgr_cleanup);
*btmgrP = btmgr;
@@ -103,9 +101,7 @@ KS_DECLARE(ks_status_t) blade_transportmgr_destroy(blade_transportmgr_t **btmgrP
btmgr = *btmgrP;
*btmgrP = NULL;
ks_assert(btmgr);
pool = btmgr->pool;
pool = ks_pool_get(btmgr);
ks_pool_close(&pool);
@@ -201,7 +197,7 @@ KS_DECLARE(ks_status_t) blade_transportmgr_transport_add(blade_transportmgr_t *b
ks_assert(btmgr);
ks_assert(bt);
key = ks_pstrdup(btmgr->pool, blade_transport_name_get(bt));
key = ks_pstrdup(ks_pool_get(btmgr), blade_transport_name_get(bt));
ks_hash_insert(btmgr->transports, (void *)key, (void *)bt);
ks_log(KS_LOG_DEBUG, "Transport Added: %s\n", key);
libs/libblade/src/blade_tuple.c
+3 -6
View File
@@ -34,14 +34,12 @@
#include "blade.h"
struct blade_tuple_s {
ks_pool_t *pool;
void *value1;
void *value2;
};
static void blade_tuple_cleanup(ks_pool_t *pool, void *ptr, void *arg, ks_pool_cleanup_action_t action, ks_pool_cleanup_type_t type)
static void blade_tuple_cleanup(void *ptr, void *arg, ks_pool_cleanup_action_t action, ks_pool_cleanup_type_t type)
{
//blade_tuple_t *bt = (blade_tuple_t *)ptr;
@@ -65,11 +63,10 @@ KS_DECLARE(ks_status_t) blade_tuple_create(blade_tuple_t **btP, ks_pool_t *pool,
ks_assert(pool);
bt = ks_pool_alloc(pool, sizeof(blade_tuple_t));
bt->pool = pool;
bt->value1 = value1;
bt->value2 = value2;
ks_pool_set_cleanup(pool, bt, NULL, blade_tuple_cleanup);
ks_pool_set_cleanup(bt, NULL, blade_tuple_cleanup);
*btP = bt;
@@ -81,7 +78,7 @@ KS_DECLARE(ks_status_t) blade_tuple_destroy(blade_tuple_t **btP)
ks_assert(btP);
ks_assert(*btP);
ks_pool_free((*btP)->pool, btP);
ks_pool_free(btP);
return KS_STATUS_SUCCESS;
}
libs/libblade/src/blade_upstreammgr.c
+12 -16
View File
@@ -35,7 +35,6 @@
struct blade_upstreammgr_s {
blade_handle_t *handle;
ks_pool_t *pool;
// local node id, can be used to get the upstream session, provided by upstream "blade.connect" response
const char *localid;
@@ -50,7 +49,7 @@ struct blade_upstreammgr_s {
};
static void blade_upstreammgr_cleanup(ks_pool_t *pool, void *ptr, void *arg, ks_pool_cleanup_action_t action, ks_pool_cleanup_type_t type)
static void blade_upstreammgr_cleanup(void *ptr, void *arg, ks_pool_cleanup_action_t action, ks_pool_cleanup_type_t type)
{
//blade_upstreammgr_t *bumgr = (blade_upstreammgr_t *)ptr;
@@ -78,20 +77,19 @@ KS_DECLARE(ks_status_t) blade_upstreammgr_create(blade_upstreammgr_t **bumgrP, b
bumgr = ks_pool_alloc(pool, sizeof(blade_upstreammgr_t));
bumgr->handle = bh;
bumgr->pool = pool;
//ks_hash_create(&bumgr->routes, KS_HASH_MODE_CASE_INSENSITIVE, KS_HASH_FLAG_RWLOCK | KS_HASH_FLAG_DUP_CHECK | KS_HASH_FLAG_FREE_KEY | KS_HASH_FLAG_FREE_VALUE, bumgr->pool);
//ks_hash_create(&bumgr->routes, KS_HASH_MODE_CASE_INSENSITIVE, KS_HASH_FLAG_RWLOCK | KS_HASH_FLAG_DUP_CHECK | KS_HASH_FLAG_FREE_KEY | KS_HASH_FLAG_FREE_VALUE, pool);
//ks_assert(bumgr->routes);
ks_rwl_create(&bumgr->localid_rwl, bumgr->pool);
ks_rwl_create(&bumgr->localid_rwl, pool);
ks_assert(bumgr->localid_rwl);
ks_rwl_create(&bumgr->masterid_rwl, bumgr->pool);
ks_rwl_create(&bumgr->masterid_rwl, pool);
ks_assert(bumgr->masterid_rwl);
ks_hash_create(&bumgr->realms, KS_HASH_MODE_CASE_INSENSITIVE, KS_HASH_FLAG_RWLOCK | KS_HASH_FLAG_DUP_CHECK | KS_HASH_FLAG_FREE_KEY, bumgr->pool);
ks_hash_create(&bumgr->realms, KS_HASH_MODE_CASE_INSENSITIVE, KS_HASH_FLAG_RWLOCK | KS_HASH_FLAG_DUP_CHECK | KS_HASH_FLAG_FREE_KEY, pool);
ks_assert(bumgr->realms);
ks_pool_set_cleanup(pool, bumgr, NULL, blade_upstreammgr_cleanup);
ks_pool_set_cleanup(bumgr, NULL, blade_upstreammgr_cleanup);
*bumgrP = bumgr;
@@ -109,9 +107,7 @@ KS_DECLARE(ks_status_t) blade_upstreammgr_destroy(blade_upstreammgr_t **bumgrP)
bumgr = *bumgrP;
*bumgrP = NULL;
ks_assert(bumgr);
pool = bumgr->pool;
pool = ks_pool_get(bumgr);
ks_pool_close(&pool);
@@ -140,8 +136,8 @@ KS_DECLARE(ks_status_t) blade_upstreammgr_localid_set(blade_upstreammgr_t *bumgr
goto done;
}
if (bumgr->localid) ks_pool_free(bumgr->pool, &bumgr->localid);
if (id) bumgr->localid = ks_pstrdup(bumgr->pool, id);
if (bumgr->localid) ks_pool_free(&bumgr->localid);
if (id) bumgr->localid = ks_pstrdup(ks_pool_get(bumgr), id);
ks_log(KS_LOG_DEBUG, "LocalID: %s\n", id);
@@ -212,8 +208,8 @@ KS_DECLARE(ks_status_t) blade_upstreammgr_masterid_set(blade_upstreammgr_t *bumg
ks_assert(bumgr);
ks_rwl_write_lock(bumgr->masterid_rwl);
if (bumgr->masterid) ks_pool_free(bumgr->pool, &bumgr->masterid);
if (id) bumgr->masterid = ks_pstrdup(bumgr->pool, id);
if (bumgr->masterid) ks_pool_free(&bumgr->masterid);
if (id) bumgr->masterid = ks_pstrdup(ks_pool_get(bumgr), id);
ks_log(KS_LOG_DEBUG, "MasterID: %s\n", id);
@@ -272,7 +268,7 @@ KS_DECLARE(ks_status_t) blade_upstreammgr_realm_add(blade_upstreammgr_t *bumgr,
ks_assert(bumgr);
ks_assert(realm);
key = ks_pstrdup(bumgr->pool, realm);
key = ks_pstrdup(ks_pool_get(bumgr), realm);
ks_hash_insert(bumgr->realms, (void *)key, (void *)KS_TRUE);
ks_log(KS_LOG_DEBUG, "Realm Added: %s\n", key);
libs/libblade/src/include/blade_connection.h
-1
View File
@@ -41,7 +41,6 @@ KS_DECLARE(ks_status_t) blade_connection_destroy(blade_connection_t **bcP);
KS_DECLARE(ks_status_t) blade_connection_startup(blade_connection_t *bc, blade_connection_direction_t direction);
KS_DECLARE(ks_status_t) blade_connection_shutdown(blade_connection_t *bc);
KS_DECLARE(blade_handle_t *) blade_connection_handle_get(blade_connection_t *bc);
KS_DECLARE(ks_pool_t *) blade_connection_pool_get(blade_connection_t *bc);
KS_DECLARE(const char *) blade_connection_id_get(blade_connection_t *bc);
KS_DECLARE(ks_status_t) blade_connection_read_lock(blade_connection_t *bc, ks_bool_t block);
KS_DECLARE(ks_status_t) blade_connection_read_unlock(blade_connection_t *bc);
libs/libblade/src/include/blade_stack.h
-1
View File
@@ -45,7 +45,6 @@ KS_DECLARE(ks_status_t) blade_handle_destroy(blade_handle_t **bhP);
KS_DECLARE(ks_status_t) blade_handle_create(blade_handle_t **bhP);
KS_DECLARE(ks_status_t) blade_handle_startup(blade_handle_t *bh, config_setting_t *config);
KS_DECLARE(ks_status_t) blade_handle_shutdown(blade_handle_t *bh);
KS_DECLARE(ks_pool_t *) blade_handle_pool_get(blade_handle_t *bh);
KS_DECLARE(ks_thread_pool_t *) blade_handle_tpool_get(blade_handle_t *bh);
KS_DECLARE(blade_transportmgr_t *) blade_handle_transportmgr_get(blade_handle_t *bh);