WIP; ds3_connection_pool queue

Author: DenverM80

src/ds3_connection.c

@@ -27,19 +27,33 @@ ds3_connection_pool* ds3_connection_pool_init(void) {
     return ds3_connection_pool_init_with_size(CONNECTION_POOL_SIZE);
 }
 
+void _ds3_queue_init(ds3_connection_pool* pool) {
+    int index;
+    pool->queue = g_new0(int, pool->size);
+    for (index = 0; index < pool->size; index++) {
+        pool->queue[index] = index; // init to default
+    }
+}
+
+void _ds3_queue_free(ds3_connection_pool* pool) {
+    g_free(pool->queue);
+}
+
 ds3_connection_pool* ds3_connection_pool_init_with_size(uint16_t pool_size) {
     printf("ds3_connection_pool_init_with_size(%u)\n", pool_size);
     ds3_connection_pool* pool = g_new0(ds3_connection_pool, 1);
     pool->connections = g_new0(ds3_connection*, pool_size);
-    pool->num_connections = pool_size;
+    pool->size = pool_size;
+
+    _ds3_queue_init(pool);
+
     g_mutex_init(&pool->mutex);
     g_cond_init(&pool->available_connections);
     pool->ref_count = 1;
     return pool;
 }
 
 void ds3_connection_pool_clear(ds3_connection_pool* pool, ds3_bool already_locked) {
-    printf("ds3_connection_pool_clear(%s)\n", (already_locked ? "locked" : "not locked"));
     int index;
 
     if (pool == NULL) {
@@ -49,63 +63,73 @@ void ds3_connection_pool_clear(ds3_connection_pool* pool, ds3_bool already_locke
     if (already_locked == False) {
         g_mutex_lock(&pool->mutex);
     }
+    printf("ds3_connection_pool_clear(%s)\n", (already_locked ? "locked" : "not locked"));
 
-    for (index = 0; index < pool->num_connections; index++) {
+    for (index = 0; index < pool->size; index++) {
         if (pool->connections[index] != NULL) {
             curl_easy_cleanup(pool->connections[index]);
         }
     }
 
+    _ds3_queue_free(pool);
+
     g_free(pool->connections);
     g_mutex_unlock(&pool->mutex);
     g_mutex_clear(&pool->mutex); // an attempt to clear a locked mutex is undefined
     g_cond_clear(&pool->available_connections);
 }
 
-static int _pool_inc(int index, uint16_t num_connections) {
-    printf("_pool_inc(%d, %u) :[%d]\n", index, num_connections, (index+1) % num_connections);
-    return (index+1) % num_connections;
+static int _queue_inc(int index, uint16_t size) {
+    printf("_pool_inc(%d, %u) :[%d]\n", index, size, (index+1) % size);
+    return (index+1) % size;
 }
 
-static int _pool_full(ds3_connection_pool* pool) {
-    printf("_pool_full(): head[%d] tail[%d] : [%d]\n", pool->head, pool->tail, (_pool_inc(pool->head, pool->num_connections) == pool->tail) );
-    return (_pool_inc(pool->head, pool->num_connections) == pool->tail);
+static int _queue_full(ds3_connection_pool* pool) {
+    return (_pool_inc(pool->head, pool->size) == pool->tail);
 }
 
 ds3_connection* ds3_connection_acquire(ds3_connection_pool* pool) {
-    printf("ds3_connection_acquire() BEGIN: head[%d] tail[%d]\n", pool->head, pool->tail);
     ds3_connection* connection = NULL;
+    int next_connection_index;
 
     g_mutex_lock(&pool->mutex);
+    printf("ds3_connection_acquire() BEGIN: head[%d] tail[%d]\n", pool->head, pool->tail);
     while (_pool_full(pool)) {
         g_cond_wait(&pool->available_connections, &pool->mutex);
     }
 
-    if (pool->connections[pool->head] == NULL) {
+    next_connection_index = pool->queue[pool->head];
+    if (pool->connections[next_connection_index] == NULL) {
+        connection = g_new0(ds3_connection, 1);
         connection = curl_easy_init();
 
-        pool->connections[pool->head] = connection;
+        pool->connections[next_connection_index] = connection;
     } else {
-        connection = pool->connections[pool->head];
+        connection = pool->connections[next_connection_index];
     }
-    pool->head = _pool_inc(pool->head, pool->num_connections);
+    pool->head = _pool_inc(pool->head, pool->size);
 
+    printf("ds3_connection_acquire() END: head[%d] tail[%d]\n", pool->head, pool->tail);
     g_mutex_unlock(&pool->mutex);
 
-    printf("ds3_connection_acquire() END: head[%d] tail[%d]\n", pool->head, pool->tail);
     return connection;
 }
 
 void ds3_connection_release(ds3_connection_pool* pool, ds3_connection* connection) {
-    printf("ds3_connection_release() BEGIN: head[%d] tail[%d]\n", pool->head, pool->tail);
+    int tail_connection_index;
+
     g_mutex_lock(&pool->mutex);
+    printf("ds3_connection_release() BEGIN: head[%d] tail[%d]\n", pool->head, pool->tail);
 
     curl_easy_reset(connection);
-    pool->tail = _pool_inc(pool->tail, pool->num_connections);
+    tail_connection_index = pool->queue[pool->tail];
+
+    pool->connections[tail_connection_index] = connection;
+    pool->tail = _pool_inc(pool->tail, pool->size);
 
+    printf("ds3_connection_release() END: head[%d] tail[%d]\n", pool->head, pool->tail);
     g_mutex_unlock(&pool->mutex);
     g_cond_signal(&pool->available_connections);
-    printf("ds3_connection_release() END: head[%d] tail[%d]\n", pool->head, pool->tail);
 }
 
 void ds3_connection_pool_inc_ref(ds3_connection_pool* pool) {

src/ds3_connection.h

@@ -37,7 +37,8 @@ typedef CURL ds3_connection;
 //-- Opaque struct
 struct _ds3_connection_pool{
     ds3_connection** connections;
-    uint16_t         num_connections;
+    int*             queue;
+    uint16_t         size;
     int              head;
     int              tail;
     ds3_mutex        mutex;

test/test.cpp

@@ -47,18 +47,31 @@ BOOST_GLOBAL_FIXTURE( BoostTestFixture );
 
 void log_timestamp(char* string_buff, long buff_size)
 {
-    
-    g_snprintf(string_buff, buff_size, "%s", );
+    time_t ltime;
+    struct tm result;
+    struct timeval tv;
+    char   usec_buff[8];
+
+    gettimeofday(&tv, NULL);
+    millisec = lrint(tv.tv_usec/1000.0); // Round to nearest millisec
+
+    ltime = time(NULL);
+    localtime_r(&ltime, &result);
+
+    strftime(string_buff, buff_size, "%Y:%m:%dT%H:%M:%S", tm);
+    strcat(string_buff, ".");
+    sprintf(usec_buff,"%d", (int)tmnow.tv_usec);
+    strcat(string_buff, usec_buff);
 }
 
 void test_log(const char* message, void* user_data) {
     char timebuffer[32];
     log_timestamp(timebuffer, 32);
     if (user_data) {
         int client_num = *((int*)user_data);
-        fprintf(stderr, "%s Client[%d] %s\n", client_num, message);
+        fprintf(stderr, "%s Client[%d] %s\n", timebuffer, client_num, message);
     } else {
-        fprintf(stderr, "%s %s\n", message);
+        fprintf(stderr, "%s %s\n", timebuffer, message);
     }
 }