stellar-stellar/src/session/session_manager.cpp

#include <stdlib.h>
#include <assert.h>

#include "timestamp.h"
#include "session_manager.h"
#include "session_pool.h"
#include "session_table.h"
#include "session_timer.h"
#include "session_private.h"

struct session_manager
{
     struct session_pool *sess_pool;
     struct session_table *sess_table;
     struct session_timer *sess_timer;

     session_event_cb event_cb;
     void *arg;

     struct session *queue_head_ptr;
     struct session *queue_tail_ptr;
};

#define HANDSHAKE_TIMEOUT_MS (5 * 1000)
#define DATA_TIMEOUT_MS (60 * 1000)

/******************************************************************************
 * private API
 ******************************************************************************/

static void session_manager_handle_new_session(struct session_manager *mgr, struct session_key *key, struct session *sess, const struct packet *pkt)
{
     // TODO
}

static void session_manager_handle_old_session(struct session_manager *mgr, struct session_key *key, struct session *sess, const struct packet *pkt)
{
     // TODO
}

// Enqueue ready session to queue tail
static void session_manager_enqueue_ready_session(struct session_manager *mgr, struct session *sess)
{
     if (mgr == NULL || sess == NULL)
     {
          return;
     }

     if (mgr->queue_head_ptr == NULL)
     {
          mgr->queue_head_ptr = sess;
          mgr->queue_tail_ptr = sess;
          sess->next_ready_ptr = NULL;
     }
     else
     {
          mgr->queue_tail_ptr->next_ready_ptr = sess;
          mgr->queue_tail_ptr = sess;
          sess->next_ready_ptr = NULL;
     }
}

// Dequeue ready session from queue head
struct session *session_manager_ready_session_dequeue(struct session_manager *mgr)
{
     if (mgr == NULL)
     {
          return NULL;
     }

     struct session *sess = mgr->queue_head_ptr;
     if (sess == NULL)
     {
          return NULL;
     }

     if (mgr->queue_head_ptr == mgr->queue_tail_ptr)
     {
          mgr->queue_head_ptr = NULL;
          mgr->queue_tail_ptr = NULL;
     }
     else
     {
          mgr->queue_head_ptr = sess->next_ready_ptr;
     }

     sess->next_ready_ptr = NULL;
     return sess;
}

// handshake expire set to discard
static void handshake_expire_cb(struct session *sess, void *arg)
{
     struct session_manager *mgr = (struct session_manager *)arg;
     assert(mgr != NULL);

     session_set_state(sess, SESSION_STATE_DISCARD);
     session_push_event(sess, SESSION_EVENT_CLOSING);
     session_manager_enqueue_ready_session(mgr, sess);
}

// data expire set to closing
static void data_expire_cb(struct session *sess, void *arg)
{
     struct session_manager *mgr = (struct session_manager *)arg;
     assert(mgr != NULL);

     session_set_state(sess, SESSION_STATE_CLOSING);
     session_push_event(sess, SESSION_EVENT_CLOSING);
     session_manager_enqueue_ready_session(mgr, sess);
}

/******************************************************************************
 * public API
 ******************************************************************************/

struct session_manager *session_manager_create(uint64_t max_session_num)
{
     struct session_manager *mgr = (struct session_manager *)calloc(1, sizeof(struct session_manager));
     if (mgr == NULL)
     {
          return NULL;
     }

     mgr->sess_pool = session_pool_create(max_session_num);
     if (mgr->sess_pool == NULL)
     {
          goto error;
     }

     mgr->sess_table = session_table_create();
     if (mgr->sess_table == NULL)
     {
          goto error;
     }

     mgr->sess_timer = session_timer_create();
     if (mgr->sess_timer == NULL)
     {
          goto error;
     }

     mgr->queue_head_ptr = NULL;
     mgr->queue_tail_ptr = NULL;

     return mgr;

error:
     session_manager_destroy(mgr);
     return NULL;
}

void session_manager_destroy(struct session_manager *mgr)
{
     if (mgr)
     {
          session_timer_destroy(mgr->sess_timer);
          session_table_destroy(mgr->sess_table);
          session_pool_destroy(mgr->sess_pool);
          free(mgr);
          mgr = NULL;
     }
}

void session_manager_set_session_eventcb(struct session_manager *mgr, session_event_cb cb, void *arg)
{
     mgr->event_cb = cb;
     mgr->arg = arg;
}

struct session *session_manager_find_session(struct session_manager *mgr, const struct packet *pkt)
{
     struct session_key key;
     // TODO packet to key

     struct session *sess = session_table_find_session(mgr->sess_table, &key);
     if (sess == NULL)
     {
          if (session_pool_get_count(mgr->sess_pool) == 1)
          {
               struct session *oldest_sess = session_table_find_oldest_session(mgr->sess_table);
               assert(oldest_sess == NULL);
               session_set_state(oldest_sess, SESSION_STATE_DISCARD);
               session_push_event(oldest_sess, SESSION_EVENT_CLOSING);
               session_manager_enqueue_ready_session(mgr, oldest_sess);
          }

          sess = session_pool_alloc(mgr->sess_pool);
          assert(sess != NULL);

          session_manager_handle_new_session(mgr, &key, sess, pkt);
     }
     else
     {
          session_manager_handle_old_session(mgr, &key, sess, pkt);
     }

     return sess;
}

void session_manager_dispatch(struct session_manager *mgr)
{
     uint32_t event;
     struct session *sess;

     void *cb_arg = mgr->arg;
     session_event_cb event_cb = mgr->event_cb;

     while (1)
     {
          sess = session_timer_expire_session(mgr->sess_timer, timestamp_get_msec());
          if (sess == NULL)
          {
               break;
          }
          session_run_expirecb(sess);
     }

     while (1)
     {
          sess = session_manager_ready_session_dequeue(mgr);
          if (sess == NULL)
          {
               break;
          }

          while (1)
          {
               if (session_pop_event(sess, &event) == false)
               {
                    break;
               }

               if (event_cb)
               {
                    event_cb(sess, event, cb_arg);
               }
          }
     };
}