bluesky/store-bdb.c

   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 #include <stdint.h>
  10 #include <stdlib.h>
  11 #include <glib.h>
  12 #include <string.h>
  13 #include <db.h>
  14 #include <errno.h>
  15
  16 #include "bluesky-private.h"
  17 #include "libs3.h"
  18
  19 /* A storage layer that writes to Berkeley DB locally. */
  20
  21 typedef struct {
  22     GThreadPool *thread_pool;
  23     DB_ENV *env;
  24     DB *db;
  25 } BDBStore;
  26
  27 static void bdbstore_task(gpointer a, gpointer s)
  28 {
  29     int res;
  30     BlueSkyStoreAsync *async = (BlueSkyStoreAsync *)a;
  31     BDBStore *store = (BDBStore *)s;
  32
  33     async->status = ASYNC_RUNNING;
  34     async->exec_time = bluesky_now_hires();
  35
  36     DBT key;
  37     memset(&key, 0, sizeof(key));
  38
  39     key.data = async->key;
  40     key.size = strlen(async->key);
  41
  42     DBT value;
  43     memset(&value, 0, sizeof(value));
  44
  45     if (async->op == STORE_OP_GET) {
  46         value.flags = DB_DBT_MALLOC;
  47
  48         res = store->db->get(store->db, NULL, &key, &value, 0);
  49
  50         async->result = res;
  51         async->data = NULL;
  52
  53         if (res != 0) {
  54             fprintf(stderr, "BDB read failure: %s\n", db_strerror(res));
  55         } else {
  56             async->data = bluesky_string_new(value.data, value.size);
  57             async->result = 0;
  58         }
  59
  60     } else if (async->op == STORE_OP_PUT) {
  61         value.data = async->data->data;
  62         value.size = async->data->len;
  63
  64         res = store->db->put(store->db, NULL, &key, &value, 0);
  65
  66         if (res != 0) {
  67             fprintf(stderr, "BDB write failure: %s\n", db_strerror(res));
  68         }
  69
  70         async->result = 0;
  71     }
  72
  73     bluesky_store_async_mark_complete(async);
  74     bluesky_store_async_unref(async);
  75 }
  76
  77 static gpointer bdbstore_new(const gchar *path)
  78 {
  79     int res;
  80     BDBStore *store = g_new0(BDBStore, 1);
  81     store->thread_pool = g_thread_pool_new(bdbstore_task, store, 16, FALSE,
  82                                            NULL);
  83
  84     res = db_env_create(&store->env, 0);
  85
  86     if (res != 0) {
  87         fprintf(stderr, "db_env_create failure: %s\n", db_strerror(res));
  88         return NULL;
  89     }
  90
  91     res = store->env->open(store->env, path,
  92                            DB_CREATE | DB_RECOVER | DB_INIT_LOCK | DB_INIT_LOG
  93                             | DB_INIT_MPOOL | DB_INIT_TXN | DB_THREAD,
  94                            0644);
  95
  96     if (res != 0) {
  97         fprintf(stderr, "BDB open failure: %s\n",
  98                 db_strerror(res));
  99         return NULL;
 100     }
 101
 102     res = db_create(&store->db, store->env, 0);
 103
 104     if (res != 0) {
 105         fprintf(stderr, "DB create failed: %s\n", db_strerror(res));
 106         return NULL;
 107     }
 108
 109     uint32_t flags = DB_CREATE | DB_THREAD | DB_AUTO_COMMIT;
 110
 111     res = store->db->open(store->db,
 112                           NULL, /* TXN */
 113                           "store.db",
 114                           "store",
 115                           DB_BTREE,
 116                           flags,
 117                           0644);
 118
 119     if (res != 0) {
 120         fprintf(stderr, "DB open failed: %s\n",
 121                 db_strerror(res));
 122     }
 123
 124     return store;
 125 }
 126
 127 static void bdbstore_destroy(gpointer s)
 128 {
 129     BDBStore *store = (BDBStore *)store;
 130
 131     if (store->db) {
 132         store->db->close(store->db, 0);
 133     }
 134
 135     if (store->env) {
 136         store->env->close(store->env, 0);
 137     }
 138
 139     g_free(store);
 140 }
 141
 142 static void bdbstore_submit(gpointer s, BlueSkyStoreAsync *async)
 143 {
 144     BDBStore *store = (BDBStore *)s;
 145     g_return_if_fail(async->status == ASYNC_NEW);
 146     g_return_if_fail(async->op != STORE_OP_NONE);
 147
 148     switch (async->op) {
 149     case STORE_OP_GET:
 150     case STORE_OP_PUT:
 151         async->status = ASYNC_PENDING;
 152         bluesky_store_async_ref(async);
 153         g_thread_pool_push(store->thread_pool, async, NULL);
 154         break;
 155
 156     default:
 157         g_warning("Uknown operation type for BDBStore: %d\n", async->op);
 158         bluesky_store_async_mark_complete(async);
 159         break;
 160     }
 161 }
 162
 163 static void bdbstore_cleanup(gpointer store, BlueSkyStoreAsync *async)
 164 {
 165 }
 166
 167 static BlueSkyStoreImplementation store_impl = {
 168     .create = bdbstore_new,
 169     .destroy = bdbstore_destroy,
 170     .submit = bdbstore_submit,
 171     .cleanup = bdbstore_cleanup,
 172 };
 173
 174 void bluesky_store_init_bdb(void)
 175 {
 176     bluesky_store_register(&store_impl, "bdb");
 177 }
 178