Added all files

21 files changed, 7130 insertions, 0 deletions

diff --git a/xlators/features/Makefile.am b/xlators/features/Makefile.am
new file mode 100644
index 00000000000..9ac9b6f19de
--- /dev/null
+++ b/xlators/features/Makefile.am
@@ -0,0 +1,3 @@
+SUBDIRS = locks trash path-convertor filter quota
+
+CLEANFILES = 
diff --git a/xlators/features/filter/Makefile.am b/xlators/features/filter/Makefile.am
new file mode 100644
index 00000000000..d471a3f9243
--- /dev/null
+++ b/xlators/features/filter/Makefile.am
@@ -0,0 +1,3 @@
+SUBDIRS = src
+
+CLEANFILES = 
diff --git a/xlators/features/filter/src/Makefile.am b/xlators/features/filter/src/Makefile.am
new file mode 100644
index 00000000000..fa0b92214a9
--- /dev/null
+++ b/xlators/features/filter/src/Makefile.am
@@ -0,0 +1,13 @@
+xlator_LTLIBRARIES = filter.la
+xlatordir = $(libdir)/glusterfs/$(PACKAGE_VERSION)/xlator/features
+
+filter_la_LDFLAGS = -module -avoidversion
+
+filter_la_SOURCES = filter.c
+filter_la_LIBADD = $(top_builddir)/libglusterfs/src/libglusterfs.la 
+
+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/features/filter/src/filter.c b/xlators/features/filter/src/filter.c
new file mode 100644
index 00000000000..67ea45d3a23
--- /dev/null
+++ b/xlators/features/filter/src/filter.c
@@ -0,0 +1,1768 @@
+/*
+  Copyright (c) 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 _CONFIG_H
+#define _CONFIG_H
+#include "config.h"
+#endif
+
+#include <sys/types.h>
+#include <sys/stat.h>
+#include <unistd.h>
+
+#include "glusterfs.h"
+#include "logging.h"
+#include "dict.h"
+#include "xlator.h"
+
+#define GF_FILTER_NOBODY_UID         65534
+#define GF_FILTER_NOBODY_GID         65534
+#define GF_FILTER_ROOT_UID           0
+#define GF_FILTER_ROOT_GID           0
+
+#define GF_MAXIMUM_FILTERING_ALLOWED 32
+
+/*
+  option root-filtering on (off by default)
+  option translate-uid <uid-range=newuid,uid=newuid>
+  option translate-gid <gid-range=newgid,gid=newgid>
+  option read-only <yes|true>
+  option fixed-uid <uid>
+  option fixed-gid <gid>
+  option filter-uid <uid-range,uid>
+  option filter-gid <gid-range,gid> // not supported yet
+
+*/
+
+struct gf_filter {
+	/* Flags */
+	gf_boolean_t complete_read_only;
+	char fixed_uid_set;
+	char fixed_gid_set;
+	char partial_filter;
+
+	/* Options */
+	/* Mapping/Filtering/Translate whatever you want to call */
+	int translate_num_uid_entries;
+	int translate_num_gid_entries;
+	int translate_input_uid[GF_MAXIMUM_FILTERING_ALLOWED][2];
+	int translate_output_uid[GF_MAXIMUM_FILTERING_ALLOWED];
+	int translate_input_gid[GF_MAXIMUM_FILTERING_ALLOWED][2];
+	int translate_output_gid[GF_MAXIMUM_FILTERING_ALLOWED];
+
+	/* Fixed uid/gid */
+	int fixed_uid;
+	int fixed_gid;
+
+	/* Filter */
+	int filter_num_uid_entries;
+	int filter_num_gid_entries;
+	int filter_input_uid[GF_MAXIMUM_FILTERING_ALLOWED][2];
+	int filter_input_gid[GF_MAXIMUM_FILTERING_ALLOWED][2];
+	
+};
+
+/* update_frame: The main logic of the whole translator.
+   Return values:
+   0: no change
+   // TRANSLATE
+   1: only uid changed 
+   2: only gid changed
+   3: both uid/gid changed
+   // FILTER
+   4: uid in filter range
+   5: gid in filter range  // not supported yet
+   6: complete fs is readonly
+*/
+
+#define GF_FILTER_NO_CHANGE    0
+#define	GF_FILTER_MAP_UID      1
+#define GF_FILTER_MAP_GID      2
+#define GF_FILTER_MAP_BOTH     3
+#define GF_FILTER_FILTER_UID   4
+#define GF_FILTER_FILTER_GID   5
+#define GF_FILTER_RO_FS        6
+
+static int32_t
+update_frame (call_frame_t *frame,
+	      inode_t *inode,
+	      struct gf_filter *filter)
+{
+	uid_t    uid = 0;
+	int32_t  idx = 0;
+	int32_t  ret = 0;
+	int32_t  dictret = 0;
+	uint64_t tmp_uid = 0;
+	
+	for (idx = 0; idx < filter->translate_num_uid_entries; idx++) {
+		if ((frame->root->uid >=filter->translate_input_uid[idx][0]) &&
+		    (frame->root->uid <=filter->translate_input_uid[idx][1])) {
+			dictret = inode_ctx_get (inode, frame->this, &tmp_uid);
+			uid = (uid_t)tmp_uid;
+			if (dictret == 0) {
+				if (frame->root->uid != uid)
+					ret = GF_FILTER_MAP_UID;
+			} else {
+				ret = GF_FILTER_MAP_UID;
+			}
+			break;
+		}
+	}
+	
+	for (idx = 0; idx < filter->translate_num_gid_entries; idx++) {
+		if ((frame->root->gid >=filter->translate_input_gid[idx][0]) &&
+		    (frame->root->gid <=filter->translate_input_gid[idx][1])) {
+			if (ret == GF_FILTER_NO_CHANGE) 
+				ret = GF_FILTER_MAP_GID;
+			else 
+				ret = GF_FILTER_MAP_BOTH;
+			break;
+		}
+	}
+
+
+	if (filter->complete_read_only)
+		return GF_FILTER_RO_FS;
+	
+	if (filter->partial_filter) {
+		dictret = inode_ctx_get (inode, frame->this, &tmp_uid);
+		uid = (uid_t)tmp_uid;
+		if (dictret != -1) {
+			for (idx = 0; idx < filter->filter_num_uid_entries; 
+			     idx++) {
+				if ((uid >=filter->filter_input_uid[idx][0]) &&
+				    (uid <=filter->filter_input_uid[idx][1])) {
+					return GF_FILTER_FILTER_UID;
+				}
+			}
+		}
+	}
+
+	return ret;
+}
+
+/* if 'root' don't change the uid/gid */
+static int32_t
+update_stat (struct stat *stbuf,
+	     struct gf_filter *filter)
+{
+	int32_t idx = 0;
+	for (idx = 0; idx < filter->translate_num_uid_entries; idx++) {
+		if (stbuf->st_uid == GF_FILTER_ROOT_UID)
+			continue;
+		if ((stbuf->st_uid >= filter->translate_input_uid[idx][0]) &&
+		    (stbuf->st_uid <= filter->translate_input_uid[idx][1])) {
+			stbuf->st_uid = filter->translate_output_uid[idx];
+			break;
+		}
+	}
+	
+	for (idx = 0; idx < filter->translate_num_gid_entries; idx++) {
+		if (stbuf->st_gid == GF_FILTER_ROOT_GID)
+			continue;
+		if ((stbuf->st_gid >= filter->translate_input_gid[idx][0]) &&
+		    (stbuf->st_gid <= filter->translate_input_gid[idx][1])) {
+			stbuf->st_gid = filter->translate_output_gid[idx];
+			break;
+		}
+	}
+
+	if (filter->fixed_uid_set) {
+		stbuf->st_uid = filter->fixed_uid;
+	}
+
+	if (filter->fixed_gid_set) {
+		stbuf->st_gid = filter->fixed_gid;
+	}
+	
+	return 0;
+}
+
+static int32_t 
+filter_lookup_cbk (call_frame_t *frame,
+		   void *cookie,
+		   xlator_t *this,
+		   int32_t op_ret,
+		   int32_t op_errno,
+		   inode_t *inode,
+		   struct stat *buf,
+		   dict_t *dict)
+{
+	int ret = 0;
+	if (op_ret >= 0) {
+		update_stat (buf, this->private);
+		ret = inode_ctx_put (inode, this, (uint64_t)(long)buf->st_uid);
+		if (ret == -1) {
+			gf_log (this->name, GF_LOG_ERROR,
+				"couldn't set context");
+		}
+	}
+	STACK_UNWIND (frame, op_ret, op_errno, inode, buf, dict);
+	return 0;
+}
+
+int32_t 
+filter_lookup (call_frame_t *frame,
+	       xlator_t *this,
+	       loc_t *loc,
+	       dict_t *xattr_req)
+{
+	STACK_WIND (frame,
+		    filter_lookup_cbk,
+		    FIRST_CHILD(this),
+		    FIRST_CHILD(this)->fops->lookup,
+		    loc,
+		    xattr_req);
+	return 0;
+}
+
+
+static int32_t
+filter_stat_cbk (call_frame_t *frame,
+		 void *cookie,
+		 xlator_t *this,
+		 int32_t op_ret,
+		 int32_t op_errno,
+		 struct stat *buf)
+{
+	if (op_ret >= 0) {
+		update_stat (buf, this->private);
+	}
+	STACK_UNWIND (frame, op_ret, op_errno, buf);
+	return 0;
+}
+
+int32_t
+filter_stat (call_frame_t *frame,
+	     xlator_t *this,
+	     loc_t *loc)
+{
+	STACK_WIND (frame,
+		    filter_stat_cbk,
+		    FIRST_CHILD(this),
+		    FIRST_CHILD(this)->fops->stat,
+		    loc);
+	return 0;
+}
+
+static int32_t
+filter_chmod_cbk (call_frame_t *frame,
+		  void *cookie,
+		  xlator_t *this,
+		  int32_t op_ret,
+		  int32_t op_errno,
+		  struct stat *buf)
+{
+	if (op_ret >= 0) {
+		update_stat (buf, this->private);
+	}
+	STACK_UNWIND (frame, op_ret, op_errno, buf);
+	return 0;
+}
+
+int32_t
+filter_chmod (call_frame_t *frame,
+	      xlator_t *this,
+	      loc_t *loc,
+	      mode_t mode)
+{
+	int32_t ret = 0;
+	ret = update_frame (frame, loc->inode, this->private);
+	switch (ret) {
+	case GF_FILTER_MAP_UID:
+		if (loc->inode->st_mode & S_IWGRP)
+			break;
+	case GF_FILTER_MAP_BOTH:
+		if (loc->inode->st_mode & S_IWOTH)
+			break;
+		gf_log (this->name, GF_LOG_DEBUG, "%s: returning permission denied", loc->path);
+		STACK_UNWIND (frame, -1, EPERM, NULL);
+		return 0;
+		
+	case GF_FILTER_FILTER_UID:
+	case GF_FILTER_FILTER_GID:
+	case GF_FILTER_RO_FS:
+		STACK_UNWIND (frame, -1, EROFS, NULL);
+		return 0;
+	default:
+		break;
+	}
+
+	STACK_WIND (frame,
+		    filter_chmod_cbk,
+		    FIRST_CHILD(this),
+		    FIRST_CHILD(this)->fops->chmod,
+		    loc,
+		    mode);
+	return 0;
+}
+
+
+static int32_t
+filter_fchmod_cbk (call_frame_t *frame,
+		   void *cookie,
+		   xlator_t *this,
+		   int32_t op_ret,
+		   int32_t op_errno,
+		   struct stat *buf)
+{
+	if (op_ret >= 0) {
+		update_stat (buf, this->private);
+	}
+	STACK_UNWIND (frame,
+		      op_ret,
+		      op_errno,
+		      buf);
+	return 0;
+}
+
+int32_t 
+filter_fchmod (call_frame_t *frame,
+	       xlator_t *this,
+	       fd_t *fd,
+	       mode_t mode)
+{
+	STACK_WIND (frame,
+		    filter_fchmod_cbk,
+		    FIRST_CHILD(this),
+		    FIRST_CHILD(this)->fops->fchmod,
+		    fd,
+		    mode);
+	return 0;
+}
+
+static int32_t
+filter_chown_cbk (call_frame_t *frame,
+		  void *cookie,
+		  xlator_t *this,
+		  int32_t op_ret,
+		  int32_t op_errno,
+		  struct stat *buf)
+{
+	if (op_ret >= 0) {
+		update_stat (buf, this->private);
+	}
+	STACK_UNWIND (frame, op_ret, op_errno, buf);
+	return 0;
+}
+
+int32_t
+filter_chown (call_frame_t *frame,
+	      xlator_t *this,
+	      loc_t *loc,
+	      uid_t uid,
+	      gid_t gid)
+{
+	int32_t ret = 0;
+	ret = update_frame (frame, loc->inode, this->private);
+	switch (ret) {
+	case GF_FILTER_MAP_UID:
+		if (loc->inode->st_mode & S_IWGRP)
+			break;
+	case GF_FILTER_MAP_BOTH:
+		if (loc->inode->st_mode & S_IWOTH)
+			break;
+		gf_log (this->name, GF_LOG_DEBUG, "%s: returning permission denied", loc->path);
+		STACK_UNWIND (frame, -1, EPERM, NULL);
+		return 0;
+		
+	case GF_FILTER_FILTER_UID:
+	case GF_FILTER_FILTER_GID:
+	case GF_FILTER_RO_FS:
+		STACK_UNWIND (frame, -1, EROFS, NULL);
+		return 0;
+	default:
+		break;
+	}			
+
+	STACK_WIND (frame,	      
+		    filter_chown_cbk,
+		    FIRST_CHILD(this),
+		    FIRST_CHILD(this)->fops->chown,
+		    loc,
+		    uid,
+		    gid);
+	return 0;
+}
+
+static int32_t
+filter_fchown_cbk (call_frame_t *frame,
+		   void *cookie,
+		   xlator_t *this,
+		   int32_t op_ret,
+		   int32_t op_errno,
+		   struct stat *buf)
+{
+	if (op_ret >= 0) {
+		update_stat (buf, this->private);
+	}
+	STACK_UNWIND (frame, op_ret, op_errno, buf);
+	return 0;
+}
+
+int32_t 
+filter_fchown (call_frame_t *frame,
+	       xlator_t *this,
+	       fd_t *fd,
+	       uid_t uid,
+	       gid_t gid)
+{
+	STACK_WIND (frame,	      
+		    filter_fchown_cbk,
+		    FIRST_CHILD(this),
+		    FIRST_CHILD(this)->fops->fchown,
+		    fd,
+		    uid,
+		    gid);
+	return 0;
+}
+
+static int32_t
+filter_truncate_cbk (call_frame_t *frame,
+		     void *cookie,
+		     xlator_t *this,
+		     int32_t op_ret,
+		     int32_t op_errno,
+		     struct stat *buf)
+{
+	if (op_ret >= 0) {
+		update_stat (buf, this->private);
+	}
+	STACK_UNWIND (frame, op_ret, op_errno, buf);
+	return 0;
+}
+
+int32_t
+filter_truncate (call_frame_t *frame,
+		 xlator_t *this,
+		 loc_t *loc,
+		 off_t offset)
+{
+	int32_t ret = 0;
+	ret = update_frame (frame, loc->inode, this->private);
+	switch (ret) {
+	case GF_FILTER_MAP_UID:
+		if (loc->inode->st_mode & S_IWGRP)
+			break;
+	case GF_FILTER_MAP_BOTH:
+		if (loc->inode->st_mode & S_IWOTH)
+			break;
+		gf_log (this->name, GF_LOG_DEBUG, "%s: returning permission denied", loc->path);
+		STACK_UNWIND (frame, -1, EPERM, NULL);
+		return 0;
+		
+	case GF_FILTER_FILTER_UID:
+	case GF_FILTER_FILTER_GID:
+	case GF_FILTER_RO_FS:
+		STACK_UNWIND (frame, -1, EROFS, NULL);
+		return 0;
+	}			
+
+	STACK_WIND (frame,
+		    filter_truncate_cbk,
+		    FIRST_CHILD(this),
+		    FIRST_CHILD(this)->fops->truncate,
+		    loc,
+		    offset);
+	return 0;
+}
+
+static int32_t
+filter_ftruncate_cbk (call_frame_t *frame,
+		      void *cookie,
+		      xlator_t *this,
+		      int32_t op_ret,
+		      int32_t op_errno,
+		      struct stat *buf)
+{
+	if (op_ret >= 0) {
+		update_stat (buf, this->private);
+	}
+	STACK_UNWIND (frame, op_ret, op_errno, buf);
+	return 0;
+}
+
+int32_t
+filter_ftruncate (call_frame_t *frame,
+		  xlator_t *this,
+		  fd_t *fd,
+		  off_t offset)
+{
+	STACK_WIND (frame,
+		    filter_ftruncate_cbk,
+		    FIRST_CHILD(this),
+		    FIRST_CHILD(this)->fops->ftruncate,
+		    fd,
+		    offset);
+	return 0;
+}
+
+int32_t 
+filter_utimens_cbk (call_frame_t *frame,
+		    void *cookie,
+		    xlator_t *this,
+		    int32_t op_ret,
+		    int32_t op_errno,
+		    struct stat *buf)
+{
+	if (op_ret >= 0) {
+		update_stat (buf, this->private);
+	}
+	STACK_UNWIND (frame, op_ret, op_errno, buf);
+	return 0;
+}
+
+
+int32_t 
+filter_utimens (call_frame_t *frame,
+		xlator_t *this,
+		loc_t *loc,
+		struct timespec tv[2])
+{
+	int32_t ret = 0;
+	ret = update_frame (frame, loc->inode, this->private);
+	switch (ret) {
+	case GF_FILTER_MAP_UID:
+		if (loc->inode->st_mode & S_IWGRP)
+			break;
+	case GF_FILTER_MAP_BOTH:
+		if (loc->inode->st_mode & S_IWOTH)
+			break;
+		gf_log (this->name, GF_LOG_DEBUG, "%s: returning permission denied", loc->path);
+		STACK_UNWIND (frame, -1, EPERM, NULL);
+		return 0;
+		
+	case GF_FILTER_FILTER_UID:
+	case GF_FILTER_FILTER_GID:
+	case GF_FILTER_RO_FS:
+		STACK_UNWIND (frame, -1, EROFS, NULL);
+		return 0;
+	}
+
+	STACK_WIND (frame,
+		    filter_utimens_cbk,
+		    FIRST_CHILD(this),
+		    FIRST_CHILD(this)->fops->utimens,
+		    loc,
+		    tv);
+	return 0;
+}
+
+static int32_t
+filter_readlink_cbk (call_frame_t *frame,
+		     void *cookie,
+		     xlator_t *this,
+		     int32_t op_ret,
+		     int32_t op_errno,
+		     const char *path)
+{
+	STACK_UNWIND (frame, op_ret, op_errno, path);
+	return 0;
+}
+
+int32_t
+filter_readlink (call_frame_t *frame,
+		 xlator_t *this,
+		 loc_t *loc,
+		 size_t size)
+{
+	int32_t ret = 0;
+	ret = update_frame (frame, loc->inode, this->private);
+	switch (ret) {
+	case GF_FILTER_MAP_UID:
+		if (loc->inode->st_mode & S_IRGRP)
+			break;
+	case GF_FILTER_MAP_BOTH:
+		if (loc->inode->st_mode & S_IROTH)
+			break;
+		gf_log (this->name, GF_LOG_DEBUG, "%s: returning permission denied", loc->path);
+		STACK_UNWIND (frame, -1, EPERM, NULL);
+		return 0;
+	}			
+	STACK_WIND (frame,
+		    filter_readlink_cbk,
+		    FIRST_CHILD(this),
+		    FIRST_CHILD(this)->fops->readlink,
+		    loc,
+		    size);
+	return 0;
+}
+
+
+static int32_t
+filter_mknod_cbk (call_frame_t *frame,
+		  void *cookie,
+		  xlator_t *this,
+		  int32_t op_ret,
+		  int32_t op_errno,
+		  inode_t *inode,
+		  struct stat *buf)
+{
+	int ret = 0;
+
+	if (op_ret >= 0) {
+		update_stat (buf, this->private);
+		ret = inode_ctx_put (inode, this, (uint64_t)(long)buf->st_uid);
+		if (ret == -1) {
+			gf_log (this->name, GF_LOG_ERROR,
+				"couldn't set context");
+		}
+	}
+	STACK_UNWIND (frame, op_ret, op_errno, inode, buf);
+	return 0;
+}
+
+int32_t
+filter_mknod (call_frame_t *frame,
+	      xlator_t *this,
+	      loc_t *loc,
+	      mode_t mode,
+	      dev_t rdev)
+{
+	int ret = 0;
+	inode_t *parent = loc->parent;
+	ret = update_frame (frame, loc->inode, this->private);
+	switch (ret) {		
+	case GF_FILTER_MAP_UID:
+		if (parent->st_mode & S_IWGRP)
+			break;
+	case GF_FILTER_MAP_BOTH:
+		if (parent->st_mode & S_IWOTH)
+			break;
+		gf_log (this->name, GF_LOG_DEBUG, "%s: returning permission denied", loc->path);
+		STACK_UNWIND (frame, -1, EPERM);
+		return 0;
+
+	case GF_FILTER_FILTER_UID:
+	case GF_FILTER_FILTER_GID:
+	case GF_FILTER_RO_FS:
+		STACK_UNWIND (frame, -1, EROFS, NULL, NULL);
+		return 0;
+	}
+	STACK_WIND (frame,
+		    filter_mknod_cbk,
+		    FIRST_CHILD(this),
+		    FIRST_CHILD(this)->fops->mknod,
+		    loc, mode, rdev);
+	return 0;
+}
+
+static int32_t
+filter_mkdir_cbk (call_frame_t *frame,
+		  void *cookie,
+		  xlator_t *this,
+		  int32_t op_ret,
+		  int32_t op_errno,
+		  inode_t *inode,
+		  struct stat *buf)
+{
+	int ret = 0;
+	if (op_ret >= 0) {
+		update_stat (buf, this->private);
+		ret = inode_ctx_put (inode, this, (uint64_t)(long)buf->st_uid);
+		if (ret == -1) {
+			gf_log (this->name, GF_LOG_ERROR,
+				"couldn't set context");
+		}
+	}
+	STACK_UNWIND (frame, op_ret, op_errno, inode, buf);
+	return 0;
+}
+
+int32_t
+filter_mkdir (call_frame_t *frame,
+	      xlator_t *this,
+	      loc_t *loc,
+	      mode_t mode)
+{
+	int ret = 0;
+	inode_t *parent = loc->parent;
+	ret = update_frame (frame, loc->inode, this->private);
+	switch (ret) {		
+	case GF_FILTER_MAP_UID:
+		if (parent->st_mode & S_IWGRP)
+			break;
+	case GF_FILTER_MAP_BOTH:
+		if (parent->st_mode & S_IWOTH)
+			break;
+		gf_log (this->name, GF_LOG_DEBUG, "%s: returning permission denied", loc->path);
+		STACK_UNWIND (frame, -1, EPERM);
+		return 0;
+
+	case GF_FILTER_FILTER_UID:
+	case GF_FILTER_FILTER_GID:
+	case GF_FILTER_RO_FS:
+		STACK_UNWIND (frame, -1, EROFS, NULL, NULL);
+		return 0;
+	}
+	STACK_WIND (frame,
+		    filter_mkdir_cbk,
+		    FIRST_CHILD(this),
+		    FIRST_CHILD(this)->fops->mkdir,
+		    loc, mode);
+	return 0;
+}
+
+static int32_t
+filter_unlink_cbk (call_frame_t *frame,
+		   void *cookie,
+		   xlator_t *this,
+		   int32_t op_ret,
+		   int32_t op_errno)
+{
+	STACK_UNWIND (frame, op_ret, op_errno);
+	return 0;
+}
+
+int32_t
+filter_unlink (call_frame_t *frame,
+	       xlator_t *this,
+	       loc_t *loc)
+{
+	int32_t ret = 0;
+	inode_t *parent = loc->parent;
+	if (!parent)
+		parent = inode_parent (loc->inode, 0, NULL);
+	ret = update_frame (frame, loc->inode, this->private);
+	switch (ret) {
+	case GF_FILTER_MAP_UID:
+		if (parent->st_mode & S_IWGRP)
+			break;
+		if (loc->inode->st_mode & S_IWGRP)
+			break;
+	case GF_FILTER_MAP_BOTH:
+		if (parent->st_mode & S_IWOTH)
+			break;
+		if (loc->inode->st_mode & S_IWOTH)
+			break;
+		gf_log (this->name, GF_LOG_DEBUG, "%s: returning permission denied", loc->path);
+		STACK_UNWIND (frame, -1, EPERM);
+		return 0;
+	case GF_FILTER_FILTER_UID:
+	case GF_FILTER_FILTER_GID:
+	case GF_FILTER_RO_FS:
+		STACK_UNWIND (frame, -1, EROFS);
+		return 0;
+	}
+	STACK_WIND (frame,
+		    filter_unlink_cbk,
+		    FIRST_CHILD(this),
+		    FIRST_CHILD(this)->fops->unlink,
+		    loc);
+	return 0;
+}
+
+static int32_t
+filter_rmdir_cbk (call_frame_t *frame,
+		  void *cookie,
+		  xlator_t *this,
+		  int32_t op_ret,
+		  int32_t op_errno)
+{
+	STACK_UNWIND (frame, op_ret, op_errno);
+	return 0;
+}
+
+int32_t
+filter_rmdir (call_frame_t *frame,
+	      xlator_t *this,
+	      loc_t *loc)
+{
+	int32_t ret = 0;
+	inode_t *parent = loc->parent;
+	if (!parent)
+		parent = inode_parent (loc->inode, 0, NULL);
+	ret = update_frame (frame, loc->inode, this->private);
+	switch (ret) {
+	case GF_FILTER_MAP_UID:
+		if (parent->st_mode & S_IWGRP)
+			break;
+		if (loc->inode->st_mode & S_IWGRP)
+			break;
+	case GF_FILTER_MAP_BOTH:
+		if (parent->st_mode & S_IWOTH)
+			break;
+		if (loc->inode->st_mode & S_IWOTH)
+			break;
+		gf_log (this->name, GF_LOG_DEBUG, "%s: returning permission denied", loc->path);
+		STACK_UNWIND (frame, -1, EPERM);
+		return 0;
+	case GF_FILTER_FILTER_UID:
+	case GF_FILTER_FILTER_GID:
+	case GF_FILTER_RO_FS:
+			STACK_UNWIND (frame, -1, EROFS);
+			return 0;
+	}			
+	STACK_WIND (frame,
+		    filter_rmdir_cbk,
+		    FIRST_CHILD(this),
+		    FIRST_CHILD(this)->fops->rmdir,
+		    loc);
+	return 0;
+}
+
+static int32_t
+filter_symlink_cbk (call_frame_t *frame,
+		    void *cookie,
+		    xlator_t *this,
+		    int32_t op_ret,
+		    int32_t op_errno,
+		    inode_t *inode,
+		    struct stat *buf)
+{
+	int ret = 0;
+	if (op_ret >= 0) {
+		update_stat (buf, this->private);
+		ret = inode_ctx_put (inode, this, (uint64_t)(long)buf->st_uid);
+		if (ret == -1) {
+			gf_log (this->name, GF_LOG_ERROR,
+				"couldn't set context");
+		}
+	}
+	STACK_UNWIND (frame, op_ret, op_errno, inode, buf);
+	return 0;
+}
+
+int32_t
+filter_symlink (call_frame_t *frame,
+		xlator_t *this,
+		const char *linkpath,
+		loc_t *loc)
+{
+	int ret = 0;
+	inode_t *parent = loc->parent;
+	ret = update_frame (frame, loc->inode, this->private);
+	switch (ret) {		
+	case GF_FILTER_MAP_UID:
+		if (parent->st_mode & S_IWGRP)
+			break;
+	case GF_FILTER_MAP_BOTH:
+		if (parent->st_mode & S_IWOTH)
+			break;
+		gf_log (this->name, GF_LOG_DEBUG, "%s: returning permission denied", loc->path);
+		STACK_UNWIND (frame, -1, EPERM);
+		return 0;
+
+	case GF_FILTER_FILTER_UID:
+	case GF_FILTER_FILTER_GID:
+	case GF_FILTER_RO_FS:
+		STACK_UNWIND (frame, -1, EROFS, NULL, NULL);
+		return 0;
+	}
+	STACK_WIND (frame,
+		    filter_symlink_cbk,
+		    FIRST_CHILD(this),
+		    FIRST_CHILD(this)->fops->symlink,
+		    linkpath, loc);
+	return 0;
+}
+
+
+static int32_t
+filter_rename_cbk (call_frame_t *frame,
+		   void *cookie,
+		   xlator_t *this,
+		   int32_t op_ret,
+		   int32_t op_errno,
+		   struct stat *buf)
+{
+	if (op_ret >= 0) {
+		update_stat (buf, this->private);
+	}
+	STACK_UNWIND (frame, op_ret, op_errno, buf);
+	return 0;
+}
+
+int32_t
+filter_rename (call_frame_t *frame,
+	       xlator_t *this,
+	       loc_t *oldloc,
+	       loc_t *newloc)
+{
+	int32_t ret = 0;
+	inode_t *parent = oldloc->parent;
+	if (!parent)
+		parent = inode_parent (oldloc->inode, 0, NULL);
+	ret = update_frame (frame, oldloc->inode, this->private);
+	switch (ret) {		
+	case GF_FILTER_MAP_UID:
+		if (parent->st_mode & S_IWGRP)
+			break;
+		if (oldloc->inode->st_mode & S_IWGRP)
+			break;
+	case GF_FILTER_MAP_BOTH:
+		if (parent->st_mode & S_IWOTH)
+			break;
+		if (oldloc->inode->st_mode & S_IWOTH)
+			break;
+		gf_log (this->name, GF_LOG_DEBUG, 
+			"%s -> %s: returning permission denied", oldloc->path, newloc->path);
+		STACK_UNWIND (frame, -1, EPERM);
+		return 0;
+
+	case GF_FILTER_FILTER_UID:
+	case GF_FILTER_FILTER_GID:
+	case GF_FILTER_RO_FS:
+		STACK_UNWIND (frame, -1, EROFS, NULL);
+		return 0;
+	}
+	STACK_WIND (frame,
+		    filter_rename_cbk,
+		    FIRST_CHILD(this),
+		    FIRST_CHILD(this)->fops->rename,
+		    oldloc, newloc);
+	return 0;
+}
+
+
+static int32_t
+filter_link_cbk (call_frame_t *frame,
+		 void *cookie,
+		 xlator_t *this,
+		 int32_t op_ret,
+		 int32_t op_errno,
+		 inode_t *inode,
+		 struct stat *buf)
+{
+	int ret = 0;
+	if (op_ret >= 0) {
+		update_stat (buf, this->private);
+		ret = inode_ctx_put (inode, this, (uint64_t)(long)buf->st_uid);
+		if (ret == -1) {
+			gf_log (this->name, GF_LOG_ERROR,
+				"couldn't set context");
+		}
+	}
+	STACK_UNWIND (frame, op_ret, op_errno, inode,	buf);
+	return 0;
+}
+
+int32_t
+filter_link (call_frame_t *frame,
+	     xlator_t *this,
+	     loc_t *oldloc,
+	     loc_t *newloc)
+{
+	int ret = 0;
+	ret = update_frame (frame, oldloc->inode, this->private);
+	switch (ret) {
+	case GF_FILTER_FILTER_UID:
+	case GF_FILTER_FILTER_GID:
+	case GF_FILTER_RO_FS:
+		STACK_UNWIND (frame, -1, EROFS, NULL, NULL);
+		return 0;
+	}
+	STACK_WIND (frame,
+		    filter_link_cbk,
+		    FIRST_CHILD(this),
+		    FIRST_CHILD(this)->fops->link,
+		    oldloc, newloc);
+	return 0;
+}
+
+
+static int32_t
+filter_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)
+{
+	int ret = 0;
+	if (op_ret >= 0) {
+		update_stat (buf, this->private);
+		ret = inode_ctx_put (inode, this, (uint64_t)(long)buf->st_uid);
+		if (ret == -1) {
+			gf_log (this->name, GF_LOG_ERROR,
+				"couldn't set context");
+		}
+	}
+	STACK_UNWIND (frame, op_ret, op_errno, fd, inode, buf);
+	return 0;
+}
+
+int32_t
+filter_create (call_frame_t *frame,
+	       xlator_t *this,
+	       loc_t *loc,
+	       int32_t flags,
+	       mode_t mode, fd_t *fd)
+{
+	int ret = 0;
+	inode_t *parent = loc->parent;
+	ret = update_frame (frame, loc->inode, this->private);
+	switch (ret) {		
+	case GF_FILTER_MAP_UID:
+		if (parent->st_mode & S_IWGRP)
+			break;
+	case GF_FILTER_MAP_BOTH:
+		if (parent->st_mode & S_IWOTH)
+			break;
+		gf_log (this->name, GF_LOG_DEBUG, "%s: returning permission denied", loc->path);
+		STACK_UNWIND (frame, -1, EPERM);
+		return 0;
+
+	case GF_FILTER_FILTER_UID:
+	case GF_FILTER_FILTER_GID:
+	case GF_FILTER_RO_FS:
+		STACK_UNWIND (frame, -1, EROFS, NULL, NULL, NULL);
+		return 0;
+	}
+	STACK_WIND (frame, filter_create_cbk,
+		    FIRST_CHILD(this),
+		    FIRST_CHILD(this)->fops->create,
+		    loc, flags, mode, fd);
+	return 0;
+}
+
+static int32_t
+filter_open_cbk (call_frame_t *frame,
+		 void *cookie,
+		 xlator_t *this,
+		 int32_t op_ret,
+		 int32_t op_errno,
+		 fd_t *fd)
+{
+	STACK_UNWIND (frame, op_ret, op_errno, fd);
+	return 0;
+}
+
+int32_t
+filter_open (call_frame_t *frame,
+	     xlator_t *this,
+	     loc_t *loc,
+	     int32_t flags, 
+	     fd_t *fd)
+{
+	int32_t ret = 0;
+	ret = update_frame (frame, loc->inode, this->private);
+	switch (ret) {
+	case GF_FILTER_MAP_UID:
+		if (loc->inode->st_mode & S_IWGRP)
+			break;
+		if (!((flags & O_WRONLY) || (flags & O_RDWR)) 
+		    && (loc->inode->st_mode & S_IRGRP))
+			break;
+	case GF_FILTER_MAP_BOTH:
+		if (loc->inode->st_mode & S_IWOTH)
+				break;
+		if (!((flags & O_WRONLY) || (flags & O_RDWR))
+		    && (loc->inode->st_mode & S_IROTH))
+			break;
+		gf_log (this->name, GF_LOG_DEBUG, 
+			"%s: returning permission denied (mode: 0%o, flag=0%o)", 
+			loc->path, loc->inode->st_mode, flags);
+		STACK_UNWIND (frame, -1, EPERM, fd);
+		return 0;
+	case GF_FILTER_FILTER_UID:
+	case GF_FILTER_FILTER_GID:
+	case GF_FILTER_RO_FS:
+		if (!((flags & O_WRONLY) || (flags & O_RDWR)))
+			break;
+		STACK_UNWIND (frame, -1, EROFS, NULL);
+		return 0;
+		
+	}
+	STACK_WIND (frame,
+		    filter_open_cbk,
+		    FIRST_CHILD(this),
+		    FIRST_CHILD(this)->fops->open,
+		    loc, flags, fd);
+	return 0;
+}
+
+static int32_t
+filter_readv_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)
+{
+	if (op_ret >= 0) {
+		update_stat (stbuf, this->private);
+	}
+	STACK_UNWIND (frame,
+		      op_ret,
+		      op_errno,
+		      vector,
+		      count,
+		      stbuf);
+	return 0;
+}
+
+int32_t
+filter_readv (call_frame_t *frame,
+	      xlator_t *this,
+	      fd_t *fd,
+	      size_t size,
+	      off_t offset)
+{
+	STACK_WIND (frame,
+		    filter_readv_cbk,
+		    FIRST_CHILD(this),
+		    FIRST_CHILD(this)->fops->readv,
+		    fd,
+		    size,
+		    offset);
+	return 0;
+}
+
+
+static int32_t
+filter_writev_cbk (call_frame_t *frame,
+		   void *cookie,
+		   xlator_t *this,
+		   int32_t op_ret,
+		   int32_t op_errno,
+		   struct stat *stbuf)
+{
+	if (op_ret >= 0) {
+		update_stat (stbuf, this->private);
+	}
+	STACK_UNWIND (frame,
+		      op_ret,
+		      op_errno,
+		      stbuf);
+	return 0;
+}
+
+int32_t
+filter_writev (call_frame_t *frame,
+	       xlator_t *this,
+	       fd_t *fd,
+	       struct iovec *vector,
+	       int32_t count,
+	       off_t off)
+{
+	int32_t ret = 0;
+	ret = update_frame (frame, fd->inode, this->private);
+	switch (ret) {
+	case GF_FILTER_FILTER_UID:
+	case GF_FILTER_FILTER_GID:
+	case GF_FILTER_RO_FS:
+		STACK_UNWIND (frame, -1, EROFS, NULL);
+		return 0;
+	}
+
+	STACK_WIND (frame,
+		    filter_writev_cbk,
+		    FIRST_CHILD(this),
+		    FIRST_CHILD(this)->fops->writev,
+		    fd,
+		    vector,
+		    count,
+		    off);
+	return 0;
+}
+
+static int32_t
+filter_fstat_cbk (call_frame_t *frame,
+		  void *cookie,
+		  xlator_t *this,
+		  int32_t op_ret,
+		  int32_t op_errno,
+		  struct stat *buf)
+{
+	if (op_ret >= 0) {
+		update_stat (buf, this->private);
+	}
+	STACK_UNWIND (frame,
+		      op_ret,
+		      op_errno,
+		      buf);
+	return 0;
+}
+
+int32_t
+filter_fstat (call_frame_t *frame,
+	      xlator_t *this,
+	      fd_t *fd)
+{
+	STACK_WIND (frame,
+		    filter_fstat_cbk,
+		    FIRST_CHILD(this),
+		    FIRST_CHILD(this)->fops->fstat,
+		    fd);
+	return 0;
+}
+
+static int32_t
+filter_opendir_cbk (call_frame_t *frame,
+		    void *cookie,
+		    xlator_t *this,
+		    int32_t op_ret,
+		    int32_t op_errno,
+		    fd_t *fd)
+{
+	STACK_UNWIND (frame,
+		      op_ret,
+		      op_errno,
+		      fd);
+	return 0;
+}
+
+int32_t
+filter_opendir (call_frame_t *frame,
+		xlator_t *this,
+		loc_t *loc, fd_t *fd)
+{
+	int32_t ret = 0;
+	ret = update_frame (frame, loc->inode, this->private);
+	switch (ret) {
+	case GF_FILTER_MAP_UID:
+		if (loc->inode->st_mode & S_IWGRP)
+			break;
+		if (loc->inode->st_mode & S_IRGRP)
+			break;
+	case GF_FILTER_MAP_BOTH:
+		if (loc->inode->st_mode & S_IWOTH)
+			break;
+		if (loc->inode->st_mode & S_IROTH)
+			break;
+		gf_log (this->name, GF_LOG_DEBUG, "%s: returning permission denied", loc->path);
+		STACK_UNWIND (frame, -1, EPERM, fd);
+		return 0;
+	}			
+	STACK_WIND (frame,
+		    filter_opendir_cbk,
+		    FIRST_CHILD(this),
+		    FIRST_CHILD(this)->fops->opendir,
+		    loc, fd);
+	return 0;
+}
+
+
+static int32_t
+filter_setxattr_cbk (call_frame_t *frame,
+		     void *cookie,
+		     xlator_t *this,
+		     int32_t op_ret,
+		     int32_t op_errno)
+{
+	STACK_UNWIND (frame,
+		      op_ret,
+		      op_errno);
+	return 0;
+}
+
+int32_t
+filter_setxattr (call_frame_t *frame,
+		 xlator_t *this,
+		 loc_t *loc,
+		 dict_t *dict,
+		 int32_t flags)
+{
+
+	int32_t ret = 0;
+	ret = update_frame (frame, loc->inode, this->private);
+	switch (ret) {
+	case GF_FILTER_MAP_UID:
+		if (loc->inode->st_mode & S_IWGRP)
+			break;
+	case GF_FILTER_MAP_BOTH:
+		if (loc->inode->st_mode & S_IWOTH)
+			break;
+		gf_log (this->name, GF_LOG_DEBUG, "%s: returning permission denied", loc->path);
+		STACK_UNWIND (frame, -1, EPERM);
+		return 0;
+	case GF_FILTER_FILTER_UID:
+	case GF_FILTER_FILTER_GID:
+	case GF_FILTER_RO_FS:
+		STACK_UNWIND (frame, -1, EROFS);
+		return 0;
+	}			
+
+	STACK_WIND (frame,
+		    filter_setxattr_cbk,
+		    FIRST_CHILD(this),
+		    FIRST_CHILD(this)->fops->setxattr,
+		    loc,
+		    dict,
+		    flags);
+	return 0;
+}
+
+static int32_t
+filter_getxattr_cbk (call_frame_t *frame,
+		     void *cookie,
+		     xlator_t *this,
+		     int32_t op_ret,
+		     int32_t op_errno,
+		     dict_t *dict)
+{
+	STACK_UNWIND (frame,
+		      op_ret,
+		      op_errno,
+		      dict);
+	return 0;
+}
+
+int32_t
+filter_getxattr (call_frame_t *frame,
+		 xlator_t *this,
+		 loc_t *loc,
+		 const char *name)
+{
+	int32_t ret = 0;
+	ret = update_frame (frame, loc->inode, this->private);
+	switch (ret) {
+	case GF_FILTER_MAP_UID:
+		if (loc->inode->st_mode & S_IRGRP)
+			break;
+	case GF_FILTER_MAP_BOTH:
+		if (loc->inode->st_mode & S_IROTH)
+			break;
+		gf_log (this->name, GF_LOG_DEBUG, "%s: returning permission denied", loc->path);
+		STACK_UNWIND (frame, -1, EPERM, NULL);
+		return 0;
+	}			
+
+	STACK_WIND (frame,
+		    filter_getxattr_cbk,
+		    FIRST_CHILD(this),
+		    FIRST_CHILD(this)->fops->getxattr,
+		    loc,
+		    name);
+	return 0;
+}
+
+static int32_t
+filter_removexattr_cbk (call_frame_t *frame,
+			void *cookie,
+			xlator_t *this,
+			int32_t op_ret,
+			int32_t op_errno)
+{
+	STACK_UNWIND (frame, op_ret, op_errno);
+	return 0;
+}
+
+int32_t
+filter_removexattr (call_frame_t *frame,
+		    xlator_t *this,
+		    loc_t *loc,
+		    const char *name)
+{
+	int32_t ret = 0;
+	ret = update_frame (frame, loc->inode, this->private);
+	switch (ret) {
+	case GF_FILTER_MAP_UID:
+		if (loc->inode->st_mode & S_IWGRP)
+			break;
+	case GF_FILTER_MAP_BOTH:
+		if (loc->inode->st_mode & S_IWOTH)
+			break;
+		gf_log (this->name, GF_LOG_DEBUG, "%s: returning permission denied", loc->path);
+		STACK_UNWIND (frame, -1, EPERM);
+		return 0;
+	case GF_FILTER_FILTER_UID:
+	case GF_FILTER_FILTER_GID:
+	case GF_FILTER_RO_FS:
+		STACK_UNWIND (frame, -1, EROFS);
+		return 0;
+	}			
+
+	STACK_WIND (frame,
+		    filter_removexattr_cbk,
+		    FIRST_CHILD(this),
+		    FIRST_CHILD(this)->fops->removexattr,
+		    loc,
+		    name);
+	return 0;
+}
+
+int32_t 
+init (xlator_t *this)
+{
+	char *value = NULL;
+	char *tmp_str = NULL;
+	char *tmp_str1 = NULL;
+	char *tmp_str2 = NULL;
+	char *dup_str = NULL;
+	char *input_value_str1 = NULL;
+	char *input_value_str2 = NULL;
+	char *output_value_str = NULL;
+	int32_t input_value = 0;
+	int32_t output_value = 0;
+	data_t *option_data = NULL;
+	struct gf_filter *filter = NULL;
+	gf_boolean_t tmp_bool = 0;
+
+	if (!this->children || this->children->next) {
+		gf_log (this->name,
+			GF_LOG_ERROR,
+			"translator not configured with exactly one child");
+		return -1;
+	}
+	
+	if (!this->parents) {
+		gf_log (this->name, GF_LOG_WARNING,
+			"dangling volume. check volfile ");
+	}
+
+	filter = CALLOC (sizeof (*filter), 1);
+	ERR_ABORT (filter);
+	
+	if (dict_get (this->options, "read-only")) {
+		value = data_to_str (dict_get (this->options, "read-only"));
+		if (gf_string2boolean (value, &filter->complete_read_only) == -1) {
+			gf_log (this->name, GF_LOG_ERROR,
+				"wrong value provided for 'read-only'");
+			return -1;
+		}
+	}
+
+	if (dict_get (this->options, "root-squashing")) {
+		value = data_to_str (dict_get (this->options, "root-squashing"));
+		if (gf_string2boolean (value, &tmp_bool) == -1) {
+			gf_log (this->name, GF_LOG_ERROR,
+				"wrong value provided for 'root-squashing'");
+			return -1;
+		}
+		if (tmp_bool) {
+			filter->translate_num_uid_entries = 1;
+			filter->translate_num_gid_entries = 1;
+			filter->translate_input_uid[0][0] = GF_FILTER_ROOT_UID; /* root */
+			filter->translate_input_uid[0][1] = GF_FILTER_ROOT_UID; /* root */
+			filter->translate_input_gid[0][0] = GF_FILTER_ROOT_GID; /* root */
+			filter->translate_input_gid[0][1] = GF_FILTER_ROOT_GID; /* root */
+			filter->translate_output_uid[0] = GF_FILTER_NOBODY_UID;
+			filter->translate_output_gid[0] = GF_FILTER_NOBODY_GID;
+		}
+	}
+
+	if (dict_get (this->options, "translate-uid")) {
+		option_data = dict_get (this->options, "translate-uid");
+		value = strtok_r (option_data->data, ",", &tmp_str);
+		while (value) {
+			dup_str = strdup (value);
+			input_value_str1 = strtok_r (dup_str, "=", &tmp_str1);
+			if (input_value_str1) {
+				/* Check for n-m */
+				char *temp_string = strdup (input_value_str1);
+				input_value_str2 = strtok_r (temp_string, "-", &tmp_str2);
+				if (gf_string2int (input_value_str2, &input_value) != 0) {
+					gf_log (this->name, GF_LOG_ERROR, 
+						"invalid number format \"%s\"", 
+						input_value_str2);
+					return -1;
+				}
+				filter->translate_input_uid[filter->translate_num_uid_entries][0] = input_value;
+				input_value_str2 = strtok_r (NULL, "-", &tmp_str2);
+				if (input_value_str2) {
+					if (gf_string2int (input_value_str2, &input_value) != 0) {
+						gf_log (this->name, GF_LOG_ERROR, 
+							"invalid number format \"%s\"", 
+							input_value_str2);
+						return -1;
+					}
+				}
+				filter->translate_input_uid[filter->translate_num_uid_entries][1] = input_value;
+				FREE (temp_string);
+				output_value_str = strtok_r (NULL, "=", &tmp_str1);
+				if (output_value_str) {
+					if (gf_string2int (output_value_str, &output_value) != 0) {
+						gf_log (this->name, GF_LOG_ERROR, 
+							"invalid number format \"%s\"", 
+							output_value_str);
+						return -1;
+					}
+				} else {
+					gf_log (this->name, GF_LOG_ERROR, 
+						"mapping string not valid");
+					return -1;
+				}
+			} else {
+				gf_log (this->name, GF_LOG_ERROR, 
+					"mapping string not valid");
+				return -1;
+			}
+			filter->translate_output_uid[filter->translate_num_uid_entries]   = output_value;
+			gf_log (this->name, 
+				GF_LOG_DEBUG, 
+				"pair %d: input uid '%d' will be changed to uid '%d'", 
+				filter->translate_num_uid_entries, input_value, output_value);
+
+			filter->translate_num_uid_entries++;
+			if (filter->translate_num_uid_entries == GF_MAXIMUM_FILTERING_ALLOWED)
+				break;
+			value = strtok_r (NULL, ",", &tmp_str);
+			FREE (dup_str);
+		}
+	}
+
+	tmp_str1 = NULL;
+	tmp_str2 = NULL;
+	tmp_str  = NULL;
+
+	if (dict_get (this->options, "translate-gid")) {
+		option_data = dict_get (this->options, "translate-gid");
+		value = strtok_r (option_data->data, ",", &tmp_str);
+		while (value) {
+			dup_str = strdup (value);
+			input_value_str1 = strtok_r (dup_str, "=", &tmp_str1);
+			if (input_value_str1) {
+				/* Check for n-m */
+				char *temp_string = strdup (input_value_str1);
+				input_value_str2 = strtok_r (temp_string, "-", &tmp_str2);
+				if (gf_string2int (input_value_str2, &input_value) != 0) {
+					gf_log (this->name, GF_LOG_ERROR, 
+						"invalid number format \"%s\"", 
+						input_value_str2);
+					return -1;
+				}
+				filter->translate_input_gid[filter->translate_num_gid_entries][0] = input_value;
+				input_value_str2 = strtok_r (NULL, "-", &tmp_str2);
+				if (input_value_str2) {
+					if (gf_string2int (input_value_str2, &input_value) != 0) {
+						gf_log (this->name, GF_LOG_ERROR, 
+							"invalid number format \"%s\"", 
+							input_value_str2);
+						return -1;
+					}
+				}
+				filter->translate_input_gid[filter->translate_num_gid_entries][1] = input_value;
+				FREE (temp_string);
+				output_value_str = strtok_r (NULL, "=", &tmp_str1);
+				if (output_value_str) {
+					if (gf_string2int (output_value_str, &output_value) != 0) {
+						gf_log (this->name, GF_LOG_ERROR, 
+							"invalid number format \"%s\"", 
+							output_value_str);
+						return -1;
+					}
+				} else {
+					gf_log (this->name, GF_LOG_ERROR, 
+						"translate-gid value not valid");
+					return -1;
+				}
+			} else {
+				gf_log (this->name, GF_LOG_ERROR, 
+					"translate-gid value not valid");
+				return -1;
+			}
+			
+			filter->translate_output_gid[filter->translate_num_gid_entries] = output_value;
+			
+			gf_log (this->name, GF_LOG_DEBUG, 
+				"pair %d: input gid '%d' will be changed to gid '%d'", 
+				filter->translate_num_gid_entries, input_value, output_value);
+			
+			filter->translate_num_gid_entries++;
+			if (filter->translate_num_gid_entries == GF_MAXIMUM_FILTERING_ALLOWED)
+				break;
+			value = strtok_r (NULL, ",", &tmp_str);
+			FREE (dup_str);
+		}
+	}
+
+	tmp_str  = NULL;
+	tmp_str1 = NULL;
+
+	if (dict_get (this->options, "filter-uid")) {
+		option_data = dict_get (this->options, "filter-uid");
+		value = strtok_r (option_data->data, ",", &tmp_str);
+		while (value) {
+			dup_str = strdup (value);
+			/* Check for n-m */
+			input_value_str1 = strtok_r (dup_str, "-", &tmp_str1);
+			if (gf_string2int (input_value_str1, &input_value) != 0) {
+				gf_log (this->name, GF_LOG_ERROR, 
+					"invalid number format \"%s\"", 
+					input_value_str1);
+				return -1;
+			}
+			filter->filter_input_uid[filter->filter_num_uid_entries][0] = input_value;
+			input_value_str1 = strtok_r (NULL, "-", &tmp_str1);
+			if (input_value_str1) {
+				if (gf_string2int (input_value_str1, &input_value) != 0) {
+					gf_log (this->name, GF_LOG_ERROR, 
+						"invalid number format \"%s\"", 
+						input_value_str1);
+					return -1;
+				}
+			}
+			filter->filter_input_uid[filter->filter_num_uid_entries][1] = input_value;
+
+			gf_log (this->name, 
+				GF_LOG_DEBUG, 
+				"filter [%d]: input uid(s) '%s' will be filtered", 
+				filter->filter_num_uid_entries, dup_str);
+			
+			filter->filter_num_uid_entries++;
+			if (filter->filter_num_uid_entries == GF_MAXIMUM_FILTERING_ALLOWED)
+				break;
+			value = strtok_r (NULL, ",", &tmp_str);
+			FREE (dup_str);
+		}
+		filter->partial_filter = 1;
+	}
+
+	tmp_str  = NULL;
+	tmp_str1 = NULL;
+
+	if (dict_get (this->options, "filter-gid")) {
+		option_data = dict_get (this->options, "filter-gid");
+		value = strtok_r (option_data->data, ",", &tmp_str);
+		while (value) {
+			dup_str = strdup (value);
+			/* Check for n-m */
+			input_value_str1 = strtok_r (dup_str, "-", &tmp_str1);
+			if (gf_string2int (input_value_str1, &input_value) != 0) {
+				gf_log (this->name, GF_LOG_ERROR, 
+					"invalid number format \"%s\"", 
+					input_value_str1);
+				return -1;
+			}
+			filter->filter_input_gid[filter->filter_num_gid_entries][0] = input_value;
+			input_value_str1 = strtok_r (NULL, "-", &tmp_str1);
+			if (input_value_str1) {
+				if (gf_string2int (input_value_str1, &input_value) != 0) {
+					gf_log (this->name, GF_LOG_ERROR, 
+						"invalid number format \"%s\"", 
+						input_value_str1);
+					return -1;
+				}
+			}
+			filter->filter_input_gid[filter->filter_num_gid_entries][1] = input_value;
+
+			gf_log (this->name, 
+				GF_LOG_DEBUG, 
+				"filter [%d]: input gid(s) '%s' will be filtered", 
+				filter->filter_num_gid_entries, dup_str);
+			
+			filter->filter_num_gid_entries++;
+			if (filter->filter_num_gid_entries == GF_MAXIMUM_FILTERING_ALLOWED)
+				break;
+			value = strtok_r (NULL, ",", &tmp_str);
+			FREE (dup_str);
+		}
+		gf_log (this->name, GF_LOG_ERROR, "this option is not supported currently.. exiting");
+		return -1;
+		filter->partial_filter = 1;
+	}
+
+	if (dict_get (this->options, "fixed-uid")) {
+		option_data = dict_get (this->options, "fixed-uid");
+		if (gf_string2int (option_data->data, &input_value) != 0) {
+			gf_log (this->name, GF_LOG_ERROR, 
+				"invalid number format \"%s\"", 
+				option_data->data);
+			return -1;
+		}
+		filter->fixed_uid = input_value;
+		filter->fixed_uid_set = 1;
+	}
+
+	if (dict_get (this->options, "fixed-gid")) {
+		option_data = dict_get (this->options, "fixed-gid");
+		if (gf_string2int (option_data->data, &input_value) != 0) {
+			gf_log (this->name, GF_LOG_ERROR, 
+				"invalid number format \"%s\"", 
+				option_data->data);
+			return -1;
+		}
+		filter->fixed_gid = input_value;
+		filter->fixed_gid_set = 1;
+	}
+
+	this->private = filter;
+	return 0;
+}
+
+
+void
+fini (xlator_t *this)
+{
+	struct gf_filter *filter = this->private;
+
+	FREE (filter);
+
+	return;
+}
+
+
+struct xlator_fops fops = {
+	.lookup      = filter_lookup,
+	.stat        = filter_stat,
+	.fstat       = filter_fstat,
+	.chmod       = filter_chmod,
+	.fchmod      = filter_fchmod,
+	.readlink    = filter_readlink,
+	.mknod       = filter_mknod,
+	.mkdir       = filter_mkdir,
+	.unlink      = filter_unlink,
+	.rmdir       = filter_rmdir,
+	.symlink     = filter_symlink,
+	.rename      = filter_rename,
+	.link        = filter_link,
+	.chown       = filter_chown,
+	.fchown      = filter_fchown,
+	.truncate    = filter_truncate,
+	.ftruncate   = filter_ftruncate,
+	.create      = filter_create,
+	.open        = filter_open,
+	.readv       = filter_readv,
+	.writev      = filter_writev,
+	.setxattr    = filter_setxattr,
+	.getxattr    = filter_getxattr,
+	.removexattr = filter_removexattr,
+	.opendir     = filter_opendir,
+	.utimens     = filter_utimens,
+};
+
+struct xlator_mops mops = {
+};
+
+struct xlator_cbks cbks = {
+};
+
+struct volume_options options[] = {
+	{ .key  = { "root-squashing" }, 
+	  .type = GF_OPTION_TYPE_BOOL 
+	},
+	{ .key  = { "read-only" }, 
+	  .type = GF_OPTION_TYPE_BOOL
+	},
+	{ .key  = { "fixed-uid" },  
+	  .type = GF_OPTION_TYPE_INT
+	},
+	{ .key  = { "fixed-gid" },  
+	  .type = GF_OPTION_TYPE_INT
+	},
+	{ .key  = { "translate-uid" },  
+	  .type = GF_OPTION_TYPE_ANY 
+	},
+	{ .key  = { "translate-gid" },  
+	  .type = GF_OPTION_TYPE_ANY
+	},
+	{ .key  = { "filter-uid" },  
+	  .type = GF_OPTION_TYPE_ANY 
+	},
+	{ .key  = { "filter-gid" },  
+	  .type = GF_OPTION_TYPE_ANY 
+	},
+	{ .key = {NULL} },
+};
diff --git a/xlators/features/locks/Makefile.am b/xlators/features/locks/Makefile.am
new file mode 100644
index 00000000000..d471a3f9243
--- /dev/null
+++ b/xlators/features/locks/Makefile.am
@@ -0,0 +1,3 @@
+SUBDIRS = src
+
+CLEANFILES = 
diff --git a/xlators/features/locks/src/Makefile.am b/xlators/features/locks/src/Makefile.am
new file mode 100644
index 00000000000..ec4a953eb91
--- /dev/null
+++ b/xlators/features/locks/src/Makefile.am
@@ -0,0 +1,20 @@
+xlator_LTLIBRARIES = locks.la
+xlatordir = $(libdir)/glusterfs/$(PACKAGE_VERSION)/xlator/features
+
+locks_la_LDFLAGS = -module -avoidversion
+
+locks_la_SOURCES = common.c posix.c internal.c
+locks_la_LIBADD = $(top_builddir)/libglusterfs/src/libglusterfs.la 
+
+noinst_HEADERS = locks.h common.h
+
+AM_CFLAGS = -fPIC -D_FILE_OFFSET_BITS=64 -D_GNU_SOURCE -Wall -fno-strict-aliasing -D$(GF_HOST_OS) \
+	-I$(top_srcdir)/libglusterfs/src $(GF_CFLAGS) -shared -nostartfiles
+
+CLEANFILES = 
+
+uninstall-local:
+	rm -f $(DESTDIR)$(xlatordir)/posix-locks.so
+
+install-data-hook:
+	ln -sf locks.so $(DESTDIR)$(xlatordir)/posix-locks.so
\ No newline at end of file
diff --git a/xlators/features/locks/src/common.c b/xlators/features/locks/src/common.c
new file mode 100644
index 00000000000..9ac1250cc57
--- /dev/null
+++ b/xlators/features/locks/src/common.c
@@ -0,0 +1,561 @@
+/*
+  Copyright (c) 2006, 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/>.
+*/
+
+#include <unistd.h>
+#include <fcntl.h>
+#include <limits.h>
+#include <pthread.h>
+
+#ifndef _CONFIG_H
+#define _CONFIG_H
+#include "config.h"
+#endif
+
+#include "glusterfs.h"
+#include "compat.h"
+#include "xlator.h"
+#include "inode.h"
+#include "logging.h"
+#include "common-utils.h"
+
+#include "locks.h"
+
+
+int
+pl_is_lock_grantable (pl_inode_t *pl_inode, posix_lock_t *lock,
+		      gf_lk_domain_t dom);
+static void
+__insert_and_merge (pl_inode_t *pl_inode, posix_lock_t *lock,
+		    gf_lk_domain_t dom);
+
+
+pl_inode_t *
+pl_inode_get (xlator_t *this, inode_t *inode)
+{
+	pl_inode_t *pl_inode     = NULL;
+	mode_t      st_mode      = 0;
+	uint64_t    tmp_pl_inode = 0;
+	int         ret          = 0;
+	
+	LOCK (&inode->lock);
+	{
+		ret = inode_ctx_get (inode, this, &tmp_pl_inode);
+		if (ret == 0) {
+			pl_inode = (pl_inode_t *)(long)tmp_pl_inode;
+			goto out;
+		}
+
+		pl_inode = CALLOC (1, sizeof (*pl_inode));
+		if (!pl_inode) {
+			gf_log (this->name, GF_LOG_ERROR,
+				"out of memory :(");
+			goto out;
+		}
+
+		st_mode  = inode->st_mode;
+		if ((st_mode & S_ISGID) && !(st_mode & S_IXGRP))
+			pl_inode->mandatory = 1;
+
+
+		pthread_mutex_init (&pl_inode->mutex, NULL);
+		
+		INIT_LIST_HEAD (&pl_inode->dir_list);
+		INIT_LIST_HEAD (&pl_inode->ext_list);
+		INIT_LIST_HEAD (&pl_inode->int_list);
+		INIT_LIST_HEAD (&pl_inode->rw_list);
+
+		ret = inode_ctx_put (inode, this, (uint64_t)(long)pl_inode);
+	}
+out:
+	UNLOCK (&inode->lock);
+	return pl_inode;
+}
+
+
+/* Create a new posix_lock_t */
+posix_lock_t *
+new_posix_lock (struct flock *flock, transport_t *transport, pid_t client_pid)
+{
+	posix_lock_t *lock = NULL;
+
+	lock = CALLOC (1, sizeof (posix_lock_t));
+	if (!lock) {
+		return NULL;
+	}
+
+	lock->fl_start = flock->l_start;
+	lock->fl_type  = flock->l_type;
+
+	if (flock->l_len == 0)
+		lock->fl_end = LLONG_MAX;
+	else
+		lock->fl_end = flock->l_start + flock->l_len - 1;
+
+	lock->transport  = transport;
+	lock->client_pid = client_pid;
+
+	INIT_LIST_HEAD (&lock->list);
+
+	return lock;
+}
+
+
+/* Delete a lock from the inode's lock list */
+void
+__delete_lock (pl_inode_t *pl_inode, posix_lock_t *lock)
+{
+	list_del_init (&lock->list);
+}
+
+
+/* Destroy a posix_lock */
+void
+__destroy_lock (posix_lock_t *lock)
+{
+	free (lock);
+}
+
+
+/* Convert a posix_lock to a struct flock */
+void
+posix_lock_to_flock (posix_lock_t *lock, struct flock *flock)
+{
+	flock->l_pid   = lock->client_pid;
+	flock->l_type  = lock->fl_type;
+	flock->l_start = lock->fl_start;
+
+	if (lock->fl_end == 0)
+		flock->l_len = LLONG_MAX;
+	else
+		flock->l_len = lock->fl_end - lock->fl_start + 1;
+}
+
+
+/* Insert the lock into the inode's lock list */
+void
+pl_insert_lock (pl_inode_t *pl_inode, posix_lock_t *lock, gf_lk_domain_t dom)
+{
+	list_add_tail (&lock->list, DOMAIN_HEAD (pl_inode, dom));
+
+	return;
+}
+
+
+/* Return true if the locks overlap, false otherwise */
+int
+locks_overlap (posix_lock_t *l1, posix_lock_t *l2)
+{
+	/* 
+	   Note:
+	   FUSE always gives us absolute offsets, so no need to worry 
+	   about SEEK_CUR or SEEK_END
+	*/
+
+	return ((l1->fl_end >= l2->fl_start) &&
+		(l2->fl_end >= l1->fl_start));
+}
+
+
+/* Return true if the locks have the same owner */
+int
+same_owner (posix_lock_t *l1, posix_lock_t *l2)
+{
+	return ((l1->client_pid == l2->client_pid) &&
+		(l1->transport  == l2->transport));
+}
+
+
+/* Delete all F_UNLCK locks */
+void
+__delete_unlck_locks (pl_inode_t *pl_inode, gf_lk_domain_t dom)
+{
+	posix_lock_t *l = NULL;
+	posix_lock_t *tmp = NULL;
+
+	list_for_each_entry_safe (l, tmp, DOMAIN_HEAD (pl_inode, dom), list) {
+		if (l->fl_type == F_UNLCK) {
+			__delete_lock (pl_inode, l);
+			__destroy_lock (l);
+		}
+	}
+}
+
+
+/* Add two locks */
+static posix_lock_t *
+add_locks (posix_lock_t *l1, posix_lock_t *l2)
+{
+	posix_lock_t *sum = NULL;
+
+	sum = CALLOC (1, sizeof (posix_lock_t));
+	if (!sum)
+		return NULL;
+
+	sum->fl_start = min (l1->fl_start, l2->fl_start);
+	sum->fl_end   = max (l1->fl_end, l2->fl_end);
+
+	return sum;
+}
+
+/* Subtract two locks */
+struct _values {
+	posix_lock_t *locks[3];
+};
+
+/* {big} must always be contained inside {small} */
+static struct _values
+subtract_locks (posix_lock_t *big, posix_lock_t *small)
+{
+	struct _values v = { .locks = {0, 0, 0} };
+  
+	if ((big->fl_start == small->fl_start) && 
+	    (big->fl_end   == small->fl_end)) {  
+		/* both edges coincide with big */
+		v.locks[0] = CALLOC (1, sizeof (posix_lock_t));
+		ERR_ABORT (v.locks[0]);
+		memcpy (v.locks[0], big, sizeof (posix_lock_t));
+		v.locks[0]->fl_type = small->fl_type;
+	}
+	else if ((small->fl_start > big->fl_start) &&
+		 (small->fl_end   < big->fl_end)) {
+		/* both edges lie inside big */
+		v.locks[0] = CALLOC (1, sizeof (posix_lock_t));
+		ERR_ABORT (v.locks[0]);
+		v.locks[1] = CALLOC (1, sizeof (posix_lock_t));
+		ERR_ABORT (v.locks[1]);
+		v.locks[2] = CALLOC (1, sizeof (posix_lock_t));
+		ERR_ABORT (v.locks[2]);
+
+		memcpy (v.locks[0], big, sizeof (posix_lock_t));
+		v.locks[0]->fl_end = small->fl_start - 1;
+
+		memcpy (v.locks[1], small, sizeof (posix_lock_t));
+		memcpy (v.locks[2], big, sizeof (posix_lock_t));
+		v.locks[2]->fl_start = small->fl_end + 1;
+	}
+	/* one edge coincides with big */
+	else if (small->fl_start == big->fl_start) {
+		v.locks[0] = CALLOC (1, sizeof (posix_lock_t));
+		ERR_ABORT (v.locks[0]);
+		v.locks[1] = CALLOC (1, sizeof (posix_lock_t));
+		ERR_ABORT (v.locks[1]);
+    
+		memcpy (v.locks[0], big, sizeof (posix_lock_t));
+		v.locks[0]->fl_start = small->fl_end + 1;
+    
+		memcpy (v.locks[1], small, sizeof (posix_lock_t));
+	}
+	else if (small->fl_end   == big->fl_end) {
+		v.locks[0] = CALLOC (1, sizeof (posix_lock_t));
+		ERR_ABORT (v.locks[0]);
+		v.locks[1] = CALLOC (1, sizeof (posix_lock_t));
+		ERR_ABORT (v.locks[1]);
+
+		memcpy (v.locks[0], big, sizeof (posix_lock_t));
+		v.locks[0]->fl_end = small->fl_start - 1;
+    
+		memcpy (v.locks[1], small, sizeof (posix_lock_t));
+	}
+	else {
+		gf_log ("posix-locks", GF_LOG_DEBUG, 
+			"unexpected case in subtract_locks");
+	}
+
+	return v;
+}
+
+/* 
+   Start searching from {begin}, and return the first lock that
+   conflicts, NULL if no conflict
+   If {begin} is NULL, then start from the beginning of the list
+*/
+static posix_lock_t *
+first_overlap (pl_inode_t *pl_inode, posix_lock_t *lock,
+	       gf_lk_domain_t dom)
+{
+	posix_lock_t *l = NULL;
+
+	list_for_each_entry (l, DOMAIN_HEAD (pl_inode, dom), list) {
+		if (l->blocked)
+			continue;
+
+		if (locks_overlap (l, lock))
+			return l;
+	}
+
+	return NULL;
+}
+
+
+
+/* Return true if lock is grantable */
+int
+pl_is_lock_grantable (pl_inode_t *pl_inode, posix_lock_t *lock,
+		      gf_lk_domain_t dom)
+{
+	posix_lock_t *l = NULL;
+	int           ret = 1;
+
+	list_for_each_entry (l, DOMAIN_HEAD (pl_inode, dom), list) {
+		if (!l->blocked && locks_overlap (lock, l)) {
+			if (((l->fl_type == F_WRLCK)
+			     || (lock->fl_type == F_WRLCK))
+			    && (lock->fl_type != F_UNLCK)
+			    && !same_owner (l, lock)) {
+				ret = 0;
+				break;
+			}
+		}
+	}
+	return ret;
+}
+
+
+extern void do_blocked_rw (pl_inode_t *);
+
+
+static void
+__insert_and_merge (pl_inode_t *pl_inode, posix_lock_t *lock,
+		    gf_lk_domain_t dom)
+{
+	posix_lock_t  *conf = NULL;
+	posix_lock_t  *t = NULL;
+	posix_lock_t  *sum = NULL;
+	int            i = 0;
+	struct _values v = { .locks = {0, 0, 0} };
+
+	list_for_each_entry_safe (conf, t, DOMAIN_HEAD (pl_inode, dom), list) {
+		if (!locks_overlap (conf, lock))
+			continue;
+
+		if (same_owner (conf, lock)) {
+			if (conf->fl_type == lock->fl_type) {
+				sum = add_locks (lock, conf);
+
+				sum->fl_type    = lock->fl_type;
+				sum->transport  = lock->transport;
+				sum->client_pid = lock->client_pid;
+
+				__delete_lock (pl_inode, conf); 
+				__destroy_lock (conf);
+
+				__destroy_lock (lock);
+				__insert_and_merge (pl_inode, sum, dom);
+
+				return;
+			} else {
+				sum = add_locks (lock, conf);
+
+				sum->fl_type    = conf->fl_type;
+				sum->transport  = conf->transport;
+				sum->client_pid = conf->client_pid;
+
+				v = subtract_locks (sum, lock);
+	
+				__delete_lock (pl_inode, conf);
+				__destroy_lock (conf);
+
+				__delete_lock (pl_inode, lock);
+				__destroy_lock (lock);
+
+				__destroy_lock (sum);
+
+				for (i = 0; i < 3; i++) {
+					if (!v.locks[i])
+						continue;
+
+					if (v.locks[i]->fl_type == F_UNLCK) {
+						__destroy_lock (v.locks[i]);
+						continue;
+					}
+					__insert_and_merge (pl_inode,
+							    v.locks[i], dom);
+				}
+
+				__delete_unlck_locks (pl_inode, dom);
+				return; 
+			}
+		}
+
+		if (lock->fl_type == F_UNLCK) {
+			continue;
+		}
+
+		if ((conf->fl_type == F_RDLCK) && (lock->fl_type == F_RDLCK)) {
+			pl_insert_lock (pl_inode, lock, dom);
+			return;
+		}
+	}
+
+	/* no conflicts, so just insert */
+	if (lock->fl_type != F_UNLCK) {
+		pl_insert_lock (pl_inode, lock, dom);
+	} else {
+		__destroy_lock (lock);
+	}
+}
+
+
+void
+__grant_blocked_locks (xlator_t *this, pl_inode_t *pl_inode,
+		       gf_lk_domain_t dom, struct list_head *granted)
+{
+	struct list_head  tmp_list;
+	posix_lock_t     *l = NULL;
+	posix_lock_t     *tmp = NULL;
+	posix_lock_t     *conf = NULL;
+
+	INIT_LIST_HEAD (&tmp_list);
+
+	list_for_each_entry_safe (l, tmp, DOMAIN_HEAD (pl_inode, dom), list) {
+		if (l->blocked) {
+			conf = first_overlap (pl_inode, l, dom);
+			if (conf)
+				continue;
+
+			l->blocked = 0;
+			list_move_tail (&l->list, &tmp_list);
+		}
+	}
+
+	list_for_each_entry_safe (l, tmp, &tmp_list, list) {
+		list_del_init (&l->list);
+
+		if (pl_is_lock_grantable (pl_inode, l, dom)) {
+			conf = CALLOC (1, sizeof (*conf));
+
+			if (!conf) {
+				l->blocked = 1;
+				pl_insert_lock (pl_inode, l, dom);
+				continue;
+			}
+
+			conf->frame = l->frame;
+			l->frame = NULL;
+
+			posix_lock_to_flock (l, &conf->user_flock);
+
+			gf_log (this->name, GF_LOG_DEBUG,
+				"%s (pid=%d) %"PRId64" - %"PRId64" => Granted",
+				l->fl_type == F_UNLCK ? "Unlock" : "Lock",
+				l->client_pid,
+				l->user_flock.l_start,
+				l->user_flock.l_len);
+
+			__insert_and_merge (pl_inode, l, dom);
+
+			list_add (&conf->list, granted);
+		} else {
+			l->blocked = 1;
+			pl_insert_lock (pl_inode, l, dom);
+		}
+	}
+}
+
+
+void
+grant_blocked_locks (xlator_t *this, pl_inode_t *pl_inode, gf_lk_domain_t dom)
+{
+	struct list_head granted_list;
+	posix_lock_t     *tmp = NULL;
+	posix_lock_t     *lock = NULL;
+
+	INIT_LIST_HEAD (&granted_list);
+
+	pthread_mutex_lock (&pl_inode->mutex);
+	{
+		__grant_blocked_locks (this, pl_inode, dom, &granted_list);
+	}
+	pthread_mutex_unlock (&pl_inode->mutex);
+
+	list_for_each_entry_safe (lock, tmp, &granted_list, list) {
+		list_del_init (&lock->list);
+
+		STACK_UNWIND (lock->frame, 0, 0, &lock->user_flock);
+
+		FREE (lock);
+	}
+
+	return;
+}
+
+
+int
+pl_setlk (xlator_t *this, pl_inode_t *pl_inode, posix_lock_t *lock,
+	  int can_block,  gf_lk_domain_t dom)
+{
+	int              ret = 0;
+
+	errno = 0;
+
+	pthread_mutex_lock (&pl_inode->mutex);
+	{
+		if (pl_is_lock_grantable (pl_inode, lock, dom)) {
+			gf_log (this->name, GF_LOG_DEBUG,
+				"%s (pid=%d) %"PRId64" - %"PRId64" => OK",
+				lock->fl_type == F_UNLCK ? "Unlock" : "Lock",
+				lock->client_pid,
+				lock->user_flock.l_start,
+				lock->user_flock.l_len);
+			__insert_and_merge (pl_inode, lock, dom);
+		} else if (can_block) {
+			gf_log (this->name, GF_LOG_DEBUG,
+				"%s (pid=%d) %"PRId64" - %"PRId64" => Blocked",
+				lock->fl_type == F_UNLCK ? "Unlock" : "Lock",
+				lock->client_pid,
+				lock->user_flock.l_start,
+				lock->user_flock.l_len);
+			lock->blocked = 1;
+			pl_insert_lock (pl_inode, lock, dom);
+			ret = -1;
+		} else {
+			gf_log (this->name, GF_LOG_DEBUG,
+				"%s (pid=%d) %"PRId64" - %"PRId64" => NOK",
+				lock->fl_type == F_UNLCK ? "Unlock" : "Lock",
+				lock->client_pid,
+				lock->user_flock.l_start,
+				lock->user_flock.l_len);
+			errno = EAGAIN;
+			ret = -1;
+		}
+	}
+	pthread_mutex_unlock (&pl_inode->mutex);
+
+	grant_blocked_locks (this, pl_inode, dom);
+
+	do_blocked_rw (pl_inode);
+
+	return ret;
+}
+
+
+posix_lock_t *
+pl_getlk (pl_inode_t *pl_inode, posix_lock_t *lock, gf_lk_domain_t dom)
+{
+	posix_lock_t *conf = NULL;
+
+	conf = first_overlap (pl_inode, lock, dom);
+
+	if (conf == NULL) {
+		lock->fl_type = F_UNLCK;
+		return lock;
+	}
+
+	return conf;
+}
diff --git a/xlators/features/locks/src/common.h b/xlators/features/locks/src/common.h
new file mode 100644
index 00000000000..135f33011bf
--- /dev/null
+++ b/xlators/features/locks/src/common.h
@@ -0,0 +1,59 @@
+/*
+  Copyright (c) 2006, 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 __COMMON_H__
+#define __COMMON_H__
+
+posix_lock_t *
+new_posix_lock (struct flock *flock, transport_t *transport, pid_t client_pid);
+
+pl_inode_t *
+pl_inode_get (xlator_t *this, inode_t *inode);
+
+posix_lock_t *
+pl_getlk (pl_inode_t *inode, posix_lock_t *lock, gf_lk_domain_t domain);
+
+int
+pl_setlk (xlator_t *this, pl_inode_t *inode, posix_lock_t *lock,
+	  int can_block, gf_lk_domain_t domain);
+
+int
+pl_is_lock_grantable (pl_inode_t *pl_inode, posix_lock_t *lock,
+		      gf_lk_domain_t dom);
+
+void
+pl_insert_lock (pl_inode_t *pl_inode, posix_lock_t *lock, gf_lk_domain_t dom);
+
+void
+grant_blocked_locks (xlator_t *this, pl_inode_t *inode, gf_lk_domain_t domain);
+
+void
+posix_lock_to_flock (posix_lock_t *lock, struct flock *flock);
+
+int
+locks_overlap (posix_lock_t *l1, posix_lock_t *l2);
+
+int
+same_owner (posix_lock_t *l1, posix_lock_t *l2);
+
+void __delete_lock (pl_inode_t *, posix_lock_t *);
+
+void __destroy_lock (posix_lock_t *);
+
+#endif /* __COMMON_H__ */
diff --git a/xlators/features/locks/src/internal.c b/xlators/features/locks/src/internal.c
new file mode 100644
index 00000000000..7f454a78e22
--- /dev/null
+++ b/xlators/features/locks/src/internal.c
@@ -0,0 +1,762 @@
+/*
+  Copyright (c) 2006, 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 _CONFIG_H
+#define _CONFIG_H
+#include "config.h"
+#endif
+
+#include "glusterfs.h"
+#include "compat.h"
+#include "xlator.h"
+#include "inode.h"
+#include "logging.h"
+#include "common-utils.h"
+#include "list.h"
+
+#include "locks.h"
+#include "common.h"
+
+
+
+static int
+delete_locks_of_transport (pl_inode_t *pinode, transport_t *trans)
+{
+	posix_lock_t *tmp = NULL;
+	posix_lock_t *l = NULL;
+
+	list_for_each_entry_safe (l, tmp, &pinode->dir_list, list) {
+		if (l->transport == trans) {
+			__delete_lock (pinode, tmp);
+			__destroy_lock (tmp);
+		}
+	}
+
+	return 0;
+}
+
+
+static posix_lock_t *
+__find_exact_matching_lock (pl_inode_t *pinode, posix_lock_t *lock)
+{
+	posix_lock_t *l     = NULL;
+	posix_lock_t *match = NULL;
+
+	list_for_each_entry (l, DOMAIN_HEAD (pinode, GF_LOCK_INTERNAL), list) {
+		if (same_owner (l, lock)
+		    && (l->fl_start == lock->fl_start)
+		    && (l->fl_end   == lock->fl_end)) {
+			match = l;
+			break;
+		}
+	}
+
+	return match;
+}
+
+/**
+ * pl_inodelk: 
+ *
+ * This fop provides fcntl-style locking on files for internal
+ * purposes. Locks held through this fop reside in a domain different
+ * from those held by applications. This fop is for the use of AFR.
+ */
+
+
+static int
+pl_inodelk_common (call_frame_t *frame, xlator_t *this,
+		   inode_t *inode, int32_t cmd, struct flock *flock)
+{
+	int32_t op_ret   = -1;
+	int32_t op_errno = 0;
+	int     can_block = 0;
+
+	posix_locks_private_t * priv       = NULL;
+	transport_t *           transport  = NULL;
+	pid_t                   client_pid = -1;
+	pl_inode_t *            pinode     = NULL;
+	
+	posix_lock_t *          reqlock    = NULL;
+	posix_lock_t *          matchlock  = NULL;  /* steady, fire! */
+
+	VALIDATE_OR_GOTO (frame, unwind);
+	VALIDATE_OR_GOTO (inode, unwind);
+	VALIDATE_OR_GOTO (flock, unwind);
+	
+	if ((flock->l_start < 0) || (flock->l_len < 0)) {
+		op_errno = EINVAL;
+		goto unwind;
+	}
+
+	transport  = frame->root->trans;
+	client_pid = frame->root->pid;
+
+	priv = (posix_locks_private_t *) this->private;
+
+	VALIDATE_OR_GOTO (priv, unwind);
+
+	pinode = pl_inode_get (this, inode);
+	if (!pinode) {
+		gf_log (this->name, GF_LOG_ERROR,
+			"out of memory :(");
+		op_errno = ENOMEM;
+		goto unwind;
+	}
+
+	if (client_pid == 0) {
+		/* 
+		   special case: this means release all locks 
+		   from this transport
+		*/
+
+		gf_log (this->name, GF_LOG_DEBUG,
+			"releasing all locks from transport %p", transport);
+
+		delete_locks_of_transport (pinode, transport);
+		goto unwind;
+	}
+
+	reqlock = new_posix_lock (flock, transport, client_pid);
+	if (!reqlock) {
+		gf_log (this->name, GF_LOG_ERROR,
+			"out of memory :(");
+		op_ret = -1;
+		op_errno = ENOMEM;
+		goto unwind;
+	}
+
+	pthread_mutex_lock (&pinode->mutex);
+	{
+		switch (cmd) {
+		case F_SETLKW:
+			can_block = 1;
+			reqlock->frame = frame;
+			reqlock->this  = this;
+			
+			/* fall through */
+			
+		case F_SETLK:
+			memcpy (&reqlock->user_flock, flock, sizeof (struct flock));
+			
+			switch (flock->l_type) {
+				
+			case F_WRLCK:
+				if (!pl_is_lock_grantable (pinode, reqlock, GF_LOCK_INTERNAL)) {
+					if (can_block) {
+						gf_log (this->name, GF_LOG_DEBUG,
+							"%s (pid=%d) %"PRId64" - %"PRId64" => blocked",
+							reqlock->fl_type == F_UNLCK ? "unlock" : "lock",
+							reqlock->client_pid,
+							reqlock->user_flock.l_start,
+							reqlock->user_flock.l_len);
+						pl_insert_lock (pinode, reqlock, GF_LOCK_INTERNAL);
+						
+						goto unlock;
+					}
+					
+					__destroy_lock (reqlock);
+					
+					
+					gf_log (this->name, GF_LOG_DEBUG,
+						"%s (pid=%d) %"PRId64" - %"PRId64" => NOK",
+						reqlock->fl_type == F_UNLCK ? "unlock" : "lock",
+						reqlock->client_pid, reqlock->user_flock.l_start,
+						reqlock->user_flock.l_len);
+					op_errno = EAGAIN;
+					
+					goto unlock;
+				}
+				
+				gf_log (this->name, GF_LOG_DEBUG,
+					"%s (pid=%d) %"PRId64" - %"PRId64" => OK",
+					reqlock->fl_type == F_UNLCK ? "unlock" : "lock",
+					reqlock->client_pid,
+					reqlock->user_flock.l_start,
+					reqlock->user_flock.l_len);
+				pl_insert_lock (pinode, reqlock, GF_LOCK_INTERNAL);
+				
+				break;
+				
+			case F_UNLCK:
+				matchlock = __find_exact_matching_lock (pinode, reqlock);
+
+				__destroy_lock (reqlock);
+				if (!matchlock) {
+					op_errno = EINVAL;
+					goto unlock;
+				}
+				
+				__delete_lock (pinode, matchlock);
+				__destroy_lock (matchlock);
+				
+				break;
+				
+			default:
+				op_errno = ENOTSUP;
+				gf_log (this->name, GF_LOG_ERROR,
+					"lock type %d not supported for [F]INODELK",
+					flock->l_type);
+				goto unlock;
+			}
+			
+			
+			break;
+			
+		default:
+			op_errno = ENOTSUP;
+			gf_log (this->name, GF_LOG_ERROR,
+				"lock command F_GETLK not supported for [F]INODELK (cmd=%d)", 
+				cmd);
+			goto unlock;
+		}
+		
+		op_ret = 0;
+		
+	unlock:
+		if (pinode)
+			pthread_mutex_unlock (&pinode->mutex);
+	}
+
+unwind:
+	STACK_UNWIND (frame, op_ret, op_errno);
+	return 0;
+}
+
+
+int
+pl_inodelk (call_frame_t *frame, xlator_t *this,
+	    loc_t *loc, int32_t cmd, struct flock *flock)
+{
+	return pl_inodelk_common (frame, this, loc->inode, cmd, flock);
+}
+
+
+int
+pl_finodelk (call_frame_t *frame, xlator_t *this,
+	     fd_t *fd, int32_t cmd, struct flock *flock)
+{
+	return pl_inodelk_common (frame, this, fd->inode, cmd, flock);
+}
+
+
+/**
+ * types_conflict - do two types of lock conflict?
+ * @t1: type
+ * @t2: type
+ *
+ * two read locks do not conflict
+ * any other case conflicts
+ */
+
+static int
+types_conflict (entrylk_type t1, entrylk_type t2)
+{
+	return !((t1 == ENTRYLK_RDLCK) && (t2 == ENTRYLK_RDLCK));
+}
+
+/**
+ * all_names - does a basename represent all names?
+ * @basename: name to check
+ */
+
+#define all_names(basename) ((basename == NULL) ? 1 : 0)
+
+/**
+ * names_conflict - do two names conflict?
+ * @n1: name
+ * @n2: name
+ */
+
+static int 
+names_conflict (const char *n1, const char *n2)
+{
+	return all_names (n1) || all_names (n2) || !strcmp (n1, n2);
+}
+
+
+static int 
+names_equal (const char *n1, const char *n2)
+{
+	return (n1 == NULL && n2 == NULL) || (n1 && n2 && !strcmp (n1, n2));
+}
+
+/**
+ * lock_grantable - is this lock grantable?
+ * @inode: inode in which to look
+ * @basename: name we're trying to lock
+ * @type: type of lock
+ */
+
+static pl_entry_lock_t *
+__lock_grantable (pl_inode_t *pinode, const char *basename, entrylk_type type)
+{
+	pl_entry_lock_t *lock = NULL;
+
+	if (list_empty (&pinode->dir_list))
+		return NULL;
+
+	list_for_each_entry (lock, &pinode->dir_list, inode_list) {
+		if (names_conflict (lock->basename, basename) &&
+		    types_conflict (lock->type, type))
+			return lock;
+	}
+
+	return NULL;
+}
+
+/**
+ * find_most_matching_lock - find the lock struct which most matches in order of:
+ *                           lock on the exact basename ||
+ *                           an all_names lock
+ *                      
+ *
+ * @inode: inode in which to look
+ * @basename: name to search for
+ */
+
+static pl_entry_lock_t * 
+__find_most_matching_lock (pl_inode_t *pinode, const char *basename)
+{
+	pl_entry_lock_t *lock;
+	pl_entry_lock_t *all = NULL;
+	pl_entry_lock_t *exact = NULL;
+
+	if (list_empty (&pinode->dir_list)) 
+		return NULL;
+
+	list_for_each_entry (lock, &pinode->dir_list, inode_list) {
+		if (all_names (lock->basename))
+			all = lock;
+		else if (names_equal (lock->basename, basename))
+			exact = lock;
+	}
+
+	return (exact ? exact : all);
+}
+
+
+/**
+ * insert_new_lock - insert a new dir lock into the inode with the given parameters
+ * @pinode: inode to insert into
+ * @basename: basename for the lock
+ * @type: type of the lock
+ */
+
+static pl_entry_lock_t *
+new_entrylk_lock (pl_inode_t *pinode, const char *basename, entrylk_type type,
+		  transport_t *trans)
+{
+	pl_entry_lock_t *newlock = NULL;
+
+	newlock = CALLOC (sizeof (pl_entry_lock_t), 1);
+	if (!newlock) {
+		goto out;
+	}
+
+	newlock->basename = basename ? strdup (basename) : NULL;
+	newlock->type     = type;
+	newlock->trans    = trans;
+
+	if (type == ENTRYLK_RDLCK)
+		newlock->read_count = 1;
+
+	INIT_LIST_HEAD (&newlock->inode_list);
+	INIT_LIST_HEAD (&newlock->blocked_locks);
+
+out:
+	return newlock;
+}
+
+/**
+ * lock_name - lock a name in a directory
+ * @inode: inode for the directory in which to lock
+ * @basename: name of the entry to lock
+ *            if null, lock the entire directory
+ *            
+ * the entire directory being locked is represented as: a single
+ * pl_entry_lock_t present in the entrylk_locks list with its
+ * basename = NULL
+ */
+
+int
+__lock_name (pl_inode_t *pinode, const char *basename, entrylk_type type,
+	     call_frame_t *frame, xlator_t *this, int nonblock)
+{
+	pl_entry_lock_t *lock    = NULL;
+	pl_entry_lock_t *conf    = NULL;
+
+	transport_t *trans = NULL;
+
+	int ret = -EINVAL;
+
+	trans = frame->root->trans;
+
+	conf = __lock_grantable (pinode, basename, type);
+	if (conf) {
+		ret = -EAGAIN;
+		if (nonblock)
+			goto out;
+
+		lock = new_entrylk_lock (pinode, basename, type, trans);
+
+		if (!lock) {
+			ret = -ENOMEM;
+			goto out;
+		}
+
+		gf_log (this->name, GF_LOG_DEBUG,
+			"blocking lock: {pinode=%p, basename=%s}",
+			pinode, basename);
+
+		lock->frame   = frame;
+		lock->this    = this;
+		lock->blocked = 1;
+
+		list_add (&lock->blocked_locks, &conf->blocked_locks);
+
+
+		goto out;
+	}
+		
+	switch (type) {
+	case ENTRYLK_RDLCK:
+		lock = __find_most_matching_lock (pinode, basename);
+
+		if (lock && names_equal (lock->basename, basename)) {
+			lock->read_count++;
+
+			FREE (lock->basename);
+			FREE (lock);
+
+			lock = NULL;
+		} else {
+			lock = new_entrylk_lock (pinode, basename, type, trans);
+
+			if (!lock) {
+				ret = -ENOMEM;
+				goto out;
+			}
+
+			list_add (&lock->inode_list, &pinode->dir_list);
+		}
+		break;
+
+	case ENTRYLK_WRLCK:
+		lock = new_entrylk_lock (pinode, basename, type, trans);
+			
+		if (!lock) {
+			ret = -ENOMEM;
+			goto out;
+		}
+
+		list_add (&lock->inode_list, &pinode->dir_list);
+		break;
+	}
+
+	ret = 0;
+out:
+	return ret;
+}
+
+
+/**
+ * unlock_name - unlock a name in a directory
+ * @inode: inode for the directory to unlock in
+ * @basename: name of the entry to unlock
+ *            if null, unlock the entire directory
+ */
+
+pl_entry_lock_t *
+__unlock_name (pl_inode_t *pinode, const char *basename, entrylk_type type)
+{
+	pl_entry_lock_t *lock = NULL;
+	pl_entry_lock_t *ret_lock = NULL;
+
+	lock = __find_most_matching_lock (pinode, basename);
+	
+	if (!lock) {
+		gf_log ("locks", GF_LOG_DEBUG,
+			"unlock on %s (type=%s) attempted but no matching lock found",
+			basename, type == ENTRYLK_RDLCK ? "ENTRYLK_RDLCK" : 
+			"ENTRYLK_WRLCK");
+		goto out;
+	}
+	
+	if (names_equal (lock->basename, basename)
+	    && lock->type == type) {
+		if (type == ENTRYLK_RDLCK) {
+			lock->read_count--;
+		}
+		if (type == ENTRYLK_WRLCK || lock->read_count == 0) {
+			list_del (&lock->inode_list);
+			ret_lock = lock;
+		}
+	} else {
+		gf_log ("locks", GF_LOG_ERROR,
+			"unlock for a non-existing lock!");
+		goto out;
+	}
+
+out:
+	return ret_lock;
+}
+
+
+void
+__grant_blocked_entry_locks (xlator_t *this, pl_inode_t *pl_inode,
+			     pl_entry_lock_t *lock,
+			     struct list_head *granted)
+{
+	int              bl_ret = 0;
+	pl_entry_lock_t *bl   = NULL;
+	pl_entry_lock_t *tmp  = NULL;
+
+	list_for_each_entry_safe (bl, tmp, &lock->blocked_locks, 
+				  blocked_locks) {
+		list_del_init (&bl->blocked_locks);
+
+		/* TODO: error checking */
+
+		gf_log ("locks", GF_LOG_DEBUG,
+			"trying to unblock: {pinode=%p, basename=%s}",
+			pl_inode, bl->basename);
+					
+		bl_ret = __lock_name (pl_inode, bl->basename, bl->type,
+				      bl->frame, bl->this, 0);
+
+		if (bl_ret == 0) {
+			list_add (&bl->blocked_locks, granted);
+		} else {
+			if (bl->basename)
+				FREE (bl->basename);
+			FREE (bl);
+		}
+	}	
+	return;
+}
+
+
+void
+grant_blocked_entry_locks (xlator_t *this, pl_inode_t *pl_inode,
+			   pl_entry_lock_t *unlocked)
+{
+	struct list_head  granted_list;
+	pl_entry_lock_t  *tmp = NULL;
+	pl_entry_lock_t  *lock = NULL;
+
+	INIT_LIST_HEAD (&granted_list);
+
+	pthread_mutex_lock (&pl_inode->mutex);
+	{
+		__grant_blocked_entry_locks (this, pl_inode, unlocked,
+					     &granted_list);
+	}
+	pthread_mutex_unlock (&pl_inode->mutex);
+
+	list_for_each_entry_safe (lock, tmp, &granted_list, blocked_locks) {
+		list_del_init (&lock->blocked_locks);
+
+		STACK_UNWIND (lock->frame, 0, 0);
+
+		FREE (lock->basename);
+		FREE (lock);
+	}
+
+	FREE (unlocked->basename);
+	FREE (unlocked);
+
+	return;
+}
+
+
+/**
+ * release_entry_locks_for_transport: release all entry locks from this
+ * transport for this loc_t
+ */
+
+static int
+release_entry_locks_for_transport (xlator_t *this, pl_inode_t *pinode,
+				   transport_t *trans)
+{
+	pl_entry_lock_t  *lock;
+	pl_entry_lock_t  *tmp;
+	struct list_head  granted;
+
+	INIT_LIST_HEAD (&granted);
+
+	pthread_mutex_lock (&pinode->mutex);
+	{
+		if (list_empty (&pinode->dir_list)) {
+			goto unlock;
+		}
+
+		list_for_each_entry_safe (lock, tmp, &pinode->dir_list,
+					  inode_list) {
+			if (lock->trans != trans)
+				continue;
+
+			list_del_init (&lock->inode_list);
+			__grant_blocked_entry_locks (this, pinode, lock,
+						     &granted);
+
+			FREE (lock->basename);
+			FREE (lock);
+		}
+	}
+unlock:
+	pthread_mutex_unlock (&pinode->mutex);
+
+	list_for_each_entry_safe (lock, tmp, &granted, blocked_locks) {
+		list_del_init (&lock->blocked_locks);
+
+		STACK_UNWIND (lock->frame, 0, 0);
+
+		FREE (lock->basename);
+		FREE (lock);
+	}
+
+	return 0;
+}
+
+
+/**
+ * pl_entrylk:
+ * 
+ * Locking on names (directory entries)
+ */
+
+int
+pl_entrylk_common (call_frame_t *frame, xlator_t *this,
+		   inode_t *inode, const char *basename, 
+		   entrylk_cmd cmd, entrylk_type type)
+{
+	int32_t op_ret   = -1;
+	int32_t op_errno = 0;
+
+	transport_t * transport = NULL;
+	pid_t pid = -1;
+
+	pl_inode_t *       pinode = NULL; 
+	int                ret    = -1;
+	pl_entry_lock_t   *unlocked = NULL;
+	char               unwind = 1;
+
+	pinode = pl_inode_get (this, inode);
+	if (!pinode) {
+		gf_log (this->name, GF_LOG_ERROR,
+			"out of memory :(");
+		op_errno = ENOMEM;
+		goto out;
+	}
+
+	pid       = frame->root->pid;
+	transport = frame->root->trans;
+
+	if (pid == 0) {
+		/* 
+		   this is a special case that means release
+		   all locks from this transport 
+		*/
+
+		gf_log (this->name, GF_LOG_DEBUG,
+			"releasing locks for transport %p", transport);
+
+		release_entry_locks_for_transport (this, pinode, transport);
+		op_ret = 0;
+
+		goto out;
+	}
+
+	switch (cmd) {
+	case ENTRYLK_LOCK:
+		pthread_mutex_lock (&pinode->mutex);
+		{
+			ret = __lock_name (pinode, basename, type,
+					   frame, this, 0);
+		}
+		pthread_mutex_unlock (&pinode->mutex);
+
+		if (ret < 0) {
+			if (ret == -EAGAIN)
+				unwind = 0;
+			op_errno = -ret;
+			goto out;
+		}
+
+		break;
+
+	case ENTRYLK_LOCK_NB:
+		pthread_mutex_lock (&pinode->mutex);
+		{
+			ret = __lock_name (pinode, basename, type,
+					   frame, this, 1);
+		}
+		pthread_mutex_unlock (&pinode->mutex);
+
+		if (ret < 0) {
+			op_errno = -ret;
+			goto out;
+		}
+
+		break;
+
+	case ENTRYLK_UNLOCK:
+		pthread_mutex_lock (&pinode->mutex);
+		{
+			unlocked = __unlock_name (pinode, basename, type);
+		}
+		pthread_mutex_unlock (&pinode->mutex);
+
+		if (unlocked)
+			grant_blocked_entry_locks (this, pinode, unlocked);
+
+		break;
+
+	default:
+		gf_log (this->name, GF_LOG_ERROR,
+			"unexpected case!");
+		goto out;
+	}
+
+	op_ret = 0;
+out:
+	if (unwind) {
+		STACK_UNWIND (frame, op_ret, op_errno);
+	}
+	
+	return 0;
+}
+
+
+int
+pl_entrylk (call_frame_t *frame, xlator_t *this,
+	    loc_t *loc, const char *basename, 
+	    entrylk_cmd cmd, entrylk_type type)
+{
+	return pl_entrylk_common (frame, this, loc->inode, basename, cmd, type);
+}
+
+
+int
+pl_fentrylk (call_frame_t *frame, xlator_t *this,
+	     fd_t *fd, const char *basename, 
+	     entrylk_cmd cmd, entrylk_type type)
+{
+	return pl_entrylk_common (frame, this, fd->inode, basename, cmd, type);	
+}
diff --git a/xlators/features/locks/src/locks.h b/xlators/features/locks/src/locks.h
new file mode 100644
index 00000000000..8ed7bb63f1c
--- /dev/null
+++ b/xlators/features/locks/src/locks.h
@@ -0,0 +1,111 @@
+/*
+   Copyright (c) 2006, 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 __POSIX_LOCKS_H__
+#define __POSIX_LOCKS_H__
+
+#ifndef _CONFIG_H
+#define _CONFIG_H
+#include "config.h"
+#endif
+
+#include "compat-errno.h"
+#include "transport.h"
+#include "stack.h"
+#include "call-stub.h"
+
+struct __pl_fd;
+
+struct __posix_lock {
+	struct list_head   list;
+
+	short              fl_type;
+	off_t              fl_start;
+	off_t              fl_end;
+
+	short              blocked;    /* waiting to acquire */
+	struct flock       user_flock; /* the flock supplied by the user */
+	xlator_t          *this;       /* required for blocked locks */
+	fd_t              *fd;
+
+	call_frame_t      *frame;
+
+	/* These two together serve to uniquely identify each process
+	   across nodes */
+
+	transport_t       *transport;     /* to identify client node */
+	pid_t              client_pid;    /* pid of client process */
+};
+typedef struct __posix_lock posix_lock_t;
+
+struct __pl_rw_req_t {
+	struct list_head      list;
+	call_stub_t          *stub;
+	posix_lock_t          region;
+};
+typedef struct __pl_rw_req_t pl_rw_req_t;
+
+
+struct __entry_lock {
+	struct list_head  inode_list;    /* list_head back to pl_inode_t */
+	struct list_head  blocked_locks; /* locks blocked due to this lock */
+
+	call_frame_t     *frame;
+	xlator_t         *this;
+	int               blocked;
+	
+	const char       *basename;
+	entrylk_type      type;
+	unsigned int      read_count;    /* number of read locks */
+	transport_t      *trans;
+};
+typedef struct __entry_lock pl_entry_lock_t;
+
+
+/* The "simulated" inode. This contains a list of all the locks associated 
+   with this file */
+
+struct __pl_inode {
+	pthread_mutex_t  mutex;
+
+	struct list_head dir_list;       /* list of entry locks */
+	struct list_head ext_list;       /* list of fcntl locks */
+	struct list_head int_list;       /* list of internal locks */
+	struct list_head rw_list;        /* list of waiting r/w requests */
+	int              mandatory;      /* if mandatory locking is enabled */
+};
+typedef struct __pl_inode pl_inode_t;
+
+#define DOMAIN_HEAD(pl_inode, dom) (dom == GF_LOCK_POSIX	\
+				    ? &pl_inode->ext_list	\
+				    : &pl_inode->int_list)
+
+
+struct __pl_fd {
+	gf_boolean_t nonblocking;       /* whether O_NONBLOCK has been set */
+};
+typedef struct __pl_fd pl_fd_t;
+
+
+typedef struct {
+	gf_boolean_t    mandatory;      /* if mandatory locking is enabled */
+} posix_locks_private_t;
+
+
+#endif /* __POSIX_LOCKS_H__ */
diff --git a/xlators/features/locks/src/posix.c b/xlators/features/locks/src/posix.c
new file mode 100644
index 00000000000..e2b336607c4
--- /dev/null
+++ b/xlators/features/locks/src/posix.c
@@ -0,0 +1,834 @@
+/*
+  Copyright (c) 2006, 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/>.
+*/
+
+#include <unistd.h>
+#include <fcntl.h>
+#include <limits.h>
+#include <pthread.h>
+
+#ifndef _CONFIG_H
+#define _CONFIG_H
+#include "config.h"
+#endif
+
+#include "glusterfs.h"
+#include "compat.h"
+#include "xlator.h"
+#include "inode.h"
+#include "logging.h"
+#include "common-utils.h"
+
+#include "locks.h"
+#include "common.h"
+
+#ifndef LLONG_MAX
+#define LLONG_MAX LONG_LONG_MAX /* compat with old gcc */
+#endif /* LLONG_MAX */
+
+/* Forward declarations */
+
+
+void do_blocked_rw (pl_inode_t *);
+static int __rw_allowable (pl_inode_t *, posix_lock_t *, glusterfs_fop_t);
+
+struct _truncate_ops {
+	loc_t  loc;
+	fd_t  *fd;
+	off_t  offset;
+	enum {TRUNCATE, FTRUNCATE} op;
+};
+
+
+int
+pl_truncate_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
+		 int32_t op_ret, int32_t op_errno, struct stat *buf)
+{
+	struct _truncate_ops *local = NULL;
+
+	local = frame->local;
+
+	if (local->op == TRUNCATE)
+		loc_wipe (&local->loc);
+
+	STACK_UNWIND (frame, op_ret, op_errno, buf);
+	return 0;
+}
+
+
+static int
+truncate_allowed (pl_inode_t *pl_inode, 
+		  transport_t *transport, pid_t client_pid, 
+		  off_t offset)
+{
+	posix_lock_t *l = NULL;
+	posix_lock_t  region = {.list = {0, }, };
+	int           ret = 1;
+
+	region.fl_start   = offset;
+	region.fl_end     = LLONG_MAX;
+	region.transport  = transport;
+	region.client_pid = client_pid;
+
+	pthread_mutex_lock (&pl_inode->mutex);
+	{
+		list_for_each_entry (l, &pl_inode->ext_list, list) {
+			if (!l->blocked
+			    && locks_overlap (&region, l)
+			    && !same_owner (&region, l)) {
+				ret = 0;
+				break;
+			}
+		}
+	}
+	pthread_mutex_unlock (&pl_inode->mutex);
+
+	return ret;
+}
+
+
+static int
+truncate_stat_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
+		   int32_t op_ret, int32_t op_errno, struct stat *buf)
+{
+	posix_locks_private_t *priv = NULL;
+	struct _truncate_ops  *local = NULL;
+	inode_t               *inode = NULL;
+	pl_inode_t            *pl_inode = NULL;
+
+
+	priv = this->private;
+	local = frame->local;
+
+	if (op_ret != 0) {
+		gf_log (this->name, GF_LOG_ERROR, 
+			"got error (errno=%d, stderror=%s) from child", 
+			op_errno, strerror (op_errno));
+		goto unwind;
+	}
+
+	if (local->op == TRUNCATE)
+		inode = local->loc.inode;
+	else
+		inode = local->fd->inode;
+
+	pl_inode = pl_inode_get (this, inode);
+	if (!pl_inode) {
+		gf_log (this->name, GF_LOG_ERROR,
+			"unable to get pl_inode from %p", inode);
+		op_errno = ENOMEM;
+		goto unwind;
+	}
+
+	if (priv->mandatory
+	    && pl_inode->mandatory
+	    && !truncate_allowed (pl_inode, frame->root->trans,
+				  frame->root->pid, local->offset)) {
+		op_errno = EAGAIN;
+		goto unwind;
+	}
+
+	switch (local->op) {
+	case TRUNCATE:
+		STACK_WIND (frame, pl_truncate_cbk, FIRST_CHILD (this),
+			    FIRST_CHILD (this)->fops->truncate,
+			    &local->loc, local->offset);
+		break;
+	case FTRUNCATE:
+		STACK_WIND (frame, pl_truncate_cbk, FIRST_CHILD (this),
+			    FIRST_CHILD (this)->fops->ftruncate,
+			    local->fd, local->offset);
+		break;
+	}
+
+	return 0;
+
+unwind:
+	if (local->op == TRUNCATE)
+		loc_wipe (&local->loc);
+
+	STACK_UNWIND (frame, -1, ENOMEM, buf);
+	return 0;
+}
+
+
+int
+pl_truncate (call_frame_t *frame, xlator_t *this,
+	     loc_t *loc, off_t offset)
+{
+	struct _truncate_ops *local = NULL;
+
+	local = CALLOC (1, sizeof (struct _truncate_ops));
+	if (!local) {
+		gf_log (this->name, GF_LOG_ERROR,
+			"out of memory :(");
+		goto unwind;
+	}
+
+	local->op         = TRUNCATE;
+	local->offset     = offset;
+	loc_copy (&local->loc, loc);
+
+	frame->local = local;
+
+	STACK_WIND (frame, truncate_stat_cbk, FIRST_CHILD (this),
+		    FIRST_CHILD (this)->fops->stat, loc);
+
+	return 0;
+
+unwind:
+	STACK_UNWIND (frame, -1, ENOMEM, NULL);
+
+	return 0;
+}
+
+
+int
+pl_ftruncate (call_frame_t *frame, xlator_t *this,
+	      fd_t *fd, off_t offset)
+{
+	struct _truncate_ops *local = NULL;
+
+	local = CALLOC (1, sizeof (struct _truncate_ops));
+	if (!local) {
+		gf_log (this->name, GF_LOG_ERROR,
+			"out of memory :(");
+		goto unwind;
+	}
+
+	local->op         = FTRUNCATE;
+	local->offset     = offset;
+	local->fd         = fd;
+
+	frame->local = local;
+
+	STACK_WIND (frame, truncate_stat_cbk, FIRST_CHILD(this),
+		    FIRST_CHILD(this)->fops->fstat, fd);
+	return 0;
+
+unwind:
+	STACK_UNWIND (frame, -1, ENOMEM, NULL);
+
+	return 0;
+}
+
+
+static void
+__delete_locks_of_owner (pl_inode_t *pl_inode,
+			 transport_t *transport, pid_t pid)
+{
+	posix_lock_t *tmp = NULL;
+	posix_lock_t *l = NULL;
+
+	/* TODO: what if it is a blocked lock with pending l->frame */
+
+	list_for_each_entry_safe (l, tmp, &pl_inode->ext_list, list) {
+		if ((l->transport == transport)
+		    && (l->client_pid == pid)) {
+			__delete_lock (pl_inode, l);
+			__destroy_lock (l);
+		}
+	}
+
+	list_for_each_entry_safe (l, tmp, &pl_inode->int_list, list) {
+		if ((l->transport == transport)
+		    && (l->client_pid == pid)) {
+			__delete_lock (pl_inode, l);
+			__destroy_lock (l);
+		}
+	}
+
+	return;
+}
+
+
+int
+pl_flush_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
+	      int32_t op_ret, int32_t op_errno)
+{
+	STACK_UNWIND (frame, op_ret, op_errno);
+
+	return 0;
+}
+
+
+int
+pl_flush (call_frame_t *frame, xlator_t *this,
+	  fd_t *fd)
+{
+	posix_locks_private_t *priv = NULL;
+	pl_inode_t            *pl_inode = NULL;
+
+	priv = this->private;
+
+	pl_inode = pl_inode_get (this, fd->inode);
+	if (!pl_inode) {
+		gf_log (this->name, GF_LOG_ERROR, "returning EBADFD");
+		STACK_UNWIND (frame, -1, EBADFD);
+		return 0;
+	}
+
+	pthread_mutex_lock (&pl_inode->mutex);
+	{
+		__delete_locks_of_owner (pl_inode, frame->root->trans,
+					 frame->root->pid);
+	}
+	pthread_mutex_unlock (&pl_inode->mutex);
+
+	grant_blocked_locks (this, pl_inode, GF_LOCK_POSIX);
+	grant_blocked_locks (this, pl_inode, GF_LOCK_INTERNAL);
+
+	do_blocked_rw (pl_inode);
+
+	STACK_WIND (frame, pl_flush_cbk, FIRST_CHILD(this),
+		    FIRST_CHILD(this)->fops->flush, fd);
+	return 0;
+}
+
+
+int
+pl_open_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
+	     int32_t op_ret, int32_t op_errno, fd_t *fd)
+{
+	STACK_UNWIND (frame, op_ret, op_errno, fd);
+
+	return 0;
+}
+
+
+int
+pl_open (call_frame_t *frame, xlator_t *this,
+	 loc_t *loc, int32_t flags, fd_t *fd)
+{
+	/* why isn't O_TRUNC being handled ? */
+	STACK_WIND (frame, pl_open_cbk, 
+		    FIRST_CHILD(this), FIRST_CHILD(this)->fops->open, 
+		    loc, flags & ~O_TRUNC, fd);
+
+	return 0;
+}
+
+
+int
+pl_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)
+{
+	STACK_UNWIND (frame, op_ret, op_errno, fd, inode, buf);
+
+	return 0;
+}
+
+
+int
+pl_create (call_frame_t *frame, xlator_t *this,
+	   loc_t *loc, int32_t flags, mode_t mode, fd_t *fd)
+{
+	STACK_WIND (frame, pl_create_cbk,
+		    FIRST_CHILD (this), FIRST_CHILD (this)->fops->create, 
+		    loc, flags, mode, fd);
+	return 0;
+}
+
+
+int
+pl_readv_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;
+}
+
+int
+pl_writev_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;
+}
+
+
+void
+do_blocked_rw (pl_inode_t *pl_inode)
+{
+	struct list_head  wind_list;
+	pl_rw_req_t      *rw = NULL;
+	pl_rw_req_t      *tmp = NULL;
+
+	INIT_LIST_HEAD (&wind_list);
+
+	pthread_mutex_lock (&pl_inode->mutex);
+	{
+		list_for_each_entry_safe (rw, tmp, &pl_inode->rw_list, list) {
+			if (__rw_allowable (pl_inode, &rw->region,
+					    rw->stub->fop)) {
+				list_del_init (&rw->list);
+				list_add_tail (&rw->list, &wind_list);
+			}
+		}
+	}
+	pthread_mutex_unlock (&pl_inode->mutex);
+
+	list_for_each_entry_safe (rw, tmp, &wind_list, list) {
+		list_del_init (&rw->list);
+		call_resume (rw->stub);
+		free (rw);
+	}
+
+	return;
+}
+
+
+static int
+__rw_allowable (pl_inode_t *pl_inode, posix_lock_t *region,
+		glusterfs_fop_t op)
+{
+	posix_lock_t *l = NULL;
+	int           ret = 1;
+
+	list_for_each_entry (l, &pl_inode->ext_list, list) {
+		if (locks_overlap (l, region) && !same_owner (l, region)) {
+			if ((op == GF_FOP_READ) && (l->fl_type != F_WRLCK))
+				continue;
+			ret = 0;
+			break;
+		}
+	}
+
+	return ret;
+}
+
+
+int
+pl_readv_cont (call_frame_t *frame, xlator_t *this,
+	       fd_t *fd, size_t size, off_t offset)
+{
+	STACK_WIND (frame, pl_readv_cbk,
+		    FIRST_CHILD (this), FIRST_CHILD (this)->fops->readv,
+		    fd, size, offset);
+
+	return 0;
+}
+
+
+int
+pl_readv (call_frame_t *frame, xlator_t *this,
+	  fd_t *fd, size_t size, off_t offset)
+{
+	posix_locks_private_t *priv = NULL;
+	pl_inode_t            *pl_inode = NULL;
+	pl_rw_req_t           *rw = NULL;
+	posix_lock_t           region = {.list = {0, }, };
+	int                    op_ret = 0;
+	int                    op_errno = 0;
+	char                   allowable = 0;
+
+
+	priv = this->private;
+
+	pl_inode = pl_inode_get (this, fd->inode);
+
+	if (priv->mandatory && pl_inode->mandatory) {
+		region.fl_start   = offset;
+		region.fl_end     = offset + size - 1;
+		region.transport  = frame->root->trans;
+		region.client_pid = frame->root->pid;
+    
+		pthread_mutex_lock (&pl_inode->mutex);
+		{
+			allowable = __rw_allowable (pl_inode, &region,
+						    GF_FOP_READ);
+			if (allowable)
+				goto unlock;
+
+			if (fd->flags & O_NONBLOCK) {
+				gf_log (this->name, GF_LOG_DEBUG,
+					"returning EWOULDBLOCK");
+				op_errno = EWOULDBLOCK;
+				op_ret = -1;
+				goto unlock;
+			}
+
+			rw = CALLOC (1, sizeof (*rw));
+			if (!rw) {
+				gf_log (this->name, GF_LOG_ERROR,
+					"out of memory :(");
+				op_errno = ENOMEM;
+				op_ret = -1;
+				goto unlock;
+			}
+
+			rw->stub = fop_readv_stub (frame, pl_readv_cont,
+						   fd, size, offset);
+			if (!rw->stub) {
+				gf_log (this->name, GF_LOG_ERROR,
+					"out of memory :(");
+				op_errno = ENOMEM;
+				op_ret = -1;
+				free (rw);
+				goto unlock;
+			}
+
+			rw->region = region;
+
+			list_add_tail (&rw->list, &pl_inode->rw_list);
+		}
+	unlock:
+		pthread_mutex_unlock (&pl_inode->mutex);
+
+		goto unwind;
+	}
+
+
+	STACK_WIND (frame, pl_readv_cbk, 
+		    FIRST_CHILD (this), FIRST_CHILD (this)->fops->readv,
+		    fd, size, offset);
+	return 0;
+
+unwind:
+	if (op_ret == -1)
+		STACK_UNWIND (frame, -1, op_errno, NULL, 0, NULL);
+
+	return 0;
+}
+
+
+int
+pl_writev_cont (call_frame_t *frame, xlator_t *this, fd_t *fd,
+		struct iovec *vector, int count, off_t offset)
+{
+	STACK_WIND (frame, pl_writev_cbk,
+		    FIRST_CHILD (this), FIRST_CHILD (this)->fops->writev,
+		    fd, vector, count, offset);
+
+	return 0;
+}
+
+
+int
+pl_writev (call_frame_t *frame, xlator_t *this, fd_t *fd,
+	   struct iovec *vector, int32_t count, off_t offset)
+{
+	posix_locks_private_t *priv = NULL;
+	pl_inode_t            *pl_inode = NULL;
+	pl_rw_req_t           *rw = NULL;
+	posix_lock_t           region = {.list = {0, }, };
+	int                    op_ret = 0;
+	int                    op_errno = 0;
+	char                   allowable = 0;
+
+
+	priv = this->private;
+
+	pl_inode = pl_inode_get (this, fd->inode);
+	if (priv->mandatory && pl_inode->mandatory) {
+		region.fl_start   = offset;
+		region.fl_end     = offset + iov_length (vector, count) - 1;
+		region.transport  = frame->root->trans;
+		region.client_pid = frame->root->pid;
+    
+		pthread_mutex_lock (&pl_inode->mutex);
+		{
+			allowable = __rw_allowable (pl_inode, &region,
+						    GF_FOP_WRITE);
+			if (allowable)
+				goto unlock;
+
+			if (fd->flags & O_NONBLOCK) {
+				gf_log (this->name, GF_LOG_DEBUG,
+					"returning EWOULDBLOCK");
+				op_errno = EWOULDBLOCK;
+				op_ret = -1;
+				goto unlock;
+			}
+
+			rw = CALLOC (1, sizeof (*rw));
+			if (!rw) {
+				gf_log (this->name, GF_LOG_ERROR,
+					"out of memory :(");
+				op_errno = ENOMEM;
+				op_ret = -1;
+				goto unlock;
+			}
+
+			rw->stub = fop_writev_stub (frame, pl_writev_cont,
+						    fd, vector, count, offset);
+			if (!rw->stub) {
+				gf_log (this->name, GF_LOG_ERROR,
+					"out of memory :(");
+				op_errno = ENOMEM;
+				op_ret = -1;
+				free (rw);
+				goto unlock;
+			}
+
+			rw->region = region;
+
+			list_add_tail (&rw->list, &pl_inode->rw_list);
+		}
+	unlock:
+		pthread_mutex_unlock (&pl_inode->mutex);
+
+		goto unwind;
+	}
+
+
+	STACK_WIND (frame, pl_writev_cbk, 
+		    FIRST_CHILD (this), FIRST_CHILD (this)->fops->writev,
+		    fd, vector, count, offset);
+	return 0;
+
+unwind:
+	if (op_ret == -1)
+		STACK_UNWIND (frame, -1, op_errno, NULL, 0, NULL);
+
+	return 0;
+}
+
+
+int
+pl_lk (call_frame_t *frame, xlator_t *this,
+       fd_t *fd, int32_t cmd, struct flock *flock)
+{
+	transport_t           *transport = NULL;
+	pid_t                  client_pid = 0;
+	posix_locks_private_t *priv = NULL;
+	pl_inode_t            *pl_inode = NULL;
+	int                    op_ret = 0;
+	int                    op_errno = 0;
+	int                    can_block = 0;
+	posix_lock_t          *reqlock = NULL;
+	posix_lock_t          *conf = NULL;
+	int                    ret = 0;
+
+	transport  = frame->root->trans;
+	client_pid = frame->root->pid;
+	priv       = this->private;
+
+	pl_inode = pl_inode_get (this, fd->inode);
+	if (!pl_inode) {
+		gf_log (this->name, GF_LOG_ERROR,
+			"out of memory :(");
+		op_ret = -1;
+		op_errno = ENOMEM;
+		goto unwind;
+	}
+
+	reqlock = new_posix_lock (flock, transport, client_pid);
+	if (!reqlock) {
+		gf_log (this->name, GF_LOG_ERROR,
+			"out of memory :(");
+		op_ret = -1;
+		op_errno = ENOMEM;
+		goto unwind;
+	}
+
+	switch (cmd) {
+
+#if F_GETLK != F_GETLK64
+	case F_GETLK64:
+#endif
+	case F_GETLK:
+		conf = pl_getlk (pl_inode, reqlock, GF_LOCK_POSIX);
+		posix_lock_to_flock (conf, flock);
+		__destroy_lock (reqlock);
+
+		break;
+
+#if F_SETLKW != F_SETLKW64
+	case F_SETLKW64:
+#endif
+	case F_SETLKW:
+		can_block = 1;
+		reqlock->frame = frame;
+		reqlock->this  = this;
+		reqlock->fd    = fd;
+
+		/* fall through */
+
+#if F_SETLK != F_SETLK64
+	case F_SETLK64:
+#endif
+	case F_SETLK:
+		memcpy (&reqlock->user_flock, flock, sizeof (struct flock));
+		ret = pl_setlk (this, pl_inode, reqlock,
+				can_block, GF_LOCK_POSIX);
+
+		if (ret == -1) {
+			if (can_block)
+				goto out;
+
+			gf_log (this->name, GF_LOG_DEBUG, "returning EAGAIN");
+			op_ret = -1;
+			op_errno = EAGAIN;
+			__destroy_lock (reqlock);
+		}
+	}
+
+unwind:
+	STACK_UNWIND (frame, op_ret, op_errno, flock);
+out:
+	return 0;
+}
+
+
+/* TODO: this function just logs, no action required?? */
+int
+pl_forget (xlator_t *this,
+	   inode_t *inode)
+{
+	pl_inode_t   *pl_inode = NULL;
+
+	pl_inode = pl_inode_get (this, inode);
+
+	if (!list_empty (&pl_inode->rw_list)) {
+		gf_log (this->name, GF_LOG_CRITICAL,
+			"pending R/W requests found!");
+	}
+
+	if (!list_empty (&pl_inode->ext_list)) {
+		gf_log (this->name, GF_LOG_CRITICAL,
+			"Pending fcntl locks found!");
+	}
+
+	if (!list_empty (&pl_inode->int_list)) {
+		gf_log (this->name, GF_LOG_CRITICAL,
+			"Pending internal locks found!");
+	}
+
+	if (!list_empty (&pl_inode->dir_list)) {
+		gf_log (this->name, GF_LOG_CRITICAL,
+			"Pending entry locks found!");
+	}
+
+	FREE (pl_inode);
+
+	return 0;
+}
+
+
+int
+init (xlator_t *this)
+{
+	posix_locks_private_t *priv = NULL;
+	xlator_list_t         *trav = NULL;
+	data_t                *mandatory = NULL;
+
+	if (!this->children || this->children->next) {
+		gf_log (this->name, GF_LOG_ERROR, 
+			"FATAL: posix-locks should have exactly one child");
+		return -1;
+	}
+
+	if (!this->parents) {
+		gf_log (this->name, GF_LOG_WARNING,
+			"dangling volume. check volfile ");
+	}
+
+	trav = this->children;
+	while (trav->xlator->children)
+		trav = trav->xlator->children;
+
+	if (strncmp ("storage/", trav->xlator->type, 8)) {
+		gf_log (this->name, GF_LOG_ERROR,
+			"'posix-locks' not loaded over storage translator");
+		return -1;
+	}
+
+	priv = CALLOC (1, sizeof (*priv));
+
+	mandatory = dict_get (this->options, "mandatory-locks");
+	if (mandatory) {
+		if (gf_string2boolean (mandatory->data,
+				       &priv->mandatory) == -1) {
+			gf_log (this->name, GF_LOG_ERROR,
+				"'mandatory-locks' takes only boolean "
+				"options");
+			return -1;
+		}
+	}
+
+	this->private = priv;
+	return 0;
+}
+
+
+int
+fini (xlator_t *this)
+{
+	posix_locks_private_t *priv = NULL;
+
+	priv = this->private;
+	free (priv);
+
+	return 0;
+}
+
+
+int
+pl_inodelk (call_frame_t *frame, xlator_t *this, 
+	    loc_t *loc, int32_t cmd, struct flock *flock);
+
+int
+pl_finodelk (call_frame_t *frame, xlator_t *this, 
+	     fd_t *fd, int32_t cmd, struct flock *flock);
+
+int
+pl_entrylk (call_frame_t *frame, xlator_t *this, 
+	    loc_t *loc, const char *basename, 
+	    entrylk_cmd cmd, entrylk_type type);
+
+int
+pl_fentrylk (call_frame_t *frame, xlator_t *this, 
+	     fd_t *fd, const char *basename, 
+	     entrylk_cmd cmd, entrylk_type type);
+
+struct xlator_fops fops = {
+	.create      = pl_create,
+	.truncate    = pl_truncate,
+	.ftruncate   = pl_ftruncate,
+	.open        = pl_open,
+	.readv       = pl_readv,
+	.writev      = pl_writev,
+	.lk          = pl_lk,
+	.inodelk     = pl_inodelk,
+	.finodelk    = pl_finodelk,
+	.entrylk     = pl_entrylk,
+	.fentrylk    = pl_fentrylk,
+	.flush       = pl_flush,
+};
+
+
+struct xlator_mops mops = {
+};
+
+
+struct xlator_cbks cbks = {
+	.forget      = pl_forget,
+};
+
+
+struct volume_options options[] = {
+	{ .key  = { "mandatory-locks", "mandatory" }, 
+	  .type = GF_OPTION_TYPE_BOOL 
+	},
+	{ .key = {NULL} },
+};
diff --git a/xlators/features/locks/tests/unit-test.c b/xlators/features/locks/tests/unit-test.c
new file mode 100644
index 00000000000..6a1bfbf6871
--- /dev/null
+++ b/xlators/features/locks/tests/unit-test.c
@@ -0,0 +1,75 @@
+/*
+  Copyright (c) 2006, 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 _CONFIG_H
+#define _CONFIG_H
+#include "config.h"
+#endif
+
+#include "glusterfs.h"
+#include "compat.h"
+#include "xlator.h"
+#include "inode.h"
+#include "logging.h"
+#include "common-utils.h"
+#include "list.h"
+
+#include "locks.h"
+#include "common.h"
+
+#define expect(cond) if (!(cond)) { goto out; }
+
+extern int lock_name (pl_inode_t *, const char *, entrylk_type);
+extern int unlock_name (pl_inode_t *, const char *, entrylk_type);
+
+int main (int argc, char **argv)
+{
+	int ret = 1;
+	int r = -1;
+
+	pl_inode_t *pinode = CALLOC (sizeof (pl_inode_t), 1);
+	pthread_mutex_init (&pinode->dir_lock_mutex, NULL);
+	INIT_LIST_HEAD (&pinode->gf_dir_locks);
+
+	r = lock_name (pinode, NULL, ENTRYLK_WRLCK); expect (r == 0);
+	{
+		r = lock_name (pinode, "foo", ENTRYLK_WRLCK); expect (r == -EAGAIN);
+	}
+	r = unlock_name (pinode, NULL, ENTRYLK_WRLCK); expect (r == 0);
+
+	r = lock_name (pinode, "foo", ENTRYLK_RDLCK); expect (r == 0);
+	{
+		r = lock_name (pinode, "foo", ENTRYLK_RDLCK); expect (r == 0);
+		{
+			r = lock_name (pinode, "foo", ENTRYLK_WRLCK); expect (r == -EAGAIN);
+		}
+		r = unlock_name (pinode, "foo", ENTRYLK_RDLCK); expect (r == 0);
+	}
+	r = unlock_name (pinode, "foo", ENTRYLK_RDLCK); expect (r == 0);
+	
+	r = lock_name (pinode, "foo", ENTRYLK_WRLCK); expect (r == 0);
+	r = unlock_name (pinode, "foo", ENTRYLK_WRLCK); expect (r == 0);
+
+	r = lock_name (pinode, "baz", ENTRYLK_WRLCK); expect (r == 0);
+	r = lock_name (pinode, "baz", ENTRYLK_RDLCK); expect (r == -EAGAIN);
+
+	ret = 0;
+out:
+	return ret;
+}
diff --git a/xlators/features/path-convertor/Makefile.am b/xlators/features/path-convertor/Makefile.am
new file mode 100644
index 00000000000..d471a3f9243
--- /dev/null
+++ b/xlators/features/path-convertor/Makefile.am
@@ -0,0 +1,3 @@
+SUBDIRS = src
+
+CLEANFILES = 
diff --git a/xlators/features/path-convertor/src/Makefile.am b/xlators/features/path-convertor/src/Makefile.am
new file mode 100644
index 00000000000..1fde1935238
--- /dev/null
+++ b/xlators/features/path-convertor/src/Makefile.am
@@ -0,0 +1,14 @@
+
+xlator_LTLIBRARIES = path-converter.la
+xlatordir = $(libdir)/glusterfs/$(PACKAGE_VERSION)/xlator/features
+
+path_converter_la_LDFLAGS = -module -avoidversion 
+
+path_converter_la_SOURCES = path.c
+path_converter_la_LIBADD = $(top_builddir)/libglusterfs/src/libglusterfs.la 
+
+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/features/path-convertor/src/path.c b/xlators/features/path-convertor/src/path.c
new file mode 100644
index 00000000000..41ef1d8a823
--- /dev/null
+++ b/xlators/features/path-convertor/src/path.c
@@ -0,0 +1,1217 @@
+/*
+  Copyright (c) 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/>.
+*/
+
+/* TODO: add gf_log to all the cases returning errors */
+
+#ifndef _CONFIG_H
+#define _CONFIG_H
+#include "config.h"
+#endif
+
+/**
+ * xlators/features/path-translator:
+ *    This translator converts the path it gets into user specified targets.
+ */
+
+#include <sys/types.h>
+#include <regex.h>
+#include <time.h>
+#include <errno.h>
+#include "glusterfs.h"
+#include "xlator.h"
+
+typedef struct path_private
+{
+	int32_t this_len;
+	int32_t start_off;
+	int32_t end_off;
+	char *this;
+	char *that;
+	char *path;
+	regex_t *preg;
+} path_private_t;
+
+static char *
+name_this_to_that (xlator_t *xl, const char *path, const char *name)
+{
+	path_private_t *priv = xl->private;
+	char priv_path[ZR_PATH_MAX] = {0,};
+	char *tmp_name = NULL;
+	int32_t path_len = strlen (path);
+	int32_t name_len = strlen (name) - ZR_FILE_CONTENT_STRLEN;
+	int32_t total_len = path_len + name_len;
+	int32_t i = 0, j = 0;
+
+	if (path_len >= priv->end_off)
+		return (char *)name;
+
+	if (priv->end_off && (total_len > priv->end_off)) {
+		j = priv->start_off;
+		tmp_name = CALLOC (1, (total_len + ZR_FILE_CONTENT_STRLEN));
+		ERR_ABORT (tmp_name);
+
+		/* Get the complete path for the file first */
+		strcpy (tmp_name, path);
+		strcat (tmp_name, name + ZR_FILE_CONTENT_STRLEN);
+
+		strncpy (priv_path, tmp_name, priv->start_off);
+		for (i = priv->start_off; i < priv->end_off; i++) {
+			if (tmp_name[i] == '/')
+				continue;
+			priv_path[j++] = tmp_name[i];
+		}
+		memcpy ((priv_path + j), 
+			(tmp_name + priv->end_off), 
+			(total_len - priv->end_off));
+		priv_path[(total_len - (priv->end_off - j))] = '\0';
+
+		strcpy (tmp_name, ZR_FILE_CONTENT_STR);
+		strcat (tmp_name, priv_path);
+
+		return tmp_name;
+	}
+
+	return (char *)name;
+}
+
+/* This function should return 
+ *  NULL - 
+ *  converted path - if path match
+ *  same path - if it doesn't match
+ */
+static char *
+path_this_to_that (xlator_t *xl, const char *path)
+{
+	path_private_t *priv = xl->private;
+	char *priv_path = NULL;
+	int32_t path_len = strlen (path);
+	int32_t i = 0, j = 0;
+
+	if (priv->end_off && (path_len > priv->start_off)) {
+		priv_path = CALLOC (1, path_len);
+		ERR_ABORT (priv_path);
+
+		if (priv->start_off && (path_len > priv->start_off))
+			memcpy (priv_path, path, priv->start_off);
+		if (path_len > priv->end_off) {
+			j = priv->start_off;
+			for (i = priv->start_off; i < priv->end_off; i++) {
+				if (path[i] == '/')
+					continue;
+				priv_path[j++] = path[i];
+			}
+			memcpy ((priv_path + j), 
+				(path + priv->end_off), 
+				(path_len - priv->end_off));
+			priv_path[(path_len - (priv->end_off - j))] = '\0';
+		}
+		return priv_path;
+	}
+	return (char *)path;
+}
+
+int32_t 
+path_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)
+{
+	STACK_UNWIND (frame, op_ret, op_errno, fd, inode, buf);
+	return 0;
+}
+
+int32_t 
+path_open_cbk (call_frame_t *frame,
+	       void *cookie,
+	       xlator_t *this,
+	       int32_t op_ret,
+	       int32_t op_errno,
+	       fd_t *fd)
+{
+	STACK_UNWIND (frame, op_ret, op_errno, fd);
+	return 0;
+}
+
+int32_t 
+path_getdents_cbk (call_frame_t *frame,
+		   void *cookie,
+		   xlator_t *this,
+		   int32_t op_ret,
+		   int32_t op_errno,
+		   dir_entry_t *entries,
+		   int32_t count)
+{
+	STACK_UNWIND (frame, op_ret, op_errno, entries, count);
+	return 0;
+}
+
+int32_t 
+path_readdir_cbk (call_frame_t *frame,
+		  void *cookie,
+		  xlator_t *this,
+		  int32_t op_ret,
+		  int32_t op_errno,
+		  gf_dirent_t *buf)
+{
+	STACK_UNWIND (frame, op_ret, op_errno, buf);
+	return 0;
+}
+
+
+int32_t 
+path_readlink_cbk (call_frame_t *frame,
+		   void *cookie,
+		   xlator_t *this,
+		   int32_t op_ret,
+		   int32_t op_errno,
+		   const char *buf)
+{
+	STACK_UNWIND (frame, op_ret, op_errno, buf);
+	return 0;
+}
+
+int32_t 
+path_lookup_cbk (call_frame_t *frame,
+		 void *cookie,
+		 xlator_t *this,
+		 int32_t op_ret,
+		 int32_t op_errno,
+		 inode_t *inode,
+		 struct stat *buf,
+		 dict_t *xattr)
+{
+	STACK_UNWIND (frame, op_ret, op_errno, inode, buf, xattr);
+	return 0;
+}
+
+
+int32_t 
+path_symlink_cbk (call_frame_t *frame,
+		  void *cookie,
+		  xlator_t *this,
+		  int32_t op_ret,
+		  int32_t op_errno,
+		  inode_t *inode,
+		  struct stat *buf)
+{
+	STACK_UNWIND (frame, op_ret, op_errno, inode, buf);
+	return 0;
+}
+
+int32_t 
+path_mknod_cbk (call_frame_t *frame,
+		void *cookie,
+		xlator_t *this,
+		int32_t op_ret,
+		int32_t op_errno,
+		inode_t *inode,
+		struct stat *buf)
+{
+	STACK_UNWIND (frame, op_ret, op_errno, inode, buf);
+	return 0;
+}
+  
+
+int32_t 
+path_mkdir_cbk (call_frame_t *frame,
+		void *cookie,
+		xlator_t *this,
+		int32_t op_ret,
+		int32_t op_errno,
+		inode_t *inode,
+		struct stat *buf)
+{
+	STACK_UNWIND (frame, op_ret, op_errno, inode, buf);
+	return 0;
+}
+  
+int32_t 
+path_link_cbk (call_frame_t *frame,
+	       void *cookie,
+	       xlator_t *this,
+	       int32_t op_ret,
+	       int32_t op_errno,
+	       inode_t *inode,
+	       struct stat *buf)
+{
+	STACK_UNWIND (frame, op_ret, op_errno, inode, buf);
+	return 0;
+}
+
+int32_t 
+path_opendir_cbk (call_frame_t *frame,
+		  void *cookie,
+		  xlator_t *this,
+		  int32_t op_ret,
+		  int32_t op_errno,
+		  fd_t *fd)
+{
+	STACK_UNWIND (frame, op_ret, op_errno, fd);
+	return 0;
+}
+
+
+int32_t 
+path_common_buf_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;
+}
+
+int32_t 
+path_common_dict_cbk (call_frame_t *frame,
+		      void *cookie,
+		      xlator_t *this,
+		      int32_t op_ret,
+		      int32_t op_errno,
+		      dict_t *dict)
+{
+	STACK_UNWIND (frame, op_ret, op_errno, dict);
+	return 0;
+}
+
+int32_t 
+path_common_cbk (call_frame_t *frame,
+		 void *cookie,
+		 xlator_t *this,
+		 int32_t op_ret,
+		 int32_t op_errno)
+{
+	STACK_UNWIND (frame, op_ret, op_errno);
+	return 0;
+}
+
+/* */
+int32_t 
+path_lookup (call_frame_t *frame,
+	     xlator_t *this,
+	     loc_t *loc,
+	     dict_t *xattr_req)
+{
+	char *loc_path = (char *)loc->path;
+	char *tmp_path = NULL;
+	
+	if (!(tmp_path = path_this_to_that (this, loc->path))) {
+		STACK_UNWIND (frame, -1, ENOENT, NULL, NULL);
+		return 0;
+	}
+	loc->path = tmp_path;
+
+	STACK_WIND (frame, path_lookup_cbk,
+		    FIRST_CHILD(this), 
+		    FIRST_CHILD(this)->fops->lookup, 
+		    loc, xattr_req);
+
+	loc->path = loc_path;	
+	if (tmp_path != loc_path)
+		FREE (tmp_path);
+
+	return 0;
+}
+
+int32_t 
+path_stat (call_frame_t *frame,
+	   xlator_t *this,
+	   loc_t *loc)
+{
+	char *loc_path = (char *)loc->path;
+	char *tmp_path = NULL;
+	
+	if (!(tmp_path = path_this_to_that (this, loc->path))) {
+		STACK_UNWIND (frame, -1, ENOENT, NULL, NULL);
+		return 0;
+	}
+	loc->path = tmp_path;
+
+	STACK_WIND (frame, 
+		    path_common_buf_cbk, 
+		    FIRST_CHILD(this), 
+		    FIRST_CHILD(this)->fops->stat, 
+		    loc);
+  
+	loc->path = loc_path;	
+	if (tmp_path != loc_path)
+		FREE (tmp_path);
+
+	return 0;
+}
+
+int32_t 
+path_readlink (call_frame_t *frame,
+	       xlator_t *this,
+	       loc_t *loc,
+	       size_t size)
+{
+	char *loc_path = (char *)loc->path;
+	char *tmp_path = NULL;
+	
+	if (!(tmp_path = path_this_to_that (this, loc->path))) {
+		STACK_UNWIND (frame, -1, ENOENT, NULL, NULL);
+		return 0;
+	}
+	loc->path = tmp_path;
+
+	STACK_WIND (frame, 
+		    path_readlink_cbk,
+		    FIRST_CHILD(this), 
+		    FIRST_CHILD(this)->fops->readlink, 
+		    loc, 
+		    size);
+  
+	loc->path = loc_path;	
+	if (tmp_path != loc_path)
+		FREE (tmp_path);
+
+	return 0;
+}
+
+int32_t 
+path_mknod (call_frame_t *frame,
+	    xlator_t *this,
+	    loc_t *loc,
+	    mode_t mode,
+	    dev_t dev)
+{
+	char *loc_path = (char *)loc->path;
+	char *tmp_path = NULL;
+	
+	if (!(tmp_path = path_this_to_that (this, loc->path))) {
+		STACK_UNWIND (frame, -1, ENOENT, NULL, NULL);
+		return 0;
+	}
+	loc->path = tmp_path;
+
+	STACK_WIND (frame, 
+		    path_mknod_cbk,
+		    FIRST_CHILD(this), 
+		    FIRST_CHILD(this)->fops->mknod, 
+		    loc, 
+		    mode, 
+		    dev);
+
+	loc->path = loc_path;	
+	if (tmp_path != loc_path)
+		FREE (tmp_path);
+
+	return 0;
+}
+
+int32_t 
+path_mkdir (call_frame_t *frame,
+	    xlator_t *this,
+	    loc_t *loc,
+	    mode_t mode)
+{
+	char *loc_path = (char *)loc->path;
+	char *tmp_path = NULL;
+	
+	if (!(tmp_path = path_this_to_that (this, loc->path))) {
+		STACK_UNWIND (frame, -1, ENOENT, NULL, NULL);
+		return 0;
+	}
+	loc->path = tmp_path;
+
+	STACK_WIND (frame, 
+		    path_mkdir_cbk,
+		    FIRST_CHILD(this), 
+		    FIRST_CHILD(this)->fops->mkdir, 
+		    loc, 
+		    mode);
+
+	loc->path = loc_path;	
+	if (tmp_path != loc_path)
+		FREE (tmp_path);
+
+	return 0;
+}
+
+int32_t 
+path_unlink (call_frame_t *frame,
+	     xlator_t *this,
+	     loc_t *loc)
+{
+	char *loc_path = (char *)loc->path;
+	char *tmp_path = NULL;
+	
+	if (!(tmp_path = path_this_to_that (this, loc->path))) {
+		STACK_UNWIND (frame, -1, ENOENT, NULL, NULL);
+		return 0;
+	}
+	loc->path = tmp_path;
+
+	STACK_WIND (frame, 
+		    path_common_cbk, 
+		    FIRST_CHILD(this), 
+		    FIRST_CHILD(this)->fops->unlink, 
+		    loc);
+
+	loc->path = loc_path;	
+	if (tmp_path != loc_path)
+		FREE (tmp_path);
+
+	return 0;
+}
+
+int32_t 
+path_rmdir (call_frame_t *frame,
+	    xlator_t *this,
+	    loc_t *loc)
+{
+	char *loc_path = (char *)loc->path;
+	char *tmp_path = NULL;
+	
+	if (!(tmp_path = path_this_to_that (this, loc->path))) {
+		STACK_UNWIND (frame, -1, ENOENT, NULL, NULL);
+		return 0;
+	}
+	loc->path = tmp_path;
+
+	STACK_WIND (frame, 
+		    path_common_cbk,
+		    FIRST_CHILD(this), 
+		    FIRST_CHILD(this)->fops->rmdir, 
+		    loc);
+  
+	loc->path = loc_path;	
+	if (tmp_path != loc_path)
+		FREE (tmp_path);
+
+	return 0;
+}
+
+int32_t 
+path_symlink (call_frame_t *frame,
+	      xlator_t *this,
+	      const char *linkpath,
+	      loc_t *loc)
+{
+	char *loc_path = (char *)loc->path;
+	char *tmp_path = NULL;
+	
+	if (!(tmp_path = path_this_to_that (this, loc->path))) {
+		STACK_UNWIND (frame, -1, ENOENT, NULL, NULL);
+		return 0;
+	}
+	loc->path = tmp_path;
+
+	STACK_WIND (frame, 
+		    path_symlink_cbk,
+		    FIRST_CHILD(this), 
+		    FIRST_CHILD(this)->fops->symlink, 
+		    linkpath,
+		    loc);
+  
+	loc->path = loc_path;	
+	if (tmp_path != loc_path)
+		FREE (tmp_path);
+
+	return 0;
+}
+
+int32_t 
+path_rename (call_frame_t *frame,
+	     xlator_t *this,
+	     loc_t *oldloc,
+	     loc_t *newloc)
+{  
+	char *oldloc_path = (char *)oldloc->path;
+	char *tmp_oldloc_path = NULL;
+
+	char *newloc_path = (char *)newloc->path;
+	char *tmp_newloc_path = NULL;
+	
+	if (!(tmp_oldloc_path = path_this_to_that (this, oldloc->path))) {
+		STACK_UNWIND (frame, -1, ENOENT, NULL, NULL);
+		return 0;
+	}
+	oldloc->path = tmp_oldloc_path;
+
+	if (!(tmp_newloc_path = path_this_to_that (this, newloc->path))) {
+		STACK_UNWIND (frame, -1, ENOENT, NULL, NULL);
+		return 0;
+	}
+	newloc->path = tmp_newloc_path;
+
+	STACK_WIND (frame, 
+		    path_common_buf_cbk,
+		    FIRST_CHILD(this), 
+		    FIRST_CHILD(this)->fops->rename, 
+		    oldloc,
+		    newloc);
+  
+	oldloc->path = oldloc_path;	
+	if (tmp_oldloc_path != oldloc_path)
+		FREE (tmp_oldloc_path);
+
+	newloc->path = newloc_path;	
+	if (tmp_newloc_path != newloc_path)
+		FREE (tmp_newloc_path);
+
+	return 0;
+}
+
+int32_t 
+path_link (call_frame_t *frame,
+	   xlator_t *this,
+	   loc_t *oldloc,
+	   loc_t *newloc)
+{
+	char *oldloc_path = (char *)oldloc->path;
+	char *tmp_oldloc_path = NULL;
+
+	char *newloc_path = (char *)newloc->path;
+	char *tmp_newloc_path = NULL;
+	
+	if (!(tmp_oldloc_path = path_this_to_that (this, oldloc->path))) {
+		STACK_UNWIND (frame, -1, ENOENT, NULL, NULL);
+		return 0;
+	}
+	oldloc->path = tmp_oldloc_path;
+
+	if (!(tmp_newloc_path = path_this_to_that (this, newloc->path))) {
+		STACK_UNWIND (frame, -1, ENOENT, NULL, NULL);
+		return 0;
+	}
+	newloc->path = tmp_newloc_path;
+
+	STACK_WIND (frame, 
+		    path_link_cbk, 
+		    FIRST_CHILD(this), 
+		    FIRST_CHILD(this)->fops->link, 
+		    oldloc, 
+		    newloc);
+
+	oldloc->path = oldloc_path;	
+	if (tmp_oldloc_path != oldloc_path)
+		FREE (tmp_oldloc_path);
+
+	newloc->path = newloc_path;	
+	if (tmp_newloc_path != newloc_path)
+		FREE (tmp_newloc_path);
+
+	return 0;
+}
+
+int32_t 
+path_chmod (call_frame_t *frame,
+	    xlator_t *this,
+	    loc_t *loc,
+	    mode_t mode)
+{
+	char *loc_path = (char *)loc->path;
+	char *tmp_path = NULL;
+	
+	if (!(tmp_path = path_this_to_that (this, loc->path))) {
+		STACK_UNWIND (frame, -1, ENOENT, NULL, NULL);
+		return 0;
+	}
+	loc->path = tmp_path;
+
+	STACK_WIND (frame, 
+		    path_common_buf_cbk, 
+		    FIRST_CHILD(this), 
+		    FIRST_CHILD(this)->fops->chmod, 
+		    loc, 
+		    mode);
+  
+	loc->path = loc_path;	
+	if (tmp_path != loc_path)
+		FREE (tmp_path);
+
+	return 0;
+}
+
+int32_t 
+path_chown (call_frame_t *frame,
+	    xlator_t *this,
+	    loc_t *loc,
+	    uid_t uid,
+	    gid_t gid)
+{
+	char *loc_path = (char *)loc->path;
+	char *tmp_path = NULL;
+	
+	if (!(tmp_path = path_this_to_that (this, loc->path))) {
+		STACK_UNWIND (frame, -1, ENOENT, NULL, NULL);
+		return 0;
+	}
+	loc->path = tmp_path;
+
+	STACK_WIND (frame, 
+		    path_common_buf_cbk, 
+		    FIRST_CHILD(this), 
+		    FIRST_CHILD(this)->fops->chown, 
+		    loc, 
+		    uid,
+		    gid);
+
+	loc->path = loc_path;	
+	if (tmp_path != loc_path)
+		FREE (tmp_path);
+
+	return 0;
+}
+
+int32_t 
+path_truncate (call_frame_t *frame,
+	       xlator_t *this,
+	       loc_t *loc,
+	       off_t offset)
+{
+	char *loc_path = (char *)loc->path;
+	char *tmp_path = NULL;
+	
+	if (!(tmp_path = path_this_to_that (this, loc->path))) {
+		STACK_UNWIND (frame, -1, ENOENT, NULL, NULL);
+		return 0;
+	}
+	loc->path = tmp_path;
+
+	STACK_WIND (frame, 
+		    path_common_buf_cbk, 
+		    FIRST_CHILD(this), 
+		    FIRST_CHILD(this)->fops->truncate, 
+		    loc, 
+		    offset);
+  
+	loc->path = loc_path;	
+	if (tmp_path != loc_path)
+		FREE (tmp_path);
+
+	return 0;
+}
+
+int32_t 
+path_utimens (call_frame_t *frame,
+	      xlator_t *this,
+	      loc_t *loc,
+	      struct timespec tv[2])
+{
+	char *loc_path = (char *)loc->path;
+	char *tmp_path = NULL;
+	
+	if (!(tmp_path = path_this_to_that (this, loc->path))) {
+		STACK_UNWIND (frame, -1, ENOENT, NULL, NULL);
+		return 0;
+	}
+	loc->path = tmp_path;
+
+	STACK_WIND (frame, 
+		    path_common_buf_cbk, 
+		    FIRST_CHILD(this), 
+		    FIRST_CHILD(this)->fops->utimens, 
+		    loc, 
+		    tv);
+
+	loc->path = loc_path;	
+	if (tmp_path != loc_path)
+		FREE (tmp_path);
+
+	return 0;
+}
+
+int32_t 
+path_open (call_frame_t *frame,
+	   xlator_t *this,
+	   loc_t *loc,
+	   int32_t flags,
+	   fd_t *fd)
+{
+	char *loc_path = (char *)loc->path;
+	char *tmp_path = NULL;
+	
+	if (!(tmp_path = path_this_to_that (this, loc->path))) {
+		STACK_UNWIND (frame, -1, ENOENT, NULL, NULL);
+		return 0;
+	}
+	loc->path = tmp_path;
+
+	STACK_WIND (frame, 
+		    path_open_cbk, 
+		    FIRST_CHILD(this), 
+		    FIRST_CHILD(this)->fops->open, 
+		    loc, 
+		    flags,
+		    fd);
+
+	loc->path = loc_path;	
+	if (tmp_path != loc_path)
+		FREE (tmp_path);
+
+	return 0;
+}
+
+int32_t 
+path_create (call_frame_t *frame,
+	     xlator_t *this,
+	     loc_t *loc,
+	     int32_t flags,
+	     mode_t mode,
+	     fd_t *fd)
+{
+	char *loc_path = (char *)loc->path;
+	char *tmp_path = NULL;
+	
+	if (!(tmp_path = path_this_to_that (this, loc->path))) {
+		STACK_UNWIND (frame, -1, ENOENT, NULL, NULL);
+		return 0;
+	}
+	loc->path = tmp_path;
+
+	STACK_WIND (frame, 
+		    path_create_cbk, 
+		    FIRST_CHILD(this), 
+		    FIRST_CHILD(this)->fops->create, 
+		    loc, 
+		    flags,
+		    mode,
+		    fd);
+
+	loc->path = loc_path;	
+	if (tmp_path != loc_path)
+		FREE (tmp_path);
+
+	return 0;
+}
+
+int32_t 
+path_setxattr (call_frame_t *frame,
+	       xlator_t *this,
+	       loc_t *loc,
+	       dict_t *dict,
+	       int32_t flags)
+{
+	char *tmp_name = NULL;
+	data_pair_t *trav = dict->members_list;
+	char *loc_path = (char *)loc->path;
+	char *tmp_path = NULL;
+	
+	if (!(tmp_path = path_this_to_that (this, loc->path))) {
+		STACK_UNWIND (frame, -1, ENOENT, NULL, NULL);
+		return 0;
+	}
+	loc->path = tmp_path;
+
+	if (ZR_FILE_CONTENT_REQUEST(trav->key)) {
+		tmp_name = name_this_to_that (this, loc->path, trav->key);
+		if (tmp_name != trav->key) {
+			trav->key = tmp_name;
+		} else {
+			tmp_name = NULL;
+		}
+	}
+
+	STACK_WIND (frame, 
+		    path_common_cbk, 
+		    FIRST_CHILD(this), 
+		    FIRST_CHILD(this)->fops->setxattr, 
+		    loc, 
+		    dict,
+		    flags);
+
+	loc->path = loc_path;	
+	if (tmp_path != loc_path)
+		FREE (tmp_path);
+
+	if (tmp_name)
+		FREE (tmp_name);
+
+	return 0;
+}
+
+int32_t 
+path_getxattr (call_frame_t *frame,
+	       xlator_t *this,
+	       loc_t *loc,
+	       const char *name)
+{
+	char *tmp_name = (char *)name;
+	char *loc_path = (char *)loc->path;
+	char *tmp_path = NULL;
+	
+	if (!(tmp_path = path_this_to_that (this, loc->path))) {
+		STACK_UNWIND (frame, -1, ENOENT, NULL, NULL);
+		return 0;
+	}
+	loc->path = tmp_path;
+
+	if (ZR_FILE_CONTENT_REQUEST(name)) {
+		tmp_name = name_this_to_that (this, loc->path, name);
+	}
+
+	STACK_WIND (frame, 
+		    path_common_dict_cbk, 
+		    FIRST_CHILD(this), 
+		    FIRST_CHILD(this)->fops->getxattr,
+		    loc, 
+		    tmp_name);
+
+	loc->path = loc_path;	
+	if (tmp_path != loc_path)
+		FREE (tmp_path);
+
+	if (tmp_name != name)
+		FREE (tmp_name);
+
+	return 0;
+}
+
+int32_t 
+path_removexattr (call_frame_t *frame,
+		  xlator_t *this,
+		  loc_t *loc,
+		  const char *name)
+{
+	char *tmp_name = (char *)name;
+	char *loc_path = (char *)loc->path;
+	char *tmp_path = NULL;
+	
+	if (!(tmp_path = path_this_to_that (this, loc->path))) {
+		STACK_UNWIND (frame, -1, ENOENT, NULL, NULL);
+		return 0;
+	}
+	loc->path = tmp_path;
+
+	if (ZR_FILE_CONTENT_REQUEST(name)) {
+		tmp_name = name_this_to_that (this, loc->path, name);
+	}
+
+	STACK_WIND (frame, 
+		    path_common_cbk, 
+		    FIRST_CHILD(this), 
+		    FIRST_CHILD(this)->fops->removexattr, 
+		    loc, 
+		    tmp_name);
+
+	loc->path = loc_path;	
+	if (tmp_path != loc_path)
+		FREE (tmp_path);
+
+	if (tmp_name != name)
+		FREE (tmp_name);
+
+	return 0;
+}
+
+int32_t 
+path_opendir (call_frame_t *frame,
+	      xlator_t *this,
+	      loc_t *loc,
+	      fd_t *fd)
+{
+	char *loc_path = (char *)loc->path;
+	char *tmp_path = NULL;
+	
+	if (!(tmp_path = path_this_to_that (this, loc->path))) {
+		STACK_UNWIND (frame, -1, ENOENT, NULL, NULL);
+		return 0;
+	}
+	loc->path = tmp_path;
+
+	STACK_WIND (frame, 
+		    path_opendir_cbk, 
+		    FIRST_CHILD(this), 
+		    FIRST_CHILD(this)->fops->opendir, 
+		    loc, 
+		    fd);
+
+	loc->path = loc_path;	
+	if (tmp_path != loc_path)
+		FREE (tmp_path);
+
+	return 0;
+}
+
+int32_t 
+path_access (call_frame_t *frame,
+	     xlator_t *this,
+	     loc_t *loc,
+	     int32_t mask)
+{
+	char *loc_path = (char *)loc->path;
+	char *tmp_path = NULL;
+	
+	if (!(tmp_path = path_this_to_that (this, loc->path))) {
+		STACK_UNWIND (frame, -1, ENOENT, NULL, NULL);
+		return 0;
+	}
+	loc->path = tmp_path;
+
+	STACK_WIND (frame, 
+		    path_common_cbk, 
+		    FIRST_CHILD(this), 
+		    FIRST_CHILD(this)->fops->access, 
+		    loc, 
+		    mask);
+
+	loc->path = loc_path;
+	if (tmp_path != loc_path)
+		FREE (tmp_path);
+
+	return 0;
+}
+
+int32_t
+path_checksum_cbk (call_frame_t *frame,
+		   void *cookie,
+		   xlator_t *this,
+		   int32_t op_ret,
+		   int32_t op_errno,
+		   uint8_t *fchecksum,
+		   uint8_t *dchecksum)
+{
+	STACK_UNWIND (frame, op_ret, op_errno, fchecksum, dchecksum);
+	return 0;
+}
+
+int32_t
+path_checksum (call_frame_t *frame,
+	       xlator_t *this,
+	       loc_t *loc,
+	       int32_t flag)
+{
+	char *loc_path = (char *)loc->path;
+	char *tmp_path = NULL;
+	
+	if (!(tmp_path = path_this_to_that (this, loc->path))) {
+		STACK_UNWIND (frame, -1, ENOENT, NULL, NULL);
+		return 0;
+	}
+	loc->path = tmp_path;
+
+	STACK_WIND (frame,
+		    path_checksum_cbk,
+		    FIRST_CHILD(this), 
+		    FIRST_CHILD(this)->fops->checksum, 
+		    loc, 
+		    flag);
+
+	loc->path = loc_path;	
+	if (tmp_path != loc_path)
+		FREE (tmp_path);
+
+	return 0;
+}
+
+
+int32_t
+path_entrylk (call_frame_t *frame, xlator_t *this,
+	      loc_t *loc, const char *basename,
+	      entrylk_cmd cmd, entrylk_type type)
+{
+	char *loc_path = (char *)loc->path;
+	char *tmp_path = NULL;
+	
+	if (!(tmp_path = path_this_to_that (this, loc->path))) {
+		STACK_UNWIND (frame, -1, ENOENT, NULL, NULL);
+		return 0;
+	}
+	loc->path = tmp_path;
+
+	STACK_WIND (frame, path_common_cbk,
+		    FIRST_CHILD(this),
+		    FIRST_CHILD(this)->fops->entrylk,
+		    loc, basename, cmd, type);
+
+	loc->path = loc_path;	
+	if (tmp_path != loc_path)
+		FREE (tmp_path);
+
+	return 0;
+}
+
+int32_t
+path_inodelk (call_frame_t *frame, xlator_t *this,
+		 loc_t *loc, int32_t cmd, struct flock *lock)
+{
+	char *loc_path = (char *)loc->path;
+	char *tmp_path = NULL;
+	
+	if (!(tmp_path = path_this_to_that (this, loc->path))) {
+		STACK_UNWIND (frame, -1, ENOENT, NULL, NULL);
+		return 0;
+	}
+	loc->path = tmp_path;
+
+	STACK_WIND (frame,
+		    path_common_cbk,
+		    FIRST_CHILD(this),
+		    FIRST_CHILD(this)->fops->inodelk,
+		    loc, cmd, lock);
+
+	loc->path = loc_path;	
+	if (tmp_path != loc_path)
+		FREE (tmp_path);
+
+	return 0;
+}
+
+
+int32_t
+path_xattrop (call_frame_t *frame,
+	      xlator_t *this,
+	      loc_t *loc,
+	      gf_xattrop_flags_t flags,
+	      dict_t *dict)
+{
+	char *loc_path = (char *)loc->path;
+	char *tmp_path = NULL;
+	
+	if (!(tmp_path = path_this_to_that (this, loc->path))) {
+		STACK_UNWIND (frame, -1, ENOENT, NULL, NULL);
+		return 0;
+	}
+	loc->path = tmp_path;
+
+	STACK_WIND (frame,
+		    path_common_dict_cbk,
+		    FIRST_CHILD(this),
+		    FIRST_CHILD(this)->fops->xattrop,
+		    loc,
+		    flags,
+		    dict);
+
+	loc->path = loc_path;	
+	if (tmp_path != loc_path)
+		FREE (tmp_path);
+
+	return 0;
+}
+
+
+int32_t 
+init (xlator_t *this)
+{
+	dict_t *options = this->options;
+	path_private_t *priv = NULL;
+
+	if (!this->children || this->children->next) {
+		gf_log (this->name, GF_LOG_ERROR, 
+			"path translator requires exactly one subvolume");
+		return -1;
+	}
+    
+	if (!this->parents) {
+		gf_log (this->name, GF_LOG_WARNING,
+			"dangling volume. check volfile ");
+	}
+  
+	priv = CALLOC (1, sizeof (*priv));
+	ERR_ABORT (priv);
+	if (dict_get (options, "start-offset")) {
+		priv->start_off = data_to_int32 (dict_get (options, 
+							   "start-offset"));
+	}
+	if (dict_get (options, "end-offset")) {
+		priv->end_off = data_to_int32 (dict_get (options, 
+							 "end-offset"));
+	}
+
+	if (dict_get (options, "regex")) {
+		int32_t ret = 0;
+		priv->preg = CALLOC (1, sizeof (regex_t));
+		ERR_ABORT (priv->preg);
+		ret = regcomp (priv->preg, 
+			       data_to_str (dict_get (options, "regex")), 
+			       REG_EXTENDED);
+		if (ret) {
+			gf_log (this->name, GF_LOG_ERROR, 
+				"Failed to compile the 'option regex'");
+			FREE (priv);
+			return -1;
+		}
+		if (dict_get (options, "replace-with")) {
+			priv->that = data_to_str (dict_get (options, 
+							    "replace-with"));
+		} else {
+			priv->that = "";
+		}
+	}
+
+	this->private = priv;
+	return 0;
+}
+
+void
+fini (xlator_t *this)
+{
+	return;
+}
+
+struct xlator_fops fops = {
+	.stat        = path_stat,
+	.readlink    = path_readlink,
+	.mknod       = path_mknod,
+	.mkdir       = path_mkdir,
+	.unlink      = path_unlink,
+	.rmdir       = path_rmdir,
+	.symlink     = path_symlink,
+	.rename      = path_rename,
+	.link        = path_link,
+	.chmod       = path_chmod,
+	.chown       = path_chown,
+	.truncate    = path_truncate,
+	.utimens     = path_utimens,
+	.open        = path_open,
+	.setxattr    = path_setxattr,
+	.getxattr    = path_getxattr,
+	.removexattr = path_removexattr,
+	.opendir     = path_opendir,
+	.access      = path_access,
+	.create      = path_create,
+	.lookup      = path_lookup,
+	.checksum    = path_checksum,
+	.xattrop     = path_xattrop,
+	.entrylk     = path_entrylk,
+	.inodelk     = path_inodelk,
+};
+
+
+struct xlator_mops mops = {
+};
+
+
+struct xlator_cbks cbks = {
+};
+
+struct volume_options options[] = { 
+	{ .key  = {"start-offset"}, 
+	  .type = GF_OPTION_TYPE_INT, 
+	  .min  = 0, 
+	  .max  = 4095 
+	},
+	{ .key  = {"end-offset"}, 
+	  .type = GF_OPTION_TYPE_INT, 
+	  .min  = 1, 
+	  .max  = 4096 
+	},
+	{ .key  = {"replace-with"}, 
+	  .type = GF_OPTION_TYPE_ANY 
+	},
+	{ .key  = {NULL} },
+};
diff --git a/xlators/features/quota/Makefile.am b/xlators/features/quota/Makefile.am
new file mode 100644
index 00000000000..d471a3f9243
--- /dev/null
+++ b/xlators/features/quota/Makefile.am
@@ -0,0 +1,3 @@
+SUBDIRS = src
+
+CLEANFILES = 
diff --git a/xlators/features/quota/src/Makefile.am b/xlators/features/quota/src/Makefile.am
new file mode 100644
index 00000000000..886d839643c
--- /dev/null
+++ b/xlators/features/quota/src/Makefile.am
@@ -0,0 +1,13 @@
+xlator_LTLIBRARIES = quota.la
+xlatordir = $(libdir)/glusterfs/$(PACKAGE_VERSION)/xlator/features
+
+quota_la_LDFLAGS = -module -avoidversion
+
+quota_la_SOURCES = quota.c
+quota_la_LIBADD = $(top_builddir)/libglusterfs/src/libglusterfs.la 
+
+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/features/quota/src/quota.c b/xlators/features/quota/src/quota.c
new file mode 100644
index 00000000000..c898899b5e3
--- /dev/null
+++ b/xlators/features/quota/src/quota.c
@@ -0,0 +1,1056 @@
+/*
+  Copyright (c) 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 <sys/time.h>
+
+#include "xlator.h"
+#include "defaults.h"
+#include "common-utils.h"
+
+struct quota_local {
+	struct stat    stbuf;
+	inode_t       *inode;
+	char          *path;
+	fd_t          *fd;
+	off_t          offset;
+	int32_t        count;
+	struct iovec  *vector;
+	dict_t        *refs;
+	loc_t          loc;
+};
+
+
+struct quota_priv {
+	char       only_first_time;          /* Used to make sure a call is done only one time */
+	gf_lock_t  lock;                     /* Used while updating variables */
+
+	uint64_t   disk_usage_limit;         /* Used for Disk usage quota */
+	uint64_t   current_disk_usage;       /* Keep the current usage value */
+
+	uint32_t   min_free_disk_limit;        /* user specified limit, in %*/
+	uint32_t   current_free_disk;          /* current free disk space available, in % */
+	uint32_t   refresh_interval;           /* interval in seconds */
+	uint32_t   min_disk_last_updated_time; /* used for interval calculation */	
+};
+
+
+int
+quota_statvfs_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
+		   int32_t op_ret, int32_t op_errno, struct statvfs *stbuf)
+{
+	struct quota_priv *priv = this->private;
+	
+	if (op_ret >= 0) {
+		priv->current_free_disk =
+			(stbuf->f_bavail * 100) / stbuf->f_blocks;
+	}
+
+	STACK_DESTROY (frame->root);
+	return 0;
+}
+
+
+static void
+build_root_loc (xlator_t *this, loc_t *loc)
+{
+	loc->path = "/";
+}
+
+
+void
+gf_quota_usage_subtract (xlator_t *this, size_t size)
+{
+	struct quota_priv *priv = NULL;
+
+	priv = this->private;
+
+	LOCK (&priv->lock);
+	{
+		if (priv->current_disk_usage < size)
+			priv->current_disk_usage = 0;
+		else
+			priv->current_disk_usage -= size;
+	}
+	UNLOCK (&priv->lock);
+}
+
+
+void
+gf_quota_usage_add (xlator_t *this, size_t size)
+{
+	struct quota_priv *priv = this->private;
+
+	LOCK (&priv->lock);
+	{
+		priv->current_disk_usage += size;
+	}
+	UNLOCK (&priv->lock);
+}
+
+
+void 
+gf_quota_update_current_free_disk (xlator_t *this)
+{
+	call_frame_t *frame = NULL;
+	call_pool_t   *pool = NULL;
+	loc_t          loc;
+
+	pool  = this->ctx->pool;
+	frame = create_frame (this, pool);
+  
+	build_root_loc (this, &loc);
+
+	STACK_WIND (frame, quota_statvfs_cbk,
+		    this->children->xlator,
+		    this->children->xlator->fops->statfs, &loc);
+
+	return ;
+}
+
+
+int
+gf_quota_check_free_disk (xlator_t *this) 
+{
+        struct quota_priv * priv = NULL;
+	struct timeval tv = {0, 0};
+
+	priv = this->private;
+	if (priv->min_free_disk_limit) {
+		gettimeofday (&tv, NULL);
+		if (tv.tv_sec > (priv->refresh_interval + 
+				 priv->min_disk_last_updated_time)) {
+			priv->min_disk_last_updated_time = tv.tv_sec;
+			gf_quota_update_current_free_disk (this);
+		}
+		if (priv->current_free_disk <= priv->min_free_disk_limit)
+			return -1;
+	}
+
+	return 0;
+}
+
+
+int
+quota_truncate_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
+		    int32_t op_ret, int32_t op_errno, struct stat *buf)
+{
+	struct quota_priv *priv = this->private;
+	struct quota_local *local = NULL;
+
+	local = frame->local;
+
+	if ((op_ret >= 0) && priv->disk_usage_limit) {
+		gf_quota_usage_subtract (this, (local->stbuf.st_blocks -
+						buf->st_blocks) * 512);
+		loc_wipe (&local->loc);
+	}
+
+	STACK_UNWIND (frame, op_ret, op_errno, buf);
+	return 0;
+}
+
+
+int
+quota_truncate_stat_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
+			 int32_t op_ret, int32_t op_errno, struct stat *buf)
+{
+	struct quota_local *local = NULL;
+	struct quota_priv  *priv = NULL;
+
+	priv = this->private;
+	local = frame->local;
+
+	if (op_ret >= 0) {
+		local->stbuf = *buf;
+	}
+
+	STACK_WIND (frame, quota_truncate_cbk,
+		    FIRST_CHILD (this), FIRST_CHILD (this)->fops->truncate,
+		    &local->loc, local->offset);
+	return 0;
+}
+
+
+int
+quota_truncate (call_frame_t *frame, xlator_t *this,
+		loc_t *loc, off_t offset)
+{
+	struct quota_local *local = NULL;
+	struct quota_priv  *priv = NULL;
+
+	priv = this->private;
+
+	if (priv->disk_usage_limit) {
+		local = CALLOC (1, sizeof (struct quota_local));
+		frame->local  = local;
+
+		loc_copy (&local->loc, loc);
+		local->offset = offset;
+
+		STACK_WIND (frame, quota_truncate_stat_cbk,
+			    FIRST_CHILD(this),
+			    FIRST_CHILD(this)->fops->stat, loc);
+		return 0;
+	}
+
+	STACK_WIND (frame, quota_truncate_cbk,
+		    FIRST_CHILD(this),
+		    FIRST_CHILD(this)->fops->truncate,
+		    loc, offset);
+	return 0;
+}
+
+
+int
+quota_ftruncate_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
+		     int32_t op_ret, int32_t op_errno, struct stat *buf)
+{
+	struct quota_priv  *priv = NULL;
+	struct quota_local *local = NULL;
+
+	local = frame->local;
+	priv = this->private;
+
+	if ((op_ret >= 0) && priv->disk_usage_limit) {
+		gf_quota_usage_subtract (this, (local->stbuf.st_blocks -
+						buf->st_blocks) * 512);
+		fd_unref (local->fd);
+	}
+
+	STACK_UNWIND (frame, op_ret, op_errno, buf);
+	return 0;
+}
+
+
+int
+quota_ftruncate_fstat_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
+			   int32_t op_ret, int32_t op_errno, struct stat *buf)
+{
+	struct quota_local *local = NULL;
+	struct quota_priv  *priv = NULL;
+
+	priv = this->private;
+	local = frame->local;
+
+	if (op_ret >= 0) {
+		local->stbuf = *buf;
+	}
+
+	STACK_WIND (frame, quota_ftruncate_cbk,
+		    FIRST_CHILD (this), FIRST_CHILD (this)->fops->ftruncate,
+		    local->fd, local->offset);
+	return 0;
+}
+
+
+int
+quota_ftruncate (call_frame_t *frame, xlator_t *this,
+		 fd_t *fd, off_t offset)
+{
+	struct quota_local *local = NULL;
+	struct quota_priv  *priv = NULL;
+
+
+	priv = this->private;
+
+	if (priv->disk_usage_limit) {
+		local = CALLOC (1, sizeof (struct quota_local));
+		frame->local  = local;
+
+		local->fd = fd_ref (fd);
+		local->offset = offset;
+
+		STACK_WIND (frame, quota_ftruncate_fstat_cbk,
+			    FIRST_CHILD(this),
+			    FIRST_CHILD(this)->fops->fstat, fd);
+		return 0;
+	}
+
+	STACK_WIND (frame, quota_ftruncate_cbk,
+		    FIRST_CHILD(this),
+		    FIRST_CHILD(this)->fops->ftruncate,
+		    fd, offset);
+	return 0;
+}
+
+
+int
+quota_mknod_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
+		 int32_t op_ret, int32_t op_errno,
+		 inode_t *inode, struct stat *buf)
+{
+	struct quota_priv *priv = NULL;
+
+	priv = this->private;
+
+	if ((op_ret >= 0) && priv->disk_usage_limit) {
+		gf_quota_usage_add (this, buf->st_blocks * 512);
+	}
+
+	STACK_UNWIND (frame, op_ret, op_errno, inode, buf);
+	return 0;
+}
+
+
+int
+quota_mknod (call_frame_t *frame, xlator_t *this,
+	     loc_t *loc, mode_t mode, dev_t rdev)
+{
+	struct quota_priv *priv = NULL;
+
+	priv = this->private;
+
+	if (gf_quota_check_free_disk (this) == -1) {
+		gf_log (this->name, GF_LOG_ERROR, 
+			"min-free-disk limit (%u) crossed, current available is %u",
+			priv->min_free_disk_limit, priv->current_free_disk);
+		STACK_UNWIND (frame, -1, ENOSPC, NULL, NULL);
+		return 0;
+	}
+
+        if (priv->current_disk_usage > priv->disk_usage_limit) {
+		gf_log (this->name, GF_LOG_ERROR, 
+			"Disk usage limit (%"PRIu64") crossed, current usage is %"PRIu64"",
+			priv->disk_usage_limit, priv->current_disk_usage);
+		STACK_UNWIND (frame, -1, ENOSPC, NULL, NULL);
+		return 0;
+        }
+
+	STACK_WIND (frame, quota_mknod_cbk,
+		    FIRST_CHILD(this),
+		    FIRST_CHILD(this)->fops->mknod,
+		    loc, mode, rdev);
+	return 0;
+}
+
+
+int
+quota_mkdir_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
+		 int32_t op_ret, int32_t op_errno, inode_t *inode,
+		 struct stat *buf)
+{
+	struct quota_priv *priv = NULL;
+
+	priv = this->private;
+
+	if ((op_ret >= 0) && priv->disk_usage_limit) {
+		gf_quota_usage_subtract (this, buf->st_blocks * 512);
+	}
+
+	STACK_UNWIND (frame, op_ret, op_errno, inode, buf);
+	return 0;
+}
+
+
+int
+quota_mkdir (call_frame_t *frame, xlator_t *this, loc_t *loc, mode_t mode)
+{
+	struct quota_priv *priv = NULL;
+
+	priv = this->private;
+
+	if (gf_quota_check_free_disk (this) == -1) {
+		gf_log (this->name, GF_LOG_ERROR, 
+			"min-free-disk limit (%u) crossed, current available is %u",
+			priv->min_free_disk_limit, priv->current_free_disk);
+		STACK_UNWIND (frame, -1, ENOSPC, NULL, NULL);
+		return 0;
+		
+	}
+
+        if (priv->current_disk_usage > priv->disk_usage_limit) {
+		gf_log (this->name, GF_LOG_ERROR, 
+			"Disk usage limit (%"PRIu64") crossed, current usage is %"PRIu64"",
+			priv->disk_usage_limit, priv->current_disk_usage);
+		STACK_UNWIND (frame, -1, ENOSPC, NULL, NULL);
+		return 0;
+        }
+
+	STACK_WIND (frame, quota_mkdir_cbk,
+		    FIRST_CHILD(this),
+		    FIRST_CHILD(this)->fops->mkdir,
+		    loc, mode);
+
+	return 0;
+}
+
+
+int
+quota_unlink_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
+		  int32_t op_ret, int32_t op_errno)
+{
+	struct quota_local *local = NULL;
+
+	local = frame->local;
+
+	if (local) {
+		if (op_ret >= 0) {
+			gf_quota_usage_subtract (this,
+						 local->stbuf.st_blocks * 512);
+		}
+		loc_wipe (&local->loc);
+	}
+
+	STACK_UNWIND (frame, op_ret, op_errno);
+	return 0;
+}
+
+
+int
+quota_unlink_stat_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
+		       int32_t op_ret, int32_t op_errno, struct stat *buf)
+{
+	struct quota_local *local = NULL;
+
+	local = frame->local;
+
+	if (op_ret >= 0) {
+		if (buf->st_nlink == 1) {
+			local->stbuf = *buf;
+		}
+	}
+
+	STACK_WIND (frame, quota_unlink_cbk,
+		    FIRST_CHILD(this),
+		    FIRST_CHILD(this)->fops->unlink,
+		    &local->loc);
+
+	return 0;
+}
+
+
+int
+quota_unlink (call_frame_t *frame, xlator_t *this, loc_t *loc)
+{
+	struct quota_local *local = NULL;
+	struct quota_priv  *priv = NULL;
+
+	priv = this->private;
+
+	if (priv->disk_usage_limit) {
+		local = CALLOC (1, sizeof (struct quota_local));
+		frame->local = local;
+
+		loc_copy (&local->loc, loc);
+
+		STACK_WIND (frame,
+			    quota_unlink_stat_cbk,
+			    FIRST_CHILD(this),
+			    FIRST_CHILD(this)->fops->stat,
+			    loc);
+		return 0;
+	}
+
+	STACK_WIND (frame, quota_unlink_cbk,
+		    FIRST_CHILD(this),
+		    FIRST_CHILD(this)->fops->unlink,
+		    loc);
+	return 0;
+}
+
+
+int
+quota_rmdir_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
+		 int32_t op_ret, int32_t op_errno)
+{
+	struct quota_local *local = NULL;
+
+	local = frame->local;
+
+	if (local) {
+		if (op_ret >= 0) {
+			gf_quota_usage_subtract (this, local->stbuf.st_blocks * 512);
+		}
+		loc_wipe (&local->loc);
+	}
+
+	STACK_UNWIND (frame, op_ret, op_errno);
+	return 0;
+}
+
+
+int
+quota_rmdir_stat_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
+		      int32_t op_ret, int32_t op_errno, struct stat *buf)
+{
+	struct quota_local *local = NULL;
+
+	local = frame->local;
+
+	if (op_ret >= 0) {
+		local->stbuf = *buf;
+	}
+
+	STACK_WIND (frame, quota_rmdir_cbk,
+		    FIRST_CHILD(this),
+		    FIRST_CHILD(this)->fops->rmdir,
+		    &local->loc);
+
+	return 0;
+}
+
+
+int
+quota_rmdir (call_frame_t *frame, xlator_t *this, loc_t *loc)
+{
+	struct quota_local *local = NULL;
+	struct quota_priv  *priv = NULL;
+
+	priv = this->private;
+
+	if (priv->disk_usage_limit) {
+		local = CALLOC (1, sizeof (struct quota_local));
+		frame->local = local;
+
+		loc_copy (&local->loc, loc);
+
+		STACK_WIND (frame, quota_rmdir_stat_cbk,
+			    FIRST_CHILD(this),
+			    FIRST_CHILD(this)->fops->stat, loc);
+		return 0;
+	}
+
+	STACK_WIND (frame, quota_rmdir_cbk,
+		    FIRST_CHILD(this),
+		    FIRST_CHILD(this)->fops->rmdir,
+		    loc);
+	return 0;
+}
+
+
+int
+quota_symlink_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
+		   int32_t op_ret, int32_t op_errno, inode_t *inode,
+		   struct stat *buf)
+{
+	struct quota_priv *priv = NULL;
+
+	priv = this->private;
+
+	if ((op_ret >= 0) && priv->disk_usage_limit) {
+		gf_quota_usage_add (this, buf->st_blocks * 512);
+	}
+
+	STACK_UNWIND (frame, op_ret, op_errno, inode, buf);
+	return 0;
+}
+
+
+int
+quota_symlink (call_frame_t *frame, xlator_t *this,
+	       const char *linkpath, loc_t *loc)
+{
+	struct quota_priv *priv = NULL;
+
+	priv = this->private;
+
+	if (gf_quota_check_free_disk (this) == -1) {
+		gf_log (this->name, GF_LOG_ERROR, 
+			"min-free-disk limit (%u) crossed, current available is %u",
+			priv->min_free_disk_limit, priv->current_free_disk);
+		STACK_UNWIND (frame, -1, ENOSPC, NULL, NULL);
+		return 0;
+		
+	}
+        if (priv->current_disk_usage > priv->disk_usage_limit) {
+		gf_log (this->name, GF_LOG_ERROR, 
+			"Disk usage limit (%"PRIu64") crossed, current usage is %"PRIu64"",
+			priv->disk_usage_limit, priv->current_disk_usage);
+		STACK_UNWIND (frame, -1, ENOSPC, NULL, NULL);
+		return 0;
+        }
+
+	STACK_WIND (frame, quota_symlink_cbk,
+		    FIRST_CHILD(this),
+		    FIRST_CHILD(this)->fops->symlink,
+		    linkpath, loc);
+	return 0;
+}
+
+
+int
+quota_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)
+{
+	struct quota_priv *priv = this->private;
+	int                ret = 0;
+
+	if ((op_ret >= 0) && priv->disk_usage_limit) {
+		gf_quota_usage_add (this, buf->st_blocks * 512);
+
+		ret = fd_ctx_set (fd, this, 1);
+	}
+
+	STACK_UNWIND (frame, op_ret, op_errno, fd, inode, buf);
+	return 0;
+}
+
+
+int
+quota_create (call_frame_t *frame, xlator_t *this,
+	      loc_t *loc, int32_t flags, mode_t mode, fd_t *fd)
+{
+	struct quota_priv *priv = NULL;
+
+	priv = this->private;
+
+	if (gf_quota_check_free_disk (this) == -1) {
+		gf_log (this->name, GF_LOG_ERROR, 
+			"min-free-disk limit (%u) crossed, current available is %u",
+			priv->min_free_disk_limit, priv->current_free_disk);
+		STACK_UNWIND (frame, -1, ENOSPC, NULL, NULL, NULL);
+		return 0;
+		
+	}
+        if (priv->current_disk_usage > priv->disk_usage_limit) {
+		gf_log (this->name, GF_LOG_ERROR, 
+			"Disk usage limit (%"PRIu64") crossed, current usage is %"PRIu64"",
+			priv->disk_usage_limit, priv->current_disk_usage);
+		STACK_UNWIND (frame, -1, ENOSPC, NULL, NULL, NULL);
+		return 0;
+        }
+
+	STACK_WIND (frame, quota_create_cbk,
+		    FIRST_CHILD(this),
+		    FIRST_CHILD(this)->fops->create,
+		    loc, flags, mode, fd);
+	return 0;
+}
+
+
+int
+quota_open_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
+		int32_t op_ret, int32_t op_errno, fd_t *fd)
+{
+	int                ret = 0;
+
+	if (op_ret >= 0)
+		ret = fd_ctx_set (fd, this, 1);
+
+	STACK_UNWIND (frame, op_ret, op_errno, fd);
+	return 0;
+}
+
+
+int
+quota_open (call_frame_t *frame, xlator_t *this,
+	    loc_t *loc, int32_t flags, fd_t *fd)
+{
+	STACK_WIND (frame, quota_open_cbk,
+		    FIRST_CHILD(this),
+		    FIRST_CHILD(this)->fops->open,
+		    loc, flags, fd);
+	return 0;
+}
+
+
+int
+quota_writev_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
+		  int32_t op_ret, int32_t op_errno, struct stat *stbuf)
+{
+	struct quota_priv *priv = NULL;
+	struct quota_local *local = NULL;
+
+
+	priv = this->private;
+	local = frame->local;
+
+	if (priv->disk_usage_limit) {
+		if (op_ret >= 0) { 
+			gf_quota_usage_add (this, (stbuf->st_blocks -
+						   local->stbuf.st_blocks) * 512);
+		}
+		fd_unref (local->fd);
+		dict_unref (local->refs);
+	}
+
+	STACK_UNWIND (frame, op_ret, op_errno, stbuf);
+	return 0;
+}
+
+
+int
+quota_writev_fstat_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
+			int32_t op_ret,	int32_t op_errno, struct stat *buf)
+{
+	struct quota_local *local = NULL;
+	struct quota_priv  *priv = NULL;
+	int                 iovlen = 0;
+
+
+	local = frame->local;
+	priv = this->private;
+
+	if (op_ret >= 0) {
+		if (priv->current_disk_usage > priv->disk_usage_limit) {
+			iovlen = iov_length (local->vector, local->count);
+
+			if (iovlen > (buf->st_blksize - (buf->st_size % buf->st_blksize))) {
+				fd_unref (local->fd);
+				dict_unref (local->refs);
+				STACK_UNWIND (frame, -1, ENOSPC, NULL);
+				return 0;
+			}
+		}
+		local->stbuf = *buf;
+	}
+	
+	STACK_WIND (frame, quota_writev_cbk,
+		    FIRST_CHILD(this),
+		    FIRST_CHILD(this)->fops->writev,
+		    local->fd, local->vector, local->count, local->offset);
+
+	return 0;
+}
+
+
+int
+quota_writev (call_frame_t *frame, xlator_t *this, fd_t *fd,
+	      struct iovec *vector, int32_t count, off_t off)
+{
+	struct quota_local *local = NULL;
+	struct quota_priv  *priv = NULL;
+
+	priv = this->private;
+
+	if (gf_quota_check_free_disk (this) == -1) {
+		gf_log (this->name, GF_LOG_ERROR, 
+			"min-free-disk limit (%u) crossed, current available is %u",
+			priv->min_free_disk_limit, priv->current_free_disk);
+		STACK_UNWIND (frame, -1, ENOSPC, NULL);
+		return 0;
+	}
+
+	if (priv->disk_usage_limit) {
+		local = CALLOC (1, sizeof (struct quota_local));
+		local->fd     = fd_ref (fd);
+		local->refs   = dict_ref (frame->root->req_refs);
+		local->vector = vector;
+		local->count  = count;
+		local->offset = off;
+		frame->local  = local;
+
+		STACK_WIND (frame, quota_writev_fstat_cbk,
+			    FIRST_CHILD(this),
+			    FIRST_CHILD(this)->fops->fstat, fd);
+		return 0;
+	}
+
+	STACK_WIND (frame, quota_writev_cbk,
+		    FIRST_CHILD(this),
+		    FIRST_CHILD(this)->fops->writev,
+		    fd, vector, count, off);
+	return 0;
+}
+
+
+int
+quota_removexattr_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
+		       int32_t op_ret, int32_t op_errno)
+{
+	if (op_ret == -1) {
+		gf_log (this->name, GF_LOG_CRITICAL, 
+			"failed to remove the disk-usage value: %s",
+			strerror (op_errno));
+	} 
+	
+	STACK_DESTROY (frame->root);
+	return 0;
+}
+
+
+int
+quota_setxattr_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
+		    int32_t op_ret, int32_t op_errno)
+{
+	if (op_ret == -1) {
+		gf_log (this->name, GF_LOG_CRITICAL, 
+			"failed to set the disk-usage value: %s",
+			strerror (op_errno));
+	} 
+
+	STACK_DESTROY (frame->root);
+	return 0;
+}
+
+
+int
+quota_statfs_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
+		  int32_t op_ret, int32_t op_errno, struct statvfs *statvfs)
+{
+	struct quota_priv *priv = NULL;
+	uint64_t           f_blocks = 0;
+	int64_t            f_bfree = 0;
+	uint64_t           f_bused = 0;
+
+
+	priv = this->private;
+
+	if (op_ret != 0)
+		goto unwind;
+
+	f_blocks = priv->disk_usage_limit / statvfs->f_frsize;
+	f_bused = priv->current_disk_usage / statvfs->f_frsize;
+
+	if (f_blocks && (f_blocks < statvfs->f_blocks))
+		statvfs->f_blocks = f_blocks;
+
+	f_bfree = (statvfs->f_blocks - f_bused);
+
+	if (f_bfree >= 0)
+		statvfs->f_bfree = statvfs->f_bavail = f_bfree;
+	else
+		statvfs->f_bfree = statvfs->f_bavail = 0;
+
+unwind:
+	STACK_UNWIND (frame, op_ret, op_errno, statvfs);
+	return 0;
+}
+
+
+int
+quota_statfs (call_frame_t *frame, xlator_t *this, loc_t *loc)
+{
+	STACK_WIND (frame, quota_statfs_cbk,
+		    FIRST_CHILD (this), FIRST_CHILD (this)->fops->statfs, loc);
+
+	return 0;
+}
+
+
+int
+quota_getxattr_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
+		    int32_t op_ret, int32_t op_errno, dict_t *value)
+{
+	data_t *data = NULL;
+	struct quota_priv *priv = this->private;
+	
+	if (op_ret >= 0) {
+		data = dict_get (value, "trusted.glusterfs-quota-du");
+		if (data) {
+			LOCK (&priv->lock);
+			{
+				priv->current_disk_usage = data_to_uint64 (data);
+			}
+			UNLOCK (&priv->lock);
+
+			return 0;
+		}
+	} 
+
+	STACK_DESTROY (frame->root);
+
+	return 0;
+}
+
+
+void
+gf_quota_get_disk_usage (xlator_t *this)
+{
+	call_frame_t *frame = NULL;
+	call_pool_t  *pool = NULL;
+	loc_t         loc;
+
+	pool = this->ctx->pool;
+	frame = create_frame (this, pool);
+	build_root_loc (this, &loc);
+
+	STACK_WIND (frame, quota_getxattr_cbk,
+		    this->children->xlator,
+		    this->children->xlator->fops->getxattr,
+		    &loc,
+		    "trusted.glusterfs-quota-du");
+	return ;
+}
+
+
+void
+gf_quota_cache_sync (xlator_t *this)
+{
+	struct quota_priv *priv = NULL;
+	call_frame_t      *frame = NULL;
+	dict_t            *dict = get_new_dict ();
+	loc_t              loc;
+
+
+	priv = this->private;
+	build_root_loc (this, &loc);
+
+	frame = create_frame (this, this->ctx->pool);
+	dict_set (dict, "trusted.glusterfs-quota-du", 
+		  data_from_uint64 (priv->current_disk_usage));
+
+	STACK_WIND (frame, quota_setxattr_cbk,
+		    this->children->xlator,
+		    this->children->xlator->fops->setxattr,
+		    &loc, dict, 0);
+}
+
+
+int
+quota_release (xlator_t *this, fd_t *fd)
+{
+	gf_quota_cache_sync (this);
+
+	return 0;
+}
+
+
+/* notify */
+int32_t
+notify (xlator_t *this,
+	int32_t event,
+	void *data,
+	...)
+{
+	struct quota_priv *priv = this->private;
+	
+	switch (event)
+	{
+	case GF_EVENT_CHILD_UP:
+		if (priv->only_first_time) {
+			priv->only_first_time = 0;
+			if (priv->disk_usage_limit) {
+				gf_quota_get_disk_usage (this);
+			}
+		}
+	default:
+		default_notify (this, event, data);
+		break;
+	}
+
+	return 0;
+}
+
+
+int32_t 
+init (xlator_t *this)
+{
+	int     ret  = 0;
+        data_t *data = NULL;
+	struct quota_priv *_private = NULL;
+
+	if (!this->children || this->children->next) {
+		gf_log (this->name, GF_LOG_ERROR, 
+			"FATAL: quota should have exactly one child");
+		return -1;
+	}
+	
+	if (!this->parents) {
+		gf_log (this->name, GF_LOG_WARNING,
+			"dangling volume. check volfile ");
+	}
+
+	_private = CALLOC (1, sizeof (struct quota_priv));
+        _private->disk_usage_limit = 0;
+        data = dict_get (this->options, "disk-usage-limit");
+        if (data) {
+		if (gf_string2bytesize (data->data, &_private->disk_usage_limit) != 0) {
+                        gf_log (this->name, GF_LOG_ERROR, 
+                                "invalid number '%s' for disk-usage limit", data->data);
+			ret = -1;
+			goto out;
+                }
+
+		LOCK_INIT (&_private->lock);
+		_private->current_disk_usage = 0;
+	}
+	
+        _private->min_free_disk_limit = 0;
+        data = dict_get (this->options, "min-free-disk-limit");
+        if (data) {
+		if (gf_string2percent (data->data, &_private->min_free_disk_limit) != 0) {
+                        gf_log (this->name, GF_LOG_ERROR, 
+                                "invalid percent '%s' for min-free-disk limit", data->data);
+			ret = -1;
+			goto out;
+                }
+		_private->refresh_interval = 20; /* 20seconds is default */
+		data = dict_get (this->options, "refresh-interval");
+		if (data) {
+			if (gf_string2time (data->data, 
+					    &_private->refresh_interval)!= 0) {
+				gf_log (this->name, GF_LOG_ERROR, 
+					"invalid time '%s' for refresh "
+					"interval", data->data);
+				ret = -1;
+				goto out;
+			}
+		}
+        }
+
+	_private->only_first_time = 1;
+        this->private = (void *)_private;
+	ret = 0;
+ out:
+	return ret;
+}
+
+void 
+fini (xlator_t *this)
+{
+	struct quota_priv *_private = this->private;
+
+	if (_private) {
+		gf_quota_cache_sync (this);
+		this->private = NULL;
+	}
+	
+	return ;
+}
+
+struct xlator_fops fops = {
+	.create      = quota_create,
+	.open        = quota_open,
+	.truncate    = quota_truncate,
+	.ftruncate   = quota_ftruncate,
+	.writev      = quota_writev,
+	.unlink      = quota_unlink,
+	.rmdir       = quota_rmdir,
+	.mknod       = quota_mknod,
+	.mkdir       = quota_mkdir,
+	.symlink     = quota_symlink,
+	.statfs      = quota_statfs,
+};
+
+struct xlator_mops mops = {
+};
+
+struct xlator_cbks cbks = {
+	.release     = quota_release
+};
+
+struct volume_options options[] = {
+	{ .key  = {"min-free-disk-limit"}, 
+	  .type = GF_OPTION_TYPE_PERCENT
+	},
+	{ .key  = {"refresh-interval"}, 
+	  .type = GF_OPTION_TYPE_TIME
+	},
+	{ .key  = {"disk-usage-limit"}, 
+	  .type = GF_OPTION_TYPE_SIZET 
+	},
+	{ .key = {NULL} },
+};
diff --git a/xlators/features/trash/Makefile.am b/xlators/features/trash/Makefile.am
new file mode 100644
index 00000000000..d471a3f9243
--- /dev/null
+++ b/xlators/features/trash/Makefile.am
@@ -0,0 +1,3 @@
+SUBDIRS = src
+
+CLEANFILES = 
diff --git a/xlators/features/trash/src/Makefile.am b/xlators/features/trash/src/Makefile.am
new file mode 100644
index 00000000000..d61f608aaa8
--- /dev/null
+++ b/xlators/features/trash/src/Makefile.am
@@ -0,0 +1,13 @@
+xlator_LTLIBRARIES = trash.la
+xlatordir = $(libdir)/glusterfs/$(PACKAGE_VERSION)/xlator/features
+
+trash_la_LDFLAGS = -module -avoidversion 
+
+trash_la_SOURCES = trash.c
+trash_la_LIBADD = $(top_builddir)/libglusterfs/src/libglusterfs.la 
+
+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/features/trash/src/trash.c b/xlators/features/trash/src/trash.c
new file mode 100644
index 00000000000..c8e7357ee08
--- /dev/null
+++ b/xlators/features/trash/src/trash.c
@@ -0,0 +1,596 @@
+/*
+  Copyright (c) 2006, 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 _CONFIG_H
+#define _CONFIG_H
+#include "config.h"
+#endif
+
+
+#include "glusterfs.h"
+#include "logging.h"
+#include "dict.h"
+#include "xlator.h"
+#include "defaults.h"
+
+#include <libgen.h>
+
+/* TODO: currently it can work only above posix, no other translators 
+ *       between them. Not a good thing. Try making more reliable methods.
+ */
+
+struct trash_struct {
+	inode_t *inode;
+	loc_t loc1;
+	loc_t loc2;
+	char origpath[ZR_PATH_MAX];
+	char newpath[ZR_PATH_MAX];
+	char oldpath[ZR_PATH_MAX]; // used only in case of rename
+};
+typedef struct trash_struct trash_local_t;
+
+struct trash_priv {
+	char trash_dir[ZR_PATH_MAX];
+};
+typedef struct trash_priv trash_private_t;
+
+int32_t
+trash_unlink_rename_cbk (call_frame_t *frame,
+			 void *cookie,
+			 xlator_t *this,
+			 int32_t op_ret,
+			 int32_t op_errno,
+			 struct stat *buf);
+int32_t
+trash_rename_rename_cbk (call_frame_t *frame,
+			 void *cookie,
+			 xlator_t *this,
+			 int32_t op_ret,
+			 int32_t op_errno,
+			 struct stat *buf);
+
+/**
+ * trash_common_unwind_cbk -
+ */
+int32_t
+trash_common_unwind_cbk (call_frame_t *frame,
+			 void *cookie,
+			 xlator_t *this,
+			 int32_t op_ret,
+			 int32_t op_errno)
+{
+	trash_local_t *local = frame->local;
+
+	if (local->loc1.path)
+		loc_wipe (&local->loc1);
+	
+	if (local->loc2.path)
+		loc_wipe (&local->loc2);
+
+	STACK_UNWIND (frame, op_ret, op_errno);
+	return 0;
+}
+
+/**
+ * trash_common_unwind_buf_cbk -
+ */
+int32_t
+trash_common_unwind_buf_cbk (call_frame_t *frame,
+			     void *cookie,
+			     xlator_t *this,
+			     int32_t op_ret,
+			     int32_t op_errno,
+			     struct stat *buf)
+{
+	trash_local_t *local = frame->local;
+
+	if (local->loc1.path)
+		loc_wipe (&local->loc1);
+	
+	if (local->loc2.path)
+		loc_wipe (&local->loc2);
+
+	STACK_UNWIND (frame, op_ret, op_errno, buf);
+	return 0;
+}
+
+int32_t
+trash_mkdir_cbk (call_frame_t *frame,
+		 void *cookie,
+		 xlator_t *this,
+		 int32_t op_ret,
+		 int32_t op_errno,
+		 inode_t *inode,
+		 struct stat *stbuf)
+{
+	trash_local_t *local = frame->local;
+	char *tmp_str = strdup (local->newpath);
+	int32_t count = 0;
+	char *tmp_path = NULL;
+	char *tmp_dirname = NULL;
+
+	if (op_ret == -1 && op_errno == ENOENT) {
+		tmp_dirname = strchr (tmp_str, '/');
+		while (tmp_dirname) {
+			count = tmp_dirname - tmp_str;
+			if (count == 0)
+				count = 1;
+			tmp_path = CALLOC (1, count + 1);
+			ERR_ABORT (tmp_path);
+			memcpy (tmp_path, local->newpath, count);
+			loc_t tmp_loc = {
+				.inode = NULL,
+				.path = tmp_path,
+			};
+
+			/* TODO:create the directory with proper permissions */
+			STACK_WIND_COOKIE (frame,
+					   trash_mkdir_cbk,
+					   tmp_path,
+					   this->children->xlator,
+					   this->children->xlator->fops->mkdir,
+					   &tmp_loc,
+					   0777);
+			tmp_dirname = strchr (tmp_str + count + 1, '/');
+		}
+		free (cookie);
+		free (tmp_str);
+		return 0;
+	}
+	char *dir_name = dirname (tmp_str);
+	if (strcmp((char*)cookie, dir_name) == 0) {
+		loc_t new_loc = {
+			.inode = NULL,
+			.path = local->newpath
+		};
+		STACK_WIND (frame,
+			    trash_unlink_rename_cbk,
+			    this->children->xlator,
+			    this->children->xlator->fops->rename,
+			    &local->loc2,
+			    &new_loc);
+
+	}
+	free (cookie); /* strdup (dir_name) was sent here :) */
+	free (tmp_str);
+	return 0;
+}
+
+/**
+ * trash_unlink_rename_cbk -
+ */
+int32_t
+trash_unlink_rename_cbk (call_frame_t *frame,
+			 void *cookie,
+			 xlator_t *this,
+			 int32_t op_ret,
+			 int32_t op_errno,
+			 struct stat *buf)
+{
+	trash_local_t *local = frame->local;
+	if (op_ret == -1 && op_errno == ENOENT) {
+		/* check for the errno, if its ENOENT create directory and call
+		 * rename later
+		 */
+		char *tmp_str = strdup (local->newpath);
+		char *dir_name = dirname (tmp_str);
+		loc_t tmp_loc = {
+			.inode = NULL,
+			.path = dir_name,
+		};
+		/* TODO: create the directory with proper permissions */
+		STACK_WIND_COOKIE (frame,
+				   trash_mkdir_cbk,
+				   strdup (dir_name),
+				   this->children->xlator,
+				   this->children->xlator->fops->mkdir,
+				   &tmp_loc,
+				   0777);
+		free (tmp_str);
+	} else if (op_ret == -1 && op_errno == ENOTDIR) {
+		gf_log (this->name, GF_LOG_WARNING,
+			"Target exists, cannot keep the copy, deleting");
+		STACK_WIND (frame,
+			    trash_common_unwind_cbk,
+			    this->children->xlator,
+			    this->children->xlator->fops->unlink,
+			    &local->loc2);
+	} else if (op_ret == -1 && op_errno == EISDIR) {
+		gf_log (this->name, GF_LOG_WARNING,
+			"Target exists as a directory, cannot keep the copy, "
+			"deleting");
+		STACK_WIND (frame,
+			    trash_common_unwind_cbk,
+			    this->children->xlator,
+			    this->children->xlator->fops->unlink,
+			    &local->loc2);
+	} else {
+		/* */
+		STACK_UNWIND (frame, 0, op_errno);
+	}
+
+	return 0;
+}
+
+
+/**
+ * trash_unlink -
+ */
+int32_t
+trash_unlink (call_frame_t *frame,
+	      xlator_t *this,
+	      loc_t *loc)
+{
+	trash_private_t *priv = this->private;
+	trash_local_t *local = NULL;
+	time_t       utime = 0;
+	struct tm   *tm = NULL;
+	char         timestr[256];
+
+	if (strncmp (loc->path, priv->trash_dir, 
+		     strlen(priv->trash_dir)) == 0) {
+		/* Trying to rename from the trash can dir, do the
+		   actual unlink */
+		STACK_WIND (frame,
+			    trash_common_unwind_cbk,
+			    this->children->xlator,
+			    this->children->xlator->fops->unlink,
+			    loc);
+	} else {
+		local = CALLOC (1, sizeof (trash_local_t));
+		if (!local) {
+			STACK_UNWIND (frame, -1, ENOMEM);
+			return 0;
+		}
+		frame->local = local;
+		
+		loc_copy (&local->loc2, loc);
+
+		strcpy (local->newpath, priv->trash_dir);
+		strcat (local->newpath, loc->path);
+
+		utime = time (NULL);
+		tm    = localtime (&utime);
+		strftime (timestr, 256, ".%Y%m%d%H%M%S", tm); 
+		strcat (local->newpath, timestr);
+
+		{
+			loc_t new_loc = {
+				.inode = NULL,
+				.path = local->newpath
+			};
+			STACK_WIND (frame,
+				    trash_unlink_rename_cbk,
+				    this->children->xlator,
+				    this->children->xlator->fops->rename,
+				    loc,
+				    &new_loc);
+		}
+	}
+	return 0;
+}
+
+/* */
+int32_t
+trash_rename_mkdir_cbk (call_frame_t *frame,
+			void *cookie,
+			xlator_t *this,
+			int32_t op_ret,
+			int32_t op_errno,
+			inode_t *inode,
+			struct stat *stbuf)
+{
+	trash_local_t *local = frame->local;
+	char *tmp_str = strdup (local->newpath);
+
+	if (op_ret == -1 && op_errno == ENOENT) {
+		int32_t count = 0;
+		char *tmp_path = NULL;
+		char *tmp_dirname = strchr (tmp_str, '/');
+
+		while (tmp_dirname) {
+			count = tmp_dirname - tmp_str;
+			if (count == 0)
+				count = 1;
+			tmp_path = CALLOC (1, count + 2);
+			ERR_ABORT (tmp_path);
+			memcpy (tmp_path, local->newpath, count);
+			loc_t tmp_loc = {
+				.inode = NULL,
+				.path = tmp_path,
+			};
+
+			/* TODO:create the directory with proper permissions */
+			STACK_WIND_COOKIE (frame,
+					   trash_rename_mkdir_cbk,
+					   tmp_path,
+					   this->children->xlator,
+					   this->children->xlator->fops->mkdir,
+					   &tmp_loc,
+					   0777);
+			tmp_dirname = strchr (tmp_str + count + 1, '/');
+		}
+		free (cookie);
+		free (tmp_str);
+		return 0;
+	}
+	char *dir_name = dirname (tmp_str);
+	if (strcmp((char*)cookie, dir_name) == 0) {
+		loc_t new_loc = {
+			.inode = NULL,
+			.path = local->newpath
+		};
+		STACK_WIND (frame,
+			    trash_rename_rename_cbk,
+			    this->children->xlator,
+			    this->children->xlator->fops->rename,
+			    &local->loc2,
+			    &new_loc);
+
+	}
+	free (cookie); /* strdup (dir_name) was sent here :) */
+	free (tmp_str);
+	return 0;
+}
+
+
+/**
+ * trash_unlink_rename_cbk -
+ */
+int32_t
+trash_rename_rename_cbk (call_frame_t *frame,
+			 void *cookie,
+			 xlator_t *this,
+			 int32_t op_ret,
+			 int32_t op_errno,
+			 struct stat *buf)
+{
+	trash_local_t *local = frame->local;
+	if (op_ret == -1 && op_errno == ENOENT) {
+		/* check for the errno, if its ENOENT create directory and call
+		 * rename later
+		 */
+		char *tmp_str = strdup (local->newpath);
+		char *dir_name = dirname (tmp_str);
+		loc_t tmp_loc = {
+			.inode = NULL,
+			.path = dir_name,
+		};
+		/* TODO: create the directory with proper permissions */
+		STACK_WIND_COOKIE (frame,
+				   trash_rename_mkdir_cbk,
+				   strdup (dir_name),
+				   this->children->xlator,
+				   this->children->xlator->fops->mkdir,
+				   &tmp_loc,
+				   0777);
+		free (tmp_str);
+		return 0;
+	} else if (op_ret == -1 && op_errno == ENOTDIR) {
+		gf_log (this->name, GF_LOG_WARNING,
+			"Target exists, cannot keep the dest entry %s, "
+			"renaming",
+			local->loc2.path);
+	} else if (op_ret == -1 && op_errno == EISDIR) {
+		gf_log (this->name, GF_LOG_WARNING,
+			"Target exists as a directory, cannot keep the "
+			"copy %s, renaming",
+			local->loc2.path);
+	}
+	loc_t new_loc = {
+		.inode = NULL,
+		.parent = local->loc2.parent,
+		.path = local->loc2.path,
+	};
+	STACK_WIND (frame,
+		    trash_common_unwind_buf_cbk,
+		    this->children->xlator,
+		    this->children->xlator->fops->rename,
+		    &local->loc1,
+		    &new_loc);
+
+	return 0;
+}
+
+/**
+ * trash_rename_lookup_cbk -
+ */
+int32_t
+trash_rename_lookup_cbk (call_frame_t *frame,
+			 void *cookie,
+			 xlator_t *this,
+			 int32_t op_ret,
+			 int32_t op_errno,
+			 inode_t *inode,
+			 struct stat *buf,
+			 dict_t *xattr)
+{
+	trash_local_t *local = frame->local;
+
+	if (op_ret == -1) {
+		STACK_WIND (frame,
+			    trash_common_unwind_buf_cbk,
+			    this->children->xlator,
+			    this->children->xlator->fops->rename,
+			    &local->loc1,
+			    &local->loc2);
+		return 0;
+	}
+
+	loc_t oldloc = {
+		.parent = local->loc2.parent,
+		.inode = inode,
+		.path = local->loc2.path,
+	};
+	loc_t newloc = {
+		.inode = NULL,
+		.path = local->newpath
+	};
+	STACK_WIND (frame,
+		    trash_rename_rename_cbk,
+		    this->children->xlator,
+		    this->children->xlator->fops->rename,
+		    &oldloc,
+		    &newloc);
+
+	return 0;
+}
+
+
+/**
+ * trash_rename -
+ */
+int32_t
+trash_rename (call_frame_t *frame,
+	      xlator_t *this,
+	      loc_t *oldloc,
+	      loc_t *newloc)
+{
+	trash_private_t *priv = this->private;
+	trash_local_t *local = NULL;
+	time_t       utime = 0;
+	struct tm   *tm = NULL;
+	char         timestr[256];
+
+	if (strncmp (oldloc->path, priv->trash_dir, 
+		     strlen(priv->trash_dir)) == 0) {
+		/* Trying to rename from the trash can dir, 
+		   do the actual rename */
+		STACK_WIND (frame,
+			    trash_common_unwind_buf_cbk,
+			    this->children->xlator,
+			    this->children->xlator->fops->rename,
+			    oldloc,
+			    newloc);
+	} else {
+		/* Trying to rename a regular file from GlusterFS */
+		local = CALLOC (1, sizeof (trash_local_t));
+		if (!local) {
+			STACK_UNWIND (frame, -1, ENOMEM, NULL);
+			return 0;
+		}
+		frame->local = local;
+		loc_copy (&local->loc1, oldloc);
+		loc_copy (&local->loc2, newloc);
+
+		strcpy (local->newpath, priv->trash_dir);
+		strcat (local->newpath, newloc->path);
+
+		utime = time (NULL);
+		tm    = localtime (&utime);
+		strftime (timestr, 256, ".%Y%m%d%H%M%S", tm); 
+		strcat (local->newpath, timestr);
+
+		/* Send a lookup call on newloc, to ensure we are not 
+		   overwriting */
+		STACK_WIND (frame,
+			    trash_rename_lookup_cbk,
+			    this->children->xlator,
+			    this->children->xlator->fops->lookup,
+			    newloc,
+			    0);
+	}
+	return 0;
+}
+
+/**
+ * trash_init -
+ */
+int32_t
+init (xlator_t *this)
+{
+  	data_t *trash_dir = NULL;
+	xlator_list_t *trav = NULL;
+	trash_private_t *_priv = NULL;
+
+	/* Create .trashcan directory in init */
+	if (!this->children || this->children->next) {
+		gf_log (this->name, GF_LOG_ERROR,
+			"not configured with exactly one child. exiting");
+		return -1;
+	}
+
+	if (!this->parents) {
+		gf_log (this->name, GF_LOG_WARNING,
+			"dangling volume. check volfile ");
+	}
+
+	trav = this->children;
+	while (trav->xlator->children)
+		trav = trav->xlator->children;
+
+	if (strncmp ("storage/", trav->xlator->type, 8))
+	{
+		gf_log (this->name, GF_LOG_ERROR,
+			"'trash' translator not loaded over storage "
+			"translator, not a supported setup");
+		return -1;
+	}
+
+	_priv = CALLOC (1, sizeof (*_priv));
+	ERR_ABORT (_priv);
+
+	trash_dir = dict_get (this->options, "trash-dir");
+	if (!trash_dir) {
+		gf_log (this->name, GF_LOG_WARNING,
+			"no option specified for 'trash-dir', "
+			"using \"/.trashcan/\"");
+		strcpy (_priv->trash_dir, "/.trashcan");
+	} else {
+		/* Need a path with '/' as the first char, if not 
+		   given, append it */
+		if (trash_dir->data[0] == '/') {
+			strcpy (_priv->trash_dir, trash_dir->data);
+		} else {
+			strcpy (_priv->trash_dir, "/");
+			strcat (_priv->trash_dir, trash_dir->data);
+		}
+	}
+
+	this->private = (void *)_priv;
+	return 0;
+}
+
+void
+fini (xlator_t *this)
+{
+	trash_private_t *priv = this->private;
+	FREE (priv);
+	return;
+}
+
+
+struct xlator_fops fops = {
+	.unlink = trash_unlink,
+	.rename = trash_rename,
+};
+
+struct xlator_mops mops = {
+
+};
+
+struct xlator_cbks cbks = {
+};
+
+struct volume_options options[] = {
+	{ .key  = { "trash-dir" }, 
+	  .type = GF_OPTION_TYPE_PATH 
+	},
+	{ .key  = {NULL} },
+};