Orangefs: kernel client part 2

Signed-off-by: Mike Marshall <hubcap@omnibond.com>

fs/orangefs/acl.c

+/*
+ * (C) 2001 Clemson University and The University of Chicago
+ *
+ * See COPYING in top-level directory.
+ */
+
+#include "protocol.h"
+#include "pvfs2-kernel.h"
+#include "pvfs2-bufmap.h"
+#include <linux/posix_acl_xattr.h>
+#include <linux/fs_struct.h>
+
+struct posix_acl *pvfs2_get_acl(struct inode *inode, int type)
+{
+	struct posix_acl *acl;
+	int ret;
+	char *key = NULL, *value = NULL;
+
+	switch (type) {
+	case ACL_TYPE_ACCESS:
+		key = PVFS2_XATTR_NAME_ACL_ACCESS;
+		break;
+	case ACL_TYPE_DEFAULT:
+		key = PVFS2_XATTR_NAME_ACL_DEFAULT;
+		break;
+	default:
+		gossip_err("pvfs2_get_acl: bogus value of type %d\n", type);
+		return ERR_PTR(-EINVAL);
+	}
+	/*
+	 * Rather than incurring a network call just to determine the exact
+	 * length of the attribute, I just allocate a max length to save on
+	 * the network call. Conceivably, we could pass NULL to
+	 * pvfs2_inode_getxattr() to probe the length of the value, but
+	 * I don't do that for now.
+	 */
+	value = kmalloc(PVFS_MAX_XATTR_VALUELEN, GFP_KERNEL);
+	if (value == NULL)
+		return ERR_PTR(-ENOMEM);
+
+	gossip_debug(GOSSIP_ACL_DEBUG,
+		     "inode %pU, key %s, type %d\n",
+		     get_khandle_from_ino(inode),
+		     key,
+		     type);
+	ret = pvfs2_inode_getxattr(inode,
+				   "",
+				   key,
+				   value,
+				   PVFS_MAX_XATTR_VALUELEN);
+	/* if the key exists, convert it to an in-memory rep */
+	if (ret > 0) {
+		acl = posix_acl_from_xattr(&init_user_ns, value, ret);
+	} else if (ret == -ENODATA || ret == -ENOSYS) {
+		acl = NULL;
+	} else {
+		gossip_err("inode %pU retrieving acl's failed with error %d\n",
+			   get_khandle_from_ino(inode),
+			   ret);
+		acl = ERR_PTR(ret);
+	}
+	/* kfree(NULL) is safe, so don't worry if value ever got used */
+	kfree(value);
+	return acl;
+}
+
+int pvfs2_set_acl(struct inode *inode, struct posix_acl *acl, int type)
+{
+	struct pvfs2_inode_s *pvfs2_inode = PVFS2_I(inode);
+	int error = 0;
+	void *value = NULL;
+	size_t size = 0;
+	const char *name = NULL;
+
+	switch (type) {
+	case ACL_TYPE_ACCESS:
+		name = PVFS2_XATTR_NAME_ACL_ACCESS;
+		if (acl) {
+			umode_t mode = inode->i_mode;
+			/*
+			 * can we represent this with the traditional file
+			 * mode permission bits?
+			 */
+			error = posix_acl_equiv_mode(acl, &mode);
+			if (error < 0) {
+				gossip_err("%s: posix_acl_equiv_mode err: %d\n",
+					   __func__,
+					   error);
+				return error;
+			}
+
+			if (inode->i_mode != mode)
+				SetModeFlag(pvfs2_inode);
+			inode->i_mode = mode;
+			mark_inode_dirty_sync(inode);
+			if (error == 0)
+				acl = NULL;
+		}
+		break;
+	case ACL_TYPE_DEFAULT:
+		name = PVFS2_XATTR_NAME_ACL_DEFAULT;
+		break;
+	default:
+		gossip_err("%s: invalid type %d!\n", __func__, type);
+		return -EINVAL;
+	}
+
+	gossip_debug(GOSSIP_ACL_DEBUG,
+		     "%s: inode %pU, key %s type %d\n",
+		     __func__, get_khandle_from_ino(inode),
+		     name,
+		     type);
+
+	if (acl) {
+		size = posix_acl_xattr_size(acl->a_count);
+		value = kmalloc(size, GFP_KERNEL);
+		if (!value)
+			return -ENOMEM;
+
+		error = posix_acl_to_xattr(&init_user_ns, acl, value, size);
+		if (error < 0)
+			goto out;
+	}
+
+	gossip_debug(GOSSIP_ACL_DEBUG,
+		     "%s: name %s, value %p, size %zd, acl %p\n",
+		     __func__, name, value, size, acl);
+	/*
+	 * Go ahead and set the extended attribute now. NOTE: Suppose acl
+	 * was NULL, then value will be NULL and size will be 0 and that
+	 * will xlate to a removexattr. However, we don't want removexattr
+	 * complain if attributes does not exist.
+	 */
+	error = pvfs2_inode_setxattr(inode, "", name, value, size, 0);
+
+out:
+	kfree(value);
+	if (!error)
+		set_cached_acl(inode, type, acl);
+	return error;
+}
+
+int pvfs2_init_acl(struct inode *inode, struct inode *dir)
+{
+	struct pvfs2_inode_s *pvfs2_inode = PVFS2_I(inode);
+	struct posix_acl *default_acl, *acl;
+	umode_t mode = inode->i_mode;
+	int error = 0;
+
+	ClearModeFlag(pvfs2_inode);
+
+	error = posix_acl_create(dir, &mode, &default_acl, &acl);
+	if (error)
+		return error;
+
+	if (default_acl) {
+		error = pvfs2_set_acl(inode, default_acl, ACL_TYPE_DEFAULT);
+		posix_acl_release(default_acl);
+	}
+
+	if (acl) {
+		if (!error)
+			error = pvfs2_set_acl(inode, acl, ACL_TYPE_ACCESS);
+		posix_acl_release(acl);
+	}
+
+	/* If mode of the inode was changed, then do a forcible ->setattr */
+	if (mode != inode->i_mode) {
+		SetModeFlag(pvfs2_inode);
+		inode->i_mode = mode;
+		pvfs2_flush_inode(inode);
+	}
+
+	return error;
+}

fs/orangefs/dcache.c

+/*
+ * (C) 2001 Clemson University and The University of Chicago
+ *
+ * See COPYING in top-level directory.
+ */
+
+/*
+ *  Implementation of dentry (directory cache) functions.
+ */
+
+#include "protocol.h"
+#include "pvfs2-kernel.h"
+
+/* Returns 1 if dentry can still be trusted, else 0. */
+static int pvfs2_revalidate_lookup(struct dentry *dentry)
+{
+	struct dentry *parent_dentry = dget_parent(dentry);
+	struct inode *parent_inode = parent_dentry->d_inode;
+	struct pvfs2_inode_s *parent = PVFS2_I(parent_inode);
+	struct inode *inode = dentry->d_inode;
+	struct pvfs2_kernel_op_s *new_op;
+	int ret = 0;
+	int err = 0;
+
+	gossip_debug(GOSSIP_DCACHE_DEBUG, "%s: attempting lookup.\n", __func__);
+
+	new_op = op_alloc(PVFS2_VFS_OP_LOOKUP);
+	if (!new_op)
+		goto out_put_parent;
+
+	new_op->upcall.req.lookup.sym_follow = PVFS2_LOOKUP_LINK_NO_FOLLOW;
+	new_op->upcall.req.lookup.parent_refn = parent->refn;
+	strncpy(new_op->upcall.req.lookup.d_name,
+		dentry->d_name.name,
+		PVFS2_NAME_LEN);
+
+	gossip_debug(GOSSIP_DCACHE_DEBUG,
+		     "%s:%s:%d interrupt flag [%d]\n",
+		     __FILE__,
+		     __func__,
+		     __LINE__,
+		     get_interruptible_flag(parent_inode));
+
+	err = service_operation(new_op, "pvfs2_lookup",
+			get_interruptible_flag(parent_inode));
+	if (err)
+		goto out_drop;
+
+	if (new_op->downcall.status != 0 ||
+	    !match_handle(new_op->downcall.resp.lookup.refn.khandle, inode)) {
+		gossip_debug(GOSSIP_DCACHE_DEBUG,
+			"%s:%s:%d "
+			"lookup failure |%s| or no match |%s|.\n",
+			__FILE__,
+			__func__,
+			__LINE__,
+			new_op->downcall.status ? "true" : "false",
+			match_handle(new_op->downcall.resp.lookup.refn.khandle,
+					inode) ? "false" : "true");
+		gossip_debug(GOSSIP_DCACHE_DEBUG,
+			     "%s:%s:%d revalidate failed\n",
+			     __FILE__, __func__, __LINE__);
+		goto out_drop;
+	}
+
+	ret = 1;
+out_release_op:
+	op_release(new_op);
+out_put_parent:
+	dput(parent_dentry);
+	return ret;
+out_drop:
+	d_drop(dentry);
+	goto out_release_op;
+}
+
+/*
+ * Verify that dentry is valid.
+ *
+ * Should return 1 if dentry can still be trusted, else 0
+ */
+static int pvfs2_d_revalidate(struct dentry *dentry, unsigned int flags)
+{
+	struct inode *inode;
+	int ret = 0;
+
+	if (flags & LOOKUP_RCU)
+		return -ECHILD;
+
+	gossip_debug(GOSSIP_DCACHE_DEBUG, "%s: called on dentry %p.\n",
+		     __func__, dentry);
+
+	/* find inode from dentry */
+	if (!dentry->d_inode) {
+		gossip_debug(GOSSIP_DCACHE_DEBUG, "%s: negative dentry.\n",
+			     __func__);
+		goto invalid_exit;
+	}
+
+	gossip_debug(GOSSIP_DCACHE_DEBUG, "%s: inode valid.\n", __func__);
+	inode = dentry->d_inode;
+
+	/*
+	 * first perform a lookup to make sure that the object not only
+	 * exists, but is still in the expected place in the name space
+	 */
+	if (!is_root_handle(inode)) {
+		if (!pvfs2_revalidate_lookup(dentry))
+			goto invalid_exit;
+	} else {
+		gossip_debug(GOSSIP_DCACHE_DEBUG,
+			     "%s: root handle, lookup skipped.\n",
+			     __func__);
+	}
+
+	/* now perform getattr */
+	gossip_debug(GOSSIP_DCACHE_DEBUG,
+		     "%s: doing getattr: inode: %p, handle: %pU\n",
+		     __func__,
+		     inode,
+		     get_khandle_from_ino(inode));
+	ret = pvfs2_inode_getattr(inode, PVFS_ATTR_SYS_ALL_NOHINT);
+	gossip_debug(GOSSIP_DCACHE_DEBUG,
+		     "%s: getattr %s (ret = %d), returning %s for dentry i_count=%d\n",
+		     __func__,
+		     (ret == 0 ? "succeeded" : "failed"),
+		     ret,
+		     (ret == 0 ? "valid" : "INVALID"),
+		     atomic_read(&inode->i_count));
+	if (ret != 0)
+		goto invalid_exit;
+
+	/* dentry is valid! */
+	return 1;
+
+invalid_exit:
+	return 0;
+}
+
+const struct dentry_operations pvfs2_dentry_operations = {
+	.d_revalidate = pvfs2_d_revalidate,
+};

fs/orangefs/devpvfs2-req.c

+/*
+ * (C) 2001 Clemson University and The University of Chicago
+ *
+ * Changes by Acxiom Corporation to add protocol version to kernel
+ * communication, Copyright Acxiom Corporation, 2005.
+ *
+ * See COPYING in top-level directory.
+ */
+
+#include "protocol.h"
+#include "pvfs2-kernel.h"
+#include "pvfs2-dev-proto.h"
+#include "pvfs2-bufmap.h"
+
+#include <linux/debugfs.h>
+#include <linux/slab.h>
+
+/* this file implements the /dev/pvfs2-req device node */
+
+static int open_access_count;
+
+#define DUMP_DEVICE_ERROR()                                                   \
+do {                                                                          \
+	gossip_err("*****************************************************\n");\
+	gossip_err("PVFS2 Device Error:  You cannot open the device file ");  \
+	gossip_err("\n/dev/%s more than once.  Please make sure that\nthere " \
+		   "are no ", PVFS2_REQDEVICE_NAME);                          \
+	gossip_err("instances of a program using this device\ncurrently "     \
+		   "running. (You must verify this!)\n");                     \
+	gossip_err("For example, you can use the lsof program as follows:\n");\
+	gossip_err("'lsof | grep %s' (run this as root)\n",                   \
+		   PVFS2_REQDEVICE_NAME);                                     \
+	gossip_err("  open_access_count = %d\n", open_access_count);          \
+	gossip_err("*****************************************************\n");\
+} while (0)
+
+static int hash_func(__u64 tag, int table_size)
+{
+	return tag % ((unsigned int)table_size);
+}
+
+static void pvfs2_devreq_add_op(struct pvfs2_kernel_op_s *op)
+{
+	int index = hash_func(op->tag, hash_table_size);
+
+	spin_lock(&htable_ops_in_progress_lock);
+	list_add_tail(&op->list, &htable_ops_in_progress[index]);
+	spin_unlock(&htable_ops_in_progress_lock);
+}
+
+static struct pvfs2_kernel_op_s *pvfs2_devreq_remove_op(__u64 tag)
+{
+	struct pvfs2_kernel_op_s *op, *next;
+	int index;
+
+	index = hash_func(tag, hash_table_size);
+
+	spin_lock(&htable_ops_in_progress_lock);
+	list_for_each_entry_safe(op,
+				 next,
+				 &htable_ops_in_progress[index],
+				 list) {
+		if (op->tag == tag) {
+			list_del(&op->list);
+			spin_unlock(&htable_ops_in_progress_lock);
+			return op;
+		}
+	}
+
+	spin_unlock(&htable_ops_in_progress_lock);
+	return NULL;
+}
+
+static int pvfs2_devreq_open(struct inode *inode, struct file *file)
+{
+	int ret = -EINVAL;
+
+	if (!(file->f_flags & O_NONBLOCK)) {
+		gossip_err("pvfs2: device cannot be opened in blocking mode\n");
+		goto out;
+	}
+	ret = -EACCES;
+	gossip_debug(GOSSIP_DEV_DEBUG, "pvfs2-client-core: opening device\n");
+	mutex_lock(&devreq_mutex);
+
+	if (open_access_count == 0) {
+		ret = generic_file_open(inode, file);
+		if (ret == 0)
+			open_access_count++;
+	} else {
+		DUMP_DEVICE_ERROR();
+	}
+	mutex_unlock(&devreq_mutex);
+
+out:
+
+	gossip_debug(GOSSIP_DEV_DEBUG,
+		     "pvfs2-client-core: open device complete (ret = %d)\n",
+		     ret);
+	return ret;
+}
+
+static ssize_t pvfs2_devreq_read(struct file *file,
+				 char __user *buf,
+				 size_t count, loff_t *offset)
+{
+	int ret = 0;
+	ssize_t len = 0;
+	struct pvfs2_kernel_op_s *cur_op = NULL;
+	static __s32 magic = PVFS2_DEVREQ_MAGIC;
+	__s32 proto_ver = PVFS_KERNEL_PROTO_VERSION;
+
+	if (!(file->f_flags & O_NONBLOCK)) {
+		/* We do not support blocking reads/opens any more */
+		gossip_err("pvfs2: blocking reads are not supported! (pvfs2-client-core bug)\n");
+		return -EINVAL;
+	} else {
+		struct pvfs2_kernel_op_s *op = NULL, *temp = NULL;
+		/* get next op (if any) from top of list */
+		spin_lock(&pvfs2_request_list_lock);
+		list_for_each_entry_safe(op, temp, &pvfs2_request_list, list) {
+			__s32 fsid = fsid_of_op(op);
+			/*
+			 * Check if this op's fsid is known and needs
+			 * remounting
+			 */
+			if (fsid != PVFS_FS_ID_NULL &&
+			    fs_mount_pending(fsid) == 1) {
+				gossip_debug(GOSSIP_DEV_DEBUG,
+					     "Skipping op tag %llu %s\n",
+					     llu(op->tag),
+					     get_opname_string(op));
+				continue;
+			} else {
+				/*
+				 * op does not belong to any particular fsid
+				 * or already mounted.. let it through
+				 */
+				cur_op = op;
+				spin_lock(&cur_op->lock);
+				list_del(&cur_op->list);
+				cur_op->op_linger_tmp--;
+				/*
+				 * if there is a trailer, re-add it to
+				 * the request list.
+				 */
+				if (cur_op->op_linger == 2 &&
+				    cur_op->op_linger_tmp == 1) {
+					if (cur_op->upcall.trailer_size <= 0 ||
+					    cur_op->upcall.trailer_buf == NULL)
+						gossip_err("BUG:trailer_size is %ld and trailer buf is %p\n", (long)cur_op->upcall.trailer_size, cur_op->upcall.trailer_buf);
+					/* re-add it to the head of the list */
+					list_add(&cur_op->list,
+						 &pvfs2_request_list);
+				}
+				spin_unlock(&cur_op->lock);
+				break;
+			}
+		}
+		spin_unlock(&pvfs2_request_list_lock);
+	}
+
+	if (cur_op) {
+		spin_lock(&cur_op->lock);
+
+		gossip_debug(GOSSIP_DEV_DEBUG,
+			     "client-core: reading op tag %llu %s\n",
+			     llu(cur_op->tag), get_opname_string(cur_op));
+		if (op_state_in_progress(cur_op) || op_state_serviced(cur_op)) {
+			if (cur_op->op_linger == 1)
+				gossip_err("WARNING: Current op already queued...skipping\n");
+		} else if (cur_op->op_linger == 1 ||
+			   (cur_op->op_linger == 2 &&
+			    cur_op->op_linger_tmp == 0)) {
+			/*
+			 * atomically move the operation to the
+			 * htable_ops_in_progress
+			 */
+			set_op_state_inprogress(cur_op);
+			pvfs2_devreq_add_op(cur_op);
+		}
+
+		spin_unlock(&cur_op->lock);
+
+		/* 2 cases
+		 * a) OPs with no trailers
+		 * b) OPs with trailers, Stage 1
+		 * Either way push the upcall out
+		 */
+		if (cur_op->op_linger == 1 ||
+		   (cur_op->op_linger == 2 && cur_op->op_linger_tmp == 1)) {
+			len = MAX_ALIGNED_DEV_REQ_UPSIZE;
+			if ((size_t) len <= count) {
+			    ret = copy_to_user(buf,
+					       &proto_ver,
+					       sizeof(__s32));
+			    if (ret == 0) {
+				ret = copy_to_user(buf + sizeof(__s32),
+						   &magic,
+						   sizeof(__s32));
+				if (ret == 0) {
+				    ret = copy_to_user(buf+2 * sizeof(__s32),
+						       &cur_op->tag,
+						       sizeof(__u64));
+				    if (ret == 0) {
+					ret = copy_to_user(
+						buf +
+						  2 *
+						  sizeof(__s32) +
+						  sizeof(__u64),
+						&cur_op->upcall,
+						sizeof(struct pvfs2_upcall_s));
+				    }
+				}
+			    }
+
+			    if (ret) {
+				gossip_err("Failed to copy data to user space\n");
+				len = -EFAULT;
+			    }
+			} else {
+				gossip_err
+				    ("Failed to copy data to user space\n");
+				len = -EIO;
+			}
+		}
+		/* Stage 2: Push the trailer out */
+		else if (cur_op->op_linger == 2 && cur_op->op_linger_tmp == 0) {
+			len = cur_op->upcall.trailer_size;
+			if ((size_t) len <= count) {
+				ret = copy_to_user(buf,
+						   cur_op->upcall.trailer_buf,
+						   len);
+				if (ret) {
+					gossip_err("Failed to copy trailer to user space\n");
+					len = -EFAULT;
+				}
+			} else {
+				gossip_err("Read buffer for trailer is too small (%ld as opposed to %ld)\n",
+					(long)count,
+					(long)len);
+				len = -EIO;
+			}
+		} else {
+			gossip_err("cur_op: %p (op_linger %d), (op_linger_tmp %d), erroneous request list?\n",
+				cur_op,
+				cur_op->op_linger,
+				cur_op->op_linger_tmp);
+			len = 0;
+		}
+	} else if (file->f_flags & O_NONBLOCK) {
+		/*
+		 * if in non-blocking mode, return EAGAIN since no requests are
+		 * ready yet
+		 */
+		len = -EAGAIN;
+	}
+	return len;
+}
+
+/* Function for writev() callers into the device */
+static ssize_t pvfs2_devreq_writev(struct file *file,
+				   const struct iovec *iov,
+				   size_t count,
+				   loff_t *offset)
+{
+	struct pvfs2_kernel_op_s *op = NULL;
+	void *buffer = NULL;
+	void *ptr = NULL;
+	unsigned long i = 0;
+	static int max_downsize = MAX_ALIGNED_DEV_REQ_DOWNSIZE;
+	int ret = 0, num_remaining = max_downsize;
+	int notrailer_count = 4; /* num elements in iovec without trailer */
+	int payload_size = 0;
+	__s32 magic = 0;
+	__s32 proto_ver = 0;
+	__u64 tag = 0;
+	ssize_t total_returned_size = 0;
+
+	/* Either there is a trailer or there isn't */
+	if (count != notrailer_count && count != (notrailer_count + 1)) {
+		gossip_err("Error: Number of iov vectors is (%ld) and notrailer count is %d\n",
+			count,
+			notrailer_count);
+		return -EPROTO;
+	}
+	buffer = dev_req_alloc();
+	if (!buffer)
+		return -ENOMEM;
+	ptr = buffer;
+
+	for (i = 0; i < notrailer_count; i++) {
+		if (iov[i].iov_len > num_remaining) {
+			gossip_err
+			    ("writev error: Freeing buffer and returning\n");
+			dev_req_release(buffer);
+			return -EMSGSIZE;
+		}
+		ret = copy_from_user(ptr, iov[i].iov_base, iov[i].iov_len);
+		if (ret) {
+			gossip_err("Failed to copy data from user space\n");
+			dev_req_release(buffer);
+			return -EIO;
+		}
+		num_remaining -= iov[i].iov_len;
+		ptr += iov[i].iov_len;
+		payload_size += iov[i].iov_len;
+	}
+	total_returned_size = payload_size;
+
+	/* these elements are currently 8 byte aligned (8 bytes for (version +
+	 * magic) 8 bytes for tag).  If you add another element, either
+	 * make it 8 bytes big, or use get_unaligned when asigning.
+	 */
+	ptr = buffer;
+	proto_ver = *((__s32 *) ptr);
+	ptr += sizeof(__s32);
+
+	magic = *((__s32 *) ptr);
+	ptr += sizeof(__s32);
+
+	tag = *((__u64 *) ptr);
+	ptr += sizeof(__u64);
+
+	if (magic != PVFS2_DEVREQ_MAGIC) {
+		gossip_err("Error: Device magic number does not match.\n");
+		dev_req_release(buffer);
+		return -EPROTO;
+	}
+
+	/*
+	 * proto_ver = 20902 for 2.9.2
+	 */
+
+	op = pvfs2_devreq_remove_op(tag);
+	if (op) {
+		/* Increase ref count! */
+		get_op(op);
+		/* cut off magic and tag from payload size */
+		payload_size -= (2 * sizeof(__s32) + sizeof(__u64));
+		if (payload_size <= sizeof(struct pvfs2_downcall_s))
+			/* copy the passed in downcall into the op */
+			memcpy(&op->downcall,
+			       ptr,
+			       sizeof(struct pvfs2_downcall_s));
+		else
+			gossip_debug(GOSSIP_DEV_DEBUG,
+				     "writev: Ignoring %d bytes\n",
+				     payload_size);
+
+		/* Do not allocate needlessly if client-core forgets
+		 * to reset trailer size on op errors.
+		 */
+		if (op->downcall.status == 0 && op->downcall.trailer_size > 0) {
+			gossip_debug(GOSSIP_DEV_DEBUG,
+				     "writev: trailer size %ld\n",
+				     (unsigned long)op->downcall.trailer_size);
+			if (count != (notrailer_count + 1)) {
+				gossip_err("Error: trailer size (%ld) is non-zero, no trailer elements though? (%ld)\n", (unsigned long)op->downcall.trailer_size, count);
+				dev_req_release(buffer);
+				put_op(op);
+				return -EPROTO;
+			}
+			if (iov[notrailer_count].iov_len >
+			    op->downcall.trailer_size) {
+				gossip_err("writev error: trailer size (%ld) != iov_len (%ld)\n", (unsigned long)op->downcall.trailer_size, (unsigned long)iov[notrailer_count].iov_len);
+				dev_req_release(buffer);
+				put_op(op);
+				return -EMSGSIZE;
+			}
+			/* Allocate a buffer large enough to hold the
+			 * trailer bytes.
+			 */
+			op->downcall.trailer_buf =
+			    vmalloc(op->downcall.trailer_size);
+			if (op->downcall.trailer_buf != NULL) {
+				gossip_debug(GOSSIP_DEV_DEBUG, "vmalloc: %p\n",
+					     op->downcall.trailer_buf);
+				ret = copy_from_user(op->downcall.trailer_buf,
+						     iov[notrailer_count].
+						     iov_base,
+						     iov[notrailer_count].
+						     iov_len);
+				if (ret) {
+					gossip_err("Failed to copy trailer data from user space\n");
+					dev_req_release(buffer);
+					gossip_debug(GOSSIP_DEV_DEBUG,
+						     "vfree: %p\n",
+						     op->downcall.trailer_buf);
+					vfree(op->downcall.trailer_buf);
+					op->downcall.trailer_buf = NULL;
+					put_op(op);
+					return -EIO;
+				}
+			} else {
+				/* Change downcall status */
+				op->downcall.status = -ENOMEM;
+				gossip_err("writev: could not vmalloc for trailer!\n");
+			}
+		}
+
+		/* if this operation is an I/O operation and if it was
+		 * initiated on behalf of a *synchronous* VFS I/O operation,
+		 * only then we need to wait
+		 * for all data to be copied before we can return to avoid
+		 * buffer corruption and races that can pull the buffers
+		 * out from under us.
+		 *
+		 * Essentially we're synchronizing with other parts of the
+		 * vfs implicitly by not allowing the user space
+		 * application reading/writing this device to return until
+		 * the buffers are done being used.
+		 */
+		if ((op->upcall.type == PVFS2_VFS_OP_FILE_IO &&
+		     op->upcall.req.io.async_vfs_io == PVFS_VFS_SYNC_IO) ||
+		     op->upcall.type == PVFS2_VFS_OP_FILE_IOX) {
+			int timed_out = 0;
+			DECLARE_WAITQUEUE(wait_entry, current);
+
+			/* tell the vfs op waiting on a waitqueue
+			 * that this op is done
+			 */
+			spin_lock(&op->lock);
+			set_op_state_serviced(op);
+			spin_unlock(&op->lock);
+
+			add_wait_queue_exclusive(&op->io_completion_waitq,
+						 &wait_entry);
+			wake_up_interruptible(&op->waitq);
+
+			while (1) {
+				set_current_state(TASK_INTERRUPTIBLE);
+
+				spin_lock(&op->lock);
+				if (op->io_completed) {
+					spin_unlock(&op->lock);
+					break;
+				}
+				spin_unlock(&op->lock);
+
+				if (!signal_pending(current)) {
+					int timeout =
+					    MSECS_TO_JIFFIES(1000 *
+							     op_timeout_secs);
+					if (!schedule_timeout(timeout)) {
+						gossip_debug(GOSSIP_DEV_DEBUG, "*** I/O wait time is up\n");
+						timed_out = 1;
+						break;
+					}
+					continue;
+				}
+
+				gossip_debug(GOSSIP_DEV_DEBUG, "*** signal on I/O wait -- aborting\n");
+				break;
+			}
+
+			set_current_state(TASK_RUNNING);
+			remove_wait_queue(&op->io_completion_waitq,
+					  &wait_entry);
+
+			/* NOTE: for I/O operations we handle releasing the op
+			 * object except in the case of timeout.  the reason we
+			 * can't free the op in timeout cases is that the op
+			 * service logic in the vfs retries operations using
+			 * the same op ptr, thus it can't be freed.
+			 */
+			if (!timed_out)
+				op_release(op);
+		} else {
+
+			/*
+			 * tell the vfs op waiting on a waitqueue that
+			 * this op is done
+			 */
+			spin_lock(&op->lock);
+			set_op_state_serviced(op);
+			spin_unlock(&op->lock);
+			/*
+			   for every other operation (i.e. non-I/O), we need to
+			   wake up the callers for downcall completion
+			   notification
+			 */
+			wake_up_interruptible(&op->waitq);
+		}
+	} else {
+		/* ignore downcalls that we're not interested in */
+		gossip_debug(GOSSIP_DEV_DEBUG,
+			     "WARNING: No one's waiting for tag %llu\n",
+			     llu(tag));
+	}
+	dev_req_release(buffer);
+
+	return total_returned_size;
+}
+
+static ssize_t pvfs2_devreq_write_iter(struct kiocb *iocb,
+				      struct iov_iter *iter)
+{
+	return pvfs2_devreq_writev(iocb->ki_filp,
+				   iter->iov,
+				   iter->nr_segs,
+				   &iocb->ki_pos);
+}
+
+/* Returns whether any FS are still pending remounted */
+static int mark_all_pending_mounts(void)
+{
+	int unmounted = 1;
+	struct pvfs2_sb_info_s *pvfs2_sb = NULL;
+
+	spin_lock(&pvfs2_superblocks_lock);
+	list_for_each_entry(pvfs2_sb, &pvfs2_superblocks, list) {
+		/* All of these file system require a remount */
+		pvfs2_sb->mount_pending = 1;
+		unmounted = 0;
+	}
+	spin_unlock(&pvfs2_superblocks_lock);
+	return unmounted;
+}
+
+/*
+ * Determine if a given file system needs to be remounted or not
+ *  Returns -1 on error
+ *           0 if already mounted
+ *           1 if needs remount
+ */
+int fs_mount_pending(__s32 fsid)
+{
+	int mount_pending = -1;
+	struct pvfs2_sb_info_s *pvfs2_sb = NULL;
+
+	spin_lock(&pvfs2_superblocks_lock);
+	list_for_each_entry(pvfs2_sb, &pvfs2_superblocks, list) {
+		if (pvfs2_sb->fs_id == fsid) {
+			mount_pending = pvfs2_sb->mount_pending;
+			break;
+		}
+	}
+	spin_unlock(&pvfs2_superblocks_lock);
+	return mount_pending;
+}
+
+/*
+ * NOTE: gets called when the last reference to this device is dropped.
+ * Using the open_access_count variable, we enforce a reference count
+ * on this file so that it can be opened by only one process at a time.
+ * the devreq_mutex is used to make sure all i/o has completed
+ * before we call pvfs_bufmap_finalize, and similar such tricky
+ * situations
+ */
+static int pvfs2_devreq_release(struct inode *inode, struct file *file)
+{
+	int unmounted = 0;
+
+	gossip_debug(GOSSIP_DEV_DEBUG,
+		     "%s:pvfs2-client-core: exiting, closing device\n",
+		     __func__);
+
+	mutex_lock(&devreq_mutex);
+	pvfs_bufmap_finalize();
+
+	open_access_count--;
+
+	unmounted = mark_all_pending_mounts();
+	gossip_debug(GOSSIP_DEV_DEBUG, "PVFS2 Device Close: Filesystem(s) %s\n",
+		     (unmounted ? "UNMOUNTED" : "MOUNTED"));
+	mutex_unlock(&devreq_mutex);
+
+	/*
+	 * Walk through the list of ops in the request list, mark them
+	 * as purged and wake them up.
+	 */
+	purge_waiting_ops();
+	/*
+	 * Walk through the hash table of in progress operations; mark
+	 * them as purged and wake them up
+	 */
+	purge_inprogress_ops();
+	gossip_debug(GOSSIP_DEV_DEBUG,
+		     "pvfs2-client-core: device close complete\n");
+	return 0;
+}
+
+int is_daemon_in_service(void)
+{
+	int in_service;
+
+	/*
+	 * What this function does is checks if client-core is alive
+	 * based on the access count we maintain on the device.
+	 */
+	mutex_lock(&devreq_mutex);
+	in_service = open_access_count == 1 ? 0 : -EIO;
+	mutex_unlock(&devreq_mutex);
+	return in_service;
+}
+
+static inline long check_ioctl_command(unsigned int command)
+{
+	/* Check for valid ioctl codes */
+	if (_IOC_TYPE(command) != PVFS_DEV_MAGIC) {
+		gossip_err("device ioctl magic numbers don't match! Did you rebuild pvfs2-client-core/libpvfs2? [cmd %x, magic %x != %x]\n",
+			command,
+			_IOC_TYPE(command),
+			PVFS_DEV_MAGIC);
+		return -EINVAL;
+	}
+	/* and valid ioctl commands */
+	if (_IOC_NR(command) >= PVFS_DEV_MAXNR || _IOC_NR(command) <= 0) {
+		gossip_err("Invalid ioctl command number [%d >= %d]\n",
+			   _IOC_NR(command), PVFS_DEV_MAXNR);
+		return -ENOIOCTLCMD;
+	}
+	return 0;
+}
+
+static long dispatch_ioctl_command(unsigned int command, unsigned long arg)
+{
+	static __s32 magic = PVFS2_DEVREQ_MAGIC;
+	static __s32 max_up_size = MAX_ALIGNED_DEV_REQ_UPSIZE;
+	static __s32 max_down_size = MAX_ALIGNED_DEV_REQ_DOWNSIZE;
+	struct PVFS_dev_map_desc user_desc;
+	int ret = 0;
+	struct dev_mask_info_s mask_info = { 0 };
+	struct dev_mask2_info_s mask2_info = { 0, 0 };
+	int upstream_kmod = 1;
+	struct list_head *tmp = NULL;
+	struct pvfs2_sb_info_s *pvfs2_sb = NULL;
+
+	/* mtmoore: add locking here */
+
+	switch (command) {
+	case PVFS_DEV_GET_MAGIC:
+		return ((put_user(magic, (__s32 __user *) arg) == -EFAULT) ?
+			-EIO :
+			0);
+	case PVFS_DEV_GET_MAX_UPSIZE:
+		return ((put_user(max_up_size,
+				  (__s32 __user *) arg) == -EFAULT) ?
+					-EIO :
+					0);
+	case PVFS_DEV_GET_MAX_DOWNSIZE:
+		return ((put_user(max_down_size,
+				  (__s32 __user *) arg) == -EFAULT) ?
+					-EIO :
+					0);
+	case PVFS_DEV_MAP:
+		ret = copy_from_user(&user_desc,
+				     (struct PVFS_dev_map_desc __user *)
+				     arg,
+				     sizeof(struct PVFS_dev_map_desc));
+		return ret ? -EIO : pvfs_bufmap_initialize(&user_desc);
+	case PVFS_DEV_REMOUNT_ALL:
+		gossip_debug(GOSSIP_DEV_DEBUG,
+			     "pvfs2_devreq_ioctl: got PVFS_DEV_REMOUNT_ALL\n");
+
+		/*
+		 * remount all mounted pvfs2 volumes to regain the lost
+		 * dynamic mount tables (if any) -- NOTE: this is done
+		 * without keeping the superblock list locked due to the
+		 * upcall/downcall waiting.  also, the request semaphore is
+		 * used to ensure that no operations will be serviced until
+		 * all of the remounts are serviced (to avoid ops between
+		 * mounts to fail)
+		 */
+		ret = mutex_lock_interruptible(&request_mutex);
+		if (ret < 0)
+			return ret;
+		gossip_debug(GOSSIP_DEV_DEBUG,
+			     "pvfs2_devreq_ioctl: priority remount in progress\n");
+		list_for_each(tmp, &pvfs2_superblocks) {
+			pvfs2_sb =
+				list_entry(tmp, struct pvfs2_sb_info_s, list);
+			if (pvfs2_sb && (pvfs2_sb->sb)) {
+				gossip_debug(GOSSIP_DEV_DEBUG,
+					     "Remounting SB %p\n",
+					     pvfs2_sb);
+
+				ret = pvfs2_remount(pvfs2_sb->sb);
+				if (ret) {
+					gossip_debug(GOSSIP_DEV_DEBUG,
+						     "SB %p remount failed\n",
+						     pvfs2_sb);
+						break;
+				}
+			}
+		}
+		gossip_debug(GOSSIP_DEV_DEBUG,
+			     "pvfs2_devreq_ioctl: priority remount complete\n");
+		mutex_unlock(&request_mutex);
+		return ret;
+
+	case PVFS_DEV_UPSTREAM:
+		ret = copy_to_user((void __user *)arg,
+				    &upstream_kmod,
+				    sizeof(upstream_kmod));
+
+		if (ret != 0)
+			return -EIO;
+		else
+			return ret;
+
+	case PVFS_DEV_CLIENT_MASK:
+		ret = copy_from_user(&mask2_info,
+				     (void __user *)arg,
+				     sizeof(struct dev_mask2_info_s));
+
+		if (ret != 0)
+			return -EIO;
+
+		client_debug_mask.mask1 = mask2_info.mask1_value;
+		client_debug_mask.mask2 = mask2_info.mask2_value;
+
+		pr_info("%s: client debug mask has been been received "
+			":%llx: :%llx:\n",
+			__func__,
+			(unsigned long long)client_debug_mask.mask1,
+			(unsigned long long)client_debug_mask.mask2);
+
+		return ret;
+
+	case PVFS_DEV_CLIENT_STRING:
+		ret = copy_from_user(&client_debug_array_string,
+				     (void __user *)arg,
+				     PVFS2_MAX_DEBUG_STRING_LEN);
+		if (ret != 0) {
+			pr_info("%s: "
+				"PVFS_DEV_CLIENT_STRING: copy_from_user failed"
+				"\n",
+				__func__);
+			return -EIO;
+		}
+
+		pr_info("%s: client debug array string has been been received."
+			"\n",
+			__func__);
+
+		if (!help_string_initialized) {
+
+			/* Free the "we don't know yet" default string... */
+			kfree(debug_help_string);
+
+			/* build a proper debug help string */
+			if (orangefs_prepare_debugfs_help_string(0)) {
+				gossip_err("%s: "
+					   "prepare_debugfs_help_string failed"
+					   "\n",
+					   __func__);
+				return -EIO;
+			}
+
+			/* Replace the boilerplate boot-time debug-help file. */
+			debugfs_remove(help_file_dentry);
+
+			help_file_dentry =
+				debugfs_create_file(
+					ORANGEFS_KMOD_DEBUG_HELP_FILE,
+					0444,
+					debug_dir,
+					debug_help_string,
+					&debug_help_fops);
+
+			if (!help_file_dentry) {
+				gossip_err("%s: debugfs_create_file failed for"
+					   " :%s:!\n",
+					   __func__,
+					   ORANGEFS_KMOD_DEBUG_HELP_FILE);
+				return -EIO;
+			}
+		}
+
+		debug_mask_to_string(&client_debug_mask, 1);
+
+		debugfs_remove(client_debug_dentry);
+
+		pvfs2_client_debug_init();
+
+		help_string_initialized++;
+
+		return ret;
+
+	case PVFS_DEV_DEBUG:
+		ret = copy_from_user(&mask_info,
+				     (void __user *)arg,
+				     sizeof(mask_info));
+
+		if (ret != 0)
+			return -EIO;
+
+		if (mask_info.mask_type == KERNEL_MASK) {
+			if ((mask_info.mask_value == 0)
+			    && (kernel_mask_set_mod_init)) {
+				/*
+				 * the kernel debug mask was set when the
+				 * kernel module was loaded; don't override
+				 * it if the client-core was started without
+				 * a value for PVFS2_KMODMASK.
+				 */
+				return 0;
+			}
+			debug_mask_to_string(&mask_info.mask_value,
+					     mask_info.mask_type);
+			gossip_debug_mask = mask_info.mask_value;
+			pr_info("PVFS: kernel debug mask has been modified to "
+				":%s: :%llx:\n",
+				kernel_debug_string,
+				(unsigned long long)gossip_debug_mask);
+		} else if (mask_info.mask_type == CLIENT_MASK) {
+			debug_mask_to_string(&mask_info.mask_value,
+					     mask_info.mask_type);
+			pr_info("PVFS: client debug mask has been modified to"
+				":%s: :%llx:\n",
+				client_debug_string,
+				llu(mask_info.mask_value));
+		} else {
+			gossip_lerr("Invalid mask type....\n");
+			return -EINVAL;
+		}
+
+		return ret;
+
+	default:
+		return -ENOIOCTLCMD;
+	}
+	return -ENOIOCTLCMD;
+}
+
+static long pvfs2_devreq_ioctl(struct file *file,
+			       unsigned int command, unsigned long arg)
+{
+	long ret;
+
+	/* Check for properly constructed commands */
+	ret = check_ioctl_command(command);
+	if (ret < 0)
+		return (int)ret;
+
+	return (int)dispatch_ioctl_command(command, arg);
+}
+
+#ifdef CONFIG_COMPAT		/* CONFIG_COMPAT is in .config */
+
+/*  Compat structure for the PVFS_DEV_MAP ioctl */
+struct PVFS_dev_map_desc32 {
+	compat_uptr_t ptr;
+	__s32 total_size;
+	__s32 size;
+	__s32 count;
+};
+
+static unsigned long translate_dev_map26(unsigned long args, long *error)
+{
+	struct PVFS_dev_map_desc32 __user *p32 = (void __user *)args;
+	/*
+	 * Depending on the architecture, allocate some space on the
+	 * user-call-stack based on our expected layout.
+	 */
+	struct PVFS_dev_map_desc __user *p =
+	    compat_alloc_user_space(sizeof(*p));
+	u32 addr;
+
+	*error = 0;
+	/* get the ptr from the 32 bit user-space */
+	if (get_user(addr, &p32->ptr))
+		goto err;
+	/* try to put that into a 64-bit layout */
+	if (put_user(compat_ptr(addr), &p->ptr))
+		goto err;
+	/* copy the remaining fields */
+	if (copy_in_user(&p->total_size, &p32->total_size, sizeof(__s32)))
+		goto err;
+	if (copy_in_user(&p->size, &p32->size, sizeof(__s32)))
+		goto err;
+	if (copy_in_user(&p->count, &p32->count, sizeof(__s32)))
+		goto err;
+	return (unsigned long)p;
+err:
+	*error = -EFAULT;
+	return 0;
+}
+
+/*
+ * 32 bit user-space apps' ioctl handlers when kernel modules
+ * is compiled as a 64 bit one
+ */
+static long pvfs2_devreq_compat_ioctl(struct file *filp, unsigned int cmd,
+				      unsigned long args)
+{
+	long ret;
+	unsigned long arg = args;
+
+	/* Check for properly constructed commands */
+	ret = check_ioctl_command(cmd);
+	if (ret < 0)
+		return ret;
+	if (cmd == PVFS_DEV_MAP) {
+		/*
+		 * convert the arguments to what we expect internally
+		 * in kernel space
+		 */
+		arg = translate_dev_map26(args, &ret);
+		if (ret < 0) {
+			gossip_err("Could not translate dev map\n");
+			return ret;
+		}
+	}
+	/* no other ioctl requires translation */
+	return dispatch_ioctl_command(cmd, arg);
+}
+
+static int pvfs2_ioctl32_init(void)
+{
+	return 0;
+}
+
+static void pvfs2_ioctl32_cleanup(void)
+{
+	return;
+}
+
+#endif /* CONFIG_COMPAT is in .config */
+
+/* the assigned character device major number */
+static int pvfs2_dev_major;
+
+/*
+ * Initialize pvfs2 device specific state:
+ * Must be called at module load time only
+ */
+int pvfs2_dev_init(void)
+{
+	int ret;
+
+	/* register the ioctl32 sub-system */
+	ret = pvfs2_ioctl32_init();
+	if (ret < 0)
+		return ret;
+
+	/* register pvfs2-req device  */
+	pvfs2_dev_major = register_chrdev(0,
+					  PVFS2_REQDEVICE_NAME,
+					  &pvfs2_devreq_file_operations);
+	if (pvfs2_dev_major < 0) {
+		gossip_debug(GOSSIP_DEV_DEBUG,
+			     "Failed to register /dev/%s (error %d)\n",
+			     PVFS2_REQDEVICE_NAME, pvfs2_dev_major);
+		pvfs2_ioctl32_cleanup();
+		return pvfs2_dev_major;
+	}
+
+	gossip_debug(GOSSIP_DEV_DEBUG,
+		     "*** /dev/%s character device registered ***\n",
+		     PVFS2_REQDEVICE_NAME);
+	gossip_debug(GOSSIP_DEV_DEBUG, "'mknod /dev/%s c %d 0'.\n",
+		     PVFS2_REQDEVICE_NAME, pvfs2_dev_major);
+	return 0;
+}
+
+void pvfs2_dev_cleanup(void)
+{
+	unregister_chrdev(pvfs2_dev_major, PVFS2_REQDEVICE_NAME);
+	gossip_debug(GOSSIP_DEV_DEBUG,
+		     "*** /dev/%s character device unregistered ***\n",
+		     PVFS2_REQDEVICE_NAME);
+	/* unregister the ioctl32 sub-system */
+	pvfs2_ioctl32_cleanup();
+}
+
+static unsigned int pvfs2_devreq_poll(struct file *file,
+				      struct poll_table_struct *poll_table)
+{
+	int poll_revent_mask = 0;
+
+	if (open_access_count == 1) {
+		poll_wait(file, &pvfs2_request_list_waitq, poll_table);
+
+		spin_lock(&pvfs2_request_list_lock);
+		if (!list_empty(&pvfs2_request_list))
+			poll_revent_mask |= POLL_IN;
+		spin_unlock(&pvfs2_request_list_lock);
+	}
+	return poll_revent_mask;
+}
+
+const struct file_operations pvfs2_devreq_file_operations = {
+	.owner = THIS_MODULE,
+	.read = pvfs2_devreq_read,
+	.write_iter = pvfs2_devreq_write_iter,
+	.open = pvfs2_devreq_open,
+	.release = pvfs2_devreq_release,
+	.unlocked_ioctl = pvfs2_devreq_ioctl,
+
+#ifdef CONFIG_COMPAT		/* CONFIG_COMPAT is in .config */
+	.compat_ioctl = pvfs2_devreq_compat_ioctl,
+#endif
+	.poll = pvfs2_devreq_poll
+};

fs/orangefs/dir.c

+/*
+ * (C) 2001 Clemson University and The University of Chicago
+ *
+ * See COPYING in top-level directory.
+ */
+
+#include "protocol.h"
+#include "pvfs2-kernel.h"
+#include "pvfs2-bufmap.h"
+
+struct readdir_handle_s {
+	int buffer_index;
+	struct pvfs2_readdir_response_s readdir_response;
+	void *dents_buf;
+};
+
+/*
+ * decode routine needed by kmod to make sense of the shared page for readdirs.
+ */
+static long decode_dirents(char *ptr, struct pvfs2_readdir_response_s *readdir)
+{
+	int i;
+	struct pvfs2_readdir_response_s *rd =
+		(struct pvfs2_readdir_response_s *) ptr;
+	char *buf = ptr;
+	char **pptr = &buf;
+
+	readdir->token = rd->token;
+	readdir->pvfs_dirent_outcount = rd->pvfs_dirent_outcount;
+	readdir->dirent_array = kmalloc(readdir->pvfs_dirent_outcount *
+					sizeof(*readdir->dirent_array),
+					GFP_KERNEL);
+	if (readdir->dirent_array == NULL)
+		return -ENOMEM;
+	*pptr += offsetof(struct pvfs2_readdir_response_s, dirent_array);
+	for (i = 0; i < readdir->pvfs_dirent_outcount; i++) {
+		dec_string(pptr, &readdir->dirent_array[i].d_name,
+			   &readdir->dirent_array[i].d_length);
+		readdir->dirent_array[i].khandle =
+			*(struct pvfs2_khandle *) *pptr;
+		*pptr += 16;
+	}
+	return (unsigned long)*pptr - (unsigned long)ptr;
+}
+
+static long readdir_handle_ctor(struct readdir_handle_s *rhandle, void *buf,
+				int buffer_index)
+{
+	long ret;
+
+	if (buf == NULL) {
+		gossip_err
+		    ("Invalid NULL buffer specified in readdir_handle_ctor\n");
+		return -ENOMEM;
+	}
+	if (buffer_index < 0) {
+		gossip_err
+		    ("Invalid buffer index specified in readdir_handle_ctor\n");
+		return -EINVAL;
+	}
+	rhandle->buffer_index = buffer_index;
+	rhandle->dents_buf = buf;
+	ret = decode_dirents(buf, &rhandle->readdir_response);
+	if (ret < 0) {
+		gossip_err("Could not decode readdir from buffer %ld\n", ret);
+		rhandle->buffer_index = -1;
+		gossip_debug(GOSSIP_DIR_DEBUG, "vfree %p\n", buf);
+		vfree(buf);
+		rhandle->dents_buf = NULL;
+	}
+	return ret;
+}
+
+static void readdir_handle_dtor(struct pvfs2_bufmap *bufmap,
+		struct readdir_handle_s *rhandle)
+{
+	if (rhandle == NULL)
+		return;
+
+	/* kfree(NULL) is safe */
+	kfree(rhandle->readdir_response.dirent_array);
+	rhandle->readdir_response.dirent_array = NULL;
+
+	if (rhandle->buffer_index >= 0) {
+		readdir_index_put(bufmap, rhandle->buffer_index);
+		rhandle->buffer_index = -1;
+	}
+	if (rhandle->dents_buf) {
+		gossip_debug(GOSSIP_DIR_DEBUG, "vfree %p\n",
+			     rhandle->dents_buf);
+		vfree(rhandle->dents_buf);
+		rhandle->dents_buf = NULL;
+	}
+}
+
+/*
+ * Read directory entries from an instance of an open directory.
+ *
+ * \note This routine was converted for the readdir to iterate change
+ *       in "struct file_operations". "converted" mostly amounts to
+ *       changing occurrences of "readdir" and "filldir" in the
+ *       comments to "iterate" and "dir_emit". Also filldir calls
+ *       were changed to dir_emit calls.
+ *
+ * \param dir_emit callback function called for each entry read.
+ *
+ * \retval <0 on error
+ * \retval 0  when directory has been completely traversed
+ * \retval >0 if we don't call dir_emit for all entries
+ *
+ * \note If the dir_emit call-back returns non-zero, then iterate should
+ *       assume that it has had enough, and should return as well.
+ */
+static int pvfs2_readdir(struct file *file, struct dir_context *ctx)
+{
+	struct pvfs2_bufmap *bufmap = NULL;
+	int ret = 0;
+	int buffer_index;
+	__u64 *ptoken = file->private_data;
+	__u64 pos = 0;
+	ino_t ino = 0;
+	struct dentry *dentry = file->f_path.dentry;
+	struct pvfs2_kernel_op_s *new_op = NULL;
+	struct pvfs2_inode_s *pvfs2_inode = PVFS2_I(dentry->d_inode);
+	int buffer_full = 0;
+	struct readdir_handle_s rhandle;
+	int i = 0;
+	int len = 0;
+	ino_t current_ino = 0;
+	char *current_entry = NULL;
+	long bytes_decoded;
+
+	gossip_ldebug(GOSSIP_DIR_DEBUG,
+		      "%s: ctx->pos:%lld, token = %llu\n",
+		      __func__,
+		      lld(ctx->pos),
+		      llu(*ptoken));
+
+	pos = (__u64) ctx->pos;
+
+	/* are we done? */
+	if (pos == PVFS_READDIR_END) {
+		gossip_debug(GOSSIP_DIR_DEBUG,
+			     "Skipping to termination path\n");
+		return 0;
+	}
+
+	gossip_debug(GOSSIP_DIR_DEBUG,
+		     "pvfs2_readdir called on %s (pos=%llu)\n",
+		     dentry->d_name.name, llu(pos));
+
+	rhandle.buffer_index = -1;
+	rhandle.dents_buf = NULL;
+	memset(&rhandle.readdir_response, 0, sizeof(rhandle.readdir_response));
+
+	new_op = op_alloc(PVFS2_VFS_OP_READDIR);
+	if (!new_op)
+		return -ENOMEM;
+
+	new_op->uses_shared_memory = 1;
+	new_op->upcall.req.readdir.refn = pvfs2_inode->refn;
+	new_op->upcall.req.readdir.max_dirent_count = MAX_DIRENT_COUNT_READDIR;
+
+	gossip_debug(GOSSIP_DIR_DEBUG,
+		     "%s: upcall.req.readdir.refn.khandle: %pU\n",
+		     __func__,
+		     &new_op->upcall.req.readdir.refn.khandle);
+
+	/*
+	 * NOTE: the position we send to the readdir upcall is out of
+	 * sync with ctx->pos since:
+	 * 1. pvfs2 doesn't include the "." and ".." entries that are
+	 *    added below.
+	 * 2. the introduction of distributed directory logic makes token no
+	 *    longer be related to f_pos and pos. Instead an independent
+	 *    variable is used inside the function and stored in the
+	 *    private_data of the file structure.
+	 */
+	new_op->upcall.req.readdir.token = *ptoken;
+
+get_new_buffer_index:
+	ret = readdir_index_get(&bufmap, &buffer_index);
+	if (ret < 0) {
+		gossip_lerr("pvfs2_readdir: readdir_index_get() failure (%d)\n",
+			    ret);
+		goto out_free_op;
+	}
+	new_op->upcall.req.readdir.buf_index = buffer_index;
+
+	ret = service_operation(new_op,
+				"pvfs2_readdir",
+				get_interruptible_flag(dentry->d_inode));
+
+	gossip_debug(GOSSIP_DIR_DEBUG,
+		     "Readdir downcall status is %d.  ret:%d\n",
+		     new_op->downcall.status,
+		     ret);
+
+	if (ret == -EAGAIN && op_state_purged(new_op)) {
+		/*
+		 * readdir shared memory aread has been wiped due to
+		 * pvfs2-client-core restarting, so we must get a new
+		 * index into the shared memory.
+		 */
+		gossip_debug(GOSSIP_DIR_DEBUG,
+			"%s: Getting new buffer_index for retry of readdir..\n",
+			 __func__);
+		readdir_index_put(bufmap, buffer_index);
+		goto get_new_buffer_index;
+	}
+
+	if (ret == -EIO && op_state_purged(new_op)) {
+		gossip_err("%s: Client is down. Aborting readdir call.\n",
+			__func__);
+		readdir_index_put(bufmap, buffer_index);
+		goto out_free_op;
+	}
+
+	if (ret < 0 || new_op->downcall.status != 0) {
+		gossip_debug(GOSSIP_DIR_DEBUG,
+			     "Readdir request failed.  Status:%d\n",
+			     new_op->downcall.status);
+		readdir_index_put(bufmap, buffer_index);
+		if (ret >= 0)
+			ret = new_op->downcall.status;
+		goto out_free_op;
+	}
+
+	bytes_decoded =
+		readdir_handle_ctor(&rhandle,
+				    new_op->downcall.trailer_buf,
+				    buffer_index);
+	if (bytes_decoded < 0) {
+		gossip_err("pvfs2_readdir: Could not decode trailer buffer into a readdir response %d\n",
+			ret);
+		ret = bytes_decoded;
+		readdir_index_put(bufmap, buffer_index);
+		goto out_free_op;
+	}
+
+	if (bytes_decoded != new_op->downcall.trailer_size) {
+		gossip_err("pvfs2_readdir: # bytes decoded (%ld) != trailer size (%ld)\n",
+			bytes_decoded,
+			(long)new_op->downcall.trailer_size);
+		ret = -EINVAL;
+		goto out_destroy_handle;
+	}
+
+	if (pos == 0) {
+		ino = get_ino_from_khandle(dentry->d_inode);
+		gossip_debug(GOSSIP_DIR_DEBUG,
+			     "%s: calling dir_emit of \".\" with pos = %llu\n",
+			     __func__,
+			     llu(pos));
+		ret = dir_emit(ctx, ".", 1, ino, DT_DIR);
+		if (ret < 0)
+			goto out_destroy_handle;
+		ctx->pos++;
+		gossip_ldebug(GOSSIP_DIR_DEBUG,
+			      "%s: ctx->pos:%lld\n",
+			      __func__,
+			      lld(ctx->pos));
+		pos++;
+	}
+
+	if (pos == 1) {
+		ino = get_parent_ino_from_dentry(dentry);
+		gossip_debug(GOSSIP_DIR_DEBUG,
+			     "%s: calling dir_emit of \"..\" with pos = %llu\n",
+			     __func__,
+			     llu(pos));
+		ret = dir_emit(ctx, "..", 2, ino, DT_DIR);
+		if (ret < 0)
+			goto out_destroy_handle;
+		ctx->pos++;
+		gossip_ldebug(GOSSIP_DIR_DEBUG,
+			      "%s: ctx->pos:%lld\n",
+			      __func__,
+			      lld(ctx->pos));
+		pos++;
+	}
+
+	for (i = 0; i < rhandle.readdir_response.pvfs_dirent_outcount; i++) {
+		len = rhandle.readdir_response.dirent_array[i].d_length;
+		current_entry = rhandle.readdir_response.dirent_array[i].d_name;
+		current_ino = pvfs2_khandle_to_ino(
+			&(rhandle.readdir_response.dirent_array[i].khandle));
+
+		gossip_debug(GOSSIP_DIR_DEBUG,
+			     "calling dir_emit for %s with len %d, pos %ld\n",
+			     current_entry,
+			     len,
+			     (unsigned long)pos);
+		ret =
+		    dir_emit(ctx, current_entry, len, current_ino, DT_UNKNOWN);
+		if (ret < 0) {
+			gossip_debug(GOSSIP_DIR_DEBUG,
+				     "dir_emit() failed. ret:%d\n",
+				     ret);
+			if (i < 2) {
+				gossip_err("dir_emit failed on one of the first two true PVFS directory entries.\n");
+				gossip_err("Duplicate entries may appear.\n");
+			}
+			buffer_full = 1;
+			break;
+		}
+		ctx->pos++;
+		gossip_ldebug(GOSSIP_DIR_DEBUG,
+			      "%s: ctx->pos:%lld\n",
+			      __func__,
+			      lld(ctx->pos));
+
+		pos++;
+	}
+
+	/* this means that all of the dir_emit calls succeeded */
+	if (i == rhandle.readdir_response.pvfs_dirent_outcount) {
+		/* update token */
+		*ptoken = rhandle.readdir_response.token;
+	} else {
+		/* this means a dir_emit call failed */
+		if (rhandle.readdir_response.token == PVFS_READDIR_END) {
+			/*
+			 * If PVFS hit end of directory, then there
+			 * is no way to do math on the token that it
+			 * returned. Instead we go by ctx->pos but
+			 * back up to account for the artificial .
+			 * and .. entries.
+			 */
+			ctx->pos -= 3;
+		} else {
+			/*
+			 * this means a dir_emit call failed. !!! need to set
+			 * back to previous ctx->pos, no middle value allowed
+			 */
+			pos -= (i - 1);
+			ctx->pos -= (i - 1);
+		}
+		gossip_debug(GOSSIP_DIR_DEBUG,
+			"at least one dir_emit call failed. Setting ctx->pos to: %lld\n",
+			lld(ctx->pos));
+	}
+
+	/*
+	 * Did we hit the end of the directory?
+	 */
+	if (rhandle.readdir_response.token == PVFS_READDIR_END &&
+	    !buffer_full) {
+		gossip_debug(GOSSIP_DIR_DEBUG, "End of dir detected; setting ctx->pos to PVFS_READDIR_END.\n");
+		ctx->pos = PVFS_READDIR_END;
+	}
+
+	gossip_debug(GOSSIP_DIR_DEBUG,
+		     "pos = %llu, token = %llu"
+		     ", ctx->pos should have been %lld\n",
+		     llu(pos),
+		     llu(*ptoken),
+		     lld(ctx->pos));
+
+out_destroy_handle:
+	readdir_handle_dtor(bufmap, &rhandle);
+out_free_op:
+	op_release(new_op);
+	gossip_debug(GOSSIP_DIR_DEBUG, "pvfs2_readdir returning %d\n", ret);
+	return ret;
+}
+
+static int pvfs2_dir_open(struct inode *inode, struct file *file)
+{
+	__u64 *ptoken;
+
+	file->private_data = kmalloc(sizeof(__u64), GFP_KERNEL);
+	if (!file->private_data)
+		return -ENOMEM;
+
+	ptoken = file->private_data;
+	*ptoken = PVFS_READDIR_START;
+	return 0;
+}
+
+static int pvfs2_dir_release(struct inode *inode, struct file *file)
+{
+	pvfs2_flush_inode(inode);
+	kfree(file->private_data);
+	return 0;
+}
+
+/** PVFS2 implementation of VFS directory operations */
+const struct file_operations pvfs2_dir_operations = {
+	.read = generic_read_dir,
+	.iterate = pvfs2_readdir,
+	.open = pvfs2_dir_open,
+	.release = pvfs2_dir_release,
+};

fs/orangefs/file.c

+/*
+ * (C) 2001 Clemson University and The University of Chicago
+ *
+ * See COPYING in top-level directory.
+ */
+
+/*
+ *  Linux VFS file operations.
+ */
+
+#include "protocol.h"
+#include "pvfs2-kernel.h"
+#include "pvfs2-bufmap.h"
+#include <linux/fs.h>
+#include <linux/pagemap.h>
+
+#define wake_up_daemon_for_return(op)			\
+do {							\
+	spin_lock(&op->lock);                           \
+	op->io_completed = 1;                           \
+	spin_unlock(&op->lock);                         \
+	wake_up_interruptible(&op->io_completion_waitq);\
+} while (0)
+
+/*
+ * Copy to client-core's address space from the buffers specified
+ * by the iovec upto total_size bytes.
+ * NOTE: the iovector can either contain addresses which
+ *       can futher be kernel-space or user-space addresses.
+ *       or it can pointers to struct page's
+ */
+static int precopy_buffers(struct pvfs2_bufmap *bufmap,
+			   int buffer_index,
+			   const struct iovec *vec,
+			   unsigned long nr_segs,
+			   size_t total_size,
+			   int from_user)
+{
+	int ret = 0;
+
+	/*
+	 * copy data from application/kernel by pulling it out
+	 * of the iovec.
+	 */
+	/* Are we copying from User Virtual Addresses? */
+	if (from_user)
+		ret = pvfs_bufmap_copy_iovec_from_user(
+			bufmap,
+			buffer_index,
+			vec,
+			nr_segs,
+			total_size);
+	/* Are we copying from Kernel Virtual Addresses? */
+	else
+		ret = pvfs_bufmap_copy_iovec_from_kernel(
+			bufmap,
+			buffer_index,
+			vec,
+			nr_segs,
+			total_size);
+	if (ret < 0)
+		gossip_err("%s: Failed to copy-in buffers. Please make sure that the pvfs2-client is running. %ld\n",
+			__func__,
+			(long)ret);
+	return ret;
+}
+
+/*
+ * Copy from client-core's address space to the buffers specified
+ * by the iovec upto total_size bytes.
+ * NOTE: the iovector can either contain addresses which
+ *       can futher be kernel-space or user-space addresses.
+ *       or it can pointers to struct page's
+ */
+static int postcopy_buffers(struct pvfs2_bufmap *bufmap,
+			    int buffer_index,
+			    const struct iovec *vec,
+			    int nr_segs,
+			    size_t total_size,
+			    int to_user)
+{
+	int ret = 0;
+
+	/*
+	 * copy data to application/kernel by pushing it out to
+	 * the iovec. NOTE; target buffers can be addresses or
+	 * struct page pointers.
+	 */
+	if (total_size) {
+		/* Are we copying to User Virtual Addresses? */
+		if (to_user)
+			ret = pvfs_bufmap_copy_to_user_iovec(
+				bufmap,
+				buffer_index,
+				vec,
+				nr_segs,
+				total_size);
+		/* Are we copying to Kern Virtual Addresses? */
+		else
+			ret = pvfs_bufmap_copy_to_kernel_iovec(
+				bufmap,
+				buffer_index,
+				vec,
+				nr_segs,
+				total_size);
+		if (ret < 0)
+			gossip_err("%s: Failed to copy-out buffers.  Please make sure that the pvfs2-client is running (%ld)\n",
+				__func__,
+				(long)ret);
+	}
+	return ret;
+}
+
+/*
+ * Post and wait for the I/O upcall to finish
+ */
+static ssize_t wait_for_direct_io(enum PVFS_io_type type, struct inode *inode,
+		loff_t *offset, struct iovec *vec, unsigned long nr_segs,
+		size_t total_size, loff_t readahead_size, int to_user)
+{
+	struct pvfs2_inode_s *pvfs2_inode = PVFS2_I(inode);
+	struct pvfs2_khandle *handle = &pvfs2_inode->refn.khandle;
+	struct pvfs2_bufmap *bufmap = NULL;
+	struct pvfs2_kernel_op_s *new_op = NULL;
+	int buffer_index = -1;
+	ssize_t ret;
+
+	new_op = op_alloc(PVFS2_VFS_OP_FILE_IO);
+	if (!new_op) {
+		ret = -ENOMEM;
+		goto out;
+	}
+	/* synchronous I/O */
+	new_op->upcall.req.io.async_vfs_io = PVFS_VFS_SYNC_IO;
+	new_op->upcall.req.io.readahead_size = readahead_size;
+	new_op->upcall.req.io.io_type = type;
+	new_op->upcall.req.io.refn = pvfs2_inode->refn;
+
+populate_shared_memory:
+	/* get a shared buffer index */
+	ret = pvfs_bufmap_get(&bufmap, &buffer_index);
+	if (ret < 0) {
+		gossip_debug(GOSSIP_FILE_DEBUG,
+			     "%s: pvfs_bufmap_get failure (%ld)\n",
+			     __func__, (long)ret);
+		goto out;
+	}
+	gossip_debug(GOSSIP_FILE_DEBUG,
+		     "%s(%pU): GET op %p -> buffer_index %d\n",
+		     __func__,
+		     handle,
+		     new_op,
+		     buffer_index);
+
+	new_op->uses_shared_memory = 1;
+	new_op->upcall.req.io.buf_index = buffer_index;
+	new_op->upcall.req.io.count = total_size;
+	new_op->upcall.req.io.offset = *offset;
+
+	gossip_debug(GOSSIP_FILE_DEBUG,
+		     "%s(%pU): copy_to_user %d nr_segs %lu, offset: %llu total_size: %zd\n",
+		     __func__,
+		     handle,
+		     to_user,
+		     nr_segs,
+		     llu(*offset),
+		     total_size);
+	/*
+	 * Stage 1: copy the buffers into client-core's address space
+	 * precopy_buffers only pertains to writes.
+	 */
+	if (type == PVFS_IO_WRITE) {
+		ret = precopy_buffers(bufmap,
+				      buffer_index,
+				      vec,
+				      nr_segs,
+				      total_size,
+				      to_user);
+		if (ret < 0)
+			goto out;
+	}
+
+	gossip_debug(GOSSIP_FILE_DEBUG,
+		     "%s(%pU): Calling post_io_request with tag (%llu)\n",
+		     __func__,
+		     handle,
+		     llu(new_op->tag));
+
+	/* Stage 2: Service the I/O operation */
+	ret = service_operation(new_op,
+				type == PVFS_IO_WRITE ?
+					"file_write" :
+					"file_read",
+				get_interruptible_flag(inode));
+
+	/*
+	 * If service_operation() returns -EAGAIN #and# the operation was
+	 * purged from pvfs2_request_list or htable_ops_in_progress, then
+	 * we know that the client was restarted, causing the shared memory
+	 * area to be wiped clean.  To restart a  write operation in this
+	 * case, we must re-copy the data from the user's iovec to a NEW
+	 * shared memory location. To restart a read operation, we must get
+	 * a new shared memory location.
+	 */
+	if (ret == -EAGAIN && op_state_purged(new_op)) {
+		pvfs_bufmap_put(bufmap, buffer_index);
+		gossip_debug(GOSSIP_FILE_DEBUG,
+			     "%s:going to repopulate_shared_memory.\n",
+			     __func__);
+		goto populate_shared_memory;
+	}
+
+	if (ret < 0) {
+		handle_io_error(); /* defined in pvfs2-kernel.h */
+		/*
+		   don't write an error to syslog on signaled operation
+		   termination unless we've got debugging turned on, as
+		   this can happen regularly (i.e. ctrl-c)
+		 */
+		if (ret == -EINTR)
+			gossip_debug(GOSSIP_FILE_DEBUG,
+				     "%s: returning error %ld\n", __func__,
+				     (long)ret);
+		else
+			gossip_err("%s: error in %s handle %pU, returning %zd\n",
+				__func__,
+				type == PVFS_IO_READ ?
+					"read from" : "write to",
+				handle, ret);
+		goto out;
+	}
+
+	/*
+	 * Stage 3: Post copy buffers from client-core's address space
+	 * postcopy_buffers only pertains to reads.
+	 */
+	if (type == PVFS_IO_READ) {
+		ret = postcopy_buffers(bufmap,
+				       buffer_index,
+				       vec,
+				       nr_segs,
+				       new_op->downcall.resp.io.amt_complete,
+				       to_user);
+		if (ret < 0) {
+			/*
+			 * put error codes in downcall so that handle_io_error()
+			 * preserves it properly
+			 */
+			new_op->downcall.status = ret;
+			handle_io_error();
+			goto out;
+		}
+	}
+	gossip_debug(GOSSIP_FILE_DEBUG,
+	    "%s(%pU): Amount written as returned by the sys-io call:%d\n",
+	    __func__,
+	    handle,
+	    (int)new_op->downcall.resp.io.amt_complete);
+
+	ret = new_op->downcall.resp.io.amt_complete;
+
+	/*
+	   tell the device file owner waiting on I/O that this read has
+	   completed and it can return now.  in this exact case, on
+	   wakeup the daemon will free the op, so we *cannot* touch it
+	   after this.
+	 */
+	wake_up_daemon_for_return(new_op);
+	new_op = NULL;
+
+out:
+	if (buffer_index >= 0) {
+		pvfs_bufmap_put(bufmap, buffer_index);
+		gossip_debug(GOSSIP_FILE_DEBUG,
+			     "%s(%pU): PUT buffer_index %d\n",
+			     __func__, handle, buffer_index);
+		buffer_index = -1;
+	}
+	if (new_op) {
+		op_release(new_op);
+		new_op = NULL;
+	}
+	return ret;
+}
+
+/*
+ * The reason we need to do this is to be able to support readv and writev
+ * that are larger than (pvfs_bufmap_size_query()) Default is
+ * PVFS2_BUFMAP_DEFAULT_DESC_SIZE MB. What that means is that we will
+ * create a new io vec descriptor for those memory addresses that
+ * go beyond the limit. Return value for this routine is negative in case
+ * of errors and 0 in case of success.
+ *
+ * Further, the new_nr_segs pointer is updated to hold the new value
+ * of number of iovecs, the new_vec pointer is updated to hold the pointer
+ * to the new split iovec, and the size array is an array of integers holding
+ * the number of iovecs that straddle pvfs_bufmap_size_query().
+ * The max_new_nr_segs value is computed by the caller and returned.
+ * (It will be (count of all iov_len/ block_size) + 1).
+ */
+static int split_iovecs(unsigned long max_new_nr_segs,		/* IN */
+			unsigned long nr_segs,			/* IN */
+			const struct iovec *original_iovec,	/* IN */
+			unsigned long *new_nr_segs,		/* OUT */
+			struct iovec **new_vec,			/* OUT */
+			unsigned long *seg_count,		/* OUT */
+			unsigned long **seg_array)		/* OUT */
+{
+	unsigned long seg;
+	unsigned long count = 0;
+	unsigned long begin_seg;
+	unsigned long tmpnew_nr_segs = 0;
+	struct iovec *new_iovec = NULL;
+	struct iovec *orig_iovec;
+	unsigned long *sizes = NULL;
+	unsigned long sizes_count = 0;
+
+	if (nr_segs <= 0 ||
+	    original_iovec == NULL ||
+	    new_nr_segs == NULL ||
+	    new_vec == NULL ||
+	    seg_count == NULL ||
+	    seg_array == NULL ||
+	    max_new_nr_segs <= 0) {
+		gossip_err("Invalid parameters to split_iovecs\n");
+		return -EINVAL;
+	}
+	*new_nr_segs = 0;
+	*new_vec = NULL;
+	*seg_count = 0;
+	*seg_array = NULL;
+	/* copy the passed in iovec descriptor to a temp structure */
+	orig_iovec = kmalloc_array(nr_segs,
+				   sizeof(*orig_iovec),
+				   PVFS2_BUFMAP_GFP_FLAGS);
+	if (orig_iovec == NULL) {
+		gossip_err(
+		    "split_iovecs: Could not allocate memory for %lu bytes!\n",
+		    (unsigned long)(nr_segs * sizeof(*orig_iovec)));
+		return -ENOMEM;
+	}
+	new_iovec = kcalloc(max_new_nr_segs,
+			    sizeof(*new_iovec),
+			    PVFS2_BUFMAP_GFP_FLAGS);
+	if (new_iovec == NULL) {
+		kfree(orig_iovec);
+		gossip_err(
+		    "split_iovecs: Could not allocate memory for %lu bytes!\n",
+		    (unsigned long)(max_new_nr_segs * sizeof(*new_iovec)));
+		return -ENOMEM;
+	}
+	sizes = kcalloc(max_new_nr_segs,
+			sizeof(*sizes),
+			PVFS2_BUFMAP_GFP_FLAGS);
+	if (sizes == NULL) {
+		kfree(new_iovec);
+		kfree(orig_iovec);
+		gossip_err(
+		    "split_iovecs: Could not allocate memory for %lu bytes!\n",
+		    (unsigned long)(max_new_nr_segs * sizeof(*sizes)));
+		return -ENOMEM;
+	}
+	/* copy the passed in iovec to a temp structure */
+	memcpy(orig_iovec, original_iovec, nr_segs * sizeof(*orig_iovec));
+	begin_seg = 0;
+repeat:
+	for (seg = begin_seg; seg < nr_segs; seg++) {
+		if (tmpnew_nr_segs >= max_new_nr_segs ||
+		    sizes_count >= max_new_nr_segs) {
+			kfree(sizes);
+			kfree(orig_iovec);
+			kfree(new_iovec);
+			gossip_err
+			    ("split_iovecs: exceeded the index limit (%lu)\n",
+			    tmpnew_nr_segs);
+			return -EINVAL;
+		}
+		if (count + orig_iovec[seg].iov_len <
+		    pvfs_bufmap_size_query()) {
+			count += orig_iovec[seg].iov_len;
+			memcpy(&new_iovec[tmpnew_nr_segs],
+			       &orig_iovec[seg],
+			       sizeof(*new_iovec));
+			tmpnew_nr_segs++;
+			sizes[sizes_count]++;
+		} else {
+			new_iovec[tmpnew_nr_segs].iov_base =
+			    orig_iovec[seg].iov_base;
+			new_iovec[tmpnew_nr_segs].iov_len =
+			    (pvfs_bufmap_size_query() - count);
+			tmpnew_nr_segs++;
+			sizes[sizes_count]++;
+			sizes_count++;
+			begin_seg = seg;
+			orig_iovec[seg].iov_base +=
+			    (pvfs_bufmap_size_query() - count);
+			orig_iovec[seg].iov_len -=
+			    (pvfs_bufmap_size_query() - count);
+			count = 0;
+			break;
+		}
+	}
+	if (seg != nr_segs)
+		goto repeat;
+	else
+		sizes_count++;
+
+	*new_nr_segs = tmpnew_nr_segs;
+	/* new_iovec is freed by the caller */
+	*new_vec = new_iovec;
+	*seg_count = sizes_count;
+	/* seg_array is also freed by the caller */
+	*seg_array = sizes;
+	kfree(orig_iovec);
+	return 0;
+}
+
+static long bound_max_iovecs(const struct iovec *curr, unsigned long nr_segs,
+			     ssize_t *total_count)
+{
+	unsigned long i;
+	long max_nr_iovecs;
+	ssize_t total;
+	ssize_t count;
+
+	total = 0;
+	count = 0;
+	max_nr_iovecs = 0;
+	for (i = 0; i < nr_segs; i++) {
+		const struct iovec *iv = &curr[i];
+
+		count += iv->iov_len;
+		if (unlikely((ssize_t) (count | iv->iov_len) < 0))
+			return -EINVAL;
+		if (total + iv->iov_len < pvfs_bufmap_size_query()) {
+			total += iv->iov_len;
+			max_nr_iovecs++;
+		} else {
+			total =
+			    (total + iv->iov_len - pvfs_bufmap_size_query());
+			max_nr_iovecs += (total / pvfs_bufmap_size_query() + 2);
+		}
+	}
+	*total_count = count;
+	return max_nr_iovecs;
+}
+
+/*
+ * Common entry point for read/write/readv/writev
+ * This function will dispatch it to either the direct I/O
+ * or buffered I/O path depending on the mount options and/or
+ * augmented/extended metadata attached to the file.
+ * Note: File extended attributes override any mount options.
+ */
+static ssize_t do_readv_writev(enum PVFS_io_type type, struct file *file,
+		loff_t *offset, const struct iovec *iov, unsigned long nr_segs)
+{
+	struct inode *inode = file->f_mapping->host;
+	struct pvfs2_inode_s *pvfs2_inode = PVFS2_I(inode);
+	struct pvfs2_khandle *handle = &pvfs2_inode->refn.khandle;
+	ssize_t ret;
+	ssize_t total_count;
+	unsigned int to_free;
+	size_t count;
+	unsigned long seg;
+	unsigned long new_nr_segs = 0;
+	unsigned long max_new_nr_segs = 0;
+	unsigned long seg_count = 0;
+	unsigned long *seg_array = NULL;
+	struct iovec *iovecptr = NULL;
+	struct iovec *ptr = NULL;
+
+	total_count = 0;
+	ret = -EINVAL;
+	count = 0;
+	to_free = 0;
+
+	/* Compute total and max number of segments after split */
+	max_new_nr_segs = bound_max_iovecs(iov, nr_segs, &count);
+	if (max_new_nr_segs < 0) {
+		gossip_lerr("%s: could not bound iovec %lu\n",
+			    __func__,
+			    max_new_nr_segs);
+		goto out;
+	}
+
+	gossip_debug(GOSSIP_FILE_DEBUG,
+		"%s-BEGIN(%pU): count(%d) after estimate_max_iovecs.\n",
+		__func__,
+		handle,
+		(int)count);
+
+	if (type == PVFS_IO_WRITE) {
+		gossip_debug(GOSSIP_FILE_DEBUG,
+			     "%s(%pU): proceeding with offset : %llu, "
+			     "size %d\n",
+			     __func__,
+			     handle,
+			     llu(*offset),
+			     (int)count);
+	}
+
+	if (count == 0) {
+		ret = 0;
+		goto out;
+	}
+
+	/*
+	 * if the total size of data transfer requested is greater than
+	 * the kernel-set blocksize of PVFS2, then we split the iovecs
+	 * such that no iovec description straddles a block size limit
+	 */
+
+	gossip_debug(GOSSIP_FILE_DEBUG,
+		     "%s: pvfs_bufmap_size:%d\n",
+		     __func__,
+		     pvfs_bufmap_size_query());
+
+	if (count > pvfs_bufmap_size_query()) {
+		/*
+		 * Split up the given iovec description such that
+		 * no iovec descriptor straddles over the block-size limitation.
+		 * This makes us our job easier to stage the I/O.
+		 * In addition, this function will also compute an array
+		 * with seg_count entries that will store the number of
+		 * segments that straddle the block-size boundaries.
+		 */
+		ret = split_iovecs(max_new_nr_segs,	/* IN */
+				   nr_segs,		/* IN */
+				   iov,			/* IN */
+				   &new_nr_segs,	/* OUT */
+				   &iovecptr,		/* OUT */
+				   &seg_count,		/* OUT */
+				   &seg_array);		/* OUT */
+		if (ret < 0) {
+			gossip_err("%s: Failed to split iovecs to satisfy larger than blocksize readv/writev request %zd\n",
+				__func__,
+				ret);
+			goto out;
+		}
+		gossip_debug(GOSSIP_FILE_DEBUG,
+			     "%s: Splitting iovecs from %lu to %lu"
+			     " [max_new %lu]\n",
+			     __func__,
+			     nr_segs,
+			     new_nr_segs,
+			     max_new_nr_segs);
+		/* We must free seg_array and iovecptr */
+		to_free = 1;
+	} else {
+		new_nr_segs = nr_segs;
+		/* use the given iovec description */
+		iovecptr = (struct iovec *)iov;
+		/* There is only 1 element in the seg_array */
+		seg_count = 1;
+		/* and its value is the number of segments passed in */
+		seg_array = &nr_segs;
+		/* We dont have to free up anything */
+		to_free = 0;
+	}
+	ptr = iovecptr;
+
+	gossip_debug(GOSSIP_FILE_DEBUG,
+		     "%s(%pU) %zd@%llu\n",
+		     __func__,
+		     handle,
+		     count,
+		     llu(*offset));
+	gossip_debug(GOSSIP_FILE_DEBUG,
+		     "%s(%pU): new_nr_segs: %lu, seg_count: %lu\n",
+		     __func__,
+		     handle,
+		     new_nr_segs, seg_count);
+
+/* PVFS2_KERNEL_DEBUG is a CFLAGS define. */
+#ifdef PVFS2_KERNEL_DEBUG
+	for (seg = 0; seg < new_nr_segs; seg++)
+		gossip_debug(GOSSIP_FILE_DEBUG,
+			     "%s: %d) %p to %p [%d bytes]\n",
+			     __func__,
+			     (int)seg + 1,
+			     iovecptr[seg].iov_base,
+			     iovecptr[seg].iov_base + iovecptr[seg].iov_len,
+			     (int)iovecptr[seg].iov_len);
+	for (seg = 0; seg < seg_count; seg++)
+		gossip_debug(GOSSIP_FILE_DEBUG,
+			     "%s: %zd) %lu\n",
+			     __func__,
+			     seg + 1,
+			     seg_array[seg]);
+#endif
+	seg = 0;
+	while (total_count < count) {
+		size_t each_count;
+		size_t amt_complete;
+
+		/* how much to transfer in this loop iteration */
+		each_count =
+		   (((count - total_count) > pvfs_bufmap_size_query()) ?
+			pvfs_bufmap_size_query() :
+			(count - total_count));
+
+		gossip_debug(GOSSIP_FILE_DEBUG,
+			     "%s(%pU): size of each_count(%d)\n",
+			     __func__,
+			     handle,
+			     (int)each_count);
+		gossip_debug(GOSSIP_FILE_DEBUG,
+			     "%s(%pU): BEFORE wait_for_io: offset is %d\n",
+			     __func__,
+			     handle,
+			     (int)*offset);
+
+		ret = wait_for_direct_io(type, inode, offset, ptr,
+				seg_array[seg], each_count, 0, 1);
+		gossip_debug(GOSSIP_FILE_DEBUG,
+			     "%s(%pU): return from wait_for_io:%d\n",
+			     __func__,
+			     handle,
+			     (int)ret);
+
+		if (ret < 0)
+			goto out;
+
+		/* advance the iovec pointer */
+		ptr += seg_array[seg];
+		seg++;
+		*offset += ret;
+		total_count += ret;
+		amt_complete = ret;
+
+		gossip_debug(GOSSIP_FILE_DEBUG,
+			     "%s(%pU): AFTER wait_for_io: offset is %d\n",
+			     __func__,
+			     handle,
+			     (int)*offset);
+
+		/*
+		 * if we got a short I/O operations,
+		 * fall out and return what we got so far
+		 */
+		if (amt_complete < each_count)
+			break;
+	} /*end while */
+
+	if (total_count > 0)
+		ret = total_count;
+out:
+	if (to_free) {
+		kfree(iovecptr);
+		kfree(seg_array);
+	}
+	if (ret > 0) {
+		if (type == PVFS_IO_READ) {
+			file_accessed(file);
+		} else {
+			SetMtimeFlag(pvfs2_inode);
+			inode->i_mtime = CURRENT_TIME;
+			mark_inode_dirty_sync(inode);
+		}
+	}
+
+	gossip_debug(GOSSIP_FILE_DEBUG,
+		     "%s(%pU): Value(%d) returned.\n",
+		     __func__,
+		     handle,
+		     (int)ret);
+
+	return ret;
+}
+
+/*
+ * Read data from a specified offset in a file (referenced by inode).
+ * Data may be placed either in a user or kernel buffer.
+ */
+ssize_t pvfs2_inode_read(struct inode *inode,
+			 char __user *buf,
+			 size_t count,
+			 loff_t *offset,
+			 loff_t readahead_size)
+{
+	struct pvfs2_inode_s *pvfs2_inode = PVFS2_I(inode);
+	size_t bufmap_size;
+	struct iovec vec;
+	ssize_t ret = -EINVAL;
+
+	g_pvfs2_stats.reads++;
+
+	vec.iov_base = buf;
+	vec.iov_len = count;
+
+	bufmap_size = pvfs_bufmap_size_query();
+	if (count > bufmap_size) {
+		gossip_debug(GOSSIP_FILE_DEBUG,
+			     "%s: count is too large (%zd/%zd)!\n",
+			     __func__, count, bufmap_size);
+		return -EINVAL;
+	}
+
+	gossip_debug(GOSSIP_FILE_DEBUG,
+		     "%s(%pU) %zd@%llu\n",
+		     __func__,
+		     &pvfs2_inode->refn.khandle,
+		     count,
+		     llu(*offset));
+
+	ret = wait_for_direct_io(PVFS_IO_READ, inode, offset, &vec, 1,
+			count, readahead_size, 0);
+	if (ret > 0)
+		*offset += ret;
+
+	gossip_debug(GOSSIP_FILE_DEBUG,
+		     "%s(%pU): Value(%zd) returned.\n",
+		     __func__,
+		     &pvfs2_inode->refn.khandle,
+		     ret);
+
+	return ret;
+}
+
+static ssize_t pvfs2_file_read_iter(struct kiocb *iocb, struct iov_iter *iter)
+{
+	struct file *file = iocb->ki_filp;
+	loff_t pos = *(&iocb->ki_pos);
+	ssize_t rc = 0;
+	unsigned long nr_segs = iter->nr_segs;
+
+	BUG_ON(iocb->private);
+
+	gossip_debug(GOSSIP_FILE_DEBUG, "pvfs2_file_read_iter\n");
+
+	g_pvfs2_stats.reads++;
+
+	rc = do_readv_writev(PVFS_IO_READ,
+			     file,
+			     &pos,
+			     iter->iov,
+			     nr_segs);
+	iocb->ki_pos = pos;
+
+	return rc;
+}
+
+static ssize_t pvfs2_file_write_iter(struct kiocb *iocb, struct iov_iter *iter)
+{
+	struct file *file = iocb->ki_filp;
+	loff_t pos = *(&iocb->ki_pos);
+	unsigned long nr_segs = iter->nr_segs;
+	ssize_t rc;
+
+	BUG_ON(iocb->private);
+
+	gossip_debug(GOSSIP_FILE_DEBUG, "pvfs2_file_write_iter\n");
+
+	mutex_lock(&file->f_mapping->host->i_mutex);
+
+	/* Make sure generic_write_checks sees an up to date inode size. */
+	if (file->f_flags & O_APPEND) {
+		rc = pvfs2_inode_getattr(file->f_mapping->host,
+					 PVFS_ATTR_SYS_SIZE);
+		if (rc) {
+			gossip_err("%s: pvfs2_inode_getattr failed, rc:%zd:.\n",
+				   __func__, rc);
+			goto out;
+		}
+	}
+
+	if (file->f_pos > i_size_read(file->f_mapping->host))
+		pvfs2_i_size_write(file->f_mapping->host, file->f_pos);
+
+	rc = generic_write_checks(iocb, iter);
+
+	if (rc <= 0) {
+		gossip_err("%s: generic_write_checks failed, rc:%zd:.\n",
+			   __func__, rc);
+		goto out;
+	}
+
+	rc = do_readv_writev(PVFS_IO_WRITE,
+			     file,
+			     &pos,
+			     iter->iov,
+			     nr_segs);
+	if (rc < 0) {
+		gossip_err("%s: do_readv_writev failed, rc:%zd:.\n",
+			   __func__, rc);
+		goto out;
+	}
+
+	iocb->ki_pos = pos;
+	g_pvfs2_stats.writes++;
+
+out:
+
+	mutex_unlock(&file->f_mapping->host->i_mutex);
+	return rc;
+}
+
+/*
+ * Perform a miscellaneous operation on a file.
+ */
+long pvfs2_ioctl(struct file *file, unsigned int cmd, unsigned long arg)
+{
+	int ret = -ENOTTY;
+	__u64 val = 0;
+	unsigned long uval;
+
+	gossip_debug(GOSSIP_FILE_DEBUG,
+		     "pvfs2_ioctl: called with cmd %d\n",
+		     cmd);
+
+	/*
+	 * we understand some general ioctls on files, such as the immutable
+	 * and append flags
+	 */
+	if (cmd == FS_IOC_GETFLAGS) {
+		val = 0;
+		ret = pvfs2_xattr_get_default(file->f_path.dentry,
+					      "user.pvfs2.meta_hint",
+					      &val,
+					      sizeof(val),
+					      0);
+		if (ret < 0 && ret != -ENODATA)
+			return ret;
+		else if (ret == -ENODATA)
+			val = 0;
+		uval = val;
+		gossip_debug(GOSSIP_FILE_DEBUG,
+			     "pvfs2_ioctl: FS_IOC_GETFLAGS: %llu\n",
+			     (unsigned long long)uval);
+		return put_user(uval, (int __user *)arg);
+	} else if (cmd == FS_IOC_SETFLAGS) {
+		ret = 0;
+		if (get_user(uval, (int __user *)arg))
+			return -EFAULT;
+		/*
+		 * PVFS_MIRROR_FL is set internally when the mirroring mode
+		 * is turned on for a file. The user is not allowed to turn
+		 * on this bit, but the bit is present if the user first gets
+		 * the flags and then updates the flags with some new
+		 * settings. So, we ignore it in the following edit. bligon.
+		 */
+		if ((uval & ~PVFS_MIRROR_FL) &
+		    (~(FS_IMMUTABLE_FL | FS_APPEND_FL | FS_NOATIME_FL))) {
+			gossip_err("pvfs2_ioctl: the FS_IOC_SETFLAGS only supports setting one of FS_IMMUTABLE_FL|FS_APPEND_FL|FS_NOATIME_FL\n");
+			return -EINVAL;
+		}
+		val = uval;
+		gossip_debug(GOSSIP_FILE_DEBUG,
+			     "pvfs2_ioctl: FS_IOC_SETFLAGS: %llu\n",
+			     (unsigned long long)val);
+		ret = pvfs2_xattr_set_default(file->f_path.dentry,
+					      "user.pvfs2.meta_hint",
+					      &val,
+					      sizeof(val),
+					      0,
+					      0);
+	}
+
+	return ret;
+}
+
+/*
+ * Memory map a region of a file.
+ */
+static int pvfs2_file_mmap(struct file *file, struct vm_area_struct *vma)
+{
+	gossip_debug(GOSSIP_FILE_DEBUG,
+		     "pvfs2_file_mmap: called on %s\n",
+		     (file ?
+			(char *)file->f_path.dentry->d_name.name :
+			(char *)"Unknown"));
+
+	/* set the sequential readahead hint */
+	vma->vm_flags |= VM_SEQ_READ;
+	vma->vm_flags &= ~VM_RAND_READ;
+	return generic_file_mmap(file, vma);
+}
+
+#define mapping_nrpages(idata) ((idata)->nrpages)
+
+/*
+ * Called to notify the module that there are no more references to
+ * this file (i.e. no processes have it open).
+ *
+ * \note Not called when each file is closed.
+ */
+int pvfs2_file_release(struct inode *inode, struct file *file)
+{
+	gossip_debug(GOSSIP_FILE_DEBUG,
+		     "pvfs2_file_release: called on %s\n",
+		     file->f_path.dentry->d_name.name);
+
+	pvfs2_flush_inode(inode);
+
+	/*
+	   remove all associated inode pages from the page cache and mmap
+	   readahead cache (if any); this forces an expensive refresh of
+	   data for the next caller of mmap (or 'get_block' accesses)
+	 */
+	if (file->f_path.dentry->d_inode &&
+	    file->f_path.dentry->d_inode->i_mapping &&
+	    mapping_nrpages(&file->f_path.dentry->d_inode->i_data))
+		truncate_inode_pages(file->f_path.dentry->d_inode->i_mapping,
+				     0);
+	return 0;
+}
+
+/*
+ * Push all data for a specific file onto permanent storage.
+ */
+int pvfs2_fsync(struct file *file, loff_t start, loff_t end, int datasync)
+{
+	int ret = -EINVAL;
+	struct pvfs2_inode_s *pvfs2_inode =
+		PVFS2_I(file->f_path.dentry->d_inode);
+	struct pvfs2_kernel_op_s *new_op = NULL;
+
+	/* required call */
+	filemap_write_and_wait_range(file->f_mapping, start, end);
+
+	new_op = op_alloc(PVFS2_VFS_OP_FSYNC);
+	if (!new_op)
+		return -ENOMEM;
+	new_op->upcall.req.fsync.refn = pvfs2_inode->refn;
+
+	ret = service_operation(new_op,
+			"pvfs2_fsync",
+			get_interruptible_flag(file->f_path.dentry->d_inode));
+
+	gossip_debug(GOSSIP_FILE_DEBUG,
+		     "pvfs2_fsync got return value of %d\n",
+		     ret);
+
+	op_release(new_op);
+
+	pvfs2_flush_inode(file->f_path.dentry->d_inode);
+	return ret;
+}
+
+/*
+ * Change the file pointer position for an instance of an open file.
+ *
+ * \note If .llseek is overriden, we must acquire lock as described in
+ *       Documentation/filesystems/Locking.
+ *
+ * Future upgrade could support SEEK_DATA and SEEK_HOLE but would
+ * require much changes to the FS
+ */
+loff_t pvfs2_file_llseek(struct file *file, loff_t offset, int origin)
+{
+	int ret = -EINVAL;
+	struct inode *inode = file->f_path.dentry->d_inode;
+
+	if (!inode) {
+		gossip_err("pvfs2_file_llseek: invalid inode (NULL)\n");
+		return ret;
+	}
+
+	if (origin == PVFS2_SEEK_END) {
+		/*
+		 * revalidate the inode's file size.
+		 * NOTE: We are only interested in file size here,
+		 * so we set mask accordingly.
+		 */
+		ret = pvfs2_inode_getattr(inode, PVFS_ATTR_SYS_SIZE);
+		if (ret) {
+			gossip_debug(GOSSIP_FILE_DEBUG,
+				     "%s:%s:%d calling make bad inode\n",
+				     __FILE__,
+				     __func__,
+				     __LINE__);
+			pvfs2_make_bad_inode(inode);
+			return ret;
+		}
+	}
+
+	gossip_debug(GOSSIP_FILE_DEBUG,
+		     "pvfs2_file_llseek: offset is %ld | origin is %d | "
+		     "inode size is %lu\n",
+		     (long)offset,
+		     origin,
+		     (unsigned long)file->f_path.dentry->d_inode->i_size);
+
+	return generic_file_llseek(file, offset, origin);
+}
+
+/*
+ * Support local locks (locks that only this kernel knows about)
+ * if Orangefs was mounted -o local_lock.
+ */
+int pvfs2_lock(struct file *filp, int cmd, struct file_lock *fl)
+{
+	int rc = -ENOLCK;
+
+	if (PVFS2_SB(filp->f_inode->i_sb)->flags & PVFS2_OPT_LOCAL_LOCK) {
+		if (cmd == F_GETLK) {
+			rc = 0;
+			posix_test_lock(filp, fl);
+		} else {
+			rc = posix_lock_file(filp, fl, NULL);
+		}
+	}
+
+	return rc;
+}
+
+/** PVFS2 implementation of VFS file operations */
+const struct file_operations pvfs2_file_operations = {
+	.llseek		= pvfs2_file_llseek,
+	.read_iter	= pvfs2_file_read_iter,
+	.write_iter	= pvfs2_file_write_iter,
+	.lock		= pvfs2_lock,
+	.unlocked_ioctl	= pvfs2_ioctl,
+	.mmap		= pvfs2_file_mmap,
+	.open		= generic_file_open,
+	.release	= pvfs2_file_release,
+	.fsync		= pvfs2_fsync,
+};

fs/orangefs/inode.c

+/*
+ * (C) 2001 Clemson University and The University of Chicago
+ *
+ * See COPYING in top-level directory.
+ */
+
+/*
+ *  Linux VFS inode operations.
+ */
+
+#include "protocol.h"
+#include "pvfs2-kernel.h"
+#include "pvfs2-bufmap.h"
+
+static int read_one_page(struct page *page)
+{
+	void *page_data;
+	int ret;
+	int max_block;
+	ssize_t bytes_read = 0;
+	struct inode *inode = page->mapping->host;
+	const __u32 blocksize = PAGE_CACHE_SIZE;	/* inode->i_blksize */
+	const __u32 blockbits = PAGE_CACHE_SHIFT;	/* inode->i_blkbits */
+
+	gossip_debug(GOSSIP_INODE_DEBUG,
+		    "pvfs2_readpage called with page %p\n",
+		     page);
+	page_data = pvfs2_kmap(page);
+
+	max_block = ((inode->i_size / blocksize) + 1);
+
+	if (page->index < max_block) {
+		loff_t blockptr_offset = (((loff_t) page->index) << blockbits);
+
+		bytes_read = pvfs2_inode_read(inode,
+					      page_data,
+					      blocksize,
+					      &blockptr_offset,
+					      inode->i_size);
+	}
+	/* only zero remaining unread portions of the page data */
+	if (bytes_read > 0)
+		memset(page_data + bytes_read, 0, blocksize - bytes_read);
+	else
+		memset(page_data, 0, blocksize);
+	/* takes care of potential aliasing */
+	flush_dcache_page(page);
+	if (bytes_read < 0) {
+		ret = bytes_read;
+		SetPageError(page);
+	} else {
+		SetPageUptodate(page);
+		if (PageError(page))
+			ClearPageError(page);
+		ret = 0;
+	}
+	pvfs2_kunmap(page);
+	/* unlock the page after the ->readpage() routine completes */
+	unlock_page(page);
+	return ret;
+}
+
+static int pvfs2_readpage(struct file *file, struct page *page)
+{
+	return read_one_page(page);
+}
+
+static int pvfs2_readpages(struct file *file,
+			   struct address_space *mapping,
+			   struct list_head *pages,
+			   unsigned nr_pages)
+{
+	int page_idx;
+	int ret;
+
+	gossip_debug(GOSSIP_INODE_DEBUG, "pvfs2_readpages called\n");
+
+	for (page_idx = 0; page_idx < nr_pages; page_idx++) {
+		struct page *page;
+
+		page = list_entry(pages->prev, struct page, lru);
+		list_del(&page->lru);
+		if (!add_to_page_cache(page,
+				       mapping,
+				       page->index,
+				       GFP_KERNEL)) {
+			ret = read_one_page(page);
+			gossip_debug(GOSSIP_INODE_DEBUG,
+				"failure adding page to cache, read_one_page returned: %d\n",
+				ret);
+	      } else {
+			page_cache_release(page);
+	      }
+	}
+	BUG_ON(!list_empty(pages));
+	return 0;
+}
+
+static void pvfs2_invalidatepage(struct page *page,
+				 unsigned int offset,
+				 unsigned int length)
+{
+	gossip_debug(GOSSIP_INODE_DEBUG,
+		     "pvfs2_invalidatepage called on page %p "
+		     "(offset is %u)\n",
+		     page,
+		     offset);
+
+	ClearPageUptodate(page);
+	ClearPageMappedToDisk(page);
+	return;
+
+}
+
+static int pvfs2_releasepage(struct page *page, gfp_t foo)
+{
+	gossip_debug(GOSSIP_INODE_DEBUG,
+		     "pvfs2_releasepage called on page %p\n",
+		     page);
+	return 0;
+}
+
+/*
+ * Having a direct_IO entry point in the address_space_operations
+ * struct causes the kernel to allows us to use O_DIRECT on
+ * open. Nothing will ever call this thing, but in the future we
+ * will need to be able to use O_DIRECT on open in order to support
+ * AIO. Modeled after NFS, they do this too.
+ */
+/*
+static ssize_t pvfs2_direct_IO(int rw,
+			struct kiocb *iocb,
+			struct iov_iter *iter,
+			loff_t offset)
+{
+	gossip_debug(GOSSIP_INODE_DEBUG,
+		     "pvfs2_direct_IO: %s\n",
+		     iocb->ki_filp->f_path.dentry->d_name.name);
+
+	return -EINVAL;
+}
+*/
+
+struct backing_dev_info pvfs2_backing_dev_info = {
+	.name = "pvfs2",
+	.ra_pages = 0,
+	.capabilities = BDI_CAP_NO_ACCT_DIRTY | BDI_CAP_NO_WRITEBACK,
+};
+
+/** PVFS2 implementation of address space operations */
+const struct address_space_operations pvfs2_address_operations = {
+	.readpage = pvfs2_readpage,
+	.readpages = pvfs2_readpages,
+	.invalidatepage = pvfs2_invalidatepage,
+	.releasepage = pvfs2_releasepage,
+/*	.direct_IO = pvfs2_direct_IO */
+};
+
+static int pvfs2_setattr_size(struct inode *inode, struct iattr *iattr)
+{
+	struct pvfs2_inode_s *pvfs2_inode = PVFS2_I(inode);
+	struct pvfs2_kernel_op_s *new_op;
+	loff_t orig_size = i_size_read(inode);
+	int ret = -EINVAL;
+
+	gossip_debug(GOSSIP_INODE_DEBUG,
+		     "%s: %pU: Handle is %pU | fs_id %d | size is %llu\n",
+		     __func__,
+		     get_khandle_from_ino(inode),
+		     &pvfs2_inode->refn.khandle,
+		     pvfs2_inode->refn.fs_id,
+		     iattr->ia_size);
+
+	truncate_setsize(inode, iattr->ia_size);
+
+	new_op = op_alloc(PVFS2_VFS_OP_TRUNCATE);
+	if (!new_op)
+		return -ENOMEM;
+
+	new_op->upcall.req.truncate.refn = pvfs2_inode->refn;
+	new_op->upcall.req.truncate.size = (__s64) iattr->ia_size;
+
+	ret = service_operation(new_op, __func__,
+				get_interruptible_flag(inode));
+
+	/*
+	 * the truncate has no downcall members to retrieve, but
+	 * the status value tells us if it went through ok or not
+	 */
+	gossip_debug(GOSSIP_INODE_DEBUG,
+		     "pvfs2: pvfs2_truncate got return value of %d\n",
+		     ret);
+
+	op_release(new_op);
+
+	if (ret != 0)
+		return ret;
+
+	/*
+	 * Only change the c/mtime if we are changing the size or we are
+	 * explicitly asked to change it.  This handles the semantic difference
+	 * between truncate() and ftruncate() as implemented in the VFS.
+	 *
+	 * The regular truncate() case without ATTR_CTIME and ATTR_MTIME is a
+	 * special case where we need to update the times despite not having
+	 * these flags set.  For all other operations the VFS set these flags
+	 * explicitly if it wants a timestamp update.
+	 */
+	if (orig_size != i_size_read(inode) &&
+	    !(iattr->ia_valid & (ATTR_CTIME | ATTR_MTIME))) {
+		iattr->ia_ctime = iattr->ia_mtime =
+			current_fs_time(inode->i_sb);
+		iattr->ia_valid |= ATTR_CTIME | ATTR_MTIME;
+	}
+
+	return ret;
+}
+
+/*
+ * Change attributes of an object referenced by dentry.
+ */
+int pvfs2_setattr(struct dentry *dentry, struct iattr *iattr)
+{
+	int ret = -EINVAL;
+	struct inode *inode = dentry->d_inode;
+
+	gossip_debug(GOSSIP_INODE_DEBUG,
+		     "pvfs2_setattr: called on %s\n",
+		     dentry->d_name.name);
+
+	ret = inode_change_ok(inode, iattr);
+	if (ret)
+		goto out;
+
+	if ((iattr->ia_valid & ATTR_SIZE) &&
+	    iattr->ia_size != i_size_read(inode)) {
+		ret = pvfs2_setattr_size(inode, iattr);
+		if (ret)
+			goto out;
+	}
+
+	setattr_copy(inode, iattr);
+	mark_inode_dirty(inode);
+
+	ret = pvfs2_inode_setattr(inode, iattr);
+	gossip_debug(GOSSIP_INODE_DEBUG,
+		     "pvfs2_setattr: inode_setattr returned %d\n",
+		     ret);
+
+	if (!ret && (iattr->ia_valid & ATTR_MODE))
+		/* change mod on a file that has ACLs */
+		ret = posix_acl_chmod(inode, inode->i_mode);
+
+out:
+	gossip_debug(GOSSIP_INODE_DEBUG, "pvfs2_setattr: returning %d\n", ret);
+	return ret;
+}
+
+/*
+ * Obtain attributes of an object given a dentry
+ */
+int pvfs2_getattr(struct vfsmount *mnt,
+		  struct dentry *dentry,
+		  struct kstat *kstat)
+{
+	int ret = -ENOENT;
+	struct inode *inode = dentry->d_inode;
+	struct pvfs2_inode_s *pvfs2_inode = NULL;
+
+	gossip_debug(GOSSIP_INODE_DEBUG,
+		     "pvfs2_getattr: called on %s\n",
+		     dentry->d_name.name);
+
+	/*
+	 * Similar to the above comment, a getattr also expects that all
+	 * fields/attributes of the inode would be refreshed. So again, we
+	 * dont have too much of a choice but refresh all the attributes.
+	 */
+	ret = pvfs2_inode_getattr(inode, PVFS_ATTR_SYS_ALL_NOHINT);
+	if (ret == 0) {
+		generic_fillattr(inode, kstat);
+		/* override block size reported to stat */
+		pvfs2_inode = PVFS2_I(inode);
+		kstat->blksize = pvfs2_inode->blksize;
+	} else {
+		/* assume an I/O error and flag inode as bad */
+		gossip_debug(GOSSIP_INODE_DEBUG,
+			     "%s:%s:%d calling make bad inode\n",
+			     __FILE__,
+			     __func__,
+			     __LINE__);
+		pvfs2_make_bad_inode(inode);
+	}
+	return ret;
+}
+
+/* PVFS2 implementation of VFS inode operations for files */
+struct inode_operations pvfs2_file_inode_operations = {
+	.get_acl = pvfs2_get_acl,
+	.set_acl = pvfs2_set_acl,
+	.setattr = pvfs2_setattr,
+	.getattr = pvfs2_getattr,
+	.setxattr = generic_setxattr,
+	.getxattr = generic_getxattr,
+	.listxattr = pvfs2_listxattr,
+	.removexattr = generic_removexattr,
+};
+
+static int pvfs2_init_iops(struct inode *inode)
+{
+	inode->i_mapping->a_ops = &pvfs2_address_operations;
+
+	switch (inode->i_mode & S_IFMT) {
+	case S_IFREG:
+		inode->i_op = &pvfs2_file_inode_operations;
+		inode->i_fop = &pvfs2_file_operations;
+		inode->i_blkbits = PAGE_CACHE_SHIFT;
+		break;
+	case S_IFLNK:
+		inode->i_op = &pvfs2_symlink_inode_operations;
+		break;
+	case S_IFDIR:
+		inode->i_op = &pvfs2_dir_inode_operations;
+		inode->i_fop = &pvfs2_dir_operations;
+		break;
+	default:
+		gossip_debug(GOSSIP_INODE_DEBUG,
+			     "%s: unsupported mode\n",
+			     __func__);
+		return -EINVAL;
+	}
+
+	return 0;
+}
+
+/*
+ * Given a PVFS2 object identifier (fsid, handle), convert it into a ino_t type
+ * that will be used as a hash-index from where the handle will
+ * be searched for in the VFS hash table of inodes.
+ */
+static inline ino_t pvfs2_handle_hash(struct pvfs2_object_kref *ref)
+{
+	if (!ref)
+		return 0;
+	return pvfs2_khandle_to_ino(&(ref->khandle));
+}
+
+/*
+ * Called to set up an inode from iget5_locked.
+ */
+static int pvfs2_set_inode(struct inode *inode, void *data)
+{
+	struct pvfs2_object_kref *ref = (struct pvfs2_object_kref *) data;
+	struct pvfs2_inode_s *pvfs2_inode = NULL;
+
+	/* Make sure that we have sane parameters */
+	if (!data || !inode)
+		return 0;
+	pvfs2_inode = PVFS2_I(inode);
+	if (!pvfs2_inode)
+		return 0;
+	pvfs2_inode->refn.fs_id = ref->fs_id;
+	pvfs2_inode->refn.khandle = ref->khandle;
+	return 0;
+}
+
+/*
+ * Called to determine if handles match.
+ */
+static int pvfs2_test_inode(struct inode *inode, void *data)
+{
+	struct pvfs2_object_kref *ref = (struct pvfs2_object_kref *) data;
+	struct pvfs2_inode_s *pvfs2_inode = NULL;
+
+	pvfs2_inode = PVFS2_I(inode);
+	return (!PVFS_khandle_cmp(&(pvfs2_inode->refn.khandle), &(ref->khandle))
+		&& pvfs2_inode->refn.fs_id == ref->fs_id);
+}
+
+/*
+ * Front-end to lookup the inode-cache maintained by the VFS using the PVFS2
+ * file handle.
+ *
+ * @sb: the file system super block instance.
+ * @ref: The PVFS2 object for which we are trying to locate an inode structure.
+ */
+struct inode *pvfs2_iget(struct super_block *sb, struct pvfs2_object_kref *ref)
+{
+	struct inode *inode = NULL;
+	unsigned long hash;
+	int error;
+
+	hash = pvfs2_handle_hash(ref);
+	inode = iget5_locked(sb, hash, pvfs2_test_inode, pvfs2_set_inode, ref);
+	if (!inode || !(inode->i_state & I_NEW))
+		return inode;
+
+	error = pvfs2_inode_getattr(inode, PVFS_ATTR_SYS_ALL_NOHINT);
+	if (error) {
+		iget_failed(inode);
+		return ERR_PTR(error);
+	}
+
+	inode->i_ino = hash;	/* needed for stat etc */
+	pvfs2_init_iops(inode);
+	unlock_new_inode(inode);
+
+	gossip_debug(GOSSIP_INODE_DEBUG,
+		     "iget handle %pU, fsid %d hash %ld i_ino %lu\n",
+		     &ref->khandle,
+		     ref->fs_id,
+		     hash,
+		     inode->i_ino);
+
+	return inode;
+}
+
+/*
+ * Allocate an inode for a newly created file and insert it into the inode hash.
+ */
+struct inode *pvfs2_new_inode(struct super_block *sb, struct inode *dir,
+		int mode, dev_t dev, struct pvfs2_object_kref *ref)
+{
+	unsigned long hash = pvfs2_handle_hash(ref);
+	struct inode *inode;
+	int error;
+
+	gossip_debug(GOSSIP_INODE_DEBUG,
+		     "pvfs2_get_custom_inode_common: called\n"
+		     "(sb is %p | MAJOR(dev)=%u | MINOR(dev)=%u mode=%o)\n",
+		     sb,
+		     MAJOR(dev),
+		     MINOR(dev),
+		     mode);
+
+	inode = new_inode(sb);
+	if (!inode)
+		return NULL;
+
+	pvfs2_set_inode(inode, ref);
+	inode->i_ino = hash;	/* needed for stat etc */
+
+	error = pvfs2_inode_getattr(inode, PVFS_ATTR_SYS_ALL_NOHINT);
+	if (error)
+		goto out_iput;
+
+	pvfs2_init_iops(inode);
+
+	inode->i_mode = mode;
+	inode->i_uid = current_fsuid();
+	inode->i_gid = current_fsgid();
+	inode->i_atime = inode->i_mtime = inode->i_ctime = CURRENT_TIME;
+	inode->i_size = PAGE_CACHE_SIZE;
+	inode->i_rdev = dev;
+
+	error = insert_inode_locked4(inode, hash, pvfs2_test_inode, ref);
+	if (error < 0)
+		goto out_iput;
+
+	gossip_debug(GOSSIP_INODE_DEBUG,
+		     "Initializing ACL's for inode %pU\n",
+		     get_khandle_from_ino(inode));
+	pvfs2_init_acl(inode, dir);
+	return inode;
+
+out_iput:
+	iput(inode);
+	return ERR_PTR(error);
+}