X-Git-Url: https://git.whamcloud.com/?p=fs%2Flustre-release.git;a=blobdiff_plain;f=lustre%2Fofd%2Fofd_dev.c;h=3166fb744e0ee0f8f4c6401b4961dcce3311b65e;hp=91e28b49d7e317f9de22f21fcea6c6c75f9ae362;hb=e5f552b70dccbd2fdf21ec7b7053a01bcbe062c2;hpb=b1491d26271f074dc6f99cde037403337c0b2151 diff --git a/lustre/ofd/ofd_dev.c b/lustre/ofd/ofd_dev.c index 91e28b4..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,6 +904,10 @@ int ofd_get_info_hdl(struct tgt_session_info *tsi) if (rc) RETURN(err_serious(rc)); + fid = &fm_key->oa.o_oi.oi_fid; + + CDEBUG(D_INODE, "get FIEMAP of object "DFID"\n", PFID(fid)); + replylen = fiemap_count_to_size(fm_key->fiemap.fm_extent_count); req_capsule_set_size(tsi->tsi_pill, &RMF_FIEMAP_VAL, RCL_SERVER, replylen); @@ -884,12 +920,6 @@ int ofd_get_info_hdl(struct tgt_session_info *tsi) if (fiemap == NULL) RETURN(-ENOMEM); - rc = ostid_to_fid(fid, &fm_key->oa.o_oi, 0); - if (rc != 0) - RETURN(rc); - - CDEBUG(D_INODE, "get FIEMAP of object "DFID"\n", PFID(fid)); - *fiemap = fm_key->fiemap; rc = ofd_fiemap_get(tsi->tsi_env, ofd, fid, fiemap); @@ -1111,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; @@ -1132,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; @@ -1144,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) && @@ -1163,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); @@ -1186,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 { @@ -1194,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); @@ -1301,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); @@ -1310,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; } @@ -1319,6 +1449,8 @@ static int ofd_destroy_hdl(struct tgt_session_info *tsi) struct ost_body *repbody; struct ofd_device *ofd = ofd_exp(tsi->tsi_exp); struct ofd_thread_info *fti = tsi2ofd_info(tsi); + struct lu_fid *fid = &fti->fti_fid; + obd_id oid; obd_count count; int rc = 0; @@ -1339,8 +1471,11 @@ static int ofd_destroy_hdl(struct tgt_session_info *tsi) ldlm_request_cancel(tgt_ses_req(tsi), dlm, 0); } + *fid = body->oa.o_oi.oi_fid; + oid = ostid_id(&body->oa.o_oi); + LASSERT(oid != 0); + repbody = req_capsule_server_get(tsi->tsi_pill, &RMF_OST_BODY); - repbody->oa.o_oi = body->oa.o_oi; /* check that o_misc makes sense */ if (body->oa.o_valid & OBD_MD_FLOBJCOUNT) @@ -1350,37 +1485,39 @@ static int ofd_destroy_hdl(struct tgt_session_info *tsi) CDEBUG(D_HA, "%s: Destroy object "DOSTID" count %d\n", ofd_name(ofd), POSTID(&body->oa.o_oi), count); + while (count > 0) { int lrc; - lrc = ostid_to_fid(&fti->fti_fid, &repbody->oa.o_oi, 0); - if (lrc != 0) { - if (rc == 0) - rc = lrc; - GOTO(out, rc); - } - lrc = ofd_destroy_by_fid(tsi->tsi_env, ofd, &fti->fti_fid, 0); + lrc = ofd_destroy_by_fid(tsi->tsi_env, ofd, fid, 0); if (lrc == -ENOENT) { CDEBUG(D_INODE, "%s: destroying non-existent object "DFID"\n", - ofd_name(ofd), PFID(&fti->fti_fid)); + ofd_name(ofd), PFID(fid)); /* rewrite rc with -ENOENT only if it is 0 */ if (rc == 0) rc = lrc; } else if (lrc != 0) { CERROR("%s: error destroying object "DFID": %d\n", - ofd_name(ofd), PFID(&fti->fti_fid), - rc); + ofd_name(ofd), PFID(fid), lrc); rc = lrc; } + count--; - ostid_inc_id(&repbody->oa.o_oi); + oid++; + lrc = fid_set_id(fid, oid); + if (unlikely(lrc != 0 && count > 0)) + GOTO(out, rc = lrc); } ofd_counter_incr(tsi->tsi_exp, LPROC_OFD_STATS_DESTROY, tsi->tsi_jobid, 1); + + GOTO(out, rc); + out: - RETURN(rc); + fid_to_ostid(fid, &repbody->oa.o_oi); + return rc; } static int ofd_statfs_hdl(struct tgt_session_info *tsi) @@ -1527,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); @@ -1555,7 +1692,6 @@ out: return rc; } - static int ofd_quotactl(struct tgt_session_info *tsi) { struct obd_quotactl *oqctl, *repoqc; @@ -1585,6 +1721,302 @@ static int ofd_quotactl(struct tgt_session_info *tsi) RETURN(rc); } +/* High priority request handlers for OFD */ + +/* prolong locks for the current service time of the corresponding + * portal (= OST_IO_PORTAL) + */ +static inline int prolong_timeout(struct ptlrpc_request *req) +{ + struct ptlrpc_service_part *svcpt = req->rq_rqbd->rqbd_svcpt; + + if (AT_OFF) + return obd_timeout / 2; + + return max(at_est2timeout(at_get(&svcpt->scp_at_estimate)), + ldlm_timeout); +} + +static int ofd_prolong_one_lock(struct tgt_session_info *tsi, + struct ldlm_lock *lock, + struct ldlm_extent *extent, int timeout) +{ + + if (lock->l_flags & LDLM_FL_DESTROYED) /* lock already cancelled */ + return 0; + + /* XXX: never try to grab resource lock here because we're inside + * exp_bl_list_lock; in ldlm_lockd.c to handle waiting list we take + * res lock and then exp_bl_list_lock. */ + + if (!(lock->l_flags & LDLM_FL_AST_SENT)) + /* ignore locks not being cancelled */ + return 0; + + LDLM_DEBUG(lock, "refreshed for req x"LPU64" ext("LPU64"->"LPU64") " + "to %ds.\n", tgt_ses_req(tsi)->rq_xid, extent->start, + extent->end, timeout); + + /* OK. this is a possible lock the user holds doing I/O + * let's refresh eviction timer for it */ + ldlm_refresh_waiting_lock(lock, timeout); + return 1; +} + +static int ofd_prolong_extent_locks(struct tgt_session_info *tsi, + __u64 start, __u64 end) +{ + struct obd_export *exp = tsi->tsi_exp; + struct obdo *oa = &tsi->tsi_ost_body->oa; + struct ldlm_extent extent = { + .start = start, + .end = end + }; + struct ldlm_lock *lock; + int timeout = prolong_timeout(tgt_ses_req(tsi)); + int lock_count = 0; + + ENTRY; + + if (oa->o_valid & OBD_MD_FLHANDLE) { + /* mostly a request should be covered by only one lock, try + * fast path. */ + lock = ldlm_handle2lock(&oa->o_handle); + if (lock != NULL) { + /* Fast path to check if the lock covers the whole IO + * region exclusively. */ + if (lock->l_granted_mode == LCK_PW && + ldlm_extent_contain(&lock->l_policy_data.l_extent, + &extent)) { + /* bingo */ + LASSERT(lock->l_export == exp); + lock_count = ofd_prolong_one_lock(tsi, lock, + &extent, timeout); + LDLM_LOCK_PUT(lock); + RETURN(lock_count); + } + LDLM_LOCK_PUT(lock); + } + } + + spin_lock_bh(&exp->exp_bl_list_lock); + list_for_each_entry(lock, &exp->exp_bl_list, l_exp_list) { + LASSERT(lock->l_flags & LDLM_FL_AST_SENT); + LASSERT(lock->l_resource->lr_type == LDLM_EXTENT); + + if (!ldlm_res_eq(&tsi->tsi_resid, &lock->l_resource->lr_name)) + continue; + + if (!ldlm_extent_overlap(&lock->l_policy_data.l_extent, + &extent)) + continue; + + lock_count += ofd_prolong_one_lock(tsi, lock, &extent, timeout); + } + spin_unlock_bh(&exp->exp_bl_list_lock); + + RETURN(lock_count); +} + +/** + * Returns 1 if the given PTLRPC matches the given LDLM lock, or 0 if it does + * not. + */ +static int ofd_rw_hpreq_lock_match(struct ptlrpc_request *req, + struct ldlm_lock *lock) +{ + struct niobuf_remote *rnb; + struct obd_ioobj *ioo; + ldlm_mode_t mode; + struct ldlm_extent ext; + __u32 opc = lustre_msg_get_opc(req->rq_reqmsg); + + ENTRY; + + ioo = req_capsule_client_get(&req->rq_pill, &RMF_OBD_IOOBJ); + LASSERT(ioo != NULL); + + rnb = req_capsule_client_get(&req->rq_pill, &RMF_NIOBUF_REMOTE); + LASSERT(rnb != NULL); + + ext.start = rnb->offset; + rnb += ioo->ioo_bufcnt - 1; + ext.end = rnb->offset + rnb->len - 1; + + LASSERT(lock->l_resource != NULL); + if (!ostid_res_name_eq(&ioo->ioo_oid, &lock->l_resource->lr_name)) + RETURN(0); + + mode = LCK_PW; + if (opc == OST_READ) + mode |= LCK_PR; + + if (!(lock->l_granted_mode & mode)) + RETURN(0); + + RETURN(ldlm_extent_overlap(&lock->l_policy_data.l_extent, &ext)); +} + +/** + * High-priority queue request check for whether the given PTLRPC request + * (\a req) is blocking an LDLM lock cancel. + * + * Returns 1 if the given given PTLRPC request (\a req) is blocking an LDLM lock + * cancel, 0 if it is not, and -EFAULT if the request is malformed. + * + * Only OST_READs, OST_WRITEs and OST_PUNCHes go on the h-p RPC queue. This + * function looks only at OST_READs and OST_WRITEs. + */ +static int ofd_rw_hpreq_check(struct ptlrpc_request *req) +{ + struct tgt_session_info *tsi; + struct obd_ioobj *ioo; + struct niobuf_remote *rnb; + __u64 start, end; + int lock_count; + + ENTRY; + + /* Don't use tgt_ses_info() to get session info, because lock_match() + * can be called while request has no processing thread yet. */ + tsi = lu_context_key_get(&req->rq_session, &tgt_session_key); + LASSERT(tsi != NULL); + + /* + * Use LASSERT below because malformed RPCs should have + * been filtered out in tgt_hpreq_handler(). + */ + ioo = req_capsule_client_get(&req->rq_pill, &RMF_OBD_IOOBJ); + LASSERT(ioo != NULL); + + rnb = req_capsule_client_get(&req->rq_pill, &RMF_NIOBUF_REMOTE); + LASSERT(rnb != NULL); + LASSERT(!(rnb->flags & OBD_BRW_SRVLOCK)); + + start = rnb->offset; + rnb += ioo->ioo_bufcnt - 1; + end = rnb->offset + rnb->len - 1; + + DEBUG_REQ(D_RPCTRACE, req, "%s %s: refresh rw locks: "DFID + " ("LPU64"->"LPU64")\n", + tgt_name(tsi->tsi_tgt), current->comm, + PFID(&tsi->tsi_fid), start, end); + + lock_count = ofd_prolong_extent_locks(tsi, start, end); + + CDEBUG(D_DLMTRACE, "%s: refreshed %u locks timeout for req %p.\n", + tgt_name(tsi->tsi_tgt), lock_count, req); + + RETURN(lock_count > 0); +} + +static void ofd_rw_hpreq_fini(struct ptlrpc_request *req) +{ + ofd_rw_hpreq_check(req); +} + +/** + * Like tgt_rw_hpreq_lock_match(), but for OST_PUNCH RPCs. + */ +static int ofd_punch_hpreq_lock_match(struct ptlrpc_request *req, + struct ldlm_lock *lock) +{ + struct tgt_session_info *tsi; + + /* Don't use tgt_ses_info() to get session info, because lock_match() + * can be called while request has no processing thread yet. */ + tsi = lu_context_key_get(&req->rq_session, &tgt_session_key); + LASSERT(tsi != NULL); + + LASSERT(tsi->tsi_ost_body != NULL); + if (tsi->tsi_ost_body->oa.o_valid & OBD_MD_FLHANDLE && + tsi->tsi_ost_body->oa.o_handle.cookie == lock->l_handle.h_cookie) + return 1; + + return 0; +} + +/** + * Like ost_rw_hpreq_check(), but for OST_PUNCH RPCs. + */ +static int ofd_punch_hpreq_check(struct ptlrpc_request *req) +{ + struct tgt_session_info *tsi; + struct obdo *oa; + int lock_count; + + ENTRY; + + /* Don't use tgt_ses_info() to get session info, because lock_match() + * can be called while request has no processing thread yet. */ + tsi = lu_context_key_get(&req->rq_session, &tgt_session_key); + LASSERT(tsi != NULL); + oa = &tsi->tsi_ost_body->oa; + + LASSERT(!(oa->o_valid & OBD_MD_FLFLAGS && + oa->o_flags & OBD_FL_SRVLOCK)); + + CDEBUG(D_DLMTRACE, + "%s: refresh locks: "LPU64"/"LPU64" ("LPU64"->"LPU64")\n", + tgt_name(tsi->tsi_tgt), tsi->tsi_resid.name[0], + tsi->tsi_resid.name[1], oa->o_size, oa->o_blocks); + + lock_count = ofd_prolong_extent_locks(tsi, oa->o_size, oa->o_blocks); + + CDEBUG(D_DLMTRACE, "%s: refreshed %u locks timeout for req %p.\n", + tgt_name(tsi->tsi_tgt), lock_count, req); + + RETURN(lock_count > 0); +} + +static void ofd_punch_hpreq_fini(struct ptlrpc_request *req) +{ + ofd_punch_hpreq_check(req); +} + +struct ptlrpc_hpreq_ops ofd_hpreq_rw = { + .hpreq_lock_match = ofd_rw_hpreq_lock_match, + .hpreq_check = ofd_rw_hpreq_check, + .hpreq_fini = ofd_rw_hpreq_fini +}; + +struct ptlrpc_hpreq_ops ofd_hpreq_punch = { + .hpreq_lock_match = ofd_punch_hpreq_lock_match, + .hpreq_check = ofd_punch_hpreq_check, + .hpreq_fini = ofd_punch_hpreq_fini +}; + +/** Assign high priority operations to the IO requests */ +static void ofd_hp_brw(struct tgt_session_info *tsi) +{ + struct niobuf_remote *rnb; + struct obd_ioobj *ioo; + + ENTRY; + + ioo = req_capsule_client_get(tsi->tsi_pill, &RMF_OBD_IOOBJ); + LASSERT(ioo != NULL); /* must exist after request preprocessing */ + if (ioo->ioo_bufcnt > 0) { + rnb = req_capsule_client_get(tsi->tsi_pill, &RMF_NIOBUF_REMOTE); + LASSERT(rnb != NULL); /* must exist after request preprocessing */ + + /* no high priority if server lock is needed */ + if (rnb->flags & OBD_BRW_SRVLOCK) + return; + } + tgt_ses_req(tsi)->rq_ops = &ofd_hpreq_rw; +} + +static void ofd_hp_punch(struct tgt_session_info *tsi) +{ + LASSERT(tsi->tsi_ost_body != NULL); /* must exists if we are here */ + /* no high-priority if server lock is needed */ + if (tsi->tsi_ost_body->oa.o_valid & OBD_MD_FLFLAGS && + tsi->tsi_ost_body->oa.o_flags & OBD_FL_SRVLOCK) + return; + tgt_ses_req(tsi)->rq_ops = &ofd_hpreq_punch; +} + #define OBD_FAIL_OST_READ_NET OBD_FAIL_OST_BRW_NET #define OBD_FAIL_OST_WRITE_NET OBD_FAIL_OST_BRW_NET #define OST_BRW_READ OST_READ @@ -1609,11 +2041,15 @@ TGT_OST_HDL(0 | HABEO_REFERO | MUTABOR, TGT_OST_HDL(0 | HABEO_REFERO | MUTABOR, OST_DESTROY, ofd_destroy_hdl), TGT_OST_HDL(0 | HABEO_REFERO, OST_STATFS, ofd_statfs_hdl), -TGT_OST_HDL(HABEO_CORPUS| HABEO_REFERO, OST_BRW_READ, tgt_brw_read), +TGT_OST_HDL_HP(HABEO_CORPUS| HABEO_REFERO, + OST_BRW_READ, tgt_brw_read, + ofd_hp_brw), /* don't set CORPUS flag for brw_write because -ENOENT may be valid case */ -TGT_OST_HDL(MUTABOR, OST_BRW_WRITE, tgt_brw_write), -TGT_OST_HDL(HABEO_CORPUS| HABEO_REFERO | MUTABOR, - OST_PUNCH, ofd_punch_hdl), +TGT_OST_HDL_HP(HABEO_CORPUS| MUTABOR, OST_BRW_WRITE, tgt_brw_write, + ofd_hp_brw), +TGT_OST_HDL_HP(HABEO_CORPUS| HABEO_REFERO | MUTABOR, + OST_PUNCH, ofd_punch_hdl, + ofd_hp_punch), TGT_OST_HDL(HABEO_CORPUS| HABEO_REFERO, OST_SYNC, ofd_sync_hdl), TGT_OST_HDL(0 | HABEO_REFERO, OST_QUOTACTL, ofd_quotactl), }; @@ -1635,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 }, { @@ -1645,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 } }; @@ -1695,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); @@ -1798,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: @@ -1813,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); @@ -1836,7 +2292,7 @@ static void ofd_fini(const struct lu_env *env, struct ofd_device *m) ofd_stack_fini(env, m, &m->ofd_dt_dev.dd_lu_dev); ofd_procfs_fini(m); - LASSERT(cfs_atomic_read(&d->ld_ref) == 0); + LASSERT(atomic_read(&d->ld_ref) == 0); server_put_mount(obd->obd_name); EXIT; }