LU-5731 osp: flush async updates for osp_sync

Current osp_sync() only considers the async requests that are
handled by the osp_sync_thread, but ignores the async updates
that are handled directly by the background ptlrpcd threads.
Usually, such async updates are for LFSCK remote repairing.
This patch will flush all of them when dt_sync() is called.

Signed-off-by: Fan Yong <fan.yong@intel.com>
Change-Id: I0e6d54120acbd8ab82cf776222277ae3b805812d
Reviewed-on: http://review.whamcloud.com/12359
Tested-by: Jenkins
Tested-by: Maloo <hpdd-maloo@intel.com>
Reviewed-by: Alex Zhuravlev <alexey.zhuravlev@intel.com>
Reviewed-by: Lai Siyao <lai.siyao@intel.com>
Reviewed-by: Oleg Drokin <oleg.drokin@intel.com>

diff --git a/lustre/osp/osp_dev.c b/lustre/osp/osp_dev.c
index 61e697f..8b038a1 100644 (file)
--- a/lustre/osp/osp_dev.c
+++ b/lustre/osp/osp_dev.c
@@ -694,6 +694,20 @@ static int osp_sync(const struct lu_env *env, struct dt_device *dev)
                RETURN(-ENOTCONN);
 
        id = d->opd_syn_last_used_id;
                RETURN(-ENOTCONN);
 
        id = d->opd_syn_last_used_id;

+       down_write(&d->opd_async_updates_rwsem);
+
+       CDEBUG(D_OTHER, "%s: async updates %d\n", d->opd_obd->obd_name,
+              atomic_read(&d->opd_async_updates_count));
+
+       /* make sure the connection is fine */
+       expire = cfs_time_shift(obd_timeout);
+       lwi = LWI_TIMEOUT(expire - cfs_time_current(), osp_sync_timeout, d);
+       rc = l_wait_event(d->opd_syn_barrier_waitq,
+                         atomic_read(&d->opd_async_updates_count) == 0,
+                         &lwi);
+       up_write(&d->opd_async_updates_rwsem);
+       if (rc != 0)
+               GOTO(out, rc);

 
        CDEBUG(D_OTHER, "%s: id: used %lu, processed %lu\n",
               d->opd_obd->obd_name, id, d->opd_syn_last_processed_id);
 
        CDEBUG(D_OTHER, "%s: id: used %lu, processed %lu\n",
               d->opd_obd->obd_name, id, d->opd_syn_last_processed_id);


@@ -867,6 +881,9 @@ static int osp_init0(const struct lu_env *env, struct osp_device *osp,
        ENTRY;
 
        mutex_init(&osp->opd_async_requests_mutex);
        ENTRY;
 
        mutex_init(&osp->opd_async_requests_mutex);

+       INIT_LIST_HEAD(&osp->opd_async_updates);
+       init_rwsem(&osp->opd_async_updates_rwsem);
+       atomic_set(&osp->opd_async_updates_count, 0);

 
        obd = class_name2obd(lustre_cfg_string(cfg, 0));
        if (obd == NULL) {
 
        obd = class_name2obd(lustre_cfg_string(cfg, 0));
        if (obd == NULL) {

diff --git a/lustre/osp/osp_internal.h b/lustre/osp/osp_internal.h
index 4ce2eec..d519151 100644 (file)
--- a/lustre/osp/osp_internal.h
+++ b/lustre/osp/osp_internal.h
@@ -198,6 +198,9 @@ struct osp_device {
        struct dt_update_request        *opd_async_requests;
        /* Protect current operations on opd_async_requests. */
        struct mutex                     opd_async_requests_mutex;
        struct dt_update_request        *opd_async_requests;
        /* Protect current operations on opd_async_requests. */
        struct mutex                     opd_async_requests_mutex;

+       struct list_head                 opd_async_updates;
+       struct rw_semaphore              opd_async_updates_rwsem;
+       atomic_t                         opd_async_updates_count;

 };
 
 #define opd_pre_lock                   opd_pre->osp_pre_lock
 };
 
 #define opd_pre_lock                   opd_pre->osp_pre_lock

diff --git a/lustre/osp/osp_trans.c b/lustre/osp/osp_trans.c
index 25d92ea..cf9d4d2 100644 (file)
--- a/lustre/osp/osp_trans.c
+++ b/lustre/osp/osp_trans.c
@@ -63,6 +63,8 @@
 
 struct osp_async_update_args {
        struct dt_update_request *oaua_update;
 
 struct osp_async_update_args {
        struct dt_update_request *oaua_update;

+       atomic_t                 *oaua_count;
+       wait_queue_head_t        *oaua_waitq;

        bool                      oaua_flow_control;
 };
 
        bool                      oaua_flow_control;
 };
 


@@ -195,6 +197,9 @@ static int osp_async_update_interpret(const struct lu_env *env,
                index++;
        }
 
                index++;
        }
 

+       if (oaua->oaua_count != NULL && atomic_dec_and_test(oaua->oaua_count))
+               wake_up_all(oaua->oaua_waitq);
+

        dt_update_request_destroy(dt_update);
 
        return 0;
        dt_update_request_destroy(dt_update);
 
        return 0;


@@ -238,6 +243,9 @@ int osp_unplug_async_request(const struct lu_env *env,
 
                args = ptlrpc_req_async_args(req);
                args->oaua_update = update;
 
                args = ptlrpc_req_async_args(req);
                args->oaua_update = update;

+               args->oaua_count = NULL;
+               args->oaua_waitq = NULL;
+               args->oaua_flow_control = false;

                req->rq_interpret_reply = osp_async_update_interpret;
                ptlrpcd_add_req(req, PDL_POLICY_LOCAL, -1);
        }
                req->rq_interpret_reply = osp_async_update_interpret;
                ptlrpcd_add_req(req, PDL_POLICY_LOCAL, -1);
        }


@@ -453,11 +461,19 @@ static int osp_trans_trigger(const struct lu_env *env, struct osp_device *osp,
                rc = out_prep_update_req(env, osp->opd_obd->u.cli.cl_import,
                                         dt_update->dur_buf.ub_req, &req);
                if (rc == 0) {
                rc = out_prep_update_req(env, osp->opd_obd->u.cli.cl_import,
                                         dt_update->dur_buf.ub_req, &req);
                if (rc == 0) {

+                       down_read(&osp->opd_async_updates_rwsem);
+

                        args = ptlrpc_req_async_args(req);
                        args->oaua_update = dt_update;
                        args = ptlrpc_req_async_args(req);
                        args->oaua_update = dt_update;

+                       args->oaua_count = &osp->opd_async_updates_count;
+                       args->oaua_waitq = &osp->opd_syn_barrier_waitq;

                        args->oaua_flow_control = flow_control;
                        req->rq_interpret_reply =
                                osp_async_update_interpret;
                        args->oaua_flow_control = flow_control;
                        req->rq_interpret_reply =
                                osp_async_update_interpret;

+
+                       atomic_inc(args->oaua_count);
+                       up_read(&osp->opd_async_updates_rwsem);
+

                        ptlrpcd_add_req(req, PDL_POLICY_LOCAL, -1);
                } else {
                        dt_update_request_destroy(dt_update);
                        ptlrpcd_add_req(req, PDL_POLICY_LOCAL, -1);
                } else {
                        dt_update_request_destroy(dt_update);