ORNL-28: Solve reconnecting race between IR and SR

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

diff --git a/lustre/ldlm/ldlm_lib.c b/lustre/ldlm/ldlm_lib.c
index f55e5be..ab65cb0 100644 (file)
--- a/lustre/ldlm/ldlm_lib.c
+++ b/lustre/ldlm/ldlm_lib.c
@@ -193,6 +193,31 @@ out:
         RETURN(rc);
 }
 
+/**
+ * Find conn uuid by peer nid. @peer is a server nid. This function is used
+ * to find a conn uuid of @imp which can reach @peer.
+ */
+int client_import_find_conn(struct obd_import *imp, lnet_nid_t peer,
+                            struct obd_uuid *uuid)
+{
+        struct obd_import_conn *conn;
+        int rc = -ENOENT;
+        ENTRY;
+
+        cfs_spin_lock(&imp->imp_lock);
+        cfs_list_for_each_entry(conn, &imp->imp_conn_list, oic_item) {
+                /* check if conn uuid does have this peer nid */
+                if (class_check_uuid(&conn->oic_uuid, peer)) {
+                        *uuid = conn->oic_uuid;
+                        rc = 0;
+                        break;
+                }
+        }
+        cfs_spin_unlock(&imp->imp_lock);
+        RETURN(rc);
+}
+EXPORT_SYMBOL(client_import_find_conn);
+
 void client_destroy_import(struct obd_import *imp)
 {
         /* drop security policy instance after all rpc finished/aborted
@@ -446,7 +471,7 @@ int client_connect_import(const struct lu_env *env,
                 imp->imp_connect_flags_orig = data->ocd_connect_flags;
         }
 
-        rc = ptlrpc_connect_import(imp, NULL);
+        rc = ptlrpc_connect_import(imp);
         if (rc != 0) {
                 LASSERT (imp->imp_state == LUSTRE_IMP_DISCON);
                 GOTO(out_ldlm, rc);
@@ -666,6 +691,7 @@ int target_handle_connect(struct ptlrpc_request *req)
         int rc = 0;
         int mds_conn = 0;
         struct obd_connect_data *data, *tmpdata;
+        int size, tmpsize;
         lnet_nid_t *client_nid = NULL;
         ENTRY;
 
@@ -684,7 +710,7 @@ int target_handle_connect(struct ptlrpc_request *req)
 
         if (!target || target->obd_stopping || !target->obd_set_up) {
                 LCONSOLE_ERROR_MSG(0x137, "UUID '%s' is not available "
-                                   " for connect (%s)\n", str,
+                                   "for connect (%s)\n", str,
                                    !target ? "no target" :
                                    (target->obd_stopping ? "stopping" :
                                    "not set up"));
@@ -733,6 +759,8 @@ int target_handle_connect(struct ptlrpc_request *req)
 
         conn = *tmp;
 
+        size = req_capsule_get_size(&req->rq_pill, &RMF_CONNECT_DATA,
+                                    RCL_CLIENT);
         data = req_capsule_client_get(&req->rq_pill, &RMF_CONNECT_DATA);
         if (!data)
                 GOTO(out, rc = -EPROTO);
@@ -923,13 +951,21 @@ dont_check_exports:
         }
         if (rc)
                 GOTO(out, rc);
+
+        LASSERT(target->u.obt.obt_magic == OBT_MAGIC);
+        data->ocd_instance = target->u.obt.obt_instance;
+
         /* Return only the parts of obd_connect_data that we understand, so the
          * client knows that we don't understand the rest. */
         if (data) {
-                 tmpdata = req_capsule_server_get(&req->rq_pill,
-                                                  &RMF_CONNECT_DATA);
-                  //data->ocd_connect_flags &= OBD_CONNECT_SUPPORTED;
-                 *tmpdata = *data;
+                tmpsize = req_capsule_get_size(&req->rq_pill, &RMF_CONNECT_DATA,
+                                               RCL_SERVER);
+                tmpdata = req_capsule_server_get(&req->rq_pill,
+                                                 &RMF_CONNECT_DATA);
+                /* Don't use struct assignment here, because the client reply
+                 * buffer may be smaller/larger than the local struct
+                 * obd_connect_data. */
+                memcpy(tmpdata, data, min(tmpsize, size));
         }
 
         /* If all else goes well, this is our RPC return code. */
@@ -1007,7 +1043,8 @@ dont_check_exports:
                         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_next_recovery_transno &&
+                     data->ocd_transno > target->obd_last_committed)
                         target->obd_next_recovery_transno = data->ocd_transno;
                 target->obd_connected_clients++;
                 cfs_atomic_inc(&target->obd_req_replay_clients);
@@ -1184,9 +1221,16 @@ static void target_exp_dequeue_req_replay(struct ptlrpc_request *req)
 #ifdef __KERNEL__
 static void target_finish_recovery(struct obd_device *obd)
 {
+        time_t elapsed_time = max_t(time_t, 1, cfs_time_current_sec() -
+                                    obd->obd_recovery_start);
         ENTRY;
-        LCONSOLE_INFO("%s: sending delayed replies to recovered clients\n",
-                      obd->obd_name);
+
+        LCONSOLE_INFO("%s: Recovery over after %d:%.02d, of %d clients "
+                      "%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,
+                      obd->obd_stale_clients == 1 ? "was" : "were");
 
         ldlm_reprocess_all_ns(obd->obd_namespace);
         cfs_spin_lock(&obd->obd_recovery_task_lock);
@@ -1729,7 +1773,7 @@ static int target_recovery_thread(void *arg)
         env.le_ctx.lc_thread = thread;
         thread->t_data = NULL;
 
-        CERROR("%s: started recovery thread pid %d\n", obd->obd_name,
+        CDEBUG(D_HA, "%s: started recovery thread pid %d\n", obd->obd_name,
                cfs_curproc_pid());
         trd->trd_processing_task = cfs_curproc_pid();