Added all files

6 files changed, 2804 insertions, 0 deletions

diff --git a/xlators/performance/io-cache/Makefile.am b/xlators/performance/io-cache/Makefile.am
new file mode 100644
index 00000000000..d471a3f9243
--- /dev/null
+++ b/xlators/performance/io-cache/Makefile.am
@@ -0,0 +1,3 @@
+SUBDIRS = src
+
+CLEANFILES = 
diff --git a/xlators/performance/io-cache/src/Makefile.am b/xlators/performance/io-cache/src/Makefile.am
new file mode 100644
index 00000000000..b1bf5bfbf71
--- /dev/null
+++ b/xlators/performance/io-cache/src/Makefile.am
@@ -0,0 +1,14 @@
+xlator_LTLIBRARIES = io-cache.la
+xlatordir = $(libdir)/glusterfs/$(PACKAGE_VERSION)/xlator/performance
+
+io_cache_la_LDFLAGS = -module -avoidversion 
+
+io_cache_la_SOURCES = io-cache.c page.c ioc-inode.c
+io_cache_la_LIBADD = $(top_builddir)/libglusterfs/src/libglusterfs.la
+
+noinst_HEADERS = io-cache.h
+
+AM_CFLAGS = -fPIC -D_FILE_OFFSET_BITS=64 -D_GNU_SOURCE -Wall -D$(GF_HOST_OS) \
+	-I$(top_srcdir)/libglusterfs/src -shared -nostartfiles $(GF_CFLAGS)
+
+CLEANFILES = 
diff --git a/xlators/performance/io-cache/src/io-cache.c b/xlators/performance/io-cache/src/io-cache.c
new file mode 100644
index 00000000000..f367cdb88de
--- /dev/null
+++ b/xlators/performance/io-cache/src/io-cache.c
@@ -0,0 +1,1478 @@
+/*
+  Copyright (c) 2007, 2008, 2009 Z RESEARCH, Inc. <http://www.zresearch.com>
+  This file is part of GlusterFS.
+
+  GlusterFS is free software; you can redistribute it and/or modify
+  it under the terms of the GNU General Public License as published
+  by the Free Software Foundation; either version 3 of the License,
+  or (at your option) any later version.
+
+  GlusterFS is distributed in the hope that it will be useful, but
+  WITHOUT ANY WARRANTY; without even the implied warranty of
+  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
+  General Public License for more details.
+
+  You should have received a copy of the GNU General Public License
+  along with this program.  If not, see
+  <http://www.gnu.org/licenses/>.
+*/
+
+#ifndef _CONFIG_H
+#define _CONFIG_H
+#include "config.h"
+#endif
+
+#include "glusterfs.h"
+#include "logging.h"
+#include "dict.h"
+#include "xlator.h"
+#include "io-cache.h"
+#include <assert.h>
+#include <sys/time.h>
+
+static uint32_t
+ioc_get_priority (ioc_table_t *table, 
+		  const char *path);
+
+static uint32_t
+ioc_get_priority (ioc_table_t *table, 
+		  const char *path);
+
+static inline ioc_inode_t *
+ioc_inode_reupdate (ioc_inode_t *ioc_inode)
+{
+	ioc_table_t *table = ioc_inode->table;
+
+	list_add_tail (&ioc_inode->inode_lru, 
+		       &table->inode_lru[ioc_inode->weight]);
+  
+	return ioc_inode;
+}
+
+static inline ioc_inode_t *
+ioc_get_inode (dict_t *dict,
+	       char *name)
+{
+	ioc_inode_t *ioc_inode = NULL;
+	data_t *ioc_inode_data = dict_get (dict, name);
+	ioc_table_t *table = NULL;
+
+	if (ioc_inode_data) {
+		ioc_inode = data_to_ptr (ioc_inode_data);
+		table = ioc_inode->table;
+
+		ioc_table_lock (table);
+		{
+			if (list_empty (&ioc_inode->inode_lru)) {
+				ioc_inode = ioc_inode_reupdate (ioc_inode);
+			}
+		}
+		ioc_table_unlock (table);
+	}
+  
+	return ioc_inode;
+}
+
+int32_t
+ioc_inode_need_revalidate (ioc_inode_t *ioc_inode)
+{
+	int8_t need_revalidate = 0;
+	struct timeval tv = {0,};
+	int32_t ret = -1;
+	ioc_table_t *table = ioc_inode->table;
+
+	ret = gettimeofday (&tv, NULL);
+
+	if (time_elapsed (&tv, &ioc_inode->tv) >= table->cache_timeout)
+		need_revalidate = 1;
+
+	return need_revalidate;
+}
+
+/*
+ * __ioc_inode_flush - flush all the cached pages of the given inode
+ *
+ * @ioc_inode: 
+ *
+ * assumes lock is held
+ */
+int32_t
+__ioc_inode_flush (ioc_inode_t *ioc_inode)
+{
+	ioc_page_t *curr = NULL, *next = NULL;
+	int32_t destroy_size = 0;
+	int32_t ret = 0;
+
+	list_for_each_entry_safe (curr, next, &ioc_inode->pages, pages) {
+		ret = ioc_page_destroy (curr);
+    
+		if (ret != -1) 
+			destroy_size += ret;
+	}
+  
+	return destroy_size;
+}
+
+void
+ioc_inode_flush (ioc_inode_t *ioc_inode)
+{
+	int32_t destroy_size = 0;    
+
+	ioc_inode_lock (ioc_inode);
+	{
+		destroy_size = __ioc_inode_flush (ioc_inode);
+	}
+	ioc_inode_unlock (ioc_inode);
+  
+	if (destroy_size) {
+		ioc_table_lock (ioc_inode->table);
+		{
+			ioc_inode->table->cache_used -= destroy_size;
+		}
+		ioc_table_unlock (ioc_inode->table);
+	}
+
+	return;
+}
+
+/* 
+ * ioc_utimens_cbk -
+ * 
+ * @frame:
+ * @cookie:
+ * @this:
+ * @op_ret:
+ * @op_errno:
+ * @stbuf:
+ *
+ */
+int32_t
+ioc_utimens_cbk (call_frame_t *frame,
+		 void *cookie,
+		 xlator_t *this,
+		 int32_t op_ret,
+		 int32_t op_errno,
+		 struct stat *stbuf)
+{
+	STACK_UNWIND (frame, op_ret, op_errno, stbuf);
+	return 0;
+}
+
+/* 
+ * ioc_utimens -
+ * 
+ * @frame:
+ * @this:
+ * @loc:
+ * @tv:
+ *
+ */
+int32_t
+ioc_utimens (call_frame_t *frame,
+	     xlator_t *this,
+	     loc_t *loc,
+	     struct timespec *tv)
+{
+	uint64_t ioc_inode = 0;
+	inode_ctx_get (loc->inode, this, &ioc_inode);
+
+	if (ioc_inode)
+		ioc_inode_flush ((ioc_inode_t *)(long)ioc_inode);
+
+	STACK_WIND (frame, ioc_utimens_cbk,
+		    FIRST_CHILD (this),
+		    FIRST_CHILD (this)->fops->utimens,
+		    loc, tv);
+	return 0;
+}
+
+int32_t
+ioc_lookup_cbk (call_frame_t *frame,
+		void *cookie,
+		xlator_t *this,
+		int32_t op_ret,
+		int32_t op_errno,
+		inode_t *inode,
+		struct stat *stbuf,
+		dict_t *dict)
+{
+	ioc_inode_t *ioc_inode = NULL;
+	ioc_local_t *local = frame->local;
+	ioc_table_t *table = this->private;
+	ioc_page_t  *page = NULL;
+	data_t      *page_data = NULL;
+	data_t      *content_data = NULL;
+	char        *src = NULL;
+	char        *dst = NULL;
+	char         need_unref = 0;
+	uint8_t      cache_still_valid = 0;
+	uint32_t     weight = 0;
+	uint64_t     tmp_ioc_inode = 0;
+	char        *buf = NULL;
+	char        *tmp = NULL;
+	int          i;
+	
+	if (op_ret != 0) 
+		goto out;
+
+	inode_ctx_get (inode, this, &tmp_ioc_inode);
+	ioc_inode = (ioc_inode_t *)(long)tmp_ioc_inode;
+	if (ioc_inode) {
+		cache_still_valid = ioc_cache_still_valid (ioc_inode, 
+							   stbuf);
+		
+		if (!cache_still_valid) {
+			ioc_inode_flush (ioc_inode);
+		} 
+		/* update the time-stamp of revalidation */
+		ioc_inode_lock (ioc_inode);
+		{
+			gettimeofday (&ioc_inode->tv, NULL);
+		}
+		ioc_inode_unlock (ioc_inode);
+		
+		ioc_table_lock (ioc_inode->table);
+		{
+			list_move_tail (&ioc_inode->inode_lru,
+					&table->inode_lru[ioc_inode->weight]);
+		}
+		ioc_table_unlock (ioc_inode->table);
+	}
+	
+	if (local && stbuf->st_size && 
+	    local->need_xattr >= stbuf->st_size) {
+		if (!ioc_inode) {
+			weight = ioc_get_priority (table, 
+						   local->file_loc.path);
+			ioc_inode = ioc_inode_update (table, 
+						      inode, weight);
+			inode_ctx_put (inode, this, 
+				       (uint64_t)(long)ioc_inode);
+		}
+		
+		ioc_inode_lock (ioc_inode);
+		{
+			content_data = dict_get (dict, "glusterfs.content");
+			page = ioc_page_get (ioc_inode, 0);
+			
+			if (content_data) {
+				if (page) {
+					dict_unref (page->ref);
+					free (page->vector);
+					page->vector = NULL;
+					
+					ioc_table_lock (table);
+					{
+						table->cache_used -= 
+							page->size;
+					}
+					ioc_table_unlock (table);
+				} else {
+					page = ioc_page_create (ioc_inode, 0);
+				}
+				
+				dst = CALLOC (1, stbuf->st_size);
+				page->ref = dict_ref (get_new_dict ());
+				page_data = data_from_dynptr (dst, 
+							      stbuf->st_size);
+				dict_set (page->ref, NULL, page_data);
+				
+				src = data_to_ptr (content_data);
+				memcpy (dst, src, stbuf->st_size);
+
+				page->vector = CALLOC (1, 
+						       sizeof (*page->vector));
+				page->vector->iov_base = dst;
+				page->vector->iov_len = stbuf->st_size;
+				page->count = 1;
+      
+				page->waitq = NULL;
+				page->size = stbuf->st_size;
+				page->ready = 1;
+
+				ioc_table_lock (table);
+				{
+					table->cache_used += page->size;
+				}
+				ioc_table_unlock (table);
+				
+			} else {
+				if (!(page && page->ready)) {
+					gf_log (this->name, GF_LOG_DEBUG,
+						"page not present");
+					
+					ioc_inode_unlock (ioc_inode);
+					STACK_WIND (frame,
+						    ioc_lookup_cbk,
+						    FIRST_CHILD (this),
+						    FIRST_CHILD (this)->fops->lookup,
+						    &local->file_loc,
+						    local->xattr_req);
+					return 0;
+				} 
+				buf = CALLOC (1, stbuf->st_size);
+				tmp = buf;
+
+				for (i = 0; i < page->count; i++) {
+					memcpy (tmp, page->vector[i].iov_base, 
+						page->vector[i].iov_len);
+					tmp += page->vector[i].iov_len;
+				}
+				
+				gf_log (this->name, GF_LOG_DEBUG,
+					"serving file %s from cache", 
+					local->file_loc.path);
+				
+				if (!dict) {
+					need_unref = 1;
+					dict = dict_ref (
+						get_new_dict ());
+				}
+				dict_set (dict, "glusterfs.content",
+					  data_from_dynptr (buf, 
+							    stbuf->st_size));
+			}
+
+			ioc_inode->mtime = stbuf->st_mtime;
+			gettimeofday (&ioc_inode->tv, NULL);
+		}
+		ioc_inode_unlock (ioc_inode);
+		
+		if (content_data && 
+		    ioc_need_prune (ioc_inode->table)) {
+			ioc_prune (ioc_inode->table);
+		}
+	}
+
+ out:
+	STACK_UNWIND (frame, op_ret, op_errno, inode, stbuf, dict);
+
+	if (need_unref) {
+		dict_unref (dict);
+	}
+
+	return 0;
+}
+
+int32_t 
+ioc_lookup (call_frame_t *frame,
+	    xlator_t *this,
+	    loc_t *loc,
+	    dict_t *xattr_req)
+{
+	uint64_t content_limit = 0;
+
+	if (GF_FILE_CONTENT_REQUESTED(xattr_req, &content_limit)) {
+		uint64_t     tmp_ioc_inode = 0;
+		ioc_inode_t *ioc_inode = NULL;
+		ioc_page_t  *page = NULL;
+		ioc_local_t *local = CALLOC (1, sizeof (*local));
+
+		local->need_xattr = content_limit;
+		local->file_loc.path = loc->path;
+		local->file_loc.inode = loc->inode;
+		frame->local = local;
+
+		inode_ctx_get (loc->inode, this, &tmp_ioc_inode);
+		ioc_inode = (ioc_inode_t *)(long)tmp_ioc_inode;
+
+		if (ioc_inode) {
+			ioc_inode_lock (ioc_inode);
+			{
+				page = ioc_page_get (ioc_inode, 0);
+				if ((content_limit <= 
+				     ioc_inode->table->page_size) && 
+				    page && page->ready) {
+					local->need_xattr = -1;
+				}
+			}
+			ioc_inode_unlock (ioc_inode);
+		}
+	}
+
+	STACK_WIND (frame,
+		    ioc_lookup_cbk,
+		    FIRST_CHILD (this),
+		    FIRST_CHILD (this)->fops->lookup,
+		    loc,
+		    xattr_req);
+	return 0;
+}
+
+/*
+ * ioc_forget - 
+ *
+ * @frame:
+ * @this:
+ * @inode:
+ *
+ */
+int32_t
+ioc_forget (xlator_t *this,
+	    inode_t *inode)
+{
+	uint64_t ioc_inode = 0;
+
+	inode_ctx_get (inode, this, &ioc_inode);
+
+	if (ioc_inode)
+		ioc_inode_destroy ((ioc_inode_t *)(long)ioc_inode);
+	    
+	return 0;
+}
+
+
+/* 
+ * ioc_cache_validate_cbk - 
+ *
+ * @frame:
+ * @cookie:
+ * @this:
+ * @op_ret:
+ * @op_errno:
+ * @buf
+ *
+ */
+int32_t
+ioc_cache_validate_cbk (call_frame_t *frame,
+			void *cookie,
+			xlator_t *this,
+			int32_t op_ret,
+			int32_t op_errno,
+			struct stat *stbuf)
+{
+	ioc_local_t *local = frame->local;
+	ioc_inode_t *ioc_inode = NULL;
+	size_t destroy_size = 0;
+	struct stat *local_stbuf = stbuf;
+
+	ioc_inode = local->inode;
+
+	if ((op_ret == -1) || 
+	    ((op_ret >= 0) && !ioc_cache_still_valid(ioc_inode, stbuf))) {
+		gf_log (ioc_inode->table->xl->name, GF_LOG_DEBUG,
+			"cache for inode(%p) is invalid. flushing all pages",
+			ioc_inode);
+		/* NOTE: only pages with no waiting frames are flushed by 
+		 * ioc_inode_flush. page_fault will be generated for all 
+		 * the pages which have waiting frames by ioc_inode_wakeup()
+		 */
+		ioc_inode_lock (ioc_inode);
+		{
+			destroy_size = __ioc_inode_flush (ioc_inode);
+			if (op_ret >= 0)
+				ioc_inode->mtime = stbuf->st_mtime;
+		}
+		ioc_inode_unlock (ioc_inode);
+		local_stbuf = NULL;
+	}
+
+	if (destroy_size) {
+		ioc_table_lock (ioc_inode->table);
+		{
+			ioc_inode->table->cache_used -= destroy_size;
+		}
+		ioc_table_unlock (ioc_inode->table);
+	}
+
+	if (op_ret < 0)
+		local_stbuf = NULL;
+  
+	ioc_inode_lock (ioc_inode);
+	{
+		gettimeofday (&ioc_inode->tv, NULL);
+	}
+	ioc_inode_unlock (ioc_inode);
+
+	ioc_inode_wakeup (frame, ioc_inode, local_stbuf);
+  
+	/* any page-fault initiated by ioc_inode_wakeup() will have its own 
+	 * fd_ref on fd, safe to unref validate frame's private copy 
+	 */
+	fd_unref (local->fd);
+
+	STACK_DESTROY (frame->root);
+
+	return 0;
+}
+
+static int32_t
+ioc_wait_on_inode (ioc_inode_t *ioc_inode, 
+		   ioc_page_t *page)
+{
+	ioc_waitq_t *waiter = NULL, *trav = NULL;
+	uint32_t page_found = 0;
+
+	trav = ioc_inode->waitq;
+
+	while (trav) {
+		if (trav->data == page) {
+			page_found = 1;
+			break;
+		}
+		trav = trav->next;
+	}
+  
+	if (!page_found) {
+		waiter = CALLOC (1, sizeof (ioc_waitq_t));
+		ERR_ABORT (waiter);
+		waiter->data = page;
+		waiter->next = ioc_inode->waitq;
+		ioc_inode->waitq = waiter;
+	}
+  
+	return 0;
+}
+
+/*
+ * ioc_cache_validate -
+ *
+ * @frame:
+ * @ioc_inode:
+ * @fd:
+ *
+ */
+static int32_t
+ioc_cache_validate (call_frame_t *frame,
+		    ioc_inode_t *ioc_inode,
+		    fd_t *fd,
+		    ioc_page_t *page)
+{
+	call_frame_t *validate_frame = NULL;
+	ioc_local_t *validate_local = NULL;
+
+	validate_local = CALLOC (1, sizeof (ioc_local_t));
+	ERR_ABORT (validate_local);
+	validate_frame = copy_frame (frame);
+	validate_local->fd = fd_ref (fd);
+	validate_local->inode = ioc_inode;
+	validate_frame->local = validate_local;
+    
+	STACK_WIND (validate_frame,
+		    ioc_cache_validate_cbk,
+		    FIRST_CHILD (frame->this),
+		    FIRST_CHILD (frame->this)->fops->fstat,
+		    fd);
+
+	return 0;
+}
+
+static inline uint32_t
+is_match (const char *path,
+	  const char *pattern)
+{
+	char *pathname = strdup (path);
+	int32_t ret = 0;
+
+	ret = fnmatch (pattern, path, FNM_NOESCAPE);
+  
+	free (pathname);
+  
+	return (ret == 0);
+}
+
+static uint32_t
+ioc_get_priority (ioc_table_t *table, 
+		  const char *path)
+{
+	uint32_t priority = 0;
+	struct ioc_priority *curr = NULL;
+  
+	list_for_each_entry (curr, &table->priority_list, list) {
+		if (is_match (path, curr->pattern)) 
+			priority = curr->priority;
+	}
+
+	return priority;
+}
+
+/* 
+ * ioc_open_cbk - open callback for io cache
+ *
+ * @frame: call frame
+ * @cookie:
+ * @this:
+ * @op_ret:
+ * @op_errno:
+ * @fd:
+ *
+ */
+int32_t
+ioc_open_cbk (call_frame_t *frame,
+	      void *cookie,
+	      xlator_t *this,
+	      int32_t op_ret,
+	      int32_t op_errno,
+	      fd_t *fd)
+{
+	uint64_t     tmp_ioc_inode = 0;
+	ioc_local_t *local = frame->local;
+	ioc_table_t *table = this->private;
+	ioc_inode_t *ioc_inode = NULL;
+	inode_t *inode = local->file_loc.inode;
+	uint32_t weight = 0;
+	const char *path = local->file_loc.path;
+
+	if (op_ret != -1) {
+		/* look for ioc_inode corresponding to this fd */
+		LOCK (&fd->inode->lock);
+		//{
+
+		inode_ctx_get (fd->inode, this, &tmp_ioc_inode);
+		ioc_inode = (ioc_inode_t *)(long)tmp_ioc_inode;
+      
+		if (!ioc_inode) {
+			/* this is the first time someone is opening this 
+			   file, assign weight 
+			*/
+			weight = ioc_get_priority (table, path);
+ 
+			ioc_inode = ioc_inode_update (table, inode, weight);
+			inode_ctx_put (fd->inode, this, 
+				       (uint64_t)(long)ioc_inode);
+		} else {
+			ioc_table_lock (ioc_inode->table);
+			//{
+			list_move_tail (&ioc_inode->inode_lru,
+					&table->inode_lru[ioc_inode->weight]);
+			//}
+			ioc_table_unlock (ioc_inode->table);
+		}
+
+		//}
+		UNLOCK (&fd->inode->lock);
+
+		/* If mandatory locking has been enabled on this file,
+		   we disable caching on it */
+		if (((inode->st_mode & S_ISGID) && 
+		     !(inode->st_mode & S_IXGRP))) {
+			fd_ctx_set (fd, this, 1);
+		}
+  
+		/* If O_DIRECT open, we disable caching on it */
+		if ((local->flags & O_DIRECT)){
+			/* O_DIRECT is only for one fd, not the inode 
+			 * as a whole 
+			 */
+			fd_ctx_set (fd, this, 1);
+		}
+	}
+
+	FREE (local);
+	frame->local = NULL;
+
+	STACK_UNWIND (frame, op_ret, op_errno, fd);
+
+	return 0;
+}
+
+/*
+ * ioc_create_cbk - create callback for io cache
+ *
+ * @frame: call frame
+ * @cookie:
+ * @this:
+ * @op_ret:
+ * @op_errno:
+ * @fd:
+ * @inode:
+ * @buf:
+ *
+ */
+int32_t
+ioc_create_cbk (call_frame_t *frame,
+		void *cookie,
+		xlator_t *this,
+		int32_t op_ret,
+		int32_t op_errno,
+		fd_t *fd,
+		inode_t *inode,
+		struct stat *buf)
+{
+	ioc_local_t *local = frame->local;
+	ioc_table_t *table = this->private;
+	ioc_inode_t *ioc_inode = NULL;
+	uint32_t weight = 0;
+	const char *path = local->file_loc.path;
+
+	if (op_ret != -1) {
+		{
+			/* assign weight */
+			weight = ioc_get_priority (table, path);
+
+			ioc_inode = ioc_inode_update (table, inode, weight);
+			LOCK (&fd->inode->lock);
+			{
+				inode_ctx_put (fd->inode, this, 
+					       (uint64_t)(long)ioc_inode);
+			}
+			UNLOCK (&fd->inode->lock);
+		}
+		/* If mandatory locking has been enabled on this file,
+		   we disable caching on it */
+		if ((inode->st_mode & S_ISGID) && 
+		    !(inode->st_mode & S_IXGRP)) {
+			fd_ctx_set (fd, this, 1);
+		}
+
+		/* If O_DIRECT open, we disable caching on it */
+		if (local->flags & O_DIRECT){
+			/* O_DIRECT is only for one fd, not the inode 
+			 * as a whole 
+			 */
+			fd_ctx_set (fd, this, 1);
+		}
+    
+	}
+  
+	frame->local = NULL;
+	FREE (local);
+
+	STACK_UNWIND (frame, op_ret, op_errno, fd, inode, buf);
+
+	return 0;
+}
+
+/*
+ * ioc_open - open fop for io cache
+ * @frame:
+ * @this:
+ * @loc:
+ * @flags:
+ *
+ */
+int32_t
+ioc_open (call_frame_t *frame,
+	  xlator_t *this,
+	  loc_t *loc,
+	  int32_t flags,
+	  fd_t *fd)
+{
+  
+	ioc_local_t *local = CALLOC (1, sizeof (ioc_local_t));
+	ERR_ABORT (local);
+
+	local->flags = flags;
+	local->file_loc.path = loc->path;
+	local->file_loc.inode = loc->inode;
+  
+	frame->local = local;
+  
+	STACK_WIND (frame,
+		    ioc_open_cbk,
+		    FIRST_CHILD(this),
+		    FIRST_CHILD(this)->fops->open,
+		    loc,
+		    flags,
+		    fd);
+
+	return 0;
+}
+
+/*
+ * ioc_create - create fop for io cache
+ * 
+ * @frame:
+ * @this:
+ * @pathname:
+ * @flags:
+ * @mode:
+ *
+ */
+int32_t
+ioc_create (call_frame_t *frame,
+	    xlator_t *this,
+	    loc_t *loc,
+	    int32_t flags,
+	    mode_t mode,
+	    fd_t *fd)
+{
+	ioc_local_t *local = CALLOC (1, sizeof (ioc_local_t));
+	ERR_ABORT (local);
+
+	local->flags = flags;
+	local->file_loc.path = loc->path;
+	frame->local = local;
+
+	STACK_WIND (frame, ioc_create_cbk,
+		    FIRST_CHILD(this),
+		    FIRST_CHILD(this)->fops->create,
+		    loc, flags, mode, fd);
+	return 0;
+}
+
+
+
+
+/*
+ * ioc_release - release fop for io cache
+ * 
+ * @frame:
+ * @this:
+ * @fd:
+ *
+ */
+int32_t
+ioc_release (xlator_t *this,
+	     fd_t *fd)
+{
+	return 0;
+}
+
+/* 
+ * ioc_readv_disabled_cbk 
+ * @frame:
+ * @cookie:
+ * @this:
+ * @op_ret:
+ * @op_errno:
+ * @vector:
+ * @count:
+ *
+ */ 
+int32_t
+ioc_readv_disabled_cbk (call_frame_t *frame, 
+			void *cookie,
+			xlator_t *this,
+			int32_t op_ret,
+			int32_t op_errno,
+			struct iovec *vector,
+			int32_t count,
+			struct stat *stbuf)
+{
+	STACK_UNWIND (frame, op_ret, op_errno, vector, count, stbuf);
+	return 0;
+}
+
+
+int32_t
+ioc_need_prune (ioc_table_t *table)
+{
+	int64_t cache_difference = 0;
+  
+	ioc_table_lock (table);
+	{
+		cache_difference = table->cache_used - table->cache_size;
+	}
+	ioc_table_unlock (table);
+
+	if (cache_difference > 0)
+		return 1;
+	else 
+		return 0;
+}
+
+/*
+ * dispatch_requests -
+ * 
+ * @frame:
+ * @inode:
+ *
+ * 
+ */
+static void
+dispatch_requests (call_frame_t *frame,
+		   ioc_inode_t *ioc_inode,
+		   fd_t *fd,
+		   off_t offset,
+		   size_t size)
+{
+	ioc_local_t *local = frame->local;
+	ioc_table_t *table = ioc_inode->table;
+	ioc_page_t  *trav = NULL;
+	ioc_waitq_t *waitq = NULL;
+	off_t   rounded_offset = 0;
+	off_t   rounded_end = 0;
+	off_t   trav_offset = 0;
+	int32_t fault = 0;
+	int8_t  need_validate = 0;
+	int8_t  might_need_validate = 0;  /* if a page exists, do we need 
+					    to validate it? */
+
+	rounded_offset = floor (offset, table->page_size);
+	rounded_end = roof (offset + size, table->page_size);
+	trav_offset = rounded_offset;
+
+	/* once a frame does read, it should be waiting on something */
+	local->wait_count++;
+
+	/* Requested region can fall in three different pages,
+	 * 1. Ready - region is already in cache, we just have to serve it.
+	 * 2. In-transit - page fault has been generated on this page, we need
+	 *    to wait till the page is ready
+	 * 3. Fault - page is not in cache, we have to generate a page fault
+	 */
+
+	might_need_validate = ioc_inode_need_revalidate (ioc_inode);
+
+	while (trav_offset < rounded_end) {
+		size_t trav_size = 0;
+		off_t local_offset = 0;
+
+		ioc_inode_lock (ioc_inode);
+		//{
+
+		/* look for requested region in the cache */
+		trav = ioc_page_get (ioc_inode, trav_offset);
+
+		local_offset = max (trav_offset, offset);
+		trav_size = min (((offset+size) - local_offset), 
+				 table->page_size);
+
+		if (!trav) {
+			/* page not in cache, we need to generate page fault */
+			trav = ioc_page_create (ioc_inode, trav_offset);
+			fault = 1;
+			if (!trav) {
+				gf_log (frame->this->name, GF_LOG_CRITICAL,
+					"ioc_page_create returned NULL");
+			}
+		} 
+
+		ioc_wait_on_page (trav, frame, local_offset, trav_size);
+
+		if (trav->ready) {
+			/* page found in cache */
+			if (!might_need_validate) {
+				/* fresh enough */
+				gf_log (frame->this->name, GF_LOG_DEBUG,
+					"cache hit for trav_offset=%"PRId64""
+					"/local_offset=%"PRId64"",
+					trav_offset, local_offset);
+				waitq = ioc_page_wakeup (trav);
+			} else {
+				/* if waitq already exists, fstat revalidate is
+				   already on the way */
+				if (!ioc_inode->waitq) {
+					need_validate = 1;
+				}
+				ioc_wait_on_inode (ioc_inode, trav);
+			}
+		}
+
+		//}
+		ioc_inode_unlock (ioc_inode);
+    
+		ioc_waitq_return (waitq);
+		waitq = NULL;
+
+		if (fault) {
+			fault = 0;
+			/* new page created, increase the table->cache_used */
+			ioc_page_fault (ioc_inode, frame, fd, trav_offset);
+		}
+
+		if (need_validate) {
+			need_validate = 0;
+			gf_log (frame->this->name, GF_LOG_DEBUG,
+				"sending validate request for "
+				"inode(%"PRId64") at offset=%"PRId64"",
+				fd->inode->ino, trav_offset);
+			ioc_cache_validate (frame, ioc_inode, fd, trav);
+		}
+    
+		trav_offset += table->page_size;
+	}
+
+	ioc_frame_return (frame);
+
+	if (ioc_need_prune (ioc_inode->table)) {
+		ioc_prune (ioc_inode->table);
+	}
+
+	return;
+}
+
+
+/*
+ * ioc_readv -
+ * 
+ * @frame:
+ * @this:
+ * @fd:
+ * @size:
+ * @offset:
+ *
+ */
+int32_t
+ioc_readv (call_frame_t *frame,
+	   xlator_t *this,
+	   fd_t *fd,
+	   size_t size,
+	   off_t offset)
+{
+	uint64_t     tmp_ioc_inode = 0;
+	ioc_inode_t *ioc_inode = NULL;
+	ioc_local_t *local = NULL;
+	uint32_t     weight = 0;
+
+	inode_ctx_get (fd->inode, this, &tmp_ioc_inode);
+	ioc_inode = (ioc_inode_t *)(long)tmp_ioc_inode;
+	if (!ioc_inode) {
+		/* caching disabled, go ahead with normal readv */
+		STACK_WIND (frame, 
+			    ioc_readv_disabled_cbk,
+			    FIRST_CHILD (frame->this), 
+			    FIRST_CHILD (frame->this)->fops->readv,
+			    fd, 
+			    size, 
+			    offset);
+		return 0;
+	}
+
+	if (!fd_ctx_get (fd, this, NULL)) {
+		/* disable caching for this fd, go ahead with normal readv */
+		STACK_WIND (frame, 
+			    ioc_readv_disabled_cbk,
+			    FIRST_CHILD (frame->this), 
+			    FIRST_CHILD (frame->this)->fops->readv,
+			    fd, 
+			    size, 
+			    offset);
+		return 0;
+	}
+
+	local = (ioc_local_t *) CALLOC (1, sizeof (ioc_local_t));
+	ERR_ABORT (local);
+	INIT_LIST_HEAD (&local->fill_list);
+
+	frame->local = local;  
+	local->pending_offset = offset;
+	local->pending_size = size;
+	local->offset = offset;
+	local->size = size;
+	local->inode = ioc_inode;
+
+	gf_log (this->name, GF_LOG_DEBUG,
+		"NEW REQ (%p) offset = %"PRId64" && size = %"GF_PRI_SIZET"", 
+		frame, offset, size);
+
+	weight = ioc_inode->weight;
+
+	ioc_table_lock (ioc_inode->table);
+	{
+		list_move_tail (&ioc_inode->inode_lru, 
+				&ioc_inode->table->inode_lru[weight]);
+	}
+	ioc_table_unlock (ioc_inode->table);
+
+	dispatch_requests (frame, ioc_inode, fd, offset, size);
+  
+	return 0;
+}
+
+/*
+ * ioc_writev_cbk -
+ * 
+ * @frame:
+ * @cookie:
+ * @this:
+ * @op_ret:
+ * @op_errno:
+ *
+ */
+int32_t
+ioc_writev_cbk (call_frame_t *frame,
+		void *cookie,
+		xlator_t *this,
+		int32_t op_ret,
+		int32_t op_errno,
+		struct stat *stbuf)
+{
+	ioc_local_t *local     = frame->local;
+	uint64_t     ioc_inode = 0;
+
+	inode_ctx_get (local->fd->inode, this, &ioc_inode);
+  
+	if (ioc_inode)
+		ioc_inode_flush ((ioc_inode_t *)(long)ioc_inode);
+
+	STACK_UNWIND (frame, op_ret, op_errno, stbuf);
+	return 0;
+}
+
+/*
+ * ioc_writev
+ * 
+ * @frame:
+ * @this:
+ * @fd:
+ * @vector:
+ * @count:
+ * @offset:
+ *
+ */
+int32_t
+ioc_writev (call_frame_t *frame,
+	    xlator_t *this,
+	    fd_t *fd,
+	    struct iovec *vector,
+	    int32_t count,
+	    off_t offset)
+{
+	ioc_local_t *local     = NULL;
+	uint64_t     ioc_inode = 0;
+	
+	local = CALLOC (1, sizeof (ioc_local_t));
+	ERR_ABORT (local);
+
+	/* TODO: why is it not fd_ref'ed */
+	local->fd = fd;
+	frame->local = local;
+
+	inode_ctx_get (fd->inode, this, &ioc_inode);
+	if (ioc_inode)
+		ioc_inode_flush ((ioc_inode_t *)(long)ioc_inode);
+
+	STACK_WIND (frame,
+		    ioc_writev_cbk,
+		    FIRST_CHILD(this),
+		    FIRST_CHILD(this)->fops->writev,
+		    fd,
+		    vector,
+		    count,
+		    offset);
+
+	return 0;
+}
+
+/*
+ * ioc_truncate_cbk -
+ * 
+ * @frame:
+ * @cookie:
+ * @this:
+ * @op_ret:
+ * @op_errno:
+ * @buf:
+ *
+ */
+int32_t 
+ioc_truncate_cbk (call_frame_t *frame,
+		  void *cookie,
+		  xlator_t *this,
+		  int32_t op_ret,
+		  int32_t op_errno,
+		  struct stat *buf)
+{
+
+	STACK_UNWIND (frame, op_ret, op_errno, buf);
+	return 0;
+}
+
+/*
+ * ioc_truncate -
+ * 
+ * @frame:
+ * @this:
+ * @loc:
+ * @offset:
+ *
+ */
+int32_t 
+ioc_truncate (call_frame_t *frame,
+	      xlator_t *this,
+	      loc_t *loc,
+	      off_t offset)
+{
+	uint64_t ioc_inode = 0;
+	inode_ctx_get (loc->inode, this, &ioc_inode);
+
+	if (ioc_inode)
+		ioc_inode_flush ((ioc_inode_t *)(long)ioc_inode);
+
+	STACK_WIND (frame,
+		    ioc_truncate_cbk,
+		    FIRST_CHILD(this),
+		    FIRST_CHILD(this)->fops->truncate,
+		    loc,
+		    offset);
+	return 0;
+}
+
+/*
+ * ioc_ftruncate -
+ * 
+ * @frame:
+ * @this:
+ * @fd:
+ * @offset:
+ *
+ */
+int32_t
+ioc_ftruncate (call_frame_t *frame,
+	       xlator_t *this,
+	       fd_t *fd,
+	       off_t offset)
+{
+	uint64_t ioc_inode = 0;
+	inode_ctx_get (fd->inode, this, &ioc_inode);
+
+	if (ioc_inode)
+		ioc_inode_flush ((ioc_inode_t *)(long)ioc_inode);
+
+	STACK_WIND (frame,
+		    ioc_truncate_cbk,
+		    FIRST_CHILD(this),
+		    FIRST_CHILD(this)->fops->ftruncate,
+		    fd,
+		    offset);
+	return 0;
+}
+
+int32_t
+ioc_lk_cbk (call_frame_t *frame,
+	    void *cookie,
+	    xlator_t *this,
+	    int32_t op_ret,
+	    int32_t op_errno,
+	    struct flock *lock)
+{
+	STACK_UNWIND (frame, op_ret, op_errno, lock);
+	return 0;
+}
+
+int32_t 
+ioc_lk (call_frame_t *frame,
+	xlator_t *this,
+	fd_t *fd,
+	int32_t cmd,
+	struct flock *lock)
+{
+	ioc_inode_t *ioc_inode = NULL;
+	uint64_t     tmp_inode = 0;
+
+	inode_ctx_get (fd->inode, this, &tmp_inode);
+	ioc_inode = (ioc_inode_t *)(long)tmp_inode;
+	if (!ioc_inode) {
+		gf_log (this->name, GF_LOG_ERROR,
+			"inode context is NULL: returning EBADFD");
+		STACK_UNWIND (frame, -1, EBADFD, NULL);
+		return 0;
+	}
+
+	ioc_inode_lock (ioc_inode);
+	{
+		gettimeofday (&ioc_inode->tv, NULL);
+	}
+	ioc_inode_unlock (ioc_inode);
+
+	STACK_WIND (frame, ioc_lk_cbk, 
+		    FIRST_CHILD (this),
+		    FIRST_CHILD (this)->fops->lk, fd, cmd, lock);
+	return 0;
+}
+
+int32_t
+ioc_get_priority_list (const char *opt_str, struct list_head *first)
+{
+	int32_t max_pri = 0;
+	char *tmp_str = NULL;
+	char *tmp_str1 = NULL;
+	char *tmp_str2 = NULL;
+	char *dup_str = NULL;
+	char *stripe_str = NULL;
+	char *pattern = NULL;
+	char *priority = NULL;
+	char *string = strdup (opt_str);
+	struct ioc_priority *curr = NULL;
+
+	/* Get the pattern for cache priority. 
+	 * "option priority *.jpg:1,abc*:2" etc 
+	 */
+	/* TODO: inode_lru in table is statically hard-coded to 5, 
+	 * should be changed to run-time configuration 
+	 */
+	stripe_str = strtok_r (string, ",", &tmp_str);
+	while (stripe_str) {
+		curr = CALLOC (1, sizeof (struct ioc_priority));
+		ERR_ABORT (curr);
+		list_add_tail (&curr->list, first);
+
+		dup_str = strdup (stripe_str);
+		pattern = strtok_r (dup_str, ":", &tmp_str1);
+		if (!pattern)
+			return -1;
+		priority = strtok_r (NULL, ":", &tmp_str1);
+		if (!priority)
+			return -1;
+		gf_log ("io-cache", 
+			GF_LOG_DEBUG, 
+			"ioc priority : pattern %s : priority %s", 
+			pattern,
+			priority);
+		curr->pattern = strdup (pattern);
+		curr->priority = strtol (priority, &tmp_str2, 0);
+		if (tmp_str2 && (*tmp_str2))
+			return -1;
+		else
+			max_pri = max (max_pri, curr->priority);
+		stripe_str = strtok_r (NULL, ",", &tmp_str);
+	}
+
+	return max_pri;
+}
+
+/*
+ * init - 
+ * @this:
+ *
+ */
+int32_t 
+init (xlator_t *this)
+{
+	ioc_table_t *table;
+	dict_t *options = this->options;
+	uint32_t index = 0;
+	char *page_size_string = NULL;
+	char *cache_size_string = NULL;
+
+	if (!this->children || this->children->next) {
+		gf_log (this->name, GF_LOG_ERROR,
+			"FATAL: io-cache not configured with exactly "
+			"one child");
+		return -1;
+	}
+
+	if (!this->parents) {
+		gf_log (this->name, GF_LOG_WARNING,
+			"dangling volume. check volfile ");
+	}
+
+	table = (void *) CALLOC (1, sizeof (*table));
+	ERR_ABORT (table);
+  
+	table->xl = this;
+	table->page_size = IOC_PAGE_SIZE;
+	table->cache_size = IOC_CACHE_SIZE;
+
+	if (dict_get (options, "page-size"))
+		page_size_string = data_to_str (dict_get (options, 
+							  "page-size"));
+
+	if (page_size_string) {
+		if (gf_string2bytesize (page_size_string, 
+					&table->page_size) != 0) {
+			gf_log ("io-cache", GF_LOG_ERROR, 
+				"invalid number format \"%s\" of "
+				"\"option page-size\"", 
+				page_size_string);
+			return -1;
+		}
+		gf_log (this->name, GF_LOG_DEBUG, 
+			"using page-size %"PRIu64"",  table->page_size);
+	}
+  
+	if (dict_get (options, "cache-size"))
+		cache_size_string = data_to_str (dict_get (options, 
+							   "cache-size"));
+	if (cache_size_string) {
+		if (gf_string2bytesize (cache_size_string, 
+					&table->cache_size) != 0) {
+			gf_log ("io-cache", GF_LOG_ERROR, 
+				"invalid number format \"%s\" of "
+				"\"option cache-size\"", 
+				cache_size_string);
+			return -1;
+		}
+      
+		gf_log (this->name, GF_LOG_DEBUG, 
+			"using cache-size %"PRIu64"", table->cache_size);
+	}
+  
+	table->cache_timeout = 1;
+
+	if (dict_get (options, "cache-timeout")) {
+		table->cache_timeout = 
+			data_to_uint32 (dict_get (options,
+						  "cache-timeout"));
+		gf_log (this->name, GF_LOG_DEBUG,
+			"Using %d seconds to revalidate cache",
+			table->cache_timeout);
+	}
+
+	INIT_LIST_HEAD (&table->priority_list);
+	if (dict_get (options, "priority")) {
+		char *option_list = data_to_str (dict_get (options, 
+							   "priority"));
+		gf_log (this->name, GF_LOG_DEBUG,
+			"option path %s", option_list);
+		/* parse the list of pattern:priority */
+		table->max_pri = ioc_get_priority_list (option_list, 
+							&table->priority_list);
+    
+		if (table->max_pri == -1)
+			return -1;
+	}
+	table->max_pri ++;
+	INIT_LIST_HEAD (&table->inodes);
+  
+	table->inode_lru = CALLOC (table->max_pri, sizeof (struct list_head));
+	ERR_ABORT (table->inode_lru);
+	for (index = 0; index < (table->max_pri); index++)
+		INIT_LIST_HEAD (&table->inode_lru[index]);
+
+	pthread_mutex_init (&table->table_lock, NULL);
+	this->private = table;
+	return 0;
+}
+
+/*
+ * fini -
+ * 
+ * @this:
+ *
+ */
+void
+fini (xlator_t *this)
+{
+	ioc_table_t *table = this->private;
+
+	pthread_mutex_destroy (&table->table_lock);
+	FREE (table);
+
+	this->private = NULL;
+	return;
+}
+
+struct xlator_fops fops = {
+	.open        = ioc_open,
+	.create      = ioc_create,
+	.readv       = ioc_readv,
+	.writev      = ioc_writev,
+	.truncate    = ioc_truncate,
+	.ftruncate   = ioc_ftruncate,
+	.utimens     = ioc_utimens,
+	.lookup      = ioc_lookup,
+	.lk          = ioc_lk
+};
+
+struct xlator_mops mops = {
+};
+
+struct xlator_cbks cbks = {
+	.forget      = ioc_forget,
+  	.release     = ioc_release
+};
+
+struct volume_options options[] = {
+	{ .key  = {"priority"}, 
+	  .type = GF_OPTION_TYPE_ANY 
+	},
+	{ .key  = {"cache-timeout", "force-revalidate-timeout"},
+	  .type = GF_OPTION_TYPE_INT,
+	  .min  = 0, 
+	  .max  = 60 
+	}, 
+	{ .key  = {"page-size"}, 
+	  .type = GF_OPTION_TYPE_SIZET, 
+	  .min  = 16 * GF_UNIT_KB, 
+	  .max  =  4 * GF_UNIT_MB 
+	},
+	{ .key  = {"cache-size"}, 
+	  .type = GF_OPTION_TYPE_SIZET,
+	  .min  = 4 * GF_UNIT_MB, 
+	  .max  = 6 * GF_UNIT_GB 
+	},
+	{ .key = {NULL} },
+};
diff --git a/xlators/performance/io-cache/src/io-cache.h b/xlators/performance/io-cache/src/io-cache.h
new file mode 100644
index 00000000000..e997f6e7ca6
--- /dev/null
+++ b/xlators/performance/io-cache/src/io-cache.h
@@ -0,0 +1,330 @@
+/*
+   Copyright (c) 2007, 2008 Z RESEARCH, Inc. <http://www.zresearch.com>
+   This file is part of GlusterFS.
+
+   GlusterFS is free software; you can redistribute it and/or modify
+   it under the terms of the GNU General Public License as published
+   by the Free Software Foundation; either version 3 of the License,
+   or (at your option) any later version.
+
+   GlusterFS is distributed in the hope that it will be useful, but
+   WITHOUT ANY WARRANTY; without even the implied warranty of
+   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
+   General Public License for more details.
+
+   You should have received a copy of the GNU General Public License
+   along with this program.  If not, see
+   <http://www.gnu.org/licenses/>.
+*/
+
+#ifndef __IO_CACHE_H
+#define __IO_CACHE_H
+
+#ifndef _CONFIG_H
+#define _CONFIG_H
+#include "config.h"
+#endif
+
+#include <sys/types.h>
+#include "compat-errno.h"
+
+#include "glusterfs.h"
+#include "logging.h"
+#include "dict.h"
+#include "xlator.h"
+#include "common-utils.h"
+#include "call-stub.h"
+#include <sys/time.h>
+#include <fnmatch.h>
+
+#define IOC_PAGE_SIZE    (1024 * 128)   /* 128KB */
+#define IOC_CACHE_SIZE   (32 * 1024 * 1024)
+
+struct ioc_table;
+struct ioc_local;
+struct ioc_page;
+struct ioc_inode;
+
+struct ioc_priority {
+	struct list_head list;
+	char *pattern;
+	uint32_t priority;
+};
+
+/*
+ * ioc_waitq - this structure is used to represents the waiting 
+ *             frames on a page
+ *
+ * @next: pointer to next object in waitq
+ * @data: pointer to the frame which is waiting
+ */
+struct ioc_waitq {
+	struct ioc_waitq *next;
+	void *data;
+	off_t pending_offset;
+	size_t pending_size;
+};
+
+/*
+ * ioc_fill - 
+ *
+ */
+struct ioc_fill {
+	struct list_head list;  /* list of ioc_fill structures of a frame */
+	off_t offset;          
+	size_t size;           
+	struct iovec *vector;  
+	int32_t count;
+	dict_t *refs;
+};
+
+struct ioc_local {
+	mode_t mode;
+	int32_t flags;
+	loc_t file_loc;
+	off_t offset;
+	size_t size;
+	int32_t op_ret;
+	int32_t op_errno;
+	struct list_head fill_list;      /* list of ioc_fill structures */
+	off_t pending_offset;            /* offset from this frame should continue */
+	size_t pending_size;             /* size of data this frame is waiting on */
+	struct ioc_inode *inode;
+	int32_t wait_count;
+	pthread_mutex_t local_lock;
+	struct ioc_waitq *waitq;
+	void *stub;
+	fd_t *fd;
+	int32_t need_xattr;
+	dict_t *xattr_req;
+};
+
+/*
+ * ioc_page - structure to store page of data from file 
+ *
+ */
+struct ioc_page {
+	struct list_head pages;
+	struct list_head page_lru;
+	struct ioc_inode *inode;   /* inode this page belongs to */
+	struct ioc_priority *priority;
+	char dirty;
+	char ready;
+	struct iovec *vector;
+	int32_t count;
+	off_t offset;
+	size_t size;
+	struct ioc_waitq *waitq;
+	dict_t *ref;
+	pthread_mutex_t page_lock;
+};
+
+struct ioc_inode {
+	struct ioc_table *table;
+	struct list_head pages;      /* list of pages of this inode */
+	struct list_head inode_list; /* list of inodes, maintained by io-cache translator */
+	struct list_head inode_lru;
+	struct list_head page_lru;
+	struct ioc_waitq *waitq;
+	pthread_mutex_t inode_lock;
+	uint32_t weight;             /* weight of the inode, increases on each read */
+	time_t mtime;             /* mtime of the server file when last cached */
+	struct timeval tv;           /* time-stamp at last re-validate */
+};
+
+struct ioc_table {
+	uint64_t page_size;
+	uint64_t cache_size;
+	uint64_t cache_used;
+	struct list_head inodes; /* list of inodes cached */
+	struct list_head active; 
+	struct list_head *inode_lru;
+	struct list_head priority_list;
+	int32_t readv_count;
+	pthread_mutex_t table_lock;
+	xlator_t *xl;
+	uint32_t inode_count;
+	int32_t cache_timeout;
+	int32_t max_pri;
+};
+
+typedef struct ioc_table ioc_table_t;
+typedef struct ioc_local ioc_local_t;
+typedef struct ioc_page ioc_page_t;
+typedef struct ioc_inode ioc_inode_t;
+typedef struct ioc_waitq ioc_waitq_t;
+typedef struct ioc_fill ioc_fill_t;
+
+void *
+str_to_ptr (char *string);
+
+char *
+ptr_to_str (void *ptr);
+
+int32_t 
+ioc_readv_disabled_cbk (call_frame_t *frame,
+			void *cookie,
+			xlator_t *this,
+			int32_t op_ret,
+			int32_t op_errno,
+			struct iovec *vector,
+			int32_t count,
+			struct stat *stbuf);
+
+ioc_page_t *
+ioc_page_get (ioc_inode_t *ioc_inode,
+	      off_t offset);
+
+ioc_page_t *
+ioc_page_create (ioc_inode_t *ioc_inode,
+		 off_t offset);
+
+void
+ioc_page_fault (ioc_inode_t *ioc_inode,
+		call_frame_t *frame,
+		fd_t *fd,
+		off_t offset);
+void
+ioc_wait_on_page (ioc_page_t *page,
+		  call_frame_t *frame,
+		  off_t offset,
+		  size_t size);
+
+ioc_waitq_t *
+ioc_page_wakeup (ioc_page_t *page);
+
+void
+ioc_page_flush (ioc_page_t *page);
+
+ioc_waitq_t *
+ioc_page_error (ioc_page_t *page,
+		int32_t op_ret,
+		int32_t op_errno);
+void
+ioc_page_purge (ioc_page_t *page);
+
+void
+ioc_frame_return (call_frame_t *frame);
+
+void
+ioc_waitq_return (ioc_waitq_t *waitq);
+
+void
+ioc_frame_fill (ioc_page_t *page,
+		call_frame_t *frame,
+		off_t offset,
+		size_t size);
+
+#define ioc_inode_lock(ioc_inode)					\
+	do {								\
+		gf_log (ioc_inode->table->xl->name, GF_LOG_DEBUG,	\
+			"locked inode(%p)", ioc_inode);			\
+		pthread_mutex_lock (&ioc_inode->inode_lock);		\
+	} while (0)
+
+
+#define ioc_inode_unlock(ioc_inode)					\
+	do {								\
+		gf_log (ioc_inode->table->xl->name, GF_LOG_DEBUG,	\
+			"unlocked inode(%p)", ioc_inode);		\
+		pthread_mutex_unlock (&ioc_inode->inode_lock);		\
+	} while (0)
+
+
+#define ioc_table_lock(table)					\
+	do {							\
+		gf_log (table->xl->name, GF_LOG_DEBUG,		\
+			"locked table(%p)", table);		\
+		pthread_mutex_lock (&table->table_lock);	\
+	} while (0)
+
+
+#define ioc_table_unlock(table)					\
+	do {							\
+		gf_log (table->xl->name, GF_LOG_DEBUG,		\
+			"unlocked table(%p)", table);		\
+		pthread_mutex_unlock (&table->table_lock);	\
+	} while (0)
+
+
+#define ioc_local_lock(local)						\
+	do {								\
+		gf_log (local->inode->table->xl->name, GF_LOG_DEBUG,	\
+			"locked local(%p)", local);			\
+		pthread_mutex_lock (&local->local_lock);		\
+	} while (0)
+
+
+#define ioc_local_unlock(local)						\
+	do {								\
+		gf_log (local->inode->table->xl->name, GF_LOG_DEBUG,	\
+			"unlocked local(%p)", local);			\
+		pthread_mutex_unlock (&local->local_lock);		\
+	} while (0)
+
+
+#define ioc_page_lock(page)						\
+	do {								\
+		gf_log (page->inode->table->xl->name, GF_LOG_DEBUG,	\
+			"locked page(%p)", page);			\
+		pthread_mutex_lock (&page->page_lock);			\
+	} while (0)
+
+
+#define ioc_page_unlock(page)						\
+	do {								\
+		gf_log (page->inode->table->xl->name, GF_LOG_DEBUG,	\
+			"unlocked page(%p)", page);			\
+		pthread_mutex_unlock (&page->page_lock);		\
+	} while (0)
+
+
+static inline uint64_t
+time_elapsed (struct timeval *now,
+	      struct timeval *then)
+{
+	uint64_t sec = now->tv_sec - then->tv_sec;
+
+	if (sec)
+		return sec;
+  
+	return 0;
+}
+
+ioc_inode_t *
+ioc_inode_search (ioc_table_t *table,
+		  inode_t *inode);
+
+void 
+ioc_inode_destroy (ioc_inode_t *ioc_inode);
+
+ioc_inode_t *
+ioc_inode_update (ioc_table_t *table,
+		  inode_t *inode,
+		  uint32_t weight);
+
+int64_t 
+ioc_page_destroy (ioc_page_t *page);
+
+int32_t
+__ioc_inode_flush (ioc_inode_t *ioc_inode);
+
+void
+ioc_inode_flush (ioc_inode_t *ioc_inode);
+
+void
+ioc_inode_wakeup (call_frame_t *frame, 
+		  ioc_inode_t *ioc_inode, 
+		  struct stat *stbuf);
+
+int8_t
+ioc_cache_still_valid (ioc_inode_t *ioc_inode,
+		       struct stat *stbuf);
+
+int32_t
+ioc_prune (ioc_table_t *table);
+
+int32_t
+ioc_need_prune (ioc_table_t *table);
+
+#endif /* __READ_AHEAD_H */
diff --git a/xlators/performance/io-cache/src/ioc-inode.c b/xlators/performance/io-cache/src/ioc-inode.c
new file mode 100644
index 00000000000..2e2e561dd23
--- /dev/null
+++ b/xlators/performance/io-cache/src/ioc-inode.c
@@ -0,0 +1,201 @@
+/*
+  Copyright (c) 2007, 2008, 2009 Z RESEARCH, Inc. <http://www.zresearch.com>
+  This file is part of GlusterFS.
+
+  GlusterFS is free software; you can redistribute it and/or modify
+  it under the terms of the GNU General Public License as published
+  by the Free Software Foundation; either version 3 of the License,
+  or (at your option) any later version.
+
+  GlusterFS is distributed in the hope that it will be useful, but
+  WITHOUT ANY WARRANTY; without even the implied warranty of
+  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
+  General Public License for more details.
+
+  You should have received a copy of the GNU General Public License
+  along with this program.  If not, see
+  <http://www.gnu.org/licenses/>.
+*/
+
+#ifndef _CONFIG_H
+#define _CONFIG_H
+#include "config.h"
+#endif
+
+#include "io-cache.h"
+
+
+/*
+ * str_to_ptr - convert a string to pointer
+ * @string: string
+ *
+ */
+void *
+str_to_ptr (char *string)
+{
+	void *ptr = (void *)strtoul (string, NULL, 16);
+	return ptr;
+}
+
+
+/*
+ * ptr_to_str - convert a pointer to string
+ * @ptr: pointer
+ *
+ */
+char *
+ptr_to_str (void *ptr)
+{
+	char *str;
+	asprintf (&str, "%p", ptr);
+	return str;
+}
+
+void
+ioc_inode_wakeup (call_frame_t *frame,
+		  ioc_inode_t *ioc_inode, 
+		  struct stat *stbuf)
+{
+	ioc_waitq_t *waiter = NULL, *waited = NULL;
+	ioc_waitq_t *page_waitq = NULL;
+	int8_t cache_still_valid = 1;
+	ioc_local_t *local = frame->local;
+	int8_t need_fault = 0;
+	ioc_page_t *waiter_page = NULL;
+
+	ioc_inode_lock (ioc_inode);
+	{
+		waiter = ioc_inode->waitq;
+		ioc_inode->waitq = NULL;
+	}
+	ioc_inode_unlock (ioc_inode);
+
+	if (stbuf)
+		cache_still_valid = ioc_cache_still_valid (ioc_inode, stbuf);
+	else
+		cache_still_valid = 0;
+
+	if (!waiter) {
+		gf_log (frame->this->name, GF_LOG_DEBUG,
+			"cache validate called without any "
+			"page waiting to be validated");
+	}
+
+	while (waiter) {
+		waiter_page = waiter->data;
+		page_waitq = NULL;
+    
+		if (waiter_page) {
+			if (cache_still_valid) {
+				/* cache valid, wake up page */
+				ioc_inode_lock (ioc_inode);
+				{
+					page_waitq = 
+						ioc_page_wakeup (waiter_page);
+				}
+				ioc_inode_unlock (ioc_inode);
+				if (page_waitq)
+					ioc_waitq_return (page_waitq);
+			} else {
+				/* cache invalid, generate page fault and set 
+				 * page->ready = 0, to avoid double faults  
+				 */
+				ioc_inode_lock (ioc_inode);
+	
+				if (waiter_page->ready) {
+					waiter_page->ready = 0;
+					need_fault = 1;
+				} else {
+					gf_log (frame->this->name, 
+						GF_LOG_DEBUG,
+						"validate frame(%p) is waiting"
+						"for in-transit page = %p",
+						frame, waiter_page);
+				}
+	
+				ioc_inode_unlock (ioc_inode);
+      
+				if (need_fault) {
+					need_fault = 0;
+					ioc_page_fault (ioc_inode, frame, 
+							local->fd, 
+							waiter_page->offset);
+				}
+			}
+		}
+
+		waited = waiter;
+		waiter = waiter->next;
+    
+		waited->data = NULL;
+		free (waited);
+	}
+}
+
+/* 
+ * ioc_inode_update - create a new ioc_inode_t structure and add it to 
+ *                    the table table. fill in the fields which are derived 
+ *                    from inode_t corresponding to the file
+ * 
+ * @table: io-table structure
+ * @inode: inode structure
+ *
+ * not for external reference
+ */
+ioc_inode_t *
+ioc_inode_update (ioc_table_t *table, 
+		  inode_t *inode,
+		  uint32_t weight)
+{
+	ioc_inode_t *ioc_inode = CALLOC (1, sizeof (ioc_inode_t));
+	ERR_ABORT (ioc_inode);
+  
+	ioc_inode->table = table;
+ 
+	/* initialize the list for pages */
+	INIT_LIST_HEAD (&ioc_inode->pages);
+	INIT_LIST_HEAD (&ioc_inode->page_lru);
+
+	ioc_table_lock (table);
+
+	table->inode_count++;
+	list_add (&ioc_inode->inode_list, &table->inodes);
+	list_add_tail (&ioc_inode->inode_lru, &table->inode_lru[weight]);
+
+	gf_log (table->xl->name,
+		GF_LOG_DEBUG,
+		"adding to inode_lru[%d]", weight);
+
+	ioc_table_unlock (table);
+
+	pthread_mutex_init (&ioc_inode->inode_lock, NULL);
+	ioc_inode->weight = weight;
+  
+	return ioc_inode;
+}
+
+
+/* 
+ * ioc_inode_destroy - destroy an ioc_inode_t object.
+ *
+ * @inode: inode to destroy
+ *
+ * to be called only from ioc_forget. 
+ */
+void
+ioc_inode_destroy (ioc_inode_t *ioc_inode)
+{
+	ioc_table_t *table = ioc_inode->table;
+
+	ioc_table_lock (table);
+	table->inode_count--;
+	list_del (&ioc_inode->inode_list);
+	list_del (&ioc_inode->inode_lru);
+	ioc_table_unlock (table);
+  
+	ioc_inode_flush (ioc_inode);
+
+	pthread_mutex_destroy (&ioc_inode->inode_lock);
+	free (ioc_inode);
+}
+
diff --git a/xlators/performance/io-cache/src/page.c b/xlators/performance/io-cache/src/page.c
new file mode 100644
index 00000000000..e549f0bb547
--- /dev/null
+++ b/xlators/performance/io-cache/src/page.c
@@ -0,0 +1,778 @@
+/*
+   Copyright (c) 2007, 2008, 2009 Z RESEARCH, Inc. <http://www.zresearch.com>
+   This file is part of GlusterFS.
+
+   GlusterFS is free software; you can redistribute it and/or modify
+   it under the terms of the GNU General Public License as published
+   by the Free Software Foundation; either version 3 of the License,
+   or (at your option) any later version.
+
+   GlusterFS is distributed in the hope that it will be useful, but
+   WITHOUT ANY WARRANTY; without even the implied warranty of
+   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
+   General Public License for more details.
+
+   You should have received a copy of the GNU General Public License
+   along with this program.  If not, see
+   <http://www.gnu.org/licenses/>.
+*/
+
+#ifndef _CONFIG_H
+#define _CONFIG_H
+#include "config.h"
+#endif
+
+#include "glusterfs.h"
+#include "logging.h"
+#include "dict.h"
+#include "xlator.h"
+#include "io-cache.h"
+#include <assert.h>
+#include <sys/time.h>
+
+ioc_page_t *
+ioc_page_get (ioc_inode_t *ioc_inode,
+	      off_t offset)
+{
+	int8_t       found = 0;
+	ioc_page_t  *page = NULL;
+	ioc_table_t *table = ioc_inode->table;
+	off_t        rounded_offset = floor (offset, table->page_size);
+
+	if (list_empty (&ioc_inode->pages)) {
+		return NULL;
+	}
+
+	list_for_each_entry (page, &ioc_inode->pages, pages) {
+		if (page->offset == rounded_offset) {
+			found = 1;
+			break;
+		}
+	}
+
+	/* was previously returning ioc_inode itself.., 
+	 * 1st of its type and found one more downstairs :O */
+	if (!found){
+		page = NULL;
+	} else {
+		/* push the page to the end of the lru list */
+		list_move_tail (&page->page_lru, &ioc_inode->page_lru);
+	}
+
+	return page;
+}
+
+
+/*
+ * ioc_page_destroy -
+ *
+ * @page:
+ *
+ */
+int64_t
+ioc_page_destroy (ioc_page_t *page)
+{
+	int64_t page_size = 0;
+
+	page_size = page->size;
+
+	if (page->waitq) {
+		/* frames waiting on this page, do not destroy this page */
+		page_size = -1;
+	} else {
+
+		list_del (&page->pages);
+		list_del (&page->page_lru);
+    
+		gf_log (page->inode->table->xl->name, GF_LOG_DEBUG,
+			"destroying page = %p, offset = %"PRId64" "
+			"&& inode = %p",
+			page, page->offset, page->inode);
+    
+		if (page->vector){
+			dict_unref (page->ref);
+			free (page->vector);
+			page->vector = NULL;
+		}
+    
+		page->inode = NULL;
+    
+	}
+
+	if (page_size != -1) {
+		pthread_mutex_destroy (&page->page_lock);
+		free (page);
+	}
+
+	return page_size;
+}
+
+/*
+ * ioc_prune - prune the cache. we have a limit to the number of pages we
+ *             can have in-memory.
+ *
+ * @table: ioc_table_t of this translator
+ *
+ */
+int32_t
+ioc_prune (ioc_table_t *table)
+{
+	ioc_inode_t *curr = NULL, *next_ioc_inode = NULL;
+	ioc_page_t *page = NULL, *next = NULL;
+	int32_t ret = -1;
+	int32_t index = 0;
+	uint64_t size_to_prune = 0;
+	uint64_t size_pruned = 0;
+
+	ioc_table_lock (table);
+	{
+		size_to_prune = table->cache_used - table->cache_size;
+		/* take out the least recently used inode */
+		for (index=0; index < table->max_pri; index++) {
+			list_for_each_entry_safe (curr, next_ioc_inode, 
+						  &table->inode_lru[index], 
+						  inode_lru) {
+				/* prune page-by-page for this inode, till 
+				 * we reach the equilibrium */
+				ioc_inode_lock (curr);
+				/* { */
+
+				list_for_each_entry_safe (page, next, 
+							  &curr->page_lru, 
+							  page_lru) {
+					/* done with all pages, and not 
+					 * reached equilibrium yet??
+					 * continue with next inode in 
+					 * lru_list */
+					size_pruned += page->size;
+					ret = ioc_page_destroy (page);
+
+					if (ret != -1)
+						table->cache_used -= ret;
+	    
+					gf_log (table->xl->name,
+						GF_LOG_DEBUG,
+						"index = %d && table->cache_"
+						"used = %"PRIu64" && table->"
+						"cache_size = %"PRIu64, 
+						index, table->cache_used, 
+						table->cache_size);
+	    
+					if (size_pruned >= size_to_prune)
+						break;
+				} /* list_for_each_entry_safe(page...) */
+				if (list_empty (&curr->pages)) {
+					list_del_init (&curr->inode_lru);
+				}
+
+				/* } */ 
+				ioc_inode_unlock (curr);
+	
+				if (size_pruned >= size_to_prune)
+					break;
+			} /* list_for_each_entry_safe (curr...) */
+      
+			if (size_pruned >= size_to_prune)
+				break;
+		} /* for(index=0;...) */
+
+	} /* ioc_inode_table locked region end */
+	ioc_table_unlock (table);
+
+	return 0;
+}
+
+/*
+ * ioc_page_create - create a new page. 
+ *
+ * @ioc_inode: 
+ * @offset:
+ *
+ */
+ioc_page_t *
+ioc_page_create (ioc_inode_t *ioc_inode,
+		 off_t offset)
+{
+	ioc_table_t *table = ioc_inode->table;
+	ioc_page_t *page = NULL;
+	off_t rounded_offset = floor (offset, table->page_size);
+	ioc_page_t *newpage = CALLOC (1, sizeof (*newpage));
+	ERR_ABORT (newpage);
+  
+	if (ioc_inode)
+		table = ioc_inode->table;
+	else {
+		return NULL;
+	}
+   
+	newpage->offset = rounded_offset;
+	newpage->inode = ioc_inode;
+	pthread_mutex_init (&newpage->page_lock, NULL);
+
+	list_add_tail (&newpage->page_lru, &ioc_inode->page_lru);
+	list_add_tail (&newpage->pages, &ioc_inode->pages);
+
+	page = newpage;
+
+	gf_log ("io-cache", GF_LOG_DEBUG,
+		"returning new page %p", page);
+	return page;
+}
+
+/* 
+ * ioc_wait_on_page - pause a frame to wait till the arrival of a page. 
+ * here we need to handle the case when the frame who calls wait_on_page 
+ * himself has caused page_fault 
+ *
+ * @page: page to wait on
+ * @frame: call frame who is waiting on page
+ *
+ */
+void
+ioc_wait_on_page (ioc_page_t *page,
+		  call_frame_t *frame,
+		  off_t offset,
+		  size_t size)
+{
+	ioc_waitq_t *waitq = NULL;
+	ioc_local_t *local = frame->local;
+
+	waitq = CALLOC (1, sizeof (*waitq));
+	ERR_ABORT (waitq);
+  
+	gf_log (frame->this->name, GF_LOG_DEBUG,
+		"frame(%p) waiting on page = %p, offset=%"PRId64", "
+		"size=%"GF_PRI_SIZET"",
+		frame, page, offset, size);
+
+	waitq->data = frame;
+	waitq->next = page->waitq;
+	waitq->pending_offset = offset;
+	waitq->pending_size = size;
+	page->waitq = waitq;
+	/* one frame can wait only once on a given page, 
+	 * local->wait_count is number of pages a frame is waiting on */
+	ioc_local_lock (local);
+	{
+		local->wait_count++;
+	}
+	ioc_local_unlock (local);
+}
+
+
+/*
+ * ioc_cache_still_valid - see if cached pages ioc_inode are still valid 
+ * against given stbuf
+ *
+ * @ioc_inode:
+ * @stbuf:
+ *
+ * assumes ioc_inode is locked
+ */
+int8_t
+ioc_cache_still_valid (ioc_inode_t *ioc_inode,
+		       struct stat *stbuf)
+{
+	int8_t cache_still_valid = 1;
+  
+#if 0
+	if (!stbuf || (stbuf->st_mtime != ioc_inode->mtime) || 
+	    (stbuf->st_mtim.tv_nsec != ioc_inode->stbuf.st_mtim.tv_nsec))
+		cache_still_valid = 0;
+
+#else
+	if (!stbuf || (stbuf->st_mtime != ioc_inode->mtime))
+		cache_still_valid = 0;
+
+#endif
+
+#if 0
+	/* talk with avati@zresearch.com to enable this section */
+	if (!ioc_inode->mtime && stbuf) {
+		cache_still_valid = 1;
+		ioc_inode->mtime = stbuf->st_mtime;
+	}
+#endif
+
+	return cache_still_valid;
+}
+
+
+void
+ioc_waitq_return (ioc_waitq_t *waitq)
+{
+	ioc_waitq_t  *trav   = NULL;
+	ioc_waitq_t  *next   = NULL;
+	call_frame_t *frame = NULL;
+
+	for (trav = waitq; trav; trav = next) {
+		next = trav->next;
+
+		frame = trav->data;
+		ioc_frame_return (frame);
+		free (trav);
+	}
+}
+
+
+int
+ioc_fault_cbk (call_frame_t *frame,
+	       void *cookie,
+	       xlator_t *this,
+	       int32_t op_ret,
+	       int32_t op_errno,
+	       struct iovec *vector,
+	       int32_t count,
+	       struct stat *stbuf)
+{
+	ioc_local_t *local = frame->local;
+	off_t offset = local->pending_offset;
+	ioc_inode_t *ioc_inode = local->inode;
+	ioc_table_t *table = ioc_inode->table;
+	ioc_page_t *page = NULL;
+	off_t trav_offset = 0;
+	size_t payload_size = 0;
+	int32_t destroy_size = 0;
+	size_t page_size = 0;
+	ioc_waitq_t *waitq = NULL;
+
+	trav_offset = offset;  
+	payload_size = op_ret;
+
+	ioc_inode_lock (ioc_inode);
+	{
+		if (op_ret == -1 || 
+		    (op_ret >= 0 && 
+		     !ioc_cache_still_valid(ioc_inode, stbuf))) {
+			gf_log (ioc_inode->table->xl->name, GF_LOG_DEBUG,
+				"cache for inode(%p) is invalid. flushing "
+				"all pages", ioc_inode);
+			destroy_size = __ioc_inode_flush (ioc_inode);
+		} 
+    
+		if (op_ret >= 0)
+			ioc_inode->mtime = stbuf->st_mtime;
+    
+		gettimeofday (&ioc_inode->tv, NULL);
+    
+		if (op_ret < 0) {
+			/* error, readv returned -1 */
+			page = ioc_page_get (ioc_inode, offset);
+			if (page)
+				waitq = ioc_page_error (page, op_ret, 
+							op_errno);
+		} else {
+			gf_log (ioc_inode->table->xl->name, GF_LOG_DEBUG,
+				"op_ret = %d", op_ret);
+			page = ioc_page_get (ioc_inode, offset);
+			if (!page) {
+				/* page was flushed */
+				/* some serious bug ? */
+				gf_log (this->name, GF_LOG_DEBUG,
+					"wasted copy: %"PRId64"[+%"PRId64"] "
+					"ioc_inode=%p", offset, 
+					table->page_size, ioc_inode);
+			} else {
+				if (page->vector) {
+					dict_unref (page->ref);
+					free (page->vector);
+					page->vector = NULL;
+				}
+	
+				/* keep a copy of the page for our cache */
+				page->vector = iov_dup (vector, count);
+				page->count = count;
+				if (frame->root->rsp_refs) {
+					dict_ref (frame->root->rsp_refs);
+					page->ref = frame->root->rsp_refs;
+				} else {
+					/* TODO: we have got a response to 
+					 * our request and no data */
+					gf_log (this->name, GF_LOG_CRITICAL,
+						"frame>root>rsp_refs is null");
+				} /* if(frame->root->rsp_refs) */
+	
+				/* page->size should indicate exactly how 
+				 * much the readv call to the child
+				 * translator returned. earlier op_ret 
+				 * from child translator was used, which 
+				 * gave rise to a bug where reads from 
+				 * io-cached volume were resulting in 0 
+				 * byte replies */
+				page_size = iov_length(vector, count);
+	
+				page->size = page_size;
+
+				if (page->waitq) {
+					/* wake up all the frames waiting on 
+					 * this page, including 
+					 * the frame which triggered fault */
+					waitq = ioc_page_wakeup (page);
+				} /* if(page->waitq) */
+			} /* if(!page)...else */
+		} /* if(op_ret < 0)...else */
+	} /* ioc_inode locked region end */
+	ioc_inode_unlock (ioc_inode);
+
+	ioc_waitq_return (waitq);
+
+	if (page_size) {
+		ioc_table_lock (table);
+		{
+			table->cache_used += page_size;
+		}
+		ioc_table_unlock (table);
+	}
+
+	if (destroy_size) {
+		ioc_table_lock (table);
+		{
+			table->cache_used -= destroy_size;
+		}
+		ioc_table_unlock (table);
+	}
+
+	if (ioc_need_prune (ioc_inode->table)) {
+		ioc_prune (ioc_inode->table);
+	}
+
+	gf_log (this->name, GF_LOG_DEBUG, "fault frame %p returned", frame);
+	pthread_mutex_destroy (&local->local_lock);
+
+	fd_unref (local->fd);
+
+	STACK_DESTROY (frame->root);
+	return 0;
+}
+
+/*
+ * ioc_page_fault -
+ * 
+ * @ioc_inode:
+ * @frame:
+ * @fd:
+ * @offset:
+ *
+ */
+void
+ioc_page_fault (ioc_inode_t *ioc_inode,
+		call_frame_t *frame,
+		fd_t *fd,
+		off_t offset)
+{
+	ioc_table_t *table = ioc_inode->table;
+	call_frame_t *fault_frame = copy_frame (frame);
+	ioc_local_t *fault_local = CALLOC (1, sizeof (ioc_local_t));
+	ERR_ABORT (fault_local);
+
+	/* NOTE: copy_frame() means, the frame the fop whose fd_ref we 
+	 * are using till now won't be valid till we get reply from server. 
+	 * we unref this fd, in fault_cbk */
+	fault_local->fd = fd_ref (fd);
+
+	fault_frame->local = fault_local;
+	pthread_mutex_init (&fault_local->local_lock, NULL);
+
+	INIT_LIST_HEAD (&fault_local->fill_list);
+	fault_local->pending_offset = offset;
+	fault_local->pending_size = table->page_size;
+	fault_local->inode = ioc_inode;
+
+	gf_log (frame->this->name, GF_LOG_DEBUG,
+		"stack winding page fault for offset = %"PRId64" with "
+		"frame %p", offset, fault_frame);
+  
+	STACK_WIND (fault_frame, ioc_fault_cbk,
+		    FIRST_CHILD(fault_frame->this),
+		    FIRST_CHILD(fault_frame->this)->fops->readv,
+		    fd, table->page_size, offset);
+	return;
+}
+
+void
+ioc_frame_fill (ioc_page_t *page,
+		call_frame_t *frame,
+		off_t offset,
+		size_t size)
+{
+	ioc_local_t *local = frame->local;
+	ioc_fill_t *fill = NULL;
+	off_t src_offset = 0;
+	off_t dst_offset = 0;
+	ssize_t copy_size = 0;
+	ioc_inode_t *ioc_inode = page->inode;
+  
+	gf_log (frame->this->name, GF_LOG_DEBUG,
+		"frame (%p) offset = %"PRId64" && size = %"GF_PRI_SIZET" "
+		"&& page->size = %"GF_PRI_SIZET" && wait_count = %d", 
+		frame, offset, size, page->size, local->wait_count);
+
+	/* immediately move this page to the end of the page_lru list */
+	list_move_tail (&page->page_lru, &ioc_inode->page_lru);
+	/* fill local->pending_size bytes from local->pending_offset */
+	if (local->op_ret != -1 && page->size) {
+		if (offset > page->offset)
+			/* offset is offset in file, convert it to offset in 
+			 * page */
+			src_offset = offset - page->offset;
+		/*FIXME: since offset is the offset within page is the 
+		 * else case valid? */
+		else
+			/* local->pending_offset is in previous page. do not
+			 * fill until we have filled all previous pages */
+			dst_offset = page->offset - offset;
+
+		/* we have to copy from offset to either end of this page 
+		 * or till the requested size */
+		copy_size = min (page->size - src_offset,
+				 size - dst_offset);
+
+		if (copy_size < 0) {
+			/* if page contains fewer bytes and the required offset
+			   is beyond the page size in the page */
+			copy_size = src_offset = 0;
+		}
+    
+		gf_log (page->inode->table->xl->name, GF_LOG_DEBUG,
+			"copy_size = %"GF_PRI_SIZET" && src_offset = "
+			"%"PRId64" && dst_offset = %"PRId64"",
+			copy_size, src_offset, dst_offset);
+
+		{
+			ioc_fill_t *new = CALLOC (1, sizeof (*new));
+			ERR_ABORT (new);
+			new->offset = page->offset;
+			new->size = copy_size;
+			new->refs = dict_ref (page->ref);
+			new->count = iov_subset (page->vector,
+						 page->count,
+						 src_offset,
+						 src_offset + copy_size,
+						 NULL);
+			new->vector = CALLOC (new->count, 
+					      sizeof (struct iovec));
+			ERR_ABORT (new->vector);
+			new->count = iov_subset (page->vector,
+						 page->count,
+						 src_offset,
+						 src_offset + copy_size,
+						 new->vector);
+
+
+
+			/* add the ioc_fill to fill_list for this frame */
+			if (list_empty (&local->fill_list)) {
+				/* if list is empty, then this is the first 
+				 * time we are filling frame, add the 
+				 * ioc_fill_t to the end of list */
+				list_add_tail (&new->list, &local->fill_list);
+			} else {
+				int8_t found = 0;
+				/* list is not empty, we need to look for 
+				 * where this offset fits in list */
+				list_for_each_entry (fill, &local->fill_list, 
+						     list) {
+					if (fill->offset > new->offset) {
+						found = 1;
+						break;
+					}
+				}
+
+				if (found) {
+					found = 0;
+					list_add_tail (&new->list, 
+						       &fill->list);
+				} else {
+					list_add_tail (&new->list, 
+						       &local->fill_list);
+				}
+			}
+		}
+		local->op_ret += copy_size;
+	}
+}
+
+/*
+ * ioc_frame_unwind - frame unwinds only from here 
+ *
+ * @frame: call frame to unwind
+ *
+ * to be used only by ioc_frame_return(), when a frame has
+ * finished waiting on all pages, required
+ *
+ */
+static void
+ioc_frame_unwind (call_frame_t *frame)
+{
+	ioc_local_t *local = frame->local;
+	ioc_fill_t *fill = NULL, *next = NULL;
+	int32_t count = 0;
+	struct iovec *vector = NULL;
+	int32_t copied = 0;
+	dict_t *refs = NULL;
+	struct stat stbuf = {0,};
+	int32_t op_ret = 0;
+
+	//  ioc_local_lock (local);
+	refs = get_new_dict ();
+
+	frame->local = NULL;
+
+	if (list_empty (&local->fill_list)) {
+		gf_log (frame->this->name, GF_LOG_DEBUG,
+			"frame(%p) has 0 entries in local->fill_list "
+			"(offset = %"PRId64" && size = %"GF_PRI_SIZET")",
+			frame, local->offset, local->size);
+	}
+
+	list_for_each_entry (fill, &local->fill_list, list) {
+		count += fill->count;
+	}
+
+	vector = CALLOC (count, sizeof (*vector));
+	ERR_ABORT (vector);
+  
+	list_for_each_entry_safe (fill, next, &local->fill_list, list) {
+		memcpy (((char *)vector) + copied,
+			fill->vector,
+			fill->count * sizeof (*vector));
+    
+		copied += (fill->count * sizeof (*vector));
+
+		dict_copy (fill->refs, refs);
+
+		list_del (&fill->list);
+		dict_unref (fill->refs);
+		free (fill->vector);
+		free (fill);
+	}
+  
+	frame->root->rsp_refs = dict_ref (refs);
+  
+	op_ret = iov_length (vector, count);
+	gf_log (frame->this->name, GF_LOG_DEBUG,
+		"frame(%p) unwinding with op_ret=%d", frame, op_ret);
+
+	//  ioc_local_unlock (local);
+
+	STACK_UNWIND (frame,
+		      op_ret,
+		      local->op_errno,
+		      vector,
+		      count,
+		      &stbuf);
+
+	dict_unref (refs);
+    
+	pthread_mutex_destroy (&local->local_lock);
+	free (local);
+	free (vector);
+
+	return;
+}
+
+/*
+ * ioc_frame_return -
+ * @frame:
+ *
+ * to be called only when a frame is waiting on an in-transit page
+ */
+void
+ioc_frame_return (call_frame_t *frame)
+{
+	ioc_local_t *local = frame->local;
+	int32_t wait_count;
+	assert (local->wait_count > 0);
+
+	ioc_local_lock (local);
+	{
+		wait_count = --local->wait_count;
+	}
+	ioc_local_unlock (local);
+
+	if (!wait_count) {
+		ioc_frame_unwind (frame);
+	} 
+
+	return;
+}
+
+/* 
+ * ioc_page_wakeup -
+ * @page:
+ *
+ * to be called only when a frame is waiting on an in-transit page
+ */
+ioc_waitq_t *
+ioc_page_wakeup (ioc_page_t *page)
+{
+	ioc_waitq_t *waitq = NULL, *trav = NULL;
+	call_frame_t *frame = NULL;
+
+	waitq = page->waitq;
+	page->waitq = NULL;
+
+	trav = waitq;
+	page->ready = 1;
+
+	gf_log (page->inode->table->xl->name, GF_LOG_DEBUG,
+		"page is %p && waitq = %p", page, waitq);
+  
+	for (trav = waitq; trav; trav = trav->next) {
+		frame = trav->data; 
+		ioc_frame_fill (page, frame, trav->pending_offset, 
+				trav->pending_size);
+	}
+	
+	return waitq;
+}
+
+
+/*
+ * ioc_page_error -
+ * @page:
+ * @op_ret:
+ * @op_errno:
+ *
+ */
+ioc_waitq_t *
+ioc_page_error (ioc_page_t *page,
+		int32_t op_ret,
+		int32_t op_errno)
+{
+	ioc_waitq_t *waitq = NULL, *trav = NULL;
+	call_frame_t *frame = NULL;
+	int64_t ret = 0;
+	ioc_table_t *table = NULL;
+	ioc_local_t *local = NULL;
+
+	waitq = page->waitq;
+	page->waitq = NULL;
+  
+	gf_log (page->inode->table->xl->name, GF_LOG_DEBUG,
+		"page error for page = %p & waitq = %p", page, waitq);
+
+	for (trav = waitq; trav; trav = trav->next) {
+
+		frame = trav->data;
+
+		local = frame->local;
+		ioc_local_lock (local);
+		{
+			if (local->op_ret != -1) {
+				local->op_ret = op_ret;
+				local->op_errno = op_errno;
+			}
+		}
+		ioc_local_unlock (local);
+	}
+
+	table = page->inode->table;
+	ret = ioc_page_destroy (page);
+
+	if (ret != -1) {
+		table->cache_used -= ret;
+	}
+
+	return waitq;
+}