Whamcloud - gitweb
* merge b_recovery into HEAD
[fs/lustre-release.git] / lustre / ptlrpc / recover.c
index 272ffb9..acdecf8 100644 (file)
@@ -6,11 +6,11 @@
  * This code is issued under the GNU General Public License.
  * See the file COPYING in this distribution
  *
- * Copryright (C) 1996 Peter J. Braam <braam@stelias.com>
- * Copryright (C) 1999 Stelias Computing Inc. <braam@stelias.com>
- * Copryright (C) 1999 Seagate Technology Inc.
- * Copryright (C) 2001 Mountain View Data, Inc.
- * Copryright (C) 2002 Cluster File Systems, Inc.
+ * Copyright (C) 1996 Peter J. Braam <braam@stelias.com>
+ * Copyright (C) 1999 Stelias Computing Inc. <braam@stelias.com>
+ * Copyright (C) 1999 Seagate Technology Inc.
+ * Copyright (C) 2001 Mountain View Data, Inc.
+ * Copyright (C) 2002 Cluster File Systems, Inc.
  *
  */
 
@@ -32,7 +32,7 @@ int ptlrpc_reconnect_import(struct obd_import *imp, int rq_opc)
         char *tmp[] = {cli->cl_target_uuid, obd->obd_uuid };
         struct ptlrpc_connection *conn = imp->imp_connection;
         struct lustre_handle old_hdl;
-        struct ptlrpc_request *request; 
+        struct ptlrpc_request *request;
         struct obd_export *ldlmexp;
         int rc;
 
@@ -40,7 +40,6 @@ int ptlrpc_reconnect_import(struct obd_import *imp, int rq_opc)
         request->rq_level = LUSTRE_CONN_NEW;
         request->rq_replen = lustre_msg_size(0, NULL);
         /*
-
          * This address is the export that represents our client-side LDLM
          * service (for ASTs).  We should only have one on this list, so we
          * just grab the first one.
@@ -52,24 +51,55 @@ int ptlrpc_reconnect_import(struct obd_import *imp, int rq_opc)
         request->rq_reqmsg->addr = (__u64)(unsigned long)ldlmexp;
         request->rq_reqmsg->cookie = ldlmexp->exp_cookie;
         rc = ptlrpc_queue_wait(request);
-        rc = ptlrpc_check_status(request, rc);
-        if (rc) {
+        switch (rc) {
+            case EALREADY:
+            case -EALREADY:
+                /* already connected! */
+                memset(&old_hdl, 0, sizeof(old_hdl));
+                if (!memcmp(&old_hdl.addr, &request->rq_repmsg->addr,
+                            sizeof (old_hdl.addr)) &&
+                    !memcmp(&old_hdl.cookie, &request->rq_repmsg->cookie,
+                            sizeof (old_hdl.cookie))) {
+                        CERROR("%s@%s didn't like our handle %Lx/%Lx, failed\n",
+                               cli->cl_target_uuid, conn->c_remote_uuid,
+                               (__u64)(unsigned long)ldlmexp,
+                               ldlmexp->exp_cookie);
+                        GOTO(out_disc, rc = -ENOTCONN);
+                }
+
+                old_hdl.addr = request->rq_repmsg->addr;
+                old_hdl.cookie = request->rq_repmsg->cookie;
+                if (memcmp(&imp->imp_handle, &old_hdl, sizeof(old_hdl))) {
+                        CERROR("%s@%s changed handle from %Lx/%Lx to %Lx/%Lx; "
+                               "copying, but this may foreshadow disaster\n",
+                               cli->cl_target_uuid, conn->c_remote_uuid,
+                               old_hdl.addr, old_hdl.cookie,
+                               imp->imp_handle.addr, imp->imp_handle.cookie);
+                        imp->imp_handle.addr = request->rq_repmsg->addr;
+                        imp->imp_handle.cookie = request->rq_repmsg->cookie;
+                        GOTO(out_disc, rc = EALREADY);
+                }
+
+                CERROR("reconnected to %s@%s after partition\n",
+                       cli->cl_target_uuid, conn->c_remote_uuid);
+                GOTO(out_disc, rc = EALREADY);
+            case 0:
+                old_hdl = imp->imp_handle;
+                imp->imp_handle.addr = request->rq_repmsg->addr;
+                imp->imp_handle.cookie = request->rq_repmsg->cookie;
+                CERROR("now connected to %s@%s (%Lx/%Lx, was %Lx/%Lx)!\n",
+                       cli->cl_target_uuid, conn->c_remote_uuid,
+                       imp->imp_handle.addr, imp->imp_handle.cookie,
+                       old_hdl.addr, old_hdl.cookie);
+                GOTO(out_disc, rc = 0);
+            default:
                 CERROR("cannot connect to %s@%s: rc = %d\n",
                        cli->cl_target_uuid, conn->c_remote_uuid, rc);
-                ptlrpc_free_req(request);
-                GOTO(out_disc, rc = -ENOTCONN);
+                GOTO(out_disc, rc = -ENOTCONN); /* XXX preserve rc? */
         }
-        
-        old_hdl = imp->imp_handle;
-        imp->imp_handle.addr = request->rq_repmsg->addr;
-        imp->imp_handle.cookie = request->rq_repmsg->cookie;
-        CERROR("reconnected to %s@%s (%Lx/%Lx, was %Lx/%Lx)!\n",
-               cli->cl_target_uuid, conn->c_remote_uuid,
-               imp->imp_handle.addr, imp->imp_handle.cookie,
-               old_hdl.addr, old_hdl.cookie);
-        ptlrpc_req_finished(request);
 
  out_disc:
+        ptlrpc_req_finished(request);
         return rc;
 }
 
@@ -80,8 +110,6 @@ int ptlrpc_run_recovery_upcall(struct ptlrpc_connection *conn)
         int rc;
 
         ENTRY;
-        conn->c_level = LUSTRE_CONN_RECOVD;
-
         argv[0] = obd_recovery_upcall;
         argv[1] = conn->c_remote_uuid;
         argv[2] = NULL;
@@ -108,120 +136,120 @@ int ptlrpc_run_recovery_upcall(struct ptlrpc_connection *conn)
         RETURN(0);
 }
 
-#define REPLAY_COMMITTED     0 /* Fully processed (commit + reply) */
-#define REPLAY_REPLAY        1 /* Forced-replay (e.g. open) */
-#define REPLAY_RESEND        2 /* Resend required. */
-#define REPLAY_RESEND_IGNORE 3 /* Resend, ignore the reply (already saw it) */
-#define REPLAY_RESTART       4 /* Have to restart the call, sorry! */
-
-static int replay_state(struct ptlrpc_request *req, __u64 last_xid)
+int ptlrpc_replay(struct obd_import *imp, int send_last_flag)
 {
-        /* This request must always be replayed. */
-        if (req->rq_flags & PTL_RPC_FL_REPLAY)
-                return REPLAY_REPLAY;
+        int rc = 0;
+        struct list_head *tmp, *pos;
+        struct ptlrpc_request *req;
+        __u64 committed = imp->imp_peer_committed_transno;
+        ENTRY;
 
-        /* Uncommitted request */
-        if (req->rq_xid > last_xid) {
-                if (req->rq_flags & PTL_RPC_FL_REPLIED) {
-                        /* Saw reply, so resend and ignore new reply. */
-                        return REPLAY_RESEND_IGNORE;
-                }
+        /* It might have committed some after we last spoke, so make sure we
+         * get rid of them now.
+         */
+        spin_lock(&imp->imp_lock);
 
-                /* Didn't see reply either, so resend. */
-                return REPLAY_RESEND;
+        ptlrpc_free_committed(imp);
+
+        CDEBUG(D_HA, "import %p from %s has committed "LPD64"\n",
+               imp, imp->imp_obd->u.cli.cl_target_uuid, committed);
+
+        list_for_each(tmp, &imp->imp_replay_list) {
+                req = list_entry(tmp, struct ptlrpc_request, rq_list);
+                DEBUG_REQ(D_HA, req, "RETAINED: ");
         }
 
-        /* This request has been committed and we saw the reply.  Goodbye! */
-        if (req->rq_flags & PTL_RPC_FL_REPLIED)
-                return REPLAY_COMMITTED;
+        list_for_each_safe(tmp, pos, &imp->imp_replay_list) {
+                req = list_entry(tmp, struct ptlrpc_request, rq_list);
+
+                if (req->rq_transno == imp->imp_max_transno &&
+                    send_last_flag) {
+                        req->rq_reqmsg->flags |= MSG_LAST_REPLAY;
+                        DEBUG_REQ(D_HA, req, "LAST_REPLAY:");
+                } else {
+                        DEBUG_REQ(D_HA, req, "REPLAY:");
+                }
+
+                rc = ptlrpc_replay_req(req);
+                req->rq_reqmsg->flags &= ~MSG_LAST_REPLAY;
 
-        /* Request committed, but we didn't see the reply: have to restart. */
-        return REPLAY_RESTART;
+                if (rc) {
+                        CERROR("recovery replay error %d for req %Ld\n",
+                               rc, req->rq_xid);
+                        GOTO(out, rc);
+                }
+        }
+
+ out:
+        spin_unlock(&imp->imp_lock);
+        return rc;
 }
 
-static char *replay_state2str(int state) {
-        static char *state_strings[] = {
-                "COMMITTED", "REPLAY", "RESEND", "RESEND_IGNORE", "RESTART"
-        };
-        static char *unknown_state = "UNKNOWN";
+#define NO_RESEND     0 /* No action required. */
+#define RESEND        1 /* Resend required. */
+#define RESEND_IGNORE 2 /* Resend, ignore the reply (already saw it). */
+#define RESTART       3 /* Have to restart the call, sorry! */
 
-        if (state < 0 || 
-            state > (sizeof(state_strings) / sizeof(state_strings[0]))) {
-                return unknown_state;
+static int resend_type(struct ptlrpc_request *req, __u64 committed)
+{
+        if (req->rq_transno < committed) {
+                if (req->rq_flags & PTL_RPC_FL_REPLIED) {
+                        /* Saw the reply and it was committed, no biggie. */
+                        DEBUG_REQ(D_HA, req, "NO_RESEND");
+                        return NO_RESEND;
+                }
+                /* Request committed, but no reply: have to restart. */
+                return RESTART;
+        }
+
+        if (req->rq_flags & PTL_RPC_FL_REPLIED) {
+                /* Saw reply, so resend and ignore new reply. */
+                return RESEND_IGNORE;
         }
 
-        return state_strings[state];
+        /* Didn't see reply either, so resend. */
+        return RESEND;
+
 }
 
-int ptlrpc_replay(struct ptlrpc_connection *conn)
+int ptlrpc_resend(struct obd_import *imp)
 {
         int rc = 0;
         struct list_head *tmp, *pos;
         struct ptlrpc_request *req;
-        ENTRY;
-
-        spin_lock(&conn->c_lock);
+        __u64 committed = imp->imp_peer_committed_transno;
 
-        CDEBUG(D_HA, "connection %p to %s has last_xid "LPD64"\n",
-               conn, conn->c_remote_uuid, conn->c_last_xid);
+        ENTRY;
 
-        list_for_each(tmp, &conn->c_sending_head) {
-                int state;
+        spin_lock(&imp->imp_lock);
+        list_for_each(tmp, &imp->imp_sending_list) {
                 req = list_entry(tmp, struct ptlrpc_request, rq_list);
-                state = replay_state(req, conn->c_last_xid);
-                DEBUG_REQ(D_HA, req, "SENDING: %s: ", replay_state2str(state));
+                DEBUG_REQ(D_HA, req, "SENDING: ");
         }
 
-        list_for_each(tmp, &conn->c_delayed_head) {
-                int state;
+        list_for_each_safe(tmp, pos, &imp->imp_sending_list) {
                 req = list_entry(tmp, struct ptlrpc_request, rq_list);
-                state = replay_state(req, conn->c_last_xid);
-                DEBUG_REQ(D_HA, req, "DELAYED: ");
-        }
 
-        list_for_each_safe(tmp, pos, &conn->c_sending_head) { 
-                req = list_entry(tmp, struct ptlrpc_request, rq_list);
-                
-                switch (replay_state(req, conn->c_last_xid)) {
-                    case REPLAY_REPLAY:
-                        DEBUG_REQ(D_HA, req, "REPLAY:");
-                        rc = ptlrpc_replay_req(req);
-#if 0
-#error We should not hold a spinlock over such a lengthy operation.
-#error If necessary, drop spinlock, do operation, re-get spinlock, restart loop.
-#error If we need to avoid re-processint items, then delete them from the list
-#error as they are replayed and re-add at the tail of this list, so the next
-#error item to process will always be at the head of the list.
-#endif
-                        if (rc) {
-                                CERROR("recovery replay error %d for req %Ld\n",
-                                       rc, req->rq_xid);
-                                GOTO(out, rc);
-                        }
+                switch(resend_type(req, committed)) {
+                    case NO_RESEND:
                         break;
 
-
-                    case REPLAY_COMMITTED:
-                        DEBUG_REQ(D_HA, req, "COMMITTED:");
-                        /* XXX commit now? */
+                    case RESTART:
+                        DEBUG_REQ(D_HA, req, "RESTART:");
+                        ptlrpc_restart_req(req);
                         break;
 
-                    case REPLAY_RESEND_IGNORE:
+                    case RESEND_IGNORE:
                         DEBUG_REQ(D_HA, req, "RESEND_IGNORE:");
-                        rc = ptlrpc_replay_req(req); 
+                        rc = ptlrpc_replay_req(req);
                         if (rc) {
-                                CERROR("request resend error %d for req %Ld\n",
-                                       rc, req->rq_xid); 
-                                GOTO(out, rc);
+                                DEBUG_REQ(D_ERROR, req, "error %d resending:",
+                                          rc);
+                                ptlrpc_restart_req(req); /* might as well */
                         }
                         break;
 
-                    case REPLAY_RESTART:
-                        DEBUG_REQ(D_HA, req, "RESTART:");
-                        ptlrpc_restart_req(req);
-                        break;
-
-                    case REPLAY_RESEND:
+                    case RESEND:
                         DEBUG_REQ(D_HA, req, "RESEND:");
                         ptlrpc_resend_req(req);
                         break;
@@ -229,23 +257,21 @@ int ptlrpc_replay(struct ptlrpc_connection *conn)
                     default:
                         LBUG();
                 }
-
         }
 
-        conn->c_level = LUSTRE_CONN_FULL;
-        recovd_conn_fixed(conn);
+        RETURN(rc);
+}
 
-        CERROR("recovery complete on conn %p(%s), waking delayed reqs\n",
-               conn, conn->c_remote_uuid);
-        /* Finally, continue processing requests that blocked for recovery. */
-        list_for_each_safe(tmp, pos, &conn->c_delayed_head) { 
+void ptlrpc_wake_delayed(struct obd_import *imp)
+{
+        struct list_head *tmp, *pos;
+        struct ptlrpc_request *req;
+
+        spin_lock(&imp->imp_lock);
+        list_for_each_safe(tmp, pos, &imp->imp_delayed_list) {
                 req = list_entry(tmp, struct ptlrpc_request, rq_list);
-                DEBUG_REQ(D_HA, req, "WAKING: ");
-                ptlrpc_continue_req(req);
+                DEBUG_REQ(D_HA, req, "waking:");
+                wake_up(&req->rq_wait_for_rep);
         }
-
-        EXIT;
- out:
-        spin_unlock(&conn->c_lock);
-        return rc;
+        spin_unlock(&imp->imp_lock);
 }