Whamcloud - gitweb
- it's valid that MDC/LMV notifies MDS
[fs/lustre-release.git] / lustre / ldlm / ldlm_lib.c
index 57d1058..fb70b65 100644 (file)
@@ -46,7 +46,7 @@ int client_obd_setup(struct obd_device *obddev, obd_count len, void *buf)
         int rq_portal, rp_portal, connect_op;
         char *name = obddev->obd_type->typ_name;
         char *mgmt_name = NULL;
-        int rc = 0;
+        int rc;
         struct obd_device *mgmt_obd;
         mgmtcli_register_for_events_t register_f;
         ENTRY;
@@ -112,7 +112,7 @@ int client_obd_setup(struct obd_device *obddev, obd_count len, void *buf)
         cli->cl_max_pages_per_rpc = PTLRPC_MAX_BRW_PAGES;
         cli->cl_max_rpcs_in_flight = OSC_MAX_RIF_DEFAULT;
 
-        ldlm_get_ref();
+        rc = ldlm_get_ref();
         if (rc) {
                 CERROR("ldlm_get_ref failed: %d\n", rc);
                 GOTO(err, rc);
@@ -219,7 +219,8 @@ int client_obd_cleanup(struct obd_device *obddev, int flags)
 
 int client_connect_import(struct lustre_handle *dlm_handle,
                           struct obd_device *obd,
-                          struct obd_uuid *cluuid)
+                          struct obd_uuid *cluuid,
+                          unsigned long connect_flags)
 {
         struct client_obd *cli = &obd->u.cli;
         struct obd_import *imp = cli->cl_import;
@@ -250,6 +251,7 @@ int client_connect_import(struct lustre_handle *dlm_handle,
                 GOTO(out_ldlm, rc);
 
         exp->exp_connection = ptlrpc_connection_addref(imp->imp_connection);
+        imp->imp_connect_flags = connect_flags;
         rc = ptlrpc_connect_import(imp, NULL);
         if (rc != 0) {
                 LASSERT (imp->imp_state == LUSTRE_IMP_DISCON);
@@ -333,9 +335,9 @@ int client_disconnect_export(struct obd_export *exp, int failover)
  * -------------------------------------------------------------------------- */
 
 int target_handle_reconnect(struct lustre_handle *conn, struct obd_export *exp,
-                            struct obd_uuid *cluuid)
+                            struct obd_uuid *cluuid, int initial_conn)
 {
-        if (exp->exp_connection) {
+        if (exp->exp_connection && !initial_conn) {
                 struct lustre_handle *hdl;
                 hdl = &exp->exp_imp_reverse->imp_remote_handle;
                 /* Might be a re-connect after a partition. */
@@ -361,8 +363,9 @@ int target_handle_reconnect(struct lustre_handle *conn, struct obd_export *exp,
         RETURN(0);
 }
 
-int target_handle_connect(struct ptlrpc_request *req, svc_handler_t handler)
+int target_handle_connect(struct ptlrpc_request *req)
 {
+        unsigned long connect_flags = 0, *cfp;
         struct obd_device *target;
         struct obd_export *export = NULL;
         struct obd_import *revimp;
@@ -372,8 +375,10 @@ int target_handle_connect(struct ptlrpc_request *req, svc_handler_t handler)
         struct obd_uuid remote_uuid;
         struct list_head *p;
         char *str, *tmp;
-        int rc = 0, abort_recovery;
+        int rc = 0;
         unsigned long flags;
+        int initial_conn = 0;
+        char peer_str[PTL_NALFMT_SIZE];
         ENTRY;
 
         OBD_RACE(OBD_FAIL_TGT_CONN_RACE); 
@@ -393,6 +398,7 @@ int target_handle_connect(struct ptlrpc_request *req, svc_handler_t handler)
         
         if (!target || target->obd_stopping || !target->obd_set_up) {
                 CERROR("UUID '%s' is not available for connect\n", str);
+
                 GOTO(out, rc = -ENODEV);
         }
 
@@ -406,14 +412,19 @@ int target_handle_connect(struct ptlrpc_request *req, svc_handler_t handler)
         obd_str2uuid (&cluuid, str);
 
         /* XXX extract a nettype and format accordingly */
-        snprintf(remote_uuid.uuid, sizeof remote_uuid,
-                 "NET_"LPX64"_UUID", req->rq_peer.peer_nid);
-
-        spin_lock_bh(&target->obd_processing_task_lock);
-        abort_recovery = target->obd_abort_recovery;
-        spin_unlock_bh(&target->obd_processing_task_lock);
-        if (abort_recovery)
-                target_abort_recovery(target);
+        switch (sizeof(ptl_nid_t)) {
+                /* NB the casts only avoid compiler warnings */
+        case 8:
+                snprintf(remote_uuid.uuid, sizeof remote_uuid,
+                         "NET_"LPX64"_UUID", (__u64)req->rq_peer.peer_nid);
+                break;
+        case 4:
+                snprintf(remote_uuid.uuid, sizeof remote_uuid,
+                         "NET_%x_UUID", (__u32)req->rq_peer.peer_nid);
+                break;
+        default:
+                LBUG();
+        }
 
         tmp = lustre_msg_buf(req->rq_reqmsg, 2, sizeof conn);
         if (tmp == NULL)
@@ -421,10 +432,17 @@ int target_handle_connect(struct ptlrpc_request *req, svc_handler_t handler)
 
         memcpy(&conn, tmp, sizeof conn);
 
+        cfp = lustre_msg_buf(req->rq_reqmsg, 3, sizeof(unsigned long));
+        LASSERT(cfp != NULL);
+        connect_flags = *cfp;
+
         rc = lustre_pack_reply(req, 0, NULL, NULL);
         if (rc)
                 GOTO(out, rc);
-
+        
+        if (lustre_msg_get_op_flags(req->rq_reqmsg) & MSG_CONNECT_INITIAL)
+                initial_conn = 1;
+        
         /* lctl gets a backstage, all-access pass. */
         if (obd_uuid_equals(&cluuid, &target->obd_uuid))
                 goto dont_check_exports;
@@ -436,7 +454,8 @@ int target_handle_connect(struct ptlrpc_request *req, svc_handler_t handler)
                         spin_unlock(&target->obd_dev_lock);
                         LASSERT(export->exp_obd == target);
 
-                        rc = target_handle_reconnect(&conn, export, &cluuid);
+                        rc = target_handle_reconnect(&conn, export, &cluuid,
+                                                     initial_conn);
                         break;
                 }
                 export = NULL;
@@ -444,7 +463,22 @@ int target_handle_connect(struct ptlrpc_request *req, svc_handler_t handler)
         /* If we found an export, we already unlocked. */
         if (!export) {
                 spin_unlock(&target->obd_dev_lock);
-        } else if (req->rq_reqmsg->conn_cnt == 1) {
+        } else if (req->rq_export == NULL && 
+                   atomic_read(&export->exp_rpc_count) > 0) {
+                CWARN("%s: refuse connection from %s/%s to 0x%p/%d\n",
+                      target->obd_name, cluuid.uuid,
+                      ptlrpc_peernid2str(&req->rq_peer, peer_str),
+                      export, atomic_read(&export->exp_refcount));
+                GOTO(out, rc = -EBUSY);
+        } else if (req->rq_export != NULL &&
+                   atomic_read(&export->exp_rpc_count) > 1) {
+                CWARN("%s: refuse reconnection from %s@%s to 0x%p/%d\n",
+                      target->obd_name, cluuid.uuid,
+                      ptlrpc_peernid2str(&req->rq_peer, peer_str),
+                      export, atomic_read(&export->exp_rpc_count));
+                GOTO(out, rc = -EBUSY);
+        }
+        else if (req->rq_reqmsg->conn_cnt == 1 && !initial_conn) {
                 CERROR("%s reconnected with 1 conn_cnt; cookies not random?\n",
                        cluuid.uuid);
                 GOTO(out, rc = -EALREADY);
@@ -452,28 +486,34 @@ int target_handle_connect(struct ptlrpc_request *req, svc_handler_t handler)
 
         /* Tell the client if we're in recovery. */
         /* If this is the first client, start the recovery timer */
+        CWARN("%s: connection from %s@%s/%lu %s\n", target->obd_name, cluuid.uuid,
+              ptlrpc_peernid2str(&req->rq_peer, peer_str), *cfp,
+              target->obd_recovering ? "(recovering)" : "");
         if (target->obd_recovering) {
                 lustre_msg_add_op_flags(req->rq_repmsg, MSG_CONNECT_RECOVERING);
-                target_start_recovery_timer(target, handler);
+                target_start_recovery_timer(target);
         }
-
+#if 0
         /* Tell the client if we support replayable requests */
         if (target->obd_replayable)
                 lustre_msg_add_op_flags(req->rq_repmsg, MSG_CONNECT_REPLAYABLE);
-
+#endif
         if (export == NULL) {
                 if (target->obd_recovering) {
-                        CERROR("denying connection for new client %s: "
+                        CERROR("denying connection for new client %s@%s: "
                                "%d clients in recovery for %lds\n", cluuid.uuid,
+                               ptlrpc_peernid2str(&req->rq_peer, peer_str),
                                target->obd_recoverable_clients,
                                (target->obd_recovery_timer.expires-jiffies)/HZ);
                         rc = -EBUSY;
                 } else {
  dont_check_exports:
-                        rc = obd_connect(&conn, target, &cluuid);
+                        rc = obd_connect(&conn, target, &cluuid, connect_flags);
                 }
         }
-
+        /* Tell the client if we support replayable requests */
+        if (target->obd_replayable)
+                lustre_msg_add_op_flags(req->rq_repmsg, MSG_CONNECT_REPLAYABLE);
 
         /* If all else goes well, this is our RPC return code. */
         req->rq_status = 0;
@@ -481,11 +521,6 @@ int target_handle_connect(struct ptlrpc_request *req, svc_handler_t handler)
         if (rc && rc != EALREADY)
                 GOTO(out, rc);
 
-        /* XXX track this all the time? */
-        if (target->obd_recovering) {
-                target->obd_connected_clients++;
-        }
-
         req->rq_repmsg->handle = conn;
 
         /* If the client and the server are the same node, we will already
@@ -503,13 +538,16 @@ int target_handle_connect(struct ptlrpc_request *req, svc_handler_t handler)
         LASSERT(export != NULL);
 
         spin_lock_irqsave(&export->exp_lock, flags);
-        if (export->exp_conn_cnt >= req->rq_reqmsg->conn_cnt) {
-                CERROR("%s: already connected at a higher conn_cnt: %d > %d\n",
-                       cluuid.uuid, export->exp_conn_cnt, 
+        if (initial_conn) {
+                req->rq_repmsg->conn_cnt = export->exp_conn_cnt + 1;
+        } else if (export->exp_conn_cnt >= req->rq_reqmsg->conn_cnt) {
+                CERROR("%s@%s: already connected at a higher conn_cnt: %d > %d\n",
+                       cluuid.uuid, ptlrpc_peernid2str(&req->rq_peer, peer_str),
+                       export->exp_conn_cnt, 
                        req->rq_reqmsg->conn_cnt);
                 spin_unlock_irqrestore(&export->exp_lock, flags);
                 GOTO(out, rc = -EALREADY);
-        }
+        } 
         export->exp_conn_cnt = req->rq_reqmsg->conn_cnt;
         spin_unlock_irqrestore(&export->exp_lock, flags);
 
@@ -528,6 +566,10 @@ int target_handle_connect(struct ptlrpc_request *req, svc_handler_t handler)
                 GOTO(out, rc = 0);
         }
 
+        if (target->obd_recovering) {
+                target->obd_connected_clients++;
+        }
+
         memcpy(&conn, lustre_msg_buf(req->rq_reqmsg, 2, sizeof conn),
                sizeof conn);
 
@@ -541,6 +583,8 @@ int target_handle_connect(struct ptlrpc_request *req, svc_handler_t handler)
         revimp->imp_dlm_fake = 1;
         revimp->imp_state = LUSTRE_IMP_FULL;
         class_import_put(revimp);
+
+        rc = obd_connect_post(export);
 out:
         if (rc)
                 req->rq_status = rc;
@@ -580,6 +624,40 @@ void target_destroy_export(struct obd_export *exp)
  * Recovery functions
  */
 
+struct ptlrpc_request *
+ptlrpc_clone_req( struct ptlrpc_request *orig_req) 
+{
+        struct ptlrpc_request *copy_req;
+        struct lustre_msg *copy_reqmsg;
+
+        OBD_ALLOC(copy_req, sizeof *copy_req);
+        if (!copy_req)
+                return NULL;
+        OBD_ALLOC(copy_reqmsg, orig_req->rq_reqlen);
+        if (!copy_reqmsg){
+                OBD_FREE(copy_req, sizeof *copy_req);
+                return NULL;
+        }
+
+        memcpy(copy_req, orig_req, sizeof *copy_req);
+        memcpy(copy_reqmsg, orig_req->rq_reqmsg, orig_req->rq_reqlen);
+        /* the copied req takes over the reply state */
+        orig_req->rq_reply_state = NULL;
+
+        copy_req->rq_reqmsg = copy_reqmsg;
+        class_export_get(copy_req->rq_export);
+        INIT_LIST_HEAD(&copy_req->rq_list);
+
+        return copy_req;
+}
+void ptlrpc_free_clone( struct ptlrpc_request *req) 
+{
+        class_export_put(req->rq_export);
+        list_del(&req->rq_list);
+        OBD_FREE(req->rq_reqmsg, req->rq_reqlen);
+        OBD_FREE(req, sizeof *req);
+}
+
 static void abort_delayed_replies(struct obd_device *obd)
 {
         struct ptlrpc_request *req;
@@ -622,24 +700,13 @@ static void abort_recovery_queue(struct obd_device *obd)
         }
 }
 
-void target_abort_recovery(void *data)
+static void target_abort_recovery(void *data)
 {
         struct obd_device *obd = data;
         int rc;
 
         CERROR("disconnecting clients and aborting recovery\n");
-        spin_lock_bh(&obd->obd_processing_task_lock);
-        if (!obd->obd_recovering) {
-                spin_unlock_bh(&obd->obd_processing_task_lock);
-                EXIT;
-                return;
-        }
-
-        obd->obd_recovering = obd->obd_abort_recovery = 0;
-
-        wake_up(&obd->obd_next_transno_waitq);
-        target_cancel_recovery_timer(obd);
-        spin_unlock_bh(&obd->obd_processing_task_lock);
+        LASSERT(!obd->obd_recovering);
 
         class_disconnect_exports(obd, 0);
 
@@ -647,7 +714,8 @@ void target_abort_recovery(void *data)
         if (OBT(obd) && OBP(obd, postrecov)) {
                 rc = OBP(obd, postrecov)(obd);
                 if (rc >= 0)
-                        CWARN("Cleanup %d orphans after recovery was aborted\n", rc);
+                        CWARN("Cleanup %d orphans after recovery was aborted\n",
+                              rc);
                 else
                         CERROR("postrecov failed %d\n", rc);
         }
@@ -690,50 +758,87 @@ static void reset_recovery_timer(struct obd_device *obd)
 
 
 /* Only start it the first time called */
-void target_start_recovery_timer(struct obd_device *obd, svc_handler_t handler)
+void target_start_recovery_timer(struct obd_device *obd)
 {
         spin_lock_bh(&obd->obd_processing_task_lock);
-        if (obd->obd_recovery_handler) {
+        if (!obd->obd_recovering || timer_pending(&obd->obd_recovery_timer)) {
                 spin_unlock_bh(&obd->obd_processing_task_lock);
                 return;
         }
         CWARN("%s: starting recovery timer (%us)\n", obd->obd_name,
                OBD_RECOVERY_TIMEOUT / HZ);
-        obd->obd_recovery_handler = handler;
         obd->obd_recovery_timer.function = target_recovery_expired;
         obd->obd_recovery_timer.data = (unsigned long)obd;
-        init_timer(&obd->obd_recovery_timer);
+        mod_timer(&obd->obd_recovery_timer, jiffies + OBD_RECOVERY_TIMEOUT);
         spin_unlock_bh(&obd->obd_processing_task_lock);
+}
 
-        reset_recovery_timer(obd);
+static void target_finish_recovery(struct obd_device *obd)
+{
+        struct list_head *tmp, *n;
+        int rc2;
+
+        ldlm_reprocess_all_ns(obd->obd_namespace);
+
+        CWARN("%s: all clients recovered, calling postrecov\n",
+              obd->obd_name);
+        /* when recovery finished, cleanup orphans on mds and ost */
+        if (OBT(obd) && OBP(obd, postrecov)) {
+                rc2 = OBP(obd, postrecov)(obd);
+                if (rc2 >= 0)
+                        CWARN("%s: all clients recovered, %d MDS "
+                              "orphans deleted\n", obd->obd_name, rc2);
+                else
+                        CERROR("postrecov failed %d\n", rc2);
+        }
+
+        CWARN("%s: recovery over, sending delayed replies\n",
+              obd->obd_name);
+        list_for_each_safe(tmp, n, &obd->obd_delayed_reply_queue) {
+                struct ptlrpc_request *req;
+                req = list_entry(tmp, struct ptlrpc_request, rq_list);
+                DEBUG_REQ(D_ERROR, req, "delayed:");
+                ptlrpc_reply(req);
+                ptlrpc_free_clone(req);
+        }
+        ptlrpc_run_recovery_over_upcall(obd);
 }
 
 static int check_for_next_transno(struct obd_device *obd)
 {
-        struct ptlrpc_request *req;
+        struct ptlrpc_request *req = NULL;
         int wake_up = 0, connected, completed, queue_len, max;
         __u64 next_transno, req_transno;
 
         spin_lock_bh(&obd->obd_processing_task_lock);
-        req = list_entry(obd->obd_recovery_queue.next,
-                         struct ptlrpc_request, rq_list);
+        if (!list_empty(&obd->obd_recovery_queue)) {
+                req = list_entry(obd->obd_recovery_queue.next,
+                                 struct ptlrpc_request, rq_list);
+                req_transno = req->rq_reqmsg->transno;
+        } else {
+                req_transno = 0;
+        }
+
         max = obd->obd_max_recoverable_clients;
-        req_transno = req->rq_reqmsg->transno;
         connected = obd->obd_connected_clients;
         completed = max - obd->obd_recoverable_clients;
         queue_len = obd->obd_requests_queued_for_recovery;
         next_transno = obd->obd_next_recovery_transno;
 
+        CDEBUG(D_HA,"max: %d, connected: %d, completed: %d, queue_len: %d, "
+               "req_transno: "LPU64", next_transno: "LPU64"\n",
+               max, connected, completed, queue_len, req_transno, next_transno);
         if (obd->obd_abort_recovery) {
                 CDEBUG(D_HA, "waking for aborted recovery\n");
                 wake_up = 1;
-        } else if (!obd->obd_recovering) {
-                CDEBUG(D_HA, "waking for completed recovery (?)\n");
+        } else if (max == completed) {
+                CDEBUG(D_HA, "waking for completed recovery\n");
                 wake_up = 1;
         } else if (req_transno == next_transno) {
                 CDEBUG(D_HA, "waking for next ("LPD64")\n", next_transno);
                 wake_up = 1;
         } else if (queue_len + completed == max) {
+                LASSERT(req->rq_reqmsg->transno >= next_transno);
                 CDEBUG(D_ERROR,
                        "waking for skipped transno (skip: "LPD64
                        ", ql: %d, comp: %d, conn: %d, next: "LPD64")\n",
@@ -742,63 +847,129 @@ static int check_for_next_transno(struct obd_device *obd)
                 wake_up = 1;
         }
         spin_unlock_bh(&obd->obd_processing_task_lock);
-        LASSERT(req->rq_reqmsg->transno >= next_transno);
+        
         return wake_up;
 }
 
-static void process_recovery_queue(struct obd_device *obd)
+static struct ptlrpc_request *
+target_next_replay_req(struct obd_device *obd)
 {
-        struct ptlrpc_request *req;
-        int abort_recovery = 0;
         struct l_wait_info lwi = { 0 };
-        ENTRY;
+        struct ptlrpc_request *req;
 
-        for (;;) {
-                spin_lock_bh(&obd->obd_processing_task_lock);
-                LASSERT(obd->obd_processing_task == current->pid);
+        CDEBUG(D_HA, "Waiting for transno "LPD64"\n",
+               obd->obd_next_recovery_transno);
+        l_wait_event(obd->obd_next_transno_waitq,
+                     check_for_next_transno(obd), &lwi);
+        
+        spin_lock_bh(&obd->obd_processing_task_lock);
+        if (obd->obd_abort_recovery) {
+                req = NULL;
+        } else if (!list_empty(&obd->obd_recovery_queue)) {
                 req = list_entry(obd->obd_recovery_queue.next,
                                  struct ptlrpc_request, rq_list);
+                list_del_init(&req->rq_list);
+                obd->obd_requests_queued_for_recovery--;
+        } else {
+                req = NULL;
+        }
+        spin_unlock_bh(&obd->obd_processing_task_lock);
+        return req;
+}
 
-                if (req->rq_reqmsg->transno != obd->obd_next_recovery_transno) {
-                        spin_unlock_bh(&obd->obd_processing_task_lock);
-                        CDEBUG(D_HA, "Waiting for transno "LPD64" (1st is "
-                               LPD64")\n",
-                               obd->obd_next_recovery_transno,
-                               req->rq_reqmsg->transno);
-                        l_wait_event(obd->obd_next_transno_waitq,
-                                     check_for_next_transno(obd), &lwi);
+
+static int target_recovery_thread(void *arg)
+{
+        struct obd_device *obd = arg;
+        struct ptlrpc_request *req;
+        struct target_recovery_data *trd = &obd->obd_recovery_data;
+        unsigned long flags;
+        ENTRY;
+
+        kportal_daemonize("tgt-recov");
+
+        SIGNAL_MASK_LOCK(current, flags);
+        sigfillset(&current->blocked);
+        RECALC_SIGPENDING;
+        SIGNAL_MASK_UNLOCK(current, flags);
+
+        CERROR("%s: started recovery thread pid %d\n", obd->obd_name, 
+               current->pid);
+        trd->trd_processing_task = current->pid;
+
+        obd->obd_recovering = 1;
+        complete(&trd->trd_starting);
+
+        while (obd->obd_recovering) {
+                LASSERT(trd->trd_processing_task == current->pid);
+                req = target_next_replay_req(obd);
+                if (req != NULL) {
+                        char peer_str[PTL_NALFMT_SIZE];
+                        DEBUG_REQ(D_HA, req, "processing t"LPD64" from %s: ", 
+                                  req->rq_reqmsg->transno, 
+                                  ptlrpc_peernid2str(&req->rq_peer, peer_str));
+                        (void)trd->trd_recovery_handler(req);
+                        obd->obd_replayed_requests++;
+                        reset_recovery_timer(obd);
+                        /* bug 1580: decide how to properly sync() in recovery*/
+                        //mds_fsync_super(mds->mds_sb);
+                        ptlrpc_free_clone(req);
                         spin_lock_bh(&obd->obd_processing_task_lock);
-                        abort_recovery = obd->obd_abort_recovery;
+                        obd->obd_next_recovery_transno++;
                         spin_unlock_bh(&obd->obd_processing_task_lock);
-                        if (abort_recovery) {
-                                target_abort_recovery(obd);
-                                return;
+                } else {
+                        /* recovery is over */
+                        spin_lock_bh(&obd->obd_processing_task_lock);
+                        obd->obd_recovering = 0;
+                        target_cancel_recovery_timer(obd);
+                        if (obd->obd_abort_recovery) {
+                                obd->obd_abort_recovery = 0;
+                                spin_unlock_bh(&obd->obd_processing_task_lock);
+                                target_abort_recovery(obd); 
+                        } else {
+                                LASSERT(obd->obd_recoverable_clients == 0);
+                                spin_unlock_bh(&obd->obd_processing_task_lock);
+                                target_finish_recovery(obd);
                         }
-                        continue;
                 }
-                list_del_init(&req->rq_list);
-                obd->obd_requests_queued_for_recovery--;
-                spin_unlock_bh(&obd->obd_processing_task_lock);
+        }
 
-                DEBUG_REQ(D_HA, req, "processing: ");
-                (void)obd->obd_recovery_handler(req);
-                obd->obd_replayed_requests++;
-                reset_recovery_timer(obd);
-                /* bug 1580: decide how to properly sync() in recovery */
-                //mds_fsync_super(mds->mds_sb);
-                class_export_put(req->rq_export);
-                OBD_FREE(req->rq_reqmsg, req->rq_reqlen);
-                OBD_FREE(req, sizeof *req);
-                spin_lock_bh(&obd->obd_processing_task_lock);
-                obd->obd_next_recovery_transno++;
-                if (list_empty(&obd->obd_recovery_queue)) {
-                        obd->obd_processing_task = 0;
-                        spin_unlock_bh(&obd->obd_processing_task_lock);
-                        break;
-                }
+        trd->trd_processing_task = 0;
+        complete(&trd->trd_finishing);
+        return 0;
+}
+
+int target_start_recovery_thread(struct obd_device *obd, svc_handler_t handler)
+{
+        int rc = 0;
+        struct target_recovery_data *trd = &obd->obd_recovery_data;
+
+        memset(trd, 0, sizeof(*trd));
+        init_completion(&trd->trd_starting);
+        init_completion(&trd->trd_finishing);
+        trd->trd_recovery_handler = handler;
+
+        if (kernel_thread(target_recovery_thread, obd, 0) == 0) 
+                wait_for_completion(&trd->trd_starting);
+        else
+                rc = -ECHILD;
+
+        return rc;
+}
+
+void target_stop_recovery_thread(struct obd_device *obd)
+{
+        spin_lock_bh(&obd->obd_processing_task_lock);
+        if (obd->obd_recovery_data.trd_processing_task > 0) {
+                struct target_recovery_data *trd = &obd->obd_recovery_data;
+                CERROR("%s: aborting recovery\n", obd->obd_name);
+                obd->obd_abort_recovery = 1;
+                wake_up(&obd->obd_next_transno_waitq);
+                spin_unlock_bh(&obd->obd_processing_task_lock);
+                wait_for_completion(&trd->trd_finishing);
+        } else {
                 spin_unlock_bh(&obd->obd_processing_task_lock);
         }
-        EXIT;
 }
 
 int target_queue_recovery_request(struct ptlrpc_request *req,
@@ -807,8 +978,6 @@ int target_queue_recovery_request(struct ptlrpc_request *req,
         struct list_head *tmp;
         int inserted = 0;
         __u64 transno = req->rq_reqmsg->transno;
-        struct ptlrpc_request *saved_req;
-        struct lustre_msg *reqmsg;
 
         /* CAVEAT EMPTOR: The incoming request message has been swabbed
          * (i.e. buflens etc are in my own byte order), but type-dependent
@@ -820,16 +989,6 @@ int target_queue_recovery_request(struct ptlrpc_request *req,
                 return 1;
         }
 
-        /* XXX If I were a real man, these LBUGs would be sane cleanups. */
-        /* XXX just like the request-dup code in queue_final_reply */
-        OBD_ALLOC(saved_req, sizeof *saved_req);
-        if (!saved_req)
-                LBUG();
-        OBD_ALLOC(reqmsg, req->rq_reqlen);
-        if (!reqmsg)
-                LBUG();
-
-        spin_lock_bh(&obd->obd_processing_task_lock);
 
         /* If we're processing the queue, we want don't want to queue this
          * message.
@@ -837,23 +996,34 @@ int target_queue_recovery_request(struct ptlrpc_request *req,
          * Also, if this request has a transno less than the one we're waiting
          * for, we should process it now.  It could (and currently always will)
          * be an open request for a descriptor that was opened some time ago.
+         *
+         * Also, a resent, replayed request that has already been
+         * handled will pass through here and be processed immediately.
          */
-        if (obd->obd_processing_task == current->pid ||
+        spin_lock_bh(&obd->obd_processing_task_lock);
+        if (obd->obd_recovery_data.trd_processing_task == current->pid ||
             transno < obd->obd_next_recovery_transno) {
                 /* Processing the queue right now, don't re-add. */
+                lustre_msg_clear_flags(req->rq_reqmsg, MSG_RESENT);
                 LASSERT(list_empty(&req->rq_list));
                 spin_unlock_bh(&obd->obd_processing_task_lock);
-                OBD_FREE(reqmsg, req->rq_reqlen);
-                OBD_FREE(saved_req, sizeof *saved_req);
                 return 1;
         }
+        spin_unlock_bh(&obd->obd_processing_task_lock);
+
+        /* A resent, replayed request that is still on the queue; just drop it.
+           The queued request will handle this. */
+        if ((lustre_msg_get_flags(req->rq_reqmsg) & (MSG_RESENT | MSG_REPLAY))
+            == (MSG_RESENT | MSG_REPLAY)) {
+                DEBUG_REQ(D_ERROR, req, "dropping resent queued req");
+                return 0;
+        }
+
+        req = ptlrpc_clone_req(req);
+        if (req == NULL)
+                return -ENOMEM;
 
-        memcpy(saved_req, req, sizeof *req);
-        memcpy(reqmsg, req->rq_reqmsg, req->rq_reqlen);
-        req = saved_req;
-        req->rq_reqmsg = reqmsg;
-        class_export_get(req->rq_export);
-        INIT_LIST_HEAD(&req->rq_list);
+        spin_lock_bh(&obd->obd_processing_task_lock);
 
         /* XXX O(n^2) */
         list_for_each(tmp, &obd->obd_recovery_queue) {
@@ -872,23 +1042,9 @@ int target_queue_recovery_request(struct ptlrpc_request *req,
         }
 
         obd->obd_requests_queued_for_recovery++;
-
-        if (obd->obd_processing_task != 0) {
-                /* Someone else is processing this queue, we'll leave it to
-                 * them.
-                 */
-                wake_up(&obd->obd_next_transno_waitq);
-                spin_unlock_bh(&obd->obd_processing_task_lock);
-                return 0;
-        }
-
-        /* Nobody is processing, and we know there's (at least) one to process
-         * now, so we'll do the honours.
-         */
-        obd->obd_processing_task = current->pid;
+        wake_up(&obd->obd_next_transno_waitq);
         spin_unlock_bh(&obd->obd_processing_task_lock);
 
-        process_recovery_queue(obd);
         return 0;
 }
 
@@ -900,10 +1056,6 @@ struct obd_device * target_req2obd(struct ptlrpc_request *req)
 int target_queue_final_reply(struct ptlrpc_request *req, int rc)
 {
         struct obd_device *obd = target_req2obd(req);
-        struct ptlrpc_request *saved_req;
-        struct lustre_msg *reqmsg;
-        int recovery_done = 0;
-        int rc2;
 
         LASSERT ((rc == 0) == (req->rq_reply_state != NULL));
 
@@ -916,63 +1068,23 @@ int target_queue_final_reply(struct ptlrpc_request *req, int rc)
 
         LASSERT (!req->rq_reply_state->rs_difficult);
         LASSERT(list_empty(&req->rq_list));
-        /* XXX a bit like the request-dup code in queue_recovery_request */
-        OBD_ALLOC(saved_req, sizeof *saved_req);
-        if (!saved_req)
-                LBUG();
-        OBD_ALLOC(reqmsg, req->rq_reqlen);
-        if (!reqmsg)
-                LBUG();
-        memcpy(saved_req, req, sizeof *saved_req);
-        memcpy(reqmsg, req->rq_reqmsg, req->rq_reqlen);
-        /* the copied req takes over the reply state */
-        req->rq_reply_state = NULL;
-        req = saved_req;
-        req->rq_reqmsg = reqmsg;
-        class_export_get(req->rq_export);
-        list_add(&req->rq_list, &obd->obd_delayed_reply_queue);
+        
+        req = ptlrpc_clone_req(req);
 
         spin_lock_bh(&obd->obd_processing_task_lock);
-        --obd->obd_recoverable_clients;
-        recovery_done = (obd->obd_recoverable_clients == 0);
-        spin_unlock_bh(&obd->obd_processing_task_lock);
-
-        if (recovery_done) {
-                struct list_head *tmp, *n;
-                ldlm_reprocess_all_ns(req->rq_export->exp_obd->obd_namespace);
-                CWARN("%s: all clients recovered, sending delayed replies\n",
-                       obd->obd_name);
-                spin_lock_bh(&obd->obd_processing_task_lock);
-                obd->obd_recovering = 0;
-                target_cancel_recovery_timer(obd);
-                spin_unlock_bh(&obd->obd_processing_task_lock);
 
-                /* when recovery finished, cleanup orphans on mds and ost */
-                if (OBT(obd) && OBP(obd, postrecov)) {
-                        rc2 = OBP(obd, postrecov)(obd);
-                        if (rc2 >= 0)
-                                CWARN("%s: all clients recovered, %d MDS "
-                                      "orphans deleted\n", obd->obd_name, rc2);
-                        else
-                                CERROR("postrecov failed %d\n", rc2);
-                }
+        list_add(&req->rq_list, &obd->obd_delayed_reply_queue);
 
-                list_for_each_safe(tmp, n, &obd->obd_delayed_reply_queue) {
-                        req = list_entry(tmp, struct ptlrpc_request, rq_list);
-                        DEBUG_REQ(D_ERROR, req, "delayed:");
-                        ptlrpc_reply(req);
-                        class_export_put(req->rq_export);
-                        list_del(&req->rq_list);
-                        OBD_FREE(req->rq_reqmsg, req->rq_reqlen);
-                        OBD_FREE(req, sizeof *req);
-                }
-                ptlrpc_run_recovery_over_upcall(obd);
-        } else {
+        /* only count the first "replay over" request from each
+           export */
+        if (req->rq_export->exp_replay_needed) {
+                --obd->obd_recoverable_clients;
+                req->rq_export->exp_replay_needed = 0;
                 CWARN("%s: %d recoverable clients remain\n",
-                       obd->obd_name, obd->obd_recoverable_clients);
-                wake_up(&obd->obd_next_transno_waitq);
+                      obd->obd_name, obd->obd_recoverable_clients);
         }
-
+        wake_up(&obd->obd_next_transno_waitq);
+        spin_unlock_bh(&obd->obd_processing_task_lock);
         return 1;
 }