X-Git-Url: http://git.vrable.net/?a=blobdiff_plain;f=bluesky%2Fstore-simple.c;h=ad93a05ec11b67a8634d9782143824173da9ccca;hb=0083413280deccffea5f0e34512eb02f81633dae;hp=25eef53321e30a08c5efb125c2af8f774add2429;hpb=3545d572eecff2a6e0ad53bfae3ed0c35f232cbe;p=bluesky.git

diff --git a/bluesky/store-simple.c b/bluesky/store-simple.c
index 25eef53..ad93a05 100644
--- a/bluesky/store-simple.c
+++ b/bluesky/store-simple.c
@@ -20,11 +20,56 @@
 
 #include "bluesky-private.h"
 
+#define MAX_IDLE_CONNECTIONS 8
+
 typedef struct {
     GThreadPool *thread_pool;
     struct sockaddr_in server_addr;
+
+    /* A pool of open file connections to the server which are not currently in
+     * use. */
+    GQueue *fd_pool;
+    GMutex *fd_pool_lock;
 } SimpleStore;
 
+static int get_connection(SimpleStore *store)
+{
+    int fd = -1;
+
+    g_mutex_lock(store->fd_pool_lock);
+    if (!g_queue_is_empty(store->fd_pool)) {
+        fd = GPOINTER_TO_INT(g_queue_pop_head(store->fd_pool));
+    }
+    g_mutex_unlock(store->fd_pool_lock);
+    if (fd != -1)
+        return fd;
+
+    fd = socket(PF_INET, SOCK_STREAM, 0);
+    if (fd < 0) {
+        g_warning("Error creating simplestore socket: %m");
+        return -1;
+    }
+
+    if (connect(fd, (struct sockaddr *)&store->server_addr,
+                sizeof(store->server_addr)) < 0) {
+        g_warning("Error connecting to simplestore server: %m");
+        return -1;
+    }
+
+    return fd;
+}
+
+static void put_connection(SimpleStore *store, int fd)
+{
+    g_mutex_lock(store->fd_pool_lock);
+    g_queue_push_head(store->fd_pool, GINT_TO_POINTER(fd));
+    while (g_queue_get_length(store->fd_pool) > MAX_IDLE_CONNECTIONS) {
+        fd = GPOINTER_TO_INT(g_queue_pop_tail(store->fd_pool));
+        close(fd);
+    }
+    g_mutex_unlock(store->fd_pool_lock);
+}
+
 static gboolean write_data(int fd, const char *buf, size_t len)
 {
     while (len > 0) {
@@ -103,17 +148,8 @@ static void simplestore_task(gpointer a, gpointer b)
 
     async->status = ASYNC_RUNNING;
 
-    int fd = socket(PF_INET, SOCK_STREAM, 0);
+    int fd = get_connection(server);
     if (fd < 0) {
-        g_warning("Error creating simplestore socket: %m");
-        bluesky_store_async_mark_complete(async);
-        bluesky_store_async_unref(async);
-        return;
-    }
-
-    if (connect(fd, (struct sockaddr *)&server->server_addr,
-                sizeof(server->server_addr)) < 0) {
-        g_warning("Error connecting to simplestore server: %m");
         bluesky_store_async_mark_complete(async);
         bluesky_store_async_unref(async);
         return;
@@ -157,6 +193,7 @@ static void simplestore_task(gpointer a, gpointer b)
 
         async->data = bluesky_string_new(data, result);
         async->result = 0;
+        async->range_done = TRUE;
         break;
     }
 
@@ -190,10 +227,15 @@ static void simplestore_task(gpointer a, gpointer b)
         break;
     }
 
+    int success = (async->result == 0);
     bluesky_store_async_mark_complete(async);
     bluesky_store_async_unref(async);
 
-    close(fd);
+    if (success) {
+        put_connection(server, fd);
+    } else {
+        close(fd);
+    }
 }
 
 static char *simplestore_lookup_last(gpointer s, const char *prefix)
@@ -207,7 +249,7 @@ static gpointer simplestore_new(const gchar *path)
 
     /* TODO: Right now we leak this memory.  We should probably clean up in
      * simplestore_destroy, but it's not a big deal. */
-    const gchar *host = "127.0.0.1", *port = "8257";
+    const gchar *host = "127.0.0.1", *port = "9541";
     if (path != NULL) {
         gchar **target = g_strsplit(path, ":", 0);
         if (target[0] != NULL) {
@@ -231,7 +273,6 @@ static gpointer simplestore_new(const gchar *path)
                 gai_strerror(res));
         return NULL;
     }
-    freeaddrinfo(lookup_result);
     for (struct addrinfo *ai = lookup_result; ai != NULL; ai = ai->ai_next) {
         printf("flags=%d family=%d socktype=%d proto=%d\n",
                ai->ai_flags,
@@ -245,6 +286,10 @@ static gpointer simplestore_new(const gchar *path)
             fprintf(stderr, "Warning: Bad address record size!\n");
         }
     }
+    freeaddrinfo(lookup_result);
+
+    store->fd_pool = g_queue_new();
+    store->fd_pool_lock = g_mutex_new();
 
     store->thread_pool = g_thread_pool_new(simplestore_task, NULL,
                                            bluesky_max_threads, FALSE, NULL);