bluesky/store-kv.cc

   1 /* Blue Sky: File Systems in the Cloud
   2  *
   3  * Copyright (C) 2009  The Regents of the University of California
   4  * Written by Michael Vrable <mvrable@cs.ucsd.edu>
   5  *
   6  * TODO: Licensing
   7  */
   8
   9 /* Interface to John McCullough's simple key/value store. */
  10
  11 #include <stdint.h>
  12 #include <stdlib.h>
  13 #include <glib.h>
  14 #include <string.h>
  15
  16 #include "bluesky-private.h"
  17 #include "kvservice.h"
  18 #include "kvclient.h"
  19
  20 using namespace boost;
  21 using namespace kvstore;
  22 using namespace std;
  23
  24 static GThreadPool *thread_pool = NULL;
  25
  26 static void kvstore_task(gpointer a, gpointer b)
  27 {
  28     BlueSkyStoreAsync *async = (BlueSkyStoreAsync *)a;
  29     KeyValueClient *client = (KeyValueClient *)bluesky_store_async_get_handle(async);
  30
  31     async->status = ASYNC_RUNNING;
  32
  33     switch (async->op) {
  34     case STORE_OP_GET:
  35     {
  36         string value;
  37         if (client->Get(async->key, &value)) {
  38             async->data = bluesky_string_new(g_memdup(value.c_str(),
  39                                                     value.length()),
  40                                             value.length());
  41             async->result = 0;
  42         }
  43         break;
  44     }
  45
  46     case STORE_OP_PUT:
  47     {
  48         string value(async->data->data, async->data->len);
  49         client->Put(async->key, value);
  50         break;
  51     }
  52
  53     default:
  54         break;
  55     }
  56
  57     bluesky_store_async_mark_complete(async);
  58     bluesky_store_async_unref(async);
  59 }
  60
  61 static gpointer kvstore_new(const gchar *path)
  62 {
  63     /* TODO: Right now we leak this memory.  We should probably clean up in
  64      * kvstore_destroy, but it's not a big deal. */
  65     const gchar *host = "127.0.0.1", *port = "9090";
  66     if (path != NULL) {
  67         gchar **target = g_strsplit(path, ":", 0);
  68         if (target[0] != NULL) {
  69             host = target[0];
  70             if (target[1] != NULL) {
  71                 port = target[1];
  72             }
  73         }
  74     }
  75
  76     static volatile gsize once = 0;
  77     if (g_once_init_enter(&once)) {
  78         thread_pool = g_thread_pool_new(kvstore_task, NULL, -1, FALSE, NULL);
  79         g_once_init_leave(&once, 1);
  80     }
  81
  82     g_print("kvstore: %s port %s\n", host, port);
  83     KeyValueClient *client = new KeyValueClient(host, port);
  84     return client;
  85 }
  86
  87 static void kvstore_destroy(gpointer store)
  88 {
  89     KeyValueClient *client = (KeyValueClient *)store;
  90     delete client;
  91 }
  92
  93 static void kvstore_submit(gpointer store, BlueSkyStoreAsync *async)
  94 {
  95     KeyValueClient *client = (KeyValueClient *)store;
  96
  97     g_return_if_fail(async->status == ASYNC_NEW);
  98     g_return_if_fail(async->op != STORE_OP_NONE);
  99
 100     switch (async->op) {
 101     case STORE_OP_GET:
 102     case STORE_OP_PUT:
 103         async->status = ASYNC_PENDING;
 104         bluesky_store_async_ref(async);
 105         g_thread_pool_push(thread_pool, async, NULL);
 106         break;
 107
 108     default:
 109         g_warning("Uknown operation type for kvstore: %d\n", async->op);
 110         bluesky_store_async_mark_complete(async);
 111         break;
 112     }
 113 }
 114
 115 static void kvstore_cleanup(gpointer store, BlueSkyStoreAsync *async)
 116 {
 117     KeyValueClient *client = (KeyValueClient *)store;
 118 }
 119
 120 static BlueSkyStoreImplementation store_impl = {
 121     kvstore_new,
 122     kvstore_destroy,
 123     kvstore_submit,
 124     kvstore_cleanup,
 125 };
 126
 127 extern "C" void bluesky_store_init_kv(void)
 128 {
 129     bluesky_store_register(&store_impl, "kv");
 130 }