X-Git-Url: https://git.whamcloud.com/?a=blobdiff_plain;f=lustre%2Fobdclass%2Fllog_cat.c;h=2a0c99916894c056ec29094a2590690681d6c7fd;hb=fcfd78204a9fbd6b2c353403844ca57541d9c5cd;hp=a3c5873a92db366b3f78b3f9140e02ff60e6c4a9;hpb=abe5c61764860431d437b453310aea7742f68064;p=fs%2Flustre-release.git diff --git a/lustre/obdclass/llog_cat.c b/lustre/obdclass/llog_cat.c index a3c5873..2a0c9991 100644 --- a/lustre/obdclass/llog_cat.c +++ b/lustre/obdclass/llog_cat.c @@ -27,7 +27,7 @@ * Copyright (c) 2003, 2010, Oracle and/or its affiliates. All rights reserved. * Use is subject to license terms. * - * Copyright (c) 2011, 2012, Intel, Inc. + * Copyright (c) 2012, 2013, Intel Corporation. */ /* * This file is part of Lustre, http://www.lustre.org/ @@ -66,9 +66,9 @@ static int llog_cat_new_log(const struct lu_env *env, struct llog_handle *loghandle, struct thandle *th) { - + struct llog_thread_info *lgi = llog_info(env); + struct llog_logid_rec *rec = &lgi->lgi_logid; struct llog_log_hdr *llh; - struct llog_logid_rec rec = { { 0 }, }; int rc, index, bitmap_size; ENTRY; @@ -105,32 +105,33 @@ static int llog_cat_new_log(const struct lu_env *env, if (index == 0) index = 1; - cfs_spin_lock(&loghandle->lgh_hdr_lock); + spin_lock(&loghandle->lgh_hdr_lock); llh->llh_count++; - if (ext2_set_bit(index, llh->llh_bitmap)) { - CERROR("argh, index %u already set in log bitmap?\n", - index); - cfs_spin_unlock(&loghandle->lgh_hdr_lock); - LBUG(); /* should never happen */ - } - cfs_spin_unlock(&loghandle->lgh_hdr_lock); + if (ext2_set_bit(index, llh->llh_bitmap)) { + CERROR("argh, index %u already set in log bitmap?\n", + index); + spin_unlock(&loghandle->lgh_hdr_lock); + LBUG(); /* should never happen */ + } + spin_unlock(&loghandle->lgh_hdr_lock); cathandle->lgh_last_idx = index; llh->llh_tail.lrt_index = index; - CDEBUG(D_RPCTRACE,"new recovery log "LPX64":%x for index %u of catalog " - LPX64"\n", loghandle->lgh_id.lgl_oid, loghandle->lgh_id.lgl_ogen, - index, cathandle->lgh_id.lgl_oid); - /* build the record for this log in the catalog */ - rec.lid_hdr.lrh_len = sizeof(rec); - rec.lid_hdr.lrh_index = index; - rec.lid_hdr.lrh_type = LLOG_LOGID_MAGIC; - rec.lid_id = loghandle->lgh_id; - rec.lid_tail.lrt_len = sizeof(rec); - rec.lid_tail.lrt_index = index; + CDEBUG(D_RPCTRACE,"new recovery log "DOSTID":%x for index %u of catalog" + DOSTID"\n", POSTID(&loghandle->lgh_id.lgl_oi), + loghandle->lgh_id.lgl_ogen, index, + POSTID(&cathandle->lgh_id.lgl_oi)); + /* build the record for this log in the catalog */ + rec->lid_hdr.lrh_len = sizeof(*rec); + rec->lid_hdr.lrh_index = index; + rec->lid_hdr.lrh_type = LLOG_LOGID_MAGIC; + rec->lid_id = loghandle->lgh_id; + rec->lid_tail.lrt_len = sizeof(*rec); + rec->lid_tail.lrt_index = index; /* update the catalog: header and record */ - rc = llog_write_rec(env, cathandle, &rec.lid_hdr, + rc = llog_write_rec(env, cathandle, &rec->lid_hdr, &loghandle->u.phd.phd_cookie, 1, NULL, index, th); if (rc < 0) GOTO(out_destroy, rc); @@ -147,6 +148,9 @@ out_destroy: * * Assumes caller has already pushed us into the kernel context and is locking. * We return a lock on the handle to ensure nobody yanks it from us. + * + * This takes extra reference on llog_handle via llog_handle_get() and require + * this reference to be put by caller using llog_handle_put() */ int llog_cat_id2handle(const struct lu_env *env, struct llog_handle *cathandle, struct llog_handle **res, struct llog_logid *logid) @@ -159,44 +163,46 @@ int llog_cat_id2handle(const struct lu_env *env, struct llog_handle *cathandle, if (cathandle == NULL) RETURN(-EBADF); - cfs_down_write(&cathandle->lgh_lock); + down_write(&cathandle->lgh_lock); cfs_list_for_each_entry(loghandle, &cathandle->u.chd.chd_head, u.phd.phd_entry) { struct llog_logid *cgl = &loghandle->lgh_id; - if (cgl->lgl_oid == logid->lgl_oid) { + if (ostid_id(&cgl->lgl_oi) == ostid_id(&logid->lgl_oi) && + ostid_seq(&cgl->lgl_oi) == ostid_seq(&logid->lgl_oi)) { if (cgl->lgl_ogen != logid->lgl_ogen) { - CERROR("%s: log "LPX64" generation %x != %x\n", + CERROR("%s: log "DOSTID" generation %x != %x\n", loghandle->lgh_ctxt->loc_obd->obd_name, - logid->lgl_oid, cgl->lgl_ogen, + POSTID(&logid->lgl_oi), cgl->lgl_ogen, logid->lgl_ogen); continue; } loghandle->u.phd.phd_cat_handle = cathandle; - cfs_up_write(&cathandle->lgh_lock); + up_write(&cathandle->lgh_lock); GOTO(out, rc = 0); } } - cfs_up_write(&cathandle->lgh_lock); + up_write(&cathandle->lgh_lock); rc = llog_open(env, cathandle->lgh_ctxt, &loghandle, logid, NULL, LLOG_OPEN_EXISTS); if (rc < 0) { - CERROR("%s: error opening log id "LPX64":%x: rc = %d\n", + CERROR("%s: error opening log id "DOSTID":%x: rc = %d\n", cathandle->lgh_ctxt->loc_obd->obd_name, - logid->lgl_oid, logid->lgl_ogen, rc); - GOTO(out, rc); + POSTID(&logid->lgl_oi), logid->lgl_ogen, rc); + RETURN(rc); } rc = llog_init_handle(env, loghandle, LLOG_F_IS_PLAIN, NULL); if (rc < 0) { llog_close(env, loghandle); - GOTO(out, rc); + loghandle = NULL; + RETURN(rc); } - cfs_down_write(&cathandle->lgh_lock); - cfs_list_add(&loghandle->u.phd.phd_entry, &cathandle->u.chd.chd_head); - cfs_up_write(&cathandle->lgh_lock); + down_write(&cathandle->lgh_lock); + cfs_list_add_tail(&loghandle->u.phd.phd_entry, &cathandle->u.chd.chd_head); + up_write(&cathandle->lgh_lock); loghandle->u.phd.phd_cat_handle = cathandle; loghandle->u.phd.phd_cookie.lgc_lgl = cathandle->lgh_id; @@ -204,8 +210,9 @@ int llog_cat_id2handle(const struct lu_env *env, struct llog_handle *cathandle, loghandle->lgh_hdr->llh_cat_idx; EXIT; out: + llog_handle_get(loghandle); *res = loghandle; - return rc; + return 0; } int llog_cat_close(const struct lu_env *env, struct llog_handle *cathandle) @@ -234,15 +241,7 @@ int llog_cat_close(const struct lu_env *env, struct llog_handle *cathandle) rc); index = loghandle->u.phd.phd_cookie.lgc_index; - - LASSERT(index); - llog_cat_set_first_idx(cathandle, index); - rc = llog_cancel_rec(env, cathandle, index); - if (rc == 0) - CDEBUG(D_RPCTRACE, - "cancel plain log at index %u of " - "catalog "LPX64"\n", - index, cathandle->lgh_id.lgl_oid); + llog_cat_cleanup(env, cathandle, NULL, index); } llog_close(env, loghandle); } @@ -278,39 +277,39 @@ static struct llog_handle *llog_cat_current_log(struct llog_handle *cathandle, struct llog_handle *loghandle = NULL; ENTRY; - cfs_down_read_nested(&cathandle->lgh_lock, LLOGH_CAT); + down_read_nested(&cathandle->lgh_lock, LLOGH_CAT); loghandle = cathandle->u.chd.chd_current_log; if (loghandle) { struct llog_log_hdr *llh; - cfs_down_write_nested(&loghandle->lgh_lock, LLOGH_LOG); + down_write_nested(&loghandle->lgh_lock, LLOGH_LOG); llh = loghandle->lgh_hdr; if (llh == NULL || loghandle->lgh_last_idx < LLOG_BITMAP_SIZE(llh) - 1) { - cfs_up_read(&cathandle->lgh_lock); + up_read(&cathandle->lgh_lock); RETURN(loghandle); } else { - cfs_up_write(&loghandle->lgh_lock); + up_write(&loghandle->lgh_lock); } } - cfs_up_read(&cathandle->lgh_lock); + up_read(&cathandle->lgh_lock); /* time to use next log */ /* first, we have to make sure the state hasn't changed */ - cfs_down_write_nested(&cathandle->lgh_lock, LLOGH_CAT); + down_write_nested(&cathandle->lgh_lock, LLOGH_CAT); loghandle = cathandle->u.chd.chd_current_log; if (loghandle) { struct llog_log_hdr *llh; - cfs_down_write_nested(&loghandle->lgh_lock, LLOGH_LOG); + down_write_nested(&loghandle->lgh_lock, LLOGH_LOG); llh = loghandle->lgh_hdr; LASSERT(llh); if (loghandle->lgh_last_idx < LLOG_BITMAP_SIZE(llh) - 1) { - cfs_up_write(&cathandle->lgh_lock); + up_write(&cathandle->lgh_lock); RETURN(loghandle); } else { - cfs_up_write(&loghandle->lgh_lock); + up_write(&loghandle->lgh_lock); } } @@ -319,8 +318,8 @@ static struct llog_handle *llog_cat_current_log(struct llog_handle *cathandle, loghandle = cathandle->u.chd.chd_next_log; cathandle->u.chd.chd_current_log = loghandle; cathandle->u.chd.chd_next_log = NULL; - cfs_down_write_nested(&loghandle->lgh_lock, LLOGH_LOG); - cfs_up_write(&cathandle->lgh_lock); + down_write_nested(&loghandle->lgh_lock, LLOGH_LOG); + up_write(&cathandle->lgh_lock); LASSERT(loghandle); RETURN(loghandle); } @@ -346,15 +345,16 @@ int llog_cat_add_rec(const struct lu_env *env, struct llog_handle *cathandle, if (!llog_exist(loghandle)) { rc = llog_cat_new_log(env, cathandle, loghandle, th); if (rc < 0) { - cfs_up_write(&loghandle->lgh_lock); + up_write(&loghandle->lgh_lock); RETURN(rc); } } /* now let's try to add the record */ rc = llog_write_rec(env, loghandle, rec, reccookie, 1, buf, -1, th); - if (rc < 0) - CERROR("llog_write_rec %d: lh=%p\n", rc, loghandle); - cfs_up_write(&loghandle->lgh_lock); + if (rc < 0) + CDEBUG_LIMIT(rc == -ENOSPC ? D_HA : D_ERROR, + "llog_write_rec %d: lh=%p\n", rc, loghandle); + up_write(&loghandle->lgh_lock); if (rc == -ENOSPC) { /* try to use next log */ loghandle = llog_cat_current_log(cathandle, th); @@ -363,7 +363,7 @@ int llog_cat_add_rec(const struct lu_env *env, struct llog_handle *cathandle, if (!llog_exist(loghandle)) { rc = llog_cat_new_log(env, cathandle, loghandle, th); if (rc < 0) { - cfs_up_write(&loghandle->lgh_lock); + up_write(&loghandle->lgh_lock); RETURN(rc); } } @@ -372,7 +372,7 @@ int llog_cat_add_rec(const struct lu_env *env, struct llog_handle *cathandle, -1, th); if (rc < 0) CERROR("llog_write_rec %d: lh=%p\n", rc, loghandle); - cfs_up_write(&loghandle->lgh_lock); + up_write(&loghandle->lgh_lock); } RETURN(rc); @@ -383,6 +383,8 @@ int llog_cat_declare_add_rec(const struct lu_env *env, struct llog_handle *cathandle, struct llog_rec_hdr *rec, struct thandle *th) { + struct llog_thread_info *lgi = llog_info(env); + struct llog_logid_rec *lirec = &lgi->lgi_logid; struct llog_handle *loghandle, *next; int rc = 0; @@ -390,7 +392,7 @@ int llog_cat_declare_add_rec(const struct lu_env *env, if (cathandle->u.chd.chd_current_log == NULL) { /* declare new plain llog */ - cfs_down_write(&cathandle->lgh_lock); + down_write(&cathandle->lgh_lock); if (cathandle->u.chd.chd_current_log == NULL) { rc = llog_open(env, cathandle->lgh_ctxt, &loghandle, NULL, NULL, LLOG_OPEN_NEW); @@ -400,10 +402,10 @@ int llog_cat_declare_add_rec(const struct lu_env *env, &cathandle->u.chd.chd_head); } } - cfs_up_write(&cathandle->lgh_lock); + up_write(&cathandle->lgh_lock); } else if (cathandle->u.chd.chd_next_log == NULL) { /* declare next plain llog */ - cfs_down_write(&cathandle->lgh_lock); + down_write(&cathandle->lgh_lock); if (cathandle->u.chd.chd_next_log == NULL) { rc = llog_open(env, cathandle->lgh_ctxt, &loghandle, NULL, NULL, LLOG_OPEN_NEW); @@ -413,17 +415,19 @@ int llog_cat_declare_add_rec(const struct lu_env *env, &cathandle->u.chd.chd_head); } } - cfs_up_write(&cathandle->lgh_lock); + up_write(&cathandle->lgh_lock); } if (rc) GOTO(out, rc); + lirec->lid_hdr.lrh_len = sizeof(*lirec); + if (!llog_exist(cathandle->u.chd.chd_current_log)) { rc = llog_declare_create(env, cathandle->u.chd.chd_current_log, th); if (rc) GOTO(out, rc); - llog_declare_write_rec(env, cathandle, NULL, -1, th); + llog_declare_write_rec(env, cathandle, &lirec->lid_hdr, -1, th); } /* declare records in the llogs */ rc = llog_declare_write_rec(env, cathandle->u.chd.chd_current_log, @@ -435,9 +439,14 @@ int llog_cat_declare_add_rec(const struct lu_env *env, if (next) { if (!llog_exist(next)) { rc = llog_declare_create(env, next, th); - llog_declare_write_rec(env, cathandle, NULL, -1, th); + llog_declare_write_rec(env, cathandle, &lirec->lid_hdr, + -1, th); } - llog_declare_write_rec(env, next, rec, -1, th); + /* XXX: we hope for declarations made for existing llog + * this might be not correct with some backends + * where declarations are expected against specific + * object like ZFS with full debugging enabled */ + /*llog_declare_write_rec(env, next, rec, -1, th);*/ } out: RETURN(rc); @@ -457,31 +466,23 @@ int llog_cat_add(const struct lu_env *env, struct llog_handle *cathandle, LASSERT(ctxt); LASSERT(ctxt->loc_exp); - if (cathandle->lgh_obj != NULL) { - dt = ctxt->loc_exp->exp_obd->obd_lvfs_ctxt.dt; - LASSERT(dt); + LASSERT(cathandle->lgh_obj != NULL); + dt = lu2dt_dev(cathandle->lgh_obj->do_lu.lo_dev); - th = dt_trans_create(env, dt); - if (IS_ERR(th)) - RETURN(PTR_ERR(th)); + th = dt_trans_create(env, dt); + if (IS_ERR(th)) + RETURN(PTR_ERR(th)); - rc = llog_cat_declare_add_rec(env, cathandle, rec, th); - if (rc) - GOTO(out_trans, rc); + rc = llog_cat_declare_add_rec(env, cathandle, rec, th); + if (rc) + GOTO(out_trans, rc); - rc = dt_trans_start_local(env, dt, th); - if (rc) - GOTO(out_trans, rc); - rc = llog_cat_add_rec(env, cathandle, rec, reccookie, buf, th); + rc = dt_trans_start_local(env, dt, th); + if (rc) + GOTO(out_trans, rc); + rc = llog_cat_add_rec(env, cathandle, rec, reccookie, buf, th); out_trans: - dt_trans_stop(env, dt, th); - } else { /* lvfs compat code */ - LASSERT(cathandle->lgh_file != NULL); - rc = llog_cat_declare_add_rec(env, cathandle, rec, th); - if (rc == 0) - rc = llog_cat_add_rec(env, cathandle, rec, reccookie, - buf, th); - } + dt_trans_stop(env, dt, th); RETURN(rc); } EXPORT_SYMBOL(llog_cat_add); @@ -510,26 +511,18 @@ int llog_cat_cancel_records(const struct lu_env *env, rc = llog_cat_id2handle(env, cathandle, &loghandle, lgl); if (rc) { - CERROR("Cannot find log "LPX64"\n", lgl->lgl_oid); - break; + CERROR("%s: cannot find handle for llog "DOSTID": %d\n", + cathandle->lgh_ctxt->loc_obd->obd_name, + POSTID(&lgl->lgl_oi), rc); + failed++; + continue; } lrc = llog_cancel_rec(env, loghandle, cookies->lgc_index); - if (lrc == 1) { /* log has been destroyed */ + if (lrc == LLOG_DEL_PLAIN) { /* log has been destroyed */ index = loghandle->u.phd.phd_cookie.lgc_index; - cfs_down_write(&cathandle->lgh_lock); - if (cathandle->u.chd.chd_current_log == loghandle) - cathandle->u.chd.chd_current_log = NULL; - cfs_up_write(&cathandle->lgh_lock); - llog_close(env, loghandle); - - LASSERT(index); - llog_cat_set_first_idx(cathandle, index); - lrc = llog_cancel_rec(env, cathandle, index); - if (lrc == 0) - CDEBUG(D_RPCTRACE, "cancel plain log at index" - " %u of catalog "LPX64"\n", - index, cathandle->lgh_id.lgl_oid); + rc = llog_cat_cleanup(env, cathandle, loghandle, + index); } else if (lrc == -ENOENT) { if (rc == 0) /* ENOENT shouldn't rewrite any error */ rc = lrc; @@ -537,6 +530,7 @@ int llog_cat_cancel_records(const struct lu_env *env, failed++; rc = lrc; } + llog_handle_put(loghandle); } if (rc) CERROR("%s: fail to cancel %d of %d llog-records: rc = %d\n", @@ -560,36 +554,42 @@ int llog_cat_process_cb(const struct lu_env *env, struct llog_handle *cat_llh, CERROR("invalid record in catalog\n"); RETURN(-EINVAL); } - CDEBUG(D_HA, "processing log "LPX64":%x at index %u of catalog " - LPX64"\n", lir->lid_id.lgl_oid, lir->lid_id.lgl_ogen, - rec->lrh_index, cat_llh->lgh_id.lgl_oid); + CDEBUG(D_HA, "processing log "DOSTID":%x at index %u of catalog " + DOSTID"\n", POSTID(&lir->lid_id.lgl_oi), lir->lid_id.lgl_ogen, + rec->lrh_index, POSTID(&cat_llh->lgh_id.lgl_oi)); rc = llog_cat_id2handle(env, cat_llh, &llh, &lir->lid_id); - if (rc) { - CERROR("Cannot find handle for log "LPX64"\n", - lir->lid_id.lgl_oid); - RETURN(rc); - } - - if (rec->lrh_index < d->lpd_startcat) - /* Skip processing of the logs until startcat */ - RETURN(0); + if (rc) { + CERROR("%s: cannot find handle for llog "DOSTID": %d\n", + cat_llh->lgh_ctxt->loc_obd->obd_name, + POSTID(&lir->lid_id.lgl_oi), rc); + RETURN(rc); + } - if (d->lpd_startidx > 0) { + if (rec->lrh_index < d->lpd_startcat) { + /* Skip processing of the logs until startcat */ + rc = 0; + } else if (d->lpd_startidx > 0) { struct llog_process_cat_data cd; cd.lpcd_first_idx = d->lpd_startidx; cd.lpcd_last_idx = 0; rc = llog_process_or_fork(env, llh, d->lpd_cb, d->lpd_data, &cd, false); - /* Continue processing the next log from idx 0 */ - d->lpd_startidx = 0; - } else { + /* Continue processing the next log from idx 0 */ + d->lpd_startidx = 0; + } else { rc = llog_process_or_fork(env, llh, d->lpd_cb, d->lpd_data, NULL, false); - } + } - RETURN(rc); + /* The empty plain log was destroyed while processing */ + if (rc == LLOG_DEL_PLAIN) + rc = llog_cat_cleanup(env, cat_llh, llh, + llh->u.phd.phd_cookie.lgc_index); + llog_handle_put(llh); + + RETURN(rc); } int llog_cat_process_or_fork(const struct lu_env *env, @@ -611,18 +611,18 @@ int llog_cat_process_or_fork(const struct lu_env *env, if (llh->llh_cat_idx > cat_llh->lgh_last_idx) { struct llog_process_cat_data cd; - CWARN("catlog "LPX64" crosses index zero\n", - cat_llh->lgh_id.lgl_oid); + CWARN("catlog "DOSTID" crosses index zero\n", + POSTID(&cat_llh->lgh_id.lgl_oi)); cd.lpcd_first_idx = llh->llh_cat_idx; cd.lpcd_last_idx = 0; rc = llog_process_or_fork(env, cat_llh, llog_cat_process_cb, &d, &cd, fork); - if (rc != 0) - RETURN(rc); + if (rc != 0) + RETURN(rc); - cd.lpcd_first_idx = 0; - cd.lpcd_last_idx = cat_llh->lgh_last_idx; + cd.lpcd_first_idx = 0; + cd.lpcd_last_idx = cat_llh->lgh_last_idx; rc = llog_process_or_fork(env, cat_llh, llog_cat_process_cb, &d, &cd, fork); } else { @@ -638,102 +638,44 @@ int llog_cat_process(const struct lu_env *env, struct llog_handle *cat_llh, llog_cb_t cb, void *data, int startcat, int startidx) { return llog_cat_process_or_fork(env, cat_llh, cb, data, startcat, - startidx, 0); + startidx, false); } EXPORT_SYMBOL(llog_cat_process); -#ifdef __KERNEL__ -int llog_cat_process_thread(void *data) -{ - struct llog_process_cat_args *args = data; - struct llog_ctxt *ctxt = args->lpca_ctxt; - struct llog_handle *llh = NULL; - llog_cb_t cb = args->lpca_cb; - struct llog_thread_info *lgi; - struct lu_env env; - int rc; - ENTRY; - - cfs_daemonize_ctxt("ll_log_process"); - - rc = lu_env_init(&env, LCT_LOCAL); - if (rc) - GOTO(out, rc); - lgi = llog_info(&env); - LASSERT(lgi); - - lgi->lgi_logid = *(struct llog_logid *)(args->lpca_arg); - rc = llog_open(&env, ctxt, &llh, &lgi->lgi_logid, NULL, - LLOG_OPEN_EXISTS); - if (rc) { - CERROR("%s: cannot open llog "LPX64":%x: rc = %d\n", - ctxt->loc_obd->obd_name, lgi->lgi_logid.lgl_oid, - lgi->lgi_logid.lgl_ogen, rc); - GOTO(out_env, rc); - } - rc = llog_init_handle(&env, llh, LLOG_F_IS_CAT, NULL); - if (rc) { - CERROR("%s: llog_init_handle failed: rc = %d\n", - llh->lgh_ctxt->loc_obd->obd_name, rc); - GOTO(release_llh, rc); - } - - if (cb) { - rc = llog_cat_process(&env, llh, cb, NULL, 0, 0); - if (rc != LLOG_PROC_BREAK && rc != 0) - CERROR("%s: llog_cat_process() failed: rc = %d\n", - llh->lgh_ctxt->loc_obd->obd_name, rc); - cb(&env, llh, NULL, NULL); - } else { - CWARN("No callback function for recovery\n"); - } - - /* - * Make sure that all cached data is sent. - */ - llog_sync(ctxt, NULL, 0); - GOTO(release_llh, rc); -release_llh: - rc = llog_cat_close(&env, llh); - if (rc) - CERROR("%s: llog_cat_close() failed: rc = %d\n", - llh->lgh_ctxt->loc_obd->obd_name, rc); -out_env: - lu_env_fini(&env); -out: - llog_ctxt_put(ctxt); - OBD_FREE_PTR(args); - return rc; -} -EXPORT_SYMBOL(llog_cat_process_thread); -#endif - static int llog_cat_reverse_process_cb(const struct lu_env *env, struct llog_handle *cat_llh, struct llog_rec_hdr *rec, void *data) { - struct llog_process_data *d = data; - struct llog_logid_rec *lir = (struct llog_logid_rec *)rec; - struct llog_handle *llh; - int rc; + struct llog_process_data *d = data; + struct llog_logid_rec *lir = (struct llog_logid_rec *)rec; + struct llog_handle *llh; + int rc; - if (le32_to_cpu(rec->lrh_type) != LLOG_LOGID_MAGIC) { - CERROR("invalid record in catalog\n"); - RETURN(-EINVAL); - } - CDEBUG(D_HA, "processing log "LPX64":%x at index %u of catalog " - LPX64"\n", lir->lid_id.lgl_oid, lir->lid_id.lgl_ogen, - le32_to_cpu(rec->lrh_index), cat_llh->lgh_id.lgl_oid); + if (le32_to_cpu(rec->lrh_type) != LLOG_LOGID_MAGIC) { + CERROR("invalid record in catalog\n"); + RETURN(-EINVAL); + } + CDEBUG(D_HA, "processing log "DOSTID":%x at index %u of catalog " + DOSTID"\n", POSTID(&lir->lid_id.lgl_oi), lir->lid_id.lgl_ogen, + le32_to_cpu(rec->lrh_index), POSTID(&cat_llh->lgh_id.lgl_oi)); rc = llog_cat_id2handle(env, cat_llh, &llh, &lir->lid_id); - if (rc) { - CERROR("Cannot find handle for log "LPX64"\n", - lir->lid_id.lgl_oid); - RETURN(rc); - } + if (rc) { + CERROR("%s: cannot find handle for llog "DOSTID": %d\n", + cat_llh->lgh_ctxt->loc_obd->obd_name, + POSTID(&lir->lid_id.lgl_oi), rc); + RETURN(rc); + } rc = llog_reverse_process(env, llh, d->lpd_cb, d->lpd_data, NULL); - RETURN(rc); + + /* The empty plain was destroyed while processing */ + if (rc == LLOG_DEL_PLAIN) + rc = llog_cat_cleanup(env, cat_llh, llh, + llh->u.phd.phd_cookie.lgc_index); + + llog_handle_put(llh); + RETURN(rc); } int llog_cat_reverse_process(const struct lu_env *env, @@ -750,20 +692,20 @@ int llog_cat_reverse_process(const struct lu_env *env, d.lpd_data = data; d.lpd_cb = cb; - if (llh->llh_cat_idx > cat_llh->lgh_last_idx) { - CWARN("catalog "LPX64" crosses index zero\n", - cat_llh->lgh_id.lgl_oid); + if (llh->llh_cat_idx > cat_llh->lgh_last_idx) { + CWARN("catalog "DOSTID" crosses index zero\n", + POSTID(&cat_llh->lgh_id.lgl_oi)); cd.lpcd_first_idx = 0; cd.lpcd_last_idx = cat_llh->lgh_last_idx; rc = llog_reverse_process(env, cat_llh, llog_cat_reverse_process_cb, &d, &cd); - if (rc != 0) - RETURN(rc); + if (rc != 0) + RETURN(rc); - cd.lpcd_first_idx = le32_to_cpu(llh->llh_cat_idx); - cd.lpcd_last_idx = 0; + cd.lpcd_first_idx = le32_to_cpu(llh->llh_cat_idx); + cd.lpcd_last_idx = 0; rc = llog_reverse_process(env, cat_llh, llog_cat_reverse_process_cb, &d, &cd); @@ -802,41 +744,69 @@ int llog_cat_set_first_idx(struct llog_handle *cathandle, int index) } } out: - CDEBUG(D_RPCTRACE, "set catlog "LPX64" first idx %u\n", - cathandle->lgh_id.lgl_oid, llh->llh_cat_idx); - } + CDEBUG(D_RPCTRACE, "set catlog "DOSTID" first idx %u\n", + POSTID(&cathandle->lgh_id.lgl_oi), llh->llh_cat_idx); + } - RETURN(0); + RETURN(0); +} + +/* Cleanup deleted plain llog traces from catalog */ +int llog_cat_cleanup(const struct lu_env *env, struct llog_handle *cathandle, + struct llog_handle *loghandle, int index) +{ + int rc; + + LASSERT(index); + if (loghandle != NULL) { + /* remove destroyed llog from catalog list and + * chd_current_log variable */ + down_write(&cathandle->lgh_lock); + if (cathandle->u.chd.chd_current_log == loghandle) + cathandle->u.chd.chd_current_log = NULL; + cfs_list_del_init(&loghandle->u.phd.phd_entry); + up_write(&cathandle->lgh_lock); + LASSERT(index == loghandle->u.phd.phd_cookie.lgc_index); + /* llog was opened and keep in a list, close it now */ + llog_close(env, loghandle); + } + /* remove plain llog entry from catalog by index */ + llog_cat_set_first_idx(cathandle, index); + rc = llog_cancel_rec(env, cathandle, index); + if (rc == 0) + CDEBUG(D_HA, "cancel plain log at index" + " %u of catalog "DOSTID"\n", + index, POSTID(&cathandle->lgh_id.lgl_oi)); + return rc; } -/* callback func for llog_process in llog_obd_origin_setup */ int cat_cancel_cb(const struct lu_env *env, struct llog_handle *cathandle, struct llog_rec_hdr *rec, void *data) { struct llog_logid_rec *lir = (struct llog_logid_rec *)rec; struct llog_handle *loghandle; struct llog_log_hdr *llh; - int rc, index; + int rc; ENTRY; if (rec->lrh_type != LLOG_LOGID_MAGIC) { - CERROR("%s: invalid record in catalog\n", - loghandle->lgh_ctxt->loc_obd->obd_name); + CERROR("invalid record in catalog\n"); RETURN(-EINVAL); } - CDEBUG(D_HA, "processing log "LPX64":%x at index %u of catalog " - LPX64"\n", lir->lid_id.lgl_oid, lir->lid_id.lgl_ogen, - rec->lrh_index, cathandle->lgh_id.lgl_oid); + + CDEBUG(D_HA, "processing log "DOSTID":%x at index %u of catalog " + DOSTID"\n", POSTID(&lir->lid_id.lgl_oi), lir->lid_id.lgl_ogen, + rec->lrh_index, POSTID(&cathandle->lgh_id.lgl_oi)); rc = llog_cat_id2handle(env, cathandle, &loghandle, &lir->lid_id); if (rc) { - CERROR("%s: cannot find handle for llog "LPX64"\n", - loghandle->lgh_ctxt->loc_obd->obd_name, - lir->lid_id.lgl_oid); - if (rc == -ENOENT) { - index = rec->lrh_index; - goto cat_cleanup; + CERROR("%s: cannot find handle for llog "DOSTID": %d\n", + cathandle->lgh_ctxt->loc_obd->obd_name, + POSTID(&lir->lid_id.lgl_oi), rc); + if (rc == -ENOENT || rc == -ESTALE) { + /* remove index from catalog */ + llog_cat_cleanup(env, cathandle, NULL, rec->lrh_index); } RETURN(rc); } @@ -849,20 +819,30 @@ int cat_cancel_cb(const struct lu_env *env, struct llog_handle *cathandle, CERROR("%s: fail to destroy empty log: rc = %d\n", loghandle->lgh_ctxt->loc_obd->obd_name, rc); - index = loghandle->u.phd.phd_cookie.lgc_index; - llog_close(env, loghandle); - -cat_cleanup: - LASSERT(index); - llog_cat_set_first_idx(cathandle, index); - rc = llog_cancel_rec(env, cathandle, index); - if (rc == 0) - CDEBUG(D_HA, - "cancel log "LPX64":%x at index %u of catalog " - LPX64"\n", lir->lid_id.lgl_oid, - lir->lid_id.lgl_ogen, rec->lrh_index, - cathandle->lgh_id.lgl_oid); + llog_cat_cleanup(env, cathandle, loghandle, + loghandle->u.phd.phd_cookie.lgc_index); } + llog_handle_put(loghandle); RETURN(rc); } +EXPORT_SYMBOL(cat_cancel_cb); + +/* helper to initialize catalog llog and process it to cancel */ +int llog_cat_init_and_process(const struct lu_env *env, + struct llog_handle *llh) +{ + int rc; + + rc = llog_init_handle(env, llh, LLOG_F_IS_CAT, NULL); + if (rc) + RETURN(rc); + + rc = llog_process_or_fork(env, llh, cat_cancel_cb, NULL, NULL, false); + if (rc) + CERROR("%s: llog_process() with cat_cancel_cb failed: rc = " + "%d\n", llh->lgh_ctxt->loc_obd->obd_name, rc); + RETURN(0); +} +EXPORT_SYMBOL(llog_cat_init_and_process); +