LU-5287 export: hold exp_lock when modify exp_flags

[fs/lustre-release.git] / lustre / ldlm / ldlm_lib.c

diff --git a/lustre/ldlm/ldlm_lib.c b/lustre/ldlm/ldlm_lib.c
index 1c370e6..0eb34d7 100644 (file)
--- a/lustre/ldlm/ldlm_lib.c
+++ b/lustre/ldlm/ldlm_lib.c
@@ -43,11 +43,7 @@
 
 #define DEBUG_SUBSYSTEM S_LDLM
 
-#ifdef __KERNEL__
-# include <libcfs/libcfs.h>
-#else
-# include <liblustre.h>
-#endif
+#include <libcfs/libcfs.h>
 #include <obd.h>
 #include <obd_class.h>
 #include <lustre_dlm.h>
@@ -373,11 +369,11 @@ int client_obd_setup(struct obd_device *obddev, struct lustre_cfg *lcfg)
        /* lru for osc. */
        INIT_LIST_HEAD(&cli->cl_lru_osc);
        atomic_set(&cli->cl_lru_shrinkers, 0);
-       atomic_set(&cli->cl_lru_busy, 0);
-       atomic_set(&cli->cl_lru_in_list, 0);
+       atomic_long_set(&cli->cl_lru_busy, 0);
+       atomic_long_set(&cli->cl_lru_in_list, 0);
        INIT_LIST_HEAD(&cli->cl_lru_list);
        client_obd_list_lock_init(&cli->cl_lru_list_lock);
-       atomic_set(&cli->cl_unstable_count, 0);
+       atomic_long_set(&cli->cl_unstable_count, 0);
 
        init_waitqueue_head(&cli->cl_destroy_waitq);
        atomic_set(&cli->cl_destroy_in_flight, 0);
@@ -758,17 +754,9 @@ void target_client_add_cb(struct obd_device *obd, __u64 transno, void *cb_data,
 }
 EXPORT_SYMBOL(target_client_add_cb);
 
-#ifdef __KERNEL__
 static void
 check_and_start_recovery_timer(struct obd_device *obd,
                                struct ptlrpc_request *req, int new_client);
-#else
-static inline void
-check_and_start_recovery_timer(struct obd_device *obd,
-                               struct ptlrpc_request *req, int new_client)
-{
-}
-#endif
 
 int target_handle_connect(struct ptlrpc_request *req)
 {
@@ -1072,7 +1060,7 @@ dont_check_exports:
         if (rc)
                 GOTO(out, rc);
 
-#if LUSTRE_VERSION_CODE < OBD_OCD_VERSION(2, 6, 52, 0)
+#if LUSTRE_VERSION_CODE < OBD_OCD_VERSION(2, 6, 53, 0)
        /* 2.2.0 clients always swab nidtbl entries due to a bug, so server
         * will do the swabbing for if the client is using the same endianness.
         *
@@ -1080,10 +1068,10 @@ dont_check_exports:
         * OBD_CONNECT_MNE_SWAB flag around forever, just so long as we need
         * interop with unpatched 2.2 clients.  For newer clients, servers
         * will never do MNE swabbing, let the client handle that.  LU-1644 */
+       spin_lock(&export->exp_lock);
        export->exp_need_mne_swab = !ptlrpc_req_need_swab(req) &&
                        !(data->ocd_connect_flags & OBD_CONNECT_MNE_SWAB);
-#else
-#warning "LU-1644: Remove old OBD_CONNECT_MNE_SWAB fixup and exp_need_mne_swab"
+       spin_unlock(&export->exp_lock);
 #endif
 
         LASSERT(target->u.obt.obt_magic == OBT_MAGIC);
@@ -1382,7 +1370,6 @@ static void target_exp_dequeue_req_replay(struct ptlrpc_request *req)
        spin_unlock(&req->rq_export->exp_lock);
 }
 
-#ifdef __KERNEL__
 static void target_finish_recovery(struct obd_device *obd)
 {
         ENTRY;
@@ -1821,17 +1808,19 @@ repeat:
 
 static struct ptlrpc_request *target_next_replay_req(struct obd_device *obd)
 {
-        struct ptlrpc_request *req = NULL;
-        ENTRY;
+       struct ptlrpc_request *req = NULL;
+       ENTRY;
 
-        CDEBUG(D_HA, "Waiting for transno "LPD64"\n",
-               obd->obd_next_recovery_transno);
+       CDEBUG(D_HA, "Waiting for transno "LPD64"\n",
+               obd->obd_next_recovery_transno);
 
-        if (target_recovery_overseer(obd, check_for_next_transno,
-                                     exp_req_replay_healthy)) {
-                abort_req_replay_queue(obd);
-                abort_lock_replay_queue(obd);
-        }
+       CFS_FAIL_TIMEOUT(OBD_FAIL_TGT_REPLAY_DELAY2, cfs_fail_val);
+
+       if (target_recovery_overseer(obd, check_for_next_transno,
+                                    exp_req_replay_healthy)) {
+               abort_req_replay_queue(obd);
+               abort_lock_replay_queue(obd);
+       }
 
        spin_lock(&obd->obd_recovery_task_lock);
        if (!list_empty(&obd->obd_req_replay_queue)) {
@@ -2171,7 +2160,6 @@ void target_recovery_init(struct lu_target *lut, svc_handler_t handler)
 }
 EXPORT_SYMBOL(target_recovery_init);
 
-#endif /* __KERNEL__ */
 
 static int target_process_req_flags(struct obd_device *obd,
                                     struct ptlrpc_request *req)
@@ -2631,6 +2619,7 @@ int target_bulk_io(struct obd_export *exp, struct ptlrpc_bulk_desc *desc,
 {
        struct ptlrpc_request   *req = desc->bd_req;
        time_t                   start = cfs_time_current_sec();
+       time_t                   deadline;
        int                      rc = 0;
 
        ENTRY;
@@ -2640,7 +2629,7 @@ int target_bulk_io(struct obd_export *exp, struct ptlrpc_bulk_desc *desc,
                *lwi = LWI_INTR(NULL, NULL);
                rc = l_wait_event(exp->exp_obd->obd_evict_inprogress_waitq,
                                  !atomic_read(&exp->exp_obd->
-                                                  obd_evict_inprogress),
+                                                  obd_evict_inprogress),
                                  lwi);
        }
 
@@ -2666,8 +2655,13 @@ int target_bulk_io(struct obd_export *exp, struct ptlrpc_bulk_desc *desc,
                RETURN(0);
        }
 
+       /* limit actual bulk transfer to bulk_timeout seconds */
+       deadline = start + bulk_timeout;
+       if (deadline > req->rq_deadline)
+               deadline = req->rq_deadline;
+
        do {
-               long timeoutl = req->rq_deadline - cfs_time_current_sec();
+               long timeoutl = deadline - cfs_time_current_sec();
                cfs_duration_t timeout = timeoutl <= 0 ?
                                         CFS_TICK : cfs_time_seconds(timeoutl);
 
@@ -2680,14 +2674,17 @@ int target_bulk_io(struct obd_export *exp, struct ptlrpc_bulk_desc *desc,
                                  lustre_msg_get_conn_cnt(req->rq_reqmsg),
                                  lwi);
                LASSERT(rc == 0 || rc == -ETIMEDOUT);
-               /* Wait again if we changed deadline. */
+               /* Wait again if we changed rq_deadline. */
+               deadline = start + bulk_timeout;
+               if (deadline > req->rq_deadline)
+                       deadline = req->rq_deadline;
        } while ((rc == -ETIMEDOUT) &&
-                (req->rq_deadline > cfs_time_current_sec()));
+                (deadline > cfs_time_current_sec()));
 
        if (rc == -ETIMEDOUT) {
                DEBUG_REQ(D_ERROR, req, "timeout on bulk %s after %ld%+lds",
-                         bulk2type(desc), req->rq_deadline - start,
-                         cfs_time_current_sec() - req->rq_deadline);
+                         bulk2type(desc), deadline - start,
+                         cfs_time_current_sec() - deadline);
                ptlrpc_abort_bulk(desc);
        } else if (exp->exp_failed) {
                DEBUG_REQ(D_ERROR, req, "Eviction on bulk %s",