Whamcloud - gitweb
LU-4629 lmv: fix issue found by Klocwork Insight tool
[fs/lustre-release.git] / lustre / lmv / lmv_intent.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) 2004, 2010, Oracle and/or its affiliates. All rights reserved.
28  * Use is subject to license terms.
29  *
30  * Copyright (c) 2011, 2013, Intel Corporation.
31  */
32 /*
33  * This file is part of Lustre, http://www.lustre.org/
34  * Lustre is a trademark of Sun Microsystems, Inc.
35  */
36
37 #define DEBUG_SUBSYSTEM S_LMV
38 #ifdef __KERNEL__
39 #include <linux/slab.h>
40 #include <linux/module.h>
41 #include <linux/init.h>
42 #include <linux/slab.h>
43 #include <linux/pagemap.h>
44 #include <asm/div64.h>
45 #include <linux/seq_file.h>
46 #include <linux/namei.h>
47 #include <linux/lustre_intent.h>
48 #else
49 #include <liblustre.h>
50 #endif
51
52 #include <obd_support.h>
53 #include <lustre/lustre_idl.h>
54 #include <lustre_lib.h>
55 #include <lustre_net.h>
56 #include <lustre_dlm.h>
57 #include <lustre_mdc.h>
58 #include <obd_class.h>
59 #include <lprocfs_status.h>
60 #include "lmv_internal.h"
61
62 static int lmv_intent_remote(struct obd_export *exp, void *lmm,
63                              int lmmsize, struct lookup_intent *it,
64                              const struct lu_fid *parent_fid, int flags,
65                              struct ptlrpc_request **reqp,
66                              ldlm_blocking_callback cb_blocking,
67                              __u64 extra_lock_flags)
68 {
69         struct obd_device       *obd = exp->exp_obd;
70         struct lmv_obd          *lmv = &obd->u.lmv;
71         struct ptlrpc_request   *req = NULL;
72         struct lustre_handle    plock;
73         struct md_op_data       *op_data;
74         struct lmv_tgt_desc     *tgt;
75         struct mdt_body         *body;
76         int                     pmode;
77         int                     rc = 0;
78         ENTRY;
79
80         body = req_capsule_server_get(&(*reqp)->rq_pill, &RMF_MDT_BODY);
81         if (body == NULL)
82                 RETURN(-EPROTO);
83
84         LASSERT((body->valid & OBD_MD_MDS));
85
86         /*
87          * Unfortunately, we have to lie to MDC/MDS to retrieve
88          * attributes llite needs and provideproper locking.
89          */
90         if (it->it_op & IT_LOOKUP)
91                 it->it_op = IT_GETATTR;
92
93         /*
94          * We got LOOKUP lock, but we really need attrs.
95          */
96         pmode = it->d.lustre.it_lock_mode;
97         if (pmode) {
98                 plock.cookie = it->d.lustre.it_lock_handle;
99                 it->d.lustre.it_lock_mode = 0;
100                 it->d.lustre.it_data = NULL;
101         }
102
103         LASSERT(fid_is_sane(&body->fid1));
104
105         tgt = lmv_find_target(lmv, &body->fid1);
106         if (IS_ERR(tgt))
107                 GOTO(out, rc = PTR_ERR(tgt));
108
109         OBD_ALLOC_PTR(op_data);
110         if (op_data == NULL)
111                 GOTO(out, rc = -ENOMEM);
112
113         op_data->op_fid1 = body->fid1;
114         /* Sent the parent FID to the remote MDT */
115         if (parent_fid != NULL) {
116                 /* The parent fid is only for remote open to
117                  * check whether the open is from OBF,
118                  * see mdt_cross_open */
119                 LASSERT(it->it_op & IT_OPEN);
120                 op_data->op_fid2 = *parent_fid;
121                 /* Add object FID to op_fid3, in case it needs to check stale
122                  * (M_CHECK_STALE), see mdc_finish_intent_lock */
123                 op_data->op_fid3 = body->fid1;
124         }
125
126         op_data->op_bias = MDS_CROSS_REF;
127         CDEBUG(D_INODE, "REMOTE_INTENT with fid="DFID" -> mds #%d\n",
128                PFID(&body->fid1), tgt->ltd_idx);
129
130         rc = md_intent_lock(tgt->ltd_exp, op_data, lmm, lmmsize, it,
131                             flags, &req, cb_blocking, extra_lock_flags);
132         if (rc)
133                 GOTO(out_free_op_data, rc);
134
135         /*
136          * LLite needs LOOKUP lock to track dentry revocation in order to
137          * maintain dcache consistency. Thus drop UPDATE|PERM lock here
138          * and put LOOKUP in request.
139          */
140         if (it->d.lustre.it_lock_mode != 0) {
141                 it->d.lustre.it_remote_lock_handle =
142                                         it->d.lustre.it_lock_handle;
143                 it->d.lustre.it_remote_lock_mode = it->d.lustre.it_lock_mode;
144         }
145
146         if (pmode) {
147                 it->d.lustre.it_lock_handle = plock.cookie;
148                 it->d.lustre.it_lock_mode = pmode;
149         }
150
151         EXIT;
152 out_free_op_data:
153         OBD_FREE_PTR(op_data);
154 out:
155         if (rc && pmode)
156                 ldlm_lock_decref(&plock, pmode);
157
158         ptlrpc_req_finished(*reqp);
159         *reqp = req;
160         return rc;
161 }
162
163 #ifdef __KERNEL__
164 int lmv_revalidate_slaves(struct obd_export *exp, struct mdt_body *mbody,
165                           struct lmv_stripe_md *lsm,
166                           ldlm_blocking_callback cb_blocking,
167                           int extra_lock_flags)
168 {
169         struct obd_device      *obd = exp->exp_obd;
170         struct lmv_obd         *lmv = &obd->u.lmv;
171         struct mdt_body         *body;
172         struct md_op_data      *op_data;
173         unsigned long           size = 0;
174         unsigned long           nlink = 0;
175         obd_time                atime = 0;
176         obd_time                ctime = 0;
177         obd_time                mtime = 0;
178         int                     i;
179         int                     rc = 0;
180
181         ENTRY;
182
183         /**
184          * revalidate slaves has some problems, temporarily return,
185          * we may not need that
186          */
187         if (lsm->lsm_md_stripe_count <= 1)
188                 RETURN(0);
189
190         OBD_ALLOC_PTR(op_data);
191         if (op_data == NULL)
192                 RETURN(-ENOMEM);
193
194         /**
195          * Loop over the stripe information, check validity and update them
196          * from MDS if needed.
197          */
198         for (i = 0; i < lsm->lsm_md_stripe_count; i++) {
199                 struct lu_fid           fid;
200                 struct lookup_intent    it = { .it_op = IT_GETATTR };
201                 struct ptlrpc_request   *req = NULL;
202                 struct lustre_handle    *lockh = NULL;
203                 struct lmv_tgt_desc     *tgt = NULL;
204                 struct inode            *inode;
205
206                 fid = lsm->lsm_md_oinfo[i].lmo_fid;
207                 inode = lsm->lsm_md_oinfo[i].lmo_root;
208                 if (i == 0) {
209                         if (mbody != NULL) {
210                                 body = mbody;
211                                 goto update;
212                         } else {
213                                 goto release_lock;
214                         }
215                 }
216
217                 /*
218                  * Prepare op_data for revalidating. Note that @fid2 shluld be
219                  * defined otherwise it will go to server and take new lock
220                  * which is not needed here.
221                  */
222                 memset(op_data, 0, sizeof(*op_data));
223                 op_data->op_fid1 = fid;
224                 op_data->op_fid2 = fid;
225
226                 tgt = lmv_locate_mds(lmv, op_data, &fid);
227                 if (IS_ERR(tgt))
228                         GOTO(cleanup, rc = PTR_ERR(tgt));
229
230                 CDEBUG(D_INODE, "Revalidate slave "DFID" -> mds #%d\n",
231                        PFID(&fid), tgt->ltd_idx);
232
233                 rc = md_intent_lock(tgt->ltd_exp, op_data, NULL, 0, &it, 0,
234                                     &req, cb_blocking, extra_lock_flags);
235                 if (rc < 0)
236                         GOTO(cleanup, rc);
237
238                 lockh = (struct lustre_handle *)&it.d.lustre.it_lock_handle;
239                 if (rc > 0 && req == NULL) {
240                         /* slave inode is still valid */
241                         CDEBUG(D_INODE, "slave "DFID" is still valid.\n",
242                                PFID(&fid));
243                         rc = 0;
244                 } else {
245                         /* refresh slave from server */
246                         body = req_capsule_server_get(&req->rq_pill,
247                                                       &RMF_MDT_BODY);
248                         LASSERT(body != NULL);
249 update:
250                         if (unlikely(body->nlink < 2)) {
251                                 CERROR("%s: nlink %d < 2 corrupt stripe %d "DFID
252                                        ":" DFID"\n", obd->obd_name, body->nlink,
253                                        i, PFID(&lsm->lsm_md_oinfo[i].lmo_fid),
254                                        PFID(&lsm->lsm_md_oinfo[0].lmo_fid));
255
256                                 if (req != NULL)
257                                         ptlrpc_req_finished(req);
258
259                                 if (it.d.lustre.it_lock_mode && lockh) {
260                                         ldlm_lock_decref(lockh,
261                                                  it.d.lustre.it_lock_mode);
262                                         it.d.lustre.it_lock_mode = 0;
263                                 }
264
265                                 GOTO(cleanup, rc = -EIO);
266                         }
267
268                         if (i != 0)
269                                 md_set_lock_data(tgt->ltd_exp, &lockh->cookie,
270                                                  inode, NULL);
271
272                         i_size_write(inode, body->size);
273                         set_nlink(inode, body->nlink);
274                         LTIME_S(inode->i_atime) = body->atime;
275                         LTIME_S(inode->i_ctime) = body->ctime;
276                         LTIME_S(inode->i_mtime) = body->mtime;
277
278                         if (req != NULL)
279                                 ptlrpc_req_finished(req);
280                 }
281 release_lock:
282                 size += i_size_read(inode);
283
284                 if (i != 0)
285                         nlink += inode->i_nlink - 2;
286                 else
287                         nlink += inode->i_nlink;
288
289                 atime = LTIME_S(inode->i_atime) > atime ?
290                                 LTIME_S(inode->i_atime) : atime;
291                 ctime = LTIME_S(inode->i_ctime) > ctime ?
292                                 LTIME_S(inode->i_ctime) : ctime;
293                 mtime = LTIME_S(inode->i_mtime) > mtime ?
294                                 LTIME_S(inode->i_mtime) : mtime;
295
296                 if (it.d.lustre.it_lock_mode != 0 && lockh != NULL) {
297                         ldlm_lock_decref(lockh, it.d.lustre.it_lock_mode);
298                         it.d.lustre.it_lock_mode = 0;
299                 }
300
301                 CDEBUG(D_INODE, "i %d "DFID" size %llu, nlink %u, atime "
302                        "%lu, mtime %lu, ctime %lu.\n", i, PFID(&fid),
303                        i_size_read(inode), inode->i_nlink,
304                        LTIME_S(inode->i_atime), LTIME_S(inode->i_mtime),
305                        LTIME_S(inode->i_ctime));
306         }
307
308         /*
309          * update attr of master request.
310          */
311         CDEBUG(D_INODE, "Return refreshed attrs: size = %lu nlink %lu atime "
312                LPU64 "ctime "LPU64" mtime "LPU64" for " DFID"\n", size, nlink,
313                atime, ctime, mtime, PFID(&lsm->lsm_md_oinfo[0].lmo_fid));
314
315         if (mbody != NULL) {
316                 mbody->atime = atime;
317                 mbody->ctime = ctime;
318                 mbody->mtime = mtime;
319         }
320 cleanup:
321         OBD_FREE_PTR(op_data);
322         RETURN(rc);
323 }
324
325 #else
326
327 int lmv_revalidate_slaves(struct obd_export *exp, struct mdt_body *mbody,
328                           struct lmv_stripe_md *lsm,
329                           ldlm_blocking_callback cb_blocking,
330                           int extra_lock_flags)
331 {
332         return 0;
333 }
334
335 #endif
336
337 /*
338  * IT_OPEN is intended to open (and create, possible) an object. Parent (pid)
339  * may be split dir.
340  */
341 int lmv_intent_open(struct obd_export *exp, struct md_op_data *op_data,
342                     void *lmm, int lmmsize, struct lookup_intent *it,
343                     int flags, struct ptlrpc_request **reqp,
344                     ldlm_blocking_callback cb_blocking,
345                     __u64 extra_lock_flags)
346 {
347         struct obd_device       *obd = exp->exp_obd;
348         struct lmv_obd          *lmv = &obd->u.lmv;
349         struct lmv_tgt_desc     *tgt;
350         struct mdt_body         *body;
351         int                     rc;
352         ENTRY;
353
354         /* Note: client might open with some random flags(sanity 33b), so we can
355          * not make sure op_fid2 is being initialized with BY_FID flag */
356         if (it->it_flags & MDS_OPEN_BY_FID && fid_is_sane(&op_data->op_fid2)) {
357                 if (op_data->op_mea1 != NULL) {
358                         struct lmv_stripe_md    *lsm = op_data->op_mea1;
359                         const struct lmv_oinfo  *oinfo;
360
361                         oinfo = lsm_name_to_stripe_info(lsm, op_data->op_name,
362                                                         op_data->op_namelen);
363                         if (IS_ERR(oinfo))
364                                 RETURN(PTR_ERR(oinfo));
365                         op_data->op_fid1 = oinfo->lmo_fid;
366                 }
367
368                 tgt = lmv_find_target(lmv, &op_data->op_fid2);
369                 if (IS_ERR(tgt))
370                         RETURN(PTR_ERR(tgt));
371
372                 op_data->op_mds = tgt->ltd_idx;
373         } else {
374                 tgt = lmv_locate_mds(lmv, op_data, &op_data->op_fid1);
375                 if (IS_ERR(tgt))
376                         RETURN(PTR_ERR(tgt));
377         }
378
379         /* If it is ready to open the file by FID, do not need
380          * allocate FID at all, otherwise it will confuse MDT */
381         if ((it->it_op & IT_CREAT) &&
382             !(it->it_flags & MDS_OPEN_BY_FID)) {
383                 /*
384                  * For open with IT_CREATE and for IT_CREATE cases allocate new
385                  * fid and setup FLD for it.
386                  */
387                 op_data->op_fid3 = op_data->op_fid2;
388                 rc = lmv_fid_alloc(exp, &op_data->op_fid2, op_data);
389                 if (rc != 0)
390                         RETURN(rc);
391         }
392
393         CDEBUG(D_INODE, "OPEN_INTENT with fid1="DFID", fid2="DFID","
394                " name='%s' -> mds #%d\n", PFID(&op_data->op_fid1),
395                PFID(&op_data->op_fid2), op_data->op_name, tgt->ltd_idx);
396
397         rc = md_intent_lock(tgt->ltd_exp, op_data, lmm, lmmsize, it, flags,
398                             reqp, cb_blocking, extra_lock_flags);
399         if (rc != 0)
400                 RETURN(rc);
401         /*
402          * Nothing is found, do not access body->fid1 as it is zero and thus
403          * pointless.
404          */
405         if ((it->d.lustre.it_disposition & DISP_LOOKUP_NEG) &&
406             !(it->d.lustre.it_disposition & DISP_OPEN_CREATE) &&
407             !(it->d.lustre.it_disposition & DISP_OPEN_OPEN))
408                 RETURN(rc);
409
410         body = req_capsule_server_get(&(*reqp)->rq_pill, &RMF_MDT_BODY);
411         if (body == NULL)
412                 RETURN(-EPROTO);
413
414         /* Not cross-ref case, just get out of here. */
415         if (unlikely((body->valid & OBD_MD_MDS))) {
416                 rc = lmv_intent_remote(exp, lmm, lmmsize, it, &op_data->op_fid1,
417                                        flags, reqp, cb_blocking,
418                                        extra_lock_flags);
419                 if (rc != 0)
420                         RETURN(rc);
421
422                 body = req_capsule_server_get(&(*reqp)->rq_pill, &RMF_MDT_BODY);
423                 if (body == NULL)
424                         RETURN(-EPROTO);
425         }
426
427         RETURN(rc);
428 }
429
430 /*
431  * Handler for: getattr, lookup and revalidate cases.
432  */
433 int lmv_intent_lookup(struct obd_export *exp, struct md_op_data *op_data,
434                       void *lmm, int lmmsize, struct lookup_intent *it,
435                       int flags, struct ptlrpc_request **reqp,
436                       ldlm_blocking_callback cb_blocking,
437                       __u64 extra_lock_flags)
438 {
439         struct obd_device       *obd = exp->exp_obd;
440         struct lmv_obd          *lmv = &obd->u.lmv;
441         struct lmv_tgt_desc     *tgt = NULL;
442         struct mdt_body         *body;
443         struct lmv_stripe_md    *lsm = op_data->op_mea1;
444         int                     rc = 0;
445         ENTRY;
446
447         tgt = lmv_locate_mds(lmv, op_data, &op_data->op_fid1);
448         if (IS_ERR(tgt))
449                 RETURN(PTR_ERR(tgt));
450
451         if (!fid_is_sane(&op_data->op_fid2))
452                 fid_zero(&op_data->op_fid2);
453
454         CDEBUG(D_INODE, "LOOKUP_INTENT with fid1="DFID", fid2="DFID
455                ", name='%s' -> mds #%d lsm=%p lsm_magic=%x\n",
456                PFID(&op_data->op_fid1), PFID(&op_data->op_fid2),
457                op_data->op_name ? op_data->op_name : "<NULL>",
458                tgt->ltd_idx, lsm, lsm == NULL ? -1 : lsm->lsm_md_magic);
459
460         op_data->op_bias &= ~MDS_CROSS_REF;
461
462         rc = md_intent_lock(tgt->ltd_exp, op_data, lmm, lmmsize, it,
463                              flags, reqp, cb_blocking, extra_lock_flags);
464         if (rc < 0)
465                 RETURN(rc);
466
467         if (*reqp == NULL) {
468                 /* If RPC happens, lsm information will be revalidated
469                  * during update_inode process (see ll_update_lsm_md) */
470                 if (op_data->op_mea2 != NULL) {
471                         rc = lmv_revalidate_slaves(exp, NULL, op_data->op_mea2,
472                                                    cb_blocking,
473                                                    extra_lock_flags);
474                         if (rc != 0)
475                                 RETURN(rc);
476                 }
477                 RETURN(rc);
478         } else if (it_disposition(it, DISP_LOOKUP_NEG) &&
479                    lsm != NULL && lsm->lsm_md_magic == LMV_MAGIC_MIGRATE) {
480                 /* For migrating directory, if it can not find the child in
481                  * the source directory(master stripe), try the targeting
482                  * directory(stripe 1) */
483                 tgt = lmv_find_target(lmv, &lsm->lsm_md_oinfo[1].lmo_fid);
484                 if (IS_ERR(tgt))
485                         RETURN(PTR_ERR(tgt));
486
487                 ptlrpc_req_finished(*reqp);
488                 CDEBUG(D_INODE, "For migrating dir, try target dir "DFID"\n",
489                        PFID(&lsm->lsm_md_oinfo[1].lmo_fid));
490
491                 op_data->op_fid1 = lsm->lsm_md_oinfo[1].lmo_fid;
492                 it->d.lustre.it_disposition &= ~DISP_ENQ_COMPLETE;
493                 rc = md_intent_lock(tgt->ltd_exp, op_data, lmm, lmmsize, it,
494                                     flags, reqp, cb_blocking, extra_lock_flags);
495                 RETURN(rc);
496         }
497         /*
498          * MDS has returned success. Probably name has been resolved in
499          * remote inode. Let's check this.
500          */
501         body = req_capsule_server_get(&(*reqp)->rq_pill, &RMF_MDT_BODY);
502         if (body == NULL)
503                 RETURN(-EPROTO);
504
505         /* Not cross-ref case, just get out of here. */
506         if (unlikely((body->valid & OBD_MD_MDS))) {
507                 rc = lmv_intent_remote(exp, lmm, lmmsize, it, NULL, flags,
508                                        reqp, cb_blocking, extra_lock_flags);
509                 if (rc != 0)
510                         RETURN(rc);
511                 body = req_capsule_server_get(&(*reqp)->rq_pill, &RMF_MDT_BODY);
512                 if (body == NULL)
513                         RETURN(-EPROTO);
514         }
515
516         RETURN(rc);
517 }
518
519 int lmv_intent_lock(struct obd_export *exp, struct md_op_data *op_data,
520                     void *lmm, int lmmsize, struct lookup_intent *it,
521                     int flags, struct ptlrpc_request **reqp,
522                     ldlm_blocking_callback cb_blocking,
523                     __u64 extra_lock_flags)
524 {
525         struct obd_device *obd = exp->exp_obd;
526         int                rc;
527         ENTRY;
528
529         LASSERT(it != NULL);
530         LASSERT(fid_is_sane(&op_data->op_fid1));
531
532         CDEBUG(D_INODE, "INTENT LOCK '%s' for '%*s' on "DFID"\n",
533                LL_IT2STR(it), op_data->op_namelen, op_data->op_name,
534                PFID(&op_data->op_fid1));
535
536         rc = lmv_check_connect(obd);
537         if (rc)
538                 RETURN(rc);
539
540         if (it->it_op & (IT_LOOKUP | IT_GETATTR | IT_LAYOUT))
541                 rc = lmv_intent_lookup(exp, op_data, lmm, lmmsize, it,
542                                        flags, reqp, cb_blocking,
543                                        extra_lock_flags);
544         else if (it->it_op & IT_OPEN)
545                 rc = lmv_intent_open(exp, op_data, lmm, lmmsize, it,
546                                      flags, reqp, cb_blocking,
547                                      extra_lock_flags);
548         else
549                 LBUG();
550         RETURN(rc);
551 }