X-Git-Url: http://git.vrable.net/?a=blobdiff_plain;f=bluesky%2Fcloudlog.c;h=c99d181a11cc6f8acbebe5feadfd0b1db19d1183;hb=e49fed25d000a51b316b480c1bfb1b00c1f5d345;hp=0c5db98b5f094dbaf0e8cdc9f40167ff0ef0d729;hpb=810fdc7bdf0cd13aaa7c88d3c5af5aae24e77254;p=bluesky.git diff --git a/bluesky/cloudlog.c b/bluesky/cloudlog.c index 0c5db98..c99d181 100644 --- a/bluesky/cloudlog.c +++ b/bluesky/cloudlog.c @@ -6,6 +6,7 @@ * TODO: Licensing */ +#include #include #include #include @@ -27,6 +28,40 @@ BlueSkyCloudID bluesky_cloudlog_new_id() return id; } +gchar *bluesky_cloudlog_id_to_string(BlueSkyCloudID id) +{ + char buf[sizeof(BlueSkyCloudID) * 2 + 1]; + buf[0] = '\0'; + + for (int i = 0; i < sizeof(BlueSkyCloudID); i++) { + sprintf(&buf[2*i], "%02x", (uint8_t)(id.bytes[i])); + } + + return g_strdup(buf); +} + +BlueSkyCloudID bluesky_cloudlog_id_from_string(const gchar *idstr) +{ + BlueSkyCloudID id; + memset(&id, 0, sizeof(id)); + for (int i = 0; i < 2*sizeof(BlueSkyCloudID); i++) { + char c = idstr[i]; + if (c == '\0') { + g_warning("Short cloud id: %s\n", idstr); + break; + } + int val = 0; + if (c >= '0' && c <= '9') + val = c - '0'; + else if (c >= 'a' && c <= 'f') + val = c - 'a' + 10; + else + g_warning("Bad character in cloud id: %s\n", idstr); + id.bytes[i / 2] += val << (i % 2 ? 0 : 4); + } + return id; +} + gboolean bluesky_cloudlog_equal(gconstpointer a, gconstpointer b) { BlueSkyCloudID *id1 = (BlueSkyCloudID *)a, *id2 = (BlueSkyCloudID *)b; @@ -51,9 +86,12 @@ 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)); g_atomic_int_set(&log->refcount, 1); return log; @@ -61,25 +99,26 @@ BlueSkyCloudLog *bluesky_cloudlog_new(BlueSkyFS *fs) 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"); } } /* 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 = g_strdup("cloudlog"); - 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. @@ -90,3 +129,133 @@ void bluesky_cloudlog_insert(BlueSkyCloudLog *log) g_hash_table_insert(log->fs->locations, &log->id, log); g_mutex_unlock(log->fs->lock); } + +struct log_header { + char magic[4]; + uint32_t size; + BlueSkyCloudID id; + uint32_t pointer_count; +} __attribute__((packed)); + +struct logref { + BlueSkyCloudID id; + BlueSkyCloudPointer location; +} __attribute__((packed)); + +struct log_footer { + char refmagic[4]; + struct logref refs[0]; +}; + +BlueSkyCloudPointer bluesky_cloudlog_serialize(BlueSkyCloudLog *log, + BlueSkyCloudLogState *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); + } + + g_assert(log->data != NULL); + + log->location = state->location; + log->location.offset = state->data->len; + log->location.size + = sizeof(struct log_header) + sizeof(BlueSkyCloudID) * 0 + + log->data->len; + + struct log_header header; + memcpy(header.magic, "AgI ", 4); + header.size = GUINT32_TO_LE(log->location.size); + header.id = log->id; + header.pointer_count = GUINT32_TO_LE(0); + + g_string_append_len(state->data, (const char *)&header, sizeof(header)); + g_string_append_len(state->data, log->data->data, log->data->len); + + log->location_flags |= CLOUDLOG_CLOUD; + + 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; + + bluesky_cloudlog_ref(item); + state->inode_list = g_list_prepend(state->inode_list, item); +} + +void bluesky_cloudlog_write_log(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); + 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) + return; + + g_print("Re-mapping log entry %d/%d/%d...\n", + log->log_seq, log->log_offset, log->log_size); + + g_assert(log->location_flags & CLOUDLOG_JOURNAL); + + log->data = bluesky_log_map_object(log->fs->log, log->log_seq, + log->log_offset, log->log_size); + + g_cond_broadcast(log->cond); +}