LU-8900 snapshot: check write barrier before modification

For client sponsored modifications, the RPC service thread on the
MDT will check whether there is write barrier or not when creates
transaction handler, if yes, return -EINPROGRESS to the caller.

Generally, the check is done inside mdd_trans_create(). For those
cases that bypass mdd_trans_create(), such as some mdd_iocontrol,
quota, will be checked individually.

For open-unlink case, when close the last open handler, it will
try to destroy the orphan. But if the close-destroy is blocked
by the barrier, the orphan MDT-object will be kept in the MDT's
orphan list and will be destroyed automatically when remount the
MDT next time. Since barrier-snapshot is relative rare operation,
and race with close-destroy will be more rare, such solution is
reasonable and will not cause serious trouble.

When client get the -EINPROGRESS error, the expected behaviour is
to re-try the RPC some time later. That is the standard action on
the client-side, not only for blocked by barrier but also for some
cases, such as the case of related OI mapping is invaid and the OI
scrub is rebuilding the OI files. So there is no interoperability
trouble caused by MDT-side write barrier.

Signed-off-by: Fan Yong <fan.yong@intel.com>
Change-Id: I8c3571f7f89fc9ff7397457955ffc75543eb2164
Reviewed-on: https://review.whamcloud.com/24264
Tested-by: Jenkins
Tested-by: Maloo <hpdd-maloo@intel.com>
Reviewed-by: Niu Yawei <yawei.niu@intel.com>
Reviewed-by: Lai Siyao <lai.siyao@intel.com>
Reviewed-by: Oleg Drokin <oleg.drokin@intel.com>

diff --git a/lustre/include/lustre_barrier.h b/lustre/include/lustre_barrier.h
index 13ed590..231eae9 100644 (file)
--- a/lustre/include/lustre_barrier.h
+++ b/lustre/include/lustre_barrier.h
@@ -35,6 +35,8 @@
 #include <dt_object.h>
 #include <lustre_export.h>
 
+bool barrier_entry(struct dt_device *key);
+void barrier_exit(struct dt_device *key);
 int barrier_handler(struct dt_device *key, struct ptlrpc_request *req);
 int barrier_register(struct dt_device *key, struct dt_device *next);
 void barrier_deregister(struct dt_device *key);

diff --git a/lustre/mdd/mdd_device.c b/lustre/mdd/mdd_device.c
index 57fe3de..7192e6c 100644 (file)
--- a/lustre/mdd/mdd_device.c
+++ b/lustre/mdd/mdd_device.c
@@ -1602,20 +1602,22 @@ out:
 static int mdd_iocontrol(const struct lu_env *env, struct md_device *m,
                          unsigned int cmd, int len, void *karg)
 {
-       struct mdd_device *mdd;
+       struct mdd_device *mdd = lu2mdd_dev(&m->md_lu_dev);
        struct obd_ioctl_data *data = karg;
        int rc;
-
        ENTRY;
 
-       mdd = lu2mdd_dev(&m->md_lu_dev);
-
        /* Doesn't use obd_ioctl_data */
        switch (cmd) {
        case OBD_IOC_CHANGELOG_CLEAR: {
                struct changelog_setinfo *cs = karg;
+
+               if (unlikely(!barrier_entry(mdd->mdd_bottom)))
+                       RETURN(-EINPROGRESS);
+
                rc = mdd_changelog_clear(env, mdd, cs->cs_id,
                                         cs->cs_recno);
+               barrier_exit(mdd->mdd_bottom);
                RETURN(rc);
        }
        case OBD_IOC_GET_MNTOPT: {
@@ -1653,10 +1655,18 @@ static int mdd_iocontrol(const struct lu_env *env, struct md_device *m,
 
        switch (cmd) {
        case OBD_IOC_CHANGELOG_REG:
+               if (unlikely(!barrier_entry(mdd->mdd_bottom)))
+                       RETURN(-EINPROGRESS);
+
                rc = mdd_changelog_user_register(env, mdd, &data->ioc_u32_1);
+               barrier_exit(mdd->mdd_bottom);
                break;
        case OBD_IOC_CHANGELOG_DEREG:
+               if (unlikely(!barrier_entry(mdd->mdd_bottom)))
+                       RETURN(-EINPROGRESS);
+
                rc = mdd_changelog_user_purge(env, mdd, data->ioc_u32_1);
+               barrier_exit(mdd->mdd_bottom);
                break;
        default:
                rc = -ENOTTY;

diff --git a/lustre/mdd/mdd_object.c b/lustre/mdd/mdd_object.c
index 36de98f..2f8e51a 100644 (file)
--- a/lustre/mdd/mdd_object.c
+++ b/lustre/mdd/mdd_object.c
@@ -1756,6 +1756,7 @@ static int mdd_close(const struct lu_env *env, struct md_object *obj,
        struct thandle *handle = NULL;
        int is_orphan = 0;
        int rc;
+       bool blocked = false;
        ENTRY;
 
        if (ma->ma_valid & MA_FLAGS && ma->ma_attr_flags & MDS_KEEP_ORPHAN) {
@@ -1777,24 +1778,43 @@ static int mdd_close(const struct lu_env *env, struct md_object *obj,
 
 again:
        if (is_orphan) {
-                handle = mdd_trans_create(env, mdo2mdd(obj));
-                if (IS_ERR(handle))
-                        RETURN(PTR_ERR(handle));
+               /* mdd_trans_create() maybe failed because of barrier_entry(),
+                * under such case, the orphan MDT-object will be left in the
+                * orphan list, and when the MDT remount next time, the unused
+                * orphans will be destroyed automatically.
+                *
+                * One exception: the former mdd_finish_unlink may failed to
+                * add the orphan MDT-object to the orphan list, then if the
+                * mdd_trans_create() failed because of barrier_entry(), the
+                * MDT-object will become real orphan that is neither in the
+                * namespace nor in the orphan list. Such bad case should be
+                * very rare and will be handled by e2fsck/lfsck. */
+               handle = mdd_trans_create(env, mdo2mdd(obj));
+               if (IS_ERR(handle)) {
+                       rc = PTR_ERR(handle);
+                       if (rc != -EINPROGRESS)
+                               GOTO(stop, rc);
 
-                rc = mdd_declare_close(env, mdd_obj, ma, handle);
-                if (rc)
-                        GOTO(stop, rc);
+                       handle = NULL;
+                       blocked = true;
+                       goto cont;
+               }
+
+               rc = mdd_declare_close(env, mdd_obj, ma, handle);
+               if (rc)
+                       GOTO(stop, rc);
 
                rc = mdd_declare_changelog_store(env, mdd, NULL, NULL, handle);
                if (rc)
                        GOTO(stop, rc);
 
-                rc = mdd_trans_start(env, mdo2mdd(obj), handle);
-                if (rc)
-                        GOTO(stop, rc);
-        }
+               rc = mdd_trans_start(env, mdo2mdd(obj), handle);
+               if (rc)
+                       GOTO(stop, rc);
+       }
 
-        mdd_write_lock(env, mdd_obj, MOR_TGT_CHILD);
+cont:
+       mdd_write_lock(env, mdd_obj, MOR_TGT_CHILD);
        rc = mdd_la_get(env, mdd_obj, &ma->ma_attr);
        if (rc != 0) {
                CERROR("Failed to get lu_attr of "DFID": %d\n",
@@ -1807,14 +1827,14 @@ again:
                    ((mdd_obj->mod_flags & (ORPHAN_OBJ | DEAD_OBJ)) != 0 ||
                     ma->ma_attr.la_nlink == 0);
 
-       if (is_orphan && handle == NULL) {
+       if (is_orphan && !handle && !blocked) {
                mdd_write_unlock(env, mdd_obj);
                goto again;
        }
 
        mdd_obj->mod_count--; /*release open count */
 
-       if (!is_orphan)
+       if (!is_orphan || blocked)
                GOTO(out, rc = 0);
 
        /* Orphan object */
@@ -1850,14 +1870,14 @@ again:
        EXIT;
 
 out:
-        mdd_write_unlock(env, mdd_obj);
-
-        if (rc == 0 &&
-            (mode & (FMODE_WRITE | MDS_OPEN_APPEND | MDS_OPEN_TRUNC)) &&
-            !(ma->ma_valid & MA_FLAGS && ma->ma_attr_flags & MDS_RECOV_OPEN)) {
-                if (handle == NULL) {
-                        handle = mdd_trans_create(env, mdo2mdd(obj));
-                        if (IS_ERR(handle))
+       mdd_write_unlock(env, mdd_obj);
+
+       if (!rc && !blocked &&
+           (mode & (FMODE_WRITE | MDS_OPEN_APPEND | MDS_OPEN_TRUNC)) &&
+           !(ma->ma_valid & MA_FLAGS && ma->ma_attr_flags & MDS_RECOV_OPEN)) {
+               if (handle == NULL) {
+                       handle = mdd_trans_create(env, mdo2mdd(obj));
+                       if (IS_ERR(handle))
                                GOTO(stop, rc = PTR_ERR(handle));
 
                        rc = mdd_declare_changelog_store(env, mdd, NULL, NULL,

diff --git a/lustre/mdd/mdd_trans.c b/lustre/mdd/mdd_trans.c
index 0254e99..6481ea1 100644 (file)
--- a/lustre/mdd/mdd_trans.c
+++ b/lustre/mdd/mdd_trans.c
@@ -41,12 +41,20 @@
 #include <obd_class.h>
 #include <lprocfs_status.h>
 #include <lustre_mds.h>
+#include <lustre_barrier.h>
 
 #include "mdd_internal.h"
 
 struct thandle *mdd_trans_create(const struct lu_env *env,
                                  struct mdd_device *mdd)
 {
+       /* If blocked by the write barrier, then return "-EINPROGRESS"
+        * to the caller. Usually, such error will be forwarded to the
+        * client, and the expected behaviour is to re-try such modify
+        * RPC some time later until the barrier is thawed or expired. */
+       if (unlikely(!barrier_entry(mdd->mdd_bottom)))
+               return ERR_PTR(-EINPROGRESS);
+
         return mdd_child_ops(mdd)->dt_trans_create(env, mdd->mdd_child);
 }
 
@@ -63,6 +71,7 @@ int mdd_trans_stop(const struct lu_env *env, struct mdd_device *mdd,
 
        handle->th_result = result;
        rc = mdd_child_ops(mdd)->dt_trans_stop(env, mdd->mdd_child, handle);
+       barrier_exit(mdd->mdd_bottom);
 
        /* if operation failed, return \a result, otherwise return status of
         * dt_trans_stop */

diff --git a/lustre/mdt/mdt_handler.c b/lustre/mdt/mdt_handler.c
index e71e12f..4d1fbaf 100644 (file)
--- a/lustre/mdt/mdt_handler.c
+++ b/lustre/mdt/mdt_handler.c
@@ -61,6 +61,7 @@
 #include <lustre_swab.h>
 #include <obd.h>
 #include <obd_support.h>
+#include <lustre_barrier.h>
 
 #include <llog_swab.h>
 
@@ -2100,9 +2101,17 @@ static int mdt_quotactl(struct tgt_session_info *tsi)
        if (repoqc == NULL)
                GOTO(out_nodemap, rc = err_serious(-EFAULT));
 
+       if (oqctl->qc_cmd == Q_SETINFO || oqctl->qc_cmd == Q_SETQUOTA)
+               barrier_exit(tsi->tsi_tgt->lut_bottom);
+
        if (oqctl->qc_id != id)
                swap(oqctl->qc_id, id);
 
+       if (oqctl->qc_cmd == Q_SETINFO || oqctl->qc_cmd == Q_SETQUOTA) {
+               if (unlikely(!barrier_entry(tsi->tsi_tgt->lut_bottom)))
+                       RETURN(-EINPROGRESS);
+       }
+
        switch (oqctl->qc_cmd) {
 
        case Q_GETINFO:
@@ -4926,6 +4935,7 @@ static struct lu_object *mdt_object_alloc(const struct lu_env *env,
                spin_lock_init(&mo->mot_write_lock);
                mutex_init(&mo->mot_lov_mutex);
                init_rwsem(&mo->mot_open_sem);
+               atomic_set(&mo->mot_open_count, 0);
                RETURN(o);
        }
        RETURN(NULL);

diff --git a/lustre/target/barrier.c b/lustre/target/barrier.c
index 3710908..e52d303 100644 (file)
--- a/lustre/target/barrier.c
+++ b/lustre/target/barrier.c
@@ -237,6 +237,51 @@ void barrier_fini(void)
        LASSERT(list_empty(&barrier_instance_list));
 }
 
+bool barrier_entry(struct dt_device *key)
+{
+       struct barrier_instance *barrier;
+       bool entered = false;
+       ENTRY;
+
+       barrier = barrier_instance_find(key);
+       if (unlikely(!barrier))
+               /* Fail open */
+               RETURN(true);
+
+       read_lock(&barrier->bi_rwlock);
+       if (likely(barrier->bi_status != BS_FREEZING_P1 &&
+                  barrier->bi_status != BS_FREEZING_P2 &&
+                  barrier->bi_status != BS_FROZEN) ||
+           cfs_time_beforeq(barrier->bi_deadline, cfs_time_current_sec())) {
+               percpu_counter_inc(&barrier->bi_writers);
+               entered = true;
+       }
+       read_unlock(&barrier->bi_rwlock);
+
+       barrier_instance_put(barrier);
+       return entered;
+}
+EXPORT_SYMBOL(barrier_entry);
+
+void barrier_exit(struct dt_device *key)
+{
+       struct barrier_instance *barrier;
+
+       barrier = barrier_instance_find(key);
+       if (likely(barrier)) {
+               percpu_counter_dec(&barrier->bi_writers);
+
+               /* Avoid out-of-order execution the decreasing inflight
+                * modifications count and the check of barrier status. */
+               smp_mb();
+
+               if (unlikely(barrier->bi_status == BS_FREEZING_P1))
+                       wake_up_all(&barrier->bi_waitq);
+               barrier_instance_put(barrier);
+       }
+}
+EXPORT_SYMBOL(barrier_exit);
+
 int barrier_handler(struct dt_device *key, struct ptlrpc_request *req)
 {
        struct ldlm_gl_barrier_desc *desc;