From d35b4f5124dc94227d5358e5c80074f2b5e0b432 Mon Sep 17 00:00:00 2001 From: yury Date: Wed, 12 Nov 2008 15:40:17 +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 | 16 ++++--- lustre/obdclass/llog_cat.c | 2 +- lustre/obdclass/llog_obd.c | 1 + lustre/obdfilter/filter.c | 40 +++++++++++------ lustre/osc/osc_request.c | 19 +++++--- lustre/ptlrpc/client.c | 8 ++-- lustre/ptlrpc/recov_thread.c | 105 ++++++++++++++++++++++++++++--------------- 7 files changed, 124 insertions(+), 67 deletions(-) diff --git a/lustre/include/lustre_log.h b/lustre/include/lustre_log.h index c2f550f..d332827 100644 --- a/lustre/include/lustre_log.h +++ b/lustre/include/lustre_log.h @@ -279,17 +279,17 @@ struct llog_commit_master { */ 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. + */ + spinlock_t lcm_lock; + /** + * Llcds in flight for debugging purposes. */ - cfs_waitq_t lcm_waitq; + struct list_head lcm_llcds; /** * Commit thread name buffer. Only used for thread start. */ @@ -314,6 +314,10 @@ struct llog_canceld_ctxt { */ 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 * send this llcd (if full) and allocate new one. This is also diff --git a/lustre/obdclass/llog_cat.c b/lustre/obdclass/llog_cat.c index b9a6833..10aa38e 100644 --- a/lustre/obdclass/llog_cat.c +++ b/lustre/obdclass/llog_cat.c @@ -437,7 +437,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 b9b41fc..888b7f3 100644 --- a/lustre/obdclass/llog_obd.c +++ b/lustre/obdclass/llog_obd.c @@ -72,6 +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(ctxt, sizeof(*ctxt)); return; } diff --git a/lustre/obdfilter/filter.c b/lustre/obdfilter/filter.c index 5fc6ff7..552f1b5 100644 --- a/lustre/obdfilter/filter.c +++ b/lustre/obdfilter/filter.c @@ -2006,6 +2006,12 @@ static int filter_llog_finish(struct obd_device *obd, int count) 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. @@ -2427,12 +2433,21 @@ static int filter_disconnect(struct obd_export *exp) { 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); @@ -2443,15 +2458,6 @@ static int filter_disconnect(struct obd_export *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); } @@ -3482,13 +3488,19 @@ 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 */ 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); } diff --git a/lustre/osc/osc_request.c b/lustre/osc/osc_request.c index 3151dcd..0af9007 100644 --- a/lustre/osc/osc_request.c +++ b/lustre/osc/osc_request.c @@ -3729,14 +3729,21 @@ static int osc_reconnect(struct obd_export *exp, struct obd_device *obd, 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 4122112..b0565db 100644 --- a/lustre/ptlrpc/client.c +++ b/lustre/ptlrpc/client.c @@ -517,7 +517,7 @@ static struct ptlrpc_request *ptlrpc_prep_req_from_pool(struct ptlrpc_request_po 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_reqmsg); @@ -972,9 +972,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), @@ -1918,7 +1918,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 c524cc8..4f2f744 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); @@ -169,7 +170,7 @@ static int llcd_send(struct llog_canceld_ctxt *llcd) 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; @@ -211,27 +212,36 @@ static int llcd_send(struct llog_canceld_ctxt *llcd) * 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); @@ -251,7 +261,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); @@ -282,7 +295,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", @@ -321,9 +338,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); } /** @@ -368,7 +382,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); @@ -378,27 +391,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); @@ -423,8 +449,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); @@ -545,11 +572,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); @@ -594,13 +616,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); @@ -621,16 +645,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); } -- 1.8.3.1