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     gchar **target = g_strsplit(path, ":", 0);
  66     const gchar *host = "127.0.0.1", *port = "9090";
  67     if (target[0] != NULL) {
  68         host = target[0];
  69         if (target[1] != NULL) {
  70             port = target[1];
  71         }
  72     }
  73
  74     static volatile gsize once = 0;
  75     if (g_once_init_enter(&once)) {
  76         thread_pool = g_thread_pool_new(kvstore_task, NULL, -1, FALSE, NULL);
  77         g_once_init_leave(&once, 1);
  78     }
  79
  80     g_print("kvstore: %s port %s\n", host, port);
  81     KeyValueClient *client = new KeyValueClient(host, port);
  82     return client;
  83 }
  84
  85 static void kvstore_destroy(gpointer store)
  86 {
  87     KeyValueClient *client = (KeyValueClient *)store;
  88     delete client;
  89 }
  90
  91 static void kvstore_submit(gpointer store, BlueSkyStoreAsync *async)
  92 {
  93     KeyValueClient *client = (KeyValueClient *)store;
  94
  95     g_return_if_fail(async->status == ASYNC_NEW);
  96     g_return_if_fail(async->op != STORE_OP_NONE);
  97
  98     switch (async->op) {
  99     case STORE_OP_GET:
 100     case STORE_OP_PUT:
 101         async->status = ASYNC_PENDING;
 102         bluesky_store_async_ref(async);
 103         g_thread_pool_push(thread_pool, async, NULL);
 104         break;
 105
 106     default:
 107         g_warning("Uknown operation type for kvstore: %d\n", async->op);
 108         bluesky_store_async_mark_complete(async);
 109         break;
 110     }
 111 }
 112
 113 static void kvstore_cleanup(gpointer store, BlueSkyStoreAsync *async)
 114 {
 115     KeyValueClient *client = (KeyValueClient *)store;
 116 }
 117
 118 static BlueSkyStoreImplementation store_impl = {
 119     kvstore_new,
 120     kvstore_destroy,
 121     kvstore_submit,
 122     kvstore_cleanup,
 123 };
 124
 125 extern "C" void bluesky_store_init_kv(void)
 126 {
 127     bluesky_store_register(&store_impl, "kv");
 128 }