Implement very basic grouped fetches of objects from the cloud.

Add an item prefetch call, which for now just collects together items that
have been prefetched and when one of them is actually fetched, a range
request is sent that will cover all the objects.  Thus this doesn't improve
the time to fetch the first object (actually it will hurt), it does arrange
to have objects fetched in groups where possible which reduces the total
number of operations and the time to get subsequent objects.

diff --git a/bluesky/bluesky-private.h b/bluesky/bluesky-private.h
index dcd7c48..547fb59 100644 (file)
--- a/bluesky/bluesky-private.h
+++ b/bluesky/bluesky-private.h
@@ -389,6 +389,7 @@ struct BlueSkyCacheFile {
     gboolean fetching, ready;   // Cloud data: downloading or ready for use
     int64_t atime;              // Access time, for cache management
     BlueSkyRangeset *items;     // Locations of valid items
+    BlueSkyRangeset *prefetches;// Locations we have been requested to prefetch
 };
 
 BlueSkyLog *bluesky_log_new(const char *log_directory);

diff --git a/bluesky/bluesky.h b/bluesky/bluesky.h
index c27fb6c..fd0b610 100644 (file)
--- a/bluesky/bluesky.h
+++ b/bluesky/bluesky.h
@@ -100,6 +100,8 @@ gboolean bluesky_rangeset_insert(BlueSkyRangeset *rangeset,
 const BlueSkyRangesetItem *bluesky_rangeset_lookup(BlueSkyRangeset *rangeset,
                                                    uint64_t offset);
 const BlueSkyRangesetItem *bluesky_rangeset_lookup_next(BlueSkyRangeset *rangeset, uint64_t offset);
+void bluesky_rangeset_get_extents(BlueSkyRangeset *rangeset,
+                                  uint64_t *start, uint64_t *length);
 
 /* Storage interface.  This presents a key-value store abstraction, and can
  * have multiple implementations: in-memory, on-disk, in-cloud. */

diff --git a/bluesky/cache.c b/bluesky/cache.c
index fd4797d..7327e68 100644 (file)
--- a/bluesky/cache.c
+++ b/bluesky/cache.c
@@ -362,6 +362,8 @@ void bluesky_cachefile_gc(BlueSkyFS *fs)
                 g_atomic_int_add(&fs->log->disk_used, -cachefile->disk_used);
                 g_hash_table_remove(fs->log->mmap_cache, cachefile->filename);
                 bluesky_rangeset_free(cachefile->items);
+                if (cachefile->prefetches != NULL)
+                    bluesky_rangeset_free(cachefile->prefetches);
                 g_mutex_unlock(cachefile->lock);
                 g_mutex_free(cachefile->lock);
                 g_cond_free(cachefile->cond);

diff --git a/bluesky/cloudlog.c b/bluesky/cloudlog.c
index 7730d0e..0668da7 100644 (file)
--- a/bluesky/cloudlog.c
+++ b/bluesky/cloudlog.c
@@ -270,12 +270,66 @@ BlueSkyCloudLog *bluesky_cloudlog_get(BlueSkyFS *fs, BlueSkyCloudID id)
 
 /* 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. */
-void bluesky_cloudlog_prefetch(BlueSkyCloudLog *log)
+ * bluesky_cloudlog_fetch will complete quickly.  Item must be locked? */
+void bluesky_cloudlog_prefetch(BlueSkyCloudLog *item)
 {
-    gchar *id = bluesky_cloudlog_id_to_string(log->id);
-    g_print("Prefetch for %s\n", id);
-    g_free(id);
+    if (item->data != NULL)
+        return;
+
+    /* TODO: Some of the code here is duplicated with bluesky_log_map_object.
+     * Refactor to fix that. */
+    BlueSkyFS *fs = item->fs;
+    BlueSkyCacheFile *map = NULL;
+
+    /* First, check to see if the journal still contains a copy of the item and
+     * if so update the atime on the journal so it is likely to be kept around
+     * until we need it. */
+    if ((item->location_flags | item->pending_write) & CLOUDLOG_JOURNAL) {
+        map = bluesky_cachefile_lookup(fs, -1, item->log_seq, TRUE);
+        if (map != NULL) {
+            map->atime = bluesky_get_current_time();
+            bluesky_cachefile_unref(map);
+            g_mutex_unlock(map->lock);
+            return;
+        }
+    }
+
+    item->location_flags &= ~CLOUDLOG_JOURNAL;
+    if (!(item->location_flags & CLOUDLOG_CLOUD))
+        return;
+
+    map = bluesky_cachefile_lookup(fs,
+                                   item->location.directory,
+                                   item->location.sequence,
+                                   FALSE);
+    if (map == NULL)
+        return;
+
+    /* At this point, we have information about the log segment containing the
+     * item we need.  If our item is already fetched, we have nothing to do
+     * except update the atime.  If not, queue up a fetch of our object. */
+    const BlueSkyRangesetItem *rangeitem;
+    rangeitem = bluesky_rangeset_lookup(map->items,
+                                        item->location.offset);
+    if (rangeitem == NULL) {
+        if (map->prefetches == NULL)
+            map->prefetches = bluesky_rangeset_new();
+
+        gchar *id = bluesky_cloudlog_id_to_string(item->id);
+        g_print("Need to prefetch %s\n", id);
+        g_free(id);
+
+        bluesky_rangeset_insert(map->prefetches,
+                                item->location.offset,
+                                item->location.size, NULL);
+
+        uint64_t start, length;
+        bluesky_rangeset_get_extents(map->prefetches, &start, &length);
+        g_print("Range to prefetch: %"PRIu64" + %"PRIu64"\n", start, length);
+    }
+
+    bluesky_cachefile_unref(map);
+    g_mutex_unlock(map->lock);
 }
 
 /* Ensure that a cloud log item is loaded in memory, and if not read it in.

diff --git a/bluesky/log.c b/bluesky/log.c
index ad1dac1..4cef5fd 100644 (file)
--- a/bluesky/log.c
+++ b/bluesky/log.c
@@ -616,8 +616,20 @@ BlueSkyRCStr *bluesky_log_map_object(BlueSkyCloudLog *item, gboolean map_data)
             if (rangeitem == NULL) {
                 g_print("Item at offset 0x%zx not available, need to fetch.\n",
                         file_offset);
-                if (range_request)
-                    cloudlog_partial_fetch_start(map, file_offset, file_size);
+                if (range_request) {
+                    uint64_t start = 0, length = 0, end;
+                    if (map->prefetches != NULL)
+                        bluesky_rangeset_get_extents(map->prefetches,
+                                                     &start, &length);
+                    start = MIN(start, file_offset);
+                    end = MAX(start + length, file_offset + file_size);
+                    length = end - start;
+                    cloudlog_partial_fetch_start(map, start, length);
+                    if (map->prefetches != NULL) {
+                        bluesky_rangeset_free(map->prefetches);
+                        map->prefetches = NULL;
+                    }
+                }
                 g_cond_wait(map->cond, map->lock);
             } else if (rangeitem->start == file_offset
                        && rangeitem->length == file_size) {

diff --git a/bluesky/util.c b/bluesky/util.c
index e7514fc..c5618b9 100644 (file)
--- a/bluesky/util.c
+++ b/bluesky/util.c
@@ -312,3 +312,27 @@ const BlueSkyRangesetItem *bluesky_rangeset_lookup_next(BlueSkyRangeset *rangese
     }
     return item;
 }
+
+/* Determine the full extent of a rangeset--the smallest offset covered and the
+ * length needed to extend to the end of the last item. */
+void bluesky_rangeset_get_extents(BlueSkyRangeset *rangeset,
+                                  uint64_t *start, uint64_t *length)
+{
+    GSequenceIter *i;
+    BlueSkyRangesetItem *item;
+
+    i = g_sequence_get_begin_iter(rangeset->seq);
+    if (g_sequence_iter_is_end(i)) {
+        *start = 0;
+        *length = 0;
+        return;
+    }
+
+    item = (BlueSkyRangesetItem *)g_sequence_get(i);
+    *start = item->start;
+
+    i = g_sequence_get_end_iter(rangeset->seq);
+    i = g_sequence_iter_prev(i);
+    item = (BlueSkyRangesetItem *)g_sequence_get(i);
+    *length = (item->start + item->length) - *start;
+}