X-Git-Url: https://git.whamcloud.com/?p=fs%2Flustre-release.git;a=blobdiff_plain;f=lustre%2Fofd%2Fofd_dev.c;h=3166fb744e0ee0f8f4c6401b4961dcce3311b65e;hp=cb58b99ef005fd90b992e9be422fcb469131ca8b;hb=e5f552b70dccbd2fdf21ec7b7053a01bcbe062c2;hpb=342c14842fa996181e5d5764701585b1b2f6d6cb diff --git a/lustre/ofd/ofd_dev.c b/lustre/ofd/ofd_dev.c index cb58b99..3166fb7 100644 --- a/lustre/ofd/ofd_dev.c +++ b/lustre/ofd/ofd_dev.c @@ -362,7 +362,7 @@ static struct lu_object *ofd_object_alloc(const struct lu_env *env, ENTRY; - OBD_SLAB_ALLOC_PTR_GFP(of, ofd_object_kmem, __GFP_IO); + OBD_SLAB_ALLOC_PTR_GFP(of, ofd_object_kmem, GFP_NOFS); if (of != NULL) { struct lu_object *o; struct lu_object_header *h; @@ -381,6 +381,38 @@ static struct lu_object *ofd_object_alloc(const struct lu_env *env, extern int ost_handle(struct ptlrpc_request *req); +static int ofd_lfsck_out_notify(const struct lu_env *env, void *data, + enum lfsck_events event) +{ + struct ofd_device *ofd = data; + struct obd_device *obd = ofd_obd(ofd); + + switch (event) { + case LE_LASTID_REBUILDING: + CWARN("%s: Found crashed LAST_ID, deny creating new OST-object " + "on the device until the LAST_ID rebuilt successfully.\n", + obd->obd_name); + down_write(&ofd->ofd_lastid_rwsem); + ofd->ofd_lastid_rebuilding = 1; + up_write(&ofd->ofd_lastid_rwsem); + break; + case LE_LASTID_REBUILT: { + down_write(&ofd->ofd_lastid_rwsem); + ofd_seqs_free(env, ofd); + ofd->ofd_lastid_rebuilding = 0; + ofd->ofd_lastid_gen++; + up_write(&ofd->ofd_lastid_rwsem); + break; + } + default: + CERROR("%s: unknown lfsck event: rc = %d\n", + ofd_obd(ofd)->obd_name, event); + return -EINVAL; + } + + return 0; +} + static int ofd_prepare(const struct lu_env *env, struct lu_device *pdev, struct lu_device *dev) { @@ -393,12 +425,6 @@ static int ofd_prepare(const struct lu_env *env, struct lu_device *pdev, ENTRY; - rc = lu_env_refill((struct lu_env *)env); - if (rc != 0) { - CERROR("Failure to refill session: '%d'\n", rc); - RETURN(rc); - } - info = ofd_info_init(env, NULL); if (info == NULL) RETURN(-EFAULT); @@ -408,15 +434,21 @@ static int ofd_prepare(const struct lu_env *env, struct lu_device *pdev, if (rc != 0) RETURN(rc); - rc = lfsck_register(env, ofd->ofd_osd, &ofd->ofd_dt_dev, false); + rc = lfsck_register(env, ofd->ofd_osd, ofd->ofd_osd, obd, + ofd_lfsck_out_notify, ofd, false); if (rc != 0) { CERROR("%s: failed to initialize lfsck: rc = %d\n", obd->obd_name, rc); RETURN(rc); } + rc = lfsck_register_namespace(env, ofd->ofd_osd, ofd->ofd_namespace); + /* The LFSCK instance is registered just now, so it must be there when + * register the namespace to such instance. */ + LASSERTF(rc == 0, "register namespace failed: rc = %d\n", rc); + lsp.lsp_start = NULL; - lsp.lsp_namespace = ofd->ofd_namespace; + lsp.lsp_index_valid = 0; rc = lfsck_start(env, ofd->ofd_osd, &lsp); if (rc != 0) { CWARN("%s: auto trigger paused LFSCK failed: rc = %d\n", @@ -721,8 +753,8 @@ int ofd_set_info_hdl(struct tgt_session_info *tsi) RETURN(rc); } -static int ofd_fiemap_get(const struct lu_env *env, struct ofd_device *ofd, - struct lu_fid *fid, struct ll_user_fiemap *fiemap) +int ofd_fiemap_get(const struct lu_env *env, struct ofd_device *ofd, + struct lu_fid *fid, struct ll_user_fiemap *fiemap) { struct ofd_object *fo; int rc; @@ -863,7 +895,7 @@ int ofd_get_info_hdl(struct tgt_session_info *tsi) } else if (KEY_IS(KEY_FIEMAP)) { struct ll_fiemap_info_key *fm_key; struct ll_user_fiemap *fiemap; - struct lu_fid *fid = &fti->fti_fid; + struct lu_fid *fid; req_capsule_extend(tsi->tsi_pill, &RQF_OST_GET_INFO_FIEMAP); @@ -872,7 +904,7 @@ int ofd_get_info_hdl(struct tgt_session_info *tsi) if (rc) RETURN(err_serious(rc)); - *fid = fm_key->oa.o_oi.oi_fid; + fid = &fm_key->oa.o_oi.oi_fid; CDEBUG(D_INODE, "get FIEMAP of object "DFID"\n", PFID(fid)); @@ -1109,12 +1141,94 @@ out: return rc; } +static int ofd_orphans_destroy(const struct lu_env *env, + struct obd_export *exp, + struct ofd_device *ofd, struct obdo *oa) +{ + struct ofd_thread_info *info = ofd_info(env); + struct lu_fid *fid = &info->fti_fid; + struct ost_id *oi = &oa->o_oi; + struct ofd_seq *oseq; + obd_seq seq = ostid_seq(oi); + obd_id end_id = ostid_id(oi); + obd_id last; + obd_id oid; + int skip_orphan; + int rc = 0; + + ENTRY; + + oseq = ofd_seq_get(ofd, seq); + if (oseq == NULL) { + CERROR("%s: Can not find seq for "DOSTID"\n", + ofd_name(ofd), POSTID(oi)); + RETURN(-EINVAL); + } + + *fid = oi->oi_fid; + last = ofd_seq_last_oid(oseq); + oid = last; + + LASSERT(exp != NULL); + skip_orphan = !!(exp_connect_flags(exp) & OBD_CONNECT_SKIP_ORPHAN); + + if (OBD_FAIL_CHECK(OBD_FAIL_OST_NODESTROY)) + goto done; + + LCONSOLE(D_INFO, "%s: deleting orphan objects from "DOSTID + " to "DOSTID"\n", ofd_name(ofd), seq, end_id + 1, seq, last); + + while (oid > end_id) { + rc = fid_set_id(fid, oid); + if (unlikely(rc != 0)) + GOTO(out_put, rc); + + rc = ofd_destroy_by_fid(env, ofd, fid, 1); + if (rc != 0 && rc != -ENOENT && rc != -ESTALE && + likely(rc != -EREMCHG && rc != -EINPROGRESS)) + /* this is pretty fatal... */ + CEMERG("%s: error destroying precreated id " + DFID": rc = %d\n", + ofd_name(ofd), PFID(fid), rc); + + oid--; + if (!skip_orphan) { + ofd_seq_last_oid_set(oseq, oid); + /* update last_id on disk periodically so that if we + * restart * we don't need to re-scan all of the just + * deleted objects. */ + if ((oid & 511) == 0) + ofd_seq_last_oid_write(env, ofd, oseq); + } + } + + CDEBUG(D_HA, "%s: after destroy: set last_id to "DOSTID"\n", + ofd_name(ofd), seq, oid); + +done: + if (!skip_orphan) { + ofd_seq_last_oid_set(oseq, oid); + rc = ofd_seq_last_oid_write(env, ofd, oseq); + } else { + /* don't reuse orphan object, return last used objid */ + ostid_set_id(oi, last); + rc = 0; + } + + GOTO(out_put, rc); + +out_put: + ofd_seq_put(env, oseq); + return rc; +} + static int ofd_create_hdl(struct tgt_session_info *tsi) { struct ost_body *repbody; const struct obdo *oa = &tsi->tsi_ost_body->oa; struct obdo *rep_oa; - struct ofd_device *ofd = ofd_exp(tsi->tsi_exp); + struct obd_export *exp = tsi->tsi_exp; + struct ofd_device *ofd = ofd_exp(exp); obd_seq seq = ostid_seq(&oa->o_oi); obd_id oid = ostid_id(&oa->o_oi); struct ofd_seq *oseq; @@ -1130,6 +1244,13 @@ static int ofd_create_hdl(struct tgt_session_info *tsi) if (repbody == NULL) RETURN(-ENOMEM); + down_read(&ofd->ofd_lastid_rwsem); + /* Currently, for safe, we do not distinguish which LAST_ID is broken, + * we may do that in the future. + * Return -ENOSPC until the LAST_ID rebuilt. */ + if (unlikely(ofd->ofd_lastid_rebuilding)) + GOTO(out_sem, rc = -ENOSPC); + rep_oa = &repbody->oa; rep_oa->o_oi = oa->o_oi; @@ -1142,7 +1263,7 @@ static int ofd_create_hdl(struct tgt_session_info *tsi) if (IS_ERR(oseq)) { CERROR("%s: Can't find FID Sequence "LPX64": rc = %ld\n", ofd_name(ofd), seq, PTR_ERR(oseq)); - RETURN(-EINVAL); + GOTO(out_sem, rc = -EINVAL); } if ((oa->o_valid & OBD_MD_FLFLAGS) && @@ -1161,9 +1282,11 @@ static int ofd_create_hdl(struct tgt_session_info *tsi) /* former ofd_handle_precreate */ if ((oa->o_valid & OBD_MD_FLFLAGS) && (oa->o_flags & OBD_FL_DELORPHAN)) { + exp->exp_filter_data.fed_lastid_gen = ofd->ofd_lastid_gen; + /* destroy orphans */ if (lustre_msg_get_conn_cnt(tgt_ses_req(tsi)->rq_reqmsg) < - tsi->tsi_exp->exp_conn_cnt) { + exp->exp_conn_cnt) { CERROR("%s: dropping old orphan cleanup request\n", ofd_name(ofd)); GOTO(out_nolock, rc = 0); @@ -1184,7 +1307,7 @@ static int ofd_create_hdl(struct tgt_session_info *tsi) /* FIXME: should reset precreate_next_id on MDS */ rc = 0; } else if (diff < 0) { - rc = ofd_orphans_destroy(tsi->tsi_env, tsi->tsi_exp, + rc = ofd_orphans_destroy(tsi->tsi_env, exp, ofd, rep_oa); oseq->os_destroys_in_progress = 0; } else { @@ -1192,9 +1315,15 @@ static int ofd_create_hdl(struct tgt_session_info *tsi) oseq->os_destroys_in_progress = 0; } } else { + if (unlikely(exp->exp_filter_data.fed_lastid_gen != + ofd->ofd_lastid_gen)) { + ofd_obd_disconnect(exp); + GOTO(out_nolock, rc = -ENOTCONN); + } + mutex_lock(&oseq->os_create_lock); if (lustre_msg_get_conn_cnt(tgt_ses_req(tsi)->rq_reqmsg) < - tsi->tsi_exp->exp_conn_cnt) { + exp->exp_conn_cnt) { CERROR("%s: dropping old precreate request\n", ofd_name(ofd)); GOTO(out, rc = 0); @@ -1299,7 +1428,7 @@ static int ofd_create_hdl(struct tgt_session_info *tsi) ostid_set_id(&rep_oa->o_oi, ofd_seq_last_oid(oseq)); } EXIT; - ofd_counter_incr(tsi->tsi_exp, LPROC_OFD_STATS_CREATE, + ofd_counter_incr(exp, LPROC_OFD_STATS_CREATE, tsi->tsi_jobid, 1); out: mutex_unlock(&oseq->os_create_lock); @@ -1308,6 +1437,9 @@ out_nolock: rep_oa->o_valid |= OBD_MD_FLID | OBD_MD_FLGROUP; ofd_seq_put(tsi->tsi_env, oseq); + +out_sem: + up_read(&ofd->ofd_lastid_rwsem); return rc; } @@ -1532,7 +1664,7 @@ static int ofd_punch_hdl(struct tgt_session_info *tsi) } rc = ofd_object_punch(tsi->tsi_env, fo, start, end, &info->fti_attr, - ff); + ff, (struct obdo *)oa); if (rc) GOTO(out_put, rc); @@ -1939,8 +2071,8 @@ static struct tgt_opc_slice ofd_common_slice[] = { .tos_hs = tgt_dlm_handlers }, { - .tos_opc_start = UPDATE_OBJ, - .tos_opc_end = UPDATE_LAST_OPC, + .tos_opc_start = OUT_UPDATE_FIRST_OPC, + .tos_opc_end = OUT_UPDATE_LAST_OPC, .tos_hs = tgt_out_handlers }, { @@ -1949,6 +2081,11 @@ static struct tgt_opc_slice ofd_common_slice[] = { .tos_hs = seq_handlers }, { + .tos_opc_start = LFSCK_FIRST_OPC, + .tos_opc_end = LFSCK_LAST_OPC, + .tos_hs = tgt_lfsck_handlers + }, + { .tos_hs = NULL } }; @@ -1999,10 +2136,14 @@ static int ofd_init0(const struct lu_env *env, struct ofd_device *m, m->ofd_tot_granted = 0; m->ofd_tot_pending = 0; m->ofd_seq_count = 0; + init_waitqueue_head(&m->ofd_inconsistency_thread.t_ctl_waitq); + INIT_LIST_HEAD(&m->ofd_inconsistency_list); + spin_lock_init(&m->ofd_inconsistency_lock); spin_lock_init(&m->ofd_batch_lock); rwlock_init(&obd->u.filter.fo_sptlrpc_lock); sptlrpc_rule_set_init(&obd->u.filter.fo_sptlrpc_rset); + init_rwsem(&m->ofd_lastid_rwsem); obd->u.filter.fo_fl_oss_capa = 0; CFS_INIT_LIST_HEAD(&obd->u.filter.fo_capa_keys); @@ -2102,7 +2243,14 @@ static int ofd_init0(const struct lu_env *env, struct ofd_device *m, if (rc) GOTO(err_fini_lut, rc); + rc = ofd_start_inconsistency_verification_thread(m); + if (rc != 0) + GOTO(err_fini_fs, rc); + RETURN(0); + +err_fini_fs: + ofd_fs_cleanup(env, m); err_fini_lut: tgt_fini(env, &m->ofd_lut); err_free_ns: @@ -2117,16 +2265,20 @@ err_fini_proc: static void ofd_fini(const struct lu_env *env, struct ofd_device *m) { - struct obd_device *obd = ofd_obd(m); - struct lu_device *d = &m->ofd_dt_dev.dd_lu_dev; + struct obd_device *obd = ofd_obd(m); + struct lu_device *d = &m->ofd_dt_dev.dd_lu_dev; + struct lfsck_stop stop; - lfsck_stop(env, m->ofd_osd, true); - lfsck_degister(env, m->ofd_osd); + stop.ls_status = LS_PAUSED; + stop.ls_flags = 0; + lfsck_stop(env, m->ofd_osd, &stop); target_recovery_fini(obd); obd_exports_barrier(obd); obd_zombie_barrier(); tgt_fini(env, &m->ofd_lut); + ofd_stop_inconsistency_verification_thread(m); + lfsck_degister(env, m->ofd_osd); ofd_fs_cleanup(env, m); ofd_free_capa_keys(m);