X-Git-Url: http://git.vrable.net/?a=blobdiff_plain;f=bluesky%2Fcloudlog.c;h=3809b00b0d8faf6ed59bff7ac7648de3a04f13ee;hb=97517d1bbc647846003d1a49b6ae97c69d856b90;hp=30c2db0f947f1f15f6b2249d166c8d5c5d1e7f09;hpb=bf5f9f2e43078f7d31858be259b9c60a8c08a096;p=bluesky.git diff --git a/bluesky/cloudlog.c b/bluesky/cloudlog.c index 30c2db0..3809b00 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() { @@ -97,6 +93,25 @@ BlueSkyCloudLog *bluesky_cloudlog_new(BlueSkyFS *fs) 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 @@ -132,6 +147,7 @@ void bluesky_cloudlog_unref(BlueSkyCloudLog *log) 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); @@ -178,32 +194,62 @@ 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; + + if ((log->location_flags | log->pending_write) & CLOUDLOG_JOURNAL) { + bluesky_cloudlog_stats_update(log, -1); + log->data = bluesky_log_map_object(log->fs, -1, log->log_seq, + log->log_offset, log->log_size); + bluesky_cloudlog_stats_update(log, 1); + } else if (log->location_flags & CLOUDLOG_CLOUD) { + bluesky_cloudlog_stats_update(log, -1); + log->data = bluesky_log_map_object(log->fs, log->location.directory, + log->location.sequence, + log->location.offset, + log->location.size); + bluesky_cloudlog_stats_update(log, 1); + } else { + g_error("Unable to fetch cloudlog entry!"); + } + + 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->links->len; i++) { BlueSkyCloudLog *ref = g_array_index(log->links, BlueSkyCloudLog *, i); if (ref != NULL) - bluesky_cloudlog_serialize(ref, state); + 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); + + /* TODO: Right now offset/size are set to the raw data, but we should add + * header parsing to the code which loads objects back in. */ log->location = state->location; - log->location.offset = state->data->len; - log->location.size - = sizeof(struct log_header) + sizeof(BlueSkyCloudID) * 0 - + log->data->len; + log->location.offset = state->data->len + sizeof(struct log_header); + log->location.size = log->data->len; + /* = sizeof(struct log_header) + sizeof(BlueSkyCloudID) * 0 + + log->data->len; */ struct log_header header; memcpy(header.magic, "AgI ", 4); @@ -215,76 +261,37 @@ 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) { BlueSkyCloudLogState *state = fs->log_state; - if (state->data == NULL) - state->data = g_string_new(""); - - g_mutex_lock(fs->lock); - g_hash_table_foreach(fs->locations, find_inodes, state); - g_mutex_unlock(fs->lock); - - while (state->inode_list != NULL) { - BlueSkyCloudLog *log = (BlueSkyCloudLog *)state->inode_list->data; - bluesky_cloudlog_serialize(log, state); - bluesky_cloudlog_unref(log); - state->inode_list = g_list_delete_link(state->inode_list, - state->inode_list); - } - - if (state->data->len > 0) { - g_print("Serialized %zd bytes of data to cloud\n", state->data->len); - - BlueSkyStoreAsync *async = bluesky_store_async_new(fs->store); - async->op = STORE_OP_PUT; - async->key = g_strdup_printf("log-%08d-%08d", - state->location.directory, - state->location.sequence); - async->data = bluesky_string_new_from_gstring(state->data); - bluesky_store_async_submit(async); - bluesky_store_async_wait(async); - bluesky_store_async_unref(async); - - state->location.sequence++; - state->location.offset = 0; - } - - state->data = NULL; -} - -/* 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) + if (state->data == NULL || state->data->len == 0) return; - g_print("Re-mapping log entry %d/%d/%d...\n", - log->log_seq, log->log_offset, log->log_size); + /* TODO: Append some type of commit record to the log segment? */ - g_assert(log->location_flags & CLOUDLOG_JOURNAL); + g_print("Serializing %zd bytes of data to cloud\n", state->data->len); - log->data = bluesky_log_map_object(log->fs->log, log->log_seq, - log->log_offset, log->log_size); + BlueSkyStoreAsync *async = bluesky_store_async_new(fs->store); + async->op = STORE_OP_PUT; + async->key = g_strdup_printf("log-%08d-%08d", + state->location.directory, + state->location.sequence); + async->data = bluesky_string_new_from_gstring(state->data); + bluesky_store_async_submit(async); + //bluesky_store_async_wait(async); + bluesky_store_async_unref(async); - g_cond_broadcast(log->cond); + state->location.sequence++; + state->location.offset = 0; + state->data = g_string_new(""); }