blade_handle_t *handle;
ks_pool_t *pool;
- ks_bool_t shutdown;
- ks_thread_t *state_thread;
blade_session_state_t state;
const char *id;
ks_rwl_t *lock;
+
+ ks_mutex_t *mutex;
+ ks_cond_t *cond;
+
list_t connections;
ks_time_t ttl;
ks_status_t blade_session_state_on_ready(blade_session_t *bs);
ks_status_t blade_session_process(blade_session_t *bs, cJSON *json);
-KS_DECLARE(ks_status_t) blade_session_create(blade_session_t **bsP, blade_handle_t *bh)
+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)
+{
+ blade_session_t *bs = (blade_session_t *)ptr;
+
+ ks_assert(bs);
+
+ switch (action) {
+ case KS_MPCL_ANNOUNCE:
+ break;
+ case KS_MPCL_TEARDOWN:
+ blade_session_shutdown(bs);
+ break;
+ case KS_MPCL_DESTROY:
+ // thread has auto cleanup
+ //if (bs->state_thread) {
+ // ks_pool_free(bs->pool, &bs->state_thread);
+ //}
+
+ // @todo consider looking at supporting externally allocated memory entries that can have cleanup callbacks associated, but the memory is not freed from the pool, only linked as an external allocation for auto cleanup
+ // which would allow calling something like ks_pool_set_cleanup(bs->properties, ...) and when the pool is destroyed, it can call a callback which handles calling cJSON_Delete, which is allocated externally
+ cJSON_Delete(bs->properties);
+ bs->properties = NULL;
+ // rwl has auto cleanup
+ //ks_rwl_destroy(&bs->properties_lock);
+
+ // connection id's are created as copies within the session's pool, and will be cleaned up with the pool automatically
+ //list_iterator_start(&bs->connections);
+ //while (list_iterator_hasnext(&bs->connections)) {
+ // const char *id = (const char *)list_iterator_next(&bs->connections);
+ // ks_pool_free(bs->pool, &id);
+ //}
+ //list_iterator_stop(&bs->connections);
+
+ // called by list_destroy()
+ //list_clear(&bs->connections);
+
+ // @todo change bs->connections to a pointer, allocate the memory in the pool, and set a cleanup callback for it to call list_destroy() automatically
+ list_destroy(&bs->connections);
+ // q has auto cleanup
+ //ks_q_destroy(&bs->receiving);
+ //ks_q_destroy(&bs->sending);
+
+ // rwl has auto cleanup
+ //ks_rwl_destroy(&bs->lock);
+
+ // @todo remove this, it's just for posterity in debugging
+ //bs->state_thread = NULL;
+ bs->properties_lock = NULL;
+ bs->receiving = NULL;
+ bs->sending = NULL;
+ bs->cond = NULL;
+ bs->mutex = NULL;
+ bs->lock = NULL;
+
+ ks_pool_free(bs->pool, &bs->id);
+ break;
+ }
+}
+
+
+KS_DECLARE(ks_status_t) blade_session_create(blade_session_t **bsP, blade_handle_t *bh, const char *sid)
{
blade_session_t *bs = NULL;
ks_pool_t *pool = NULL;
- uuid_t id;
ks_assert(bsP);
ks_assert(bh);
- pool = blade_handle_pool_get(bh);
+ ks_pool_open(&pool);
+ ks_assert(pool);
bs = ks_pool_alloc(pool, sizeof(blade_session_t));
bs->handle = bh;
bs->pool = pool;
- ks_uuid(&id);
- bs->id = ks_uuid_str(pool, &id);
+ if (sid) bs->id = ks_pstrdup(pool, sid);
+ else {
+ uuid_t id;
+ ks_uuid(&id);
+ bs->id = ks_uuid_str(pool, &id);
+ }
ks_rwl_create(&bs->lock, pool);
ks_assert(bs->lock);
+ ks_mutex_create(&bs->mutex, KS_MUTEX_FLAG_DEFAULT, pool);
+ ks_assert(bs->mutex);
+ ks_cond_create_ex(&bs->cond, pool, bs->mutex);
+ ks_assert(bs->cond);
+
list_init(&bs->connections);
ks_q_create(&bs->sending, pool, 0);
ks_assert(bs->sending);
ks_rwl_create(&bs->properties_lock, pool);
ks_assert(bs->properties_lock);
- *bsP = bs;
+
+ ks_assert(ks_pool_set_cleanup(pool, bs, NULL, blade_session_cleanup) == KS_STATUS_SUCCESS);
ks_log(KS_LOG_DEBUG, "Created\n");
+ *bsP = bs;
+
return KS_STATUS_SUCCESS;
}
KS_DECLARE(ks_status_t) blade_session_destroy(blade_session_t **bsP)
{
blade_session_t *bs = NULL;
+ ks_pool_t *pool = NULL;
ks_assert(bsP);
ks_assert(*bsP);
bs = *bsP;
- blade_session_shutdown(bs);
-
- cJSON_Delete(bs->properties);
- bs->properties = NULL;
- ks_rwl_destroy(&bs->properties_lock);
-
- list_destroy(&bs->connections);
- ks_q_destroy(&bs->receiving);
- ks_q_destroy(&bs->sending);
-
- ks_rwl_destroy(&bs->lock);
-
- ks_pool_free(bs->pool, &bs->id);
-
- ks_pool_free(bs->pool, bsP);
+ pool = bs->pool;
+ //ks_pool_free(bs->pool, bsP);
+ ks_pool_close(&pool);
ks_log(KS_LOG_DEBUG, "Destroyed\n");
+ *bsP = NULL;
+
return KS_STATUS_SUCCESS;
}
KS_DECLARE(ks_status_t) blade_session_startup(blade_session_t *bs)
{
+ ks_thread_pool_t *tpool = NULL;
+
ks_assert(bs);
- blade_session_state_set(bs, BLADE_SESSION_STATE_NONE);
+ tpool = blade_handle_tpool_get(bs->handle);
+ ks_assert(tpool);
- if (ks_thread_create_ex(&bs->state_thread,
- blade_session_state_thread,
- bs,
- KS_THREAD_FLAG_DEFAULT,
- KS_THREAD_DEFAULT_STACK,
- KS_PRI_NORMAL,
- bs->pool) != KS_STATUS_SUCCESS) {
+ blade_session_state_set(bs, BLADE_SESSION_STATE_NONE);
+
+ if (ks_thread_pool_add_job(tpool, blade_session_state_thread, bs) != KS_STATUS_SUCCESS) {
// @todo error logging
return KS_STATUS_FAIL;
}
ks_assert(bs);
- if (bs->state_thread) {
- bs->shutdown = KS_TRUE;
- ks_thread_join(bs->state_thread);
- printf("FREEING SESSION THREAD %p\n", (void *)bs->state_thread);
- ks_pool_free(bs->pool, &bs->state_thread);
- bs->shutdown = KS_FALSE;
- }
+ blade_handle_sessions_remove(bs);
while (ks_q_trypop(bs->sending, (void **)&json) == KS_STATUS_SUCCESS && json) cJSON_Delete(json);
while (ks_q_trypop(bs->receiving, (void **)&json) == KS_STATUS_SUCCESS && json) cJSON_Delete(json);
- list_iterator_start(&bs->connections);
- while (list_iterator_hasnext(&bs->connections)) {
- const char *id = (const char *)list_iterator_next(&bs->connections);
- ks_pool_free(bs->pool, &id);
- }
- list_iterator_stop(&bs->connections);
- list_clear(&bs->connections);
-
ks_log(KS_LOG_DEBUG, "Stopped\n");
return KS_STATUS_SUCCESS;
return bs->handle;
}
-KS_DECLARE(ks_pool_t *) blade_session_pool_get(blade_session_t *bs)
-{
- ks_assert(bs);
-
- return bs->pool;
-}
-
KS_DECLARE(const char *) blade_session_id_get(blade_session_t *bs)
{
ks_assert(bs);
return bs->id;
}
-KS_DECLARE(void) blade_session_id_set(blade_session_t *bs, const char *id)
-{
- ks_assert(bs);
- ks_assert(id);
-
- if (bs->id) ks_pool_free(bs->pool, &bs->id);
- bs->id = ks_pstrdup(bs->pool, id);
-}
-
KS_DECLARE(blade_session_state_t) blade_session_state_get(blade_session_t *bs)
{
ks_assert(bs);
{
ks_assert(bs);
+ ks_mutex_lock(bs->mutex);
+
bs->state = state;
blade_handle_session_state_callbacks_execute(bs, BLADE_SESSION_STATE_CONDITION_PRE);
+
+ ks_cond_try_signal(bs->cond);
+ ks_mutex_unlock(bs->mutex);
}
KS_DECLARE(void) blade_session_hangup(blade_session_t *bs)
{
ks_assert(bs);
- if (bs->state != BLADE_SESSION_STATE_HANGUP && bs->state != BLADE_SESSION_STATE_DESTROY) {
+ if (!blade_session_terminating(bs)) {
ks_log(KS_LOG_DEBUG, "Session (%s) hanging up\n", bs->id);
blade_session_state_set(bs, BLADE_SESSION_STATE_HANGUP);
}
{
ks_assert(bs);
- return bs->state == BLADE_SESSION_STATE_HANGUP || bs->state == BLADE_SESSION_STATE_DESTROY;
+ return bs->state == BLADE_SESSION_STATE_HANGUP || bs->state == BLADE_SESSION_STATE_DESTROY || bs->state == BLADE_SESSION_STATE_CLEANUP;
}
KS_DECLARE(ks_status_t) blade_session_connections_add(blade_session_t *bs, const char *id)
cid = ks_pstrdup(bs->pool, id);
ks_assert(cid);
+ ks_mutex_lock(bs->mutex);
+
list_append(&bs->connections, cid);
+ bs->ttl = 0;
+
+ ks_mutex_unlock(bs->mutex);
+
ks_log(KS_LOG_DEBUG, "Session (%s) connection added (%s)\n", bs->id, id);
- bs->ttl = 0;
return ret;
}
ks_assert(bs);
+ ks_mutex_lock(bs->mutex);
+
size = list_size(&bs->connections);
for (uint32_t i = 0; i < size; ++i) {
const char *cid = (const char *)list_get_at(&bs->connections, i);
if (list_size(&bs->connections) == 0) bs->ttl = ks_time_now() + (5 * KS_USEC_PER_SEC);
+ ks_mutex_unlock(bs->mutex);
+
return ret;
}
ks_status_t blade_session_connections_choose(blade_session_t *bs, cJSON *json, blade_connection_t **bcP)
{
+ ks_status_t ret = KS_STATUS_SUCCESS;
blade_connection_t *bc = NULL;
const char *cid = NULL;
ks_assert(json);
ks_assert(bcP);
+ ks_mutex_lock(bs->mutex);
+
// @todo may be multiple connections, for now let's just assume there will be only one
// later there will need to be a way to pick which connection to use
cid = list_get_at(&bs->connections, 0);
if (!cid) {
// no connections available
- return KS_STATUS_FAIL;
+ ret = KS_STATUS_FAIL;
+ goto done;
}
bc = blade_handle_connections_get(bs->handle, cid);
if (!bc) {
// @todo error logging... this shouldn't happen
- return KS_STATUS_FAIL;
+ ret = KS_STATUS_FAIL;
+ goto done;
}
// @todo make sure the connection is in the READY state before allowing it to be choosen, just in case it is detaching or not quite fully attached
*bcP = bc;
- return KS_STATUS_SUCCESS;
+done:
+ ks_mutex_unlock(bs->mutex);
+
+ return ret;
}
KS_DECLARE(ks_status_t) blade_session_sending_push(blade_session_t *bs, cJSON *json)
{
+ ks_status_t ret = KS_STATUS_SUCCESS;
cJSON *json_copy = NULL;
ks_assert(bs);
ks_assert(json);
json_copy = cJSON_Duplicate(json, 1);
- return ks_q_push(bs->sending, json_copy);
+ if ((ret = ks_q_push(bs->sending, json_copy)) == KS_STATUS_SUCCESS) ks_cond_try_signal(bs->cond);
+ return ret;
}
KS_DECLARE(ks_status_t) blade_session_sending_pop(blade_session_t *bs, cJSON **json)
KS_DECLARE(ks_status_t) blade_session_receiving_push(blade_session_t *bs, cJSON *json)
{
+ ks_status_t ret = KS_STATUS_SUCCESS;
cJSON *json_copy = NULL;
ks_assert(bs);
ks_assert(json);
json_copy = cJSON_Duplicate(json, 1);
- return ks_q_push(bs->receiving, json_copy);
+ if ((ret = ks_q_push(bs->receiving, json_copy)) == KS_STATUS_SUCCESS) ks_cond_try_signal(bs->cond);
+ return ret;
}
KS_DECLARE(ks_status_t) blade_session_receiving_pop(blade_session_t *bs, cJSON **json)
blade_session_t *bs = NULL;
blade_session_state_t state;
cJSON *json = NULL;
+ ks_bool_t shutdown = KS_FALSE;
ks_assert(thread);
ks_assert(data);
bs = (blade_session_t *)data;
+
+ ks_mutex_lock(bs->mutex);
- while (!bs->shutdown) {
+ while (!shutdown) {
+ // Entering the call below, the mutex is expected to be locked and will be unlocked by the call
+ ks_cond_timedwait(bs->cond, 500);
+ // Leaving the call above, the mutex will be locked after being signalled, timing out, or woken up for any reason
state = bs->state;
cJSON_Delete(json);
}
}
-
+
blade_handle_session_state_callbacks_execute(bs, BLADE_SESSION_STATE_CONDITION_POST);
switch (state) {
case BLADE_SESSION_STATE_DESTROY:
blade_session_state_on_destroy(bs);
- return NULL;
+ shutdown = KS_TRUE;
+ break;
case BLADE_SESSION_STATE_HANGUP:
blade_session_state_on_hangup(bs);
break;
case BLADE_SESSION_STATE_CONNECT:
ks_log(KS_LOG_DEBUG, "Session (%s) state connect\n", bs->id);
- ks_sleep_ms(1000);
+ //ks_sleep_ms(1000);
break;
case BLADE_SESSION_STATE_ATTACH:
ks_log(KS_LOG_DEBUG, "Session (%s) state attach\n", bs->id);
- ks_sleep_ms(1000);
+ //ks_sleep_ms(1000);
break;
case BLADE_SESSION_STATE_DETACH:
ks_log(KS_LOG_DEBUG, "Session (%s) state detach\n", bs->id);
- ks_sleep_ms(1000);
+ //ks_sleep_ms(1000);
break;
case BLADE_SESSION_STATE_READY:
blade_session_state_on_ready(bs);
if (list_empty(&bs->connections) &&
bs->ttl > 0 &&
- bs->state != BLADE_SESSION_STATE_HANGUP &&
- bs->state != BLADE_SESSION_STATE_DESTROY &&
+ !blade_session_terminating(bs) &&
ks_time_now() >= bs->ttl) {
ks_log(KS_LOG_DEBUG, "Session (%s) TTL timeout\n", bs->id);
blade_session_hangup(bs);
}
}
+ ks_mutex_unlock(bs->mutex);
+
+ blade_session_destroy(&bs);
return NULL;
}
}
list_iterator_stop(&bs->connections);
+ // @todo review this code, it may end up blocking forever as the mutex for the session is locked and the connection thread will not be able to lock to remove the connection...
+ // isolate another mutex for the connection list, simclist is not intrinsicly thread safe like other containers
while (!list_empty(&bs->connections)) ks_sleep(100);
blade_session_state_set(bs, BLADE_SESSION_STATE_DESTROY);
}
}
- ks_sleep_ms(1);
+ //ks_sleep_ms(1);
return KS_STATUS_SUCCESS;
}
}
if (list_empty(&bs->connections)) {
- // @todo cache the blade_request_t here if it exists to gaurentee it's cached before a response could be received
blade_session_sending_push(bs, json);
} else {
blade_connection_t *bc = NULL;
if (blade_session_connections_choose(bs, json, &bc) != KS_STATUS_SUCCESS) return KS_STATUS_FAIL;
- // @todo cache the blade_request_t here if it exists to gaurentee it's cached before a response could be received
blade_connection_sending_push(bc, json);
blade_connection_read_unlock(bc);
}
}
}
- // @todo load DSOs
-
- // @todo call onload and onstartup callbacks for modules from DSOs
+ // @todo load internal modules, call onload and onstartup
if (ks_thread_create_ex(&bh->worker_thread,
blade_handle_worker_thread,
ks_hash_remove(bh->requests, key);
blade_request_destroy(&value);
+ // @todo note to self, fix this when switching to auto cleanup, as hash invalidates iterator when removing
}
for (it = ks_hash_first(bh->sessions, KS_UNLOCKED); it; it = ks_hash_next(&it)) {
}
while (ks_hash_count(bh->sessions) > 0) ks_sleep_ms(100);
- // @todo call onshutdown and onunload callbacks for modules from DSOs, which will unregister transports and spaces, and will disconnect remaining
- // unattached connections
+ // @todo unload internal modules, call onshutdown and onunload
while ((it = ks_hash_first(bh->events, KS_UNLOCKED))) {
void *key = NULL;
ks_hash_this(it, (const void **)&key, NULL, (void **)&value);
blade_handle_event_unregister(bh, (const char *)key);
+ // @todo note to self, fix this when switching to auto cleanup, as hash invalidates iterator when removing
}
while ((it = ks_hash_first(bh->spaces, KS_UNLOCKED))) {
ks_hash_this(it, (const void **)&key, NULL, (void **)&value);
blade_handle_space_unregister(value);
+ // @todo note to self, fix this when switching to auto cleanup, as hash invalidates iterator when removing
}
// @todo unload DSOs
{
blade_handle_t *bh = NULL;
blade_connection_t *bc = NULL;
- blade_session_t *bs = NULL;
+ //blade_session_t *bs = NULL;
ks_hash_iterator_t *it = NULL;
ks_q_t *cleanup = NULL;
blade_connection_destroy(&bc);
}
- ks_hash_write_lock(bh->sessions);
- for (it = ks_hash_first(bh->sessions, KS_UNLOCKED); it; it = ks_hash_next(&it)) {
- void *key = NULL;
- blade_session_t *value = NULL;
+ //ks_hash_write_lock(bh->sessions);
+ //for (it = ks_hash_first(bh->sessions, KS_UNLOCKED); it; it = ks_hash_next(&it)) {
+ // void *key = NULL;
+ // blade_session_t *value = NULL;
- ks_hash_this(it, (const void **)&key, NULL, (void **)&value);
+ // ks_hash_this(it, (const void **)&key, NULL, (void **)&value);
- if (blade_session_state_get(value) == BLADE_SESSION_STATE_CLEANUP) ks_q_push(cleanup, value);
- }
- ks_hash_write_unlock(bh->sessions);
+ // if (blade_session_state_get(value) == BLADE_SESSION_STATE_CLEANUP) ks_q_push(cleanup, value);
+ //}
+ //ks_hash_write_unlock(bh->sessions);
- while (ks_q_trypop(cleanup, (void **)&bs) == KS_STATUS_SUCCESS) {
- blade_handle_sessions_remove(bs);
- blade_session_destroy(&bs);
- }
+ //while (ks_q_trypop(cleanup, (void **)&bs) == KS_STATUS_SUCCESS) {
+ // blade_handle_sessions_remove(bs);
+ // blade_session_destroy(&bs);
+ //}
ks_sleep_ms(500);
}
projects / thirdparty / freeswitch.git / commitdiff
