1 /* -*- mode: c; c-basic-offset: 8; indent-tabs-mode: nil; -*-
2 * vim:expandtab:shiftwidth=8:tabstop=8:
4 * lustre/mdt/mdt_handler.c
5 * Lustre Metadata Target (mdt) request handler
7 * Copyright (c) 2006 Cluster File Systems, Inc.
8 * Author: Peter Braam <braam@clusterfs.com>
9 * Author: Andreas Dilger <adilger@clusterfs.com>
10 * Author: Phil Schwan <phil@clusterfs.com>
11 * Author: Mike Shaver <shaver@clusterfs.com>
12 * Author: Nikita Danilov <nikita@clusterfs.com>
14 * This file is part of the Lustre file system, http://www.lustre.org
15 * Lustre is a trademark of Cluster File Systems, Inc.
17 * You may have signed or agreed to another license before downloading
18 * this software. If so, you are bound by the terms and conditions
19 * of that agreement, and the following does not apply to you. See the
20 * LICENSE file included with this distribution for more information.
22 * If you did not agree to a different license, then this copy of Lustre
23 * is open source software; you can redistribute it and/or modify it
24 * under the terms of version 2 of the GNU General Public License as
25 * published by the Free Software Foundation.
27 * In either case, Lustre is distributed in the hope that it will be
28 * useful, but WITHOUT ANY WARRANTY; without even the implied warranty
29 * of MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
30 * license text for more details.
34 # define EXPORT_SYMTAB
36 #define DEBUG_SUBSYSTEM S_MDS
38 #include <linux/module.h>
40 /* LUSTRE_VERSION_CODE */
41 #include <lustre_ver.h>
43 * struct OBD_{ALLOC,FREE}*()
46 #include <obd_support.h>
47 /* struct ptlrpc_request */
48 #include <lustre_net.h>
49 /* struct obd_export */
50 #include <lustre_export.h>
51 /* struct obd_device */
54 #include <dt_object.h>
55 /* struct mds_client_data */
56 #include "../mds/mds_internal.h"
57 #include "mdt_internal.h"
60 * Initialized in mdt_mod_init().
62 unsigned long mdt_num_threads;
69 int (*mh_act)(struct mdt_thread_info *info);
71 const struct req_format *mh_fmt;
74 enum mdt_handler_flags {
76 * struct mdt_body is passed in the incoming message, and object
77 * identified by this fid exists on disk.
79 HABEO_CORPUS = (1 << 0),
81 * struct ldlm_request is passed in the incoming message.
83 HABEO_CLAVIS = (1 << 1),
85 * this request has fixed reply format, so that reply message can be
86 * packed by generic code.
88 HABEO_REFERO = (1 << 2)
91 struct mdt_opc_slice {
94 struct mdt_handler *mos_hs;
97 static struct mdt_opc_slice mdt_handlers[];
99 static int mdt_handle (struct ptlrpc_request *req);
100 static struct mdt_device *mdt_dev (struct lu_device *d);
101 static int mdt_unpack_req_pack_rep(struct mdt_thread_info *info, __u32 flags);
103 static struct lu_context_key mdt_thread_key;
104 static struct lu_object_operations mdt_obj_ops;
107 static int mdt_getstatus(struct mdt_thread_info *info)
109 struct md_device *next = info->mti_mdt->mdt_child;
111 struct mdt_body *body;
115 if (OBD_FAIL_CHECK(OBD_FAIL_MDS_GETSTATUS_PACK))
118 body = req_capsule_server_get(&info->mti_pill, &RMF_MDT_BODY);
119 result = next->md_ops->mdo_root_get(info->mti_ctxt,
122 body->valid |= OBD_MD_FLID;
125 /* the last_committed and last_xid fields are filled in for all
126 * replies already - no need to do so here also.
131 static int mdt_statfs(struct mdt_thread_info *info)
133 struct md_device *next = info->mti_mdt->mdt_child;
134 struct obd_statfs *osfs;
139 if (OBD_FAIL_CHECK(OBD_FAIL_MDS_STATFS_PACK)) {
140 CERROR(LUSTRE_MDT0_NAME": statfs lustre_pack_reply failed\n");
143 osfs = req_capsule_server_get(&info->mti_pill, &RMF_OBD_STATFS);
144 /* XXX max_age optimisation is needed here. See mds_statfs */
145 result = next->md_ops->mdo_statfs(info->mti_ctxt,
146 next, &info->mti_sfs);
147 statfs_pack(osfs, &info->mti_sfs);
153 void mdt_pack_attr2body(struct mdt_body *b, struct lu_attr *attr)
155 b->valid |= OBD_MD_FLCTIME | OBD_MD_FLUID |
156 OBD_MD_FLGID | OBD_MD_FLFLAGS | OBD_MD_FLTYPE |
157 OBD_MD_FLMODE | OBD_MD_FLNLINK | OBD_MD_FLGENER;
159 if (!S_ISREG(attr->la_mode))
160 b->valid |= OBD_MD_FLSIZE | OBD_MD_FLBLOCKS | OBD_MD_FLATIME |
163 b->atime = attr->la_atime;
164 b->mtime = attr->la_mtime;
165 b->ctime = attr->la_ctime;
166 b->mode = attr->la_mode;
167 b->size = attr->la_size;
168 b->blocks = attr->la_blocks;
169 b->uid = attr->la_uid;
170 b->gid = attr->la_gid;
171 b->flags = attr->la_flags;
172 b->nlink = attr->la_nlink;
175 static int mdt_getattr_pack_msg(struct mdt_thread_info *info)
178 const struct mdt_body *body = info->mti_body;
179 struct req_capsule *pill = &info->mti_pill;
181 struct md_object *next = mdt_object_child(info->mti_object);
182 struct lu_attr *la = &info->mti_attr;
186 rc = mo_attr_get(info->mti_ctxt, next, la);
191 if ((S_ISREG(la->la_mode) && (body->valid & OBD_MD_FLEASIZE)) ||
192 (S_ISDIR(la->la_mode) && (body->valid & OBD_MD_FLDIREA))) {
193 rc = mo_xattr_get(info->mti_ctxt, next, NULL, 0, "lov");
195 CDEBUG(D_INODE, "got %d bytes MD data for object "DFID3"\n",
196 rc, PFID3(mdt_object_fid(info->mti_object)));
198 if (rc != -ENODATA) {
199 CERROR("error getting MD "DFID3": rc = %d\n",
200 PFID3(mdt_object_fid(info->mti_object)),
204 req_capsule_set_size(pill, &RMF_MDT_MD, RCL_SERVER, 0);
205 } else if (rc > MAX_MD_SIZE) {
206 req_capsule_set_size(pill, &RMF_MDT_MD, RCL_SERVER, 0);
207 CERROR("MD size %d larger than maximum possible %u\n",
210 req_capsule_set_size(pill, &RMF_MDT_MD, RCL_SERVER, rc);
212 } else if (S_ISLNK(la->la_mode) && (body->valid & OBD_MD_LINKNAME)) {
213 /* It also uese the mdt_md to hold symname */
214 int len = min_t(int, la->la_size + 1, body->eadatasize);
215 req_capsule_set_size(pill, &RMF_MDT_MD, RCL_SERVER, len);
218 #ifdef CONFIG_FS_POSIX_ACL
219 if ((mdt_info_req(info)->rq_export->exp_connect_flags & OBD_CONNECT_ACL) &&
220 (body->valid & OBD_MD_FLACL)) {
222 rc = mo_xattr_get(info->mti_ctxt, next,
223 NULL, 0, XATTR_NAME_ACL_ACCESS);
225 if (rc != -ENODATA) {
226 CERROR("got acl size: %d\n", rc);
229 req_capsule_set_size(pill, &RMF_EADATA, RCL_SERVER, 0);
231 req_capsule_set_size(pill, &RMF_EADATA, RCL_SERVER, rc);
235 if (OBD_FAIL_CHECK(OBD_FAIL_MDS_GETATTR_PACK)) {
236 CERROR("failed MDT_GETATTR_PACK test\n");
239 rc = req_capsule_pack(&info->mti_pill);
241 CERROR("lustre_pack_reply failed: rc %d\n", rc);
248 static int mdt_getattr_internal(struct mdt_thread_info *info)
250 struct md_object *next = mdt_object_child(info->mti_object);
251 const struct mdt_body *reqbody = info->mti_body;
252 struct mdt_body *repbody;
253 struct lu_attr *la = &info->mti_attr;
261 rc = mo_attr_get(info->mti_ctxt, next, la);
263 CERROR("getattr error for "DFID3": %d\n",
264 PFID3(&reqbody->fid1), rc);
268 repbody = req_capsule_server_get(&info->mti_pill, &RMF_MDT_BODY);
269 mdt_pack_attr2body(repbody, la);
270 repbody->fid1 = *mdt_object_fid(info->mti_object);
271 repbody->valid |= OBD_MD_FLID;
274 buffer = req_capsule_server_get(&info->mti_pill, &RMF_MDT_MD);
275 length = req_capsule_get_size(&info->mti_pill, &RMF_MDT_MD,
278 if ((S_ISREG(la->la_mode) && (reqbody->valid & OBD_MD_FLEASIZE)) ||
279 (S_ISDIR(la->la_mode) && (reqbody->valid & OBD_MD_FLDIREA))) {
280 rc = mo_xattr_get(info->mti_ctxt, next,
281 buffer, length, "lov");
285 if (S_ISDIR(la->la_mode))
286 repbody->valid |= OBD_MD_FLDIREA;
288 repbody->valid |= OBD_MD_FLEASIZE;
289 repbody->eadatasize = rc;
290 } else if (S_ISLNK(la->la_mode) &&
291 (reqbody->valid & OBD_MD_LINKNAME) != 0) {
292 /* FIXME How to readlink??
293 rc = mo_xattr_get(info->mti_ctxt, next,
294 buffer, length, "readlink");
297 CERROR("readlink failed: %d\n", rc);
300 repbody->valid |= OBD_MD_LINKNAME;
301 repbody->eadatasize = rc + 1;
302 ((char*)buffer)[rc] = 0; /* NULL terminate */
303 CDEBUG(D_INODE, "read symlink dest %s\n", (char*)buffer);
307 if (reqbody->valid & OBD_MD_FLMODEASIZE) {
308 repbody->max_cookiesize = MAX_MD_SIZE; /*FIXME*/
309 repbody->max_mdsize = MAX_MD_SIZE;
310 repbody->valid |= OBD_MD_FLMODEASIZE;
314 #ifdef CONFIG_FS_POSIX_ACL
315 if ((mdt_info_req(info)->rq_export->exp_connect_flags & OBD_CONNECT_ACL) &&
316 (reqbody->valid & OBD_MD_FLACL)) {
317 buffer = req_capsule_server_get(&info->mti_pill,
319 length = req_capsule_get_size(&info->mti_pill,
322 rc = mo_xattr_get(info->mti_ctxt, next,
323 buffer, length, XATTR_NAME_ACL_ACCESS);
326 if (rc != -ENODATA) {
327 CERROR("got acl size: %d\n", rc);
332 repbody->aclsize = rc;
333 repbody->valid |= OBD_MD_FLACL;
340 static int mdt_getattr(struct mdt_thread_info *info)
344 LASSERT(info->mti_object != NULL);
345 LASSERT(lu_object_assert_exists(info->mti_ctxt,
346 &info->mti_object->mot_obj.mo_lu));
349 if (OBD_FAIL_CHECK(OBD_FAIL_MDS_GETATTR_PACK)) {
350 CERROR(LUSTRE_MDT0_NAME": getattr lustre_pack_reply failed\n");
353 result = mdt_getattr_pack_msg(info);
355 result = mdt_getattr_internal(info);
361 * UPDATE lock should be taken against parent, and be release before exit;
362 * child_bits lock should be taken against child, and be returned back:
363 * (1)normal request should release the child lock;
364 * (2)intent request will grant the lock to client.
366 static int mdt_getattr_name_lock(struct mdt_thread_info *info,
367 struct mdt_lock_handle *lhc,
370 struct mdt_object *parent = info->mti_object;
371 struct mdt_object *child;
372 struct md_object *next = mdt_object_child(info->mti_object);
375 struct mdt_lock_handle *lhp;
376 struct lu_fid child_fid;
377 struct ldlm_namespace *ns;
381 lhp = &info->mti_lh[MDT_LH_PARENT];
382 lhp->mlh_mode = LCK_CR;
383 lhc->mlh_mode = LCK_CR;
385 LASSERT(info->mti_object != NULL);
387 name = req_capsule_client_get(&info->mti_pill, &RMF_NAME);
391 ns = info->mti_mdt->mdt_namespace;
392 /*step 1: lock parent */
393 result = mdt_object_lock(ns, parent, lhp, MDS_INODELOCK_UPDATE);
397 /*step 2: lookup child's fid by name */
398 result = mdo_lookup(info->mti_ctxt, next, name, &child_fid);
400 GOTO(out_parent, result);
402 /*step 3: find the child object by fid */
403 child = mdt_object_find(info->mti_ctxt, info->mti_mdt, &child_fid);
405 GOTO(out_parent, result = PTR_ERR(child));
407 /*step 4: lock child: this lock is returned back to caller
408 * if successfully get attr.
410 result = mdt_object_lock(ns, child, lhc, child_bits);
412 GOTO(out_child, result);
414 /* finally, we can get attr for child. */
415 result = mdt_getattr_pack_msg(info);
417 struct ldlm_reply *ldlm_rep;
418 ldlm_rep = req_capsule_server_get(&info->mti_pill, &RMF_DLM_REP);
420 intent_set_disposition(ldlm_rep, DISP_IT_EXECD);
421 result = mdt_getattr_internal(info);
423 intent_set_disposition(ldlm_rep, DISP_LOOKUP_NEG);
425 intent_set_disposition(ldlm_rep, DISP_LOOKUP_POS);
428 mdt_object_unlock(ns, child, lhc);
431 mdt_object_put(info->mti_ctxt, child);
433 mdt_object_unlock(ns, parent, lhp);
437 /* normal handler: should release the child lock */
438 static int mdt_getattr_name(struct mdt_thread_info *info)
440 struct mdt_lock_handle lhc = {{0}};
445 rc = mdt_getattr_name_lock(info, &lhc, MDS_INODELOCK_UPDATE);
446 if (rc == 0 && lustre_handle_is_used(&lhc.mlh_lh))
447 ldlm_lock_decref(&lhc.mlh_lh, lhc.mlh_mode);
451 static struct lu_device_operations mdt_lu_ops;
453 static int lu_device_is_mdt(struct lu_device *d)
456 * XXX for now. Tags in lu_device_type->ldt_something are needed.
458 return ergo(d != NULL && d->ld_ops != NULL, d->ld_ops == &mdt_lu_ops);
461 static struct mdt_device *mdt_dev(struct lu_device *d)
463 LASSERT(lu_device_is_mdt(d));
464 return container_of0(d, struct mdt_device, mdt_md_dev.md_lu_dev);
467 static int mdt_connect(struct mdt_thread_info *info)
470 struct req_capsule *pill;
471 struct ptlrpc_request *req;
473 pill = &info->mti_pill;
474 req = mdt_info_req(info);
475 result = target_handle_connect(req, mdt_handle);
477 LASSERT(req->rq_export != NULL);
478 info->mti_mdt = mdt_dev(req->rq_export->exp_obd->obd_lu_dev);
483 static int mdt_disconnect(struct mdt_thread_info *info)
485 return target_handle_disconnect(mdt_info_req(info));
488 static int mdt_readpage(struct mdt_thread_info *info)
493 static int mdt_reint_internal(struct mdt_thread_info *info, __u32 op)
498 OBD_FAIL_RETURN(OBD_FAIL_MDS_REINT_UNPACK, -EFAULT);
500 rc = mdt_reint_unpack(info, op);
502 struct mdt_reint_reply *rep;
504 rep = &info->mti_reint_rep;
505 rep->mrr_body = req_capsule_server_get(&info->mti_pill,
507 if (rep->mrr_body != NULL)
509 * XXX fill other fields in @rep with pointers to
512 rc = mdt_reint_rec(info);
520 static long mdt_reint_opcode(struct mdt_thread_info *info,
521 const struct req_format **fmt)
527 ptr = req_capsule_client_get(&info->mti_pill, &RMF_REINT_OPC);
530 DEBUG_REQ(D_INODE, mdt_info_req(info), "reint opt = %ld", opc);
531 if (opc < REINT_MAX && fmt[opc] != NULL)
532 req_capsule_extend(&info->mti_pill, fmt[opc]);
534 CERROR("Unsupported opc: %ld\n", opc);
539 static int mdt_reint(struct mdt_thread_info *info)
544 struct ptlrpc_request *req;
546 static const struct req_format *reint_fmts[REINT_MAX] = {
547 [REINT_SETATTR] = &RQF_MDS_REINT_SETATTR,
548 [REINT_CREATE] = &RQF_MDS_REINT_CREATE,
549 [REINT_LINK] = &RQF_MDS_REINT_LINK,
550 [REINT_UNLINK] = &RQF_MDS_REINT_UNLINK,
551 [REINT_RENAME] = &RQF_MDS_REINT_RENAME,
552 [REINT_OPEN] = &RQF_MDS_REINT_OPEN
557 req = mdt_info_req(info);
558 opc = mdt_reint_opcode(info, reint_fmts);
560 OBD_FAIL_RETURN(OBD_FAIL_MDS_REINT_NET, 0);
562 rc = req_capsule_pack(&info->mti_pill);
564 rc = mdt_reint_internal(info, opc);
570 static int mdt_close(struct mdt_thread_info *info)
575 static int mdt_done_writing(struct mdt_thread_info *info)
580 static int mdt_pin(struct mdt_thread_info *info)
586 static int mdt_device_sync(struct mdt_device *mdt)
591 static int mdt_object_sync(struct mdt_object *m)
596 static int mdt_sync(struct mdt_thread_info *info)
598 struct mdt_body *body;
599 struct req_capsule *pill = &info->mti_pill;
603 req_capsule_set(pill, &RQF_MDS_SYNC);
605 body = req_capsule_client_get(pill, &RMF_MDT_BODY);
609 if (fid_seq(&body->fid1) == 0) {
610 /* sync the whole device */
611 rc = req_capsule_pack(pill);
613 rc = mdt_device_sync(info->mti_mdt);
616 rc = mdt_unpack_req_pack_rep(info, HABEO_CORPUS | HABEO_REFERO);
620 rc = mdt_object_sync(info->mti_object);
624 rc = mo_attr_get(info->mti_ctxt,
625 mdt_object_child(info->mti_object),
630 body = req_capsule_server_get(pill, &RMF_MDT_BODY);
631 mdt_pack_attr2body(body, &info->mti_attr);
632 body->fid1 = *mdt_object_fid(info->mti_object);
633 body->valid |= OBD_MD_FLID;
638 static int mdt_sync(struct mdt_thread_info *info)
645 static int mdt_handle_quotacheck(struct mdt_thread_info *info)
650 static int mdt_handle_quotactl(struct mdt_thread_info *info)
656 * OBD PING and other handlers.
659 static int mdt_obd_ping(struct mdt_thread_info *info)
663 result = target_handle_ping(mdt_info_req(info));
667 static int mdt_obd_log_cancel(struct mdt_thread_info *info)
672 static int mdt_obd_qc_callback(struct mdt_thread_info *info)
682 static struct ldlm_callback_suite cbs = {
683 .lcs_completion = ldlm_server_completion_ast,
684 .lcs_blocking = ldlm_server_blocking_ast,
688 static int mdt_enqueue(struct mdt_thread_info *info)
691 * info->mti_dlm_req already contains swapped and (if necessary)
692 * converted dlm request.
694 LASSERT(info->mti_dlm_req != NULL);
696 info->mti_fail_id = OBD_FAIL_LDLM_REPLY;
697 return ldlm_handle_enqueue0(info->mti_mdt->mdt_namespace,
699 info->mti_dlm_req, &cbs);
702 static int mdt_convert(struct mdt_thread_info *info)
704 LASSERT(info->mti_dlm_req);
705 return ldlm_handle_convert0(mdt_info_req(info), info->mti_dlm_req);
708 static int mdt_bl_callback(struct mdt_thread_info *info)
710 CERROR("bl callbacks should not happen on MDS\n");
715 static int mdt_cp_callback(struct mdt_thread_info *info)
717 CERROR("cp callbacks should not happen on MDS\n");
723 * Build (DLM) resource name from fid.
725 struct ldlm_res_id *fid_build_res_name(const struct lu_fid *f,
726 struct ldlm_res_id *name)
728 memset(name, 0, sizeof *name);
729 name->name[0] = fid_seq(f);
730 name->name[1] = fid_oid(f);
731 name->name[2] = fid_ver(f);
736 * Return true if resource is for object identified by fid.
738 int fid_res_name_eq(const struct lu_fid *f, const struct ldlm_res_id *name)
740 return name->name[0] == fid_seq(f) &&
741 name->name[1] == fid_oid(f) &&
742 name->name[2] == fid_ver(f);
745 /* issues dlm lock on passed @ns, @f stores it lock handle into @lh. */
746 int fid_lock(struct ldlm_namespace *ns, const struct lu_fid *f,
747 struct lustre_handle *lh, ldlm_mode_t mode,
748 ldlm_policy_data_t *policy)
750 struct ldlm_res_id res_id;
759 /* FIXME: is that correct to have @flags=0 here? */
760 rc = ldlm_cli_enqueue(NULL, NULL, ns, *fid_build_res_name(f, &res_id),
761 LDLM_IBITS, policy, mode, &flags,
762 ldlm_blocking_ast, ldlm_completion_ast, NULL,
763 NULL, NULL, 0, NULL, lh);
764 RETURN(rc == ELDLM_OK ? 0 : -EIO);
767 void fid_unlock(struct ldlm_namespace *ns, const struct lu_fid *f,
768 struct lustre_handle *lh, ldlm_mode_t mode)
770 struct ldlm_lock *lock;
773 /* FIXME: this is debug stuff, remove it later. */
774 lock = ldlm_handle2lock(lh);
776 CERROR("invalid lock handle "LPX64, lh->cookie);
780 LASSERT(fid_res_name_eq(f, &lock->l_resource->lr_name));
782 ldlm_lock_decref(lh, mode);
786 static struct mdt_object *mdt_obj(struct lu_object *o)
788 LASSERT(lu_device_is_mdt(o->lo_dev));
789 return container_of0(o, struct mdt_object, mot_obj.mo_lu);
792 struct mdt_object *mdt_object_find(const struct lu_context *ctxt,
793 struct mdt_device *d,
794 const struct lu_fid *f)
798 o = lu_object_find(ctxt, d->mdt_md_dev.md_lu_dev.ld_site, f);
800 return (struct mdt_object *)o;
805 void mdt_object_put(const struct lu_context *ctxt, struct mdt_object *o)
807 lu_object_put(ctxt, &o->mot_obj.mo_lu);
810 const struct lu_fid *mdt_object_fid(struct mdt_object *o)
812 return lu_object_fid(&o->mot_obj.mo_lu);
815 int mdt_object_lock(struct ldlm_namespace *ns, struct mdt_object *o,
816 struct mdt_lock_handle *lh, __u64 ibits)
818 ldlm_policy_data_t p = {
823 LASSERT(!lustre_handle_is_used(&lh->mlh_lh));
824 LASSERT(lh->mlh_mode != LCK_MINMODE);
826 return fid_lock(ns, mdt_object_fid(o), &lh->mlh_lh, lh->mlh_mode, &p);
829 void mdt_object_unlock(struct ldlm_namespace *ns, struct mdt_object *o,
830 struct mdt_lock_handle *lh)
832 if (lustre_handle_is_used(&lh->mlh_lh)) {
833 fid_unlock(ns, mdt_object_fid(o), &lh->mlh_lh, lh->mlh_mode);
834 lh->mlh_lh.cookie = 0;
838 struct mdt_object *mdt_object_find_lock(const struct lu_context *ctxt,
839 struct mdt_device *d,
840 const struct lu_fid *f,
841 struct mdt_lock_handle *lh,
844 struct mdt_object *o;
846 o = mdt_object_find(ctxt, d, f);
850 result = mdt_object_lock(d->mdt_namespace, o, lh, ibits);
852 mdt_object_put(ctxt, o);
860 static struct mdt_handler *mdt_handler_find(__u32 opc)
862 struct mdt_opc_slice *s;
863 struct mdt_handler *h;
866 for (s = mdt_handlers; s->mos_hs != NULL; s++) {
867 if (s->mos_opc_start <= opc && opc < s->mos_opc_end) {
868 h = s->mos_hs + (opc - s->mos_opc_start);
870 LASSERT(h->mh_opc == opc);
872 h = NULL; /* unsupported opc */
879 static inline __u64 req_exp_last_xid(struct ptlrpc_request *req)
881 return req->rq_export->exp_mds_data.med_mcd->mcd_last_xid;
884 static int mdt_lock_resname_compat(struct mdt_device *m,
885 struct ldlm_request *req)
887 /* XXX something... later. */
891 static int mdt_lock_reply_compat(struct mdt_device *m, struct ldlm_reply *rep)
893 /* XXX something... later. */
898 * Generic code handling requests that have struct mdt_body passed in:
900 * - extract mdt_body from request and save it in @info, if present;
902 * - create lu_object, corresponding to the fid in mdt_body, and save it in
905 * - if HABEO_CORPUS flag is set for this request type check whether object
906 * actually exists on storage (lu_object_exists()).
909 static int mdt_body_unpack(struct mdt_thread_info *info, __u32 flags)
912 const struct mdt_body *body;
913 struct mdt_object *obj;
914 const struct lu_context *ctx;
915 const struct req_capsule *pill;
917 ctx = info->mti_ctxt;
918 pill = &info->mti_pill;
920 body = info->mti_body = req_capsule_client_get(pill, &RMF_MDT_BODY);
922 if (fid_is_sane(&body->fid1)) {
923 obj = mdt_object_find(ctx, info->mti_mdt, &body->fid1);
925 if ((flags & HABEO_CORPUS) &&
926 !lu_object_exists(ctx,
927 &obj->mot_obj.mo_lu)) {
928 mdt_object_put(ctx, obj);
931 info->mti_object = obj;
935 result = PTR_ERR(obj);
937 CERROR("Invalid fid: "DFID3"\n", PFID3(&body->fid1));
945 static int mdt_unpack_req_pack_rep(struct mdt_thread_info *info, __u32 flags)
947 struct req_capsule *pill;
951 pill = &info->mti_pill;
953 if (req_capsule_has_field(pill, &RMF_MDT_BODY))
954 result = mdt_body_unpack(info, flags);
958 if (result == 0 && (flags & HABEO_REFERO))
959 result = req_capsule_pack(pill);
965 * Invoke handler for this request opc. Also do necessary preprocessing
966 * (according to handler ->mh_flags), and post-processing (setting of
967 * ->last_{xid,committed}).
969 static int mdt_req_handle(struct mdt_thread_info *info,
970 struct mdt_handler *h, struct ptlrpc_request *req)
977 LASSERT(h->mh_act != NULL);
978 LASSERT(h->mh_opc == req->rq_reqmsg->opc);
979 LASSERT(current->journal_info == NULL);
981 DEBUG_REQ(D_INODE, req, "%s", h->mh_name);
983 if (h->mh_fail_id != 0)
984 OBD_FAIL_RETURN(h->mh_fail_id, 0);
988 LASSERT(ergo(flags & (HABEO_CORPUS | HABEO_REFERO), h->mh_fmt != NULL));
990 req_capsule_init(&info->mti_pill,
991 req, RCL_SERVER, info->mti_rep_buf_size);
993 if (h->mh_fmt != NULL) {
994 req_capsule_set(&info->mti_pill, h->mh_fmt);
996 result = mdt_unpack_req_pack_rep(info, flags);
999 if (result == 0 && flags & HABEO_CLAVIS) {
1000 struct ldlm_request *dlm_req;
1002 LASSERT(h->mh_fmt != NULL);
1004 dlm_req = req_capsule_client_get(&info->mti_pill, &RMF_DLM_REQ);
1005 if (dlm_req != NULL) {
1006 if (info->mti_mdt->mdt_flags & MDT_CL_COMPAT_RESNAME)
1007 result = mdt_lock_resname_compat(info->mti_mdt,
1009 info->mti_dlm_req = dlm_req;
1011 CERROR("Can't unpack dlm request\n");
1020 result = h->mh_act(info);
1022 * XXX result value is unconditionally shoved into ->rq_status
1023 * (original code sometimes placed error code into ->rq_status, and
1024 * sometimes returned it to the
1025 * caller). ptlrpc_server_handle_request() doesn't check return value
1028 req->rq_status = result;
1030 LASSERT(current->journal_info == NULL);
1032 if (result == 0 && flags & HABEO_CLAVIS &&
1033 info->mti_mdt->mdt_flags & MDT_CL_COMPAT_RESNAME) {
1034 struct ldlm_reply *dlm_rep;
1036 dlm_rep = req_capsule_server_get(&info->mti_pill, &RMF_DLM_REP);
1037 if (dlm_rep != NULL)
1038 result = mdt_lock_reply_compat(info->mti_mdt, dlm_rep);
1041 /* If we're DISCONNECTing, the mds_export_data is already freed */
1042 if (result == 0 && h->mh_opc != MDS_DISCONNECT) {
1043 req->rq_reqmsg->last_xid = le64_to_cpu(req_exp_last_xid(req));
1044 target_committed_to_req(req);
1046 req_capsule_fini(&info->mti_pill);
1050 void mdt_lock_handle_init(struct mdt_lock_handle *lh)
1052 lh->mlh_lh.cookie = 0ull;
1053 lh->mlh_mode = LCK_MINMODE;
1056 void mdt_lock_handle_fini(struct mdt_lock_handle *lh)
1058 LASSERT(!lustre_handle_is_used(&lh->mlh_lh));
1061 static void mdt_thread_info_init(struct mdt_thread_info *info)
1065 info->mti_fail_id = OBD_FAIL_MDS_ALL_REPLY_NET;
1066 for (i = 0; i < ARRAY_SIZE(info->mti_rep_buf_size); i++)
1067 info->mti_rep_buf_size[i] = 0;
1068 info->mti_rep_buf_nr = i;
1069 for (i = 0; i < ARRAY_SIZE(info->mti_lh); i++)
1070 mdt_lock_handle_init(&info->mti_lh[i]);
1073 static void mdt_thread_info_fini(struct mdt_thread_info *info)
1077 if (info->mti_object != NULL) {
1078 mdt_object_put(info->mti_ctxt, info->mti_object);
1079 info->mti_object = NULL;
1081 for (i = 0; i < ARRAY_SIZE(info->mti_lh); i++)
1082 mdt_lock_handle_fini(&info->mti_lh[i]);
1085 static int mds_msg_check_version(struct lustre_msg *msg)
1089 /* TODO: enable the below check while really introducing msg version.
1090 * it's disabled because it will break compatibility with b1_4.
1096 case MDS_DISCONNECT:
1098 rc = lustre_msg_check_version(msg, LUSTRE_OBD_VERSION);
1100 CERROR("bad opc %u version %08x, expecting %08x\n",
1101 msg->opc, msg->version, LUSTRE_OBD_VERSION);
1105 case MDS_GETATTR_NAME:
1110 case MDS_DONE_WRITING:
1116 case MDS_QUOTACHECK:
1120 rc = lustre_msg_check_version(msg, LUSTRE_MDS_VERSION);
1122 CERROR("bad opc %u version %08x, expecting %08x\n",
1123 msg->opc, msg->version, LUSTRE_MDS_VERSION);
1127 case LDLM_BL_CALLBACK:
1128 case LDLM_CP_CALLBACK:
1129 rc = lustre_msg_check_version(msg, LUSTRE_DLM_VERSION);
1131 CERROR("bad opc %u version %08x, expecting %08x\n",
1132 msg->opc, msg->version, LUSTRE_DLM_VERSION);
1134 case OBD_LOG_CANCEL:
1135 case LLOG_ORIGIN_HANDLE_CREATE:
1136 case LLOG_ORIGIN_HANDLE_NEXT_BLOCK:
1137 case LLOG_ORIGIN_HANDLE_PREV_BLOCK:
1138 case LLOG_ORIGIN_HANDLE_READ_HEADER:
1139 case LLOG_ORIGIN_HANDLE_CLOSE:
1141 rc = lustre_msg_check_version(msg, LUSTRE_LOG_VERSION);
1143 CERROR("bad opc %u version %08x, expecting %08x\n",
1144 msg->opc, msg->version, LUSTRE_LOG_VERSION);
1147 CERROR("MDS unknown opcode %d\n", msg->opc);
1153 static int mdt_filter_recovery_request(struct ptlrpc_request *req,
1154 struct obd_device *obd, int *process)
1156 switch (req->rq_reqmsg->opc) {
1157 case MDS_CONNECT: /* This will never get here, but for completeness. */
1158 case OST_CONNECT: /* This will never get here, but for completeness. */
1159 case MDS_DISCONNECT:
1160 case OST_DISCONNECT:
1165 case MDS_SYNC: /* used in unmounting */
1169 *process = target_queue_recovery_request(req, obd);
1173 DEBUG_REQ(D_ERROR, req, "not permitted during recovery");
1175 /* XXX what should we set rq_status to here? */
1176 req->rq_status = -EAGAIN;
1177 RETURN(ptlrpc_error(req));
1182 * Handle recovery. Return:
1183 * +1: continue request processing;
1184 * -ve: abort immediately with the given error code;
1185 * 0: send reply with error code in req->rq_status;
1187 static int mdt_recovery(struct ptlrpc_request *req)
1191 struct obd_device *obd;
1195 if (req->rq_reqmsg->opc == MDS_CONNECT)
1198 if (req->rq_export == NULL) {
1199 CERROR("operation %d on unconnected MDS from %s\n",
1200 req->rq_reqmsg->opc,
1201 libcfs_id2str(req->rq_peer));
1202 req->rq_status = -ENOTCONN;
1206 /* sanity check: if the xid matches, the request must be marked as a
1207 * resent or replayed */
1208 LASSERTF(ergo(req->rq_xid == req_exp_last_xid(req),
1209 lustre_msg_get_flags(req->rq_reqmsg) &
1210 (MSG_RESENT | MSG_REPLAY)),
1211 "rq_xid "LPU64" matches last_xid, "
1212 "expected RESENT flag\n", req->rq_xid);
1214 /* else: note the opposite is not always true; a RESENT req after a
1215 * failover will usually not match the last_xid, since it was likely
1216 * never committed. A REPLAYed request will almost never match the
1217 * last xid, however it could for a committed, but still retained,
1220 obd = req->rq_export->exp_obd;
1222 /* Check for aborted recovery... */
1223 spin_lock_bh(&obd->obd_processing_task_lock);
1224 abort_recovery = obd->obd_abort_recovery;
1225 recovering = obd->obd_recovering;
1226 spin_unlock_bh(&obd->obd_processing_task_lock);
1227 if (abort_recovery) {
1228 target_abort_recovery(obd);
1229 } else if (recovering) {
1233 rc = mdt_filter_recovery_request(req, obd, &should_process);
1234 if (rc != 0 || !should_process) {
1242 static int mdt_reply(struct ptlrpc_request *req, struct mdt_thread_info *info)
1244 struct obd_device *obd;
1246 if (lustre_msg_get_flags(req->rq_reqmsg) & MSG_LAST_REPLAY) {
1247 if (req->rq_reqmsg->opc != OBD_PING)
1248 DEBUG_REQ(D_ERROR, req, "Unexpected MSG_LAST_REPLAY");
1250 obd = req->rq_export != NULL ? req->rq_export->exp_obd : NULL;
1251 if (obd && obd->obd_recovering) {
1252 DEBUG_REQ(D_HA, req, "LAST_REPLAY, queuing reply");
1253 RETURN(target_queue_final_reply(req, req->rq_status));
1255 /* Lost a race with recovery; let the error path
1257 req->rq_status = -ENOTCONN;
1260 target_send_reply(req, req->rq_status, info->mti_fail_id);
1261 RETURN(req->rq_status);
1264 static int mdt_handle0(struct ptlrpc_request *req, struct mdt_thread_info *info)
1266 struct mdt_handler *h;
1267 struct lustre_msg *msg;
1272 OBD_FAIL_RETURN(OBD_FAIL_MDS_ALL_REQUEST_NET | OBD_FAIL_ONCE, 0);
1274 LASSERT(current->journal_info == NULL);
1276 msg = req->rq_reqmsg;
1277 result = mds_msg_check_version(msg);
1279 result = mdt_recovery(req);
1282 h = mdt_handler_find(msg->opc);
1284 result = mdt_req_handle(info, h, req);
1286 req->rq_status = -ENOTSUPP;
1287 result = ptlrpc_error(req);
1292 result = mdt_reply(req, info);
1295 CERROR(LUSTRE_MDT0_NAME" drops mal-formed request\n");
1300 * MDT handler function called by ptlrpc service thread when request comes.
1302 * XXX common "target" functionality should be factored into separate module
1303 * shared by mdt, ost and stand-alone services like fld.
1305 static int mdt_handle(struct ptlrpc_request *req)
1308 struct lu_context *ctx;
1309 struct mdt_thread_info *info;
1312 ctx = req->rq_svc_thread->t_ctx;
1313 LASSERT(ctx != NULL);
1314 LASSERT(ctx->lc_thread == req->rq_svc_thread);
1316 info = lu_context_key_get(ctx, &mdt_thread_key);
1317 LASSERT(info != NULL);
1319 mdt_thread_info_init(info);
1320 /* it can be NULL while CONNECT */
1322 info->mti_mdt = mdt_dev(req->rq_export->exp_obd->obd_lu_dev);
1324 result = mdt_handle0(req, info);
1325 mdt_thread_info_fini(info);
1329 /*Please move these functions from mds to mdt*/
1330 int intent_disposition(struct ldlm_reply *rep, int flag)
1334 return (rep->lock_policy_res1 & flag);
1337 void intent_set_disposition(struct ldlm_reply *rep, int flag)
1341 rep->lock_policy_res1 |= flag;
1357 static int mdt_intent_getattr(enum mdt_it_code opcode,
1358 struct mdt_thread_info *info,
1359 struct ldlm_lock **,
1361 static int mdt_intent_reint(enum mdt_it_code opcode,
1362 struct mdt_thread_info *info,
1363 struct ldlm_lock **,
1366 static struct mdt_it_flavor {
1367 const struct req_format *it_fmt;
1369 int (*it_act)(enum mdt_it_code ,
1370 struct mdt_thread_info *,
1371 struct ldlm_lock **,
1374 } mdt_it_flavor[] = {
1376 .it_fmt = &RQF_LDLM_INTENT,
1377 /*.it_flags = HABEO_REFERO,*/
1379 .it_act = mdt_intent_reint,
1380 .it_reint = REINT_OPEN
1383 .it_fmt = &RQF_LDLM_INTENT,
1385 .it_act = mdt_intent_reint,
1386 .it_reint = REINT_OPEN
1389 .it_fmt = &RQF_LDLM_INTENT,
1391 .it_act = mdt_intent_reint,
1392 .it_reint = REINT_CREATE
1394 [MDT_IT_GETATTR] = {
1395 .it_fmt = &RQF_LDLM_INTENT_GETATTR,
1397 .it_act = mdt_intent_getattr
1399 [MDT_IT_READDIR] = {
1405 .it_fmt = &RQF_LDLM_INTENT_GETATTR,
1407 .it_act = mdt_intent_getattr
1410 .it_fmt = &RQF_LDLM_INTENT_UNLINK,
1412 .it_act = NULL, /* XXX can be mdt_intent_reint, ? */
1413 .it_reint = REINT_UNLINK
1420 [MDT_IT_GETXATTR] = {
1427 static int mdt_intent_getattr(enum mdt_it_code opcode,
1428 struct mdt_thread_info *info,
1429 struct ldlm_lock **lockp,
1433 struct ldlm_lock *old_lock = *lockp;
1434 struct ldlm_lock *new_lock = NULL;
1435 struct ptlrpc_request *req = mdt_info_req(info);
1436 struct ldlm_reply *ldlm_rep;
1437 struct mdt_lock_handle lhc = {{0}};
1444 child_bits = MDS_INODELOCK_LOOKUP;
1446 case MDT_IT_GETATTR:
1447 child_bits = MDS_INODELOCK_LOOKUP | MDS_INODELOCK_UPDATE;
1450 CERROR("Unhandled till now");
1455 rc = mdt_getattr_name_lock(info, &lhc, child_bits);
1458 ldlm_rep = req_capsule_server_get(&info->mti_pill, &RMF_DLM_REP);
1459 ldlm_rep->lock_policy_res2 = rc;
1461 intent_set_disposition(ldlm_rep, DISP_LOOKUP_EXECD);
1463 if (intent_disposition(ldlm_rep, DISP_LOOKUP_NEG))
1464 ldlm_rep->lock_policy_res2 = 0;
1465 if (!intent_disposition(ldlm_rep, DISP_LOOKUP_POS) ||
1466 ldlm_rep->lock_policy_res2) {
1467 RETURN(ELDLM_LOCK_ABORTED);
1470 new_lock = ldlm_handle2lock(&lhc.mlh_lh);
1471 if (new_lock == NULL && (flags & LDLM_FL_INTENT_ONLY))
1476 /* Fixup the lock to be given to the client */
1477 l_lock(&new_lock->l_resource->lr_namespace->ns_lock);
1478 new_lock->l_readers = 0;
1479 new_lock->l_writers = 0;
1481 new_lock->l_export = class_export_get(req->rq_export);
1482 list_add(&new_lock->l_export_chain,
1483 &new_lock->l_export->exp_ldlm_data.led_held_locks);
1485 new_lock->l_blocking_ast = old_lock->l_blocking_ast;
1486 new_lock->l_completion_ast = old_lock->l_completion_ast;
1488 new_lock->l_remote_handle = old_lock->l_remote_handle;
1490 new_lock->l_flags &= ~LDLM_FL_LOCAL;
1492 LDLM_LOCK_PUT(new_lock);
1493 l_unlock(&new_lock->l_resource->lr_namespace->ns_lock);
1495 RETURN(ELDLM_LOCK_REPLACED);
1498 static int mdt_intent_reint(enum mdt_it_code opcode,
1499 struct mdt_thread_info *info,
1500 struct ldlm_lock **lockp,
1504 struct ldlm_reply *rep;
1506 static const struct req_format *intent_fmts[REINT_MAX] = {
1507 [REINT_CREATE] = &RQF_LDLM_INTENT_CREATE,
1508 [REINT_OPEN] = &RQF_LDLM_INTENT_OPEN
1513 opc = mdt_reint_opcode(info, intent_fmts);
1517 if (mdt_it_flavor[opcode].it_reint != opc) {
1518 CERROR("Reint code %ld doesn't match intent: %d\n",
1523 opc = req_capsule_pack(&info->mti_pill);
1527 rep = req_capsule_server_get(&info->mti_pill, &RMF_DLM_REP);
1530 rep->lock_policy_res2 = mdt_reint_internal(info, opc);
1531 intent_set_disposition(rep, DISP_IT_EXECD);
1533 RETURN(ELDLM_LOCK_ABORTED);
1536 static int mdt_intent_code(long itcode)
1542 result = MDT_IT_OPEN;
1544 case IT_OPEN|IT_CREAT:
1545 result = MDT_IT_OCREAT;
1548 result = MDT_IT_CREATE;
1551 result = MDT_IT_READDIR;
1554 result = MDT_IT_GETATTR;
1557 result = MDT_IT_LOOKUP;
1560 result = MDT_IT_UNLINK;
1563 result = MDT_IT_TRUNC;
1566 result = MDT_IT_GETXATTR;
1569 CERROR("Unknown intent opcode: %ld\n", itcode);
1576 static int mdt_intent_opc(long itopc, struct mdt_thread_info *info,
1577 struct ldlm_lock **lockp, int flags)
1582 struct req_capsule *pill;
1583 struct mdt_it_flavor *flv;
1587 opc = mdt_intent_code(itopc);
1591 pill = &info->mti_pill;
1592 flv = &mdt_it_flavor[opc];
1594 if (flv->it_fmt != NULL)
1595 req_capsule_extend(pill, flv->it_fmt);
1597 rc = mdt_unpack_req_pack_rep(info, flv->it_flags);
1602 /* execute policy */
1603 /*XXX LASSERT( flv->it_act) */
1605 rc = flv->it_act(opc, info, lockp, flags);
1612 static int mdt_intent_policy(struct ldlm_namespace *ns,
1613 struct ldlm_lock **lockp, void *req_cookie,
1614 ldlm_mode_t mode, int flags, void *data)
1616 struct mdt_thread_info *info;
1617 struct ptlrpc_request *req = req_cookie;
1618 struct ldlm_intent *it;
1619 struct req_capsule *pill;
1620 struct ldlm_lock *lock = *lockp;
1625 LASSERT(req != NULL);
1627 info = lu_context_key_get(req->rq_svc_thread->t_ctx, &mdt_thread_key);
1628 LASSERT(info != NULL);
1629 pill = &info->mti_pill;
1630 LASSERT(pill->rc_req == req);
1632 if (req->rq_reqmsg->bufcount > MDS_REQ_INTENT_IT_OFF) {
1633 req_capsule_extend(pill, &RQF_LDLM_INTENT);
1634 it = req_capsule_client_get(pill, &RMF_LDLM_INTENT);
1636 LDLM_DEBUG(lock, "intent policy opc: %s",
1637 ldlm_it2str(it->opc));
1639 rc = mdt_intent_opc(it->opc, info, lockp, flags);
1645 /* No intent was provided */
1646 LASSERT(pill->rc_fmt == &RQF_LDLM_ENQUEUE);
1647 rc = req_capsule_pack(pill);
1655 static int mdt_seq_fini(const struct lu_context *ctx,
1656 struct mdt_device *m)
1658 struct lu_site *ls = m->mdt_md_dev.md_lu_dev.ld_site;
1661 if (ls->ls_server_seq) {
1662 seq_server_fini(ls->ls_server_seq, ctx);
1663 OBD_FREE_PTR(ls->ls_server_seq);
1664 ls->ls_server_seq = NULL;
1666 if (ls->ls_client_seq) {
1667 seq_client_fini(ls->ls_client_seq);
1668 OBD_FREE_PTR(ls->ls_client_seq);
1669 ls->ls_client_seq = NULL;
1674 static int mdt_seq_init(const struct lu_context *ctx,
1676 struct mdt_device *m)
1682 ls = m->mdt_md_dev.md_lu_dev.ld_site;
1684 OBD_ALLOC_PTR(ls->ls_server_seq);
1686 if (ls->ls_server_seq != NULL) {
1687 rc = seq_server_init(ls->ls_server_seq,
1688 m->mdt_bottom, uuid,
1694 mdt_seq_fini(ctx, m);
1698 /* XXX: this is ugly, should be something else */
1699 static int mdt_seq_ctlr_init(const struct lu_context *ctx,
1700 struct mdt_device *m,
1701 struct lustre_cfg *cfg)
1703 struct lu_site *ls = m->mdt_md_dev.md_lu_dev.ld_site;
1704 struct obd_device *mdc;
1705 struct obd_uuid uuid;
1710 index = simple_strtol(lustre_cfg_string(cfg, 2), NULL, 10);
1712 /* check if this is first MDC add and controller is not yet
1714 if (index != 0 || ls->ls_controller)
1717 uuid_str = lustre_cfg_string(cfg, 1);
1718 obd_str2uuid(&uuid, uuid_str);
1719 mdc = class_find_client_obd(&uuid, LUSTRE_MDC_NAME, NULL);
1721 CERROR("can't find controller MDC by uuid %s\n",
1724 } else if (!mdc->obd_set_up) {
1725 CERROR("target %s not set up\n", mdc->obd_name);
1728 struct lustre_handle conn = {0, };
1730 CDEBUG(D_CONFIG, "connect to controller %s(%s)\n",
1731 mdc->obd_name, mdc->obd_uuid.uuid);
1733 rc = obd_connect(&conn, mdc, &mdc->obd_uuid, NULL);
1736 CERROR("target %s connect error %d\n",
1739 ls->ls_controller = class_conn2export(&conn);
1741 OBD_ALLOC_PTR(ls->ls_client_seq);
1743 if (ls->ls_client_seq != NULL) {
1744 rc = seq_client_init(ls->ls_client_seq,
1753 LASSERT(ls->ls_server_seq != NULL);
1755 rc = seq_server_set_ctlr(ls->ls_server_seq,
1764 static void mdt_seq_ctlr_fini(struct mdt_device *m)
1770 ls = m->mdt_md_dev.md_lu_dev.ld_site;
1771 if (ls && ls->ls_controller) {
1774 rc = obd_disconnect(ls->ls_controller);
1775 ls->ls_controller = NULL;
1777 CERROR("Failure to disconnect obd: %d\n", rc);
1785 static int mdt_fld_init(const struct lu_context *ctx,
1787 struct mdt_device *m)
1793 ls = m->mdt_md_dev.md_lu_dev.ld_site;
1795 OBD_ALLOC_PTR(ls->ls_fld);
1797 if (ls->ls_fld != NULL)
1798 rc = fld_server_init(ls->ls_fld, ctx,
1799 uuid, m->mdt_bottom);
1806 static int mdt_fld_fini(const struct lu_context *ctx,
1807 struct mdt_device *m)
1809 struct lu_site *ls = m->mdt_md_dev.md_lu_dev.ld_site;
1812 if (ls && ls->ls_fld) {
1813 fld_server_fini(ls->ls_fld, ctx);
1814 OBD_FREE_PTR(ls->ls_fld);
1819 /* device init/fini methods */
1820 static void mdt_stop_ptlrpc_service(struct mdt_device *m)
1822 if (m->mdt_service != NULL) {
1823 ptlrpc_unregister_service(m->mdt_service);
1824 m->mdt_service = NULL;
1828 static int mdt_start_ptlrpc_service(struct mdt_device *m)
1831 struct ptlrpc_service_conf conf = {
1832 .psc_nbufs = MDS_NBUFS,
1833 .psc_bufsize = MDS_BUFSIZE,
1834 .psc_max_req_size = MDS_MAXREQSIZE,
1835 .psc_max_reply_size = MDS_MAXREPSIZE,
1836 .psc_req_portal = MDS_REQUEST_PORTAL,
1837 .psc_rep_portal = MDC_REPLY_PORTAL,
1838 .psc_watchdog_timeout = MDS_SERVICE_WATCHDOG_TIMEOUT,
1840 * We'd like to have a mechanism to set this on a per-device
1841 * basis, but alas...
1843 .psc_num_threads = min(max(mdt_num_threads, MDT_MIN_THREADS),
1850 ptlrpc_init_client(LDLM_CB_REQUEST_PORTAL, LDLM_CB_REPLY_PORTAL,
1851 "mdt_ldlm_client", &m->mdt_ldlm_client);
1854 ptlrpc_init_svc_conf(&conf, mdt_handle, LUSTRE_MDT0_NAME,
1855 m->mdt_md_dev.md_lu_dev.ld_proc_entry,
1857 if (m->mdt_service == NULL)
1860 rc = ptlrpc_start_threads(NULL, m->mdt_service, LUSTRE_MDT0_NAME);
1862 GOTO(err_mdt_svc, rc);
1866 ptlrpc_unregister_service(m->mdt_service);
1867 m->mdt_service = NULL;
1872 static void mdt_stack_fini(const struct lu_context *ctx,
1873 struct mdt_device *m, struct lu_device *d)
1875 /* goes through all stack */
1877 struct lu_device *n;
1878 struct obd_type *type;
1879 struct lu_device_type *ldt = d->ld_type;
1883 /* each fini() returns next device in stack of layers
1884 * * so we can avoid the recursion */
1885 n = ldt->ldt_ops->ldto_device_fini(ctx, d);
1886 ldt->ldt_ops->ldto_device_free(ctx, d);
1888 type = ldt->ldt_obd_type;
1890 class_put_type(type);
1891 /* switch to the next device in the layer */
1894 m->mdt_child = NULL;
1897 static struct lu_device *mdt_layer_setup(const struct lu_context *ctx,
1898 const char *typename,
1899 struct lu_device *child,
1900 struct lustre_cfg *cfg)
1902 struct obd_type *type;
1903 struct lu_device_type *ldt;
1904 struct lu_device *d;
1908 type = class_get_type(typename);
1910 CERROR("Unknown type: '%s'\n", typename);
1911 GOTO(out, rc = -ENODEV);
1916 CERROR("type: '%s'\n", typename);
1917 GOTO(out_type, rc = -EINVAL);
1920 ldt->ldt_obd_type = type;
1921 d = ldt->ldt_ops->ldto_device_alloc(ctx, ldt, cfg);
1923 CERROR("Cannot allocate device: '%s'\n", typename);
1924 GOTO(out_type, rc = -ENODEV);
1927 LASSERT(child->ld_site);
1928 d->ld_site = child->ld_site;
1931 rc = ldt->ldt_ops->ldto_device_init(ctx, d, child);
1933 CERROR("can't init device '%s', rc %d\n", typename, rc);
1934 GOTO(out_alloc, rc);
1940 ldt->ldt_ops->ldto_device_free(ctx, d);
1943 class_put_type(type);
1945 RETURN(ERR_PTR(rc));
1948 static int mdt_stack_init(const struct lu_context *ctx,
1949 struct mdt_device *m, struct lustre_cfg *cfg)
1951 struct lu_device *d = &m->mdt_md_dev.md_lu_dev;
1952 struct lu_device *tmp;
1955 /* init the stack */
1956 tmp = mdt_layer_setup(ctx, LUSTRE_OSD0_NAME, d, cfg);
1958 RETURN (PTR_ERR(tmp));
1960 m->mdt_bottom = lu2dt_dev(tmp);
1962 tmp = mdt_layer_setup(ctx, LUSTRE_MDD0_NAME, d, cfg);
1964 GOTO(out, rc = PTR_ERR(tmp));
1967 tmp = mdt_layer_setup(ctx, LUSTRE_CMM0_NAME, d, cfg);
1969 GOTO(out, rc = PTR_ERR(tmp));
1972 m->mdt_child = lu2md_dev(d);
1974 /* process setup config */
1975 tmp = &m->mdt_md_dev.md_lu_dev;
1976 rc = tmp->ld_ops->ldo_process_config(ctx, tmp, cfg);
1979 /* fini from last known good lu_device */
1981 mdt_stack_fini(ctx, m, d);
1986 static void mdt_fini(struct mdt_device *m)
1988 struct lu_device *d = &m->mdt_md_dev.md_lu_dev;
1989 struct lu_context ctx;
1994 rc = lu_context_init(&ctx);
1996 CERROR("Cannot initialize context: %d\n", rc);
2000 lu_context_enter(&ctx);
2002 mdt_stop_ptlrpc_service(m);
2004 /* finish the stack */
2005 mdt_stack_fini(&ctx, m, md2lu_dev(m->mdt_child));
2007 mdt_fld_fini(&ctx, m);
2008 mdt_seq_fini(&ctx, m);
2009 mdt_seq_ctlr_fini(m);
2011 LASSERT(atomic_read(&d->ld_ref) == 0);
2012 md_device_fini(&m->mdt_md_dev);
2014 if (m->mdt_namespace != NULL) {
2015 ldlm_namespace_free(m->mdt_namespace, 0);
2016 m->mdt_namespace = NULL;
2019 if (d->ld_site != NULL) {
2020 lu_site_fini(d->ld_site);
2021 OBD_FREE_PTR(d->ld_site);
2025 lu_context_exit(&ctx);
2026 lu_context_fini(&ctx);
2031 static int mdt_init0(struct mdt_device *m,
2032 struct lu_device_type *t, struct lustre_cfg *cfg)
2037 struct lu_context ctx;
2038 const char *dev = lustre_cfg_string(cfg, 0);
2039 const char *num = lustre_cfg_string(cfg, 2);
2040 struct obd_device *obd;
2043 obd = class_name2obd(dev);
2044 m->mdt_md_dev.md_lu_dev.ld_obd = obd;
2050 md_device_init(&m->mdt_md_dev, t);
2051 m->mdt_md_dev.md_lu_dev.ld_ops = &mdt_lu_ops;
2053 rc = lu_site_init(s, &m->mdt_md_dev.md_lu_dev);
2055 CERROR("can't init lu_site, rc %d\n", rc);
2056 GOTO(err_fini_site, rc);
2059 rc = lu_context_init(&ctx);
2061 GOTO(err_fini_site, rc);
2063 lu_context_enter(&ctx);
2065 /* init the stack */
2066 rc = mdt_stack_init(&ctx, m, cfg);
2068 CERROR("can't init device stack, rc %d\n", rc);
2069 GOTO(err_fini_ctx, rc);
2071 lu_context_exit(&ctx);
2073 GOTO(err_fini_ctx, rc);
2075 /* set server index */
2077 s->ls_node_id = simple_strtol(num, NULL, 10);
2079 lu_context_enter(&ctx);
2080 rc = mdt_fld_init(&ctx, obd->obd_name, m);
2081 lu_context_exit(&ctx);
2083 GOTO(err_fini_stack, rc);
2085 lu_context_enter(&ctx);
2086 rc = mdt_seq_init(&ctx, obd->obd_name, m);
2087 lu_context_exit(&ctx);
2089 GOTO(err_fini_fld, rc);
2091 lu_context_fini(&ctx);
2093 snprintf(ns_name, sizeof ns_name, LUSTRE_MDT0_NAME"-%p", m);
2094 m->mdt_namespace = ldlm_namespace_new(ns_name, LDLM_NAMESPACE_SERVER);
2095 if (m->mdt_namespace == NULL)
2096 GOTO(err_fini_seq, rc = -ENOMEM);
2098 ldlm_register_intent(m->mdt_namespace, mdt_intent_policy);
2100 rc = mdt_start_ptlrpc_service(m);
2102 GOTO(err_free_ns, rc);
2107 ldlm_namespace_free(m->mdt_namespace, 0);
2108 m->mdt_namespace = NULL;
2110 mdt_seq_fini(&ctx, m);
2112 mdt_fld_fini(&ctx, m);
2114 mdt_stack_fini(&ctx, m, md2lu_dev(m->mdt_child));
2116 lu_context_fini(&ctx);
2123 /* used by MGS to process specific configurations */
2124 static int mdt_process_config(const struct lu_context *ctx,
2125 struct lu_device *d, struct lustre_cfg *cfg)
2127 struct lu_device *next = md2lu_dev(mdt_dev(d)->mdt_child);
2131 switch (cfg->lcfg_command) {
2133 /* add mdc hook to get first MDT uuid and connect it to
2134 * ls->controller to use for seq manager. */
2135 err = mdt_seq_ctlr_init(ctx, mdt_dev(d), cfg);
2137 CERROR("can't initialize controller export, "
2140 /* all MDT specific commands should be here */
2142 /* others are passed further */
2143 err = next->ld_ops->ldo_process_config(ctx, next, cfg);
2148 static struct lu_object *mdt_object_alloc(const struct lu_context *ctxt,
2149 const struct lu_object_header *hdr,
2150 struct lu_device *d)
2152 struct mdt_object *mo;
2158 struct lu_object *o;
2159 struct lu_object_header *h;
2161 o = &mo->mot_obj.mo_lu;
2162 h = &mo->mot_header;
2163 lu_object_header_init(h);
2164 lu_object_init(o, h, d);
2165 lu_object_add_top(h, o);
2166 o->lo_ops = &mdt_obj_ops;
2172 static int mdt_object_init(const struct lu_context *ctxt, struct lu_object *o)
2174 struct mdt_device *d = mdt_dev(o->lo_dev);
2175 struct lu_device *under;
2176 struct lu_object *below;
2178 under = &d->mdt_child->md_lu_dev;
2179 below = under->ld_ops->ldo_object_alloc(ctxt, o->lo_header, under);
2180 if (below != NULL) {
2181 lu_object_add(o, below);
2187 static void mdt_object_free(const struct lu_context *ctxt, struct lu_object *o)
2189 struct mdt_object *mo = mdt_obj(o);
2190 struct lu_object_header *h;
2194 lu_object_header_fini(h);
2199 static int mdt_object_exists(const struct lu_context *ctx,
2200 const struct lu_object *o)
2202 return lu_object_exists(ctx, lu_object_next(o));
2205 static int mdt_object_print(const struct lu_context *ctxt,
2206 struct seq_file *f, const struct lu_object *o)
2208 return seq_printf(f, LUSTRE_MDT0_NAME"-object@%p", o);
2211 static struct lu_device_operations mdt_lu_ops = {
2212 .ldo_object_alloc = mdt_object_alloc,
2213 .ldo_process_config = mdt_process_config
2216 static struct lu_object_operations mdt_obj_ops = {
2217 .loo_object_init = mdt_object_init,
2218 .loo_object_free = mdt_object_free,
2219 .loo_object_print = mdt_object_print,
2220 .loo_object_exists = mdt_object_exists
2223 /* mds_connect_internal */
2224 static int mdt_connect0(struct mdt_device *mdt,
2225 struct obd_export *exp, struct obd_connect_data *data)
2228 data->ocd_connect_flags &= MDT_CONNECT_SUPPORTED;
2229 data->ocd_ibits_known &= MDS_INODELOCK_FULL;
2231 /* If no known bits (which should not happen, probably,
2232 as everybody should support LOOKUP and UPDATE bits at least)
2233 revert to compat mode with plain locks. */
2234 if (!data->ocd_ibits_known &&
2235 data->ocd_connect_flags & OBD_CONNECT_IBITS)
2236 data->ocd_connect_flags &= ~OBD_CONNECT_IBITS;
2238 if (!mdt->mdt_opts.mo_acl)
2239 data->ocd_connect_flags &= ~OBD_CONNECT_ACL;
2241 if (!mdt->mdt_opts.mo_user_xattr)
2242 data->ocd_connect_flags &= ~OBD_CONNECT_XATTR;
2244 exp->exp_connect_flags = data->ocd_connect_flags;
2245 data->ocd_version = LUSTRE_VERSION_CODE;
2246 exp->exp_mds_data.med_ibits_known = data->ocd_ibits_known;
2249 if (mdt->mdt_opts.mo_acl &&
2250 ((exp->exp_connect_flags & OBD_CONNECT_ACL) == 0)) {
2251 CWARN("%s: MDS requires ACL support but client does not\n",
2252 mdt->mdt_md_dev.md_lu_dev.ld_obd->obd_name);
2258 /* mds_connect copy */
2259 static int mdt_obd_connect(struct lustre_handle *conn, struct obd_device *obd,
2260 struct obd_uuid *cluuid,
2261 struct obd_connect_data *data)
2263 struct obd_export *exp;
2265 struct mdt_device *mdt;
2266 struct mds_export_data *med;
2267 struct mds_client_data *mcd = NULL;
2270 if (!conn || !obd || !cluuid)
2273 mdt = mdt_dev(obd->obd_lu_dev);
2275 rc = class_connect(conn, obd, cluuid);
2279 exp = class_conn2export(conn);
2280 LASSERT(exp != NULL);
2281 med = &exp->exp_mds_data;
2283 rc = mdt_connect0(mdt, exp, data);
2287 memcpy(mcd->mcd_uuid, cluuid, sizeof mcd->mcd_uuid);
2293 class_disconnect(exp);
2295 class_export_put(exp);
2300 static int mdt_obd_disconnect(struct obd_export *exp)
2302 struct mds_export_data *med = &exp->exp_mds_data;
2303 unsigned long irqflags;
2308 class_export_get(exp);
2310 /* Disconnect early so that clients can't keep using export */
2311 rc = class_disconnect(exp);
2312 //ldlm_cancel_locks_for_export(exp);
2314 /* complete all outstanding replies */
2315 spin_lock_irqsave(&exp->exp_lock, irqflags);
2316 while (!list_empty(&exp->exp_outstanding_replies)) {
2317 struct ptlrpc_reply_state *rs =
2318 list_entry(exp->exp_outstanding_replies.next,
2319 struct ptlrpc_reply_state, rs_exp_list);
2320 struct ptlrpc_service *svc = rs->rs_service;
2322 spin_lock(&svc->srv_lock);
2323 list_del_init(&rs->rs_exp_list);
2324 ptlrpc_schedule_difficult_reply(rs);
2325 spin_unlock(&svc->srv_lock);
2327 spin_unlock_irqrestore(&exp->exp_lock, irqflags);
2329 OBD_FREE_PTR(med->med_mcd);
2331 class_export_put(exp);
2335 static struct obd_ops mdt_obd_device_ops = {
2336 .o_owner = THIS_MODULE,
2337 .o_connect = mdt_obd_connect,
2338 .o_disconnect = mdt_obd_disconnect,
2341 static void mdt_device_free(const struct lu_context *ctx, struct lu_device *d)
2343 struct mdt_device *m = mdt_dev(d);
2349 static struct lu_device *mdt_device_alloc(const struct lu_context *ctx,
2350 struct lu_device_type *t,
2351 struct lustre_cfg *cfg)
2353 struct lu_device *l;
2354 struct mdt_device *m;
2360 l = &m->mdt_md_dev.md_lu_dev;
2361 result = mdt_init0(m, t, cfg);
2363 mdt_device_free(ctx, l);
2364 return ERR_PTR(result);
2368 l = ERR_PTR(-ENOMEM);
2373 * context key constructor/destructor
2376 static void *mdt_thread_init(const struct lu_context *ctx,
2377 struct lu_context_key *key)
2379 struct mdt_thread_info *info;
2382 * check that no high order allocations are incurred.
2384 CLASSERT(CFS_PAGE_SIZE >= sizeof *info);
2385 OBD_ALLOC_PTR(info);
2387 info->mti_ctxt = ctx;
2389 info = ERR_PTR(-ENOMEM);
2393 static void mdt_thread_fini(const struct lu_context *ctx,
2394 struct lu_context_key *key, void *data)
2396 struct mdt_thread_info *info = data;
2400 static struct lu_context_key mdt_thread_key = {
2401 .lct_init = mdt_thread_init,
2402 .lct_fini = mdt_thread_fini
2405 static int mdt_type_init(struct lu_device_type *t)
2407 return lu_context_key_register(&mdt_thread_key);
2410 static void mdt_type_fini(struct lu_device_type *t)
2412 lu_context_key_degister(&mdt_thread_key);
2415 static struct lu_device_type_operations mdt_device_type_ops = {
2416 .ldto_init = mdt_type_init,
2417 .ldto_fini = mdt_type_fini,
2419 .ldto_device_alloc = mdt_device_alloc,
2420 .ldto_device_free = mdt_device_free
2423 static struct lu_device_type mdt_device_type = {
2424 .ldt_tags = LU_DEVICE_MD,
2425 .ldt_name = LUSTRE_MDT0_NAME,
2426 .ldt_ops = &mdt_device_type_ops
2429 static struct lprocfs_vars lprocfs_mdt_obd_vars[] = {
2433 static struct lprocfs_vars lprocfs_mdt_module_vars[] = {
2437 LPROCFS_INIT_VARS(mdt, lprocfs_mdt_module_vars, lprocfs_mdt_obd_vars);
2439 static int __init mdt_mod_init(void)
2441 struct lprocfs_static_vars lvars;
2443 mdt_num_threads = MDT_NUM_THREADS;
2444 lprocfs_init_vars(mdt, &lvars);
2445 return class_register_type(&mdt_obd_device_ops, NULL,
2446 lvars.module_vars, LUSTRE_MDT0_NAME,
2450 static void __exit mdt_mod_exit(void)
2452 class_unregister_type(LUSTRE_MDT0_NAME);
2456 #define DEF_HNDL(prefix, base, suffix, flags, opc, fn, fmt) \
2457 [prefix ## _ ## opc - prefix ## _ ## base] = { \
2459 .mh_fail_id = OBD_FAIL_ ## prefix ## _ ## opc ## suffix, \
2460 .mh_opc = prefix ## _ ## opc, \
2461 .mh_flags = flags, \
2466 #define DEF_MDT_HNDL(flags, name, fn, fmt) \
2467 DEF_HNDL(MDS, GETATTR, _NET, flags, name, fn, fmt)
2469 * Request with a format known in advance
2471 #define DEF_MDT_HNDL_F(flags, name, fn) \
2472 DEF_HNDL(MDS, GETATTR, _NET, flags, name, fn, &RQF_MDS_ ## name)
2474 * Request with a format we do not yet know
2476 #define DEF_MDT_HNDL_0(flags, name, fn) \
2477 DEF_HNDL(MDS, GETATTR, _NET, flags, name, fn, NULL)
2479 static struct mdt_handler mdt_mds_ops[] = {
2480 DEF_MDT_HNDL_F(0, CONNECT, mdt_connect),
2481 DEF_MDT_HNDL_F(0, DISCONNECT, mdt_disconnect),
2482 DEF_MDT_HNDL_F(0 |HABEO_REFERO, GETSTATUS, mdt_getstatus),
2483 DEF_MDT_HNDL_F(HABEO_CORPUS, GETATTR, mdt_getattr),
2484 DEF_MDT_HNDL_F(HABEO_CORPUS, GETATTR_NAME, mdt_getattr_name),
2485 DEF_MDT_HNDL_F(HABEO_CORPUS|HABEO_REFERO, SETXATTR, mdt_setxattr),
2486 DEF_MDT_HNDL_F(HABEO_CORPUS, GETXATTR, mdt_getxattr),
2487 DEF_MDT_HNDL_F(0 |HABEO_REFERO, STATFS, mdt_statfs),
2488 DEF_MDT_HNDL_0(HABEO_CORPUS, READPAGE, mdt_readpage),
2489 DEF_MDT_HNDL_F(0, REINT, mdt_reint),
2490 DEF_MDT_HNDL_0(HABEO_CORPUS, CLOSE, mdt_close),
2491 DEF_MDT_HNDL_0(0, DONE_WRITING, mdt_done_writing),
2492 DEF_MDT_HNDL_0(0, PIN, mdt_pin),
2493 DEF_MDT_HNDL_0(0, SYNC, mdt_sync),
2494 DEF_MDT_HNDL_0(0, QUOTACHECK, mdt_handle_quotacheck),
2495 DEF_MDT_HNDL_0(0, QUOTACTL, mdt_handle_quotactl)
2498 #define DEF_OBD_HNDL(flags, name, fn) \
2499 DEF_HNDL(OBD, PING, _NET, flags, name, fn, NULL)
2502 static struct mdt_handler mdt_obd_ops[] = {
2503 DEF_OBD_HNDL(0, PING, mdt_obd_ping),
2504 DEF_OBD_HNDL(0, LOG_CANCEL, mdt_obd_log_cancel),
2505 DEF_OBD_HNDL(0, QC_CALLBACK, mdt_obd_qc_callback)
2508 #define DEF_DLM_HNDL_0(flags, name, fn) \
2509 DEF_HNDL(LDLM, ENQUEUE, , flags, name, fn, NULL)
2510 #define DEF_DLM_HNDL_F(flags, name, fn) \
2511 DEF_HNDL(LDLM, ENQUEUE, , flags, name, fn, &RQF_LDLM_ ## name)
2513 static struct mdt_handler mdt_dlm_ops[] = {
2514 DEF_DLM_HNDL_F(HABEO_CLAVIS, ENQUEUE, mdt_enqueue),
2515 DEF_DLM_HNDL_0(HABEO_CLAVIS, CONVERT, mdt_convert),
2516 DEF_DLM_HNDL_0(0, BL_CALLBACK, mdt_bl_callback),
2517 DEF_DLM_HNDL_0(0, CP_CALLBACK, mdt_cp_callback)
2520 static struct mdt_handler mdt_llog_ops[] = {
2523 static struct mdt_opc_slice mdt_handlers[] = {
2525 .mos_opc_start = MDS_GETATTR,
2526 .mos_opc_end = MDS_LAST_OPC,
2527 .mos_hs = mdt_mds_ops
2530 .mos_opc_start = OBD_PING,
2531 .mos_opc_end = OBD_LAST_OPC,
2532 .mos_hs = mdt_obd_ops
2535 .mos_opc_start = LDLM_ENQUEUE,
2536 .mos_opc_end = LDLM_LAST_OPC,
2537 .mos_hs = mdt_dlm_ops
2540 .mos_opc_start = LLOG_ORIGIN_HANDLE_CREATE,
2541 .mos_opc_end = LLOG_LAST_OPC,
2542 .mos_hs = mdt_llog_ops
2549 MODULE_AUTHOR("Cluster File Systems, Inc. <info@clusterfs.com>");
2550 MODULE_DESCRIPTION("Lustre Meta-data Target Prototype ("LUSTRE_MDT0_NAME")");
2551 MODULE_LICENSE("GPL");
2553 CFS_MODULE_PARM(mdt_num_threads, "ul", ulong, 0444,
2554 "number of mdt service threads to start");
2556 cfs_module(mdt, "0.0.4", mdt_mod_init, mdt_mod_exit);