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