Whamcloud - gitweb
LU-56 ptlrpc: partitioned ptlrpc service
[fs/lustre-release.git] / lustre / mdt / mdt_handler.c
1 /*
2  * GPL HEADER START
3  *
4  * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.
5  *
6  * This program is free software; you can redistribute it and/or modify
7  * it under the terms of the GNU General Public License version 2 only,
8  * as published by the Free Software Foundation.
9  *
10  * This program is distributed in the hope that it will be useful, but
11  * WITHOUT ANY WARRANTY; without even the implied warranty of
12  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
13  * General Public License version 2 for more details (a copy is included
14  * in the LICENSE file that accompanied this code).
15  *
16  * You should have received a copy of the GNU General Public License
17  * version 2 along with this program; If not, see
18  * http://www.sun.com/software/products/lustre/docs/GPLv2.pdf
19  *
20  * Please contact Sun Microsystems, Inc., 4150 Network Circle, Santa Clara,
21  * CA 95054 USA or visit www.sun.com if you need additional information or
22  * have any questions.
23  *
24  * GPL HEADER END
25  */
26 /*
27  * Copyright (c) 2007, 2010, Oracle and/or its affiliates. All rights reserved.
28  * Use is subject to license terms.
29  *
30  * Copyright (c) 2011, 2012, Whamcloud, Inc.
31  */
32 /*
33  * This file is part of Lustre, http://www.lustre.org/
34  * Lustre is a trademark of Sun Microsystems, Inc.
35  *
36  * lustre/mdt/mdt_handler.c
37  *
38  * Lustre Metadata Target (mdt) request handler
39  *
40  * Author: Peter Braam <braam@clusterfs.com>
41  * Author: Andreas Dilger <adilger@clusterfs.com>
42  * Author: Phil Schwan <phil@clusterfs.com>
43  * Author: Mike Shaver <shaver@clusterfs.com>
44  * Author: Nikita Danilov <nikita@clusterfs.com>
45  * Author: Huang Hua <huanghua@clusterfs.com>
46  * Author: Yury Umanets <umka@clusterfs.com>
47  */
48
49 #define DEBUG_SUBSYSTEM S_MDS
50
51 #include <linux/module.h>
52 /*
53  * struct OBD_{ALLOC,FREE}*()
54  */
55 #include <obd_support.h>
56 /* struct ptlrpc_request */
57 #include <lustre_net.h>
58 /* struct obd_export */
59 #include <lustre_export.h>
60 /* struct obd_device */
61 #include <obd.h>
62 /* lu2dt_dev() */
63 #include <dt_object.h>
64 #include <lustre_mds.h>
65 #include <lustre_mdt.h>
66 #include "mdt_internal.h"
67 #ifdef HAVE_QUOTA_SUPPORT
68 # include <lustre_quota.h>
69 #endif
70 #include <lustre_acl.h>
71 #include <lustre_param.h>
72 #include <lustre_fsfilt.h>
73
74 mdl_mode_t mdt_mdl_lock_modes[] = {
75         [LCK_MINMODE] = MDL_MINMODE,
76         [LCK_EX]      = MDL_EX,
77         [LCK_PW]      = MDL_PW,
78         [LCK_PR]      = MDL_PR,
79         [LCK_CW]      = MDL_CW,
80         [LCK_CR]      = MDL_CR,
81         [LCK_NL]      = MDL_NL,
82         [LCK_GROUP]   = MDL_GROUP
83 };
84
85 ldlm_mode_t mdt_dlm_lock_modes[] = {
86         [MDL_MINMODE] = LCK_MINMODE,
87         [MDL_EX]      = LCK_EX,
88         [MDL_PW]      = LCK_PW,
89         [MDL_PR]      = LCK_PR,
90         [MDL_CW]      = LCK_CW,
91         [MDL_CR]      = LCK_CR,
92         [MDL_NL]      = LCK_NL,
93         [MDL_GROUP]   = LCK_GROUP
94 };
95
96 /*
97  * Initialized in mdt_mod_init().
98  */
99 static unsigned long mdt_num_threads;
100 CFS_MODULE_PARM(mdt_num_threads, "ul", ulong, 0444,
101                 "number of mdt service threads to start");
102
103 static char *mdt_cpts;
104 CFS_MODULE_PARM(mdt_cpts, "c", charp, 0444,
105                 "CPU partitions MDT threads should run on");
106
107 static unsigned long mdt_rdpg_num_threads;
108 CFS_MODULE_PARM(mdt_rdpg_num_threads, "ul", ulong, 0444,
109                 "number of mdt readpage service threads to start");
110
111 static char *mdt_rdpg_cpts;
112 CFS_MODULE_PARM(mdt_rdpg_cpts, "c", charp, 0444,
113                 "CPU partitions MDT readpage threads should run on");
114
115 /* NB: these two should be removed along with setattr service in the future */
116 static unsigned long mdt_attr_num_threads;
117 CFS_MODULE_PARM(mdt_attr_num_threads, "ul", ulong, 0444,
118                 "number of mdt setattr service threads to start");
119
120 static char *mdt_attr_cpts;
121 CFS_MODULE_PARM(mdt_attr_cpts, "c", charp, 0444,
122                 "CPU partitions MDT setattr threads should run on");
123
124 /* ptlrpc request handler for MDT. All handlers are
125  * grouped into several slices - struct mdt_opc_slice,
126  * and stored in an array - mdt_handlers[].
127  */
128 struct mdt_handler {
129         /* The name of this handler. */
130         const char *mh_name;
131         /* Fail id for this handler, checked at the beginning of this handler*/
132         int         mh_fail_id;
133         /* Operation code for this handler */
134         __u32       mh_opc;
135         /* flags are listed in enum mdt_handler_flags below. */
136         __u32       mh_flags;
137         /* The actual handler function to execute. */
138         int (*mh_act)(struct mdt_thread_info *info);
139         /* Request format for this request. */
140         const struct req_format *mh_fmt;
141 };
142
143 enum mdt_handler_flags {
144         /*
145          * struct mdt_body is passed in the incoming message, and object
146          * identified by this fid exists on disk.
147          *
148          * "habeo corpus" == "I have a body"
149          */
150         HABEO_CORPUS = (1 << 0),
151         /*
152          * struct ldlm_request is passed in the incoming message.
153          *
154          * "habeo clavis" == "I have a key"
155          */
156         HABEO_CLAVIS = (1 << 1),
157         /*
158          * this request has fixed reply format, so that reply message can be
159          * packed by generic code.
160          *
161          * "habeo refero" == "I have a reply"
162          */
163         HABEO_REFERO = (1 << 2),
164         /*
165          * this request will modify something, so check whether the filesystem
166          * is readonly or not, then return -EROFS to client asap if necessary.
167          *
168          * "mutabor" == "I shall modify"
169          */
170         MUTABOR      = (1 << 3)
171 };
172
173 struct mdt_opc_slice {
174         __u32               mos_opc_start;
175         int                 mos_opc_end;
176         struct mdt_handler *mos_hs;
177 };
178
179 static struct mdt_opc_slice mdt_regular_handlers[];
180 static struct mdt_opc_slice mdt_readpage_handlers[];
181 static struct mdt_opc_slice mdt_xmds_handlers[];
182 static struct mdt_opc_slice mdt_seq_handlers[];
183 static struct mdt_opc_slice mdt_fld_handlers[];
184
185 static struct mdt_device *mdt_dev(struct lu_device *d);
186 static int mdt_regular_handle(struct ptlrpc_request *req);
187 static int mdt_unpack_req_pack_rep(struct mdt_thread_info *info, __u32 flags);
188 static int mdt_fid2path(const struct lu_env *env, struct mdt_device *mdt,
189                         struct getinfo_fid2path *fp);
190
191 static const struct lu_object_operations mdt_obj_ops;
192
193 int mdt_get_disposition(struct ldlm_reply *rep, int flag)
194 {
195         if (!rep)
196                 return 0;
197         return (rep->lock_policy_res1 & flag);
198 }
199
200 void mdt_clear_disposition(struct mdt_thread_info *info,
201                            struct ldlm_reply *rep, int flag)
202 {
203         if (info)
204                 info->mti_opdata &= ~flag;
205         if (rep)
206                 rep->lock_policy_res1 &= ~flag;
207 }
208
209 void mdt_set_disposition(struct mdt_thread_info *info,
210                          struct ldlm_reply *rep, int flag)
211 {
212         if (info)
213                 info->mti_opdata |= flag;
214         if (rep)
215                 rep->lock_policy_res1 |= flag;
216 }
217
218 void mdt_lock_reg_init(struct mdt_lock_handle *lh, ldlm_mode_t lm)
219 {
220         lh->mlh_pdo_hash = 0;
221         lh->mlh_reg_mode = lm;
222         lh->mlh_type = MDT_REG_LOCK;
223 }
224
225 void mdt_lock_pdo_init(struct mdt_lock_handle *lh, ldlm_mode_t lm,
226                        const char *name, int namelen)
227 {
228         lh->mlh_reg_mode = lm;
229         lh->mlh_type = MDT_PDO_LOCK;
230
231         if (name != NULL && (name[0] != '\0')) {
232                 LASSERT(namelen > 0);
233                 lh->mlh_pdo_hash = full_name_hash(name, namelen);
234         } else {
235                 LASSERT(namelen == 0);
236                 lh->mlh_pdo_hash = 0ull;
237         }
238 }
239
240 static void mdt_lock_pdo_mode(struct mdt_thread_info *info, struct mdt_object *o,
241                               struct mdt_lock_handle *lh)
242 {
243         mdl_mode_t mode;
244         ENTRY;
245
246         /*
247          * Any dir access needs couple of locks:
248          *
249          * 1) on part of dir we gonna take lookup/modify;
250          *
251          * 2) on whole dir to protect it from concurrent splitting and/or to
252          * flush client's cache for readdir().
253          *
254          * so, for a given mode and object this routine decides what lock mode
255          * to use for lock #2:
256          *
257          * 1) if caller's gonna lookup in dir then we need to protect dir from
258          * being splitted only - LCK_CR
259          *
260          * 2) if caller's gonna modify dir then we need to protect dir from
261          * being splitted and to flush cache - LCK_CW
262          *
263          * 3) if caller's gonna modify dir and that dir seems ready for
264          * splitting then we need to protect it from any type of access
265          * (lookup/modify/split) - LCK_EX --bzzz
266          */
267
268         LASSERT(lh->mlh_reg_mode != LCK_MINMODE);
269         LASSERT(lh->mlh_pdo_mode == LCK_MINMODE);
270
271         /*
272          * Ask underlaying level its opinion about preferable PDO lock mode
273          * having access type passed as regular lock mode:
274          *
275          * - MDL_MINMODE means that lower layer does not want to specify lock
276          * mode;
277          *
278          * - MDL_NL means that no PDO lock should be taken. This is used in some
279          * cases. Say, for non-splittable directories no need to use PDO locks
280          * at all.
281          */
282         mode = mdo_lock_mode(info->mti_env, mdt_object_child(o),
283                              mdt_dlm_mode2mdl_mode(lh->mlh_reg_mode));
284
285         if (mode != MDL_MINMODE) {
286                 lh->mlh_pdo_mode = mdt_mdl_mode2dlm_mode(mode);
287         } else {
288                 /*
289                  * Lower layer does not want to specify locking mode. We do it
290                  * our selves. No special protection is needed, just flush
291                  * client's cache on modification and allow concurrent
292                  * mondification.
293                  */
294                 switch (lh->mlh_reg_mode) {
295                 case LCK_EX:
296                         lh->mlh_pdo_mode = LCK_EX;
297                         break;
298                 case LCK_PR:
299                         lh->mlh_pdo_mode = LCK_CR;
300                         break;
301                 case LCK_PW:
302                         lh->mlh_pdo_mode = LCK_CW;
303                         break;
304                 default:
305                         CERROR("Not expected lock type (0x%x)\n",
306                                (int)lh->mlh_reg_mode);
307                         LBUG();
308                 }
309         }
310
311         LASSERT(lh->mlh_pdo_mode != LCK_MINMODE);
312         EXIT;
313 }
314
315 static int mdt_getstatus(struct mdt_thread_info *info)
316 {
317         struct mdt_device *mdt  = info->mti_mdt;
318         struct md_device  *next = mdt->mdt_child;
319         struct mdt_body   *repbody;
320         int                rc;
321
322         ENTRY;
323
324         rc = mdt_check_ucred(info);
325         if (rc)
326                 RETURN(err_serious(rc));
327
328         if (OBD_FAIL_CHECK(OBD_FAIL_MDS_GETSTATUS_PACK))
329                 RETURN(err_serious(-ENOMEM));
330
331         repbody = req_capsule_server_get(info->mti_pill, &RMF_MDT_BODY);
332         rc = next->md_ops->mdo_root_get(info->mti_env, next, &repbody->fid1);
333         if (rc != 0)
334                 RETURN(rc);
335
336         repbody->valid |= OBD_MD_FLID;
337
338         if (mdt->mdt_opts.mo_mds_capa &&
339             info->mti_exp->exp_connect_flags & OBD_CONNECT_MDS_CAPA) {
340                 struct mdt_object  *root;
341                 struct lustre_capa *capa;
342
343                 root = mdt_object_find(info->mti_env, mdt, &repbody->fid1);
344                 if (IS_ERR(root))
345                         RETURN(PTR_ERR(root));
346
347                 capa = req_capsule_server_get(info->mti_pill, &RMF_CAPA1);
348                 LASSERT(capa);
349                 capa->lc_opc = CAPA_OPC_MDS_DEFAULT;
350                 rc = mo_capa_get(info->mti_env, mdt_object_child(root), capa,
351                                  0);
352                 mdt_object_put(info->mti_env, root);
353                 if (rc == 0)
354                         repbody->valid |= OBD_MD_FLMDSCAPA;
355         }
356
357         RETURN(rc);
358 }
359
360 static int mdt_statfs(struct mdt_thread_info *info)
361 {
362         struct ptlrpc_request           *req = mdt_info_req(info);
363         struct md_device                *next = info->mti_mdt->mdt_child;
364         struct ptlrpc_service_part      *svcpt;
365         struct obd_statfs               *osfs;
366         int                             rc;
367
368         ENTRY;
369
370         svcpt = info->mti_pill->rc_req->rq_rqbd->rqbd_svcpt;
371
372         /* This will trigger a watchdog timeout */
373         OBD_FAIL_TIMEOUT(OBD_FAIL_MDS_STATFS_LCW_SLEEP,
374                          (MDT_SERVICE_WATCHDOG_FACTOR *
375                           at_get(&svcpt->scp_at_estimate)) + 1);
376
377         rc = mdt_check_ucred(info);
378         if (rc)
379                 RETURN(err_serious(rc));
380
381         if (OBD_FAIL_CHECK(OBD_FAIL_MDS_STATFS_PACK)) {
382                 rc = err_serious(-ENOMEM);
383         } else {
384                 osfs = req_capsule_server_get(info->mti_pill, &RMF_OBD_STATFS);
385                 rc = next->md_ops->mdo_statfs(info->mti_env, next, osfs);
386         }
387
388         if (rc == 0)
389                 mdt_counter_incr(req, LPROC_MDT_STATFS);
390
391         RETURN(rc);
392 }
393
394 /**
395  * Pack SOM attributes into the reply.
396  * Call under a DLM UPDATE lock.
397  */
398 static void mdt_pack_size2body(struct mdt_thread_info *info,
399                                struct mdt_object *mo)
400 {
401         struct mdt_body *b;
402         struct md_attr *ma = &info->mti_attr;
403
404         LASSERT(ma->ma_attr.la_valid & LA_MODE);
405         b = req_capsule_server_get(info->mti_pill, &RMF_MDT_BODY);
406
407         /* Check if Size-on-MDS is supported, if this is a regular file,
408          * if SOM is enabled on the object and if SOM cache exists and valid.
409          * Otherwise do not pack Size-on-MDS attributes to the reply. */
410         if (!(mdt_conn_flags(info) & OBD_CONNECT_SOM) ||
411             !S_ISREG(ma->ma_attr.la_mode) ||
412             !mdt_object_is_som_enabled(mo) ||
413             !(ma->ma_valid & MA_SOM))
414                 return;
415
416         b->valid |= OBD_MD_FLSIZE | OBD_MD_FLBLOCKS;
417         b->size = ma->ma_som->msd_size;
418         b->blocks = ma->ma_som->msd_blocks;
419 }
420
421 void mdt_pack_attr2body(struct mdt_thread_info *info, struct mdt_body *b,
422                         const struct lu_attr *attr, const struct lu_fid *fid)
423 {
424         struct md_attr *ma = &info->mti_attr;
425
426         LASSERT(ma->ma_valid & MA_INODE);
427
428         b->atime      = attr->la_atime;
429         b->mtime      = attr->la_mtime;
430         b->ctime      = attr->la_ctime;
431         b->mode       = attr->la_mode;
432         b->size       = attr->la_size;
433         b->blocks     = attr->la_blocks;
434         b->uid        = attr->la_uid;
435         b->gid        = attr->la_gid;
436         b->flags      = attr->la_flags;
437         b->nlink      = attr->la_nlink;
438         b->rdev       = attr->la_rdev;
439
440         /*XXX should pack the reply body according to lu_valid*/
441         b->valid |= OBD_MD_FLCTIME | OBD_MD_FLUID   |
442                     OBD_MD_FLGID   | OBD_MD_FLTYPE  |
443                     OBD_MD_FLMODE  | OBD_MD_FLNLINK | OBD_MD_FLFLAGS |
444                     OBD_MD_FLATIME | OBD_MD_FLMTIME ;
445
446         if (!S_ISREG(attr->la_mode)) {
447                 b->valid |= OBD_MD_FLSIZE | OBD_MD_FLBLOCKS | OBD_MD_FLRDEV;
448         } else if (ma->ma_need & MA_LOV && ma->ma_lmm_size == 0) {
449                 /* means no objects are allocated on osts. */
450                 LASSERT(!(ma->ma_valid & MA_LOV));
451                 /* just ignore blocks occupied by extend attributes on MDS */
452                 b->blocks = 0;
453                 /* if no object is allocated on osts, the size on mds is valid. b=22272 */
454                 b->valid |= OBD_MD_FLSIZE | OBD_MD_FLBLOCKS;
455         }
456
457         if (fid) {
458                 b->fid1 = *fid;
459                 b->valid |= OBD_MD_FLID;
460
461                 /* FIXME: these should be fixed when new igif ready.*/
462                 b->ino  =  fid_oid(fid);       /* 1.6 compatibility */
463                 b->generation = fid_ver(fid);  /* 1.6 compatibility */
464                 b->valid |= OBD_MD_FLGENER;    /* 1.6 compatibility */
465
466                 CDEBUG(D_INODE, DFID": nlink=%d, mode=%o, size="LPU64"\n",
467                                 PFID(fid), b->nlink, b->mode, b->size);
468         }
469
470         if (info)
471                 mdt_body_reverse_idmap(info, b);
472
473         if (b->valid & OBD_MD_FLSIZE)
474                 CDEBUG(D_VFSTRACE, DFID": returning size %llu\n",
475                        PFID(fid), (unsigned long long)b->size);
476 }
477
478 static inline int mdt_body_has_lov(const struct lu_attr *la,
479                                    const struct mdt_body *body)
480 {
481         return ((S_ISREG(la->la_mode) && (body->valid & OBD_MD_FLEASIZE)) ||
482                 (S_ISDIR(la->la_mode) && (body->valid & OBD_MD_FLDIREA )) );
483 }
484
485 void mdt_client_compatibility(struct mdt_thread_info *info)
486 {
487         struct mdt_body       *body;
488         struct ptlrpc_request *req = mdt_info_req(info);
489         struct obd_export     *exp = req->rq_export;
490         struct md_attr        *ma = &info->mti_attr;
491         struct lu_attr        *la = &ma->ma_attr;
492         ENTRY;
493
494         if (exp->exp_connect_flags & OBD_CONNECT_LAYOUTLOCK)
495                 /* the client can deal with 16-bit lmm_stripe_count */
496                 RETURN_EXIT;
497
498         body = req_capsule_server_get(info->mti_pill, &RMF_MDT_BODY);
499
500         if (!mdt_body_has_lov(la, body))
501                 RETURN_EXIT;
502
503         /* now we have a reply with a lov for a client not compatible with the
504          * layout lock so we have to clean the layout generation number */
505         if (S_ISREG(la->la_mode))
506                 ma->ma_lmm->lmm_layout_gen = 0;
507         EXIT;
508 }
509
510
511 static int mdt_getattr_internal(struct mdt_thread_info *info,
512                                 struct mdt_object *o, int ma_need)
513 {
514         struct md_object        *next = mdt_object_child(o);
515         const struct mdt_body   *reqbody = info->mti_body;
516         struct ptlrpc_request   *req = mdt_info_req(info);
517         struct md_attr          *ma = &info->mti_attr;
518         struct lu_attr          *la = &ma->ma_attr;
519         struct req_capsule      *pill = info->mti_pill;
520         const struct lu_env     *env = info->mti_env;
521         struct mdt_body         *repbody;
522         struct lu_buf           *buffer = &info->mti_buf;
523         int                     rc;
524         ENTRY;
525
526         if (OBD_FAIL_CHECK(OBD_FAIL_MDS_GETATTR_PACK))
527                 RETURN(err_serious(-ENOMEM));
528
529         repbody = req_capsule_server_get(pill, &RMF_MDT_BODY);
530
531         ma->ma_valid = 0;
532
533         rc = mdt_object_exists(o);
534         if (rc < 0) {
535                 /* This object is located on remote node.*/
536                 repbody->fid1 = *mdt_object_fid(o);
537                 repbody->valid = OBD_MD_FLID | OBD_MD_MDS;
538                 GOTO(out, rc = 0);
539         }
540
541         buffer->lb_buf = req_capsule_server_get(pill, &RMF_MDT_MD);
542         buffer->lb_len = req_capsule_get_size(pill, &RMF_MDT_MD, RCL_SERVER);
543
544         /* If it is dir object and client require MEA, then we got MEA */
545         if (S_ISDIR(lu_object_attr(&next->mo_lu)) &&
546             reqbody->valid & OBD_MD_MEA) {
547                 /* Assumption: MDT_MD size is enough for lmv size. */
548                 ma->ma_lmv = buffer->lb_buf;
549                 ma->ma_lmv_size = buffer->lb_len;
550                 ma->ma_need = MA_LMV | MA_INODE;
551         } else {
552                 ma->ma_lmm = buffer->lb_buf;
553                 ma->ma_lmm_size = buffer->lb_len;
554                 ma->ma_need = MA_LOV | MA_INODE;
555         }
556
557         if (S_ISDIR(lu_object_attr(&next->mo_lu)) &&
558             reqbody->valid & OBD_MD_FLDIREA  &&
559             lustre_msg_get_opc(req->rq_reqmsg) == MDS_GETATTR) {
560                 /* get default stripe info for this dir. */
561                 ma->ma_need |= MA_LOV_DEF;
562         }
563         ma->ma_need |= ma_need;
564         if (ma->ma_need & MA_SOM)
565                 ma->ma_som = &info->mti_u.som.data;
566
567         rc = mo_attr_get(env, next, ma);
568         if (unlikely(rc)) {
569                 CERROR("getattr error for "DFID": %d\n",
570                         PFID(mdt_object_fid(o)), rc);
571                 RETURN(rc);
572         }
573
574         if (likely(ma->ma_valid & MA_INODE))
575                 mdt_pack_attr2body(info, repbody, la, mdt_object_fid(o));
576         else
577                 RETURN(-EFAULT);
578
579         if (mdt_body_has_lov(la, reqbody)) {
580                 if (ma->ma_valid & MA_LOV) {
581                         LASSERT(ma->ma_lmm_size);
582                         mdt_dump_lmm(D_INFO, ma->ma_lmm);
583                         repbody->eadatasize = ma->ma_lmm_size;
584                         if (S_ISDIR(la->la_mode))
585                                 repbody->valid |= OBD_MD_FLDIREA;
586                         else
587                                 repbody->valid |= OBD_MD_FLEASIZE;
588                 }
589                 if (ma->ma_valid & MA_LMV) {
590                         LASSERT(S_ISDIR(la->la_mode));
591                         repbody->eadatasize = ma->ma_lmv_size;
592                         repbody->valid |= (OBD_MD_FLDIREA|OBD_MD_MEA);
593                 }
594         } else if (S_ISLNK(la->la_mode) &&
595                    reqbody->valid & OBD_MD_LINKNAME) {
596                 buffer->lb_buf = ma->ma_lmm;
597                 /* eadatasize from client includes NULL-terminator, so
598                  * there is no need to read it */
599                 buffer->lb_len = reqbody->eadatasize - 1;
600                 rc = mo_readlink(env, next, buffer);
601                 if (unlikely(rc <= 0)) {
602                         CERROR("readlink failed: %d\n", rc);
603                         rc = -EFAULT;
604                 } else {
605                         int print_limit = min_t(int, CFS_PAGE_SIZE - 128, rc);
606
607                         if (OBD_FAIL_CHECK(OBD_FAIL_MDS_READLINK_EPROTO))
608                                 rc -= 2;
609                         repbody->valid |= OBD_MD_LINKNAME;
610                         /* we need to report back size with NULL-terminator
611                          * because client expects that */
612                         repbody->eadatasize = rc + 1;
613                         if (repbody->eadatasize != reqbody->eadatasize)
614                                 CERROR("Read shorter symlink %d, expected %d\n",
615                                        rc, reqbody->eadatasize - 1);
616                         /* NULL terminate */
617                         ((char *)ma->ma_lmm)[rc] = 0;
618
619                         /* If the total CDEBUG() size is larger than a page, it
620                          * will print a warning to the console, avoid this by
621                          * printing just the last part of the symlink. */
622                         CDEBUG(D_INODE, "symlink dest %s%.*s, len = %d\n",
623                                print_limit < rc ? "..." : "", print_limit,
624                                (char *)ma->ma_lmm + rc - print_limit, rc);
625                         rc = 0;
626                 }
627         }
628
629         if (reqbody->valid & OBD_MD_FLMODEASIZE) {
630                 repbody->max_cookiesize = info->mti_mdt->mdt_max_cookiesize;
631                 repbody->max_mdsize = info->mti_mdt->mdt_max_mdsize;
632                 repbody->valid |= OBD_MD_FLMODEASIZE;
633                 CDEBUG(D_INODE, "I am going to change the MAX_MD_SIZE & "
634                        "MAX_COOKIE to : %d:%d\n", repbody->max_mdsize,
635                        repbody->max_cookiesize);
636         }
637
638         if (exp_connect_rmtclient(info->mti_exp) &&
639             reqbody->valid & OBD_MD_FLRMTPERM) {
640                 void *buf = req_capsule_server_get(pill, &RMF_ACL);
641
642                 /* mdt_getattr_lock only */
643                 rc = mdt_pack_remote_perm(info, o, buf);
644                 if (rc) {
645                         repbody->valid &= ~OBD_MD_FLRMTPERM;
646                         repbody->aclsize = 0;
647                         RETURN(rc);
648                 } else {
649                         repbody->valid |= OBD_MD_FLRMTPERM;
650                         repbody->aclsize = sizeof(struct mdt_remote_perm);
651                 }
652         }
653 #ifdef CONFIG_FS_POSIX_ACL
654         else if ((req->rq_export->exp_connect_flags & OBD_CONNECT_ACL) &&
655                  (reqbody->valid & OBD_MD_FLACL)) {
656                 buffer->lb_buf = req_capsule_server_get(pill, &RMF_ACL);
657                 buffer->lb_len = req_capsule_get_size(pill,
658                                                       &RMF_ACL, RCL_SERVER);
659                 if (buffer->lb_len > 0) {
660                         rc = mo_xattr_get(env, next, buffer,
661                                           XATTR_NAME_ACL_ACCESS);
662                         if (rc < 0) {
663                                 if (rc == -ENODATA) {
664                                         repbody->aclsize = 0;
665                                         repbody->valid |= OBD_MD_FLACL;
666                                         rc = 0;
667                                 } else if (rc == -EOPNOTSUPP) {
668                                         rc = 0;
669                                 } else {
670                                         CERROR("got acl size: %d\n", rc);
671                                 }
672                         } else {
673                                 repbody->aclsize = rc;
674                                 repbody->valid |= OBD_MD_FLACL;
675                                 rc = 0;
676                         }
677                 }
678         }
679 #endif
680
681         if (reqbody->valid & OBD_MD_FLMDSCAPA &&
682             info->mti_mdt->mdt_opts.mo_mds_capa &&
683             info->mti_exp->exp_connect_flags & OBD_CONNECT_MDS_CAPA) {
684                 struct lustre_capa *capa;
685
686                 capa = req_capsule_server_get(pill, &RMF_CAPA1);
687                 LASSERT(capa);
688                 capa->lc_opc = CAPA_OPC_MDS_DEFAULT;
689                 rc = mo_capa_get(env, next, capa, 0);
690                 if (rc)
691                         RETURN(rc);
692                 repbody->valid |= OBD_MD_FLMDSCAPA;
693         }
694
695 out:
696         if (rc == 0)
697                 mdt_counter_incr(req, LPROC_MDT_GETATTR);
698
699         RETURN(rc);
700 }
701
702 static int mdt_renew_capa(struct mdt_thread_info *info)
703 {
704         struct mdt_object  *obj = info->mti_object;
705         struct mdt_body    *body;
706         struct lustre_capa *capa, *c;
707         int rc;
708         ENTRY;
709
710         /* if object doesn't exist, or server has disabled capability,
711          * return directly, client will find body->valid OBD_MD_FLOSSCAPA
712          * flag not set.
713          */
714         if (!obj || !info->mti_mdt->mdt_opts.mo_oss_capa ||
715             !(info->mti_exp->exp_connect_flags & OBD_CONNECT_OSS_CAPA))
716                 RETURN(0);
717
718         body = req_capsule_server_get(info->mti_pill, &RMF_MDT_BODY);
719         LASSERT(body != NULL);
720
721         c = req_capsule_client_get(info->mti_pill, &RMF_CAPA1);
722         LASSERT(c);
723
724         capa = req_capsule_server_get(info->mti_pill, &RMF_CAPA2);
725         LASSERT(capa);
726
727         *capa = *c;
728         rc = mo_capa_get(info->mti_env, mdt_object_child(obj), capa, 1);
729         if (rc == 0)
730                 body->valid |= OBD_MD_FLOSSCAPA;
731         RETURN(rc);
732 }
733
734 static int mdt_getattr(struct mdt_thread_info *info)
735 {
736         struct mdt_object       *obj = info->mti_object;
737         struct req_capsule      *pill = info->mti_pill;
738         struct mdt_body         *reqbody;
739         struct mdt_body         *repbody;
740         mode_t                   mode;
741         int rc, rc2;
742         ENTRY;
743
744         reqbody = req_capsule_client_get(pill, &RMF_MDT_BODY);
745         LASSERT(reqbody);
746
747         if (reqbody->valid & OBD_MD_FLOSSCAPA) {
748                 rc = req_capsule_server_pack(pill);
749                 if (unlikely(rc))
750                         RETURN(err_serious(rc));
751                 rc = mdt_renew_capa(info);
752                 GOTO(out_shrink, rc);
753         }
754
755         LASSERT(obj != NULL);
756         LASSERT(lu_object_assert_exists(&obj->mot_obj.mo_lu));
757
758         mode = lu_object_attr(&obj->mot_obj.mo_lu);
759
760         /* old clients may not report needed easize, use max value then */
761         req_capsule_set_size(pill, &RMF_MDT_MD, RCL_SERVER,
762                              reqbody->eadatasize == 0 ?
763                              info->mti_mdt->mdt_max_mdsize :
764                              reqbody->eadatasize);
765
766         rc = req_capsule_server_pack(pill);
767         if (unlikely(rc != 0))
768                 RETURN(err_serious(rc));
769
770         repbody = req_capsule_server_get(pill, &RMF_MDT_BODY);
771         LASSERT(repbody != NULL);
772         repbody->eadatasize = 0;
773         repbody->aclsize = 0;
774
775         if (reqbody->valid & OBD_MD_FLRMTPERM)
776                 rc = mdt_init_ucred(info, reqbody);
777         else
778                 rc = mdt_check_ucred(info);
779         if (unlikely(rc))
780                 GOTO(out_shrink, rc);
781
782         info->mti_spec.sp_ck_split = !!(reqbody->valid & OBD_MD_FLCKSPLIT);
783         info->mti_cross_ref = !!(reqbody->valid & OBD_MD_FLCROSSREF);
784
785         /*
786          * Don't check capability at all, because rename might getattr for
787          * remote obj, and at that time no capability is available.
788          */
789         mdt_set_capainfo(info, 1, &reqbody->fid1, BYPASS_CAPA);
790         rc = mdt_getattr_internal(info, obj, 0);
791         if (reqbody->valid & OBD_MD_FLRMTPERM)
792                 mdt_exit_ucred(info);
793         EXIT;
794 out_shrink:
795         mdt_client_compatibility(info);
796         rc2 = mdt_fix_reply(info);
797         if (rc == 0)
798                 rc = rc2;
799         return rc;
800 }
801
802 static int mdt_is_subdir(struct mdt_thread_info *info)
803 {
804         struct mdt_object     *o = info->mti_object;
805         struct req_capsule    *pill = info->mti_pill;
806         const struct mdt_body *body = info->mti_body;
807         struct mdt_body       *repbody;
808         int                    rc;
809         ENTRY;
810
811         LASSERT(o != NULL);
812
813         repbody = req_capsule_server_get(pill, &RMF_MDT_BODY);
814
815         /*
816          * We save last checked parent fid to @repbody->fid1 for remote
817          * directory case.
818          */
819         LASSERT(fid_is_sane(&body->fid2));
820         LASSERT(mdt_object_exists(o) > 0);
821         rc = mdo_is_subdir(info->mti_env, mdt_object_child(o),
822                            &body->fid2, &repbody->fid1);
823         if (rc == 0 || rc == -EREMOTE)
824                 repbody->valid |= OBD_MD_FLID;
825
826         RETURN(rc);
827 }
828
829 static int mdt_raw_lookup(struct mdt_thread_info *info,
830                           struct mdt_object *parent,
831                           const struct lu_name *lname,
832                           struct ldlm_reply *ldlm_rep)
833 {
834         struct md_object *next = mdt_object_child(info->mti_object);
835         const struct mdt_body *reqbody = info->mti_body;
836         struct lu_fid *child_fid = &info->mti_tmp_fid1;
837         struct mdt_body *repbody;
838         int rc;
839         ENTRY;
840
841         if (reqbody->valid != OBD_MD_FLID)
842                 RETURN(0);
843
844         LASSERT(!info->mti_cross_ref);
845
846         /* Only got the fid of this obj by name */
847         fid_zero(child_fid);
848         rc = mdo_lookup(info->mti_env, next, lname, child_fid,
849                         &info->mti_spec);
850 #if 0
851         /* XXX is raw_lookup possible as intent operation? */
852         if (rc != 0) {
853                 if (rc == -ENOENT)
854                         mdt_set_disposition(info, ldlm_rep, DISP_LOOKUP_NEG);
855                 RETURN(rc);
856         } else
857                 mdt_set_disposition(info, ldlm_rep, DISP_LOOKUP_POS);
858
859         repbody = req_capsule_server_get(info->mti_pill, &RMF_MDT_BODY);
860 #endif
861         if (rc == 0) {
862                 repbody = req_capsule_server_get(info->mti_pill, &RMF_MDT_BODY);
863                 repbody->fid1 = *child_fid;
864                 repbody->valid = OBD_MD_FLID;
865         }
866         RETURN(1);
867 }
868
869 /*
870  * UPDATE lock should be taken against parent, and be release before exit;
871  * child_bits lock should be taken against child, and be returned back:
872  *            (1)normal request should release the child lock;
873  *            (2)intent request will grant the lock to client.
874  */
875 static int mdt_getattr_name_lock(struct mdt_thread_info *info,
876                                  struct mdt_lock_handle *lhc,
877                                  __u64 child_bits,
878                                  struct ldlm_reply *ldlm_rep)
879 {
880         struct ptlrpc_request  *req       = mdt_info_req(info);
881         struct mdt_body        *reqbody   = NULL;
882         struct mdt_object      *parent    = info->mti_object;
883         struct mdt_object      *child;
884         struct md_object       *next      = mdt_object_child(parent);
885         struct lu_fid          *child_fid = &info->mti_tmp_fid1;
886         struct lu_name         *lname     = NULL;
887         const char             *name      = NULL;
888         int                     namelen   = 0;
889         struct mdt_lock_handle *lhp       = NULL;
890         struct ldlm_lock       *lock;
891         struct ldlm_res_id     *res_id;
892         int                     is_resent;
893         int                     ma_need = 0;
894         int                     rc;
895
896         ENTRY;
897
898         is_resent = lustre_handle_is_used(&lhc->mlh_reg_lh);
899         LASSERT(ergo(is_resent,
900                      lustre_msg_get_flags(req->rq_reqmsg) & MSG_RESENT));
901
902         LASSERT(parent != NULL);
903         name = req_capsule_client_get(info->mti_pill, &RMF_NAME);
904         if (name == NULL)
905                 RETURN(err_serious(-EFAULT));
906
907         namelen = req_capsule_get_size(info->mti_pill, &RMF_NAME,
908                                        RCL_CLIENT) - 1;
909         if (!info->mti_cross_ref) {
910                 /*
911                  * XXX: Check for "namelen == 0" is for getattr by fid
912                  * (OBD_CONNECT_ATTRFID), otherwise do not allow empty name,
913                  * that is the name must contain at least one character and
914                  * the terminating '\0'
915                  */
916                 if (namelen == 0) {
917                         reqbody = req_capsule_client_get(info->mti_pill,
918                                                          &RMF_MDT_BODY);
919                         if (unlikely(reqbody == NULL))
920                                 RETURN(err_serious(-EFAULT));
921
922                         if (unlikely(!fid_is_sane(&reqbody->fid2)))
923                                 RETURN(err_serious(-EINVAL));
924
925                         name = NULL;
926                         CDEBUG(D_INODE, "getattr with lock for "DFID"/"DFID", "
927                                "ldlm_rep = %p\n",
928                                PFID(mdt_object_fid(parent)),
929                                PFID(&reqbody->fid2), ldlm_rep);
930                 } else {
931                         lname = mdt_name(info->mti_env, (char *)name, namelen);
932                         CDEBUG(D_INODE, "getattr with lock for "DFID"/%s, "
933                                "ldlm_rep = %p\n", PFID(mdt_object_fid(parent)),
934                                name, ldlm_rep);
935                 }
936         }
937         mdt_set_disposition(info, ldlm_rep, DISP_LOOKUP_EXECD);
938
939         rc = mdt_object_exists(parent);
940         if (unlikely(rc == 0)) {
941                 LU_OBJECT_DEBUG(D_INODE, info->mti_env,
942                                 &parent->mot_obj.mo_lu,
943                                 "Parent doesn't exist!\n");
944                 RETURN(-ESTALE);
945         } else if (!info->mti_cross_ref) {
946                 LASSERTF(rc > 0, "Parent "DFID" is on remote server\n",
947                          PFID(mdt_object_fid(parent)));
948         }
949         if (lname) {
950                 rc = mdt_raw_lookup(info, parent, lname, ldlm_rep);
951                 if (rc != 0) {
952                         if (rc > 0)
953                                 rc = 0;
954                         RETURN(rc);
955                 }
956         }
957
958         if (info->mti_cross_ref) {
959                 /* Only getattr on the child. Parent is on another node. */
960                 mdt_set_disposition(info, ldlm_rep, DISP_LOOKUP_POS);
961                 child = parent;
962                 CDEBUG(D_INODE, "partial getattr_name child_fid = "DFID", "
963                        "ldlm_rep=%p\n", PFID(mdt_object_fid(child)), ldlm_rep);
964
965                 if (is_resent) {
966                         /* Do not take lock for resent case. */
967                         lock = ldlm_handle2lock(&lhc->mlh_reg_lh);
968                         LASSERTF(lock != NULL, "Invalid lock handle "LPX64"\n",
969                                  lhc->mlh_reg_lh.cookie);
970                         LASSERT(fid_res_name_eq(mdt_object_fid(child),
971                                                 &lock->l_resource->lr_name));
972                         LDLM_LOCK_PUT(lock);
973                         rc = 0;
974                 } else {
975                         mdt_lock_handle_init(lhc);
976                         mdt_lock_reg_init(lhc, LCK_PR);
977
978                         /*
979                          * Object's name is on another MDS, no lookup lock is
980                          * needed here but update is.
981                          */
982                         child_bits &= ~MDS_INODELOCK_LOOKUP;
983                         child_bits |= MDS_INODELOCK_UPDATE;
984
985                         rc = mdt_object_lock(info, child, lhc, child_bits,
986                                              MDT_LOCAL_LOCK);
987                 }
988                 if (rc == 0) {
989                         /* Finally, we can get attr for child. */
990                         mdt_set_capainfo(info, 0, mdt_object_fid(child),
991                                          BYPASS_CAPA);
992                         rc = mdt_getattr_internal(info, child, 0);
993                         if (unlikely(rc != 0))
994                                 mdt_object_unlock(info, child, lhc, 1);
995                 }
996                 RETURN(rc);
997         }
998
999         if (lname) {
1000                 /* step 1: lock parent only if parent is a directory */
1001                 if (S_ISDIR(lu_object_attr(&parent->mot_obj.mo_lu))) {
1002                         lhp = &info->mti_lh[MDT_LH_PARENT];
1003                         mdt_lock_pdo_init(lhp, LCK_PR, name, namelen);
1004                         rc = mdt_object_lock(info, parent, lhp,
1005                                              MDS_INODELOCK_UPDATE,
1006                                              MDT_LOCAL_LOCK);
1007                         if (unlikely(rc != 0))
1008                                 RETURN(rc);
1009                 }
1010
1011                 /* step 2: lookup child's fid by name */
1012                 fid_zero(child_fid);
1013                 rc = mdo_lookup(info->mti_env, next, lname, child_fid,
1014                                 &info->mti_spec);
1015
1016                 if (rc != 0) {
1017                         if (rc == -ENOENT)
1018                                 mdt_set_disposition(info, ldlm_rep,
1019                                                     DISP_LOOKUP_NEG);
1020                         GOTO(out_parent, rc);
1021                 } else
1022                         mdt_set_disposition(info, ldlm_rep, DISP_LOOKUP_POS);
1023         } else {
1024                 *child_fid = reqbody->fid2;
1025                 mdt_set_disposition(info, ldlm_rep, DISP_LOOKUP_POS);
1026         }
1027
1028         /*
1029          *step 3: find the child object by fid & lock it.
1030          *        regardless if it is local or remote.
1031          */
1032         child = mdt_object_find(info->mti_env, info->mti_mdt, child_fid);
1033
1034         if (unlikely(IS_ERR(child)))
1035                 GOTO(out_parent, rc = PTR_ERR(child));
1036         if (is_resent) {
1037                 /* Do not take lock for resent case. */
1038                 lock = ldlm_handle2lock(&lhc->mlh_reg_lh);
1039                 LASSERTF(lock != NULL, "Invalid lock handle "LPX64"\n",
1040                          lhc->mlh_reg_lh.cookie);
1041
1042                 res_id = &lock->l_resource->lr_name;
1043                 if (!fid_res_name_eq(mdt_object_fid(child),
1044                                     &lock->l_resource->lr_name)) {
1045                          LASSERTF(fid_res_name_eq(mdt_object_fid(parent),
1046                                                  &lock->l_resource->lr_name),
1047                                  "Lock res_id: %lu/%lu/%lu, Fid: "DFID".\n",
1048                                  (unsigned long)res_id->name[0],
1049                                  (unsigned long)res_id->name[1],
1050                                  (unsigned long)res_id->name[2],
1051                                  PFID(mdt_object_fid(parent)));
1052                           CWARN("Although resent, but still not get child lock"
1053                                 "parent:"DFID" child:"DFID"\n",
1054                                 PFID(mdt_object_fid(parent)),
1055                                 PFID(mdt_object_fid(child)));
1056                           lustre_msg_clear_flags(req->rq_reqmsg, MSG_RESENT);
1057                           LDLM_LOCK_PUT(lock);
1058                           GOTO(relock, 0);
1059                 }
1060                 LDLM_LOCK_PUT(lock);
1061                 rc = 0;
1062         } else {
1063 relock:
1064                 OBD_FAIL_TIMEOUT(OBD_FAIL_MDS_RESEND, obd_timeout*2);
1065                 mdt_lock_handle_init(lhc);
1066                 if (child_bits == MDS_INODELOCK_LAYOUT)
1067                         mdt_lock_reg_init(lhc, LCK_CR);
1068                 else
1069                         mdt_lock_reg_init(lhc, LCK_PR);
1070
1071                 if (mdt_object_exists(child) == 0) {
1072                         LU_OBJECT_DEBUG(D_INODE, info->mti_env,
1073                                         &child->mot_obj.mo_lu,
1074                                         "Object doesn't exist!\n");
1075                         GOTO(out_child, rc = -ENOENT);
1076                 }
1077
1078                 if (!(child_bits & MDS_INODELOCK_UPDATE)) {
1079                         struct md_attr *ma = &info->mti_attr;
1080
1081                         ma->ma_valid = 0;
1082                         ma->ma_need = MA_INODE;
1083                         rc = mo_attr_get(info->mti_env,
1084                                          mdt_object_child(child), ma);
1085                         if (unlikely(rc != 0))
1086                                 GOTO(out_child, rc);
1087
1088                         /* layout lock is used only on regular files */
1089                         if ((ma->ma_valid & MA_INODE) &&
1090                             (ma->ma_attr.la_valid & LA_MODE) &&
1091                             !S_ISREG(ma->ma_attr.la_mode))
1092                                 child_bits &= ~MDS_INODELOCK_LAYOUT;
1093
1094                         /* If the file has not been changed for some time, we
1095                          * return not only a LOOKUP lock, but also an UPDATE
1096                          * lock and this might save us RPC on later STAT. For
1097                          * directories, it also let negative dentry starts
1098                          * working for this dir. */
1099                         if (ma->ma_valid & MA_INODE &&
1100                             ma->ma_attr.la_valid & LA_CTIME &&
1101                             info->mti_mdt->mdt_namespace->ns_ctime_age_limit +
1102                                 ma->ma_attr.la_ctime < cfs_time_current_sec())
1103                                 child_bits |= MDS_INODELOCK_UPDATE;
1104                 }
1105
1106                 rc = mdt_object_lock(info, child, lhc, child_bits,
1107                                      MDT_CROSS_LOCK);
1108
1109                 if (unlikely(rc != 0))
1110                         GOTO(out_child, rc);
1111         }
1112
1113         lock = ldlm_handle2lock(&lhc->mlh_reg_lh);
1114         /* Get MA_SOM attributes if update lock is given. */
1115         if (lock &&
1116             lock->l_policy_data.l_inodebits.bits & MDS_INODELOCK_UPDATE &&
1117             S_ISREG(lu_object_attr(&mdt_object_child(child)->mo_lu)))
1118                 ma_need = MA_SOM;
1119
1120         /* finally, we can get attr for child. */
1121         mdt_set_capainfo(info, 1, child_fid, BYPASS_CAPA);
1122         rc = mdt_getattr_internal(info, child, ma_need);
1123         if (unlikely(rc != 0)) {
1124                 mdt_object_unlock(info, child, lhc, 1);
1125         } else if (lock) {
1126                 /* Debugging code. */
1127                 res_id = &lock->l_resource->lr_name;
1128                 LDLM_DEBUG(lock, "Returning lock to client");
1129                 LASSERTF(fid_res_name_eq(mdt_object_fid(child),
1130                                          &lock->l_resource->lr_name),
1131                          "Lock res_id: %lu/%lu/%lu, Fid: "DFID".\n",
1132                          (unsigned long)res_id->name[0],
1133                          (unsigned long)res_id->name[1],
1134                          (unsigned long)res_id->name[2],
1135                          PFID(mdt_object_fid(child)));
1136                 mdt_pack_size2body(info, child);
1137         }
1138         if (lock)
1139                 LDLM_LOCK_PUT(lock);
1140
1141         EXIT;
1142 out_child:
1143         mdt_object_put(info->mti_env, child);
1144 out_parent:
1145         if (lhp)
1146                 mdt_object_unlock(info, parent, lhp, 1);
1147         return rc;
1148 }
1149
1150 /* normal handler: should release the child lock */
1151 static int mdt_getattr_name(struct mdt_thread_info *info)
1152 {
1153         struct mdt_lock_handle *lhc = &info->mti_lh[MDT_LH_CHILD];
1154         struct mdt_body        *reqbody;
1155         struct mdt_body        *repbody;
1156         int rc, rc2;
1157         ENTRY;
1158
1159         reqbody = req_capsule_client_get(info->mti_pill, &RMF_MDT_BODY);
1160         LASSERT(reqbody != NULL);
1161         repbody = req_capsule_server_get(info->mti_pill, &RMF_MDT_BODY);
1162         LASSERT(repbody != NULL);
1163
1164         info->mti_spec.sp_ck_split = !!(reqbody->valid & OBD_MD_FLCKSPLIT);
1165         info->mti_cross_ref = !!(reqbody->valid & OBD_MD_FLCROSSREF);
1166         repbody->eadatasize = 0;
1167         repbody->aclsize = 0;
1168
1169         rc = mdt_init_ucred(info, reqbody);
1170         if (unlikely(rc))
1171                 GOTO(out_shrink, rc);
1172
1173         rc = mdt_getattr_name_lock(info, lhc, MDS_INODELOCK_UPDATE, NULL);
1174         if (lustre_handle_is_used(&lhc->mlh_reg_lh)) {
1175                 ldlm_lock_decref(&lhc->mlh_reg_lh, lhc->mlh_reg_mode);
1176                 lhc->mlh_reg_lh.cookie = 0;
1177         }
1178         mdt_exit_ucred(info);
1179         EXIT;
1180 out_shrink:
1181         mdt_client_compatibility(info);
1182         rc2 = mdt_fix_reply(info);
1183         if (rc == 0)
1184                 rc = rc2;
1185         return rc;
1186 }
1187
1188 static const struct lu_device_operations mdt_lu_ops;
1189
1190 static int lu_device_is_mdt(struct lu_device *d)
1191 {
1192         return ergo(d != NULL && d->ld_ops != NULL, d->ld_ops == &mdt_lu_ops);
1193 }
1194
1195 static int mdt_iocontrol(unsigned int cmd, struct obd_export *exp, int len,
1196                          void *karg, void *uarg);
1197
1198 static int mdt_set_info(struct mdt_thread_info *info)
1199 {
1200         struct ptlrpc_request *req = mdt_info_req(info);
1201         char *key;
1202         void *val;
1203         int keylen, vallen, rc = 0;
1204         ENTRY;
1205
1206         rc = req_capsule_server_pack(info->mti_pill);
1207         if (rc)
1208                 RETURN(rc);
1209
1210         key = req_capsule_client_get(info->mti_pill, &RMF_SETINFO_KEY);
1211         if (key == NULL) {
1212                 DEBUG_REQ(D_HA, req, "no set_info key");
1213                 RETURN(-EFAULT);
1214         }
1215
1216         keylen = req_capsule_get_size(info->mti_pill, &RMF_SETINFO_KEY,
1217                                       RCL_CLIENT);
1218
1219         val = req_capsule_client_get(info->mti_pill, &RMF_SETINFO_VAL);
1220         if (val == NULL) {
1221                 DEBUG_REQ(D_HA, req, "no set_info val");
1222                 RETURN(-EFAULT);
1223         }
1224
1225         vallen = req_capsule_get_size(info->mti_pill, &RMF_SETINFO_VAL,
1226                                       RCL_CLIENT);
1227
1228         /* Swab any part of val you need to here */
1229         if (KEY_IS(KEY_READ_ONLY)) {
1230                 req->rq_status = 0;
1231                 lustre_msg_set_status(req->rq_repmsg, 0);
1232
1233                 cfs_spin_lock(&req->rq_export->exp_lock);
1234                 if (*(__u32 *)val)
1235                         req->rq_export->exp_connect_flags |= OBD_CONNECT_RDONLY;
1236                 else
1237                         req->rq_export->exp_connect_flags &=~OBD_CONNECT_RDONLY;
1238                 cfs_spin_unlock(&req->rq_export->exp_lock);
1239
1240         } else if (KEY_IS(KEY_CHANGELOG_CLEAR)) {
1241                 struct changelog_setinfo *cs =
1242                         (struct changelog_setinfo *)val;
1243                 if (vallen != sizeof(*cs)) {
1244                         CERROR("Bad changelog_clear setinfo size %d\n", vallen);
1245                         RETURN(-EINVAL);
1246                 }
1247                 if (ptlrpc_req_need_swab(req)) {
1248                         __swab64s(&cs->cs_recno);
1249                         __swab32s(&cs->cs_id);
1250                 }
1251
1252                 rc = mdt_iocontrol(OBD_IOC_CHANGELOG_CLEAR, info->mti_exp,
1253                                    vallen, val, NULL);
1254                 lustre_msg_set_status(req->rq_repmsg, rc);
1255
1256         } else {
1257                 RETURN(-EINVAL);
1258         }
1259         RETURN(0);
1260 }
1261
1262 static int mdt_connect(struct mdt_thread_info *info)
1263 {
1264         int rc;
1265         struct ptlrpc_request *req;
1266
1267         req = mdt_info_req(info);
1268         rc = target_handle_connect(req);
1269         if (rc == 0) {
1270                 LASSERT(req->rq_export != NULL);
1271                 info->mti_mdt = mdt_dev(req->rq_export->exp_obd->obd_lu_dev);
1272                 rc = mdt_init_sec_level(info);
1273                 if (rc == 0)
1274                         rc = mdt_init_idmap(info);
1275                 if (rc != 0)
1276                         obd_disconnect(class_export_get(req->rq_export));
1277         } else {
1278                 rc = err_serious(rc);
1279         }
1280         return rc;
1281 }
1282
1283 static int mdt_disconnect(struct mdt_thread_info *info)
1284 {
1285         int rc;
1286         ENTRY;
1287
1288         rc = target_handle_disconnect(mdt_info_req(info));
1289         if (rc)
1290                 rc = err_serious(rc);
1291         RETURN(rc);
1292 }
1293
1294 static int mdt_sendpage(struct mdt_thread_info *info,
1295                         struct lu_rdpg *rdpg, int nob)
1296 {
1297         struct ptlrpc_request   *req = mdt_info_req(info);
1298         struct obd_export       *exp = req->rq_export;
1299         struct ptlrpc_bulk_desc *desc;
1300         struct l_wait_info      *lwi = &info->mti_u.rdpg.mti_wait_info;
1301         int                      tmpcount;
1302         int                      tmpsize;
1303         int                      i;
1304         int                      rc;
1305         ENTRY;
1306
1307         desc = ptlrpc_prep_bulk_exp(req, rdpg->rp_npages, BULK_PUT_SOURCE,
1308                                     MDS_BULK_PORTAL);
1309         if (desc == NULL)
1310                 RETURN(-ENOMEM);
1311
1312         for (i = 0, tmpcount = nob;
1313                 i < rdpg->rp_npages && tmpcount > 0; i++, tmpcount -= tmpsize) {
1314                 tmpsize = min_t(int, tmpcount, CFS_PAGE_SIZE);
1315                 ptlrpc_prep_bulk_page(desc, rdpg->rp_pages[i], 0, tmpsize);
1316         }
1317
1318         LASSERT(desc->bd_nob == nob);
1319         rc = target_bulk_io(exp, desc, lwi);
1320         ptlrpc_free_bulk(desc);
1321         RETURN(rc);
1322 }
1323
1324 #ifdef HAVE_SPLIT_SUPPORT
1325 /*
1326  * Retrieve dir entry from the page and insert it to the slave object, actually,
1327  * this should be in osd layer, but since it will not in the final product, so
1328  * just do it here and do not define more moo api anymore for this.
1329  */
1330 static int mdt_write_dir_page(struct mdt_thread_info *info, struct page *page,
1331                               int size)
1332 {
1333         struct mdt_object *object = info->mti_object;
1334         struct lu_fid *lf = &info->mti_tmp_fid2;
1335         struct md_attr *ma = &info->mti_attr;
1336         struct lu_dirpage *dp;
1337         struct lu_dirent *ent;
1338         int rc = 0, offset = 0;
1339         ENTRY;
1340
1341         /* Make sure we have at least one entry. */
1342         if (size == 0)
1343                 RETURN(-EINVAL);
1344
1345         /*
1346          * Disable trans for this name insert, since it will include many trans
1347          * for this.
1348          */
1349         info->mti_no_need_trans = 1;
1350         /*
1351          * When write_dir_page, no need update parent's ctime,
1352          * and no permission check for name_insert.
1353          */
1354         ma->ma_attr.la_ctime = 0;
1355         ma->ma_attr.la_valid = LA_MODE;
1356         ma->ma_valid = MA_INODE;
1357
1358         cfs_kmap(page);
1359         dp = page_address(page);
1360         offset = (int)((__u32)lu_dirent_start(dp) - (__u32)dp);
1361
1362         for (ent = lu_dirent_start(dp); ent != NULL;
1363              ent = lu_dirent_next(ent)) {
1364                 struct lu_name *lname;
1365                 char *name;
1366
1367                 if (le16_to_cpu(ent->lde_namelen) == 0)
1368                         continue;
1369
1370                 fid_le_to_cpu(lf, &ent->lde_fid);
1371                 if (le64_to_cpu(ent->lde_hash) & MAX_HASH_HIGHEST_BIT)
1372                         ma->ma_attr.la_mode = S_IFDIR;
1373                 else
1374                         ma->ma_attr.la_mode = 0;
1375                 OBD_ALLOC(name, le16_to_cpu(ent->lde_namelen) + 1);
1376                 if (name == NULL)
1377                         GOTO(out, rc = -ENOMEM);
1378
1379                 memcpy(name, ent->lde_name, le16_to_cpu(ent->lde_namelen));
1380                 lname = mdt_name(info->mti_env, name,
1381                                  le16_to_cpu(ent->lde_namelen));
1382                 ma->ma_attr_flags |= (MDS_PERM_BYPASS | MDS_QUOTA_IGNORE);
1383                 rc = mdo_name_insert(info->mti_env,
1384                                      md_object_next(&object->mot_obj),
1385                                      lname, lf, ma);
1386                 OBD_FREE(name, le16_to_cpu(ent->lde_namelen) + 1);
1387                 if (rc) {
1388                         CERROR("Can't insert %*.*s, rc %d\n",
1389                                le16_to_cpu(ent->lde_namelen),
1390                                le16_to_cpu(ent->lde_namelen),
1391                                ent->lde_name, rc);
1392                         GOTO(out, rc);
1393                 }
1394
1395                 offset += lu_dirent_size(ent);
1396                 if (offset >= size)
1397                         break;
1398         }
1399         EXIT;
1400 out:
1401         cfs_kunmap(page);
1402         return rc;
1403 }
1404
1405 static int mdt_bulk_timeout(void *data)
1406 {
1407         ENTRY;
1408
1409         CERROR("mdt bulk transfer timeout \n");
1410
1411         RETURN(1);
1412 }
1413
1414 static int mdt_writepage(struct mdt_thread_info *info)
1415 {
1416         struct ptlrpc_request   *req = mdt_info_req(info);
1417         struct mdt_body         *reqbody;
1418         struct l_wait_info      *lwi;
1419         struct ptlrpc_bulk_desc *desc;
1420         struct page             *page;
1421         int                rc;
1422         ENTRY;
1423
1424
1425         reqbody = req_capsule_client_get(info->mti_pill, &RMF_MDT_BODY);
1426         if (reqbody == NULL)
1427                 RETURN(err_serious(-EFAULT));
1428
1429         desc = ptlrpc_prep_bulk_exp(req, 1, BULK_GET_SINK, MDS_BULK_PORTAL);
1430         if (desc == NULL)
1431                 RETURN(err_serious(-ENOMEM));
1432
1433         /* allocate the page for the desc */
1434         page = cfs_alloc_page(CFS_ALLOC_STD);
1435         if (page == NULL)
1436                 GOTO(desc_cleanup, rc = -ENOMEM);
1437
1438         CDEBUG(D_INFO, "Received page offset %d size %d \n",
1439                (int)reqbody->size, (int)reqbody->nlink);
1440
1441         ptlrpc_prep_bulk_page(desc, page, (int)reqbody->size,
1442                               (int)reqbody->nlink);
1443
1444         rc = sptlrpc_svc_prep_bulk(req, desc);
1445         if (rc != 0)
1446                 GOTO(cleanup_page, rc);
1447         /*
1448          * Check if client was evicted while we were doing i/o before touching
1449          * network.
1450          */
1451         OBD_ALLOC_PTR(lwi);
1452         if (!lwi)
1453                 GOTO(cleanup_page, rc = -ENOMEM);
1454
1455         if (desc->bd_export->exp_failed)
1456                 rc = -ENOTCONN;
1457         else
1458                 rc = ptlrpc_start_bulk_transfer (desc);
1459         if (rc == 0) {
1460                 *lwi = LWI_TIMEOUT_INTERVAL(obd_timeout * CFS_HZ / 4, CFS_HZ,
1461                                             mdt_bulk_timeout, desc);
1462                 rc = l_wait_event(desc->bd_waitq, !ptlrpc_bulk_active(desc) ||
1463                                   desc->bd_export->exp_failed, lwi);
1464                 LASSERT(rc == 0 || rc == -ETIMEDOUT);
1465                 if (rc == -ETIMEDOUT) {
1466                         DEBUG_REQ(D_ERROR, req, "timeout on bulk GET");
1467                         ptlrpc_abort_bulk(desc);
1468                 } else if (desc->bd_export->exp_failed) {
1469                         DEBUG_REQ(D_ERROR, req, "Eviction on bulk GET");
1470                         rc = -ENOTCONN;
1471                         ptlrpc_abort_bulk(desc);
1472                 } else if (!desc->bd_success ||
1473                            desc->bd_nob_transferred != desc->bd_nob) {
1474                         DEBUG_REQ(D_ERROR, req, "%s bulk GET %d(%d)",
1475                                   desc->bd_success ?
1476                                   "truncated" : "network error on",
1477                                   desc->bd_nob_transferred, desc->bd_nob);
1478                         /* XXX should this be a different errno? */
1479                         rc = -ETIMEDOUT;
1480                 }
1481         } else {
1482                 DEBUG_REQ(D_ERROR, req, "ptlrpc_bulk_get failed: rc %d", rc);
1483         }
1484         if (rc)
1485                 GOTO(cleanup_lwi, rc);
1486         rc = mdt_write_dir_page(info, page, reqbody->nlink);
1487
1488 cleanup_lwi:
1489         OBD_FREE_PTR(lwi);
1490 cleanup_page:
1491         cfs_free_page(page);
1492 desc_cleanup:
1493         ptlrpc_free_bulk(desc);
1494         RETURN(rc);
1495 }
1496 #endif
1497
1498 static int mdt_readpage(struct mdt_thread_info *info)
1499 {
1500         struct mdt_object *object = info->mti_object;
1501         struct lu_rdpg    *rdpg = &info->mti_u.rdpg.mti_rdpg;
1502         struct mdt_body   *reqbody;
1503         struct mdt_body   *repbody;
1504         int                rc;
1505         int                i;
1506         ENTRY;
1507
1508         if (OBD_FAIL_CHECK(OBD_FAIL_MDS_READPAGE_PACK))
1509                 RETURN(err_serious(-ENOMEM));
1510
1511         reqbody = req_capsule_client_get(info->mti_pill, &RMF_MDT_BODY);
1512         repbody = req_capsule_server_get(info->mti_pill, &RMF_MDT_BODY);
1513         if (reqbody == NULL || repbody == NULL)
1514                 RETURN(err_serious(-EFAULT));
1515
1516         /*
1517          * prepare @rdpg before calling lower layers and transfer itself. Here
1518          * reqbody->size contains offset of where to start to read and
1519          * reqbody->nlink contains number bytes to read.
1520          */
1521         rdpg->rp_hash = reqbody->size;
1522         if (rdpg->rp_hash != reqbody->size) {
1523                 CERROR("Invalid hash: "LPX64" != "LPX64"\n",
1524                        rdpg->rp_hash, reqbody->size);
1525                 RETURN(-EFAULT);
1526         }
1527
1528         rdpg->rp_attrs = reqbody->mode;
1529         if (info->mti_exp->exp_connect_flags & OBD_CONNECT_64BITHASH)
1530                 rdpg->rp_attrs |= LUDA_64BITHASH;
1531         rdpg->rp_count  = min_t(unsigned int, reqbody->nlink,
1532                                 PTLRPC_MAX_BRW_SIZE);
1533         rdpg->rp_npages = (rdpg->rp_count + CFS_PAGE_SIZE - 1) >>
1534                           CFS_PAGE_SHIFT;
1535         OBD_ALLOC(rdpg->rp_pages, rdpg->rp_npages * sizeof rdpg->rp_pages[0]);
1536         if (rdpg->rp_pages == NULL)
1537                 RETURN(-ENOMEM);
1538
1539         for (i = 0; i < rdpg->rp_npages; ++i) {
1540                 rdpg->rp_pages[i] = cfs_alloc_page(CFS_ALLOC_STD);
1541                 if (rdpg->rp_pages[i] == NULL)
1542                         GOTO(free_rdpg, rc = -ENOMEM);
1543         }
1544
1545         /* call lower layers to fill allocated pages with directory data */
1546         rc = mo_readpage(info->mti_env, mdt_object_child(object), rdpg);
1547         if (rc < 0)
1548                 GOTO(free_rdpg, rc);
1549
1550         /* send pages to client */
1551         rc = mdt_sendpage(info, rdpg, rc);
1552
1553         EXIT;
1554 free_rdpg:
1555
1556         for (i = 0; i < rdpg->rp_npages; i++)
1557                 if (rdpg->rp_pages[i] != NULL)
1558                         cfs_free_page(rdpg->rp_pages[i]);
1559         OBD_FREE(rdpg->rp_pages, rdpg->rp_npages * sizeof rdpg->rp_pages[0]);
1560
1561         if (OBD_FAIL_CHECK(OBD_FAIL_MDS_SENDPAGE))
1562                 RETURN(0);
1563
1564         return rc;
1565 }
1566
1567 static int mdt_reint_internal(struct mdt_thread_info *info,
1568                               struct mdt_lock_handle *lhc,
1569                               __u32 op)
1570 {
1571         struct req_capsule      *pill = info->mti_pill;
1572         struct md_quota         *mq = md_quota(info->mti_env);
1573         struct mdt_body         *repbody;
1574         int                      rc = 0, rc2;
1575         ENTRY;
1576
1577
1578         rc = mdt_reint_unpack(info, op);
1579         if (rc != 0) {
1580                 CERROR("Can't unpack reint, rc %d\n", rc);
1581                 RETURN(err_serious(rc));
1582         }
1583
1584         /* for replay (no_create) lmm is not needed, client has it already */
1585         if (req_capsule_has_field(pill, &RMF_MDT_MD, RCL_SERVER))
1586                 req_capsule_set_size(pill, &RMF_MDT_MD, RCL_SERVER,
1587                                      info->mti_rr.rr_eadatalen);
1588
1589         if (req_capsule_has_field(pill, &RMF_LOGCOOKIES, RCL_SERVER))
1590                 req_capsule_set_size(pill, &RMF_LOGCOOKIES, RCL_SERVER,
1591                                      info->mti_mdt->mdt_max_cookiesize);
1592
1593         rc = req_capsule_server_pack(pill);
1594         if (rc != 0) {
1595                 CERROR("Can't pack response, rc %d\n", rc);
1596                 RETURN(err_serious(rc));
1597         }
1598
1599         if (req_capsule_has_field(pill, &RMF_MDT_BODY, RCL_SERVER)) {
1600                 repbody = req_capsule_server_get(pill, &RMF_MDT_BODY);
1601                 LASSERT(repbody);
1602                 repbody->eadatasize = 0;
1603                 repbody->aclsize = 0;
1604         }
1605
1606         OBD_FAIL_TIMEOUT(OBD_FAIL_MDS_REINT_DELAY, 10);
1607
1608         /* for replay no cookkie / lmm need, because client have this already */
1609         if (info->mti_spec.no_create)
1610                 if (req_capsule_has_field(pill, &RMF_MDT_MD, RCL_SERVER))
1611                         req_capsule_set_size(pill, &RMF_MDT_MD, RCL_SERVER, 0);
1612
1613         rc = mdt_init_ucred_reint(info);
1614         if (rc)
1615                 GOTO(out_shrink, rc);
1616
1617         rc = mdt_fix_attr_ucred(info, op);
1618         if (rc != 0)
1619                 GOTO(out_ucred, rc = err_serious(rc));
1620
1621         if (mdt_check_resent(info, mdt_reconstruct, lhc)) {
1622                 rc = lustre_msg_get_status(mdt_info_req(info)->rq_repmsg);
1623                 GOTO(out_ucred, rc);
1624         }
1625         mq->mq_exp = info->mti_exp;
1626         rc = mdt_reint_rec(info, lhc);
1627         EXIT;
1628 out_ucred:
1629         mdt_exit_ucred(info);
1630 out_shrink:
1631         mdt_client_compatibility(info);
1632         rc2 = mdt_fix_reply(info);
1633         if (rc == 0)
1634                 rc = rc2;
1635         return rc;
1636 }
1637
1638 static long mdt_reint_opcode(struct mdt_thread_info *info,
1639                              const struct req_format **fmt)
1640 {
1641         struct mdt_rec_reint *rec;
1642         long opc;
1643
1644         opc = err_serious(-EFAULT);
1645         rec = req_capsule_client_get(info->mti_pill, &RMF_REC_REINT);
1646         if (rec != NULL) {
1647                 opc = rec->rr_opcode;
1648                 DEBUG_REQ(D_INODE, mdt_info_req(info), "reint opt = %ld", opc);
1649                 if (opc < REINT_MAX && fmt[opc] != NULL)
1650                         req_capsule_extend(info->mti_pill, fmt[opc]);
1651                 else {
1652                         CERROR("Unsupported opc: %ld\n", opc);
1653                         opc = err_serious(opc);
1654                 }
1655         }
1656         return opc;
1657 }
1658
1659 static int mdt_reint(struct mdt_thread_info *info)
1660 {
1661         long opc;
1662         int  rc;
1663
1664         static const struct req_format *reint_fmts[REINT_MAX] = {
1665                 [REINT_SETATTR]  = &RQF_MDS_REINT_SETATTR,
1666                 [REINT_CREATE]   = &RQF_MDS_REINT_CREATE,
1667                 [REINT_LINK]     = &RQF_MDS_REINT_LINK,
1668                 [REINT_UNLINK]   = &RQF_MDS_REINT_UNLINK,
1669                 [REINT_RENAME]   = &RQF_MDS_REINT_RENAME,
1670                 [REINT_OPEN]     = &RQF_MDS_REINT_OPEN,
1671                 [REINT_SETXATTR] = &RQF_MDS_REINT_SETXATTR
1672         };
1673
1674         ENTRY;
1675
1676         opc = mdt_reint_opcode(info, reint_fmts);
1677         if (opc >= 0) {
1678                 /*
1679                  * No lock possible here from client to pass it to reint code
1680                  * path.
1681                  */
1682                 rc = mdt_reint_internal(info, NULL, opc);
1683         } else {
1684                 rc = opc;
1685         }
1686
1687         info->mti_fail_id = OBD_FAIL_MDS_REINT_NET_REP;
1688         RETURN(rc);
1689 }
1690
1691 /* this should sync the whole device */
1692 static int mdt_device_sync(const struct lu_env *env, struct mdt_device *mdt)
1693 {
1694         struct dt_device *dt = mdt->mdt_bottom;
1695         int rc;
1696         ENTRY;
1697
1698         rc = dt->dd_ops->dt_sync(env, dt);
1699         RETURN(rc);
1700 }
1701
1702 /* this should sync this object */
1703 static int mdt_object_sync(struct mdt_thread_info *info)
1704 {
1705         struct md_object *next;
1706         int rc;
1707         ENTRY;
1708
1709         if (!mdt_object_exists(info->mti_object)) {
1710                 CWARN("Non existing object  "DFID"!\n",
1711                       PFID(mdt_object_fid(info->mti_object)));
1712                 RETURN(-ESTALE);
1713         }
1714         next = mdt_object_child(info->mti_object);
1715         rc = mo_object_sync(info->mti_env, next);
1716
1717         RETURN(rc);
1718 }
1719
1720 static int mdt_sync(struct mdt_thread_info *info)
1721 {
1722         struct ptlrpc_request *req = mdt_info_req(info);
1723         struct req_capsule *pill = info->mti_pill;
1724         struct mdt_body *body;
1725         int rc;
1726         ENTRY;
1727
1728         /* The fid may be zero, so we req_capsule_set manually */
1729         req_capsule_set(pill, &RQF_MDS_SYNC);
1730
1731         body = req_capsule_client_get(pill, &RMF_MDT_BODY);
1732         if (body == NULL)
1733                 RETURN(err_serious(-EINVAL));
1734
1735         if (OBD_FAIL_CHECK(OBD_FAIL_MDS_SYNC_PACK))
1736                 RETURN(err_serious(-ENOMEM));
1737
1738         if (fid_seq(&body->fid1) == 0) {
1739                 /* sync the whole device */
1740                 rc = req_capsule_server_pack(pill);
1741                 if (rc == 0)
1742                         rc = mdt_device_sync(info->mti_env, info->mti_mdt);
1743                 else
1744                         rc = err_serious(rc);
1745         } else {
1746                 /* sync an object */
1747                 rc = mdt_unpack_req_pack_rep(info, HABEO_CORPUS|HABEO_REFERO);
1748                 if (rc == 0) {
1749                         rc = mdt_object_sync(info);
1750                         if (rc == 0) {
1751                                 struct md_object *next;
1752                                 const struct lu_fid *fid;
1753                                 struct lu_attr *la = &info->mti_attr.ma_attr;
1754
1755                                 next = mdt_object_child(info->mti_object);
1756                                 info->mti_attr.ma_need = MA_INODE;
1757                                 info->mti_attr.ma_valid = 0;
1758                                 rc = mo_attr_get(info->mti_env, next,
1759                                                  &info->mti_attr);
1760                                 if (rc == 0) {
1761                                         body = req_capsule_server_get(pill,
1762                                                                 &RMF_MDT_BODY);
1763                                         fid = mdt_object_fid(info->mti_object);
1764                                         mdt_pack_attr2body(info, body, la, fid);
1765                                 }
1766                         }
1767                 } else
1768                         rc = err_serious(rc);
1769         }
1770         if (rc == 0)
1771                 mdt_counter_incr(req, LPROC_MDT_SYNC);
1772
1773         RETURN(rc);
1774 }
1775
1776 #ifdef HAVE_QUOTA_SUPPORT
1777 static int mdt_quotacheck_handle(struct mdt_thread_info *info)
1778 {
1779         struct obd_quotactl *oqctl;
1780         struct req_capsule *pill = info->mti_pill;
1781         struct obd_export *exp = info->mti_exp;
1782         struct md_quota *mq = md_quota(info->mti_env);
1783         struct md_device *next = info->mti_mdt->mdt_child;
1784         int rc;
1785         ENTRY;
1786
1787         oqctl = req_capsule_client_get(pill, &RMF_OBD_QUOTACTL);
1788         if (oqctl == NULL)
1789                 RETURN(-EPROTO);
1790
1791         /* remote client has no permission for quotacheck */
1792         if (unlikely(exp_connect_rmtclient(exp)))
1793                 RETURN(-EPERM);
1794
1795         rc = req_capsule_server_pack(pill);
1796         if (rc)
1797                 RETURN(rc);
1798
1799         mq->mq_exp = exp;
1800         rc = next->md_ops->mdo_quota.mqo_check(info->mti_env, next,
1801                                                oqctl->qc_type);
1802         RETURN(rc);
1803 }
1804
1805 static int mdt_quotactl_handle(struct mdt_thread_info *info)
1806 {
1807         struct obd_quotactl *oqctl, *repoqc;
1808         struct req_capsule *pill = info->mti_pill;
1809         struct obd_export *exp = info->mti_exp;
1810         struct md_quota *mq = md_quota(info->mti_env);
1811         struct md_device *next = info->mti_mdt->mdt_child;
1812         const struct md_quota_operations *mqo = &next->md_ops->mdo_quota;
1813         int id, rc;
1814         ENTRY;
1815
1816         oqctl = req_capsule_client_get(pill, &RMF_OBD_QUOTACTL);
1817         if (oqctl == NULL)
1818                 RETURN(-EPROTO);
1819
1820         id = oqctl->qc_id;
1821         if (exp_connect_rmtclient(exp)) {
1822                 struct ptlrpc_request *req = mdt_info_req(info);
1823                 struct mdt_export_data *med = mdt_req2med(req);
1824                 struct lustre_idmap_table *idmap = med->med_idmap;
1825
1826                 if (unlikely(oqctl->qc_cmd != Q_GETQUOTA &&
1827                              oqctl->qc_cmd != Q_GETINFO))
1828                         RETURN(-EPERM);
1829
1830
1831                 if (oqctl->qc_type == USRQUOTA)
1832                         id = lustre_idmap_lookup_uid(NULL, idmap, 0,
1833                                                      oqctl->qc_id);
1834                 else if (oqctl->qc_type == GRPQUOTA)
1835                         id = lustre_idmap_lookup_gid(NULL, idmap, 0,
1836                                                      oqctl->qc_id);
1837                 else
1838                         RETURN(-EINVAL);
1839
1840                 if (id == CFS_IDMAP_NOTFOUND) {
1841                         CDEBUG(D_QUOTA, "no mapping for id %u\n",
1842                                oqctl->qc_id);
1843                         RETURN(-EACCES);
1844                 }
1845         }
1846
1847         rc = req_capsule_server_pack(pill);
1848         if (rc)
1849                 RETURN(rc);
1850
1851         repoqc = req_capsule_server_get(pill, &RMF_OBD_QUOTACTL);
1852         LASSERT(repoqc != NULL);
1853
1854         mq->mq_exp = exp;
1855         switch (oqctl->qc_cmd) {
1856         case Q_QUOTAON:
1857                 rc = mqo->mqo_on(info->mti_env, next, oqctl->qc_type);
1858                 break;
1859         case Q_QUOTAOFF:
1860                 rc = mqo->mqo_off(info->mti_env, next, oqctl->qc_type);
1861                 break;
1862         case Q_SETINFO:
1863                 rc = mqo->mqo_setinfo(info->mti_env, next, oqctl->qc_type, id,
1864                                       &oqctl->qc_dqinfo);
1865                 break;
1866         case Q_GETINFO:
1867                 rc = mqo->mqo_getinfo(info->mti_env, next, oqctl->qc_type, id,
1868                                       &oqctl->qc_dqinfo);
1869                 break;
1870         case Q_SETQUOTA:
1871                 rc = mqo->mqo_setquota(info->mti_env, next, oqctl->qc_type, id,
1872                                        &oqctl->qc_dqblk);
1873                 break;
1874         case Q_GETQUOTA:
1875                 rc = mqo->mqo_getquota(info->mti_env, next, oqctl->qc_type, id,
1876                                        &oqctl->qc_dqblk);
1877                 break;
1878         case Q_GETOINFO:
1879                 rc = mqo->mqo_getoinfo(info->mti_env, next, oqctl->qc_type, id,
1880                                        &oqctl->qc_dqinfo);
1881                 break;
1882         case Q_GETOQUOTA:
1883                 rc = mqo->mqo_getoquota(info->mti_env, next, oqctl->qc_type, id,
1884                                         &oqctl->qc_dqblk);
1885                 break;
1886         case LUSTRE_Q_INVALIDATE:
1887                 rc = mqo->mqo_invalidate(info->mti_env, next, oqctl->qc_type);
1888                 break;
1889         case LUSTRE_Q_FINVALIDATE:
1890                 rc = mqo->mqo_finvalidate(info->mti_env, next, oqctl->qc_type);
1891                 break;
1892         default:
1893                 CERROR("unsupported mdt_quotactl command: %d\n",
1894                        oqctl->qc_cmd);
1895                 RETURN(-EFAULT);
1896         }
1897
1898         *repoqc = *oqctl;
1899         RETURN(rc);
1900 }
1901 #endif
1902
1903
1904 /*
1905  * OBD PING and other handlers.
1906  */
1907 static int mdt_obd_ping(struct mdt_thread_info *info)
1908 {
1909         int rc;
1910         ENTRY;
1911
1912         req_capsule_set(info->mti_pill, &RQF_OBD_PING);
1913
1914         rc = target_handle_ping(mdt_info_req(info));
1915         if (rc < 0)
1916                 rc = err_serious(rc);
1917         RETURN(rc);
1918 }
1919
1920 static int mdt_obd_log_cancel(struct mdt_thread_info *info)
1921 {
1922         return err_serious(-EOPNOTSUPP);
1923 }
1924
1925 static int mdt_obd_qc_callback(struct mdt_thread_info *info)
1926 {
1927         return err_serious(-EOPNOTSUPP);
1928 }
1929
1930
1931 /*
1932  * LLOG handlers.
1933  */
1934
1935 /** clone llog ctxt from child (mdd)
1936  * This allows remote llog (replicator) access.
1937  * We can either pass all llog RPCs (eg mdt_llog_create) on to child where the
1938  * context was originally set up, or we can handle them directly.
1939  * I choose the latter, but that means I need any llog
1940  * contexts set up by child to be accessable by the mdt.  So we clone the
1941  * context into our context list here.
1942  */
1943 static int mdt_llog_ctxt_clone(const struct lu_env *env, struct mdt_device *mdt,
1944                                int idx)
1945 {
1946         struct md_device  *next = mdt->mdt_child;
1947         struct llog_ctxt *ctxt;
1948         int rc;
1949
1950         if (!llog_ctxt_null(mdt2obd_dev(mdt), idx))
1951                 return 0;
1952
1953         rc = next->md_ops->mdo_llog_ctxt_get(env, next, idx, (void **)&ctxt);
1954         if (rc || ctxt == NULL) {
1955                 CERROR("Can't get mdd ctxt %d\n", rc);
1956                 return rc;
1957         }
1958
1959         rc = llog_group_set_ctxt(&mdt2obd_dev(mdt)->obd_olg, ctxt, idx);
1960         if (rc)
1961                 CERROR("Can't set mdt ctxt %d\n", rc);
1962
1963         return rc;
1964 }
1965
1966 static int mdt_llog_ctxt_unclone(const struct lu_env *env,
1967                                  struct mdt_device *mdt, int idx)
1968 {
1969         struct llog_ctxt *ctxt;
1970
1971         ctxt = llog_get_context(mdt2obd_dev(mdt), idx);
1972         if (ctxt == NULL)
1973                 return 0;
1974         /* Put once for the get we just did, and once for the clone */
1975         llog_ctxt_put(ctxt);
1976         llog_ctxt_put(ctxt);
1977         return 0;
1978 }
1979
1980 static int mdt_llog_create(struct mdt_thread_info *info)
1981 {
1982         int rc;
1983
1984         req_capsule_set(info->mti_pill, &RQF_LLOG_ORIGIN_HANDLE_CREATE);
1985         rc = llog_origin_handle_create(mdt_info_req(info));
1986         return (rc < 0 ? err_serious(rc) : rc);
1987 }
1988
1989 static int mdt_llog_destroy(struct mdt_thread_info *info)
1990 {
1991         int rc;
1992
1993         req_capsule_set(info->mti_pill, &RQF_LLOG_ORIGIN_HANDLE_DESTROY);
1994         rc = llog_origin_handle_destroy(mdt_info_req(info));
1995         return (rc < 0 ? err_serious(rc) : rc);
1996 }
1997
1998 static int mdt_llog_read_header(struct mdt_thread_info *info)
1999 {
2000         int rc;
2001
2002         req_capsule_set(info->mti_pill, &RQF_LLOG_ORIGIN_HANDLE_READ_HEADER);
2003         rc = llog_origin_handle_read_header(mdt_info_req(info));
2004         return (rc < 0 ? err_serious(rc) : rc);
2005 }
2006
2007 static int mdt_llog_next_block(struct mdt_thread_info *info)
2008 {
2009         int rc;
2010
2011         req_capsule_set(info->mti_pill, &RQF_LLOG_ORIGIN_HANDLE_NEXT_BLOCK);
2012         rc = llog_origin_handle_next_block(mdt_info_req(info));
2013         return (rc < 0 ? err_serious(rc) : rc);
2014 }
2015
2016 static int mdt_llog_prev_block(struct mdt_thread_info *info)
2017 {
2018         int rc;
2019
2020         req_capsule_set(info->mti_pill, &RQF_LLOG_ORIGIN_HANDLE_PREV_BLOCK);
2021         rc = llog_origin_handle_prev_block(mdt_info_req(info));
2022         return (rc < 0 ? err_serious(rc) : rc);
2023 }
2024
2025
2026 /*
2027  * DLM handlers.
2028  */
2029 static struct ldlm_callback_suite cbs = {
2030         .lcs_completion = ldlm_server_completion_ast,
2031         .lcs_blocking   = ldlm_server_blocking_ast,
2032         .lcs_glimpse    = NULL
2033 };
2034
2035 static int mdt_enqueue(struct mdt_thread_info *info)
2036 {
2037         struct ptlrpc_request *req;
2038         int rc;
2039
2040         /*
2041          * info->mti_dlm_req already contains swapped and (if necessary)
2042          * converted dlm request.
2043          */
2044         LASSERT(info->mti_dlm_req != NULL);
2045
2046         req = mdt_info_req(info);
2047         rc = ldlm_handle_enqueue0(info->mti_mdt->mdt_namespace,
2048                                   req, info->mti_dlm_req, &cbs);
2049         info->mti_fail_id = OBD_FAIL_LDLM_REPLY;
2050         return rc ? err_serious(rc) : req->rq_status;
2051 }
2052
2053 static int mdt_convert(struct mdt_thread_info *info)
2054 {
2055         int rc;
2056         struct ptlrpc_request *req;
2057
2058         LASSERT(info->mti_dlm_req);
2059         req = mdt_info_req(info);
2060         rc = ldlm_handle_convert0(req, info->mti_dlm_req);
2061         return rc ? err_serious(rc) : req->rq_status;
2062 }
2063
2064 static int mdt_bl_callback(struct mdt_thread_info *info)
2065 {
2066         CERROR("bl callbacks should not happen on MDS\n");
2067         LBUG();
2068         return err_serious(-EOPNOTSUPP);
2069 }
2070
2071 static int mdt_cp_callback(struct mdt_thread_info *info)
2072 {
2073         CERROR("cp callbacks should not happen on MDS\n");
2074         LBUG();
2075         return err_serious(-EOPNOTSUPP);
2076 }
2077
2078 /*
2079  * sec context handlers
2080  */
2081 static int mdt_sec_ctx_handle(struct mdt_thread_info *info)
2082 {
2083         int rc;
2084
2085         rc = mdt_handle_idmap(info);
2086
2087         if (unlikely(rc)) {
2088                 struct ptlrpc_request *req = mdt_info_req(info);
2089                 __u32                  opc;
2090
2091                 opc = lustre_msg_get_opc(req->rq_reqmsg);
2092                 if (opc == SEC_CTX_INIT || opc == SEC_CTX_INIT_CONT)
2093                         sptlrpc_svc_ctx_invalidate(req);
2094         }
2095
2096         CFS_FAIL_TIMEOUT(OBD_FAIL_SEC_CTX_HDL_PAUSE, cfs_fail_val);
2097
2098         return rc;
2099 }
2100
2101 static struct mdt_object *mdt_obj(struct lu_object *o)
2102 {
2103         LASSERT(lu_device_is_mdt(o->lo_dev));
2104         return container_of0(o, struct mdt_object, mot_obj.mo_lu);
2105 }
2106
2107 struct mdt_object *mdt_object_new(const struct lu_env *env,
2108                                   struct mdt_device *d,
2109                                   const struct lu_fid *f)
2110 {
2111         struct lu_object_conf conf = { .loc_flags = LOC_F_NEW };
2112         struct lu_object *o;
2113         struct mdt_object *m;
2114         ENTRY;
2115
2116         CDEBUG(D_INFO, "Allocate object for "DFID"\n", PFID(f));
2117         o = lu_object_find(env, &d->mdt_md_dev.md_lu_dev, f, &conf);
2118         if (unlikely(IS_ERR(o)))
2119                 m = (struct mdt_object *)o;
2120         else
2121                 m = mdt_obj(o);
2122         RETURN(m);
2123 }
2124
2125 struct mdt_object *mdt_object_find(const struct lu_env *env,
2126                                    struct mdt_device *d,
2127                                    const struct lu_fid *f)
2128 {
2129         struct lu_object *o;
2130         struct mdt_object *m;
2131         ENTRY;
2132
2133         CDEBUG(D_INFO, "Find object for "DFID"\n", PFID(f));
2134         o = lu_object_find(env, &d->mdt_md_dev.md_lu_dev, f, NULL);
2135         if (unlikely(IS_ERR(o)))
2136                 m = (struct mdt_object *)o;
2137         else
2138                 m = mdt_obj(o);
2139         RETURN(m);
2140 }
2141
2142 /**
2143  * Asyncronous commit for mdt device.
2144  *
2145  * Pass asynchonous commit call down the MDS stack.
2146  *
2147  * \param env environment
2148  * \param mdt the mdt device
2149  */
2150 static void mdt_device_commit_async(const struct lu_env *env,
2151                                     struct mdt_device *mdt)
2152 {
2153         struct dt_device *dt = mdt->mdt_bottom;
2154         int rc;
2155
2156         rc = dt->dd_ops->dt_commit_async(env, dt);
2157         if (unlikely(rc != 0))
2158                 CWARN("async commit start failed with rc = %d", rc);
2159 }
2160
2161 /**
2162  * Mark the lock as "synchonous".
2163  *
2164  * Mark the lock to deffer transaction commit to the unlock time.
2165  *
2166  * \param lock the lock to mark as "synchonous"
2167  *
2168  * \see mdt_is_lock_sync
2169  * \see mdt_save_lock
2170  */
2171 static inline void mdt_set_lock_sync(struct ldlm_lock *lock)
2172 {
2173         lock->l_ast_data = (void*)1;
2174 }
2175
2176 /**
2177  * Check whehter the lock "synchonous" or not.
2178  *
2179  * \param lock the lock to check
2180  * \retval 1 the lock is "synchonous"
2181  * \retval 0 the lock isn't "synchronous"
2182  *
2183  * \see mdt_set_lock_sync
2184  * \see mdt_save_lock
2185  */
2186 static inline int mdt_is_lock_sync(struct ldlm_lock *lock)
2187 {
2188         return lock->l_ast_data != NULL;
2189 }
2190
2191 /**
2192  * Blocking AST for mdt locks.
2193  *
2194  * Starts transaction commit if in case of COS lock conflict or
2195  * deffers such a commit to the mdt_save_lock.
2196  *
2197  * \param lock the lock which blocks a request or cancelling lock
2198  * \param desc unused
2199  * \param data unused
2200  * \param flag indicates whether this cancelling or blocking callback
2201  * \retval 0
2202  * \see ldlm_blocking_ast_nocheck
2203  */
2204 int mdt_blocking_ast(struct ldlm_lock *lock, struct ldlm_lock_desc *desc,
2205                      void *data, int flag)
2206 {
2207         struct obd_device *obd = ldlm_lock_to_ns(lock)->ns_obd;
2208         struct mdt_device *mdt = mdt_dev(obd->obd_lu_dev);
2209         int rc;
2210         ENTRY;
2211
2212         if (flag == LDLM_CB_CANCELING)
2213                 RETURN(0);
2214         lock_res_and_lock(lock);
2215         if (lock->l_blocking_ast != mdt_blocking_ast) {
2216                 unlock_res_and_lock(lock);
2217                 RETURN(0);
2218         }
2219         if (mdt_cos_is_enabled(mdt) &&
2220             lock->l_req_mode & (LCK_PW | LCK_EX) &&
2221             lock->l_blocking_lock != NULL &&
2222             lock->l_client_cookie != lock->l_blocking_lock->l_client_cookie) {
2223                 mdt_set_lock_sync(lock);
2224         }
2225         rc = ldlm_blocking_ast_nocheck(lock);
2226
2227         /* There is no lock conflict if l_blocking_lock == NULL,
2228          * it indicates a blocking ast sent from ldlm_lock_decref_internal
2229          * when the last reference to a local lock was released */
2230         if (lock->l_req_mode == LCK_COS && lock->l_blocking_lock != NULL) {
2231                 struct lu_env env;
2232
2233                 rc = lu_env_init(&env, LCT_LOCAL);
2234                 if (unlikely(rc != 0))
2235                         CWARN("lu_env initialization failed with rc = %d,"
2236                               "cannot start asynchronous commit\n", rc);
2237                 else
2238                         mdt_device_commit_async(&env, mdt);
2239                 lu_env_fini(&env);
2240         }
2241         RETURN(rc);
2242 }
2243
2244 int mdt_object_lock(struct mdt_thread_info *info, struct mdt_object *o,
2245                     struct mdt_lock_handle *lh, __u64 ibits, int locality)
2246 {
2247         struct ldlm_namespace *ns = info->mti_mdt->mdt_namespace;
2248         ldlm_policy_data_t *policy = &info->mti_policy;
2249         struct ldlm_res_id *res_id = &info->mti_res_id;
2250         int rc;
2251         ENTRY;
2252
2253         LASSERT(!lustre_handle_is_used(&lh->mlh_reg_lh));
2254         LASSERT(!lustre_handle_is_used(&lh->mlh_pdo_lh));
2255         LASSERT(lh->mlh_reg_mode != LCK_MINMODE);
2256         LASSERT(lh->mlh_type != MDT_NUL_LOCK);
2257
2258         if (mdt_object_exists(o) < 0) {
2259                 if (locality == MDT_CROSS_LOCK) {
2260                         /* cross-ref object fix */
2261                         ibits &= ~MDS_INODELOCK_UPDATE;
2262                         ibits |= MDS_INODELOCK_LOOKUP;
2263                 } else {
2264                         LASSERT(!(ibits & MDS_INODELOCK_UPDATE));
2265                         LASSERT(ibits & MDS_INODELOCK_LOOKUP);
2266                 }
2267                 /* No PDO lock on remote object */
2268                 LASSERT(lh->mlh_type != MDT_PDO_LOCK);
2269         }
2270
2271         if (lh->mlh_type == MDT_PDO_LOCK) {
2272                 /* check for exists after object is locked */
2273                 if (mdt_object_exists(o) == 0) {
2274                         /* Non-existent object shouldn't have PDO lock */
2275                         RETURN(-ESTALE);
2276                 } else {
2277                         /* Non-dir object shouldn't have PDO lock */
2278                         if (!S_ISDIR(lu_object_attr(&o->mot_obj.mo_lu)))
2279                                 RETURN(-ENOTDIR);
2280                 }
2281         }
2282
2283         memset(policy, 0, sizeof(*policy));
2284         fid_build_reg_res_name(mdt_object_fid(o), res_id);
2285
2286         /*
2287          * Take PDO lock on whole directory and build correct @res_id for lock
2288          * on part of directory.
2289          */
2290         if (lh->mlh_pdo_hash != 0) {
2291                 LASSERT(lh->mlh_type == MDT_PDO_LOCK);
2292                 mdt_lock_pdo_mode(info, o, lh);
2293                 if (lh->mlh_pdo_mode != LCK_NL) {
2294                         /*
2295                          * Do not use LDLM_FL_LOCAL_ONLY for parallel lock, it
2296                          * is never going to be sent to client and we do not
2297                          * want it slowed down due to possible cancels.
2298                          */
2299                         policy->l_inodebits.bits = MDS_INODELOCK_UPDATE;
2300                         rc = mdt_fid_lock(ns, &lh->mlh_pdo_lh, lh->mlh_pdo_mode,
2301                                           policy, res_id, LDLM_FL_ATOMIC_CB,
2302                                           &info->mti_exp->exp_handle.h_cookie);
2303                         if (unlikely(rc))
2304                                 RETURN(rc);
2305                 }
2306
2307                 /*
2308                  * Finish res_id initializing by name hash marking part of
2309                  * directory which is taking modification.
2310                  */
2311                 res_id->name[LUSTRE_RES_ID_HSH_OFF] = lh->mlh_pdo_hash;
2312         }
2313
2314         policy->l_inodebits.bits = ibits;
2315
2316         /*
2317          * Use LDLM_FL_LOCAL_ONLY for this lock. We do not know yet if it is
2318          * going to be sent to client. If it is - mdt_intent_policy() path will
2319          * fix it up and turn FL_LOCAL flag off.
2320          */
2321         rc = mdt_fid_lock(ns, &lh->mlh_reg_lh, lh->mlh_reg_mode, policy,
2322                           res_id, LDLM_FL_LOCAL_ONLY | LDLM_FL_ATOMIC_CB,
2323                           &info->mti_exp->exp_handle.h_cookie);
2324         if (rc)
2325                 mdt_object_unlock(info, o, lh, 1);
2326         else if (unlikely(OBD_FAIL_PRECHECK(OBD_FAIL_MDS_PDO_LOCK)) &&
2327                  lh->mlh_pdo_hash != 0 &&
2328                  (lh->mlh_reg_mode == LCK_PW || lh->mlh_reg_mode == LCK_EX)) {
2329                 OBD_FAIL_TIMEOUT(OBD_FAIL_MDS_PDO_LOCK, 15);
2330         }
2331
2332         RETURN(rc);
2333 }
2334
2335 /**
2336  * Save a lock within request object.
2337  *
2338  * Keep the lock referenced until whether client ACK or transaction
2339  * commit happens or release the lock immediately depending on input
2340  * parameters. If COS is ON, a write lock is converted to COS lock
2341  * before saving.
2342  *
2343  * \param info thead info object
2344  * \param h lock handle
2345  * \param mode lock mode
2346  * \param decref force immediate lock releasing
2347  */
2348 static
2349 void mdt_save_lock(struct mdt_thread_info *info, struct lustre_handle *h,
2350                    ldlm_mode_t mode, int decref)
2351 {
2352         ENTRY;
2353
2354         if (lustre_handle_is_used(h)) {
2355                 if (decref || !info->mti_has_trans ||
2356                     !(mode & (LCK_PW | LCK_EX))){
2357                         mdt_fid_unlock(h, mode);
2358                 } else {
2359                         struct mdt_device *mdt = info->mti_mdt;
2360                         struct ldlm_lock *lock = ldlm_handle2lock(h);
2361                         struct ptlrpc_request *req = mdt_info_req(info);
2362                         int no_ack = 0;
2363
2364                         LASSERTF(lock != NULL, "no lock for cookie "LPX64"\n",
2365                                  h->cookie);
2366                         CDEBUG(D_HA, "request = %p reply state = %p"
2367                                " transno = "LPD64"\n",
2368                                req, req->rq_reply_state, req->rq_transno);
2369                         if (mdt_cos_is_enabled(mdt)) {
2370                                 no_ack = 1;
2371                                 ldlm_lock_downgrade(lock, LCK_COS);
2372                                 mode = LCK_COS;
2373                         }
2374                         ptlrpc_save_lock(req, h, mode, no_ack);
2375                         if (mdt_is_lock_sync(lock)) {
2376                                 CDEBUG(D_HA, "found sync-lock,"
2377                                        " async commit started\n");
2378                                 mdt_device_commit_async(info->mti_env,
2379                                                         mdt);
2380                         }
2381                         LDLM_LOCK_PUT(lock);
2382                 }
2383                 h->cookie = 0ull;
2384         }
2385
2386         EXIT;
2387 }
2388
2389 /**
2390  * Unlock mdt object.
2391  *
2392  * Immeditely release the regular lock and the PDO lock or save the
2393  * lock in reqeuest and keep them referenced until client ACK or
2394  * transaction commit.
2395  *
2396  * \param info thread info object
2397  * \param o mdt object
2398  * \param lh mdt lock handle referencing regular and PDO locks
2399  * \param decref force immediate lock releasing
2400  */
2401 void mdt_object_unlock(struct mdt_thread_info *info, struct mdt_object *o,
2402                        struct mdt_lock_handle *lh, int decref)
2403 {
2404         ENTRY;
2405
2406         mdt_save_lock(info, &lh->mlh_pdo_lh, lh->mlh_pdo_mode, decref);
2407         mdt_save_lock(info, &lh->mlh_reg_lh, lh->mlh_reg_mode, decref);
2408
2409         EXIT;
2410 }
2411
2412 struct mdt_object *mdt_object_find_lock(struct mdt_thread_info *info,
2413                                         const struct lu_fid *f,
2414                                         struct mdt_lock_handle *lh,
2415                                         __u64 ibits)
2416 {
2417         struct mdt_object *o;
2418
2419         o = mdt_object_find(info->mti_env, info->mti_mdt, f);
2420         if (!IS_ERR(o)) {
2421                 int rc;
2422
2423                 rc = mdt_object_lock(info, o, lh, ibits,
2424                                      MDT_LOCAL_LOCK);
2425                 if (rc != 0) {
2426                         mdt_object_put(info->mti_env, o);
2427                         o = ERR_PTR(rc);
2428                 }
2429         }
2430         return o;
2431 }
2432
2433 void mdt_object_unlock_put(struct mdt_thread_info * info,
2434                            struct mdt_object * o,
2435                            struct mdt_lock_handle *lh,
2436                            int decref)
2437 {
2438         mdt_object_unlock(info, o, lh, decref);
2439         mdt_object_put(info->mti_env, o);
2440 }
2441
2442 static struct mdt_handler *mdt_handler_find(__u32 opc,
2443                                             struct mdt_opc_slice *supported)
2444 {
2445         struct mdt_opc_slice *s;
2446         struct mdt_handler   *h;
2447
2448         h = NULL;
2449         for (s = supported; s->mos_hs != NULL; s++) {
2450                 if (s->mos_opc_start <= opc && opc < s->mos_opc_end) {
2451                         h = s->mos_hs + (opc - s->mos_opc_start);
2452                         if (likely(h->mh_opc != 0))
2453                                 LASSERTF(h->mh_opc == opc,
2454                                          "opcode mismatch %d != %d\n",
2455                                          h->mh_opc, opc);
2456                         else
2457                                 h = NULL; /* unsupported opc */
2458                         break;
2459                 }
2460         }
2461         return h;
2462 }
2463
2464 static int mdt_lock_resname_compat(struct mdt_device *m,
2465                                    struct ldlm_request *req)
2466 {
2467         /* XXX something... later. */
2468         return 0;
2469 }
2470
2471 static int mdt_lock_reply_compat(struct mdt_device *m, struct ldlm_reply *rep)
2472 {
2473         /* XXX something... later. */
2474         return 0;
2475 }
2476
2477 /*
2478  * Generic code handling requests that have struct mdt_body passed in:
2479  *
2480  *  - extract mdt_body from request and save it in @info, if present;
2481  *
2482  *  - create lu_object, corresponding to the fid in mdt_body, and save it in
2483  *  @info;
2484  *
2485  *  - if HABEO_CORPUS flag is set for this request type check whether object
2486  *  actually exists on storage (lu_object_exists()).
2487  *
2488  */
2489 static int mdt_body_unpack(struct mdt_thread_info *info, __u32 flags)
2490 {
2491         const struct mdt_body    *body;
2492         struct mdt_object        *obj;
2493         const struct lu_env      *env;
2494         struct req_capsule       *pill;
2495         int                       rc;
2496         ENTRY;
2497
2498         env = info->mti_env;
2499         pill = info->mti_pill;
2500
2501         body = info->mti_body = req_capsule_client_get(pill, &RMF_MDT_BODY);
2502         if (body == NULL)
2503                 RETURN(-EFAULT);
2504
2505         if (!(body->valid & OBD_MD_FLID))
2506                 RETURN(0);
2507
2508         if (!fid_is_sane(&body->fid1)) {
2509                 CERROR("Invalid fid: "DFID"\n", PFID(&body->fid1));
2510                 RETURN(-EINVAL);
2511         }
2512
2513         /*
2514          * Do not get size or any capa fields before we check that request
2515          * contains capa actually. There are some requests which do not, for
2516          * instance MDS_IS_SUBDIR.
2517          */
2518         if (req_capsule_has_field(pill, &RMF_CAPA1, RCL_CLIENT) &&
2519             req_capsule_get_size(pill, &RMF_CAPA1, RCL_CLIENT))
2520                 mdt_set_capainfo(info, 0, &body->fid1,
2521                                  req_capsule_client_get(pill, &RMF_CAPA1));
2522
2523         obj = mdt_object_find(env, info->mti_mdt, &body->fid1);
2524         if (!IS_ERR(obj)) {
2525                 if ((flags & HABEO_CORPUS) &&
2526                     !mdt_object_exists(obj)) {
2527                         mdt_object_put(env, obj);
2528                         /* for capability renew ENOENT will be handled in
2529                          * mdt_renew_capa */
2530                         if (body->valid & OBD_MD_FLOSSCAPA)
2531                                 rc = 0;
2532                         else
2533                                 rc = -ENOENT;
2534                 } else {
2535                         info->mti_object = obj;
2536                         rc = 0;
2537                 }
2538         } else
2539                 rc = PTR_ERR(obj);
2540
2541         RETURN(rc);
2542 }
2543
2544 static int mdt_unpack_req_pack_rep(struct mdt_thread_info *info, __u32 flags)
2545 {
2546         struct req_capsule *pill = info->mti_pill;
2547         int rc;
2548         ENTRY;
2549
2550         if (req_capsule_has_field(pill, &RMF_MDT_BODY, RCL_CLIENT))
2551                 rc = mdt_body_unpack(info, flags);
2552         else
2553                 rc = 0;
2554
2555         if (rc == 0 && (flags & HABEO_REFERO)) {
2556                 /* Pack reply. */
2557                 if (req_capsule_has_field(pill, &RMF_MDT_MD, RCL_SERVER))
2558                         req_capsule_set_size(pill, &RMF_MDT_MD, RCL_SERVER,
2559                                              info->mti_body->eadatasize);
2560                 if (req_capsule_has_field(pill, &RMF_LOGCOOKIES, RCL_SERVER))
2561                         req_capsule_set_size(pill, &RMF_LOGCOOKIES, RCL_SERVER,
2562                                              info->mti_mdt->mdt_max_cookiesize);
2563
2564                 rc = req_capsule_server_pack(pill);
2565         }
2566         RETURN(rc);
2567 }
2568
2569 static int mdt_init_capa_ctxt(const struct lu_env *env, struct mdt_device *m)
2570 {
2571         struct md_device *next = m->mdt_child;
2572
2573         return next->md_ops->mdo_init_capa_ctxt(env, next,
2574                                                 m->mdt_opts.mo_mds_capa,
2575                                                 m->mdt_capa_timeout,
2576                                                 m->mdt_capa_alg,
2577                                                 m->mdt_capa_keys);
2578 }
2579
2580 /*
2581  * Invoke handler for this request opc. Also do necessary preprocessing
2582  * (according to handler ->mh_flags), and post-processing (setting of
2583  * ->last_{xid,committed}).
2584  */
2585 static int mdt_req_handle(struct mdt_thread_info *info,
2586                           struct mdt_handler *h, struct ptlrpc_request *req)
2587 {
2588         int   rc, serious = 0;
2589         __u32 flags;
2590
2591         ENTRY;
2592
2593         LASSERT(h->mh_act != NULL);
2594         LASSERT(h->mh_opc == lustre_msg_get_opc(req->rq_reqmsg));
2595         LASSERT(current->journal_info == NULL);
2596
2597         /*
2598          * Checking for various OBD_FAIL_$PREF_$OPC_NET codes. _Do_ not try
2599          * to put same checks into handlers like mdt_close(), mdt_reint(),
2600          * etc., without talking to mdt authors first. Checking same thing
2601          * there again is useless and returning 0 error without packing reply
2602          * is buggy! Handlers either pack reply or return error.
2603          *
2604          * We return 0 here and do not send any reply in order to emulate
2605          * network failure. Do not send any reply in case any of NET related
2606          * fail_id has occured.
2607          */
2608         if (OBD_FAIL_CHECK_ORSET(h->mh_fail_id, OBD_FAIL_ONCE))
2609                 RETURN(0);
2610
2611         rc = 0;
2612         flags = h->mh_flags;
2613         LASSERT(ergo(flags & (HABEO_CORPUS|HABEO_REFERO), h->mh_fmt != NULL));
2614
2615         if (h->mh_fmt != NULL) {
2616                 req_capsule_set(info->mti_pill, h->mh_fmt);
2617                 rc = mdt_unpack_req_pack_rep(info, flags);
2618         }
2619
2620         if (rc == 0 && flags & MUTABOR &&
2621             req->rq_export->exp_connect_flags & OBD_CONNECT_RDONLY)
2622                 /* should it be rq_status? */
2623                 rc = -EROFS;
2624
2625         if (rc == 0 && flags & HABEO_CLAVIS) {
2626                 struct ldlm_request *dlm_req;
2627
2628                 LASSERT(h->mh_fmt != NULL);
2629
2630                 dlm_req = req_capsule_client_get(info->mti_pill, &RMF_DLM_REQ);
2631                 if (dlm_req != NULL) {
2632                         if (unlikely(dlm_req->lock_desc.l_resource.lr_type ==
2633                                         LDLM_IBITS &&
2634                                      dlm_req->lock_desc.l_policy_data.\
2635                                         l_inodebits.bits == 0)) {
2636                                 /*
2637                                  * Lock without inodebits makes no sense and
2638                                  * will oops later in ldlm. If client miss to
2639                                  * set such bits, do not trigger ASSERTION.
2640                                  *
2641                                  * For liblustre flock case, it maybe zero.
2642                                  */
2643                                 rc = -EPROTO;
2644                         } else {
2645                                 if (info->mti_mdt->mdt_opts.mo_compat_resname)
2646                                         rc = mdt_lock_resname_compat(
2647                                                                 info->mti_mdt,
2648                                                                 dlm_req);
2649                                 info->mti_dlm_req = dlm_req;
2650                         }
2651                 } else {
2652                         rc = -EFAULT;
2653                 }
2654         }
2655
2656         /* capability setting changed via /proc, needs reinitialize ctxt */
2657         if (info->mti_mdt && info->mti_mdt->mdt_capa_conf) {
2658                 mdt_init_capa_ctxt(info->mti_env, info->mti_mdt);
2659                 info->mti_mdt->mdt_capa_conf = 0;
2660         }
2661
2662         if (likely(rc == 0)) {
2663                 /*
2664                  * Process request, there can be two types of rc:
2665                  * 1) errors with msg unpack/pack, other failures outside the
2666                  * operation itself. This is counted as serious errors;
2667                  * 2) errors during fs operation, should be placed in rq_status
2668                  * only
2669                  */
2670                 rc = h->mh_act(info);
2671                 if (rc == 0 &&
2672                     !req->rq_no_reply && req->rq_reply_state == NULL) {
2673                         DEBUG_REQ(D_ERROR, req, "MDT \"handler\" %s did not "
2674                                   "pack reply and returned 0 error\n",
2675                                   h->mh_name);
2676                         LBUG();
2677                 }
2678                 serious = is_serious(rc);
2679                 rc = clear_serious(rc);
2680         } else
2681                 serious = 1;
2682
2683         req->rq_status = rc;
2684
2685         /*
2686          * ELDLM_* codes which > 0 should be in rq_status only as well as
2687          * all non-serious errors.
2688          */
2689         if (rc > 0 || !serious)
2690                 rc = 0;
2691
2692         LASSERT(current->journal_info == NULL);
2693
2694         if (rc == 0 && (flags & HABEO_CLAVIS) &&
2695             info->mti_mdt->mdt_opts.mo_compat_resname) {
2696                 struct ldlm_reply *dlmrep;
2697
2698                 dlmrep = req_capsule_server_get(info->mti_pill, &RMF_DLM_REP);
2699                 if (dlmrep != NULL)
2700                         rc = mdt_lock_reply_compat(info->mti_mdt, dlmrep);
2701         }
2702
2703         /* If we're DISCONNECTing, the mdt_export_data is already freed */
2704         if (likely(rc == 0 && req->rq_export && h->mh_opc != MDS_DISCONNECT))
2705                 target_committed_to_req(req);
2706
2707         if (unlikely(req_is_replay(req) &&
2708                      lustre_msg_get_transno(req->rq_reqmsg) == 0)) {
2709                 DEBUG_REQ(D_ERROR, req, "transno is 0 during REPLAY");
2710                 LBUG();
2711         }
2712
2713         target_send_reply(req, rc, info->mti_fail_id);
2714         RETURN(0);
2715 }
2716
2717 void mdt_lock_handle_init(struct mdt_lock_handle *lh)
2718 {
2719         lh->mlh_type = MDT_NUL_LOCK;
2720         lh->mlh_reg_lh.cookie = 0ull;
2721         lh->mlh_reg_mode = LCK_MINMODE;
2722         lh->mlh_pdo_lh.cookie = 0ull;
2723         lh->mlh_pdo_mode = LCK_MINMODE;
2724 }
2725
2726 void mdt_lock_handle_fini(struct mdt_lock_handle *lh)
2727 {
2728         LASSERT(!lustre_handle_is_used(&lh->mlh_reg_lh));
2729         LASSERT(!lustre_handle_is_used(&lh->mlh_pdo_lh));
2730 }
2731
2732 /*
2733  * Initialize fields of struct mdt_thread_info. Other fields are left in
2734  * uninitialized state, because it's too expensive to zero out whole
2735  * mdt_thread_info (> 1K) on each request arrival.
2736  */
2737 static void mdt_thread_info_init(struct ptlrpc_request *req,
2738                                  struct mdt_thread_info *info)
2739 {
2740         int i;
2741         struct md_capainfo *ci;
2742
2743         req_capsule_init(&req->rq_pill, req, RCL_SERVER);
2744         info->mti_pill = &req->rq_pill;
2745
2746         /* lock handle */
2747         for (i = 0; i < ARRAY_SIZE(info->mti_lh); i++)
2748                 mdt_lock_handle_init(&info->mti_lh[i]);
2749
2750         /* mdt device: it can be NULL while CONNECT */
2751         if (req->rq_export) {
2752                 info->mti_mdt = mdt_dev(req->rq_export->exp_obd->obd_lu_dev);
2753                 info->mti_exp = req->rq_export;
2754         } else
2755                 info->mti_mdt = NULL;
2756         info->mti_env = req->rq_svc_thread->t_env;
2757         ci = md_capainfo(info->mti_env);
2758         memset(ci, 0, sizeof *ci);
2759         if (req->rq_export) {
2760                 if (exp_connect_rmtclient(req->rq_export))
2761                         ci->mc_auth = LC_ID_CONVERT;
2762                 else if (req->rq_export->exp_connect_flags &
2763                          OBD_CONNECT_MDS_CAPA)
2764                         ci->mc_auth = LC_ID_PLAIN;
2765                 else
2766                         ci->mc_auth = LC_ID_NONE;
2767         }
2768
2769         info->mti_fail_id = OBD_FAIL_MDS_ALL_REPLY_NET;
2770         info->mti_transno = lustre_msg_get_transno(req->rq_reqmsg);
2771         info->mti_mos = NULL;
2772
2773         memset(&info->mti_attr, 0, sizeof(info->mti_attr));
2774         info->mti_body = NULL;
2775         info->mti_object = NULL;
2776         info->mti_dlm_req = NULL;
2777         info->mti_has_trans = 0;
2778         info->mti_no_need_trans = 0;
2779         info->mti_cross_ref = 0;
2780         info->mti_opdata = 0;
2781
2782         /* To not check for split by default. */
2783         info->mti_spec.sp_ck_split = 0;
2784         info->mti_spec.no_create = 0;
2785 }
2786
2787 static void mdt_thread_info_fini(struct mdt_thread_info *info)
2788 {
2789         int i;
2790
2791         req_capsule_fini(info->mti_pill);
2792         if (info->mti_object != NULL) {
2793                 /*
2794                  * freeing an object may lead to OSD level transaction, do not
2795                  * let it mess with MDT. bz19385.
2796                  */
2797                 info->mti_no_need_trans = 1;
2798                 mdt_object_put(info->mti_env, info->mti_object);
2799                 info->mti_object = NULL;
2800         }
2801         for (i = 0; i < ARRAY_SIZE(info->mti_lh); i++)
2802                 mdt_lock_handle_fini(&info->mti_lh[i]);
2803         info->mti_env = NULL;
2804 }
2805
2806 static int mdt_filter_recovery_request(struct ptlrpc_request *req,
2807                                        struct obd_device *obd, int *process)
2808 {
2809         switch (lustre_msg_get_opc(req->rq_reqmsg)) {
2810         case MDS_CONNECT: /* This will never get here, but for completeness. */
2811         case OST_CONNECT: /* This will never get here, but for completeness. */
2812         case MDS_DISCONNECT:
2813         case OST_DISCONNECT:
2814                *process = 1;
2815                RETURN(0);
2816
2817         case MDS_CLOSE:
2818         case MDS_DONE_WRITING:
2819         case MDS_SYNC: /* used in unmounting */
2820         case OBD_PING:
2821         case MDS_REINT:
2822         case SEQ_QUERY:
2823         case FLD_QUERY:
2824         case LDLM_ENQUEUE:
2825                 *process = target_queue_recovery_request(req, obd);
2826                 RETURN(0);
2827
2828         default:
2829                 DEBUG_REQ(D_ERROR, req, "not permitted during recovery");
2830                 *process = -EAGAIN;
2831                 RETURN(0);
2832         }
2833 }
2834
2835 /*
2836  * Handle recovery. Return:
2837  *        +1: continue request processing;
2838  *       -ve: abort immediately with the given error code;
2839  *         0: send reply with error code in req->rq_status;
2840  */
2841 static int mdt_recovery(struct mdt_thread_info *info)
2842 {
2843         struct ptlrpc_request *req = mdt_info_req(info);
2844         struct obd_device *obd;
2845
2846         ENTRY;
2847
2848         switch (lustre_msg_get_opc(req->rq_reqmsg)) {
2849         case MDS_CONNECT:
2850         case SEC_CTX_INIT:
2851         case SEC_CTX_INIT_CONT:
2852         case SEC_CTX_FINI:
2853                 {
2854 #if 0
2855                         int rc;
2856
2857                         rc = mdt_handle_idmap(info);
2858                         if (rc)
2859                                 RETURN(rc);
2860                         else
2861 #endif
2862                                 RETURN(+1);
2863                 }
2864         }
2865
2866         if (unlikely(!class_connected_export(req->rq_export))) {
2867                 CERROR("operation %d on unconnected MDS from %s\n",
2868                        lustre_msg_get_opc(req->rq_reqmsg),
2869                        libcfs_id2str(req->rq_peer));
2870                 /* FIXME: For CMD cleanup, when mds_B stop, the req from
2871                  * mds_A will get -ENOTCONN(especially for ping req),
2872                  * which will cause that mds_A deactive timeout, then when
2873                  * mds_A cleanup, the cleanup process will be suspended since
2874                  * deactive timeout is not zero.
2875                  */
2876                 req->rq_status = -ENOTCONN;
2877                 target_send_reply(req, -ENOTCONN, info->mti_fail_id);
2878                 RETURN(0);
2879         }
2880
2881         /* sanity check: if the xid matches, the request must be marked as a
2882          * resent or replayed */
2883         if (req_xid_is_last(req)) {
2884                 if (!(lustre_msg_get_flags(req->rq_reqmsg) &
2885                       (MSG_RESENT | MSG_REPLAY))) {
2886                         DEBUG_REQ(D_WARNING, req, "rq_xid "LPU64" matches last_xid, "
2887                                   "expected REPLAY or RESENT flag (%x)", req->rq_xid,
2888                                   lustre_msg_get_flags(req->rq_reqmsg));
2889                         LBUG();
2890                         req->rq_status = -ENOTCONN;
2891                         RETURN(-ENOTCONN);
2892                 }
2893         }
2894
2895         /* else: note the opposite is not always true; a RESENT req after a
2896          * failover will usually not match the last_xid, since it was likely
2897          * never committed. A REPLAYed request will almost never match the
2898          * last xid, however it could for a committed, but still retained,
2899          * open. */
2900
2901         obd = req->rq_export->exp_obd;
2902
2903         /* Check for aborted recovery... */
2904         if (unlikely(obd->obd_recovering)) {
2905                 int rc;
2906                 int should_process;
2907                 DEBUG_REQ(D_INFO, req, "Got new replay");
2908                 rc = mdt_filter_recovery_request(req, obd, &should_process);
2909                 if (rc != 0 || !should_process)
2910                         RETURN(rc);
2911                 else if (should_process < 0) {
2912                         req->rq_status = should_process;
2913                         rc = ptlrpc_error(req);
2914                         RETURN(rc);
2915                 }
2916         }
2917         RETURN(+1);
2918 }
2919
2920 static int mdt_msg_check_version(struct lustre_msg *msg)
2921 {
2922         int rc;
2923
2924         switch (lustre_msg_get_opc(msg)) {
2925         case MDS_CONNECT:
2926         case MDS_DISCONNECT:
2927         case OBD_PING:
2928         case SEC_CTX_INIT:
2929         case SEC_CTX_INIT_CONT:
2930         case SEC_CTX_FINI:
2931                 rc = lustre_msg_check_version(msg, LUSTRE_OBD_VERSION);
2932                 if (rc)
2933                         CERROR("bad opc %u version %08x, expecting %08x\n",
2934                                lustre_msg_get_opc(msg),
2935                                lustre_msg_get_version(msg),
2936                                LUSTRE_OBD_VERSION);
2937                 break;
2938         case MDS_GETSTATUS:
2939         case MDS_GETATTR:
2940         case MDS_GETATTR_NAME:
2941         case MDS_STATFS:
2942         case MDS_READPAGE:
2943         case MDS_WRITEPAGE:
2944         case MDS_IS_SUBDIR:
2945         case MDS_REINT:
2946         case MDS_CLOSE:
2947         case MDS_DONE_WRITING:
2948         case MDS_PIN:
2949         case MDS_SYNC:
2950         case MDS_GETXATTR:
2951         case MDS_SETXATTR:
2952         case MDS_SET_INFO:
2953         case MDS_GET_INFO:
2954         case MDS_QUOTACHECK:
2955         case MDS_QUOTACTL:
2956         case QUOTA_DQACQ:
2957         case QUOTA_DQREL:
2958         case SEQ_QUERY:
2959         case FLD_QUERY:
2960                 rc = lustre_msg_check_version(msg, LUSTRE_MDS_VERSION);
2961                 if (rc)
2962                         CERROR("bad opc %u version %08x, expecting %08x\n",
2963                                lustre_msg_get_opc(msg),
2964                                lustre_msg_get_version(msg),
2965                                LUSTRE_MDS_VERSION);
2966                 break;
2967         case LDLM_ENQUEUE:
2968         case LDLM_CONVERT:
2969         case LDLM_BL_CALLBACK:
2970         case LDLM_CP_CALLBACK:
2971                 rc = lustre_msg_check_version(msg, LUSTRE_DLM_VERSION);
2972                 if (rc)
2973                         CERROR("bad opc %u version %08x, expecting %08x\n",
2974                                lustre_msg_get_opc(msg),
2975                                lustre_msg_get_version(msg),
2976                                LUSTRE_DLM_VERSION);
2977                 break;
2978         case OBD_LOG_CANCEL:
2979         case LLOG_ORIGIN_HANDLE_CREATE:
2980         case LLOG_ORIGIN_HANDLE_NEXT_BLOCK:
2981         case LLOG_ORIGIN_HANDLE_READ_HEADER:
2982         case LLOG_ORIGIN_HANDLE_CLOSE:
2983         case LLOG_ORIGIN_HANDLE_DESTROY:
2984         case LLOG_ORIGIN_HANDLE_PREV_BLOCK:
2985         case LLOG_CATINFO:
2986                 rc = lustre_msg_check_version(msg, LUSTRE_LOG_VERSION);
2987                 if (rc)
2988                         CERROR("bad opc %u version %08x, expecting %08x\n",
2989                                lustre_msg_get_opc(msg),
2990                                lustre_msg_get_version(msg),
2991                                LUSTRE_LOG_VERSION);
2992                 break;
2993         default:
2994                 CERROR("MDS unknown opcode %d\n", lustre_msg_get_opc(msg));
2995                 rc = -ENOTSUPP;
2996         }
2997         return rc;
2998 }
2999
3000 static int mdt_handle0(struct ptlrpc_request *req,
3001                        struct mdt_thread_info *info,
3002                        struct mdt_opc_slice *supported)
3003 {
3004         struct mdt_handler *h;
3005         struct lustre_msg  *msg;
3006         int                 rc;
3007
3008         ENTRY;
3009
3010         if (OBD_FAIL_CHECK_ORSET(OBD_FAIL_MDS_ALL_REQUEST_NET, OBD_FAIL_ONCE))
3011                 RETURN(0);
3012
3013         LASSERT(current->journal_info == NULL);
3014
3015         msg = req->rq_reqmsg;
3016         rc = mdt_msg_check_version(msg);
3017         if (likely(rc == 0)) {
3018                 rc = mdt_recovery(info);
3019                 if (likely(rc == +1)) {
3020                         h = mdt_handler_find(lustre_msg_get_opc(msg),
3021                                              supported);
3022                         if (likely(h != NULL)) {
3023                                 rc = mdt_req_handle(info, h, req);
3024                         } else {
3025                                 CERROR("The unsupported opc: 0x%x\n",
3026                                        lustre_msg_get_opc(msg) );
3027                                 req->rq_status = -ENOTSUPP;
3028                                 rc = ptlrpc_error(req);
3029                                 RETURN(rc);
3030                         }
3031                 }
3032         } else
3033                 CERROR(LUSTRE_MDT_NAME" drops mal-formed request\n");
3034         RETURN(rc);
3035 }
3036
3037 /*
3038  * MDT handler function called by ptlrpc service thread when request comes.
3039  *
3040  * XXX common "target" functionality should be factored into separate module
3041  * shared by mdt, ost and stand-alone services like fld.
3042  */
3043 static int mdt_handle_common(struct ptlrpc_request *req,
3044                              struct mdt_opc_slice *supported)
3045 {
3046         struct lu_env          *env;
3047         struct mdt_thread_info *info;
3048         int                     rc;
3049         ENTRY;
3050
3051         env = req->rq_svc_thread->t_env;
3052         LASSERT(env != NULL);
3053         LASSERT(env->le_ses != NULL);
3054         LASSERT(env->le_ctx.lc_thread == req->rq_svc_thread);
3055         info = lu_context_key_get(&env->le_ctx, &mdt_thread_key);
3056         LASSERT(info != NULL);
3057
3058         mdt_thread_info_init(req, info);
3059
3060         rc = mdt_handle0(req, info, supported);
3061
3062         mdt_thread_info_fini(info);
3063         RETURN(rc);
3064 }
3065
3066 /*
3067  * This is called from recovery code as handler of _all_ RPC types, FLD and SEQ
3068  * as well.
3069  */
3070 int mdt_recovery_handle(struct ptlrpc_request *req)
3071 {
3072         int rc;
3073         ENTRY;
3074
3075         switch (lustre_msg_get_opc(req->rq_reqmsg)) {
3076         case FLD_QUERY:
3077                 rc = mdt_handle_common(req, mdt_fld_handlers);
3078                 break;
3079         case SEQ_QUERY:
3080                 rc = mdt_handle_common(req, mdt_seq_handlers);
3081                 break;
3082         default:
3083                 rc = mdt_handle_common(req, mdt_regular_handlers);
3084                 break;
3085         }
3086
3087         RETURN(rc);
3088 }
3089
3090 static int mdt_regular_handle(struct ptlrpc_request *req)
3091 {
3092         return mdt_handle_common(req, mdt_regular_handlers);
3093 }
3094
3095 static int mdt_readpage_handle(struct ptlrpc_request *req)
3096 {
3097         return mdt_handle_common(req, mdt_readpage_handlers);
3098 }
3099
3100 static int mdt_xmds_handle(struct ptlrpc_request *req)
3101 {
3102         return mdt_handle_common(req, mdt_xmds_handlers);
3103 }
3104
3105 static int mdt_mdsc_handle(struct ptlrpc_request *req)
3106 {
3107         return mdt_handle_common(req, mdt_seq_handlers);
3108 }
3109
3110 static int mdt_mdss_handle(struct ptlrpc_request *req)
3111 {
3112         return mdt_handle_common(req, mdt_seq_handlers);
3113 }
3114
3115 static int mdt_dtss_handle(struct ptlrpc_request *req)
3116 {
3117         return mdt_handle_common(req, mdt_seq_handlers);
3118 }
3119
3120 static int mdt_fld_handle(struct ptlrpc_request *req)
3121 {
3122         return mdt_handle_common(req, mdt_fld_handlers);
3123 }
3124
3125 enum mdt_it_code {
3126         MDT_IT_OPEN,
3127         MDT_IT_OCREAT,
3128         MDT_IT_CREATE,
3129         MDT_IT_GETATTR,
3130         MDT_IT_READDIR,
3131         MDT_IT_LOOKUP,
3132         MDT_IT_UNLINK,
3133         MDT_IT_TRUNC,
3134         MDT_IT_GETXATTR,
3135         MDT_IT_LAYOUT,
3136         MDT_IT_NR
3137 };
3138
3139 static int mdt_intent_getattr(enum mdt_it_code opcode,
3140                               struct mdt_thread_info *info,
3141                               struct ldlm_lock **,
3142                               int);
3143 static int mdt_intent_reint(enum mdt_it_code opcode,
3144                             struct mdt_thread_info *info,
3145                             struct ldlm_lock **,
3146                             int);
3147
3148 static struct mdt_it_flavor {
3149         const struct req_format *it_fmt;
3150         __u32                    it_flags;
3151         int                    (*it_act)(enum mdt_it_code ,
3152                                          struct mdt_thread_info *,
3153                                          struct ldlm_lock **,
3154                                          int);
3155         long                     it_reint;
3156 } mdt_it_flavor[] = {
3157         [MDT_IT_OPEN]     = {
3158                 .it_fmt   = &RQF_LDLM_INTENT,
3159                 /*.it_flags = HABEO_REFERO,*/
3160                 .it_flags = 0,
3161                 .it_act   = mdt_intent_reint,
3162                 .it_reint = REINT_OPEN
3163         },
3164         [MDT_IT_OCREAT]   = {
3165                 .it_fmt   = &RQF_LDLM_INTENT,
3166                 .it_flags = MUTABOR,
3167                 .it_act   = mdt_intent_reint,
3168                 .it_reint = REINT_OPEN
3169         },
3170         [MDT_IT_CREATE]   = {
3171                 .it_fmt   = &RQF_LDLM_INTENT,
3172                 .it_flags = MUTABOR,
3173                 .it_act   = mdt_intent_reint,
3174                 .it_reint = REINT_CREATE
3175         },
3176         [MDT_IT_GETATTR]  = {
3177                 .it_fmt   = &RQF_LDLM_INTENT_GETATTR,
3178                 .it_flags = HABEO_REFERO,
3179                 .it_act   = mdt_intent_getattr
3180         },
3181         [MDT_IT_READDIR]  = {
3182                 .it_fmt   = NULL,
3183                 .it_flags = 0,
3184                 .it_act   = NULL
3185         },
3186         [MDT_IT_LOOKUP]   = {
3187                 .it_fmt   = &RQF_LDLM_INTENT_GETATTR,
3188                 .it_flags = HABEO_REFERO,
3189                 .it_act   = mdt_intent_getattr
3190         },
3191         [MDT_IT_UNLINK]   = {
3192                 .it_fmt   = &RQF_LDLM_INTENT_UNLINK,
3193                 .it_flags = MUTABOR,
3194                 .it_act   = NULL,
3195                 .it_reint = REINT_UNLINK
3196         },
3197         [MDT_IT_TRUNC]    = {
3198                 .it_fmt   = NULL,
3199                 .it_flags = MUTABOR,
3200                 .it_act   = NULL
3201         },
3202         [MDT_IT_GETXATTR] = {
3203                 .it_fmt   = NULL,
3204                 .it_flags = 0,
3205                 .it_act   = NULL
3206         },
3207         [MDT_IT_LAYOUT] = {
3208                 .it_fmt   = &RQF_LDLM_INTENT_GETATTR,
3209                 .it_flags = HABEO_REFERO,
3210                 .it_act   = mdt_intent_getattr
3211         }
3212 };
3213
3214 int mdt_intent_lock_replace(struct mdt_thread_info *info,
3215                             struct ldlm_lock **lockp,
3216                             struct ldlm_lock *new_lock,
3217                             struct mdt_lock_handle *lh,
3218                             int flags)
3219 {
3220         struct ptlrpc_request  *req = mdt_info_req(info);
3221         struct ldlm_lock       *lock = *lockp;
3222
3223         /*
3224          * Get new lock only for cases when possible resent did not find any
3225          * lock.
3226          */
3227         if (new_lock == NULL)
3228                 new_lock = ldlm_handle2lock_long(&lh->mlh_reg_lh, 0);
3229
3230         if (new_lock == NULL && (flags & LDLM_FL_INTENT_ONLY)) {
3231                 lh->mlh_reg_lh.cookie = 0;
3232                 RETURN(0);
3233         }
3234