X-Git-Url: http://git.vrable.net/?a=blobdiff_plain;f=bluesky%2Fcloudlog.c;h=dd16976f79aad9d6433b06820399cd877de96324;hb=8ff0fd08d6e1cc97cdb7e94b7cd97dc28c29e674;hp=f0fa4855f2eb4f2676a8a74acc6733d990bd67e3;hpb=207d6851cd0b6d4a5fb1ebfd13fc131ca935c364;p=bluesky.git diff --git a/bluesky/cloudlog.c b/bluesky/cloudlog.c index f0fa485..dd16976 100644 --- a/bluesky/cloudlog.c +++ b/bluesky/cloudlog.c @@ -3,7 +3,29 @@ * Copyright (C) 2009 The Regents of the University of California * Written by Michael Vrable * - * TODO: Licensing + * Redistribution and use in source and binary forms, with or without + * modification, are permitted provided that the following conditions + * are met: + * 1. Redistributions of source code must retain the above copyright + * notice, this list of conditions and the following disclaimer. + * 2. Redistributions in binary form must reproduce the above copyright + * notice, this list of conditions and the following disclaimer in the + * documentation and/or other materials provided with the distribution. + * 3. Neither the name of the University nor the names of its contributors + * may be used to endorse or promote products derived from this software + * without specific prior written permission. + * + * THIS SOFTWARE IS PROVIDED BY THE REGENTS AND CONTRIBUTORS ``AS IS'' AND + * ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE + * IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE + * ARE DISCLAIMED. IN NO EVENT SHALL THE REGENTS OR CONTRIBUTORS BE LIABLE + * FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL + * DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS + * OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) + * HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT + * LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY + * OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF + * SUCH DAMAGE. */ #include @@ -17,6 +39,9 @@ // 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; @@ -189,12 +214,6 @@ void bluesky_cloudlog_unref_delayed(BlueSkyCloudLog *log) 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 @@ -268,6 +287,25 @@ BlueSkyCloudLog *bluesky_cloudlog_get(BlueSkyFS *fs, BlueSkyCloudID id) 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? */ @@ -276,6 +314,14 @@ void bluesky_cloudlog_prefetch(BlueSkyCloudLog *item) 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_options.disable_read_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; @@ -446,8 +492,11 @@ BlueSkyCloudPointer bluesky_cloudlog_serialize(BlueSkyCloudLog *log, 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; } @@ -478,6 +527,13 @@ static void cloudlog_flush_complete(BlueSkyStoreAsync *async, 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"); @@ -522,6 +578,12 @@ void bluesky_cloudlog_flush(BlueSkyFS *fs) 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); @@ -538,12 +600,12 @@ void bluesky_cloudlog_flush(BlueSkyFS *fs) 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(""); @@ -616,3 +678,11 @@ void bluesky_cloudlog_decrypt(char *segment, size_t len, 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); +}