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  * Redistribution and use in source and binary forms, with or without
   7  * modification, are permitted provided that the following conditions
   8  * are met:
   9  * 1. Redistributions of source code must retain the above copyright
  10  *    notice, this list of conditions and the following disclaimer.
  11  * 2. Redistributions in binary form must reproduce the above copyright
  12  *    notice, this list of conditions and the following disclaimer in the
  13  *    documentation and/or other materials provided with the distribution.
  14  * 3. Neither the name of the University nor the names of its contributors
  15  *    may be used to endorse or promote products derived from this software
  16  *    without specific prior written permission.
  17  *
  18  * THIS SOFTWARE IS PROVIDED BY THE REGENTS AND CONTRIBUTORS ``AS IS'' AND
  19  * ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
  20  * IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
  21  * ARE DISCLAIMED.  IN NO EVENT SHALL THE REGENTS OR CONTRIBUTORS BE LIABLE
  22  * FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL
  23  * DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS
  24  * OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION)
  25  * HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT
  26  * LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY
  27  * OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF
  28  * SUCH DAMAGE.
  29  */
  30
  31 /* Interface to John McCullough's simple key/value store. */
  32
  33 #include <stdint.h>
  34 #include <stdlib.h>
  35 #include <glib.h>
  36 #include <string.h>
  37
  38 #include "bluesky-private.h"
  39 #include "kvservice.h"
  40 #include "kvclient.h"
  41
  42 using namespace boost;
  43 using namespace kvstore;
  44 using namespace std;
  45
  46 static GThreadPool *thread_pool = NULL;
  47
  48 static void kvstore_task(gpointer a, gpointer b)
  49 {
  50     BlueSkyStoreAsync *async = (BlueSkyStoreAsync *)a;
  51     KeyValueClient *client = (KeyValueClient *)bluesky_store_async_get_handle(async);
  52
  53     async->status = ASYNC_RUNNING;
  54
  55     switch (async->op) {
  56     case STORE_OP_GET:
  57     {
  58         string value;
  59         if (client->Get(async->key, &value)) {
  60             async->data = bluesky_string_new(g_memdup(value.c_str(),
  61                                                     value.length()),
  62                                             value.length());
  63             async->result = 0;
  64         } else {
  65             g_warning("Failed to fetch key %s from kvstore", async->key);
  66         }
  67         break;
  68     }
  69
  70     case STORE_OP_PUT:
  71     {
  72         string value(async->data->data, async->data->len);
  73         if (!client->Put(async->key, value)) {
  74             g_warning("Failed to store key %s to kvstore", async->key);
  75         }
  76         break;
  77     }
  78
  79     default:
  80         break;
  81     }
  82
  83     bluesky_store_async_mark_complete(async);
  84     bluesky_store_async_unref(async);
  85 }
  86
  87 static gpointer kvstore_new(const gchar *path)
  88 {
  89     /* TODO: Right now we leak this memory.  We should probably clean up in
  90      * kvstore_destroy, but it's not a big deal. */
  91     const gchar *host = "127.0.0.1", *port = "9090";
  92     if (path != NULL) {
  93         gchar **target = g_strsplit(path, ":", 0);
  94         if (target[0] != NULL) {
  95             host = target[0];
  96             if (target[1] != NULL) {
  97                 port = target[1];
  98             }
  99         }
 100     }
 101
 102     static volatile gsize once = 0;
 103     if (g_once_init_enter(&once)) {
 104         thread_pool = g_thread_pool_new(kvstore_task, NULL,
 105                                         bluesky_max_threads, FALSE, NULL);
 106         g_once_init_leave(&once, 1);
 107     }
 108
 109     g_print("kvstore: %s port %s\n", host, port);
 110     KeyValueClient *client = new KeyValueClient(host, port);
 111     return client;
 112 }
 113
 114 static void kvstore_destroy(gpointer store)
 115 {
 116     KeyValueClient *client = (KeyValueClient *)store;
 117     delete client;
 118 }
 119
 120 static void kvstore_submit(gpointer store, BlueSkyStoreAsync *async)
 121 {
 122     KeyValueClient *client = (KeyValueClient *)store;
 123
 124     g_return_if_fail(async->status == ASYNC_NEW);
 125     g_return_if_fail(async->op != STORE_OP_NONE);
 126
 127     switch (async->op) {
 128     case STORE_OP_GET:
 129     case STORE_OP_PUT:
 130         async->status = ASYNC_PENDING;
 131         bluesky_store_async_ref(async);
 132         g_thread_pool_push(thread_pool, async, NULL);
 133         break;
 134
 135     default:
 136         g_warning("Uknown operation type for kvstore: %d\n", async->op);
 137         bluesky_store_async_mark_complete(async);
 138         break;
 139     }
 140 }
 141
 142 static void kvstore_cleanup(gpointer store, BlueSkyStoreAsync *async)
 143 {
 144     KeyValueClient *client = (KeyValueClient *)store;
 145 }
 146
 147 static BlueSkyStoreImplementation store_impl = {
 148     kvstore_new,
 149     kvstore_destroy,
 150     kvstore_submit,
 151     kvstore_cleanup,
 152 };
 153
 154 extern "C" void bluesky_store_init_kv(void)
 155 {
 156     bluesky_store_register(&store_impl, "kv");
 157 }