ctxt = llog_get_context(obd, LLOG_MDS_OST_REPL_CTXT);
if (ctxt) {
/*
+ * Make sure that no cached llcds left in recov_thread. We
+ * actually do sync in disconnect time, but disconnect may
+ * not come being marked rq_no_resend = 1.
+ */
+ llog_sync(ctxt, NULL);
+ /*
* Balance class_import_get() called in llog_receptor_accept().
* This is safe to do here, as llog is already synchronized and
* its import may go.
{
struct obd_device *obd = exp->exp_obd;
struct llog_ctxt *ctxt;
- int rc, err;
+ int rc;
ENTRY;
LASSERT(exp);
class_export_get(exp);
+ /* Flush any remaining cancel messages out to the target */
+ ctxt = llog_get_context(obd, LLOG_MDS_OST_REPL_CTXT);
+ if (ctxt) {
+ if (ctxt->loc_imp == exp->exp_imp_reverse)
+ CDEBUG(D_RPCTRACE, "Reverse import disconnect\n");
+ llog_sync(ctxt, exp);
+ llog_ctxt_put(ctxt);
+ }
+
if (!(exp->exp_flags & OBD_OPT_FORCE))
filter_grant_sanity_check(obd, __FUNCTION__);
filter_grant_discard(exp);
ldlm_cancel_locks_for_export(exp);
lprocfs_exp_cleanup(exp);
-
- /* flush any remaining cancel messages out to the target */
- ctxt = llog_get_context(obd, LLOG_MDS_OST_REPL_CTXT);
- err = llog_sync(ctxt, exp);
- llog_ctxt_put(ctxt);
-
- if (err)
- CERROR("error flushing logs to MDS: rc %d\n", err);
-
class_export_put(exp);
RETURN(rc);
}
filter = &exp->exp_obd->u.filter;
- /* an objid of zero is taken to mean "sync whole filesystem" */
+ /* An objid of zero is taken to mean "sync whole filesystem" */
if (!oa || !(oa->o_valid & OBD_MD_FLID)) {
rc = fsfilt_sync(exp->exp_obd, filter->fo_obt.obt_sb);
- /* flush any remaining cancel messages out to the target */
+
+ /* Flush any remaining cancel messages out to the target */
ctxt = llog_get_context(exp->exp_obd, LLOG_MDS_OST_REPL_CTXT);
- llog_sync(ctxt, exp);
- llog_ctxt_put(ctxt);
+ if (ctxt) {
+ llog_sync(ctxt, exp);
+ llog_ctxt_put(ctxt);
+ } else {
+ CERROR("No LLOG_MDS_OST_REPL_CTXT found in obd %p\n",
+ exp->exp_obd);
+ }
RETURN(rc);
}
if (!llcd)
return NULL;
+ CFS_INIT_LIST_HEAD(&llcd->llcd_list);
llcd->llcd_size = llcd_size;
llcd->llcd_cookiebytes = 0;
atomic_inc(&llcd_count);
char *bufs[2] = { NULL, (char *)llcd->llcd_cookies };
struct obd_import *import = NULL;
struct llog_commit_master *lcm;
- struct ptlrpc_request *request;
+ struct ptlrpc_request *req;
struct llog_ctxt *ctxt;
int rc;
ENTRY;
* No need to get import here as it is already done in
* llog_receptor_accept().
*/
- request = ptlrpc_prep_req(import, LUSTRE_LOG_VERSION,
- OBD_LOG_CANCEL, 2, size, bufs);
- if (request == NULL) {
+ req = ptlrpc_prep_req(import, LUSTRE_LOG_VERSION,
+ OBD_LOG_CANCEL, 2, size, bufs);
+ if (req == NULL) {
CERROR("Can't allocate request for sending llcd %p\n",
llcd);
GOTO(exit, rc = -ENOMEM);
}
+ /*
+ * Check if we're in exit stage again. Do not send llcd in
+ * this case.
+ */
+ if (test_bit(LLOG_LCM_FL_EXIT, &lcm->lcm_flags)) {
+ ptlrpc_req_finished(req);
+ GOTO(exit, rc = -ENODEV);
+ }
+
/* bug 5515 */
- request->rq_request_portal = LDLM_CANCEL_REQUEST_PORTAL;
- request->rq_reply_portal = LDLM_CANCEL_REPLY_PORTAL;
- ptlrpc_req_set_repsize(request, 1, NULL);
- ptlrpc_at_set_req_timeout(request);
- request->rq_interpret_reply = llcd_interpret;
- request->rq_async_args.pointer_arg[0] = llcd;
- rc = ptlrpc_set_add_new_req(&lcm->lcm_pc, request);
+ req->rq_request_portal = LDLM_CANCEL_REQUEST_PORTAL;
+ req->rq_reply_portal = LDLM_CANCEL_REPLY_PORTAL;
+ ptlrpc_req_set_repsize(req, 1, NULL);
+ ptlrpc_at_set_req_timeout(req);
+ req->rq_interpret_reply = llcd_interpret;
+ req->rq_async_args.pointer_arg[0] = llcd;
+ rc = ptlrpc_set_add_new_req(&lcm->lcm_pc, req);
if (rc) {
- ptlrpc_req_finished(request);
+ ptlrpc_req_finished(req);
GOTO(exit, rc);
}
- RETURN(0);
+ RETURN(rc);
exit:
CDEBUG(D_RPCTRACE, "Refused llcd %p\n", llcd);
llcd_free(llcd);
LASSERT_SEM_LOCKED(&ctxt->loc_sem);
LASSERT(ctxt->loc_llcd == NULL);
lcm = ctxt->loc_lcm;
+ spin_lock(&lcm->lcm_lock);
atomic_inc(&lcm->lcm_count);
+ list_add_tail(&llcd->llcd_list, &lcm->lcm_llcds);
+ spin_unlock(&lcm->lcm_lock);
CDEBUG(D_RPCTRACE, "Attach llcd %p to ctxt %p (%d)\n",
llcd, ctxt, atomic_read(&lcm->lcm_count));
llcd->llcd_ctxt = llog_ctxt_get(ctxt);
llcd_print(llcd, __FUNCTION__, __LINE__);
LBUG();
}
+ spin_lock(&lcm->lcm_lock);
+ LASSERT(!list_empty(&llcd->llcd_list));
+ list_del_init(&llcd->llcd_list);
atomic_dec(&lcm->lcm_count);
+ spin_unlock(&lcm->lcm_lock);
ctxt->loc_llcd = NULL;
CDEBUG(D_RPCTRACE, "Detach llcd %p from ctxt %p (%d)\n",
llcd = llcd_detach(ctxt);
if (llcd)
llcd_free(llcd);
-
- if (atomic_read(&lcm->lcm_count) == 0)
- cfs_waitq_signal(&lcm->lcm_waitq);
}
/**
rc, lcm->lcm_name);
RETURN(rc);
}
- lcm->lcm_set = lcm->lcm_pc.pc_set;
RETURN(rc);
}
EXPORT_SYMBOL(llog_recov_thread_start);
*/
void llog_recov_thread_stop(struct llog_commit_master *lcm, int force)
{
- struct l_wait_info lwi = LWI_INTR(LWI_ON_SIGNAL_NOOP, NULL);
ENTRY;
- /**
+ /*
* Let all know that we're stopping. This will also make
* llcd_send() refuse any new llcds.
*/
set_bit(LLOG_LCM_FL_EXIT, &lcm->lcm_flags);
- /**
+ /*
* Stop processing thread. No new rpcs will be accepted for
* for processing now.
*/
ptlrpcd_stop(&lcm->lcm_pc, force);
-
+
/*
- * Wait for llcd number == 0. Note, this is infinite wait.
- * All other parts should make sure that no lost llcd is left.
+ * By this point no alive inflight llcds should be left. Only
+ * those forgotten in sync may still be attached to ctxt. Let's
+ * print them.
*/
- l_wait_event(lcm->lcm_waitq,
- atomic_read(&lcm->lcm_count) == 0, &lwi);
+ if (atomic_read(&lcm->lcm_count) != 0) {
+ struct llog_canceld_ctxt *llcd;
+ struct list_head *tmp;
+
+ CERROR("Busy llcds found (%d) on lcm %p\n",
+ atomic_read(&lcm->lcm_count) == 0, lcm);
+
+ spin_lock(&lcm->lcm_lock);
+ list_for_each(tmp, &lcm->lcm_llcds) {
+ llcd = list_entry(tmp, struct llog_canceld_ctxt,
+ llcd_list);
+ llcd_print(llcd, __FUNCTION__, __LINE__);
+ }
+ spin_unlock(&lcm->lcm_lock);
+ }
EXIT;
}
EXPORT_SYMBOL(llog_recov_thread_stop);
"ll_log_commit_%s", name);
strncpy(lcm->lcm_name, name, sizeof(lcm->lcm_name));
- cfs_waitq_init(&lcm->lcm_waitq);
atomic_set(&lcm->lcm_count, 0);
+ spin_lock_init(&lcm->lcm_lock);
+ CFS_INIT_LIST_HEAD(&lcm->lcm_llcds);
rc = llog_recov_thread_start(lcm);
if (rc) {
CERROR("Can't start commit thread, rc %d\n", rc);
GOTO(out, rc = -ENODEV);
}
- if (ctxt->loc_obd->obd_stopping) {
- CDEBUG(D_RPCTRACE, "Obd is stopping for ctxt %p\n", ctxt);
- GOTO(out, rc = -ENODEV);
- }
-
if (test_bit(LLOG_LCM_FL_EXIT, &lcm->lcm_flags)) {
CDEBUG(D_RPCTRACE, "Commit thread is stopping for ctxt %p\n",
ctxt);
}
/*
- * Copy cookies to @llcd, no matter old or new allocated one.
+ * Copy cookies to @llcd, no matter old or new allocated
+ * one.
*/
llcd_copy(llcd, cookies);
}
/*
- * Let's check if we need to send copied @cookies asap. If yes - do it.
+ * Let's check if we need to send copied @cookies asap. If yes
+ * then do it.
*/
if (llcd && (flags & OBD_LLOG_FL_SENDNOW)) {
rc = llcd_push(ctxt);
int rc = 0;
ENTRY;
+ /*
+ * Flush any remaining llcd.
+ */
mutex_down(&ctxt->loc_sem);
if (exp && (ctxt->loc_imp == exp->exp_imp_reverse)) {
- CDEBUG(D_RPCTRACE, "Reverse import disconnect\n");
/*
- * Check for llcd which might be left attached to @ctxt.
- * Let's kill it.
+ * This is ost->mds connection, we can't be sure that mds
+ * can still receive cookies, let's killed the cached llcd.
*/
+ CDEBUG(D_RPCTRACE, "Kill cached llcd\n");
llcd_put(ctxt);
mutex_up(&ctxt->loc_sem);
} else {
+ /*
+ * This is either llog_sync() from generic llog code or sync
+ * on client disconnect. In either way let's do it and send
+ * llcds to the target with waiting for completion.
+ */
+ CDEBUG(D_RPCTRACE, "Sync cached llcd\n");
mutex_up(&ctxt->loc_sem);
rc = llog_cancel(ctxt, NULL, 0, NULL, OBD_LLOG_FL_SENDNOW);
}
git://git.whamcloud.com / fs / lustre-release.git / commitdiff