Whamcloud - gitweb
git://git.whamcloud.com
/
fs
/
lustre-release.git
/ blobdiff
commit
grep
author
committer
pickaxe
?
search:
re
summary
|
shortlog
|
log
|
commit
|
commitdiff
|
tree
raw
| inline |
side by side
branch: HEAD
[fs/lustre-release.git]
/
lustre
/
mdt
/
mdt_handler.c
diff --git
a/lustre/mdt/mdt_handler.c
b/lustre/mdt/mdt_handler.c
index
34620b8
..
3182205
100644
(file)
--- a/
lustre/mdt/mdt_handler.c
+++ b/
lustre/mdt/mdt_handler.c
@@
-348,7
+348,7
@@
void mdt_pack_size2body(struct mdt_thread_info *info, struct mdt_object *o)
{
struct mdt_body *b;
struct lu_attr *attr = &info->mti_attr.ma_attr;
-
+
b = req_capsule_server_get(info->mti_pill, &RMF_MDT_BODY);
/* Check if Size-on-MDS is enabled. */
@@
-392,7
+392,7
@@
void mdt_pack_attr2body(struct mdt_thread_info *info, struct mdt_body *b,
b->ino = fid_oid(fid); /* 1.6 compatibility */
b->generation = fid_ver(fid); /* 1.6 compatibility */
b->valid |= OBD_MD_FLGENER; /* 1.6 compatibility */
-
+
CDEBUG(D_INODE, DFID": nlink=%d, mode=%o, size="LPU64"\n",
PFID(fid), b->nlink, b->mode, b->size);
}
@@
-778,7
+778,7
@@
static int mdt_getattr_name_lock(struct mdt_thread_info *info,
if (name == NULL)
RETURN(err_serious(-EFAULT));
- namelen = req_capsule_get_size(info->mti_pill, &RMF_NAME,
+ namelen = req_capsule_get_size(info->mti_pill, &RMF_NAME,
RCL_CLIENT) - 1;
LASSERT(namelen >= 0);
@@
-923,6
+923,7
@@
static int mdt_getattr_name_lock(struct mdt_thread_info *info,
LDLM_LOCK_PUT(lock);
rc = 0;
} else {
+ struct md_attr *ma = &info->mti_attr;
relock:
mdt_lock_handle_init(lhc);
mdt_lock_reg_init(lhc, LCK_PR);
@@
-932,6
+933,23
@@
relock:
&child->mot_obj.mo_lu,
"Object doesn't exist!\n");
}
+
+ ma->ma_valid = 0;
+ ma->ma_need = MA_INODE;
+ rc = mo_attr_get(info->mti_env, next, ma);
+ if (unlikely(rc != 0))
+ GOTO(out_child, rc);
+
+ /* If the file has not been changed for some time, we return
+ * not only a LOOKUP lock, but also an UPDATE lock and this
+ * might save us RPC on later STAT. For directories, it also
+ * let negative dentry starts working for this dir. */
+ if (ma->ma_valid & MA_INODE &&
+ ma->ma_attr.la_valid & LA_CTIME &&
+ info->mti_mdt->mdt_namespace->ns_ctime_age_limit +
+ ma->ma_attr.la_ctime < cfs_time_current_sec())
+ child_bits |= MDS_INODELOCK_UPDATE;
+
rc = mdt_object_lock(info, child, lhc, child_bits,
MDT_CROSS_LOCK);
@@
-1126,7
+1144,7
@@
static int mdt_sendpage(struct mdt_thread_info *info,
GOTO(free_desc, rc);
if (OBD_FAIL_CHECK(OBD_FAIL_MDS_SENDPAGE))
- GOTO(abort_bulk, rc);
+ GOTO(abort_bulk, rc
= 0
);
*lwi = LWI_TIMEOUT(obd_timeout * HZ / 4, NULL, NULL);
rc = l_wait_event(desc->bd_waitq, !ptlrpc_bulk_active(desc), lwi);
@@
-1190,7
+1208,7
@@
static int mdt_write_dir_page(struct mdt_thread_info *info, struct page *page,
ma->ma_attr.la_valid = LA_MODE;
ma->ma_valid = MA_INODE;
- kmap(page);
+
cfs_
kmap(page);
dp = page_address(page);
offset = (int)((__u32)lu_dirent_start(dp) - (__u32)dp);
@@
-1233,7
+1251,7
@@
static int mdt_write_dir_page(struct mdt_thread_info *info, struct page *page,
}
EXIT;
out:
- kunmap(page);
+
cfs_
kunmap(page);
return rc;
}
@@
-1320,7
+1338,7
@@
static int mdt_writepage(struct mdt_thread_info *info)
cleanup_lwi:
OBD_FREE_PTR(lwi);
cleanup_page:
-
__
cfs_free_page(page);
+ cfs_free_page(page);
desc_cleanup:
ptlrpc_free_bulk(desc);
RETURN(rc);
@@
-1381,7
+1399,7
@@
free_rdpg:
for (i = 0; i < rdpg->rp_npages; i++)
if (rdpg->rp_pages[i] != NULL)
-
__
cfs_free_page(rdpg->rp_pages[i]);
+ cfs_free_page(rdpg->rp_pages[i]);
OBD_FREE(rdpg->rp_pages, rdpg->rp_npages * sizeof rdpg->rp_pages[0]);
if (OBD_FAIL_CHECK(OBD_FAIL_MDS_SENDPAGE))
@@
-1628,11
+1646,6
@@
static int mdt_enqueue(struct mdt_thread_info *info)
*/
LASSERT(info->mti_dlm_req != NULL);
- if (OBD_FAIL_CHECK(OBD_FAIL_LDLM_ENQUEUE)) {
- info->mti_fail_id = OBD_FAIL_LDLM_ENQUEUE;
- return 0;
- }
-
req = mdt_info_req(info);
/*
@@
-2026,22
+2039,18
@@
static int mdt_req_handle(struct mdt_thread_info *info,
LASSERT(current->journal_info == NULL);
/*
- * Mask out OBD_FAIL_ONCE, because that will stop
- * correct handling of failed req later in ldlm due to doing
- * obd_fail_loc |= OBD_FAIL_ONCE without actually
- * correct actions like it is done in target_send_reply_msg().
+ * Checking for various OBD_FAIL_$PREF_$OPC_NET codes. _Do_ not try
+ * to put same checks into handlers like mdt_close(), mdt_reint(),
+ * etc., without talking to mdt authors first. Checking same thing
+ * there again is useless and returning 0 error wihtout packing reply
+ * is buggy! Handlers either pack reply or return error.
+ *
+ * We return 0 here and do not send any reply in order to emulate
+ * network failure. Do not send any reply in case any of NET related
+ * fail_id has occured.
*/
- if (h->mh_fail_id != 0) {
- /*
- * Set to info->mti_fail_id to handler fail_id, it will be used
- * later, and better than use default fail_id.
- */
- if (OBD_FAIL_CHECK_RESET(h->mh_fail_id && OBD_FAIL_MASK_LOC,
- h->mh_fail_id & ~OBD_FAILED)) {
- info->mti_fail_id = h->mh_fail_id;
- RETURN(0);
- }
- }
+ if (OBD_FAIL_CHECK_ORSET(h->mh_fail_id, OBD_FAIL_ONCE))
+ RETURN(0);
rc = 0;
flags = h->mh_flags;
@@
-2088,6
+2097,13
@@
static int mdt_req_handle(struct mdt_thread_info *info,
* only
*/
rc = h->mh_act(info);
+ if (rc == 0 &&
+ !req->rq_no_reply && req->rq_reply_state == NULL) {
+ DEBUG_REQ(D_ERROR, req, "MDT \"handler\" %s did not "
+ "pack reply and returned 0 error\n",
+ h->mh_name);
+ LBUG();
+ }
serious = is_serious(rc);
rc = clear_serious(rc);
} else
@@
-2123,7
+2139,8
@@
static int mdt_req_handle(struct mdt_thread_info *info,
LBUG();
}
- RETURN(rc);
+ target_send_reply(req, rc, info->mti_fail_id);
+ RETURN(0);
}
void mdt_lock_handle_init(struct mdt_lock_handle *lh)
@@
-2199,9
+2216,35
@@
static void mdt_thread_info_fini(struct mdt_thread_info *info)
info->mti_env = NULL;
}
-/* mds/handler.c */
-extern int mds_filter_recovery_request(struct ptlrpc_request *req,
- struct obd_device *obd, int *process);
+static int mdt_filter_recovery_request(struct ptlrpc_request *req,
+ struct obd_device *obd, int *process)
+{
+ switch (lustre_msg_get_opc(req->rq_reqmsg)) {
+ case MDS_CONNECT: /* This will never get here, but for completeness. */
+ case OST_CONNECT: /* This will never get here, but for completeness. */
+ case MDS_DISCONNECT:
+ case OST_DISCONNECT:
+ *process = 1;
+ RETURN(0);
+
+ case MDS_CLOSE:
+ case MDS_DONE_WRITING:
+ case MDS_SYNC: /* used in unmounting */
+ case OBD_PING:
+ case MDS_REINT:
+ case SEQ_QUERY:
+ case FLD_QUERY:
+ case LDLM_ENQUEUE:
+ *process = target_queue_recovery_request(req, obd);
+ RETURN(0);
+
+ default:
+ DEBUG_REQ(D_ERROR, req, "not permitted during recovery");
+ *process = -EAGAIN;
+ RETURN(0);
+ }
+}
+
/*
* Handle recovery. Return:
* +1: continue request processing;
@@
-2279,7
+2322,7
@@
static int mdt_recovery(struct mdt_thread_info *info)
int rc;
int should_process;
DEBUG_REQ(D_INFO, req, "Got new replay");
- rc = md
s
_filter_recovery_request(req, obd, &should_process);
+ rc = md
t
_filter_recovery_request(req, obd, &should_process);
if (rc != 0 || !should_process)
RETURN(rc);
else if (should_process < 0) {
@@
-2291,24
+2334,85
@@
static int mdt_recovery(struct mdt_thread_info *info)
RETURN(+1);
}
-static int mdt_reply(struct ptlrpc_request *req, int rc,
- struct mdt_thread_info *info)
+static int mdt_msg_check_version(struct lustre_msg *msg)
{
-
ENTRY
;
+
int rc
;
-#if 0
- if (req->rq_reply_state == NULL && rc == 0) {
- req->rq_status = rc;
- lustre_pack_reply(req, 1, NULL, NULL);
+ switch (lustre_msg_get_opc(msg)) {
+ case MDS_CONNECT:
+ case MDS_DISCONNECT:
+ case OBD_PING:
+ case SEC_CTX_INIT:
+ case SEC_CTX_INIT_CONT:
+ case SEC_CTX_FINI:
+ rc = lustre_msg_check_version(msg, LUSTRE_OBD_VERSION);
+ if (rc)
+ CERROR("bad opc %u version %08x, expecting %08x\n",
+ lustre_msg_get_opc(msg),
+ lustre_msg_get_version(msg),
+ LUSTRE_OBD_VERSION);
+ break;
+ case MDS_GETSTATUS:
+ case MDS_GETATTR:
+ case MDS_GETATTR_NAME:
+ case MDS_STATFS:
+ case MDS_READPAGE:
+ case MDS_WRITEPAGE:
+ case MDS_IS_SUBDIR:
+ case MDS_REINT:
+ case MDS_CLOSE:
+ case MDS_DONE_WRITING:
+ case MDS_PIN:
+ case MDS_SYNC:
+ case MDS_GETXATTR:
+ case MDS_SETXATTR:
+ case MDS_SET_INFO:
+ case MDS_QUOTACHECK:
+ case MDS_QUOTACTL:
+ case QUOTA_DQACQ:
+ case QUOTA_DQREL:
+ case SEQ_QUERY:
+ case FLD_QUERY:
+ rc = lustre_msg_check_version(msg, LUSTRE_MDS_VERSION);
+ if (rc)
+ CERROR("bad opc %u version %08x, expecting %08x\n",
+ lustre_msg_get_opc(msg),
+ lustre_msg_get_version(msg),
+ LUSTRE_MDS_VERSION);
+ break;
+ case LDLM_ENQUEUE:
+ case LDLM_CONVERT:
+ case LDLM_BL_CALLBACK:
+ case LDLM_CP_CALLBACK:
+ rc = lustre_msg_check_version(msg, LUSTRE_DLM_VERSION);
+ if (rc)
+ CERROR("bad opc %u version %08x, expecting %08x\n",
+ lustre_msg_get_opc(msg),
+ lustre_msg_get_version(msg),
+ LUSTRE_DLM_VERSION);
+ break;
+ case OBD_LOG_CANCEL:
+ case LLOG_ORIGIN_HANDLE_CREATE:
+ case LLOG_ORIGIN_HANDLE_NEXT_BLOCK:
+ case LLOG_ORIGIN_HANDLE_READ_HEADER:
+ case LLOG_ORIGIN_HANDLE_CLOSE:
+ case LLOG_ORIGIN_HANDLE_DESTROY:
+ case LLOG_ORIGIN_HANDLE_PREV_BLOCK:
+ case LLOG_CATINFO:
+ rc = lustre_msg_check_version(msg, LUSTRE_LOG_VERSION);
+ if (rc)
+ CERROR("bad opc %u version %08x, expecting %08x\n",
+ lustre_msg_get_opc(msg),
+ lustre_msg_get_version(msg),
+ LUSTRE_LOG_VERSION);
+ break;
+ default:
+ CERROR("MDS unknown opcode %d\n", lustre_msg_get_opc(msg));
+ rc = -ENOTSUPP;
}
-#endif
- target_send_reply(req, rc, info->mti_fail_id);
- RETURN(0);
+ return rc;
}
-/* mds/handler.c */
-extern int mds_msg_check_version(struct lustre_msg *msg);
-
static int mdt_handle0(struct ptlrpc_request *req,
struct mdt_thread_info *info,
struct mdt_opc_slice *supported)
@@
-2325,7
+2429,7
@@
static int mdt_handle0(struct ptlrpc_request *req,
LASSERT(current->journal_info == NULL);
msg = req->rq_reqmsg;
- rc = md
s
_msg_check_version(msg);
+ rc = md
t
_msg_check_version(msg);
if (likely(rc == 0)) {
rc = mdt_recovery(info);
if (likely(rc == +1)) {
@@
-2333,7
+2437,6
@@
static int mdt_handle0(struct ptlrpc_request *req,
supported);
if (likely(h != NULL)) {
rc = mdt_req_handle(info, h, req);
- rc = mdt_reply(req, rc, info);
} else {
CERROR("The unsupported opc: 0x%x\n", lustre_msg_get_opc(msg) );
req->rq_status = -ENOTSUPP;
@@
-3483,7
+3586,6
@@
err_mdt_svc:
static void mdt_stack_fini(const struct lu_env *env,
struct mdt_device *m, struct lu_device *top)
{
- struct lu_device *d = top, *n;
struct obd_device *obd = m->mdt_md_dev.md_lu_dev.ld_obd;
struct lustre_cfg_bufs *bufs;
struct lustre_cfg *lcfg;
@@
-3512,23
+3614,7
@@
static void mdt_stack_fini(const struct lu_env *env,
top->ld_ops->ldo_process_config(env, top, lcfg);
lustre_cfg_free(lcfg);
- lu_site_purge(env, top->ld_site, ~0);
- while (d != NULL) {
- struct obd_type *type;
- struct lu_device_type *ldt = d->ld_type;
-
- /* each fini() returns next device in stack of layers
- * * so we can avoid the recursion */
- n = ldt->ldt_ops->ldto_device_fini(env, d);
- lu_device_put(d);
- ldt->ldt_ops->ldto_device_free(env, d);
- type = ldt->ldt_obd_type;
- type->typ_refcnt--;
- class_put_type(type);
-
- /* switch to the next device in the layer */
- d = n;
- }
+ lu_stack_fini(env, top);
m->mdt_child = NULL;
m->mdt_bottom = NULL;
}
@@
-3659,7
+3745,7
@@
static void mdt_fini(const struct lu_env *env, struct mdt_device *m)
ENTRY;
ping_evictor_stop();
-
+
target_recovery_fini(obd);
mdt_stop_ptlrpc_service(m);
@@
-3669,7
+3755,7
@@
static void mdt_fini(const struct lu_env *env, struct mdt_device *m)
m->mdt_identity_cache = NULL;
if (m->mdt_namespace != NULL) {
- ldlm_namespace_free(m->mdt_namespace, d->ld_obd->obd_force);
+ ldlm_namespace_free(m->mdt_namespace,
NULL,
d->ld_obd->obd_force);
d->ld_obd->obd_namespace = m->mdt_namespace = NULL;
}
@@
-3683,22
+3769,13
@@
static void mdt_fini(const struct lu_env *env, struct mdt_device *m)
sptlrpc_rule_set_free(&m->mdt_sptlrpc_rset);
next->md_ops->mdo_init_capa_ctxt(env, next, 0, 0, 0, NULL);
-
del_timer
(&m->mdt_ck_timer);
+
cfs_timer_disarm
(&m->mdt_ck_timer);
mdt_ck_thread_stop(m);
/* finish the stack */
mdt_stack_fini(env, m, md2lu_dev(m->mdt_child));
if (ls) {
- if (!list_empty(&ls->ls_lru) || ls->ls_total != 0) {
- /*
- * Uh-oh, objects still exist.
- */
- static DECLARE_LU_CDEBUG_PRINT_INFO(cookie, D_ERROR);
-
- lu_site_print(env, ls, &cookie, lu_cdebug_printer);
- }
-
lu_site_fini(ls);
OBD_FREE_PTR(ls);
d->ld_site = NULL;
@@
-3713,6
+3790,15
@@
static void fsoptions_to_mdt_flags(struct mdt_device *m, char *options)
{
char *p = options;
+#ifdef CONFIG_FS_POSIX_ACL
+ /* ACLs should be enabled by default (b=13829) */
+ m->mdt_opts.mo_acl = 1;
+ LCONSOLE_INFO("Enabling ACL\n");
+#else
+ m->mdt_opts.mo_acl = 0;
+ LCONSOLE_INFO("Disabling ACL\n");
+#endif
+
if (!options)
return;
@@
-3731,16
+3817,6
@@
static void fsoptions_to_mdt_flags(struct mdt_device *m, char *options)
(memcmp(options, "nouser_xattr", len) == 0)) {
m->mdt_opts.mo_user_xattr = 0;
LCONSOLE_INFO("Disabling user_xattr\n");
- } else if ((len == sizeof("acl") - 1) &&
- (memcmp(options, "acl", len) == 0)) {
-#ifdef CONFIG_FS_POSIX_ACL
- m->mdt_opts.mo_acl = 1;
- LCONSOLE_INFO("Enabling ACL\n");
-#else
- m->mdt_opts.mo_acl = 0;
- CWARN("ignoring unsupported acl mount option\n");
- LCONSOLE_INFO("Disabling ACL\n");
-#endif
} else if ((len == sizeof("noacl") - 1) &&
(memcmp(options, "noacl", len) == 0)) {
m->mdt_opts.mo_acl = 0;
@@
-3869,7
+3945,7
@@
static int mdt_init0(const struct lu_env *env, struct mdt_device *m,
snprintf(info->mti_u.ns_name, sizeof info->mti_u.ns_name,
LUSTRE_MDT_NAME"-%p", m);
- m->mdt_namespace = ldlm_namespace_new(info->mti_u.ns_name,
+ m->mdt_namespace = ldlm_namespace_new(
obd,
info->mti_u.ns_name,
LDLM_NAMESPACE_SERVER,
LDLM_NAMESPACE_GREEDY);
if (m->mdt_namespace == NULL)
@@
-3892,9
+3968,8
@@
static int mdt_init0(const struct lu_env *env, struct mdt_device *m,
GOTO(err_free_ns, rc);
}
- m->mdt_ck_timer.function = mdt_ck_timer_callback;
- m->mdt_ck_timer.data = (unsigned long)m;
- init_timer(&m->mdt_ck_timer);
+ cfs_timer_init(&m->mdt_ck_timer, mdt_ck_timer_callback, m);
+
rc = mdt_ck_thread_start(m);
if (rc)
GOTO(err_free_ns, rc);
@@
-3932,12
+4007,12
@@
err_fs_cleanup:
target_recovery_fini(obd);
mdt_fs_cleanup(env, m);
err_capa:
-
del_timer
(&m->mdt_ck_timer);
+
cfs_timer_disarm
(&m->mdt_ck_timer);
mdt_ck_thread_stop(m);
err_free_ns:
upcall_cache_cleanup(m->mdt_identity_cache);
m->mdt_identity_cache = NULL;
- ldlm_namespace_free(m->mdt_namespace, 0);
+ ldlm_namespace_free(m->mdt_namespace,
NULL,
0);
obd->obd_namespace = m->mdt_namespace = NULL;
err_fini_seq:
mdt_seq_fini(env, m);
@@
-4185,7
+4260,8
@@
static int mdt_connect_internal(struct obd_export *exp,
static int mdt_obd_connect(const struct lu_env *env,
struct lustre_handle *conn, struct obd_device *obd,
struct obd_uuid *cluuid,
- struct obd_connect_data *data)
+ struct obd_connect_data *data,
+ void *localdata)
{
struct mdt_thread_info *info;
struct mdt_client_data *mcd;
@@
-4343,7
+4419,7
@@
static int mdt_init_export(struct obd_export *exp)
struct mdt_export_data *med = &exp->exp_mdt_data;
ENTRY;
- INIT_LIST_HEAD(&med->med_open_head);
+
CFS_
INIT_LIST_HEAD(&med->med_open_head);
spin_lock_init(&med->med_open_lock);
sema_init(&med->med_idmap_sem, 1);
med->med_idmap = NULL;
@@
-4547,11
+4623,11
@@
int mdt_postrecov(const struct lu_env *env, struct mdt_device *mdt)
struct obd_device *obd = mdt->mdt_md_dev.md_lu_dev.ld_obd;
int rc, lost;
ENTRY;
- /* if some clients didn't participate in recovery then we can possibly
+ /* if some clients didn't participate in recovery then we can possibly
* lost sequence. Now we should increase sequence for safe value */
lost = obd->obd_max_recoverable_clients - obd->obd_connected_clients;
mdt_seq_adjust(env, mdt, lost);
-
+
rc = ld->ld_ops->ldo_recovery_complete(env, ld);
RETURN(rc);
}
@@
-4591,11
+4667,14
@@
static struct lu_device* mdt_device_fini(const struct lu_env *env,
RETURN(NULL);
}
-static void mdt_device_free(const struct lu_env *env, struct lu_device *d)
+static struct lu_device *mdt_device_free(const struct lu_env *env,
+ struct lu_device *d)
{
struct mdt_device *m = mdt_dev(d);
+ ENTRY;
OBD_FREE_PTR(m);
+ RETURN(NULL);
}
static struct lu_device *mdt_device_alloc(const struct lu_env *env,