// no absolute guarantees on the size of a log segment.
#define CLOUDLOG_SEGMENT_SIZE (4 << 20)
+// Maximum number of segments to attempt to upload concurrently
+int cloudlog_concurrent_uploads = 32;
+
BlueSkyCloudID bluesky_cloudlog_new_id()
{
BlueSkyCloudID id;
g_async_queue_push(log->fs->unref_queue, log);
}
-void bluesky_cloudlog_threads_init(BlueSkyFS *fs)
-{
- fs->unref_queue = g_async_queue_new();
- g_thread_create(cloudlog_unref_thread, fs->unref_queue, FALSE, NULL);
-}
-
/* Erase the information contained within the in-memory cloud log
* representation. This does not free up the item itself, but frees the data
* and references to other log items and resets the type back to unknown. If
return item;
}
+/* Work to fetch a cloudlog item in a background thread. The item will be
+ * locked while the fetch is in progress and unlocked when it completes. */
+static GThreadPool *fetch_pool;
+
+static void background_fetch_task(gpointer p, gpointer unused)
+{
+ BlueSkyCloudLog *item = (BlueSkyCloudLog *)p;
+
+ g_mutex_lock(item->lock);
+ g_mutex_unlock(item->lock);
+ bluesky_cloudlog_unref(item);
+}
+
+void bluesky_cloudlog_background_fetch(BlueSkyCloudLog *item)
+{
+ bluesky_cloudlog_ref(item);
+ g_thread_pool_push(fetch_pool, item, NULL);
+}
+
/* Attempt to prefetch a cloud log item. This does not guarantee that it will
* be made available, but does make it more likely that a future call to
* bluesky_cloudlog_fetch will complete quickly. Item must be locked? */
if (item->data != NULL)
return;
+ /* When operating in a non log-structured mode, simply start a background
+ * fetch immediately when asked to prefetch. */
+ if (bluesky_options.disable_aggregation) {
+ bluesky_cloudlog_background_fetch(item);
+ return;
+ }
+
/* TODO: Some of the code here is duplicated with bluesky_log_map_object.
* Refactor to fix that. */
BlueSkyFS *fs = item->fs;
bluesky_cloudlog_ref(log);
g_mutex_unlock(log->lock);
- if (state->data->len > CLOUDLOG_SEGMENT_SIZE)
+ if (state->data->len > CLOUDLOG_SEGMENT_SIZE
+ || bluesky_options.disable_aggregation)
+ {
bluesky_cloudlog_flush(fs);
+ }
return log->location;
}
g_slist_free(record->items);
record->items = NULL;
record->complete = TRUE;
+
+ BlueSkyCloudLogState *state = record->fs->log_state;
+ g_mutex_lock(state->uploads_pending_lock);
+ state->uploads_pending--;
+ g_cond_broadcast(state->uploads_pending_cond);
+ g_mutex_unlock(state->uploads_pending_lock);
+
g_cond_broadcast(record->cond);
} else {
g_print("Write should be resubmitted...\n");
if (state->data == NULL || state->data->len == 0)
return;
+ g_mutex_lock(state->uploads_pending_lock);
+ while (state->uploads_pending > cloudlog_concurrent_uploads)
+ g_cond_wait(state->uploads_pending_cond, state->uploads_pending_lock);
+ state->uploads_pending++;
+ g_mutex_unlock(state->uploads_pending_lock);
+
/* TODO: Append some type of commit record to the log segment? */
g_print("Serializing %zd bytes of data to cloud\n", state->data->len);
state->location.directory,
state->location.sequence);
+ state->pending_segments = g_list_prepend(state->pending_segments, record);
+
/* Encryption of data and upload happen in the background, for additional
* parallelism when uploading large amounts of data. */
g_thread_create((GThreadFunc)cloud_flush_background, record, FALSE, NULL);
- state->pending_segments = g_list_prepend(state->pending_segments, record);
-
state->location.sequence++;
state->location.offset = 0;
state->data = g_string_new("");
remaining_size -= item_size;
}
}
+
+void bluesky_cloudlog_threads_init(BlueSkyFS *fs)
+{
+ fs->unref_queue = g_async_queue_new();
+ g_thread_create(cloudlog_unref_thread, fs->unref_queue, FALSE, NULL);
+ fetch_pool = g_thread_pool_new(background_fetch_task, NULL, 40, FALSE,
+ NULL);
+}