* Copyright (c) 2009, 2010, Oracle and/or its affiliates. All rights reserved.
* Use is subject to license terms.
*
- * Copyright (c) 2012, 2014 Intel Corporation.
+ * Copyright (c) 2012, 2015, Intel Corporation.
*/
/*
* This file is part of Lustre, http://www.lustre.org/
* The OBD Filter Device (OFD) module belongs to the Object Storage
* Server stack and connects the RPC oriented Unified Target (TGT)
* layer (see lustre/include/lu_target.h) to the storage oriented OSD
- * layer (see lustre/doc/osd-api.txt).
+ * layer (see Documentation/osd-api.txt).
*
* TGT
* | DT and OBD APIs
struct lu_device *d;
struct ofd_thread_info *info = ofd_info(env);
struct lustre_mount_info *lmi;
+ struct lustre_mount_data *lmd;
int rc;
char *osdname;
RETURN(-ENODEV);
}
+ lmd = s2lsi(lmi->lmi_sb)->lsi_lmd;
+ if (lmd != NULL && lmd->lmd_flags & LMD_FLG_SKIP_LFSCK)
+ m->ofd_skip_lfsck = 1;
+
/* find bottom osd */
OBD_ALLOC(osdname, MTI_NAME_MAXLEN);
if (osdname == NULL)
static int ofd_recovery_complete(const struct lu_env *env,
struct lu_device *dev)
{
+ struct ofd_thread_info *oti = ofd_info(env);
struct ofd_device *ofd = ofd_dev(dev);
struct lu_device *next = &ofd->ofd_osd->dd_lu_dev;
- int rc = 0, max_precreate;
+ int rc = 0;
ENTRY;
/*
* Grant space for object precreation on the self export.
- * This initial reserved space (i.e. 10MB for zfs and 280KB for ldiskfs)
+ * The initial reserved space (i.e. 10MB for zfs and 280KB for ldiskfs)
* is enough to create 10k objects. More space is then acquired for
* precreation in ofd_grant_create().
*/
- max_precreate = OST_MAX_PRECREATE * ofd->ofd_dt_conf.ddp_inodespace / 2;
- ofd_grant_connect(env, dev->ld_obd->obd_self_export, max_precreate,
- false);
+ memset(&oti->fti_ocd, 0, sizeof(oti->fti_ocd));
+ oti->fti_ocd.ocd_grant = OST_MAX_PRECREATE / 2;
+ oti->fti_ocd.ocd_grant *= ofd->ofd_dt_conf.ddp_inodespace;
+ oti->fti_ocd.ocd_connect_flags = OBD_CONNECT_GRANT |
+ OBD_CONNECT_GRANT_PARAM;
+ ofd_grant_connect(env, dev->ld_obd->obd_self_export, &oti->fti_ocd,
+ true);
rc = next->ld_ops->ldo_recovery_complete(env, next);
RETURN(rc);
}
obd->obd_uses_nid_stats = 1;
- entry = lprocfs_seq_register("exports", obd->obd_proc_entry, NULL,
- NULL);
+ entry = lprocfs_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",
ss->ss_lu = lu->ld_site;
ss->ss_node_id = ofd->ofd_lut.lut_lsd.lsd_osd_index;
+ OBD_ALLOC(name, sizeof(obd_name) * 2 + 10);
+ if (name == NULL)
+ return -ENOMEM;
+
OBD_ALLOC_PTR(ss->ss_server_seq);
if (ss->ss_server_seq == NULL)
- GOTO(out_free, rc = -ENOMEM);
-
- OBD_ALLOC(name, strlen(obd_name) + 10);
- if (!name) {
- OBD_FREE_PTR(ss->ss_server_seq);
- ss->ss_server_seq = NULL;
- GOTO(out_free, rc = -ENOMEM);
- }
+ GOTO(out_name, rc = -ENOMEM);
rc = seq_server_init(env, ss->ss_server_seq, ofd->ofd_osd, obd_name,
LUSTRE_SEQ_SERVER, ss);
if (rc) {
CERROR("%s : seq server init error %d\n", obd_name, rc);
- GOTO(out_free, rc);
+ GOTO(out_server, rc);
}
ss->ss_server_seq->lss_space.lsr_index = ss->ss_node_id;
OBD_ALLOC_PTR(ss->ss_client_seq);
if (ss->ss_client_seq == NULL)
- GOTO(out_free, rc = -ENOMEM);
+ GOTO(out_server, rc = -ENOMEM);
- snprintf(name, strlen(obd_name) + 6, "%p-super", obd_name);
+ /*
+ * It always printed as "%p", so that the name is unique in the kernel,
+ * even if the filesystem is mounted twice. So sizeof(.) * 2 is enough.
+ */
+ snprintf(name, sizeof(obd_name) * 2 + 7, "%p-super", obd_name);
rc = seq_client_init(ss->ss_client_seq, NULL, LUSTRE_SEQ_DATA,
name, NULL);
if (rc) {
CERROR("%s : seq client init error %d\n", obd_name, rc);
- GOTO(out_free, rc);
+ GOTO(out_client, rc);
}
- OBD_FREE(name, strlen(obd_name) + 10);
- name = NULL;
rc = seq_server_set_cli(env, ss->ss_server_seq, ss->ss_client_seq);
-out_free:
if (rc) {
- if (ss->ss_server_seq) {
- seq_server_fini(ss->ss_server_seq, env);
- OBD_FREE_PTR(ss->ss_server_seq);
- ss->ss_server_seq = NULL;
- }
-
- if (ss->ss_client_seq) {
- seq_client_fini(ss->ss_client_seq);
- OBD_FREE_PTR(ss->ss_client_seq);
- ss->ss_client_seq = NULL;
- }
-
- if (name) {
- OBD_FREE(name, strlen(obd_name) + 10);
- name = NULL;
- }
+out_client:
+ seq_client_fini(ss->ss_client_seq);
+ OBD_FREE_PTR(ss->ss_client_seq);
+ ss->ss_client_seq = NULL;
+out_server:
+ seq_server_fini(ss->ss_server_seq, env);
+ OBD_FREE_PTR(ss->ss_server_seq);
+ ss->ss_server_seq = NULL;
}
+out_name:
+ OBD_FREE(name, sizeof(obd_name) * 2 + 10);
return rc;
}
void *key, *val = NULL;
int keylen, vallen, rc = 0;
bool is_grant_shrink;
- struct ofd_device *ofd = ofd_exp(tsi->tsi_exp);
ENTRY;
if (vallen > 0)
obd_export_evict_by_nid(tsi->tsi_exp->exp_obd, val);
rc = 0;
- } else if (KEY_IS(KEY_CAPA_KEY)) {
- rc = ofd_update_capa_key(ofd, val);
} else if (KEY_IS(KEY_SPTLRPC_CONF)) {
rc = tgt_adapt_sptlrpc_conf(tsi->tsi_tgt, 0);
} else {
* \retval negative value on error
*/
int ofd_fiemap_get(const struct lu_env *env, struct ofd_device *ofd,
- struct lu_fid *fid, struct ll_user_fiemap *fiemap)
+ struct lu_fid *fid, struct fiemap *fiemap)
{
struct ofd_object *fo;
int rc;
*/
static int lock_zero_regions(struct ldlm_namespace *ns,
struct ldlm_res_id *res_id,
- struct ll_user_fiemap *fiemap,
+ struct fiemap *fiemap,
struct list_head *locked)
{
__u64 begin = fiemap->fm_start;
unsigned int i;
int rc = 0;
- struct ll_fiemap_extent *fiemap_start = fiemap->fm_extents;
+ struct fiemap_extent *fiemap_start = fiemap->fm_extents;
ENTRY;
ofd_seq_put(tsi->tsi_env, oseq);
} else if (KEY_IS(KEY_FIEMAP)) {
struct ll_fiemap_info_key *fm_key;
- struct ll_user_fiemap *fiemap;
+ struct fiemap *fiemap;
struct lu_fid *fid;
req_capsule_extend(tsi->tsi_pill, &RQF_OST_GET_INFO_FIEMAP);
fm_key = req_capsule_client_get(tsi->tsi_pill, &RMF_FIEMAP_KEY);
- rc = tgt_validate_obdo(tsi, &fm_key->oa);
+ rc = tgt_validate_obdo(tsi, &fm_key->lfik_oa);
if (rc)
RETURN(err_serious(rc));
- fid = &fm_key->oa.o_oi.oi_fid;
+ fid = &fm_key->lfik_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);
+ replylen = fiemap_count_to_size(
+ fm_key->lfik_fiemap.fm_extent_count);
req_capsule_set_size(tsi->tsi_pill, &RMF_FIEMAP_VAL,
RCL_SERVER, replylen);
if (fiemap == NULL)
RETURN(-ENOMEM);
- *fiemap = fm_key->fiemap;
+ *fiemap = fm_key->lfik_fiemap;
rc = ofd_fiemap_get(tsi->tsi_env, ofd, fid, fiemap);
/* LU-3219: Lock the sparse areas to make sure dirty
* flushed back from client, then call fiemap again. */
- if (fm_key->oa.o_valid & OBD_MD_FLFLAGS &&
- fm_key->oa.o_flags & OBD_FL_SRVLOCK) {
+ if (fm_key->lfik_oa.o_valid & OBD_MD_FLFLAGS &&
+ fm_key->lfik_oa.o_flags & OBD_FL_SRVLOCK) {
struct list_head locked;
INIT_LIST_HEAD(&locked);
struct lustre_handle lh = { 0 };
struct ofd_object *fo;
__u64 flags = 0;
- ldlm_mode_t lock_mode = LCK_PR;
+ enum ldlm_mode lock_mode = LCK_PR;
bool srvlock;
int rc;
ENTRY;
struct ofd_seq *oseq;
int rc = 0, diff;
int sync_trans = 0;
+ long granted = 0;
ENTRY;
if (!(oa->o_valid & OBD_MD_FLFLAGS) ||
!(oa->o_flags & OBD_FL_DELORPHAN)) {
/* don't enforce grant during orphan recovery */
- rc = ofd_grant_create(tsi->tsi_env,
- ofd_obd(ofd)->obd_self_export,
- &diff);
- if (rc) {
+ granted = ofd_grant_create(tsi->tsi_env,
+ ofd_obd(ofd)->obd_self_export,
+ &diff);
+ if (granted < 0) {
+ rc = granted;
+ granted = 0;
CDEBUG(D_HA, "%s: failed to acquire grant "
"space for precreate (%d): rc = %d\n",
ofd_name(ofd), diff, rc);
* LFSCK will eventually clean up any orphans. LU-14 */
if (diff > 5 * OST_MAX_PRECREATE) {
diff = OST_MAX_PRECREATE / 2;
- LCONSOLE_WARN("%s: precreate FID "DOSTID" is over %u "
- "larger than the LAST_ID "DOSTID", only "
- "precreating the last %u objects.\n",
- ofd_name(ofd), POSTID(&oa->o_oi),
- 5 * OST_MAX_PRECREATE,
- POSTID(&oseq->os_oi), diff);
+ LCONSOLE_WARN("%s: Too many FIDs to precreate "
+ "OST replaced or reformatted: "
+ "LFSCK will clean up",
+ ofd_name(ofd));
+
+ CDEBUG(D_HA, "%s: precreate FID "DOSTID" is over "
+ "%u larger than the LAST_ID "DOSTID", only "
+ "precreating the last %u objects.\n",
+ ofd_name(ofd), POSTID(&oa->o_oi),
+ 5 * OST_MAX_PRECREATE,
+ POSTID(&oseq->os_oi), diff);
ofd_seq_last_oid_set(oseq, ostid_id(&oa->o_oi) - diff);
}
ofd_name(ofd), rc);
if (!(oa->o_valid & OBD_MD_FLFLAGS) ||
- !(oa->o_flags & OBD_FL_DELORPHAN))
- ofd_grant_commit(tsi->tsi_env,
- ofd_obd(ofd)->obd_self_export, rc);
+ !(oa->o_flags & OBD_FL_DELORPHAN)) {
+ ofd_grant_commit(ofd_obd(ofd)->obd_self_export, granted,
+ rc);
+ granted = 0;
+ }
ostid_set_id(&rep_oa->o_oi, ofd_seq_last_oid(oseq));
}
ENTRY;
+ OBD_FAIL_TIMEOUT(OBD_FAIL_OST_PAUSE_PUNCH, cfs_fail_val);
+
/* check that we do support OBD_CONNECT_TRUNCLOCK. */
CLASSERT(OST_CONNECT_SUPPORTED & OBD_CONNECT_TRUNCLOCK);
}
/**
+ * OFD request handler for OST_LADVISE RPC.
+ *
+ * Tune cache or perfetch policies according to advices.
+ *
+ * \param[in] tsi target session environment for this request
+ *
+ * \retval 0 if successful
+ * \retval negative errno on error
+ */
+static int ofd_ladvise_hdl(struct tgt_session_info *tsi)
+{
+ struct ptlrpc_request *req = tgt_ses_req(tsi);
+ struct obd_export *exp = tsi->tsi_exp;
+ struct ofd_device *ofd = ofd_exp(exp);
+ struct ost_body *body, *repbody;
+ struct ofd_thread_info *info;
+ struct ofd_object *fo;
+ const struct lu_env *env = req->rq_svc_thread->t_env;
+ int rc = 0;
+ struct lu_ladvise *ladvise;
+ int num_advise;
+ struct ladvise_hdr *ladvise_hdr;
+ int i;
+ ENTRY;
+
+ body = tsi->tsi_ost_body;
+
+ if ((body->oa.o_valid & OBD_MD_FLID) != OBD_MD_FLID)
+ RETURN(err_serious(-EPROTO));
+
+ ladvise_hdr = req_capsule_client_get(tsi->tsi_pill,
+ &RMF_OST_LADVISE_HDR);
+ if (ladvise_hdr == NULL)
+ RETURN(err_serious(-EPROTO));
+
+ if (ladvise_hdr->lah_magic != LADVISE_MAGIC ||
+ ladvise_hdr->lah_count < 1)
+ RETURN(err_serious(-EPROTO));
+
+ if ((ladvise_hdr->lah_flags & (~LF_MASK)) != 0)
+ RETURN(err_serious(-EPROTO));
+
+ ladvise = req_capsule_client_get(tsi->tsi_pill, &RMF_OST_LADVISE);
+ if (ladvise == NULL)
+ RETURN(err_serious(-EPROTO));
+
+ num_advise = req_capsule_get_size(&req->rq_pill,
+ &RMF_OST_LADVISE, RCL_CLIENT) /
+ sizeof(*ladvise);
+ if (num_advise < ladvise_hdr->lah_count)
+ RETURN(err_serious(-EPROTO));
+
+ repbody = req_capsule_server_get(&req->rq_pill, &RMF_OST_BODY);
+ repbody->oa = body->oa;
+
+ info = ofd_info_init(env, exp);
+
+ rc = ostid_to_fid(&info->fti_fid, &body->oa.o_oi,
+ ofd->ofd_lut.lut_lsd.lsd_osd_index);
+ if (rc != 0)
+ RETURN(rc);
+
+ fo = ofd_object_find(env, ofd, &info->fti_fid);
+ if (IS_ERR(fo)) {
+ rc = PTR_ERR(fo);
+ RETURN(rc);
+ }
+ LASSERT(fo != NULL);
+
+ for (i = 0; i < num_advise; i++, ladvise++) {
+ if (ladvise->lla_end <= ladvise->lla_start) {
+ rc = err_serious(-EPROTO);
+ break;
+ }
+
+ /* Handle different advice types */
+ switch (ladvise->lla_advice) {
+ default:
+ rc = -ENOTSUPP;
+ break;
+ }
+ if (rc != 0)
+ break;
+ }
+
+ ofd_object_put(env, fo);
+ req->rq_status = rc;
+ RETURN(rc);
+}
+
+/**
* OFD request handler for OST_QUOTACTL RPC.
*
* This is part of request processing to validate incoming request fields,
*/
static int ofd_quotactl(struct tgt_session_info *tsi)
{
- struct obd_quotactl *oqctl, *repoqc;
- struct lu_nodemap *nodemap =
- tsi->tsi_exp->exp_target_data.ted_nodemap;
- int id;
- int rc;
+ struct obd_quotactl *oqctl, *repoqc;
+ struct lu_nodemap *nodemap;
+ int id;
+ int rc;
ENTRY;
if (repoqc == NULL)
RETURN(err_serious(-ENOMEM));
- /* report success for quota on/off for interoperability with current MDT
- * stack */
- if (oqctl->qc_cmd == Q_QUOTAON || oqctl->qc_cmd == Q_QUOTAOFF)
- RETURN(0);
-
*repoqc = *oqctl;
+ nodemap = nodemap_get_from_exp(tsi->tsi_exp);
+ if (IS_ERR(nodemap))
+ RETURN(PTR_ERR(nodemap));
+
id = repoqc->qc_id;
if (oqctl->qc_type == USRQUOTA)
id = nodemap_map_id(nodemap, NODEMAP_UID,
NODEMAP_CLIENT_TO_FS,
repoqc->qc_id);
+ nodemap_putref(nodemap);
+
if (repoqc->qc_id != id)
swap(repoqc->qc_id, id);
LDLM_LOCK_PUT(lock);
RETURN(lock_count);
}
+ lock->l_last_used = cfs_time_current();
LDLM_LOCK_PUT(lock);
}
}
LASSERT(lock->l_flags & LDLM_FL_AST_SENT);
LASSERT(lock->l_resource->lr_type == LDLM_EXTENT);
+ /* ignore waiting locks, no more granted locks in the list */
+ if (lock->l_granted_mode != lock->l_req_mode)
+ break;
+
if (!ldlm_res_eq(&tsi->tsi_resid, &lock->l_resource->lr_name))
continue;
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);
+ struct niobuf_remote *rnb;
+ struct obd_ioobj *ioo;
+ enum ldlm_mode mode;
+ struct ldlm_extent ext;
+ __u32 opc = lustre_msg_get_opc(req->rq_reqmsg);
ENTRY;
struct ldlm_lock *lock)
{
struct tgt_session_info *tsi;
+ struct obdo *oa;
+ struct ldlm_extent ext;
+
+ ENTRY;
/* Don't use tgt_ses_info() to get session info, because lock_match()
* can be called while request has no processing thread yet. */
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(1);
- return 0;
+ oa = &tsi->tsi_ost_body->oa;
+ ext.start = oa->o_size;
+ ext.end = oa->o_blocks;
+
+ LASSERT(lock->l_resource != NULL);
+ if (!ostid_res_name_eq(&oa->o_oi, &lock->l_resource->lr_name))
+ RETURN(0);
+
+ if (!(lock->l_granted_mode & (LCK_PW | LCK_GROUP)))
+ RETURN(0);
+
+ RETURN(ldlm_extent_overlap(&lock->l_policy_data.l_extent, &ext));
}
/**
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),
+TGT_OST_HDL(HABEO_CORPUS | HABEO_REFERO, OST_LADVISE, ofd_ladvise_hdl),
};
static struct tgt_opc_slice ofd_common_slice[] = {
.tos_hs = tgt_lfsck_handlers
},
{
+ .tos_opc_start = SEC_FIRST_OPC,
+ .tos_opc_end = SEC_LAST_OPC,
+ .tos_hs = tgt_sec_ctx_handlers
+ },
+ {
.tos_hs = NULL
}
};
ofd_slc_set(m);
m->ofd_grant_compat_disable = 0;
m->ofd_soft_sync_limit = OFD_SOFT_SYNC_LIMIT_DEFAULT;
+ m->ofd_brw_size = ONE_MB_BRW_SIZE;
/* statfs data */
spin_lock_init(&m->ofd_osfs_lock);
spin_lock_init(&m->ofd_batch_lock);
init_rwsem(&m->ofd_lastid_rwsem);
- obd->u.filter.fo_fl_oss_capa = 0;
- INIT_LIST_HEAD(&obd->u.filter.fo_capa_keys);
- obd->u.filter.fo_capa_hash = init_capa_hash();
- if (obd->u.filter.fo_capa_hash == NULL)
- RETURN(-ENOMEM);
-
m->ofd_dt_dev.dd_lu_dev.ld_ops = &ofd_lu_ops;
m->ofd_dt_dev.dd_lu_dev.ld_obd = obd;
/* set this lu_device to obd, because error handling need it */
info = ofd_info_init(env, NULL);
if (info == NULL)
- RETURN(-EFAULT);
+ GOTO(err_fini_proc, rc = -EFAULT);
rc = ofd_stack_init(env, m, cfg);
if (rc) {
dt_conf_get(env, m->ofd_osd, &m->ofd_dt_conf);
- /* Allow at most ddp_grant_reserved% of the available filesystem space
- * to be granted to clients, so that any errors in the grant overhead
- * calculations do not allow granting more space to clients than can be
- * written. Assumes that in aggregate the grant overhead calculations do
- * not have more than ddp_grant_reserved% estimation error in them. */
- m->ofd_grant_ratio =
- ofd_grant_ratio_conv(m->ofd_dt_conf.ddp_grant_reserved);
-
rc = tgt_init(env, &m->ofd_lut, obd, m->ofd_osd, ofd_common_slice,
OBD_FAIL_OST_ALL_REQUEST_NET,
OBD_FAIL_OST_ALL_REPLY_NET);
if (rc != 0)
GOTO(err_fini_fs, rc);
+ tgt_adapt_sptlrpc_conf(&m->ofd_lut, 1);
+
RETURN(0);
err_fini_fs:
stop.ls_flags = 0;
lfsck_stop(env, m->ofd_osd, &stop);
target_recovery_fini(obd);
+ if (m->ofd_namespace != NULL)
+ ldlm_namespace_free_prior(m->ofd_namespace, NULL,
+ d->ld_obd->obd_force);
+
obd_exports_barrier(obd);
obd_zombie_barrier();
lfsck_degister(env, m->ofd_osd);
ofd_fs_cleanup(env, m);
- ofd_free_capa_keys(m);
- cleanup_capa_hash(obd->u.filter.fo_capa_hash);
-
if (m->ofd_namespace != NULL) {
- ldlm_namespace_free(m->ofd_namespace, NULL,
- d->ld_obd->obd_force);
+ ldlm_namespace_free_post(m->ofd_namespace);
d->ld_obd->obd_namespace = m->ofd_namespace = NULL;
}
class_unregister_type(LUSTRE_OST_NAME);
}
-MODULE_AUTHOR("Whamcloud, Inc. <http://www.whamcloud.com/>");
+MODULE_AUTHOR("OpenSFS, Inc. <http://www.lustre.org/>");
MODULE_DESCRIPTION("Lustre Object Filtering Device");
+MODULE_VERSION(LUSTRE_VERSION_STRING);
MODULE_LICENSE("GPL");
module_init(ofd_init);
git://git.whamcloud.com / fs / lustre-release.git / blobdiff