read-only: read-only/worm translator should be in brick graph by default

Problem:
read-only/worm translator is not loaded by default in brick graph because of which
when read-only option is set through volume set volume still remains writable
untill the bricks are restarted as the translator does not have an inmemory flag
to decide whether the read-only/worm option is turned or not.

Solution:
read-only/worm should be loaded by default in brick graph and the read-only/worm
option can be toggled through volume set command. read-only/worm translator now'
has an in-memory flag to decide whether the volume is read-only or not and based
on that either reject the fop or proceed.

Change-Id: Ic79328698f6a72c50433cff15ecadb1a92acc643
BUG: 1134822
Signed-off-by: Atin Mukherjee <amukherj@redhat.com>
Reviewed-on: http://review.gluster.org/8571
Tested-by: Gluster Build System <jenkins@build.gluster.com>
Reviewed-by: Vijay Bellur <vbellur@redhat.com>

9 files changed, 401 insertions, 72 deletions

diff --git a/xlators/features/read-only/src/Makefile.am b/xlators/features/read-only/src/Makefile.am
index 4c146213742..209e4208af0 100644
--- a/xlators/features/read-only/src/Makefile.am
+++ b/xlators/features/read-only/src/Makefile.am
@@ -2,7 +2,7 @@ xlator_LTLIBRARIES = read-only.la worm.la
 
 xlatordir = $(libdir)/glusterfs/$(PACKAGE_VERSION)/xlator/features
 
-noinst_HEADERS = read-only-common.h
+noinst_HEADERS = read-only.h read-only-mem-types.h read-only-common.h
 
 read_only_la_LDFLAGS = -module -avoid-version
 
diff --git a/xlators/features/read-only/src/read-only-common.c b/xlators/features/read-only/src/read-only-common.c
index 56a7a7176aa..39937d106c5 100644
--- a/xlators/features/read-only/src/read-only-common.c
+++ b/xlators/features/read-only/src/read-only-common.c
@@ -12,14 +12,34 @@
 #include "config.h"
 #endif
 
-#include "xlator.h"
+#include "read-only.h"
+#include "read-only-mem-types.h"
 #include "defaults.h"
 
+gf_boolean_t
+is_readonly_or_worm_enabled (xlator_t *this)
+{
+        read_only_priv_t  *priv                     = NULL;
+        gf_boolean_t       readonly_or_worm_enabled = _gf_false;
+
+        priv = this->private;
+        GF_ASSERT (priv);
+
+        readonly_or_worm_enabled = priv->readonly_or_worm_enabled;
+
+        return readonly_or_worm_enabled;
+}
+
 int32_t
 ro_xattrop (call_frame_t *frame, xlator_t *this, loc_t *loc,
             gf_xattrop_flags_t flags, dict_t *dict, dict_t *xdata)
 {
-        STACK_UNWIND_STRICT (xattrop, frame, -1, EROFS, NULL, xdata);
+        if (is_readonly_or_worm_enabled (this))
+                STACK_UNWIND_STRICT (xattrop, frame, -1, EROFS, NULL, xdata);
+        else
+                STACK_WIND_TAIL (frame, FIRST_CHILD (this),
+                                 FIRST_CHILD(this)->fops->xattrop,
+                                 loc, flags, dict, xdata);
         return 0;
 }
 
@@ -27,7 +47,13 @@ int32_t
 ro_fxattrop (call_frame_t *frame, xlator_t *this,
              fd_t *fd, gf_xattrop_flags_t flags, dict_t *dict, dict_t *xdata)
 {
-        STACK_UNWIND_STRICT (fxattrop, frame, -1, EROFS, NULL, xdata);
+        if (is_readonly_or_worm_enabled (this))
+                STACK_UNWIND_STRICT (fxattrop, frame, -1, EROFS, NULL, xdata);
+        else
+                STACK_WIND_TAIL (frame, FIRST_CHILD (this),
+                                 FIRST_CHILD(this)->fops->fxattrop,
+                                 fd, flags, dict, xdata);
+
         return 0;
 }
 
@@ -36,15 +62,28 @@ ro_entrylk (call_frame_t *frame, xlator_t *this, const char *volume,
             loc_t *loc, const char *basename, entrylk_cmd cmd,
             entrylk_type type, dict_t *xdata)
 {
-        STACK_UNWIND_STRICT (entrylk, frame, -1, EROFS, xdata);
+        if (is_readonly_or_worm_enabled (this))
+                STACK_UNWIND_STRICT (entrylk, frame, -1, EROFS, xdata);
+        else
+                STACK_WIND_TAIL (frame, FIRST_CHILD (this),
+                                 FIRST_CHILD(this)->fops->entrylk,
+                                 volume, loc, basename, cmd, type, xdata);
+
         return 0;
 }
 
 int32_t
 ro_fentrylk (call_frame_t *frame, xlator_t *this, const char *volume,
-             fd_t *fd, const char *basename, entrylk_cmd cmd, entrylk_type type, dict_t *xdata)
+             fd_t *fd, const char *basename, entrylk_cmd cmd, entrylk_type type,
+             dict_t *xdata)
 {
-        STACK_UNWIND_STRICT (fentrylk, frame, -1, EROFS, xdata);
+        if (is_readonly_or_worm_enabled (this))
+                STACK_UNWIND_STRICT (fentrylk, frame, -1, EROFS, xdata);
+        else
+                STACK_WIND_TAIL (frame, FIRST_CHILD (this),
+                                 FIRST_CHILD(this)->fops->fentrylk,
+                                 volume, fd, basename, cmd, type, xdata);
+
         return 0;
 }
 
@@ -52,7 +91,13 @@ int32_t
 ro_inodelk (call_frame_t *frame, xlator_t *this, const char *volume,
             loc_t *loc, int32_t cmd, struct gf_flock *lock, dict_t *xdata)
 {
-        STACK_UNWIND_STRICT (inodelk, frame, -1, EROFS, xdata);
+        if (is_readonly_or_worm_enabled (this))
+                STACK_UNWIND_STRICT (inodelk, frame, -1, EROFS, xdata);
+        else
+                STACK_WIND_TAIL (frame, FIRST_CHILD (this),
+                                 FIRST_CHILD(this)->fops->inodelk,
+                                 volume, loc, cmd, lock, xdata);
+
         return 0;
 }
 
@@ -60,7 +105,13 @@ int32_t
 ro_finodelk (call_frame_t *frame, xlator_t *this, const char *volume,
              fd_t *fd, int32_t cmd, struct gf_flock *lock, dict_t *xdata)
 {
-        STACK_UNWIND_STRICT (finodelk, frame, -1, EROFS, xdata);
+        if (is_readonly_or_worm_enabled (this))
+                STACK_UNWIND_STRICT (finodelk, frame, -1, EROFS, xdata);
+        else
+                STACK_WIND_TAIL (frame, FIRST_CHILD (this),
+                                 FIRST_CHILD(this)->fops->finodelk,
+                                 volume, fd, cmd, lock, xdata);
+
         return 0;
 }
 
@@ -68,7 +119,13 @@ int32_t
 ro_lk (call_frame_t *frame, xlator_t *this, fd_t *fd, int cmd,
        struct gf_flock *flock, dict_t *xdata)
 {
-        STACK_UNWIND_STRICT (lk, frame, -1, EROFS, NULL, xdata);
+        if (is_readonly_or_worm_enabled (this))
+                STACK_UNWIND_STRICT (lk, frame, -1, EROFS, NULL, xdata);
+        else
+                STACK_WIND_TAIL (frame, FIRST_CHILD (this),
+                                 FIRST_CHILD(this)->fops->lk, fd, cmd, flock,
+                                 xdata);
+
         return 0;
 }
 
@@ -76,15 +133,29 @@ int32_t
 ro_setattr (call_frame_t *frame, xlator_t *this, loc_t *loc,
             struct iatt *stbuf, int32_t valid, dict_t *xdata)
 {
-        STACK_UNWIND_STRICT (setattr, frame, -1, EROFS, NULL, NULL, xdata);
-	return 0;
+        if (is_readonly_or_worm_enabled (this))
+                STACK_UNWIND_STRICT (setattr, frame, -1, EROFS, NULL, NULL,
+                                     xdata);
+        else
+                STACK_WIND_TAIL (frame, FIRST_CHILD (this),
+                                 FIRST_CHILD(this)->fops->setattr, loc, stbuf,
+                                 valid, xdata);
+
+        return 0;
 }
 
 int32_t
 ro_fsetattr (call_frame_t *frame, xlator_t *this, fd_t *fd,
              struct iatt *stbuf, int32_t valid, dict_t *xdata)
 {
-        STACK_UNWIND_STRICT (fsetattr, frame, -1, EROFS, NULL, NULL, xdata);
+        if (is_readonly_or_worm_enabled (this))
+                STACK_UNWIND_STRICT (fsetattr, frame, -1, EROFS, NULL, NULL,
+                                     xdata);
+        else
+                STACK_WIND_TAIL (frame, FIRST_CHILD (this),
+                                 FIRST_CHILD(this)->fops->fsetattr, fd, stbuf,
+                                 valid, xdata);
+
         return 0;
 }
 
@@ -92,14 +163,28 @@ ro_fsetattr (call_frame_t *frame, xlator_t *this, fd_t *fd,
 int32_t
 ro_truncate (call_frame_t *frame, xlator_t *this, loc_t *loc, off_t offset, dict_t *xdata)
 {
-        STACK_UNWIND_STRICT (truncate, frame, -1, EROFS, NULL, NULL, xdata);
+        if (is_readonly_or_worm_enabled (this))
+                STACK_UNWIND_STRICT (truncate, frame, -1, EROFS, NULL, NULL,
+                                     xdata);
+        else
+                STACK_WIND_TAIL (frame, FIRST_CHILD (this),
+                                 FIRST_CHILD(this)->fops->truncate, loc, offset,
+                                 xdata);
+
 	return 0;
 }
 
 int32_t
 ro_ftruncate (call_frame_t *frame, xlator_t *this, fd_t *fd, off_t offset, dict_t *xdata)
 {
-        STACK_UNWIND_STRICT (ftruncate, frame, -1, EROFS, NULL, NULL, xdata);
+        if (is_readonly_or_worm_enabled (this))
+                STACK_UNWIND_STRICT (ftruncate, frame, -1, EROFS, NULL, NULL,
+                                     xdata);
+        else
+                STACK_WIND_TAIL (frame, FIRST_CHILD (this),
+                                 FIRST_CHILD(this)->fops->ftruncate, fd, offset,
+                                 xdata);
+
 	return 0;
 }
 
@@ -107,8 +192,15 @@ int
 ro_mknod (call_frame_t *frame, xlator_t *this, loc_t *loc, mode_t mode,
           dev_t rdev, mode_t umask, dict_t *xdata)
 {
-        STACK_UNWIND_STRICT (mknod, frame, -1, EROFS, NULL, NULL, NULL, NULL, xdata);
-        return 0;
+        if (is_readonly_or_worm_enabled (this))
+                STACK_UNWIND_STRICT (mknod, frame, -1, EROFS, NULL, NULL, NULL,
+                                     NULL, xdata);
+        else
+                STACK_WIND_TAIL (frame, FIRST_CHILD (this),
+                                 FIRST_CHILD(this)->fops->mknod, loc, mode,
+                                 rdev, umask, xdata);
+
+	return 0;
 }
 
 
@@ -116,7 +208,14 @@ int
 ro_mkdir (call_frame_t *frame, xlator_t *this, loc_t *loc, mode_t mode,
           mode_t umask, dict_t *xdata)
 {
-        STACK_UNWIND_STRICT (mkdir, frame, -1, EROFS, NULL, NULL, NULL, NULL, xdata);
+        if (is_readonly_or_worm_enabled (this))
+                STACK_UNWIND_STRICT (mkdir, frame, -1, EROFS, NULL, NULL, NULL,
+                                     NULL, xdata);
+        else
+                STACK_WIND_TAIL (frame, FIRST_CHILD (this),
+                                 FIRST_CHILD(this)->fops->mkdir, loc, mode,
+                                 umask, xdata);
+
         return 0;
 }
 
@@ -124,7 +223,14 @@ int32_t
 ro_unlink (call_frame_t *frame, xlator_t *this, loc_t *loc, int xflag,
            dict_t *xdata)
 {
-        STACK_UNWIND_STRICT (unlink, frame, -1, EROFS, NULL, NULL, xdata);
+        if (is_readonly_or_worm_enabled (this))
+                STACK_UNWIND_STRICT (unlink, frame, -1, EROFS, NULL, NULL,
+                                     xdata);
+        else
+                STACK_WIND_TAIL (frame, FIRST_CHILD (this),
+                                 FIRST_CHILD(this)->fops->unlink, loc, xflag,
+                                 xdata);
+
         return 0;
 }
 
@@ -133,7 +239,14 @@ int
 ro_rmdir (call_frame_t *frame, xlator_t *this, loc_t *loc, int flags,
           dict_t *xdata)
 {
-        STACK_UNWIND_STRICT (rmdir, frame, -1, EROFS, NULL, NULL, xdata);
+        if (is_readonly_or_worm_enabled (this))
+                STACK_UNWIND_STRICT (rmdir, frame, -1, EROFS, NULL, NULL,
+                                     xdata);
+        else
+                STACK_WIND_TAIL (frame, FIRST_CHILD (this),
+                                 FIRST_CHILD(this)->fops->rmdir, loc, flags,
+                                 xdata);
+
         return 0;
 }
 
@@ -142,18 +255,31 @@ int
 ro_symlink (call_frame_t *frame, xlator_t *this, const char *linkpath,
             loc_t *loc, mode_t umask, dict_t *xdata)
 {
-        STACK_UNWIND_STRICT (symlink, frame, -1, EROFS, NULL, NULL, NULL,
-                             NULL, xdata);
+        if (is_readonly_or_worm_enabled (this))
+                STACK_UNWIND_STRICT (symlink, frame, -1, EROFS, NULL, NULL,
+                                     NULL, NULL, xdata);
+        else
+                STACK_WIND_TAIL (frame, FIRST_CHILD (this),
+                                 FIRST_CHILD(this)->fops->symlink, linkpath,
+                                 loc, umask, xdata);
+
         return 0;
 }
 
 
 
 int32_t
-ro_rename (call_frame_t *frame, xlator_t *this, loc_t *oldloc, loc_t *newloc, dict_t *xdata)
+ro_rename (call_frame_t *frame, xlator_t *this, loc_t *oldloc, loc_t *newloc,
+           dict_t *xdata)
 {
-        STACK_UNWIND_STRICT (rename, frame, -1, EROFS, NULL, NULL, NULL, NULL,
-                             NULL, xdata);
+        if (is_readonly_or_worm_enabled (this))
+                STACK_UNWIND_STRICT (rename, frame, -1, EROFS, NULL, NULL, NULL,
+                                     NULL, NULL, xdata);
+        else
+                STACK_WIND_TAIL (frame, FIRST_CHILD (this),
+                                 FIRST_CHILD(this)->fops->rename, oldloc,
+                                 newloc, xdata);
+
         return 0;
 }
 
@@ -161,7 +287,14 @@ ro_rename (call_frame_t *frame, xlator_t *this, loc_t *oldloc, loc_t *newloc, di
 int32_t
 ro_link (call_frame_t *frame, xlator_t *this, loc_t *oldloc, loc_t *newloc, dict_t *xdata)
 {
-        STACK_UNWIND_STRICT (link, frame, -1, EROFS, NULL, NULL, NULL, NULL, xdata);
+        if (is_readonly_or_worm_enabled (this))
+                STACK_UNWIND_STRICT (link, frame, -1, EROFS, NULL, NULL, NULL,
+                                     NULL, xdata);
+        else
+                STACK_WIND_TAIL (frame, FIRST_CHILD (this),
+                                 FIRST_CHILD(this)->fops->link, oldloc, newloc,
+                                 xdata);
+
         return 0;
 }
 
@@ -169,8 +302,14 @@ int32_t
 ro_create (call_frame_t *frame, xlator_t *this, loc_t *loc, int32_t flags,
            mode_t mode, mode_t umask, fd_t *fd, dict_t *xdata)
 {
-        STACK_UNWIND_STRICT (create, frame, -1, EROFS, NULL, NULL, NULL,
-                             NULL, NULL, xdata);
+        if (is_readonly_or_worm_enabled (this))
+                STACK_UNWIND_STRICT (create, frame, -1, EROFS, NULL, NULL, NULL,
+                                     NULL, NULL, xdata);
+        else
+                STACK_WIND_TAIL (frame, FIRST_CHILD (this),
+                                 FIRST_CHILD(this)->fops->create, loc, flags,
+                                 mode, umask, fd, xdata);
+
         return 0;
 }
 
@@ -179,16 +318,17 @@ static int32_t
 ro_open_cbk (call_frame_t *frame, void *cookie, xlator_t *this, int32_t op_ret,
              int32_t op_errno, fd_t *fd, dict_t *xdata)
 {
-	STACK_UNWIND_STRICT (open, frame, op_ret, op_errno, fd, xdata);
-	return 0;
+        STACK_UNWIND_STRICT (open, frame, op_ret, op_errno, fd, xdata);
+        return 0;
 }
 
 int32_t
 ro_open (call_frame_t *frame, xlator_t *this, loc_t *loc, int32_t flags,
          fd_t *fd, dict_t *xdata)
 {
-        if (((flags & O_ACCMODE) == O_WRONLY) ||
-              ((flags & O_ACCMODE) == O_RDWR)) {
+        if (is_readonly_or_worm_enabled (this) &&
+            (((flags & O_ACCMODE) == O_WRONLY) ||
+              ((flags & O_ACCMODE) == O_RDWR))) {
                 STACK_UNWIND_STRICT (open, frame, -1, EROFS, NULL, xdata);
                 return 0;
 	}
@@ -202,23 +342,44 @@ int32_t
 ro_fsetxattr (call_frame_t *frame, xlator_t *this, fd_t *fd, dict_t *dict,
               int32_t flags, dict_t *xdata)
 {
-        STACK_UNWIND_STRICT (fsetxattr, frame, -1, EROFS, xdata);
-	return 0;
+        if (is_readonly_or_worm_enabled (this))
+                STACK_UNWIND_STRICT (fsetxattr, frame, -1, EROFS, xdata);
+        else
+                STACK_WIND_TAIL (frame, FIRST_CHILD (this),
+                                 FIRST_CHILD(this)->fops->fsetxattr, fd, dict,
+                                 flags, xdata);
+
+        return 0;
 }
 
 int32_t
-ro_fsyncdir (call_frame_t *frame, xlator_t *this, fd_t *fd, int32_t flags, dict_t *xdata)
+ro_fsyncdir (call_frame_t *frame, xlator_t *this, fd_t *fd, int32_t flags,
+             dict_t *xdata)
 {
-        STACK_UNWIND_STRICT (fsyncdir, frame, -1, EROFS, xdata);
+        if (is_readonly_or_worm_enabled (this))
+                STACK_UNWIND_STRICT (fsyncdir, frame, -1, EROFS, xdata);
+        else
+                STACK_WIND_TAIL (frame, FIRST_CHILD (this),
+                                 FIRST_CHILD(this)->fops->fsyncdir, fd, flags,
+                                 xdata);
+
 	return 0;
 }
 
 int32_t
 ro_writev (call_frame_t *frame, xlator_t *this, fd_t *fd, struct iovec *vector,
-           int32_t count, off_t off, uint32_t flags, struct iobref *iobref, dict_t *xdata)
+           int32_t count, off_t off, uint32_t flags, struct iobref *iobref,
+           dict_t *xdata)
 {
-        STACK_UNWIND_STRICT (writev, frame, -1, EROFS, NULL, NULL, xdata);
-        return 0;
+        if (is_readonly_or_worm_enabled (this))
+                STACK_UNWIND_STRICT (writev, frame, -1, EROFS, NULL, NULL,
+                                     xdata);
+        else
+                STACK_WIND_TAIL (frame, FIRST_CHILD (this),
+                                 FIRST_CHILD(this)->fops->writev, fd, vector,
+                                 count, off, flags, iobref, xdata);
+
+	return 0;
 }
 
 
@@ -226,14 +387,26 @@ int32_t
 ro_setxattr (call_frame_t *frame, xlator_t *this, loc_t *loc, dict_t *dict,
              int32_t flags, dict_t *xdata)
 {
-        STACK_UNWIND_STRICT (setxattr, frame, -1, EROFS, xdata);
-        return 0;
+        if (is_readonly_or_worm_enabled (this))
+                STACK_UNWIND_STRICT (setxattr, frame, -1, EROFS, xdata);
+        else
+                STACK_WIND_TAIL (frame, FIRST_CHILD (this),
+                                 FIRST_CHILD(this)->fops->setxattr, loc, dict,
+                                 flags, xdata);
+
+	return 0;
 }
 
 int32_t
 ro_removexattr (call_frame_t *frame, xlator_t *this, loc_t *loc,
                 const char *name, dict_t *xdata)
 {
-        STACK_UNWIND_STRICT (removexattr, frame, -1, EROFS, xdata);
+        if (is_readonly_or_worm_enabled (this))
+                STACK_UNWIND_STRICT (removexattr, frame, -1, EROFS, xdata);
+        else
+                STACK_WIND_TAIL (frame, FIRST_CHILD (this),
+                                 FIRST_CHILD(this)->fops->removexattr, loc,
+                                 name, xdata);
+
         return 0;
 }
diff --git a/xlators/features/read-only/src/read-only-common.h b/xlators/features/read-only/src/read-only-common.h
index 5d4c7e260ed..c99ba0310bb 100644
--- a/xlators/features/read-only/src/read-only-common.h
+++ b/xlators/features/read-only/src/read-only-common.h
@@ -15,6 +15,9 @@
 #include "xlator.h"
 #include "defaults.h"
 
+gf_boolean_t
+is_readonly_or_worm_enabled (xlator_t *this);
+
 int32_t
 ro_xattrop (call_frame_t *frame, xlator_t *this, loc_t *loc,
             gf_xattrop_flags_t flags, dict_t *dict, dict_t *xdata);
diff --git a/xlators/features/read-only/src/read-only-mem-types.h b/xlators/features/read-only/src/read-only-mem-types.h
new file mode 100644
index 00000000000..940700a017d
--- /dev/null
+++ b/xlators/features/read-only/src/read-only-mem-types.h
@@ -0,0 +1,20 @@
+/*
+   Copyright (c) 2014 Red Hat, Inc. <http://www.redhat.com>
+   This file is part of GlusterFS.
+
+   This file is licensed to you under your choice of the GNU Lesser
+   General Public License, version 3 or any later version (LGPLv3 or
+   later), or the GNU General Public License, version 2 (GPLv2), in all
+   cases as published by the Free Software Foundation.
+*/
+
+#ifndef __READONLY_MEM_TYPES_H__
+#define __READONLY_MEM_TYPES_H__
+
+#include "mem-types.h"
+
+enum gf_read_only_mem_types_ {
+        gf_read_only_mt_priv_t = gf_common_mt_end + 1,
+        gf_read_only_mt_end
+};
+#endif
diff --git a/xlators/features/read-only/src/read-only.c b/xlators/features/read-only/src/read-only.c
index e49e54a1b31..173c4f51e2a 100644
--- a/xlators/features/read-only/src/read-only.c
+++ b/xlators/features/read-only/src/read-only.c
@@ -12,14 +12,31 @@
 #include "config.h"
 #endif
 
-#include "xlator.h"
 #include "defaults.h"
 #include "read-only-common.h"
+#include "read-only-mem-types.h"
+#include "read-only.h"
+
+int32_t
+mem_acct_init (xlator_t *this)
+{
+        int     ret = -1;
+
+        ret = xlator_mem_acct_init (this, gf_read_only_mt_end + 1);
+        if (ret)
+                gf_log (this->name, GF_LOG_ERROR, "Memory accounting "
+                        "initialization failed.");
+
+        return ret;
+}
 
 int32_t
 init (xlator_t *this)
 {
-	if (!this->children || this->children->next) {
+	int                     ret     = -1;
+        read_only_priv_t       *priv    = NULL;
+
+        if (!this->children || this->children->next) {
 		gf_log (this->name, GF_LOG_ERROR,
 			"translator not configured with exactly one child");
 		return -1;
@@ -30,14 +47,50 @@ init (xlator_t *this)
 			"dangling volume. check volfile ");
 	}
 
-	return 0;
+        priv = GF_CALLOC (1, sizeof (*priv), gf_read_only_mt_priv_t);
+        if (!priv)
+                goto out;
+
+        GF_OPTION_INIT ("read-only", priv->readonly_or_worm_enabled, bool, out);
+
+        this->private = priv;
+        ret = 0;
+out:
+        return ret;
 }
 
+int
+reconfigure (xlator_t *this, dict_t *options)
+{
+        read_only_priv_t  *priv             = NULL;
+        int                ret              = -1;
+        gf_boolean_t       readonly_or_worm_enabled = _gf_false;
+
+        priv = this->private;
+        GF_ASSERT (priv);
+
+        GF_OPTION_RECONF ("read-only", readonly_or_worm_enabled, options, bool,
+                          out);
+        priv->readonly_or_worm_enabled = readonly_or_worm_enabled;
+        ret = 0;
+out:
+        gf_log (this->name, GF_LOG_DEBUG, "returning %d", ret);
+        return ret;
+}
 
 void
 fini (xlator_t *this)
 {
-	return;
+        read_only_priv_t         *priv    = NULL;
+
+        priv = this->private;
+        if (!priv)
+                return;
+
+        this->private = NULL;
+        GF_FREE (priv);
+
+        return;
 }
 
 
@@ -73,5 +126,10 @@ struct xlator_cbks cbks = {
 };
 
 struct volume_options options[] = {
-	{ .key = {NULL} },
+	{ .key  = {"read-only"},
+          .type = GF_OPTION_TYPE_BOOL,
+          .default_value = "off",
+          .description = "When \"on\", makes a volume read-only. It is turned "
+                         "\"off\" by default."
+        },
 };
diff --git a/xlators/features/read-only/src/read-only.h b/xlators/features/read-only/src/read-only.h
new file mode 100644
index 00000000000..8e7e1b68081
--- /dev/null
+++ b/xlators/features/read-only/src/read-only.h
@@ -0,0 +1,21 @@
+/*
+   Copyright (c) 2014 Red Hat, Inc. <http://www.redhat.com>
+   This file is part of GlusterFS.
+
+   This file is licensed to you under your choice of the GNU Lesser
+   General Public License, version 3 or any later version (LGPLv3 or
+   later), or the GNU General Public License, version 2 (GPLv2), in all
+   cases as published by the Free Software Foundation.
+*/
+
+#ifndef __READONLY_H__
+#define __READONLY_H__
+
+#include "read-only-mem-types.h"
+#include "xlator.h"
+
+typedef struct {
+        gf_boolean_t      readonly_or_worm_enabled;
+} read_only_priv_t;
+
+#endif
diff --git a/xlators/features/read-only/src/worm.c b/xlators/features/read-only/src/worm.c
index 16c3eb3daed..f62ffa3f6f3 100644
--- a/xlators/features/read-only/src/worm.c
+++ b/xlators/features/read-only/src/worm.c
@@ -15,6 +15,21 @@
 #include "xlator.h"
 #include "defaults.h"
 #include "read-only-common.h"
+#include "read-only-mem-types.h"
+#include "read-only.h"
+
+int32_t
+mem_acct_init (xlator_t *this)
+{
+        int     ret = -1;
+
+        ret = xlator_mem_acct_init (this, gf_read_only_mt_end + 1);
+        if (ret)
+                gf_log (this->name, GF_LOG_ERROR, "Memory accounting "
+                        "initialization failed.");
+
+        return ret;
+}
 
 static int32_t
 worm_open_cbk (call_frame_t *frame, void *cookie, xlator_t *this, int32_t op_ret,
@@ -28,9 +43,10 @@ int32_t
 worm_open (call_frame_t *frame, xlator_t *this, loc_t *loc, int32_t flags,
            fd_t *fd, dict_t *xdata)
 {
-        if ((((flags & O_ACCMODE) == O_WRONLY) ||
+        if (is_readonly_or_worm_enabled (this) &&
+            ((((flags & O_ACCMODE) == O_WRONLY) ||
               ((flags & O_ACCMODE) == O_RDWR)) &&
-              !(flags & O_APPEND)) {
+              !(flags & O_APPEND))) {
                 STACK_UNWIND_STRICT (open, frame, -1, EROFS, NULL, NULL);
                 return 0;
         }
@@ -43,6 +59,9 @@ worm_open (call_frame_t *frame, xlator_t *this, loc_t *loc, int32_t flags,
 int32_t
 init (xlator_t *this)
 {
+        int                     ret     = -1;
+        read_only_priv_t       *priv    = NULL;
+
         if (!this->children || this->children->next) {
                 gf_log (this->name, GF_LOG_ERROR,
                         "translator not configured with exactly one child");
@@ -54,13 +73,49 @@ init (xlator_t *this)
                         "dangling volume. check volfile ");
         }
 
-        return 0;
+        priv = GF_CALLOC (1, sizeof (*priv), gf_read_only_mt_priv_t);
+        if (!priv)
+                goto out;
+
+        GF_OPTION_INIT ("worm", priv->readonly_or_worm_enabled, bool, out);
+
+        this->private = priv;
+        ret = 0;
+out:
+        return ret;
 }
 
+int
+reconfigure (xlator_t *this, dict_t *options)
+{
+        read_only_priv_t  *priv                     = NULL;
+        int                ret                      = -1;
+        gf_boolean_t       readonly_or_worm_enabled = _gf_false;
+
+        priv = this->private;
+        GF_ASSERT (priv);
+
+        GF_OPTION_RECONF ("worm", readonly_or_worm_enabled, options, bool, out);
+
+        priv->readonly_or_worm_enabled = readonly_or_worm_enabled;
+        ret = 0;
+out:
+        gf_log (this->name, GF_LOG_DEBUG, "returning %d", ret);
+        return ret;
+}
 
 void
 fini (xlator_t *this)
 {
+        read_only_priv_t         *priv    = NULL;
+
+        priv = this->private;
+        if (!priv)
+                return;
+
+        this->private = NULL;
+        GF_FREE (priv);
+
         return;
 }
 
@@ -84,6 +139,11 @@ struct xlator_fops fops = {
 struct xlator_cbks cbks;
 
 struct volume_options options[] = {
-        { .key = {NULL} },
+        { .key  = {"worm"},
+          .type = GF_OPTION_TYPE_BOOL,
+          .default_value = "off",
+          .description = "When \"on\", makes a volume get write once read many "
+                         " feature. It is turned \"off\" by default."
+        },
 };
 
diff --git a/xlators/mgmt/glusterd/src/glusterd-volgen.c b/xlators/mgmt/glusterd/src/glusterd-volgen.c
index 3edd4e28009..d7709750129 100644
--- a/xlators/mgmt/glusterd/src/glusterd-volgen.c
+++ b/xlators/mgmt/glusterd/src/glusterd-volgen.c
@@ -1852,15 +1852,12 @@ brick_graph_add_ro (volgen_graph_t *graph, glusterd_volinfo_t *volinfo,
                 goto out;
         }
 
-        /* Check for read-only volume option, and add it to the graph */
-        if (dict_get_str_boolean (set_dict, "features.read-only", 0)){
-                xl = volgen_graph_add (graph, "features/read-only",
-                                       volinfo->volname);
-                if (!xl) {
-                        ret = -1;
-                        goto out;
-                }
-        }
+        xl = volgen_graph_add (graph, "features/read-only", volinfo->volname);
+        if (!xl)
+                return -1;
+        ret = xlator_set_option (xl, "read-only", "off");
+                if (ret)
+                        return -1;
 
         ret = 0;
 
@@ -1886,15 +1883,9 @@ brick_graph_add_worm (volgen_graph_t *graph, glusterd_volinfo_t *volinfo,
                 goto out;
         }
 
-        /* Check for worm volume option, and add it to the graph */
-        if (dict_get_str_boolean (set_dict, "features.worm", 0)) {
-                xl = volgen_graph_add (graph, "features/worm",
-                                       volinfo->volname);
-                if (!xl) {
-                        ret = -1;
-                        goto out;
-                }
-        }
+        xl = volgen_graph_add (graph, "features/worm", volinfo->volname);
+        if (!xl)
+                return -1;
 
         ret = 0;
 
@@ -3399,6 +3390,10 @@ client_graph_builder (volgen_graph_t *graph, glusterd_volinfo_t *volinfo,
                         ret = -1;
                         goto out;
                 }
+                ret = xlator_set_option (xl, "read-only", "on");
+                if (ret)
+                        goto out;
+
         }
 
         /* Check for compress volume option, and add it to the graph on client side */
diff --git a/xlators/mgmt/glusterd/src/glusterd-volume-set.c b/xlators/mgmt/glusterd/src/glusterd-volume-set.c
index 1e2bbcad265..d95f1fcaa0a 100644
--- a/xlators/mgmt/glusterd/src/glusterd-volume-set.c
+++ b/xlators/mgmt/glusterd/src/glusterd-volume-set.c
@@ -1547,14 +1547,13 @@ struct volopt_map_entry glusterd_volopt_map[] = {
         /* Other options which don't fit any place above */
         { .key        = "features.read-only",
           .voltype    = "features/read-only",
-          .option     = "!read-only",
-          .value      = "off",
+          .option     = "read-only",
           .op_version = 1,
           .flags      = OPT_FLAG_CLIENT_OPT | OPT_FLAG_XLATOR_OPT
         },
         { .key        = "features.worm",
           .voltype    = "features/worm",
-          .option     = "!worm",
+          .option     = "worm",
           .value      = "off",
           .op_version = 2,
           .flags      = OPT_FLAG_CLIENT_OPT | OPT_FLAG_XLATOR_OPT