obd = class_name2obd(next);
if (obd == NULL) {
CERROR("%s: can't locate next device: %s\n",
- m->ofd_dt_dev.dd_lu_dev.ld_obd->obd_name, next);
+ ofd_name(m), next);
GOTO(out, rc = -ENOTCONN);
}
rc = obd_connect(NULL, exp, obd, &obd->obd_uuid, data, NULL);
if (rc) {
CERROR("%s: cannot connect to next dev %s: rc = %d\n",
- m->ofd_dt_dev.dd_lu_dev.ld_obd->obd_name, next, rc);
+ ofd_name(m), next, rc);
GOTO(out, rc);
}
LASSERT(m->ofd_osd_exp);
obd_disconnect(m->ofd_osd_exp);
- m->ofd_osd = NULL;
EXIT;
}
{ NULL }
};
+/* Some parameters were moved from ofd to osd and only their
+ * symlinks were kept in ofd by LU-3106. They are:
+ * -writehthrough_cache_enable
+ * -readcache_max_filese
+ * -read_cache_enable
+ * -brw_stats
+ * Since they are not included by the static lprocfs var list,
+ * a pre-check is added for them to avoid "unknown param" error
+ * message confuses the customer. If they are matched in this
+ * check, they will be passed to the osd directly.
+ */
+static bool match_symlink_param(char *param)
+{
+ char *sval;
+ int paramlen;
+
+ if (class_match_param(param, PARAM_OST, ¶m) == 0) {
+ sval = strchr(param, '=');
+ if (sval != NULL) {
+ paramlen = sval - param;
+ if (strncmp(param, "writethrough_cache_enable",
+ paramlen) == 0 ||
+ strncmp(param, "readcache_max_filesize",
+ paramlen) == 0 ||
+ strncmp(param, "read_cache_enable",
+ paramlen) == 0 ||
+ strncmp(param, "brw_stats", paramlen) == 0)
+ return true;
+ }
+ }
+
+ return false;
+}
+
/* used by MGS to process specific configurations */
static int ofd_process_config(const struct lu_env *env, struct lu_device *d,
struct lustre_cfg *cfg)
switch (cfg->lcfg_command) {
case LCFG_PARAM: {
- struct lprocfs_static_vars lvars;
-
+ struct obd_device *obd = ofd_obd(m);
/* For interoperability */
struct cfg_interop_param *ptr = NULL;
struct lustre_cfg *old_cfg = NULL;
}
}
- lprocfs_ofd_init_vars(&lvars);
- rc = class_process_proc_param(PARAM_OST, lvars.obd_vars, cfg,
+ if (match_symlink_param(param)) {
+ rc = next->ld_ops->ldo_process_config(env, next, cfg);
+ break;
+ }
+
+ rc = class_process_proc_seq_param(PARAM_OST, obd->obd_vars, cfg,
d->ld_obd);
- if (rc > 0 || rc == -ENOSYS)
+ if (rc > 0 || rc == -ENOSYS) {
+ CDEBUG(D_CONFIG, "pass param %s down the stack.\n",
+ param);
/* we don't understand; pass it on */
rc = next->ld_ops->ldo_process_config(env, next, cfg);
+ }
break;
}
case LCFG_SPTLRPC_CONF: {
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;
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_name(ofd), event);
+ return -EINVAL;
+ }
+
+ return 0;
+}
+
static int ofd_prepare(const struct lu_env *env, struct lu_device *pdev,
struct lu_device *dev)
{
struct ofd_device *ofd = ofd_dev(dev);
struct obd_device *obd = ofd_obd(ofd);
struct lu_device *next = &ofd->ofd_osd->dd_lu_dev;
- struct lfsck_start_param lsp;
int rc;
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);
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);
}
- lsp.lsp_start = NULL;
- lsp.lsp_namespace = ofd->ofd_namespace;
- rc = lfsck_start(env, ofd->ofd_osd, &lsp);
- if (rc != 0) {
- CWARN("%s: auto trigger paused LFSCK failed: rc = %d\n",
- obd->obd_name, rc);
- rc = 0;
- }
+ 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);
target_recovery_init(&ofd->ofd_lut, tgt_request_handle);
LASSERT(obd->obd_no_conn);
.ldo_prepare = ofd_prepare,
};
+LPROC_SEQ_FOPS(lprocfs_nid_stats_clear);
+
static int ofd_procfs_init(struct ofd_device *ofd)
{
- struct lprocfs_static_vars lvars;
struct obd_device *obd = ofd_obd(ofd);
cfs_proc_dir_entry_t *entry;
int rc = 0;
/* lprocfs must be setup before the ofd so state can be safely added
* to /proc incrementally as the ofd is setup */
- lprocfs_ofd_init_vars(&lvars);
- rc = lprocfs_obd_setup(obd, lvars.obd_vars);
+ obd->obd_vars = lprocfs_ofd_obd_vars;
+ rc = lprocfs_seq_obd_setup(obd);
if (rc) {
CERROR("%s: lprocfs_obd_setup failed: %d.\n",
obd->obd_name, rc);
obd->obd_uses_nid_stats = 1;
- entry = lprocfs_register("exports", obd->obd_proc_entry, NULL, NULL);
+ entry = lprocfs_seq_register("exports", obd->obd_proc_entry, NULL,
+ NULL);
if (IS_ERR(entry)) {
rc = PTR_ERR(entry);
CERROR("%s: error %d setting up lprocfs for %s\n",
obd->obd_proc_exports_entry = entry;
entry = lprocfs_add_simple(obd->obd_proc_exports_entry, "clear",
- lprocfs_nid_stats_clear_read,
- lprocfs_nid_stats_clear_write, obd, NULL);
+#ifndef HAVE_ONLY_PROCFS_SEQ
+ NULL, NULL,
+#endif
+ obd, &lprocfs_nid_stats_clear_fops);
if (IS_ERR(entry)) {
rc = PTR_ERR(entry);
CERROR("%s: add proc entry 'clear' failed: %d.\n",
rc = lprocfs_job_stats_init(obd, LPROC_OFD_STATS_LAST,
ofd_stats_counter_init);
if (rc)
- GOTO(remove_entry_clear, rc);
+ GOTO(obd_cleanup, rc);
RETURN(0);
-remove_entry_clear:
- lprocfs_remove_proc_entry("clear", obd->obd_proc_exports_entry);
obd_cleanup:
lprocfs_obd_cleanup(obd);
lprocfs_free_obd_stats(obd);
return rc;
}
+/**
+ * ofd_procfs_add_brw_stats_symlink - expose osd stats to ofd layer
+ *
+ * The osd interfaces to the backend file system exposes useful data
+ * such as brw_stats and read or write cache states. This same data
+ * needs to be exposed into the obdfilter (ofd) layer to maintain
+ * backwards compatibility. This function creates the symlinks in the
+ * proc layer to enable this.
+ */
static void ofd_procfs_add_brw_stats_symlink(struct ofd_device *ofd)
{
struct obd_device *obd = ofd_obd(ofd);
struct obd_device *osd_obd = ofd->ofd_osd_exp->exp_obd;
- cfs_proc_dir_entry_t *osd_root = osd_obd->obd_type->typ_procroot;
- cfs_proc_dir_entry_t *osd_dir;
- osd_dir = lprocfs_srch(osd_root, obd->obd_name);
- if (osd_dir == NULL)
+ if (obd->obd_proc_entry == NULL)
return;
- if (lprocfs_srch(osd_dir, "brw_stats") != NULL)
- lprocfs_add_symlink("brw_stats", obd->obd_proc_entry,
- "../../%s/%s/brw_stats",
- osd_root->name, osd_dir->name);
-
- if (lprocfs_srch(osd_dir, "read_cache_enable") != NULL)
- lprocfs_add_symlink("read_cache_enable", obd->obd_proc_entry,
- "../../%s/%s/read_cache_enable",
- osd_root->name, osd_dir->name);
-
- if (lprocfs_srch(osd_dir, "readcache_max_filesize") != NULL)
- lprocfs_add_symlink("readcache_max_filesize",
- obd->obd_proc_entry,
- "../../%s/%s/readcache_max_filesize",
- osd_root->name, osd_dir->name);
-
- if (lprocfs_srch(osd_dir, "writethrough_cache_enable") != NULL)
- lprocfs_add_symlink("writethrough_cache_enable",
- obd->obd_proc_entry,
- "../../%s/%s/writethrough_cache_enable",
- osd_root->name, osd_dir->name);
+ lprocfs_add_symlink("brw_stats", obd->obd_proc_entry,
+ "../../%s/%s/brw_stats",
+ osd_obd->obd_type->typ_name, obd->obd_name);
+
+ lprocfs_add_symlink("read_cache_enable", obd->obd_proc_entry,
+ "../../%s/%s/read_cache_enable",
+ osd_obd->obd_type->typ_name, obd->obd_name);
+
+ lprocfs_add_symlink("readcache_max_filesize",
+ obd->obd_proc_entry,
+ "../../%s/%s/readcache_max_filesize",
+ osd_obd->obd_type->typ_name, obd->obd_name);
+
+ lprocfs_add_symlink("writethrough_cache_enable",
+ obd->obd_proc_entry,
+ "../../%s/%s/writethrough_cache_enable",
+ osd_obd->obd_type->typ_name, obd->obd_name);
}
static void ofd_procfs_fini(struct ofd_device *ofd)
{
struct obd_device *obd = ofd_obd(ofd);
- lprocfs_remove_proc_entry("writethrough_cache_enable",
- obd->obd_proc_entry);
- lprocfs_remove_proc_entry("readcache_max_filesize",
- obd->obd_proc_entry);
- lprocfs_remove_proc_entry("read_cache_enable", obd->obd_proc_entry);
- lprocfs_remove_proc_entry("brw_stats", obd->obd_proc_entry);
- lprocfs_remove_proc_entry("clear", obd->obd_proc_exports_entry);
lprocfs_free_per_client_stats(obd);
lprocfs_obd_cleanup(obd);
lprocfs_free_obd_stats(obd);
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;
} 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);
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);
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);
* to go... deadlock! */
res = ldlm_resource_get(ofd->ofd_namespace, NULL,
&tsi->tsi_resid, LDLM_EXTENT, 0);
- if (res != NULL) {
+ if (!IS_ERR(res)) {
ldlm_res_lvbo_update(res, NULL, 0);
ldlm_resource_putref(res);
}
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;
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;
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) &&
/* 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);
/* 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 {
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);
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);
out_nolock:
- if (rc == 0)
+ if (rc == 0) {
+#if LUSTRE_VERSION_CODE < OBD_OCD_VERSION(2, 8, 53, 0)
+ struct ofd_thread_info *info = ofd_info(tsi->tsi_env);
+ struct lu_fid *fid = &info->fti_fid;
+
+ /* For compatible purpose, it needs to convert back to
+ * OST ID before put it on wire. */
+ *fid = rep_oa->o_oi.oi_fid;
+ fid_to_ostid(fid, &rep_oa->o_oi);
+#endif
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;
}
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;
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)
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)
}
rc = tgt_sync(tsi->tsi_env, tsi->tsi_tgt,
- fo != NULL ? ofd_object_child(fo) : NULL);
+ fo != NULL ? ofd_object_child(fo) : NULL,
+ repbody->oa.o_size, repbody->oa.o_blocks);
if (rc)
GOTO(put, rc);
}
rc = ofd_object_punch(tsi->tsi_env, fo, start, end, &info->fti_attr,
- ff);
+ ff, (struct obdo *)oa);
if (rc)
GOTO(out_put, rc);
* to go... deadlock! */
res = ldlm_resource_get(ns, NULL, &tsi->tsi_resid,
LDLM_EXTENT, 0);
- if (res != NULL) {
+ if (!IS_ERR(res)) {
ldlm_res_lvbo_update(res, NULL, 0);
ldlm_resource_putref(res);
}
return rc;
}
-
static int ofd_quotactl(struct tgt_session_info *tsi)
{
struct obd_quotactl *oqctl, *repoqc;
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
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),
};
.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
},
{
.tos_hs = seq_handlers
},
{
+ .tos_opc_start = LFSCK_FIRST_OPC,
+ .tos_opc_end = LFSCK_LAST_OPC,
+ .tos_hs = tgt_lfsck_handlers
+ },
+ {
.tos_hs = NULL
}
};
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);
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:
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);
ofd_stack_fini(env, m, &m->ofd_dt_dev.dd_lu_dev);
ofd_procfs_fini(m);
- LASSERT(cfs_atomic_read(&d->ld_ref) == 0);
- server_put_mount(obd->obd_name);
+ LASSERT(atomic_read(&d->ld_ref) == 0);
+ server_put_mount(obd->obd_name, true);
EXIT;
}
int __init ofd_init(void)
{
- struct lprocfs_static_vars lvars;
int rc;
rc = lu_kmem_init(ofd_caches);
return(rc);
}
- lprocfs_ofd_init_vars(&lvars);
-
- rc = class_register_type(&ofd_obd_ops, NULL, NULL,
+ rc = class_register_type(&ofd_obd_ops, NULL, true, NULL,
#ifndef HAVE_ONLY_PROCFS_SEQ
- lvars.module_vars,
+ NULL,
#endif
- LUSTRE_OST_NAME, &ofd_device_type);
+ LUSTRE_OST_NAME, &ofd_device_type);
return rc;
}
git://git.whamcloud.com / fs / lustre-release.git / blobdiff