X-Git-Url: http://git.vrable.net/?a=blobdiff_plain;f=bluesky%2Fcloudlog.c;h=6e57d0533247d8241aca6fb36624f82fb49a6dbe;hb=83fd6b61a6e092a22d4d5e59ed95f05f5e287f11;hp=e1fd44af4bb01e55c5f85c08108f4ab7eae0217a;hpb=db4915b4f7fc8f74958c4d1891dc69b76cbbe383;p=bluesky.git diff --git a/bluesky/cloudlog.c b/bluesky/cloudlog.c index e1fd44a..6e57d05 100644 --- a/bluesky/cloudlog.c +++ b/bluesky/cloudlog.c @@ -13,13 +13,9 @@ #include "bluesky-private.h" -/* The locations hash table in the file system is used to map objects to their locations. Objects are named using 128- */ - -typedef struct { - BlueSkyCloudID id; - - BlueSkyCloudPointer *cloud_loc; -} BlueSkyLocationEntry; +// Rough size limit for a log segment. This is not a firm limit and there are +// no absolute guarantees on the size of a log segment. +#define CLOUDLOG_SEGMENT_SIZE (4 << 20) BlueSkyCloudID bluesky_cloudlog_new_id() { @@ -86,36 +82,90 @@ BlueSkyCloudLog *bluesky_cloudlog_new(BlueSkyFS *fs) { BlueSkyCloudLog *log = g_new0(BlueSkyCloudLog, 1); + log->lock = g_mutex_new(); + log->cond = g_cond_new(); log->fs = fs; log->type = LOGTYPE_UNKNOWN; log->id = bluesky_cloudlog_new_id(); - log->pointers = g_array_new(FALSE, TRUE, sizeof(BlueSkyCloudID)); + log->links = g_array_new(FALSE, TRUE, sizeof(BlueSkyCloudLog *)); g_atomic_int_set(&log->refcount, 1); return log; } +/* Helper function for updating memory usage statistics for a filesystem (the + * cache_log_* variables). This will increment (type=1) or decrement (type=-1) + * the counter associated with the current state of the cloud log item. The + * item should be locked or otherwise protected from concurrent access. */ +void bluesky_cloudlog_stats_update(BlueSkyCloudLog *log, int type) +{ + BlueSkyFS *fs = log->fs; + + if (log->location_flags & CLOUDLOG_CLOUD) { + g_atomic_int_add(&fs->cache_log_cloud, type); + } else if (log->location_flags & CLOUDLOG_JOURNAL) { + g_atomic_int_add(&fs->cache_log_journal, type); + } else if (log->pending_write & CLOUDLOG_JOURNAL) { + g_atomic_int_add(&fs->cache_log_journal, type); + } else if (log->data != NULL) { + g_atomic_int_add(&fs->cache_log_dirty, type); + } +} + +/* The reference held by the hash table does not count towards the reference + * count. When a new object is created, it initially has a reference count of + * 1 for the creator, and similarly fetching an item from the hash table will + * also create a reference. If the reference count drops to zero, + * bluesky_cloudlog_unref attempts to remove the object from the hash + * table--but there is a potential race since another thread might read the + * object from the hash table at the same time. So an object with a reference + * count of zero may still be resurrected, in which case we need to abort the + * destruction. Once the object is gone from the hash table, and if the + * reference count is still zero, it can actually be deleted. */ void bluesky_cloudlog_ref(BlueSkyCloudLog *log) { + if (log == NULL) + return; + g_atomic_int_inc(&log->refcount); } void bluesky_cloudlog_unref(BlueSkyCloudLog *log) { + if (log == NULL) + return; + if (g_atomic_int_dec_and_test(&log->refcount)) { - g_print("Cloud log refcount dropped to zero.\n"); + BlueSkyFS *fs = log->fs; + + g_mutex_lock(fs->lock); + if (g_atomic_int_get(&log->refcount) > 0) { + g_mutex_unlock(fs->lock); + return; + } + + g_hash_table_remove(fs->locations, &log->id); + g_mutex_unlock(fs->lock); + + bluesky_cloudlog_stats_update(log, -1); + log->type = LOGTYPE_INVALID; + g_mutex_free(log->lock); + g_cond_free(log->cond); + for (int i = 0; i < log->links->len; i++) { + BlueSkyCloudLog *c = g_array_index(log->links, + BlueSkyCloudLog *, i); + bluesky_cloudlog_unref(c); + } + g_array_unref(log->links); + bluesky_string_unref(log->data); + g_free(log); } } /* Start a write of the object to the local log. */ -BlueSkyLogItem *bluesky_cloudlog_sync(BlueSkyCloudLog *log) +void bluesky_cloudlog_sync(BlueSkyCloudLog *log) { - BlueSkyLogItem *log_item = bluesky_log_item_new(); - log_item->key = bluesky_cloudlog_id_to_string(log->id); - log_item->data = log->data; - bluesky_string_ref(log->data); - bluesky_log_item_submit(log_item, log->fs->log); - return log_item; + bluesky_log_item_submit(log, log->fs->log); } /* Add the given entry to the global hash table containing cloud log entries. @@ -144,31 +194,46 @@ struct log_footer { struct logref refs[0]; }; +/* Ensure that a cloud log item is loaded in memory, and if not read it in. + * TODO: Make asynchronous, and make this also fetch from the cloud. Right now + * we only read from the log. Log item must be locked. */ +void bluesky_cloudlog_fetch(BlueSkyCloudLog *log) +{ + if (log->data != NULL) + return; + + g_assert((log->location_flags | log->pending_write) & CLOUDLOG_JOURNAL); + + bluesky_cloudlog_stats_update(log, -1); + log->data = bluesky_log_map_object(log->fs->log, log->log_seq, + log->log_offset, log->log_size); + bluesky_cloudlog_stats_update(log, 1); + + g_cond_broadcast(log->cond); +} + BlueSkyCloudPointer bluesky_cloudlog_serialize(BlueSkyCloudLog *log, - BlueSkyCloudLogState *state) + BlueSkyFS *fs) { + BlueSkyCloudLogState *state = fs->log_state; + if (log->location_flags & CLOUDLOG_CLOUD) { return log->location; } - g_print("Flushing object %s to cloud...\n", - bluesky_cloudlog_id_to_string(log->id)); - - for (int i = 0; i < log->pointers->len; i++) { - BlueSkyCloudID id = g_array_index(log->pointers, BlueSkyCloudID, i); - g_print(" ...checking reference %s...\n", - bluesky_cloudlog_id_to_string(id)); - g_mutex_lock(log->fs->lock); - BlueSkyCloudLog *log2 - = (BlueSkyCloudLog *)g_hash_table_lookup(log->fs->locations, &id); - // TODO: refcount - g_mutex_unlock(log->fs->lock); - g_assert(log2 != NULL); - bluesky_cloudlog_serialize(log2, state); + for (int i = 0; i < log->links->len; i++) { + BlueSkyCloudLog *ref = g_array_index(log->links, + BlueSkyCloudLog *, i); + if (ref != NULL) + bluesky_cloudlog_serialize(ref, fs); } + g_mutex_lock(log->lock); + bluesky_cloudlog_fetch(log); g_assert(log->data != NULL); + bluesky_cloudlog_stats_update(log, -1); + log->location = state->location; log->location.offset = state->data->len; log->location.size @@ -185,41 +250,25 @@ BlueSkyCloudPointer bluesky_cloudlog_serialize(BlueSkyCloudLog *log, g_string_append_len(state->data, log->data->data, log->data->len); log->location_flags |= CLOUDLOG_CLOUD; + bluesky_cloudlog_stats_update(log, 1); + g_mutex_unlock(log->lock); - return log->location; -} - -static void find_inodes(gpointer key, gpointer value, gpointer user_data) -{ - BlueSkyCloudLogState *state = (BlueSkyCloudLogState *)user_data; - BlueSkyCloudLog *item = (BlueSkyCloudLog *)value; - - if (item->type != LOGTYPE_INODE) - return; + if (state->data->len > CLOUDLOG_SEGMENT_SIZE) + bluesky_cloudlog_flush(fs); - bluesky_cloudlog_ref(item); - state->inode_list = g_list_prepend(state->inode_list, item); + return log->location; } -void bluesky_cloudlog_write_log(BlueSkyFS *fs) +/* Finish up a partially-written cloud log segment and flush it to storage. */ +void bluesky_cloudlog_flush(BlueSkyFS *fs) { - g_print("Starting cloudlog write...\n"); - BlueSkyCloudLogState *state = fs->log_state; - state->data = g_string_new(""); + if (state->data == NULL || state->data->len == 0) + return; - g_mutex_lock(fs->lock); - g_hash_table_foreach(fs->locations, find_inodes, state); - g_mutex_unlock(fs->lock); + /* TODO: Append some type of commit record to the log segment? */ - while (state->inode_list != NULL) { - BlueSkyCloudLog *log = (BlueSkyCloudLog *)state->inode_list->data; - bluesky_cloudlog_serialize(log, state); - state->inode_list = g_list_delete_link(state->inode_list, - state->inode_list); - } - - g_print("Serialized %zd bytes of data\n", state->data->len); + g_print("Serializing %zd bytes of data to cloud\n", state->data->len); BlueSkyStoreAsync *async = bluesky_store_async_new(fs->store); async->op = STORE_OP_PUT; @@ -231,7 +280,7 @@ void bluesky_cloudlog_write_log(BlueSkyFS *fs) bluesky_store_async_wait(async); bluesky_store_async_unref(async); - state->data = NULL; state->location.sequence++; state->location.offset = 0; + state->data = g_string_new(""); }