- Parallel recovery implementation, to match documented design.

- Handle failed client recovery by simply restarting recovery.  Not perfect, but
  surprisingly useful for testing.

diff --git a/lustre/include/linux/lustre_ha.h b/lustre/include/linux/lustre_ha.h
index 3866179..9f90624 100644 (file)
--- a/lustre/include/linux/lustre_ha.h
+++ b/lustre/include/linux/lustre_ha.h
@@ -11,17 +11,18 @@ struct recovd_data;
 struct recovd_obd;
 struct ptlrpc_connection;
 
-/* recovd_phase values */
+/* rd_phase/rd_next_phase values */
 #define RECOVD_IDLE              0
 #define RECOVD_PREPARING         1
 #define RECOVD_PREPARED          2
 #define RECOVD_RECOVERING        3
 #define RECOVD_RECOVERED         4
+#define RECOVD_FAILED            5
 
-/* recovd_flags bits */
-#define RECOVD_STOPPING          1  /* how cleanup tells recovd to quit */
-#define RECOVD_STOPPED           2  /* after recovd has stopped */
-#define RECOVD_FAILED            4  /* the current recovery has failed */
+/* recovd_state values */
+#define RECOVD_READY             1
+#define RECOVD_STOPPING          2  /* how cleanup tells recovd to quit */
+#define RECOVD_STOPPED           4  /* after recovd has stopped */
 
 #define PTLRPC_RECOVD_PHASE_PREPARE  1
 #define PTLRPC_RECOVD_PHASE_RECOVER  2
@@ -33,6 +34,9 @@ struct recovd_data {
         struct list_head     rd_managed_chain;
         ptlrpc_recovery_cb_t rd_recover;
         struct recovd_obd   *rd_recovd;
+        __u32                rd_phase;
+        __u32                rd_next_phase;
+        __u32                rd_flags;
 };
 
 void recovd_conn_fail(struct ptlrpc_connection *conn);

diff --git a/lustre/include/linux/obd.h b/lustre/include/linux/obd.h
index 0877378..7547394 100644 (file)
--- a/lustre/include/linux/obd.h
+++ b/lustre/include/linux/obd.h
@@ -143,17 +143,15 @@ struct echo_obd {
 };
 
 struct recovd_obd {
-        __u32                 recovd_phase;
-        __u32                 recovd_next_phase;
-        __u32                 recovd_flags;
-        struct recovd_data   *recovd_current_rd;
         spinlock_t            recovd_lock;
         struct list_head      recovd_managed_items; /* items managed  */
-        struct list_head      recovd_troubled_items; /* items in trouble */
+        struct list_head      recovd_troubled_items; /* items in recovery */
+
         wait_queue_head_t     recovd_recovery_waitq;
         wait_queue_head_t     recovd_ctl_waitq;
         wait_queue_head_t     recovd_waitq;
         struct task_struct   *recovd_thread;
+        __u32                 recovd_state;
 };
 
 struct trace_obd {

diff --git a/lustre/lib/l_net.c b/lustre/lib/l_net.c
index 6f9af2b..c4965db 100644 (file)
--- a/lustre/lib/l_net.c
+++ b/lustre/lib/l_net.c
@@ -166,6 +166,7 @@ int client_obd_connect(struct lustre_handle *conn, struct obd_device *obd,
         cli->cl_import.imp_handle.addr = request->rq_repmsg->addr;
         cli->cl_import.imp_handle.cookie = request->rq_repmsg->cookie;
 
+
         recovd_conn_manage(c, ptlrpc_recovd, ll_recover);
 
         EXIT;

diff --git a/lustre/ptlrpc/client.c b/lustre/ptlrpc/client.c
index 2c215e3..79e4174 100644 (file)
--- a/lustre/ptlrpc/client.c
+++ b/lustre/ptlrpc/client.c
@@ -480,8 +480,9 @@ int ptlrpc_queue_wait(struct ptlrpc_request *req)
 
         /* XXX probably both an import and connection level are needed */
         if (req->rq_level > conn->c_level) { 
-                CERROR("process %d waiting for recovery (%d > %d)\n", 
-                       current->pid, req->rq_level, conn->c_level);
+                CERROR("pid %d waiting for recovery (%d > %d) on conn %p(%s)\n", 
+                       current->pid, req->rq_level, conn->c_level, conn,
+                       conn->c_remote_uuid);
 
                 spin_lock(&conn->c_lock);
                 list_del(&req->rq_list);
@@ -515,11 +516,11 @@ int ptlrpc_queue_wait(struct ptlrpc_request *req)
                 /* the sleep below will time out, triggering recovery */
         }
 
-        CDEBUG(D_OTHER, "-- sleeping\n");
+        CDEBUG(D_OTHER, "-- sleeping on xid "LPD64"\n", req->rq_xid);
         lwi = LWI_TIMEOUT_INTR(req->rq_timeout * HZ, expired_request,
                                interrupted_request,req);
         l_wait_event(req->rq_wait_for_rep, ptlrpc_check_reply(req), &lwi);
-        CDEBUG(D_OTHER, "-- done\n");
+        CDEBUG(D_OTHER, "-- done sleeping on xid "LPD64"\n", req->rq_xid);
 
         /* Don't resend if we were interrupted. */
         if ((req->rq_flags & (PTL_RPC_FL_RESEND | PTL_RPC_FL_INTR)) ==

diff --git a/lustre/ptlrpc/recovd.c b/lustre/ptlrpc/recovd.c
index 3f10733..7561ba0 100644 (file)
--- a/lustre/ptlrpc/recovd.c
+++ b/lustre/ptlrpc/recovd.c
@@ -47,6 +47,7 @@ void recovd_conn_fail(struct ptlrpc_connection *conn)
                 return;
         }
 
+        CERROR("connection %p to %s failed\n", conn, conn->c_remote_uuid);
         spin_lock(&recovd->recovd_lock);
         list_del(&rd->rd_managed_chain);
         list_add_tail(&rd->rd_managed_chain, &recovd->recovd_troubled_items);
@@ -73,116 +74,126 @@ void recovd_conn_fixed(struct ptlrpc_connection *conn)
 static int recovd_check_event(struct recovd_obd *recovd)
 {
         int rc = 0;
+        struct list_head *tmp;
+
         ENTRY;
 
         spin_lock(&recovd->recovd_lock);
 
-        if (recovd->recovd_phase == RECOVD_IDLE &&
-            !list_empty(&recovd->recovd_troubled_items)) {
+        if (recovd->recovd_state == RECOVD_STOPPING)
                 GOTO(out, rc = 1);
-        }
 
-        if (recovd->recovd_flags & RECOVD_STOPPING)
-                GOTO(out, rc = 1);
+        list_for_each(tmp, &recovd->recovd_troubled_items) {
 
-        if (recovd->recovd_flags & RECOVD_FAILED) {
-                LASSERT(recovd->recovd_phase != RECOVD_IDLE && 
-                        recovd->recovd_current_rd);
-                GOTO(out, rc = 1);
-        }
+                struct recovd_data *rd = list_entry(tmp, struct recovd_data,
+                                                    rd_managed_chain);
 
-        if (recovd->recovd_phase == recovd->recovd_next_phase)
-                GOTO(out, rc = 1);
+                if (rd->rd_phase == rd->rd_next_phase ||
+                    rd->rd_phase == RECOVD_FAILED)
+                        GOTO(out, rc = 1);
+        }
 
  out:
         spin_unlock(&recovd->recovd_lock);
         RETURN(rc);
 }
 
-static int recovd_handle_event(struct recovd_obd *recovd)
+static void dump_connection_list(struct list_head *head)
 {
-        struct recovd_data *rd;
-        int rc;
-        ENTRY;
-
-        if (recovd->recovd_flags & RECOVD_FAILED) {
-
-                LASSERT(recovd->recovd_phase != RECOVD_IDLE && 
-                        recovd->recovd_current_rd);
+        struct list_head *tmp;
 
-                rd = recovd->recovd_current_rd;
-        cb_failed:
-                CERROR("recovery FAILED for rd %p (conn %p), recovering\n",
-                       rd, class_rd2conn(rd));
+        list_for_each(tmp, head) {
+                struct ptlrpc_connection *conn =
+                        list_entry(tmp, struct ptlrpc_connection,
+                                   c_recovd_data.rd_managed_chain);
+                CDEBUG(D_NET, "   %p = %s\n", conn, conn->c_remote_uuid);
+        }
+}
 
-                list_add(&rd->rd_managed_chain, &recovd->recovd_managed_items);
-                spin_unlock(&recovd->recovd_lock);
-                rd->rd_recover(rd, PTLRPC_RECOVD_PHASE_FAILURE);
-                spin_lock(&recovd->recovd_lock);
-                recovd->recovd_phase = RECOVD_IDLE;
-                recovd->recovd_next_phase = RECOVD_PREPARING;
-                
-                recovd->recovd_flags &= ~RECOVD_FAILED;
+static int recovd_handle_event(struct recovd_obd *recovd)
+{
+        struct list_head *tmp, *n;
+        int rc = 0;
+        ENTRY;
 
-                RETURN(1);
-        }
+        spin_lock(&recovd->recovd_lock);
 
-        switch (recovd->recovd_phase) {
-            case RECOVD_IDLE:
-                if (recovd->recovd_current_rd ||
-                    list_empty(&recovd->recovd_troubled_items))
+        CDEBUG(D_NET, "managed: \n");
+        dump_connection_list(&recovd->recovd_managed_items);
+        CDEBUG(D_NET, "troubled: \n");
+        dump_connection_list(&recovd->recovd_troubled_items);
+
+        /*
+         * We use _safe here because one of the callbacks, expecially
+         * FAILURE or PREPARED, could move list items around.
+         */
+        list_for_each_safe(tmp, n, &recovd->recovd_troubled_items) {
+                struct recovd_data *rd = list_entry(tmp, struct recovd_data,
+                                                    rd_managed_chain);
+
+                if (rd->rd_phase != RECOVD_FAILED &&
+                    rd->rd_phase != rd->rd_next_phase)
+                        continue;
+
+                switch (rd->rd_phase) {
+                    case RECOVD_FAILED:
+                cb_failed: /* must always reach here with recovd_lock held! */
+                        CERROR("recovery FAILED for rd %p (conn %p): %d\n",
+                               rd, class_rd2conn(rd), rc);
+                        
+                        spin_unlock(&recovd->recovd_lock);
+                        (void)rd->rd_recover(rd, PTLRPC_RECOVD_PHASE_FAILURE);
+                        spin_lock(&recovd->recovd_lock);
+                        break;
+                        
+                    case RECOVD_IDLE:
+                        if (!rd->rd_recover) {
+                                CERROR("no rd_recover for rd %p (conn %p)\n",
+                                       rd, class_rd2conn(rd));
+                                rc = -EINVAL;
+                                break;
+                        }
+                        CERROR("starting recovery for rd %p (conn %p)\n",
+                               rd, class_rd2conn(rd));
+                        rd->rd_phase = RECOVD_PREPARING;
+                        
+                        spin_unlock(&recovd->recovd_lock);
+                        rc = rd->rd_recover(rd, PTLRPC_RECOVD_PHASE_PREPARE);
+                        spin_lock(&recovd->recovd_lock);
+                        if (rc)
+                                goto cb_failed;
+                        
+                        rd->rd_next_phase = RECOVD_PREPARED;
+                        break;
+                        
+                    case RECOVD_PREPARED:
+                        rd->rd_phase = RECOVD_RECOVERING;
+                        
+                        CERROR("recovery prepared for rd %p (conn %p)\n",
+                               rd, class_rd2conn(rd));
+                        
+                        spin_unlock(&recovd->recovd_lock);
+                        rc = rd->rd_recover(rd, PTLRPC_RECOVD_PHASE_RECOVER);
+                        spin_lock(&recovd->recovd_lock);
+                        if (rc)
+                                goto cb_failed;
+                        
+                        rd->rd_next_phase = RECOVD_RECOVERED;
+                        break;
+                        
+                    case RECOVD_RECOVERED:
+                        rd->rd_phase = RECOVD_IDLE;
+                        rd->rd_next_phase = RECOVD_PREPARING;
+                        
+                        CERROR("recovery complete for rd %p (conn %p)\n",
+                               rd, class_rd2conn(rd));
+                        break;
+                        
+                    default:
                         break;
-                rd = list_entry(recovd->recovd_troubled_items.next,
-                                struct recovd_data, rd_managed_chain);
-                
-                list_del(&rd->rd_managed_chain);
-                if (!rd->rd_recover)
-                        LBUG();
-
-                CERROR("starting recovery for rd %p (conn %p)\n",
-                       rd, class_rd2conn(rd));
-                recovd->recovd_current_rd = rd;
-                recovd->recovd_flags &= ~RECOVD_FAILED;
-                recovd->recovd_phase = RECOVD_PREPARING;
-
-                spin_unlock(&recovd->recovd_lock);
-                rc = rd->rd_recover(rd, PTLRPC_RECOVD_PHASE_PREPARE);
-                spin_lock(&recovd->recovd_lock);
-                if (rc)
-                        goto cb_failed;
-                
-                recovd->recovd_next_phase = RECOVD_PREPARED;
-                break;
-
-            case RECOVD_PREPARED:
-                rd = recovd->recovd_current_rd;
-                recovd->recovd_phase = RECOVD_RECOVERING;
-
-                CERROR("recovery prepared for rd %p (conn %p), recovering\n",
-                       rd, class_rd2conn(rd));
-
-                spin_unlock(&recovd->recovd_lock);
-                rc = rd->rd_recover(rd, PTLRPC_RECOVD_PHASE_RECOVER);
-                spin_lock(&recovd->recovd_lock);
-                if (rc)
-                        goto cb_failed;
-                
-                recovd->recovd_next_phase = RECOVD_RECOVERED;
-                break;
-
-            case RECOVD_RECOVERED:
-                rd = recovd->recovd_current_rd;
-                recovd->recovd_phase = RECOVD_IDLE;
-                recovd->recovd_next_phase = RECOVD_PREPARING;
-
-                CERROR("recovery complete for rd %p (conn %p), recovering\n",
-                       rd, class_rd2conn(rd));
-                break;
-
-            default:
-                break;
+                }
         }
-
+        spin_unlock(&recovd->recovd_lock);
         RETURN(0);
 }
 
@@ -202,30 +213,21 @@ static int recovd_main(void *arg)
         sprintf(current->comm, "lustre_recovd");
         unlock_kernel();
 
-        /* Record that the  thread is running */
+        /* Signal that the thread is running. */
         recovd->recovd_thread = current;
-        recovd->recovd_flags = RECOVD_IDLE;
+        recovd->recovd_state = RECOVD_READY;
         wake_up(&recovd->recovd_ctl_waitq);
 
-        /* And now, loop forever on requests */
+        /* And now, loop forever on requests. */
         while (1) {
                 wait_event(recovd->recovd_waitq, recovd_check_event(recovd));
-
-                spin_lock(&recovd->recovd_lock);
-
-                if (recovd->recovd_flags & RECOVD_STOPPING) {
-                        spin_unlock(&recovd->recovd_lock);
-                        CERROR("lustre_recovd stopping\n");
-                        EXIT;
+                if (recovd->recovd_state == RECOVD_STOPPING)
                         break;
-                }
-
                 recovd_handle_event(recovd);
-                spin_unlock(&recovd->recovd_lock);
         }
 
         recovd->recovd_thread = NULL;
-        recovd->recovd_flags = RECOVD_STOPPED;
+        recovd->recovd_state = RECOVD_STOPPED;
         wake_up(&recovd->recovd_ctl_waitq);
         CDEBUG(D_NET, "mgr exiting process %d\n", current->pid);
         RETURN(0);
@@ -247,8 +249,6 @@ int recovd_setup(struct recovd_obd *recovd)
         init_waitqueue_head(&recovd->recovd_recovery_waitq);
         init_waitqueue_head(&recovd->recovd_ctl_waitq);
 
-        recovd->recovd_next_phase = RECOVD_PREPARING;
-        
         rc = kernel_thread(recovd_main, (void *)recovd,
                            CLONE_VM | CLONE_FS | CLONE_FILES);
         if (rc < 0) {
@@ -256,7 +256,7 @@ int recovd_setup(struct recovd_obd *recovd)
                 RETURN(-EINVAL);
         }
         wait_event(recovd->recovd_ctl_waitq,
-                   recovd->recovd_phase == RECOVD_IDLE);
+                   recovd->recovd_state == RECOVD_READY);
 
         /* exported and called by obdclass timeout handlers */
         class_signal_connection_failure = recovd_conn_fail;
@@ -268,12 +268,12 @@ int recovd_setup(struct recovd_obd *recovd)
 int recovd_cleanup(struct recovd_obd *recovd)
 {
         spin_lock(&recovd->recovd_lock);
-        recovd->recovd_flags = RECOVD_STOPPING;
+        recovd->recovd_state = RECOVD_STOPPING;
         wake_up(&recovd->recovd_waitq);
         spin_unlock(&recovd->recovd_lock);
 
         wait_event(recovd->recovd_ctl_waitq,
-                   (recovd->recovd_flags & RECOVD_STOPPED));
+                   (recovd->recovd_state == RECOVD_STOPPED));
         RETURN(0);
 }
 

diff --git a/lustre/ptlrpc/recover.c b/lustre/ptlrpc/recover.c
index ea047bf..0e44c86 100644 (file)
--- a/lustre/ptlrpc/recover.c
+++ b/lustre/ptlrpc/recover.c
@@ -115,7 +115,7 @@ static int ll_recover_reconnect(struct ptlrpc_connection *conn)
                 
                 /* replay what needs to be replayed */
                 if (req->rq_flags & PTL_RPC_FL_REPLAY) {
-                        CDEBUG(D_INODE, "req %Ld needs replay [last rcvd %Ld]\n",
+                        CDEBUG(D_NET, "req %Ld needs replay [last rcvd %Ld]\n",
                                req->rq_xid, conn->c_last_xid);
                         rc = ptlrpc_replay_req(req);
 #if 0
@@ -135,7 +135,7 @@ static int ll_recover_reconnect(struct ptlrpc_connection *conn)
                 /* server has seen req, we have reply: skip */
                 if ((req->rq_flags & PTL_RPC_FL_REPLIED)  &&
                     req->rq_xid <= conn->c_last_xid) { 
-                        CDEBUG(D_INODE,
+                        CDEBUG(D_NET,
                                "req %Ld was complete: skip [last rcvd %Ld]\n", 
                                req->rq_xid, conn->c_last_xid);
                         continue;
@@ -144,7 +144,7 @@ static int ll_recover_reconnect(struct ptlrpc_connection *conn)
                 /* server has lost req, we have reply: resend, ign reply */
                 if ((req->rq_flags & PTL_RPC_FL_REPLIED)  &&
                     req->rq_xid > conn->c_last_xid) { 
-                        CDEBUG(D_INODE, "lost req %Ld have rep: replay [last "
+                        CDEBUG(D_NET, "lost req %Ld have rep: replay [last "
                                "rcvd %Ld]\n", req->rq_xid, conn->c_last_xid);
                         rc = ptlrpc_replay_req(req); 
                         if (rc) {
@@ -157,7 +157,7 @@ static int ll_recover_reconnect(struct ptlrpc_connection *conn)
                 /* server has seen req, we have lost reply: -ERESTARTSYS */
                 if ( !(req->rq_flags & PTL_RPC_FL_REPLIED)  &&
                      req->rq_xid <= conn->c_last_xid) { 
-                        CDEBUG(D_INODE, "lost rep %Ld srv did req: restart "
+                        CDEBUG(D_NET, "lost rep %Ld srv did req: restart "
                                "[last rcvd %Ld]\n", 
                                req->rq_xid, conn->c_last_xid);
                         ptlrpc_restart_req(req);
@@ -166,7 +166,7 @@ static int ll_recover_reconnect(struct ptlrpc_connection *conn)
                 /* service has not seen req, no reply: resend */
                 if ( !(req->rq_flags & PTL_RPC_FL_REPLIED)  &&
                      req->rq_xid > conn->c_last_xid) {
-                        CDEBUG(D_INODE,
+                        CDEBUG(D_NET,
                                "lost rep/req %Ld: resend [last rcvd %Ld]\n", 
                                req->rq_xid, conn->c_last_xid);
                         ptlrpc_resend_req(req);
@@ -177,6 +177,8 @@ static int ll_recover_reconnect(struct ptlrpc_connection *conn)
         conn->c_level = LUSTRE_CONN_FULL;
         recovd_conn_fixed(conn);
 
+        CDEBUG(D_NET, "recovery complete on conn %p(%s), waking delayed reqs\n",
+               conn, conn->c_remote_uuid);
         /* Finally, continue what we delayed since recovery started */
         list_for_each_safe(tmp, pos, &conn->c_delayed_head) { 
                 req = list_entry(tmp, struct ptlrpc_request, rq_list);
@@ -189,6 +191,13 @@ static int ll_recover_reconnect(struct ptlrpc_connection *conn)
         return rc;
 }
 
+static int ll_retry_recovery(struct ptlrpc_connection *conn)
+{
+        /* XXX use a timer, sideshow bob */
+        recovd_conn_fail(conn);
+        return 0;
+}
+
 int ll_recover(struct recovd_data *rd, int phase)
 {
         struct ptlrpc_connection *conn = class_rd2conn(rd);
@@ -202,8 +211,7 @@ int ll_recover(struct recovd_data *rd, int phase)
             case PTLRPC_RECOVD_PHASE_RECOVER:
                 RETURN(ll_recover_reconnect(conn));
             case PTLRPC_RECOVD_PHASE_FAILURE:
-                fixme();
-                RETURN(0);
+                RETURN(ll_retry_recovery(conn));
         }
 
         LBUG();

diff --git a/lustre/ptlrpc/rpc.c b/lustre/ptlrpc/rpc.c
index 9485861..742e460 100644 (file)
--- a/lustre/ptlrpc/rpc.c
+++ b/lustre/ptlrpc/rpc.c
@@ -64,8 +64,6 @@ int connmgr_cleanup(struct obd_device *dev)
         RETURN(0);
 }
 
-/* should this be in llite? */
-
 int connmgr_iocontrol(long cmd, struct lustre_handle *hdl, int len, void *karg,
                       void *uarg)
 {
@@ -73,9 +71,8 @@ int connmgr_iocontrol(long cmd, struct lustre_handle *hdl, int len, void *karg,
         struct obd_device *obd = class_conn2obd(hdl);
         struct recovd_obd *recovd = &obd->u.recovd;
         struct obd_ioctl_data *data = karg;
-#if 0 && PARALLEL_RECOVERY
         struct list_head *tmp;
-#endif
+        int rc = 0;
 
         ENTRY;
 
@@ -84,7 +81,6 @@ int connmgr_iocontrol(long cmd, struct lustre_handle *hdl, int len, void *karg,
         
         /* Find the connection that's been rebuilt. */
         spin_lock(&recovd->recovd_lock);
-#if 0 && PARALLEL_RECOVERY
         list_for_each(tmp, &recovd->recovd_troubled_items) {
                 conn = list_entry(tmp, struct ptlrpc_connection,
                                   c_recovd_data.rd_managed_chain);
@@ -95,24 +91,14 @@ int connmgr_iocontrol(long cmd, struct lustre_handle *hdl, int len, void *karg,
                         break;
                 conn = NULL;
         }
-#endif
-        if (!recovd->recovd_current_rd) {
-                spin_unlock(&recovd->recovd_lock);
-                LBUG();
-        }
-
-        conn = class_rd2conn(recovd->recovd_current_rd);
-        spin_unlock(&recovd->recovd_lock);
 
-        spin_lock(&conn->c_lock);
+        if (!conn)
+                GOTO(out, rc = -EINVAL);
 
-        if (strcmp(conn->c_remote_uuid, data->ioc_inlbuf1)) {
-                CERROR("NEWCONN for %s: currently recovering %s\n",
-                       data->ioc_inlbuf1, conn->c_remote_uuid);
-                spin_unlock(&conn->c_lock);
-                RETURN(-EINVAL);
-        }
+        if (conn->c_recovd_data.rd_phase != RECOVD_PREPARING)
+                GOTO(out, rc = -EALREADY);
 
+        spin_lock(&conn->c_lock);
         if (data->ioc_inllen2) {
                 CERROR("conn %p UUID change %s -> %s\n",
                        conn, conn->c_remote_uuid, data->ioc_inlbuf2);
@@ -122,23 +108,13 @@ int connmgr_iocontrol(long cmd, struct lustre_handle *hdl, int len, void *karg,
                        conn->c_remote_uuid);
         }
         ptlrpc_readdress_connection(conn, conn->c_remote_uuid);
+        spin_unlock(&conn->c_lock);
         
-        spin_lock(&recovd->recovd_lock);
-#if 0 && PARALLEL_RECOVERY
-        if (conn->c_recovd_data->rd_state != RECOVD_PREPARING)
-                LBUG();
-        conn->c_recovd_data->rd_state = RECOVD_PREPARED;
-#endif
-        if (recovd->recovd_phase != RECOVD_PREPARING ||
-            recovd->recovd_next_phase != RECOVD_PREPARED ||
-            recovd->recovd_current_rd != &conn->c_recovd_data) {
-                LBUG();
-        }
-        recovd->recovd_phase = RECOVD_PREPARED;
+        conn->c_recovd_data.rd_phase = RECOVD_PREPARED;
         wake_up(&recovd->recovd_waitq);
+ out:
         spin_unlock(&recovd->recovd_lock);
-        spin_unlock(&conn->c_lock);
-        RETURN(0);
+        RETURN(rc);
 }