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,
  79                                         bluesky_max_threads, FALSE, NULL);
  80         g_once_init_leave(&once, 1);
  81     }
  82
  83     g_print("kvstore: %s port %s\n", host, port);
  84     KeyValueClient *client = new KeyValueClient(host, port);
  85     return client;
  86 }
  87
  88 static void kvstore_destroy(gpointer store)
  89 {
  90     KeyValueClient *client = (KeyValueClient *)store;
  91     delete client;
  92 }
  93
  94 static void kvstore_submit(gpointer store, BlueSkyStoreAsync *async)
  95 {
  96     KeyValueClient *client = (KeyValueClient *)store;
  97
  98     g_return_if_fail(async->status == ASYNC_NEW);
  99     g_return_if_fail(async->op != STORE_OP_NONE);
 100
 101     switch (async->op) {
 102     case STORE_OP_GET:
 103     case STORE_OP_PUT:
 104         async->status = ASYNC_PENDING;
 105         bluesky_store_async_ref(async);
 106         g_thread_pool_push(thread_pool, async, NULL);
 107         break;
 108
 109     default:
 110         g_warning("Uknown operation type for kvstore: %d\n", async->op);
 111         bluesky_store_async_mark_complete(async);
 112         break;
 113     }
 114 }
 115
 116 static void kvstore_cleanup(gpointer store, BlueSkyStoreAsync *async)
 117 {
 118     KeyValueClient *client = (KeyValueClient *)store;
 119 }
 120
 121 static BlueSkyStoreImplementation store_impl = {
 122     kvstore_new,
 123     kvstore_destroy,
 124     kvstore_submit,
 125     kvstore_cleanup,
 126 };
 127
 128 extern "C" void bluesky_store_init_kv(void)
 129 {
 130     bluesky_store_register(&store_impl, "kv");
 131 }