From: Michael Vrable <mvrable@cs.ucsd.edu>
Date: Mon, 31 Aug 2009 19:49:26 +0000 (-0700)
Subject: Add fetching of blocks from S3.
X-Git-Url: http://git.vrable.net/?a=commitdiff_plain;ds=inline;h=823820015089dd6185fd2f2b192576b5b7b2eaab;p=bluesky.git

Add fetching of blocks from S3.
---

diff --git a/bluesky.h b/bluesky.h
index d7bf8e3..c4bad9e 100644
--- a/bluesky.h
+++ b/bluesky.h
@@ -143,6 +143,7 @@ gboolean bluesky_directory_insert(BlueSkyInode *dir, gchar *name,
 void bluesky_directory_dump(BlueSkyInode *dir);
 
 void bluesky_block_touch(BlueSkyInode *inode, uint64_t i);
+void bluesky_block_fetch(BlueSkyFS *fs, BlueSkyBlock *block);
 void bluesky_block_flush(BlueSkyFS *fs, BlueSkyBlock *block);
 void bluesky_file_truncate(BlueSkyInode *inode, uint64_t size);
 void bluesky_file_write(BlueSkyInode *inode, uint64_t offset,
diff --git a/inode.c b/inode.c
index b1071a6..df3883b 100644
--- a/inode.c
+++ b/inode.c
@@ -300,9 +300,8 @@ void bluesky_file_read(BlueSkyInode *inode, uint64_t offset,
             memset(buf, 0, bytes);
             break;
         case BLUESKY_BLOCK_REF:
-            /* TODO: Pull in data first */
-            memset(buf, 0, bytes);
-            break;
+            bluesky_block_fetch(inode->fs, b);
+            /* Fall through */
         case BLUESKY_BLOCK_CACHED:
         case BLUESKY_BLOCK_DIRTY:
             memcpy(buf, &b->data[block_offset], bytes);
@@ -315,6 +314,22 @@ void bluesky_file_read(BlueSkyInode *inode, uint64_t offset,
     }
 }
 
+/* Read the given block from cloud-backed storage if the data is not already
+ * cached. */
+void bluesky_block_fetch(BlueSkyFS *fs, BlueSkyBlock *block)
+{
+    if (block->type != BLUESKY_BLOCK_REF)
+        return;
+
+    g_print("Fetching block from %s\n", block->ref);
+    BlueSkyRCStr *string = s3store_get(fs->store, block->ref);
+
+    g_free(block->data);
+    block->data = g_memdup(string->data, BLUESKY_BLOCK_SIZE);
+    block->type = BLUESKY_BLOCK_CACHED;
+    bluesky_string_unref(string);
+}
+
 /* Write the given block to cloud-backed storage and mark it clean. */
 void bluesky_block_flush(BlueSkyFS *fs, BlueSkyBlock *block)
 {
@@ -333,7 +348,11 @@ void bluesky_block_flush(BlueSkyFS *fs, BlueSkyBlock *block)
     s3store_put(fs->store, name, data);
     g_free(block->ref);
     block->ref = g_strdup(name);
-    block->type = BLUESKY_BLOCK_CACHED;
+
+    /* block->type = BLUESKY_BLOCK_CACHED; */
+    g_free(block->data);
+    block->data = NULL;
+    block->type = BLUESKY_BLOCK_REF;
 
     g_checksum_free(csum);
     bluesky_string_unref(data);
diff --git a/s3store.cc b/s3store.cc
index aff9080..ac748d1 100644
--- a/s3store.cc
+++ b/s3store.cc
@@ -36,9 +36,19 @@ S3Store *s3store_new()
     return store;
 }
 
-BlueSkyRCStr *s3store_get(S3Store *store, const gchar *key)
+struct get_info {
+    gchar *buf;
+    gint offset;
+};
+
+static S3Status s3store_get_handler(int bufferSize, const char *buffer,
+                                    void *callbackData)
 {
-    return NULL;
+    struct get_info *info = (struct get_info *)callbackData;
+    gint bytes = MIN(bufferSize, (int)(BLUESKY_BLOCK_SIZE - info->offset));
+    memcpy(info->buf + info->offset, buffer, bytes);
+    info->offset += bytes;
+    return S3StatusOK;
 }
 
 struct put_info {
@@ -74,6 +84,24 @@ void s3store_response_callback(S3Status status,
     }
 }
 
+BlueSkyRCStr *s3store_get(S3Store *store, const gchar *key)
+{
+    struct get_info info;
+    info.buf = (char *)g_malloc0(BLUESKY_BLOCK_SIZE);
+    info.offset = 0;
+
+    struct S3GetObjectHandler handler;
+    handler.responseHandler.propertiesCallback = s3store_properties_callback;
+    handler.responseHandler.completeCallback = s3store_response_callback;
+    handler.getObjectDataCallback = s3store_get_handler;
+
+    g_print("Starting fetch of %s from S3...\n", key);
+    S3_get_object(&store->bucket, key, NULL, 0, 0, NULL,
+                  &handler, &info);
+
+    return bluesky_string_new(info.buf, BLUESKY_BLOCK_SIZE);
+}
+
 void s3store_put(S3Store *store, const gchar *key, BlueSkyRCStr *val)
 {
     struct put_info info;