performance/stat-prefetch: implement sp_readdir.

Signed-off-by: Anand V. Avati <avati@dev.gluster.com>

BUG: 221 (stat prefetch implementation)
URL: http://bugs.gluster.com/cgi-bin/bugzilla3/show_bug.cgi?id=221

2 files changed, 169 insertions, 3 deletions

diff --git a/xlators/performance/stat-prefetch/src/stat-prefetch.c b/xlators/performance/stat-prefetch/src/stat-prefetch.c
index 5823634fa..0619545b7 100644
--- a/xlators/performance/stat-prefetch/src/stat-prefetch.c
+++ b/xlators/performance/stat-prefetch/src/stat-prefetch.c
@@ -21,6 +21,21 @@
 #include "locking.h"
 
 
+sp_cache_t *
+sp_cache_init (void)
+{
+        sp_cache_t *cache = NULL;
+
+        cache = CALLOC (1, sizeof (*cache));
+        if (cache) {
+                INIT_LIST_HEAD (&cache->entries.list);
+                LOCK_INIT (&cache->lock);
+        }
+
+        return cache;
+}
+
+
 void
 sp_local_free (sp_local_t *local)
 {
@@ -122,6 +137,29 @@ out:
         return cache;
 }
 
+
+sp_cache_t *
+sp_del_cache_fd (xlator_t *this, fd_t *fd)
+{
+        sp_cache_t *cache = NULL;
+        uint64_t    value = 0;
+        int32_t     ret   = -1;
+
+        if (fd == NULL) {
+                goto out;
+        }
+
+        ret = fd_ctx_del (fd, this, &value);
+        if (ret == -1) {
+                goto out;
+        }
+
+        cache = (void *)(long) value;
+out:
+        return cache;
+}
+
+
 sp_cache_t *
 sp_get_cache_inode (xlator_t *this, inode_t *inode, int32_t pid)
 {
@@ -143,6 +181,64 @@ out:
 }
 
 
+inline int32_t
+sp_put_cache (xlator_t *this, fd_t *fd, sp_cache_t *cache)
+{
+        return fd_ctx_set (fd, this, (long)(void *)cache);
+}
+
+
+int32_t
+sp_cache_add_entries (sp_cache_t *cache, gf_dirent_t *entries)
+{
+        gf_dirent_t  copy;
+        gf_dirent_t *entry           = NULL, *new = NULL;
+        int32_t      ret             = -1;
+        uint64_t     expected_offset = 0;
+        
+        memset (&copy, 0, sizeof (copy));
+        INIT_LIST_HEAD (&copy.list);
+
+        LOCK (&cache->lock);
+        {
+                list_for_each_entry (entry, &entries->list, list) {
+                        new = gf_dirent_for_name (entry->d_name);
+                        if (new == NULL) {
+                                gf_dirent_free (&copy);
+                                goto unlock;
+                        }
+
+                        new->d_ino  = entry->d_ino;
+                        new->d_off  = entry->d_off;
+                        new->d_len  = entry->d_len;
+                        new->d_type = entry->d_type;
+                        new->d_stat = entry->d_stat;
+
+                        list_add_tail (&new->list, &copy.list);
+
+                        expected_offset = new->d_off;
+                }
+
+                /* 
+                 * splice entries in cache to copy, so that we have a list in
+                 * ascending order of offsets
+                 */
+                list_splice_init (&cache->entries.list, &copy.list);
+
+                /* splice back the copy into cache */
+                list_splice_init (&copy.list, &cache->entries.list);
+
+                cache->expected_offset = expected_offset;
+
+                ret = 0;
+        }
+unlock:
+        UNLOCK (&cache->lock);
+
+        return ret;
+}
+
+
 int32_t
 sp_lookup_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
                int32_t op_ret, int32_t op_errno, inode_t *inode,
@@ -283,6 +379,74 @@ unwind:
 
 }
 
+
+int32_t
+sp_readdir_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
+                int32_t op_ret, int32_t op_errno, gf_dirent_t *entries)
+{
+        sp_local_t *local = NULL;
+        sp_cache_t *cache = NULL;  
+        fd_t       *fd    = NULL;
+        int32_t     ret   = 0;
+
+        local = frame->local;
+        if (local == NULL) {
+                goto out;
+        }
+
+        fd = local->fd;
+
+        cache = sp_get_cache_fd (this, fd);
+        if (cache == NULL) {
+                cache = sp_cache_init ();
+                if (cache == NULL) {
+                        goto out;
+                }
+
+                ret = sp_put_cache (this, fd, cache);
+                if (ret == -1) {
+                        goto out;
+                }
+        }
+
+        sp_cache_add_entries (cache, entries);
+
+out:
+	SP_STACK_UNWIND (frame, op_ret, op_errno, entries);
+	return 0;
+}
+
+
+int32_t
+sp_readdir (call_frame_t *frame, xlator_t *this, fd_t *fd, size_t size,
+            off_t off)
+{
+        sp_cache_t *cache    = NULL;
+        sp_local_t *local    = NULL;
+
+        cache = sp_get_cache_fd (this, fd);
+        if (cache) {
+                if (off != cache->expected_offset) {
+                        cache = sp_del_cache_fd (this, fd);
+                        if (cache) {
+                                sp_cache_free (cache);
+                        }
+                }
+        }
+
+        local = CALLOC (1, sizeof (*local));
+        if (local) {
+                local->fd = fd;
+                frame->local = local;
+        }
+
+	STACK_WIND (frame, sp_readdir_cbk, FIRST_CHILD(this),
+                    FIRST_CHILD(this)->fops->readdir, fd, size, off);
+
+        return 0;
+}
+
+
 int32_t
 sp_forget (xlator_t *this, inode_t *inode)
 {
@@ -325,7 +489,8 @@ fini (xlator_t *this)
 
 
 struct xlator_fops fops = {
-        .lookup = sp_lookup,
+        .lookup    = sp_lookup,
+        .readdir   = sp_readdir,
 };
 
 struct xlator_mops mops = {
diff --git a/xlators/performance/stat-prefetch/src/stat-prefetch.h b/xlators/performance/stat-prefetch/src/stat-prefetch.h
index cf6bdd89d..4a45261a4 100644
--- a/xlators/performance/stat-prefetch/src/stat-prefetch.h
+++ b/xlators/performance/stat-prefetch/src/stat-prefetch.h
@@ -30,7 +30,7 @@
 #include "xlator.h"
 
 struct sp_cache {
-        gf_dirent_t entries;            /* Head of list of cached dirents. */
+        gf_dirent_t entries;            /* Head of list of cached dirents */
         uint64_t    expected_offset;    /* Offset where the next read will
                                          * happen.
                                          */
@@ -39,7 +39,8 @@ struct sp_cache {
 typedef struct sp_cache sp_cache_t;
 
 struct sp_local {
-        loc_t loc;
+        loc_t  loc;
+        fd_t  *fd;
 };
 typedef struct sp_local sp_local_t;