From 00ab5a4e3a300c2d1b64be54780d265cb1b13b97 Mon Sep 17 00:00:00 2001
From: yury <yury>
Date: Wed, 12 Nov 2008 17:31:10 +0000
Subject: [PATCH] b=17037 r=tappro,wangdi
- fixes ost cleanup issue due to missed llcd_put() in the case ost does not receive disconnect from mds;
- do not sleep on hanging llcd. Instead assert on it _after_ stopping recov_thread's ptlrpcd which should kill any remeining llcds;
- fixes and cleanups, comments.
---
lustre/include/lustre_log.h | 18 ++++++----
lustre/obdclass/llog_cat.c | 2 +-
lustre/obdclass/llog_obd.c | 2 +-
lustre/obdfilter/filter.c | 59 ++++++++++++++++++-------------
lustre/osc/osc_request.c | 19 ++++++----
lustre/ptlrpc/client.c | 8 ++---
lustre/ptlrpc/recov_thread.c | 68 ++++++++++++++++++++++++------------
7 files changed, 111 insertions(+), 65 deletions(-)
diff --git a/lustre/include/lustre_log.h b/lustre/include/lustre_log.h
index 3140d5b147..215330cb2c 100644
--- a/lustre/include/lustre_log.h
+++ b/lustre/include/lustre_log.h
@@ -281,18 +281,18 @@ struct llog_commit_master {
* Number of llcds onthis lcm.
*/
atomic_t lcm_count;
- /**
- * Ptlrpc requests set. All cancel rpcs go via this request set.
- */
- struct ptlrpc_request_set *lcm_set;
/**
* Thread control structure. Used for control commit thread.
*/
struct ptlrpcd_ctl lcm_pc;
- /**
- * Busy resources waitq
+ /*
+ * Lock protecting list of llcds.
*/
- cfs_waitq_t lcm_waitq;
+ spinlock_t lcm_lock;
+ /*
+ * Llcds in flight for debugging purposes.
+ */
+ struct list_head lcm_llcds;
/**
* Commit thread name buffer. Only used for thread start.
*/
@@ -316,6 +316,10 @@ struct llog_canceld_ctxt {
* left in llcd to cookie comming cookies.
*/
int llcd_size;
+ /*
+ * Link to lcm llcds list.
+ */
+ struct list_head llcd_list;
/**
* Current llcd size while gathering cookies. This should not be
* more than ->llcd_size. Used for determining if we need to
diff --git a/lustre/obdclass/llog_cat.c b/lustre/obdclass/llog_cat.c
index 00151a69a1..a5c568bbbc 100644
--- a/lustre/obdclass/llog_cat.c
+++ b/lustre/obdclass/llog_cat.c
@@ -446,7 +446,7 @@ int llog_cat_process_thread(void *data)
if (cb) {
rc = llog_cat_process(llh, (llog_cb_t)cb, NULL);
- if (rc != LLOG_PROC_BREAK)
+ if (rc != LLOG_PROC_BREAK && rc != 0)
CERROR("llog_cat_process() failed %d\n", rc);
} else {
CWARN("No callback function for recovery\n");
diff --git a/lustre/obdclass/llog_obd.c b/lustre/obdclass/llog_obd.c
index 804049ac64..13a1e9a448 100644
--- a/lustre/obdclass/llog_obd.c
+++ b/lustre/obdclass/llog_obd.c
@@ -72,7 +72,7 @@ static void llog_ctxt_destroy(struct llog_ctxt *ctxt)
class_import_put(ctxt->loc_imp);
ctxt->loc_imp = NULL;
}
-
+ LASSERT(ctxt->loc_llcd == NULL);
OBD_FREE_PTR(ctxt);
return;
}
diff --git a/lustre/obdfilter/filter.c b/lustre/obdfilter/filter.c
index c7b65b3108..de162b1306 100644
--- a/lustre/obdfilter/filter.c
+++ b/lustre/obdfilter/filter.c
@@ -2334,19 +2334,27 @@ static int filter_llog_finish(struct obd_device *obd, int count)
ENTRY;
ctxt = llog_group_get_ctxt(&obd->obd_olg, LLOG_MDS_OST_REPL_CTXT);
- LASSERT(ctxt != NULL);
- mutex_down(&ctxt->loc_sem);
- if (ctxt->loc_imp) {
- /*
- * Balance class_import_get() in llog_receptor_accept(). This
- * is safe to do here, as llog is already synchronized and its
- * import may go.
- */
- class_import_put(ctxt->loc_imp);
- ctxt->loc_imp = NULL;
+ if (ctxt) {
+ mutex_down(&ctxt->loc_sem);
+ if (ctxt->loc_imp) {
+ /*
+ * 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() in llog_receptor_accept().
+ * This is safe to do, as llog is already synchronized
+ * and its import may go.
+ */
+ class_import_put(ctxt->loc_imp);
+ ctxt->loc_imp = NULL;
+ }
+ mutex_up(&ctxt->loc_sem);
+ llog_ctxt_put(ctxt);
}
- mutex_up(&ctxt->loc_sem);
- llog_ctxt_put(ctxt);
if (filter->fo_lcm) {
llog_recov_thread_fini(filter->fo_lcm, obd->obd_force);
@@ -2978,18 +2986,20 @@ static void filter_sync_llogs(struct obd_device *obd, struct obd_export *dexp)
(dexp == olg_min->olg_exp || dexp == NULL)) {
int err;
ctxt = llog_group_get_ctxt(olg_min,
- LLOG_MDS_OST_REPL_CTXT);
- LASSERT(ctxt != NULL);
- err = llog_sync(ctxt, olg_min->olg_exp);
- llog_ctxt_put(ctxt);
- if (err)
- CERROR("error flushing logs to MDS: rc %d\n",
- err);
+ LLOG_MDS_OST_REPL_CTXT);
+ if (ctxt) {
+ err = llog_sync(ctxt, olg_min->olg_exp);
+ llog_ctxt_put(ctxt);
+ if (err) {
+ CERROR("error flushing logs to MDS: "
+ "rc %d\n", err);
+ }
+ }
}
} while (olg_min != NULL);
}
-/* also incredibly similar to mds_disconnect */
+/* Also incredibly similar to mds_disconnect */
static int filter_disconnect(struct obd_export *exp)
{
struct obd_device *obd = exp->exp_obd;
@@ -3003,6 +3013,9 @@ static int filter_disconnect(struct obd_export *exp)
filter_grant_sanity_check(obd, __FUNCTION__);
filter_grant_discard(exp);
+ /* Flush any remaining cancel messages out to the target */
+ filter_sync_llogs(obd, exp);
+
/* Disconnect early so that clients can't keep using export */
rc = class_disconnect(exp);
if (exp->exp_obd->obd_namespace != NULL)
@@ -3012,8 +3025,6 @@ static int filter_disconnect(struct obd_export *exp)
lprocfs_exp_cleanup(exp);
- /* flush any remaining cancel messages out to the target */
- filter_sync_llogs(obd, exp);
class_export_put(exp);
RETURN(rc);
}
@@ -4087,10 +4098,10 @@ static int filter_sync(struct obd_export *exp, struct obdo *oa,
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 */
filter_sync_llogs(exp->exp_obd, exp);
RETURN(rc);
}
diff --git a/lustre/osc/osc_request.c b/lustre/osc/osc_request.c
index 2097a346b1..3a27a42384 100644
--- a/lustre/osc/osc_request.c
+++ b/lustre/osc/osc_request.c
@@ -3740,14 +3740,21 @@ static int osc_reconnect(const struct lu_env *env,
static int osc_disconnect(struct obd_export *exp)
{
struct obd_device *obd = class_exp2obd(exp);
- struct llog_ctxt *ctxt = llog_get_context(obd, LLOG_SIZE_REPL_CTXT);
+ struct llog_ctxt *ctxt;
int rc;
- if (obd->u.cli.cl_conn_count == 1)
- /* flush any remaining cancel messages out to the target */
- llog_sync(ctxt, exp);
-
- llog_ctxt_put(ctxt);
+ ctxt = llog_get_context(obd, LLOG_SIZE_REPL_CTXT);
+ if (ctxt) {
+ if (obd->u.cli.cl_conn_count == 1) {
+ /* Flush any remaining cancel messages out to the
+ * target */
+ llog_sync(ctxt, exp);
+ }
+ llog_ctxt_put(ctxt);
+ } else {
+ CDEBUG(D_HA, "No LLOG_SIZE_REPL_CTXT found in obd %p\n",
+ obd);
+ }
rc = client_disconnect_export(exp);
return rc;
diff --git a/lustre/ptlrpc/client.c b/lustre/ptlrpc/client.c
index 3285c1bb15..cbfcc5b11d 100644
--- a/lustre/ptlrpc/client.c
+++ b/lustre/ptlrpc/client.c
@@ -458,7 +458,7 @@ ptlrpc_prep_req_from_pool(struct ptlrpc_request_pool *pool)
request = list_entry(pool->prp_req_list.next, struct ptlrpc_request,
rq_list);
- list_del(&request->rq_list);
+ list_del_init(&request->rq_list);
spin_unlock(&pool->prp_lock);
LASSERT(request->rq_reqbuf);
@@ -1074,9 +1074,9 @@ static int ptlrpc_send_new_req(struct ptlrpc_request *req)
req->rq_import_generation = imp->imp_generation;
if (ptlrpc_import_delay_req(imp, req, &rc)) {
- spin_lock (&req->rq_lock);
+ spin_lock(&req->rq_lock);
req->rq_waiting = 1;
- spin_unlock (&req->rq_lock);
+ spin_unlock(&req->rq_lock);
DEBUG_REQ(D_HA, req, "req from PID %d waiting for recovery: "
"(%s != %s)", lustre_msg_get_status(req->rq_reqmsg),
@@ -2103,7 +2103,7 @@ int ptlrpc_queue_wait(struct ptlrpc_request *req)
req->rq_import_generation = imp->imp_generation;
restart:
if (ptlrpc_import_delay_req(imp, req, &rc)) {
- list_del(&req->rq_list);
+ list_del_init(&req->rq_list);
list_add_tail(&req->rq_list, &imp->imp_delayed_list);
atomic_inc(&imp->imp_inflight);
spin_unlock(&imp->imp_lock);
diff --git a/lustre/ptlrpc/recov_thread.c b/lustre/ptlrpc/recov_thread.c
index 497d2816d7..6fa95a708e 100644
--- a/lustre/ptlrpc/recov_thread.c
+++ b/lustre/ptlrpc/recov_thread.c
@@ -95,6 +95,7 @@ static struct llog_canceld_ctxt *llcd_alloc(void)
if (!llcd)
return NULL;
+ CFS_INIT_LIST_HEAD(&llcd->llcd_list);
llcd->llcd_size = llcd_size;
llcd->llcd_cookiebytes = 0;
atomic_inc(&llcd_count);
@@ -259,7 +260,10 @@ llcd_attach(struct llog_ctxt *ctxt, struct llog_canceld_ctxt *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);
@@ -290,7 +294,11 @@ static struct llog_canceld_ctxt *llcd_detach(struct llog_ctxt *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",
@@ -329,9 +337,6 @@ static void llcd_put(struct llog_ctxt *ctxt)
llcd = llcd_detach(ctxt);
if (llcd)
llcd_free(llcd);
-
- if (atomic_read(&lcm->lcm_count) == 0)
- cfs_waitq_signal(&lcm->lcm_waitq);
}
/**
@@ -376,7 +381,6 @@ int llog_recov_thread_start(struct llog_commit_master *lcm)
rc, lcm->lcm_name);
RETURN(rc);
}
- lcm->lcm_set = lcm->lcm_pc.pc_set;
RETURN(rc);
}
EXPORT_SYMBOL(llog_recov_thread_start);
@@ -386,27 +390,40 @@ 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);
@@ -431,8 +448,9 @@ struct llog_commit_master *llog_recov_thread_init(char *name)
"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);
@@ -553,11 +571,6 @@ int llog_obd_repl_cancel(struct llog_ctxt *ctxt,
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);
@@ -602,13 +615,15 @@ int llog_obd_repl_cancel(struct llog_ctxt *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);
@@ -629,16 +644,25 @@ int llog_obd_repl_sync(struct llog_ctxt *ctxt, struct obd_export *exp)
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);
}
--
GitLab