ORNL-28: remove lock contention in reconnecting

target_handle_connect() used to grab obd::obd_recovery_task_lock to update
obd_next_recovery_transno and obd_connected_clients. In this patch, I revised
this piece of code by:
- modify obd_connected_clients to cfs_atomic_t
- grab obd_recovery_task_lock only if ocd_transno is less than
  obd_next_recovery_transno

In this way, target_handle_connect() don't grab any global lock.

Change-Id: I971897de82566e239fe750cf95f2c3f1646325bc
Signed-off-by: Jinshan Xiong <jay@whamcloud.com>
Reviewed-on: http://review.whamcloud.com/1293
Tested-by: Hudson
Reviewed-by: Andreas Dilger <adilger@whamcloud.com>
Reviewed-by: Mikhail Pershin <tappro@whamcloud.com>
Reviewed-by: Oleg Drokin <green@whamcloud.com>
Tested-by: Maloo <whamcloud.maloo@gmail.com>

diff --git a/lustre/include/obd.h b/lustre/include/obd.h
index 08d483f..666c949 100644 (file)
--- a/lustre/include/obd.h
+++ b/lustre/include/obd.h
@@ -1056,7 +1056,7 @@ struct obd_device {
         time_t                  obd_eviction_timer; /* for ping evictor */
 
         int                              obd_max_recoverable_clients;
-        int                              obd_connected_clients;
+        cfs_atomic_t                     obd_connected_clients;
         int                              obd_stale_clients;
         int                              obd_delayed_clients;
         /* this lock protects all recovery list_heads, timer and

diff --git a/lustre/ldlm/ldlm_lib.c b/lustre/ldlm/ldlm_lib.c
index 60a7e3a..996da3e 100644 (file)
--- a/lustre/ldlm/ldlm_lib.c
+++ b/lustre/ldlm/ldlm_lib.c
@@ -1038,29 +1038,38 @@ dont_check_exports:
                              &export->exp_nid_hash);
         }
 
-        cfs_spin_lock(&target->obd_recovery_task_lock);
         if (target->obd_recovering && !export->exp_in_recovery) {
+                int has_transno;
+                __u64 transno = data->ocd_transno;
+
                 cfs_spin_lock(&export->exp_lock);
                 export->exp_in_recovery = 1;
                 export->exp_req_replay_needed = 1;
                 export->exp_lock_replay_needed = 1;
                 cfs_spin_unlock(&export->exp_lock);
-                if ((lustre_msg_get_op_flags(req->rq_reqmsg) & MSG_CONNECT_TRANSNO)
-                     && (data->ocd_transno == 0))
+
+                has_transno = !!(lustre_msg_get_op_flags(req->rq_reqmsg) &
+                                 MSG_CONNECT_TRANSNO);
+                if (has_transno && transno == 0)
                         CWARN("Connect with zero transno!\n");
 
-                if ((lustre_msg_get_op_flags(req->rq_reqmsg) & MSG_CONNECT_TRANSNO)
-                     && data->ocd_transno < target->obd_next_recovery_transno &&
-                     data->ocd_transno > target->obd_last_committed)
-                        target->obd_next_recovery_transno = data->ocd_transno;
-                target->obd_connected_clients++;
+                if (has_transno && transno > 0 &&
+                    transno < target->obd_next_recovery_transno &&
+                    transno > target->obd_last_committed) {
+                        /* another way is to use cmpxchg() so it will be
+                         * lock free */
+                        cfs_spin_lock(&target->obd_recovery_task_lock);
+                        if (transno < target->obd_next_recovery_transno)
+                                target->obd_next_recovery_transno = transno;
+                        cfs_spin_unlock(&target->obd_recovery_task_lock);
+                }
+
                 cfs_atomic_inc(&target->obd_req_replay_clients);
                 cfs_atomic_inc(&target->obd_lock_replay_clients);
-                if (target->obd_connected_clients ==
+                if (cfs_atomic_inc_return(&target->obd_connected_clients) ==
                     target->obd_max_recoverable_clients)
                         cfs_waitq_signal(&target->obd_next_transno_waitq);
         }
-        cfs_spin_unlock(&target->obd_recovery_task_lock);
 
         /* Tell the client we're in recovery, when client is involved in it. */
         if (target->obd_recovering)
@@ -1236,7 +1245,8 @@ static void target_finish_recovery(struct obd_device *obd)
                       "%d recovered and %d %s evicted.\n", obd->obd_name,
                       (int)elapsed_time / 60, (int)elapsed_time % 60,
                       obd->obd_max_recoverable_clients,
-                      obd->obd_connected_clients, obd->obd_stale_clients,
+                      cfs_atomic_read(&obd->obd_connected_clients),
+                      obd->obd_stale_clients,
                       obd->obd_stale_clients == 1 ? "was" : "were");
 
         ldlm_reprocess_all_ns(obd->obd_namespace);
@@ -1485,12 +1495,13 @@ static inline int exp_finished(struct obd_export *exp)
 /** Checking routines for recovery */
 static int check_for_clients(struct obd_device *obd)
 {
+        unsigned int clnts = cfs_atomic_read(&obd->obd_connected_clients);
+
         if (obd->obd_abort_recovery || obd->obd_recovery_expired)
                 return 1;
-        LASSERT(obd->obd_connected_clients <= obd->obd_max_recoverable_clients);
+        LASSERT(clnts <= obd->obd_max_recoverable_clients);
         if (obd->obd_no_conn == 0 &&
-            obd->obd_connected_clients + obd->obd_stale_clients ==
-            obd->obd_max_recoverable_clients)
+            clnts + obd->obd_stale_clients == obd->obd_max_recoverable_clients)
                 return 1;
         return 0;
 }
@@ -1511,7 +1522,7 @@ static int check_for_next_transno(struct obd_device *obd)
                 req_transno = 0;
         }
 
-        connected = obd->obd_connected_clients;
+        connected = cfs_atomic_read(&obd->obd_connected_clients);
         completed = connected - cfs_atomic_read(&obd->obd_req_replay_clients);
         queue_len = obd->obd_requests_queued_for_recovery;
         next_transno = obd->obd_next_recovery_transno;
@@ -1923,7 +1934,7 @@ static void target_recovery_expired(unsigned long castmeharder)
                " after %lds (%d clients connected)\n",
                obd->obd_name, cfs_atomic_read(&obd->obd_lock_replay_clients),
                cfs_time_current_sec()- obd->obd_recovery_start,
-               obd->obd_connected_clients);
+               cfs_atomic_read(&obd->obd_connected_clients));
 
         obd->obd_recovery_expired = 1;
         cfs_waitq_signal(&obd->obd_next_transno_waitq);

diff --git a/lustre/obdclass/genops.c b/lustre/obdclass/genops.c
index 177c0d0..6bd0869 100644 (file)
--- a/lustre/obdclass/genops.c
+++ b/lustre/obdclass/genops.c
@@ -1110,8 +1110,8 @@ void class_export_recovery_cleanup(struct obd_export *exp)
                 cfs_spin_lock(&exp->exp_lock);
                 exp->exp_in_recovery = 0;
                 cfs_spin_unlock(&exp->exp_lock);
-                LASSERT(obd->obd_connected_clients);
-                obd->obd_connected_clients--;
+                LASSERT_ATOMIC_POS(&obd->obd_connected_clients);
+                cfs_atomic_dec(&obd->obd_connected_clients);
         }
         cfs_spin_unlock(&obd->obd_recovery_task_lock);
         /** Cleanup req replay fields */

diff --git a/lustre/obdclass/lprocfs_status.c b/lustre/obdclass/lprocfs_status.c
index 7bdd31b..1610794 100644 (file)
--- a/lustre/obdclass/lprocfs_status.c
+++ b/lustre/obdclass/lprocfs_status.c
@@ -2286,7 +2286,7 @@ int lprocfs_obd_rd_recovery_status(char *page, char **start, off_t off,
                            obd->obd_recovery_end - cfs_time_current_sec()) <= 0)
                 goto out;
         if (lprocfs_obd_snprintf(&page, size, &len,"connected_clients: %d/%d\n",
-                                 obd->obd_connected_clients,
+                                 cfs_atomic_read(&obd->obd_connected_clients),
                                  obd->obd_max_recoverable_clients) <= 0)
                 goto out;
         /* Number of clients that have completed recovery */
@@ -2299,7 +2299,7 @@ int lprocfs_obd_rd_recovery_status(char *page, char **start, off_t off,
                 <=0)
                 goto out;
         if (lprocfs_obd_snprintf(&page, size, &len,"completed_clients: %d\n",
-                                 obd->obd_connected_clients -
+                                 cfs_atomic_read(&obd->obd_connected_clients) -
                                  cfs_atomic_read(&obd->obd_lock_replay_clients))
                 <=0)
                 goto out;