X-Git-Url: https://git.whamcloud.com/gitweb?a=blobdiff_plain;f=lustre%2Fptlrpc%2Frecover.c;h=6731c7dc04bcbcc036e94c321713ad342051e600;hb=fbb7ead129258897f5a5d5c9ce28d31fbbe5bca2;hp=3320a448f0b6a31ddcf8b2ee442307cdc28fd621;hpb=d22678dad183c98f78fe312f6f3bed1b7cae8535;p=fs%2Flustre-release.git diff --git a/lustre/ptlrpc/recover.c b/lustre/ptlrpc/recover.c index 3320a44..6731c7d 100644 --- a/lustre/ptlrpc/recover.c +++ b/lustre/ptlrpc/recover.c @@ -37,111 +37,26 @@ #include #include #include +#include #include #include /* for IOC_LOV_SET_OSC_ACTIVE */ #include "ptlrpc_internal.h" -enum reconnect_result { - RECON_RESULT_RECOVERING = 1, - RECON_RESULT_RECONNECTED = 2, - RECON_RESULT_EVICTED = 3, -}; - -int ptlrpc_reconnect_import(struct obd_import *imp) -{ - struct obd_device *obd = imp->imp_obd; - int rc, size[] = {sizeof(imp->imp_target_uuid), - sizeof(obd->obd_uuid), - sizeof(imp->imp_dlm_handle)}; - char *tmp[] = {imp->imp_target_uuid.uuid, - obd->obd_uuid.uuid, - (char *)&imp->imp_dlm_handle}; - struct ptlrpc_connection *conn = imp->imp_connection; - struct ptlrpc_request *req; - struct lustre_handle old_hdl; - __u64 committed_before_reconnect = imp->imp_peer_committed_transno; - ENTRY; - - req = ptlrpc_prep_req(imp, imp->imp_connect_op, 3, size, tmp); - if (!req) - RETURN(-ENOMEM); - req->rq_level = LUSTRE_CONN_NEW; - req->rq_replen = lustre_msg_size(0, NULL); - rc = ptlrpc_queue_wait(req); - if (rc) { - /* what if rc > 0 ??*/ - CERROR("cannot connect to %s@%s: rc = %d\n", - imp->imp_target_uuid.uuid, conn->c_remote_uuid.uuid, rc); - GOTO(out_disc, rc); - } - - if (lustre_msg_get_op_flags(req->rq_repmsg) & MSG_CONNECT_RECONNECT) { - memset(&old_hdl, 0, sizeof(old_hdl)); - if (!memcmp(&old_hdl, &req->rq_repmsg->handle, - sizeof (old_hdl))) { - CERROR("%s@%s didn't like our handle "LPX64 - ", failed\n", imp->imp_target_uuid.uuid, - conn->c_remote_uuid.uuid, - imp->imp_dlm_handle.cookie); - GOTO(out_disc, rc = -ENOTCONN); - } - - if (memcmp(&imp->imp_remote_handle, &req->rq_repmsg->handle, - sizeof(imp->imp_remote_handle))) { - CERROR("%s@%s changed handle from "LPX64" to "LPX64 - "; copying, but this may foreshadow disaster\n", - imp->imp_target_uuid.uuid, - conn->c_remote_uuid.uuid, - imp->imp_remote_handle.cookie, - req->rq_repmsg->handle.cookie); - imp->imp_remote_handle = req->rq_repmsg->handle; - GOTO(out_disc, rc = RECON_RESULT_RECONNECTED); - } - - CERROR("reconnected to %s@%s after partition\n", - imp->imp_target_uuid.uuid, conn->c_remote_uuid.uuid); - GOTO(out_disc, rc = RECON_RESULT_RECONNECTED); - } else if (lustre_msg_get_op_flags(req->rq_repmsg) & - MSG_CONNECT_RECOVERING) { - rc = RECON_RESULT_RECOVERING; - } else { - rc = RECON_RESULT_EVICTED; - } - - old_hdl = imp->imp_remote_handle; - imp->imp_remote_handle = req->rq_repmsg->handle; - CERROR("reconnected to %s@%s ("LPX64", was "LPX64")!\n", - imp->imp_target_uuid.uuid, conn->c_remote_uuid.uuid, - imp->imp_remote_handle.cookie, old_hdl.cookie); - if (req->rq_repmsg->last_committed < committed_before_reconnect) { - CERROR("%s went back in time (transno "LPD64 - " was committed, server claims "LPD64 - ")! is shared storage not coherent?\n", - imp->imp_target_uuid.uuid, - imp->imp_peer_committed_transno, - req->rq_repmsg->last_committed); - } - - GOTO(out_disc, rc); - - out_disc: - ptlrpc_req_finished(req); - return rc; -} +static int ptlrpc_recover_import_no_retry(struct obd_import *, char *); void ptlrpc_run_recovery_over_upcall(struct obd_device *obd) { char *argv[4]; char *envp[3]; int rc; - ENTRY; + argv[0] = obd_lustre_upcall; argv[1] = "RECOVERY_OVER"; argv[2] = obd->obd_uuid.uuid; argv[3] = NULL; - + envp[0] = "HOME=/"; envp[1] = "PATH=/sbin:/bin:/usr/sbin:/usr/bin"; envp[2] = NULL; @@ -153,18 +68,28 @@ void ptlrpc_run_recovery_over_upcall(struct obd_device *obd) argv[0], argv[1], argv[2], rc); } else { - CERROR("Invoked upcall %s %s %s", - argv[0], argv[1], argv[2]); + CWARN("Invoked upcall %s %s %s\n", + argv[0], argv[1], argv[2]); } } void ptlrpc_run_failed_import_upcall(struct obd_import* imp) { +#ifdef __KERNEL__ + unsigned long flags; char *argv[7]; char *envp[3]; int rc; - ENTRY; + + spin_lock_irqsave(&imp->imp_lock, flags); + if (imp->imp_state == LUSTRE_IMP_CLOSED) { + spin_unlock_irqrestore(&imp->imp_lock, flags); + EXIT; + return; + } + spin_unlock_irqrestore(&imp->imp_lock, flags); + argv[0] = obd_lustre_upcall; argv[1] = "FAILED_IMPORT"; argv[2] = imp->imp_target_uuid.uuid; @@ -180,37 +105,70 @@ void ptlrpc_run_failed_import_upcall(struct obd_import* imp) rc = USERMODEHELPER(argv[0], argv, envp); if (rc < 0) { CERROR("Error invoking recovery upcall %s %s %s %s %s: %d; " - "check /proc/sys/lustre/lustre_upcall\n", + "check /proc/sys/lustre/upcall\n", argv[0], argv[1], argv[2], argv[3], argv[4],rc); } else { - CERROR("Invoked upcall %s %s %s %s %s\n", - argv[0], argv[1], argv[2], argv[3], argv[4]); + CWARN("Invoked upcall %s %s %s %s %s\n", + argv[0], argv[1], argv[2], argv[3], argv[4]); + } +#else + if (imp->imp_state == LUSTRE_IMP_CLOSED) { + EXIT; + return; } + ptlrpc_recover_import(imp, NULL); +#endif +} + +/* This might block waiting for the upcall to start, so it should + * not be called from a thread that shouldn't block. (Like ptlrpcd) */ +void ptlrpc_initiate_recovery(struct obd_import *imp) +{ + ENTRY; + + LASSERT (obd_lustre_upcall != NULL); + + if (strcmp(obd_lustre_upcall, "DEFAULT") == 0) { + CDEBUG(D_HA, "%s: starting recovery without upcall\n", + imp->imp_target_uuid.uuid); + ptlrpc_connect_import(imp, NULL); + } + else if (strcmp(obd_lustre_upcall, "NONE") == 0) { + CDEBUG(D_HA, "%s: recovery disabled\n", + imp->imp_target_uuid.uuid); + } + else { + CDEBUG(D_HA, "%s: calling upcall to start recovery\n", + imp->imp_target_uuid.uuid); + ptlrpc_run_failed_import_upcall(imp); + } + + EXIT; } -int ptlrpc_replay(struct obd_import *imp) +int ptlrpc_replay_next(struct obd_import *imp, int *inflight) { int rc = 0; struct list_head *tmp, *pos; - struct ptlrpc_request *req; + struct ptlrpc_request *req = NULL; unsigned long flags; + __u64 last_transno; ENTRY; + *inflight = 0; + /* It might have committed some after we last spoke, so make sure we * get rid of them now. */ spin_lock_irqsave(&imp->imp_lock, flags); ptlrpc_free_committed(imp); + last_transno = imp->imp_last_replay_transno; spin_unlock_irqrestore(&imp->imp_lock, flags); - CDEBUG(D_HA, "import %p from %s has committed "LPD64"\n", - imp, imp->imp_target_uuid.uuid, imp->imp_peer_committed_transno); - - list_for_each(tmp, &imp->imp_replay_list) { - req = list_entry(tmp, struct ptlrpc_request, rq_list); - DEBUG_REQ(D_HA, req, "RETAINED: "); - } + CDEBUG(D_HA, "import %p from %s committed "LPU64" last "LPU64"\n", + imp, imp->imp_target_uuid.uuid, imp->imp_peer_committed_transno, + last_transno); /* Do I need to hold a lock across this iteration? We shouldn't be * racing with any additions to the list, because we're in recovery @@ -228,20 +186,37 @@ int ptlrpc_replay(struct obd_import *imp) * just a little race... */ list_for_each_safe(tmp, pos, &imp->imp_replay_list) { - req = list_entry(tmp, struct ptlrpc_request, rq_list); + req = list_entry(tmp, struct ptlrpc_request, rq_replay_list); + + /* If need to resend, stop on the matching one first. It's + possible though it's already been committed, so in that case + we'll just continue with replay */ + if (imp->imp_resend_replay && + req->rq_transno == last_transno) { + lustre_msg_add_flags(req->rq_reqmsg, MSG_RESENT); + break; + } + + if (req->rq_transno > last_transno) { + imp->imp_last_replay_transno = req->rq_transno; + break; + } - DEBUG_REQ(D_HA, req, "REPLAY:"); + req = NULL; + } - rc = ptlrpc_replay_req(req); + imp->imp_resend_replay = 0; + if (req != NULL) { + rc = ptlrpc_replay_req(req); if (rc) { - CERROR("recovery replay error %d for req "LPD64"\n", - rc, req->rq_xid); + CERROR("recovery replay error %d for req " + LPD64"\n", rc, req->rq_xid); RETURN(rc); } + *inflight = 1; } - - RETURN(0); + RETURN(rc); } int ptlrpc_resend(struct obd_import *imp) @@ -259,7 +234,10 @@ int ptlrpc_resend(struct obd_import *imp) /* Well... what if lctl recover is called twice at the same time? */ spin_lock_irqsave(&imp->imp_lock, flags); - LASSERT(imp->imp_level == LUSTRE_CONN_RECOVER); + if (imp->imp_state != LUSTRE_IMP_RECOVER) { + spin_unlock_irqrestore(&imp->imp_lock, flags); + RETURN(-1); + } spin_unlock_irqrestore(&imp->imp_lock, flags); list_for_each_safe(tmp, pos, &imp->imp_sending_list) { @@ -280,47 +258,12 @@ void ptlrpc_wake_delayed(struct obd_import *imp) list_for_each_safe(tmp, pos, &imp->imp_delayed_list) { req = list_entry(tmp, struct ptlrpc_request, rq_list); - ptlrpc_put_connection(req->rq_connection); - req->rq_connection = - ptlrpc_connection_addref(req->rq_import->imp_connection); - - if (req->rq_set) { - DEBUG_REQ(D_HA, req, "waking (set %p):", req->rq_set); - wake_up(&req->rq_set->set_waitq); - } else { - DEBUG_REQ(D_HA, req, "waking:"); - wake_up(&req->rq_wait_for_rep); - } + DEBUG_REQ(D_HA, req, "waking (set %p):", req->rq_set); + ptlrpc_wake_client_req(req); } spin_unlock_irqrestore(&imp->imp_lock, flags); } -inline void ptlrpc_invalidate_import_state(struct obd_import *imp) -{ - struct ldlm_namespace *ns = imp->imp_obd->obd_namespace; - if (ptlrpc_ldlm_namespace_cleanup == NULL) - CERROR("ptlrpc/ldlm hook is NULL! Please tell phil\n"); - else - ptlrpc_ldlm_namespace_cleanup(ns, 1 /* no network ops */); - ptlrpc_abort_inflight(imp); -} - -void ptlrpc_handle_failed_import(struct obd_import *imp) -{ - ENTRY; - if (!imp->imp_replayable) { - CDEBUG(D_HA, - "import %s@%s for %s not replayable, deactivating\n", - imp->imp_target_uuid.uuid, - imp->imp_connection->c_remote_uuid.uuid, - imp->imp_obd->obd_name); - ptlrpc_set_import_active(imp, 0); - } - - ptlrpc_run_failed_import_upcall(imp); - EXIT; -} - void ptlrpc_request_handle_notconn(struct ptlrpc_request *failed_req) { int rc; @@ -332,260 +275,121 @@ void ptlrpc_request_handle_notconn(struct ptlrpc_request *failed_req) imp->imp_obd->obd_name, imp->imp_target_uuid.uuid, imp->imp_connection->c_remote_uuid.uuid); - rc = ptlrpc_recover_import(imp, NULL); - if (rc < 0) { - ptlrpc_resend_req(failed_req); - if (rc != -EALREADY) - ptlrpc_handle_failed_import(imp); - } else if (rc == RECON_RESULT_RECOVERING) { - ptlrpc_resend_req(failed_req); - } else { - if (rc != RECON_RESULT_EVICTED) { - /* like LBUG, without the locking up */ - CERROR("unknown recover_import result %d\n", rc); - portals_debug_dumplog(); - portals_run_lbug_upcall(__FILE__, __FUNCTION__, - __LINE__); + + if (ptlrpc_set_import_discon(imp)) { + if (!imp->imp_replayable) { + CDEBUG(D_HA, "import %s@%s for %s not replayable, " + "auto-deactivating\n", + imp->imp_target_uuid.uuid, + imp->imp_connection->c_remote_uuid.uuid, + imp->imp_obd->obd_name); + ptlrpc_deactivate_import(imp); } - LASSERT(failed_req->rq_import_generation < imp->imp_generation); - spin_lock_irqsave (&failed_req->rq_lock, flags); - failed_req->rq_err = 1; - spin_unlock_irqrestore (&failed_req->rq_lock, flags); + + rc = ptlrpc_connect_import(imp, NULL); } + + /* Wait for recovery to complete and resend. If evicted, then + this request will be errored out later.*/ + spin_lock_irqsave(&failed_req->rq_lock, flags); + failed_req->rq_resend = 1; + spin_unlock_irqrestore(&failed_req->rq_lock, flags); + EXIT; } +/* + * This should only be called by the ioctl interface, currently + * with the lctl deactivate and activate commands. + */ int ptlrpc_set_import_active(struct obd_import *imp, int active) { - struct obd_device *notify_obd; - unsigned long flags; - int rc; - - LASSERT(imp->imp_obd); + struct obd_device *obd = imp->imp_obd; + int rc = 0; - notify_obd = imp->imp_obd->u.cli.cl_containing_lov; + LASSERT(obd); /* When deactivating, mark import invalid, and abort in-flight * requests. */ if (!active) { - spin_lock_irqsave(&imp->imp_lock, flags); - /* This is a bit of a hack, but invalidating replayable - * imports makes a temporary reconnect failure into a much more - * ugly -- and hard to remedy -- situation. */ - if (!imp->imp_replayable) { - CDEBUG(D_HA, "setting import %s INVALID\n", - imp->imp_target_uuid.uuid); - imp->imp_invalid = 1; - } - imp->imp_generation++; - spin_unlock_irqrestore(&imp->imp_lock, flags); - ptlrpc_invalidate_import_state(imp); - //ptlrpc_abort_inflight(imp); + ptlrpc_invalidate_import(imp, 0); + imp->imp_deactive = 1; } - if (notify_obd == NULL) - GOTO(out, rc = 0); - - /* How gross is _this_? */ - if (!list_empty(¬ify_obd->obd_exports)) { - struct lustre_handle fakeconn; - struct obd_ioctl_data ioc_data = { 0 }; - struct obd_export *exp = - list_entry(notify_obd->obd_exports.next, - struct obd_export, exp_obd_chain); - - fakeconn.cookie = exp->exp_handle.h_cookie; - ioc_data.ioc_inlbuf1 = (char *)&imp->imp_target_uuid; - ioc_data.ioc_offset = active; - rc = obd_iocontrol(IOC_LOV_SET_OSC_ACTIVE, &fakeconn, - sizeof ioc_data, &ioc_data, NULL); - if (rc) - CERROR("error %sabling %s on LOV %p/%s: %d\n", - active ? "en" : "dis", - imp->imp_target_uuid.uuid, notify_obd, - notify_obd->obd_uuid.uuid, rc); - } else { - CDEBUG(D_HA, "No exports for obd %p/%s, can't notify about " - "%p\n", notify_obd, notify_obd->obd_uuid.uuid, - imp->imp_obd->obd_uuid.uuid); - rc = -ENOENT; - } - -out: - /* When activating, mark import valid */ - if (active && !rc) { + /* When activating, mark import valid, and attempt recovery */ + if (active) { + imp->imp_deactive = 0; CDEBUG(D_HA, "setting import %s VALID\n", imp->imp_target_uuid.uuid); - spin_lock_irqsave(&imp->imp_lock, flags); - imp->imp_invalid = 0; - spin_unlock_irqrestore(&imp->imp_lock, flags); + rc = ptlrpc_recover_import(imp, NULL); } RETURN(rc); } -void ptlrpc_fail_import(struct obd_import *imp, int generation) +int ptlrpc_recover_import(struct obd_import *imp, char *new_uuid) { - unsigned long flags; - int in_recovery = 0; + int rc; ENTRY; - LASSERT (!imp->imp_dlm_fake); + /* force import to be disconnected. */ + ptlrpc_set_import_discon(imp); - spin_lock_irqsave(&imp->imp_lock, flags); - if (imp->imp_level != LUSTRE_CONN_FULL) - in_recovery = 1; - else - imp->imp_level = LUSTRE_CONN_NOTCONN; - spin_unlock_irqrestore(&imp->imp_lock, flags); + rc = ptlrpc_recover_import_no_retry(imp, new_uuid); - if (in_recovery) { - EXIT; - return; - } - - ptlrpc_handle_failed_import(imp); - EXIT; + RETURN(rc); } -static int signal_completed_replay(struct obd_import *imp) +int ptlrpc_import_in_recovery(struct obd_import *imp) { - struct ptlrpc_request *req; - int rc; - ENTRY; - - req = ptlrpc_prep_req(imp, OBD_PING, 0, NULL, NULL); - if (!req) - RETURN(-ENOMEM); - - req->rq_replen = lustre_msg_size(0, NULL); - req->rq_level = LUSTRE_CONN_RECOVER; - req->rq_reqmsg->flags |= MSG_LAST_REPLAY; - - rc = ptlrpc_queue_wait(req); - - ptlrpc_req_finished(req); - RETURN(rc); + unsigned long flags; + int in_recovery = 1; + spin_lock_irqsave(&imp->imp_lock, flags); + if (imp->imp_state == LUSTRE_IMP_FULL || + imp->imp_state == LUSTRE_IMP_CLOSED || + imp->imp_state == LUSTRE_IMP_DISCON) + in_recovery = 0; + spin_unlock_irqrestore(&imp->imp_lock, flags); + return in_recovery; } -int ptlrpc_recover_import(struct obd_import *imp, char *new_uuid) +static int ptlrpc_recover_import_no_retry(struct obd_import *imp, + char *new_uuid) { int rc; unsigned long flags; - int in_recover = 0; - int recon_result; + int in_recovery = 0; + struct l_wait_info lwi; ENTRY; spin_lock_irqsave(&imp->imp_lock, flags); - if (imp->imp_level == LUSTRE_CONN_FULL || - imp->imp_level == LUSTRE_CONN_NOTCONN) - imp->imp_level = LUSTRE_CONN_RECOVER; - else - in_recover = 1; + if (imp->imp_state != LUSTRE_IMP_DISCON) { + in_recovery = 1; + } spin_unlock_irqrestore(&imp->imp_lock, flags); - if (in_recover == 1) + if (in_recovery == 1) RETURN(-EALREADY); - if (new_uuid) { - struct ptlrpc_connection *conn; - struct obd_uuid uuid; - struct ptlrpc_peer peer; - struct obd_export *dlmexp; - - obd_str2uuid(&uuid, new_uuid); - if (ptlrpc_uuid_to_peer(&uuid, &peer)) { - CERROR("no connection found for UUID %s\n", new_uuid); - RETURN(-EINVAL); - } - - conn = ptlrpc_get_connection(&peer, &uuid); - if (!conn) - RETURN(-ENOMEM); - - CDEBUG(D_HA, "switching import %s/%s from %s to %s\n", - imp->imp_target_uuid.uuid, imp->imp_obd->obd_name, - imp->imp_connection->c_remote_uuid.uuid, - conn->c_remote_uuid.uuid); - - /* Switch the import's connection and the DLM export's - * connection (which are almost certainly the same, but we - * keep distinct refs just to make things clearer. I think. */ - if (imp->imp_connection) - ptlrpc_put_connection(imp->imp_connection); - /* We hand off the ref from ptlrpc_get_connection. */ - imp->imp_connection = conn; - - dlmexp = class_conn2export(&imp->imp_dlm_handle); - if (dlmexp->exp_connection) - ptlrpc_put_connection(dlmexp->exp_connection); - dlmexp->exp_connection = ptlrpc_connection_addref(conn); - class_export_put(dlmexp); - - } - - recon_result = ptlrpc_reconnect_import(imp); - - if (recon_result < 0) { - CERROR("failed to reconnect to %s@%s: %d\n", - imp->imp_target_uuid.uuid, - imp->imp_connection->c_remote_uuid.uuid, recon_result); - spin_lock_irqsave(&imp->imp_lock, flags); - imp->imp_level = LUSTRE_CONN_NOTCONN; - spin_unlock_irqrestore(&imp->imp_lock, flags); - RETURN(recon_result); - } - - if (recon_result == RECON_RESULT_RECOVERING) { - CDEBUG(D_HA, "replay requested by %s\n", - imp->imp_target_uuid.uuid); - rc = ptlrpc_replay(imp); - if (rc) - GOTO(out, rc); - - if (ptlrpc_ldlm_replay_locks == NULL) - CERROR("ptlrpc/ldlm hook is NULL! Please tell phil\n"); - else - rc = ptlrpc_ldlm_replay_locks(imp); - if (rc) - GOTO(out, rc); - - rc = signal_completed_replay(imp); - if (rc) - GOTO(out, rc); - } else if (recon_result == RECON_RESULT_RECONNECTED) { - CDEBUG(D_HA, "reconnected to %s@%s\n", - imp->imp_target_uuid.uuid, - imp->imp_connection->c_remote_uuid.uuid); - } else if (recon_result == RECON_RESULT_EVICTED) { - CDEBUG(D_HA, "evicted from %s@%s; invalidating\n", - imp->imp_target_uuid.uuid, - imp->imp_connection->c_remote_uuid.uuid); - ptlrpc_set_import_active(imp, 0); - } else { - LBUG(); - } - - ptlrpc_set_import_active(imp, 1); + rc = ptlrpc_connect_import(imp, new_uuid); + if (rc) + RETURN(rc); - rc = ptlrpc_resend(imp); + CDEBUG(D_HA, "%s: recovery started, waiting\n", + imp->imp_target_uuid.uuid); - spin_lock_irqsave(&imp->imp_lock, flags); - imp->imp_level = LUSTRE_CONN_FULL; - spin_unlock_irqrestore(&imp->imp_lock, flags); + lwi = LWI_TIMEOUT(MAX(obd_timeout * HZ, 1), NULL, NULL); + rc = l_wait_event(imp->imp_recovery_waitq, + !ptlrpc_import_in_recovery(imp), &lwi); + CDEBUG(D_HA, "%s: recovery finished\n", + imp->imp_target_uuid.uuid); - ptlrpc_wake_delayed(imp); - EXIT; - if (!rc) - rc = recon_result; - out: - return rc; + RETURN(rc); } void ptlrpc_fail_export(struct obd_export *exp) { int rc, already_failed; - struct lustre_handle hdl; unsigned long flags; spin_lock_irqsave(&exp->exp_lock, flags); @@ -601,8 +405,15 @@ void ptlrpc_fail_export(struct obd_export *exp) CDEBUG(D_HA, "disconnecting export %p/%s\n", exp, exp->exp_client_uuid.uuid); - hdl.cookie = exp->exp_handle.h_cookie; - rc = obd_disconnect(&hdl, 0); + + if (obd_dump_on_timeout) + portals_debug_dumplog(); + + /* Most callers into obd_disconnect are removing their own reference + * (request, for example) in addition to the one from the hash table. + * We don't have such a reference here, so make one. */ + class_export_get(exp); + rc = obd_disconnect(exp, 0); if (rc) CERROR("disconnecting export %p failed: %d\n", exp, rc); }