X-Git-Url: http://git.vrable.net/?a=blobdiff_plain;f=bluesky%2Flog.c;h=1fe61769b92870b95022fcc8933b70d3a9e689d3;hb=7795e004d36e779abdf77f8771488595839e4bee;hp=714650607df8c13f5b79f3b6e2c2a99f151da3e5;hpb=e1ca9ea030f0adc562642adca4b18557b3c9fe9c;p=bluesky.git diff --git a/bluesky/log.c b/bluesky/log.c index 7146506..1fe6176 100644 --- a/bluesky/log.c +++ b/bluesky/log.c @@ -38,10 +38,6 @@ // no absolute guarantees on the size of a log segment. #define LOG_SEGMENT_SIZE (1 << 22) -// Target amount of disk space to use for the journal and cache files, in -// kilobytes. -#define DISK_CACHE_SIZE_TARGET (64 * 1024) - #define HEADER_MAGIC 0x676f4c0a #define FOOTER_MAGIC 0x2e435243 @@ -217,15 +213,6 @@ static gpointer log_thread(gpointer d) offset += sizeof(header) + sizeof(footer) + item->data->len; - /* Since we have just written a new dirty object to the journal, - * increment the count of live dirty objects in that journal file. The - * count will be decremented when objects are deleted or written to the - * cloud. */ - if (!(item->location_flags & CLOUDLOG_CLOUD)) { - g_atomic_int_add(&log->current_log->dirty_refs, 1); - item->dirty_journal = log->current_log; - } - /* Replace the log item's string data with a memory-mapped copy of the * data, now that it has been written to the log file. (Even if it * isn't yet on disk, it should at least be in the page cache and so @@ -300,26 +287,47 @@ void bluesky_cachefile_unref(BlueSkyCacheFile *cachefile) g_atomic_int_add(&cachefile->refcount, -1); } +static void cloudlog_fetch_complete(BlueSkyStoreAsync *async, + BlueSkyCacheFile *cachefile); + +static void cloudlog_fetch_start(BlueSkyCacheFile *cachefile) +{ + g_atomic_int_inc(&cachefile->refcount); + cachefile->fetching = TRUE; + g_print("Starting fetch of %s from cloud\n", cachefile->filename); + BlueSkyStoreAsync *async = bluesky_store_async_new(cachefile->fs->store); + async->op = STORE_OP_GET; + async->key = g_strdup(cachefile->filename); + bluesky_store_async_add_notifier(async, + (GFunc)cloudlog_fetch_complete, + cachefile); + bluesky_store_async_submit(async); + bluesky_store_async_unref(async); +} + static void cloudlog_fetch_complete(BlueSkyStoreAsync *async, BlueSkyCacheFile *cachefile) { g_print("Fetch of %s from cloud complete, status = %d\n", async->key, async->result); - if (async->result < 0) - return; - g_mutex_lock(cachefile->lock); - char *pathname = g_strdup_printf("%s/%s", cachefile->log->log_directory, - cachefile->filename); - if (!g_file_set_contents(pathname, async->data->data, async->data->len, - NULL)) - { - g_print("Error writing out fetched file to cache!\n"); + if (async->result >= 0) { + char *pathname = g_strdup_printf("%s/%s", + cachefile->log->log_directory, + cachefile->filename); + if (!g_file_set_contents(pathname, async->data->data, async->data->len, + NULL)) + g_print("Error writing out fetched file to cache!\n"); + g_free(pathname); + + cachefile->fetching = FALSE; + cachefile->ready = TRUE; + } else { + g_print("Error fetching from cloud, retrying...\n"); + cloudlog_fetch_start(cachefile); } - g_free(pathname); - cachefile->fetching = FALSE; - cachefile->ready = TRUE; + bluesky_cachefile_unref(cachefile); g_cond_broadcast(cachefile->cond); g_mutex_unlock(cachefile->lock); @@ -362,6 +370,7 @@ BlueSkyCacheFile *bluesky_cachefile_lookup(BlueSkyFS *fs, g_print("Adding cache file %s\n", logname); map = g_new0(BlueSkyCacheFile, 1); + map->fs = fs; map->type = type; map->lock = g_mutex_new(); map->type = type; @@ -376,19 +385,8 @@ BlueSkyCacheFile *bluesky_cachefile_lookup(BlueSkyFS *fs, g_hash_table_insert(log->mmap_cache, map->filename, map); // If the log file is stored in the cloud, we may need to fetch it - if (clouddir >= 0) { - g_atomic_int_inc(&map->refcount); - map->fetching = TRUE; - g_print("Starting fetch of %s from cloud\n", logname); - BlueSkyStoreAsync *async = bluesky_store_async_new(fs->store); - async->op = STORE_OP_GET; - async->key = g_strdup(logname); - bluesky_store_async_add_notifier(async, - (GFunc)cloudlog_fetch_complete, - map); - bluesky_store_async_submit(async); - bluesky_store_async_unref(async); - } + if (clouddir >= 0) + cloudlog_fetch_start(map); } else { g_mutex_lock(map->lock); } @@ -511,18 +509,31 @@ void bluesky_cachefile_gc(BlueSkyFS *fs) * then we'll just skip the file on this pass. */ if (g_mutex_trylock(cachefile->lock)) { int64_t age = bluesky_get_current_time() - cachefile->atime; - g_print("%s addr=%p mapcount=%d refcount=%d dirty=%d atime_age=%f", + g_print("%s addr=%p mapcount=%d refcount=%d atime_age=%f", cachefile->filename, cachefile->addr, cachefile->mapcount, - cachefile->refcount, cachefile->dirty_refs, age / 1e6); + cachefile->refcount, age / 1e6); if (cachefile->fetching) g_print(" (fetching)"); g_print("\n"); - if (g_atomic_int_get(&fs->log->disk_used) > DISK_CACHE_SIZE_TARGET + gboolean deletion_candidate = FALSE; + if (g_atomic_int_get(&fs->log->disk_used) + > bluesky_options.cache_size && g_atomic_int_get(&cachefile->refcount) == 0 - && g_atomic_int_get(&cachefile->mapcount) == 0 - && g_atomic_int_get(&cachefile->dirty_refs) == 0) + && g_atomic_int_get(&cachefile->mapcount) == 0) { + deletion_candidate = TRUE; + } + + /* Don't allow journal files to be reclaimed until all data is + * known to be durably stored in the cloud. */ + if (cachefile->type == CLOUDLOG_JOURNAL + && cachefile->log_seq >= fs->log->journal_watermark) + { + deletion_candidate = FALSE; + } + + if (deletion_candidate) { g_print(" ...deleting\n"); if (unlinkat(fs->log->dirfd, cachefile->filename, 0) < 0) { fprintf(stderr, "Unable to unlink journal %s: %m\n",