Drop some debugging messages.
[bluesky.git] / bluesky / store.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 <glib.h>
11 #include <string.h>
12
13 #include "bluesky-private.h"
14
15 /* Interaction with cloud storage.  We expose very simple GET/PUT style
16  * interface, which different backends can implement.  Available backends
17  * (will) include Amazon S3 and a simple local store for testing purposes.
18  * Operations may be performed asynchronously. */
19
20 struct _BlueSkyStore {
21     const BlueSkyStoreImplementation *impl;
22     gpointer handle;
23
24     GMutex *lock;
25     GCond *cond_idle;
26     int pending;                /* Count of operations not yet complete. */
27 };
28
29 GHashTable *store_implementations;
30
31 /* Thread pool for calling notifier functions when an operation completes.
32  * These are called in a separate thread for locking reasons: we want to call
33  * the notifiers without the lock on the async object held, but completion
34  * occurs when the lock is held--so we need some way to defer the call.  This
35  * isn't really optimal from a cache-locality standpoint. */
36 static GThreadPool *notifier_thread_pool;
37
38 void bluesky_store_register(const BlueSkyStoreImplementation *impl,
39                             const gchar *name)
40 {
41     g_hash_table_insert(store_implementations, g_strdup(name), (gpointer)impl);
42 }
43
44 BlueSkyStore *bluesky_store_new(const gchar *type)
45 {
46     const BlueSkyStoreImplementation *impl;
47
48     impl = g_hash_table_lookup(store_implementations, type);
49     if (impl == NULL)
50         return NULL;
51
52     gpointer handle = impl->create();
53     if (handle == NULL)
54         return NULL;
55
56     BlueSkyStore *store = g_new(BlueSkyStore, 1);
57     store->impl = impl;
58     store->handle = handle;
59     store->lock = g_mutex_new();
60     store->cond_idle = g_cond_new();
61     store->pending = 0;
62     return store;
63 }
64
65 void bluesky_store_free(BlueSkyStore *store)
66 {
67     store->impl->destroy(store->handle);
68     g_free(store);
69 }
70
71 BlueSkyStoreAsync *bluesky_store_async_new(BlueSkyStore *store)
72 {
73     BlueSkyStoreAsync *async;
74
75     async = g_new(BlueSkyStoreAsync, 1);
76     async->store = store;
77     async->lock = g_mutex_new();
78     async->completion_cond = g_cond_new();
79     async->refcount = 1;
80     async->status = ASYNC_NEW;
81     async->op = STORE_OP_NONE;
82     async->key = NULL;
83     async->data = NULL;
84     async->result = -1;
85     async->notifiers = NULL;
86     async->store_private = NULL;
87
88     return async;
89 }
90
91 void bluesky_store_async_ref(BlueSkyStoreAsync *async)
92 {
93     if (async == NULL)
94         return;
95
96     g_return_if_fail(g_atomic_int_get(&async->refcount) > 0);
97
98     g_atomic_int_inc(&async->refcount);
99 }
100
101 void bluesky_store_async_unref(BlueSkyStoreAsync *async)
102 {
103     if (async == NULL)
104         return;
105
106     if (g_atomic_int_dec_and_test(&async->refcount)) {
107         async->store->impl->cleanup(async->store->handle, async);
108         g_mutex_free(async->lock);
109         g_cond_free(async->completion_cond);
110         g_free(async->key);
111         bluesky_string_unref(async->data);
112         g_free(async);
113         g_log("bluesky/store", G_LOG_LEVEL_DEBUG,
114               "freeing async");
115     }
116 }
117
118 /* Block until the given operation has completed. */
119 void bluesky_store_async_wait(BlueSkyStoreAsync *async)
120 {
121     g_return_if_fail(async != NULL);
122     g_mutex_lock(async->lock);
123
124     if (async->status == ASYNC_NEW) {
125         g_error("bluesky_store_async_wait on a new async object!\n");
126         g_mutex_unlock(async->lock);
127         return;
128     }
129
130     while (async->status != ASYNC_COMPLETE) {
131         g_cond_wait(async->completion_cond, async->lock);
132     }
133
134     g_mutex_unlock(async->lock);
135 }
136
137 /* Add a notifier function to be called when the operation completes. */
138 void bluesky_store_async_add_notifier(BlueSkyStoreAsync *async,
139                                       GFunc func, gpointer user_data)
140 {
141     struct BlueSkyNotifierList *nl = g_new(struct BlueSkyNotifierList, 1);
142     nl->next = async->notifiers;
143     nl->func = func;
144     nl->async = async; bluesky_store_async_ref(async);
145     nl->user_data = user_data;
146     if (async->status == ASYNC_COMPLETE) {
147         g_thread_pool_push(notifier_thread_pool, nl, NULL);
148     } else {
149         async->notifiers = nl;
150     }
151 }
152
153 /* Mark an asynchronous operation as complete.  This should only be called by
154  * the store implementations.  The lock should be held when calling this
155  * function.  Any notifier functions will be called, but in a separate thread
156  * and without the lock held. */
157 void bluesky_store_async_mark_complete(BlueSkyStoreAsync *async)
158 {
159     g_return_if_fail(async->status != ASYNC_COMPLETE);
160
161     g_mutex_lock(async->store->lock);
162     async->store->pending--;
163     if (async->store->pending == 0)
164         g_cond_broadcast(async->store->cond_idle);
165     g_mutex_unlock(async->store->lock);
166
167     async->status = ASYNC_COMPLETE;
168     g_cond_broadcast(async->completion_cond);
169
170     while (async->notifiers != NULL) {
171         struct BlueSkyNotifierList *nl = async->notifiers;
172         async->notifiers = nl->next;
173         g_thread_pool_push(notifier_thread_pool, nl, NULL);
174     }
175 }
176
177 void bluesky_store_async_submit(BlueSkyStoreAsync *async)
178 {
179     BlueSkyStore *store = async->store;
180
181     /* Barriers are handled specially, and not handed down the storage
182      * implementation layer. */
183     if (async->op == STORE_OP_BARRIER) {
184         async->status = ASYNC_RUNNING;
185         if (GPOINTER_TO_INT(async->store_private) == 0)
186             bluesky_store_async_mark_complete(async);
187         return;
188     }
189
190     g_log("bluesky/store", G_LOG_LEVEL_DEBUG, "submit: %s %s",
191           async->op == STORE_OP_GET ? "GET"
192             : async->op == STORE_OP_PUT ? "PUT"
193             : async->op == STORE_OP_DELETE ? "DELETE" : "???",
194           async->key);
195
196     g_mutex_lock(async->store->lock);
197     async->store->pending++;
198     g_mutex_unlock(async->store->lock);
199
200     store->impl->submit(store->handle, async);
201
202     if (bluesky_options.synchronous_stores)
203         bluesky_store_async_wait(async);
204 }
205
206 static void op_complete(gpointer a, gpointer b)
207 {
208     BlueSkyStoreAsync *async = (BlueSkyStoreAsync *)a;
209     BlueSkyStoreAsync *barrier = (BlueSkyStoreAsync *)b;
210
211     g_mutex_lock(barrier->lock);
212     barrier->store_private
213         = GINT_TO_POINTER(GPOINTER_TO_INT(barrier->store_private) - 1);
214     if (GPOINTER_TO_INT(barrier->store_private) == 0
215             && barrier->status != ASYNC_NEW) {
216         bluesky_store_async_mark_complete(barrier);
217     }
218     g_mutex_unlock(barrier->lock);
219 }
220
221 /* Add the given operation to the barrier.  The barrier will not complete until
222  * all operations added to it have completed. */
223 void bluesky_store_add_barrier(BlueSkyStoreAsync *barrier,
224                                BlueSkyStoreAsync *async)
225 {
226     g_return_if_fail(barrier->op == STORE_OP_BARRIER);
227     barrier->store_private
228         = GINT_TO_POINTER(GPOINTER_TO_INT(barrier->store_private) + 1);
229     bluesky_store_async_add_notifier(async, op_complete, barrier);
230 }
231
232 static void notifier_task(gpointer n, gpointer s)
233 {
234     struct BlueSkyNotifierList *notifier = (struct BlueSkyNotifierList *)n;
235
236     notifier->func(notifier->async, notifier->user_data);
237     bluesky_store_async_unref(notifier->async);
238     g_free(notifier);
239 }
240
241 void bluesky_store_sync(BlueSkyStore *store)
242 {
243     g_mutex_lock(store->lock);
244     g_print("Waiting for pending store operations to complete...\n");
245     while (store->pending > 0) {
246         g_cond_wait(store->cond_idle, store->lock);
247     }
248     g_mutex_unlock(store->lock);
249     g_print("Operations are complete.\n");
250 }
251
252 /* Convenience wrappers that perform a single operation synchronously. */
253 BlueSkyRCStr *bluesky_store_get(BlueSkyStore *store, const gchar *key)
254 {
255     BlueSkyStoreAsync *async = bluesky_store_async_new(store);
256     async->op = STORE_OP_GET;
257     async->key = g_strdup(key);
258     bluesky_store_async_submit(async);
259
260     bluesky_store_async_wait(async);
261
262     BlueSkyRCStr *data = async->data;
263     bluesky_string_ref(data);
264     bluesky_store_async_unref(async);
265     return data;
266 }
267
268 void bluesky_store_put(BlueSkyStore *store,
269                        const gchar *key, BlueSkyRCStr *val)
270 {
271     BlueSkyStoreAsync *async = bluesky_store_async_new(store);
272     async->op = STORE_OP_PUT;
273     async->key = g_strdup(key);
274     bluesky_string_ref(val);
275     async->data = val;
276     bluesky_store_async_submit(async);
277
278     bluesky_store_async_wait(async);
279     bluesky_store_async_unref(async);
280 }
281
282 #if 0
283 /* Simple in-memory data store for test purposes. */
284 typedef struct {
285     GMutex *lock;
286
287     /* TODO: A hashtable isn't optimal for list queries... */
288     GHashTable *store;
289 } MemStore;
290
291 static gpointer memstore_create()
292 {
293     MemStore *store = g_new(MemStore, 1);
294     store->lock = g_mutex_new();
295     store->store = g_hash_table_new_full(g_str_hash, g_str_equal,
296                                          g_free,
297                                          (GDestroyNotify)bluesky_string_unref);
298
299     return (gpointer)store;
300 }
301
302 static void memstore_destroy(gpointer store)
303 {
304     /* TODO */
305 }
306
307 static BlueSkyRCStr *memstore_get(gpointer st, const gchar *key)
308 {
309     MemStore *store = (MemStore *)st;
310     BlueSkyRCStr *s = g_hash_table_lookup(store->store, key);
311     if (s != NULL)
312         bluesky_string_ref(s);
313     return s;
314 }
315
316 static void memstore_put(gpointer s, const gchar *key, BlueSkyRCStr *val)
317 {
318     MemStore *store = (MemStore *)s;
319     bluesky_string_ref(val);
320     g_hash_table_insert(store->store, g_strdup(key), val);
321 }
322
323 static BlueSkyStoreImplementation memstore_impl = {
324     .create = memstore_create,
325     .destroy = memstore_destroy,
326     .get = memstore_get,
327     .put = memstore_put,
328 };
329
330 /* Store implementation which writes data as files to disk. */
331 static gpointer filestore_create()
332 {
333     return GINT_TO_POINTER(1);
334 }
335
336 static void filestore_destroy()
337 {
338 }
339
340 static BlueSkyRCStr *filestore_get(gpointer s, const gchar *key)
341 {
342     gchar *contents = NULL;
343     gsize length;
344     GError *error = NULL;
345
346     g_file_get_contents(key, &contents, &length, &error);
347     if (contents == NULL)
348         return NULL;
349
350     return bluesky_string_new(contents, length);
351 }
352
353 static void filestore_put(gpointer s, const gchar *key, BlueSkyRCStr *val)
354 {
355     g_file_set_contents(key, val->data, val->len, NULL);
356 }
357
358 static BlueSkyStoreImplementation filestore_impl = {
359     .create = filestore_create,
360     .destroy = filestore_destroy,
361     .get = filestore_get,
362     .put = filestore_put,
363 };
364 #endif
365
366 void bluesky_store_init()
367 {
368     store_implementations = g_hash_table_new(g_str_hash, g_str_equal);
369     notifier_thread_pool = g_thread_pool_new(notifier_task, NULL, -1, FALSE,
370                                              NULL);
371     //bluesky_store_register(&memstore_impl, "mem");
372     //bluesky_store_register(&filestore_impl, "file");
373 }