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