Whamcloud - gitweb
637f5eebf529e2895ae286005111c761646b8375
[fs/lustre-release.git] / lustre / mdt / mdt_io.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.gnu.org/licenses/gpl-2.0.html
19  *
20  * GPL HEADER END
21  */
22 /*
23  * Copyright (c) 2017, Intel Corporation.
24  */
25 /*
26  * lustre/mdt/mdt_io.c
27  *
28  * Author: Mikhail Pershin <mike.pershin@intel.com>
29  */
30
31 #define DEBUG_SUBSYSTEM S_FILTER
32
33 #include <dt_object.h>
34 #include "mdt_internal.h"
35
36 /* functions below are stubs for now, they will be implemented with
37  * grant support on MDT */
38 static inline void mdt_io_counter_incr(struct obd_export *exp, int opcode,
39                                        char *jobid, long amount)
40 {
41         return;
42 }
43
44 static inline void mdt_dom_read_lock(struct mdt_object *mo)
45 {
46         down_read(&mo->mot_dom_sem);
47 }
48
49 static inline void mdt_dom_read_unlock(struct mdt_object *mo)
50 {
51         up_read(&mo->mot_dom_sem);
52 }
53
54 static inline void mdt_dom_write_lock(struct mdt_object *mo)
55 {
56         down_write(&mo->mot_dom_sem);
57 }
58
59 static inline void mdt_dom_write_unlock(struct mdt_object *mo)
60 {
61         up_write(&mo->mot_dom_sem);
62 }
63
64 /**
65  * Lock prolongation for Data-on-MDT.
66  * This is similar to OFD code but for DOM ibits lock.
67  */
68 static inline time64_t prolong_timeout(struct ptlrpc_request *req)
69 {
70         struct ptlrpc_service_part *svcpt = req->rq_rqbd->rqbd_svcpt;
71         time64_t req_timeout;
72
73         if (AT_OFF)
74                 return obd_timeout / 2;
75
76         req_timeout = req->rq_deadline - req->rq_arrival_time.tv_sec;
77         return max_t(time64_t, at_est2timeout(at_get(&svcpt->scp_at_estimate)),
78                      req_timeout);
79 }
80
81 static void mdt_prolong_dom_lock(struct tgt_session_info *tsi,
82                                  struct ldlm_prolong_args *data)
83 {
84         struct obdo *oa = &tsi->tsi_ost_body->oa;
85         struct ldlm_lock *lock;
86
87         ENTRY;
88
89         data->lpa_timeout = prolong_timeout(tgt_ses_req(tsi));
90         data->lpa_export = tsi->tsi_exp;
91         data->lpa_resid = tsi->tsi_resid;
92
93         CDEBUG(D_RPCTRACE, "Prolong DOM lock for req %p with x%llu\n",
94                tgt_ses_req(tsi), tgt_ses_req(tsi)->rq_xid);
95
96         if (oa->o_valid & OBD_MD_FLHANDLE) {
97                 /* mostly a request should be covered by only one lock, try
98                  * fast path. */
99                 lock = ldlm_handle2lock(&oa->o_handle);
100                 if (lock != NULL) {
101                         LASSERT(lock->l_export == data->lpa_export);
102                         ldlm_lock_prolong_one(lock, data);
103                         lock->l_last_used = ktime_get();
104                         LDLM_LOCK_PUT(lock);
105                         RETURN_EXIT;
106                 }
107         }
108         EXIT;
109 }
110
111 static int mdt_rw_hpreq_lock_match(struct ptlrpc_request *req,
112                                    struct ldlm_lock *lock)
113 {
114         struct obd_ioobj *ioo;
115         enum ldlm_mode mode;
116         __u32 opc = lustre_msg_get_opc(req->rq_reqmsg);
117
118         ENTRY;
119
120         if (!(lock->l_policy_data.l_inodebits.bits & MDS_INODELOCK_DOM))
121                 RETURN(0);
122
123         ioo = req_capsule_client_get(&req->rq_pill, &RMF_OBD_IOOBJ);
124         LASSERT(ioo != NULL);
125
126         LASSERT(lock->l_resource != NULL);
127         if (!fid_res_name_eq(&ioo->ioo_oid.oi_fid, &lock->l_resource->lr_name))
128                 RETURN(0);
129
130         /* a bulk write can only hold a reference on a PW extent lock. */
131         mode = LCK_PW;
132         if (opc == OST_READ)
133                 /* whereas a bulk read can be protected by either a PR or PW
134                  * extent lock */
135                 mode |= LCK_PR;
136
137         if (!(lock->l_granted_mode & mode))
138                 RETURN(0);
139
140         RETURN(1);
141 }
142
143 static int mdt_rw_hpreq_check(struct ptlrpc_request *req)
144 {
145         struct tgt_session_info *tsi;
146         struct obd_ioobj *ioo;
147         struct niobuf_remote *rnb;
148         int opc;
149         struct ldlm_prolong_args pa = { 0 };
150
151         ENTRY;
152
153         /* Don't use tgt_ses_info() to get session info, because lock_match()
154          * can be called while request has no processing thread yet. */
155         tsi = lu_context_key_get(&req->rq_session, &tgt_session_key);
156
157         /*
158          * Use LASSERT below because malformed RPCs should have
159          * been filtered out in tgt_hpreq_handler().
160          */
161         opc = lustre_msg_get_opc(req->rq_reqmsg);
162         LASSERT(opc == OST_READ || opc == OST_WRITE);
163
164         ioo = req_capsule_client_get(&req->rq_pill, &RMF_OBD_IOOBJ);
165         LASSERT(ioo != NULL);
166
167         rnb = req_capsule_client_get(&req->rq_pill, &RMF_NIOBUF_REMOTE);
168         LASSERT(rnb != NULL);
169         LASSERT(!(rnb->rnb_flags & OBD_BRW_SRVLOCK));
170
171         pa.lpa_mode = LCK_PW;
172         if (opc == OST_READ)
173                 pa.lpa_mode |= LCK_PR;
174
175         DEBUG_REQ(D_RPCTRACE, req, "%s %s: refresh rw locks: "DFID"\n",
176                   tgt_name(tsi->tsi_tgt), current->comm, PFID(&tsi->tsi_fid));
177
178         mdt_prolong_dom_lock(tsi, &pa);
179
180         if (pa.lpa_blocks_cnt > 0) {
181                 CDEBUG(D_DLMTRACE,
182                        "%s: refreshed %u locks timeout for req %p.\n",
183                        tgt_name(tsi->tsi_tgt), pa.lpa_blocks_cnt, req);
184                 RETURN(1);
185         }
186
187         RETURN(pa.lpa_locks_cnt > 0 ? 0 : -ESTALE);
188 }
189
190 static void mdt_rw_hpreq_fini(struct ptlrpc_request *req)
191 {
192         mdt_rw_hpreq_check(req);
193 }
194
195 static struct ptlrpc_hpreq_ops mdt_hpreq_rw = {
196         .hpreq_lock_match = mdt_rw_hpreq_lock_match,
197         .hpreq_check = mdt_rw_hpreq_check,
198         .hpreq_fini = mdt_rw_hpreq_fini
199 };
200
201 /**
202  * Assign high priority operations to an IO request.
203  *
204  * Check if the incoming request is a candidate for
205  * high-priority processing. If it is, assign it a high
206  * priority operations table.
207  *
208  * \param[in] tsi       target session environment for this request
209  */
210 void mdt_hp_brw(struct tgt_session_info *tsi)
211 {
212         struct niobuf_remote    *rnb;
213         struct obd_ioobj        *ioo;
214
215         ENTRY;
216
217         ioo = req_capsule_client_get(tsi->tsi_pill, &RMF_OBD_IOOBJ);
218         LASSERT(ioo != NULL); /* must exist after request preprocessing */
219         if (ioo->ioo_bufcnt > 0) {
220                 rnb = req_capsule_client_get(tsi->tsi_pill, &RMF_NIOBUF_REMOTE);
221                 LASSERT(rnb != NULL); /* must exist after preprocessing */
222
223                 /* no high priority if server lock is needed */
224                 if (rnb->rnb_flags & OBD_BRW_SRVLOCK ||
225                     (lustre_msg_get_flags(tgt_ses_req(tsi)->rq_reqmsg) &
226                      MSG_REPLAY))
227                         return;
228         }
229         tgt_ses_req(tsi)->rq_ops = &mdt_hpreq_rw;
230 }
231
232 static int mdt_punch_hpreq_lock_match(struct ptlrpc_request *req,
233                                       struct ldlm_lock *lock)
234 {
235         struct tgt_session_info *tsi;
236         struct obdo *oa;
237
238         ENTRY;
239
240         /* Don't use tgt_ses_info() to get session info, because lock_match()
241          * can be called while request has no processing thread yet. */
242         tsi = lu_context_key_get(&req->rq_session, &tgt_session_key);
243
244         /*
245          * Use LASSERT below because malformed RPCs should have
246          * been filtered out in tgt_hpreq_handler().
247          */
248         LASSERT(tsi->tsi_ost_body != NULL);
249         if (tsi->tsi_ost_body->oa.o_valid & OBD_MD_FLHANDLE &&
250             tsi->tsi_ost_body->oa.o_handle.cookie == lock->l_handle.h_cookie)
251                 RETURN(1);
252
253         oa = &tsi->tsi_ost_body->oa;
254
255         LASSERT(lock->l_resource != NULL);
256         if (!fid_res_name_eq(&oa->o_oi.oi_fid, &lock->l_resource->lr_name))
257                 RETURN(0);
258
259         if (!(lock->l_granted_mode & LCK_PW))
260                 RETURN(0);
261
262         RETURN(1);
263 }
264
265 /**
266  * Implementation of ptlrpc_hpreq_ops::hpreq_lock_check for OST_PUNCH request.
267  *
268  * High-priority queue request check for whether the given punch request
269  * (\a req) is blocking an LDLM lock cancel. Also checks whether the request is
270  * covered by an LDLM lock.
271  *
272
273  *
274  * \param[in] req       the incoming request
275  *
276  * \retval              1 if \a req is blocking an LDLM lock cancel
277  * \retval              0 if it is not
278  * \retval              -ESTALE if lock is not found
279  */
280 static int mdt_punch_hpreq_check(struct ptlrpc_request *req)
281 {
282         struct tgt_session_info *tsi;
283         struct obdo *oa;
284         struct ldlm_prolong_args pa = { 0 };
285
286         ENTRY;
287
288         /* Don't use tgt_ses_info() to get session info, because lock_match()
289          * can be called while request has no processing thread yet. */
290         tsi = lu_context_key_get(&req->rq_session, &tgt_session_key);
291         LASSERT(tsi != NULL);
292         oa = &tsi->tsi_ost_body->oa;
293
294         LASSERT(!(oa->o_valid & OBD_MD_FLFLAGS &&
295                   oa->o_flags & OBD_FL_SRVLOCK));
296
297         pa.lpa_mode = LCK_PW;
298
299         CDEBUG(D_DLMTRACE, "%s: refresh DOM lock for "DFID"\n",
300                tgt_name(tsi->tsi_tgt), PFID(&tsi->tsi_fid));
301
302         mdt_prolong_dom_lock(tsi, &pa);
303
304
305         if (pa.lpa_blocks_cnt > 0) {
306                 CDEBUG(D_DLMTRACE,
307                        "%s: refreshed %u locks timeout for req %p.\n",
308                        tgt_name(tsi->tsi_tgt), pa.lpa_blocks_cnt, req);
309                 RETURN(1);
310         }
311
312         RETURN(pa.lpa_locks_cnt > 0 ? 0 : -ESTALE);
313 }
314
315 /**
316  * Implementation of ptlrpc_hpreq_ops::hpreq_lock_fini for OST_PUNCH request.
317  *
318  * Called after the request has been handled. It refreshes lock timeout again
319  * so that client has more time to send lock cancel RPC.
320  *
321  * \param[in] req       request which is being processed.
322  */
323 static void mdt_punch_hpreq_fini(struct ptlrpc_request *req)
324 {
325         mdt_punch_hpreq_check(req);
326 }
327
328 static struct ptlrpc_hpreq_ops mdt_hpreq_punch = {
329         .hpreq_lock_match = mdt_punch_hpreq_lock_match,
330         .hpreq_check = mdt_punch_hpreq_check,
331         .hpreq_fini = mdt_punch_hpreq_fini
332 };
333
334 void mdt_hp_punch(struct tgt_session_info *tsi)
335 {
336         LASSERT(tsi->tsi_ost_body != NULL); /* must exists if we are here */
337         /* no high-priority if server lock is needed */
338         if ((tsi->tsi_ost_body->oa.o_valid & OBD_MD_FLFLAGS &&
339              tsi->tsi_ost_body->oa.o_flags & OBD_FL_SRVLOCK) ||
340             tgt_conn_flags(tsi) & OBD_CONNECT_MDS ||
341             lustre_msg_get_flags(tgt_ses_req(tsi)->rq_reqmsg) & MSG_REPLAY)
342                 return;
343         tgt_ses_req(tsi)->rq_ops = &mdt_hpreq_punch;
344 }
345
346 static int mdt_preprw_read(const struct lu_env *env, struct obd_export *exp,
347                            struct mdt_device *mdt, struct mdt_object *mo,
348                            struct lu_attr *la, int niocount,
349                            struct niobuf_remote *rnb, int *nr_local,
350                            struct niobuf_local *lnb, char *jobid)
351 {
352         struct dt_object *dob;
353         int i, j, rc, tot_bytes = 0;
354
355         ENTRY;
356
357         mdt_dom_read_lock(mo);
358         if (!mdt_object_exists(mo))
359                 GOTO(unlock, rc = -ENOENT);
360
361         dob = mdt_obj2dt(mo);
362         /* parse remote buffers to local buffers and prepare the latter */
363         *nr_local = 0;
364         for (i = 0, j = 0; i < niocount; i++) {
365                 rc = dt_bufs_get(env, dob, rnb + i, lnb + j, 0);
366                 if (unlikely(rc < 0))
367                         GOTO(buf_put, rc);
368                 /* correct index for local buffers to continue with */
369                 j += rc;
370                 *nr_local += rc;
371                 tot_bytes += rnb[i].rnb_len;
372         }
373
374         rc = dt_attr_get(env, dob, la);
375         if (unlikely(rc))
376                 GOTO(buf_put, rc);
377
378         rc = dt_read_prep(env, dob, lnb, *nr_local);
379         if (unlikely(rc))
380                 GOTO(buf_put, rc);
381
382         mdt_io_counter_incr(exp, LPROC_MDT_IO_READ, jobid, tot_bytes);
383         RETURN(0);
384 buf_put:
385         dt_bufs_put(env, dob, lnb, *nr_local);
386 unlock:
387         mdt_dom_read_unlock(mo);
388         return rc;
389 }
390
391 static int mdt_preprw_write(const struct lu_env *env, struct obd_export *exp,
392                             struct mdt_device *mdt, struct mdt_object *mo,
393                             struct lu_attr *la, struct obdo *oa,
394                             int objcount, struct obd_ioobj *obj,
395                             struct niobuf_remote *rnb, int *nr_local,
396                             struct niobuf_local *lnb, char *jobid)
397 {
398         struct dt_object *dob;
399         int i, j, k, rc = 0, tot_bytes = 0;
400
401         ENTRY;
402
403         /* Process incoming grant info, set OBD_BRW_GRANTED flag and grant some
404          * space back if possible */
405         tgt_grant_prepare_write(env, exp, oa, rnb, obj->ioo_bufcnt);
406
407         mdt_dom_read_lock(mo);
408         if (!mdt_object_exists(mo)) {
409                 CDEBUG(D_ERROR, "%s: BRW to missing obj "DFID"\n",
410                        exp->exp_obd->obd_name, PFID(mdt_object_fid(mo)));
411                 GOTO(unlock, rc = -ENOENT);
412         }
413
414         dob = mdt_obj2dt(mo);
415         /* parse remote buffers to local buffers and prepare the latter */
416         *nr_local = 0;
417         for (i = 0, j = 0; i < obj->ioo_bufcnt; i++) {
418                 rc = dt_bufs_get(env, dob, rnb + i, lnb + j, 1);
419                 if (unlikely(rc < 0))
420                         GOTO(err, rc);
421                 /* correct index for local buffers to continue with */
422                 for (k = 0; k < rc; k++) {
423                         lnb[j + k].lnb_flags = rnb[i].rnb_flags;
424                         if (!(rnb[i].rnb_flags & OBD_BRW_GRANTED))
425                                 lnb[j + k].lnb_rc = -ENOSPC;
426                 }
427                 j += rc;
428                 *nr_local += rc;
429                 tot_bytes += rnb[i].rnb_len;
430         }
431
432         rc = dt_write_prep(env, dob, lnb, *nr_local);
433         if (likely(rc))
434                 GOTO(err, rc);
435
436         mdt_io_counter_incr(exp, LPROC_MDT_IO_WRITE, jobid, tot_bytes);
437         RETURN(0);
438 err:
439         dt_bufs_put(env, dob, lnb, *nr_local);
440 unlock:
441         mdt_dom_read_unlock(mo);
442         /* tgt_grant_prepare_write() was called, so we must commit */
443         tgt_grant_commit(exp, oa->o_grant_used, rc);
444         /* let's still process incoming grant information packed in the oa,
445          * but without enforcing grant since we won't proceed with the write.
446          * Just like a read request actually. */
447         tgt_grant_prepare_read(env, exp, oa);
448         return rc;
449 }
450
451 int mdt_obd_preprw(const struct lu_env *env, int cmd, struct obd_export *exp,
452                    struct obdo *oa, int objcount, struct obd_ioobj *obj,
453                    struct niobuf_remote *rnb, int *nr_local,
454                    struct niobuf_local *lnb)
455 {
456         struct tgt_session_info *tsi = tgt_ses_info(env);
457         struct mdt_thread_info *info = tsi2mdt_info(tsi);
458         struct lu_attr *la = &info->mti_attr.ma_attr;
459         struct mdt_device *mdt = mdt_dev(exp->exp_obd->obd_lu_dev);
460         struct mdt_object *mo;
461         char *jobid;
462         int rc = 0;
463
464         /* The default value PTLRPC_MAX_BRW_PAGES is set in tgt_brw_write()
465          * but for MDT it is different, correct it here. */
466         if (*nr_local > MD_MAX_BRW_PAGES)
467                 *nr_local = MD_MAX_BRW_PAGES;
468
469         jobid = tsi->tsi_jobid;
470
471         if (!oa || objcount != 1 || obj->ioo_bufcnt == 0) {
472                 CERROR("%s: bad parameters %p/%i/%i\n",
473                        exp->exp_obd->obd_name, oa, objcount, obj->ioo_bufcnt);
474                 rc = -EPROTO;
475         }
476
477         mo = mdt_object_find(env, mdt, &tsi->tsi_fid);
478         if (IS_ERR(mo))
479                 GOTO(out, rc = PTR_ERR(mo));
480
481         LASSERT(info->mti_object == NULL);
482         info->mti_object = mo;
483
484         if (cmd == OBD_BRW_WRITE) {
485                 la_from_obdo(la, oa, OBD_MD_FLGETATTR);
486                 rc = mdt_preprw_write(env, exp, mdt, mo, la, oa,
487                                       objcount, obj, rnb, nr_local, lnb,
488                                       jobid);
489         } else if (cmd == OBD_BRW_READ) {
490                 tgt_grant_prepare_read(env, exp, oa);
491                 rc = mdt_preprw_read(env, exp, mdt, mo, la,
492                                      obj->ioo_bufcnt, rnb, nr_local, lnb,
493                                      jobid);
494                 obdo_from_la(oa, la, LA_ATIME);
495         } else {
496                 CERROR("%s: wrong cmd %d received!\n",
497                        exp->exp_obd->obd_name, cmd);
498                 rc = -EPROTO;
499         }
500         if (rc) {
501                 lu_object_put(env, &mo->mot_obj);
502                 info->mti_object = NULL;
503         }
504 out:
505         RETURN(rc);
506 }
507
508 static int mdt_commitrw_read(const struct lu_env *env, struct mdt_device *mdt,
509                              struct mdt_object *mo, int objcount, int niocount,
510                              struct niobuf_local *lnb)
511 {
512         struct dt_object *dob;
513         int rc = 0;
514
515         ENTRY;
516
517         LASSERT(niocount > 0);
518
519         dob = mdt_obj2dt(mo);
520
521         dt_bufs_put(env, dob, lnb, niocount);
522
523         mdt_dom_read_unlock(mo);
524         RETURN(rc);
525 }
526
527 static int mdt_commitrw_write(const struct lu_env *env, struct obd_export *exp,
528                               struct mdt_device *mdt, struct mdt_object *mo,
529                               struct lu_attr *la, int objcount, int niocount,
530                               struct niobuf_local *lnb, unsigned long granted,
531                               int old_rc)
532 {
533         struct dt_device *dt = mdt->mdt_bottom;
534         struct dt_object *dob;
535         struct thandle *th;
536         int rc = 0;
537         int retries = 0;
538         int i;
539
540         ENTRY;
541
542         dob = mdt_obj2dt(mo);
543
544         if (old_rc)
545                 GOTO(out, rc = old_rc);
546
547         la->la_valid &= LA_ATIME | LA_MTIME | LA_CTIME;
548 retry:
549         if (!dt_object_exists(dob))
550                 GOTO(out, rc = -ENOENT);
551
552         th = dt_trans_create(env, dt);
553         if (IS_ERR(th))
554                 GOTO(out, rc = PTR_ERR(th));
555
556         for (i = 0; i < niocount; i++) {
557                 if (!(lnb[i].lnb_flags & OBD_BRW_ASYNC)) {
558                         th->th_sync = 1;
559                         break;
560                 }
561         }
562
563         if (OBD_FAIL_CHECK(OBD_FAIL_OST_DQACQ_NET))
564                 GOTO(out_stop, rc = -EINPROGRESS);
565
566         rc = dt_declare_write_commit(env, dob, lnb, niocount, th);
567         if (rc)
568                 GOTO(out_stop, rc);
569
570         if (la->la_valid) {
571                 /* update [mac]time if needed */
572                 rc = dt_declare_attr_set(env, dob, la, th);
573                 if (rc)
574                         GOTO(out_stop, rc);
575         }
576
577         rc = dt_trans_start(env, dt, th);
578         if (rc)
579                 GOTO(out_stop, rc);
580
581         dt_write_lock(env, dob, 0);
582         rc = dt_write_commit(env, dob, lnb, niocount, th);
583         if (rc)
584                 GOTO(unlock, rc);
585
586         if (la->la_valid) {
587                 rc = dt_attr_set(env, dob, la, th);
588                 if (rc)
589                         GOTO(unlock, rc);
590         }
591         /* get attr to return */
592         rc = dt_attr_get(env, dob, la);
593 unlock:
594         dt_write_unlock(env, dob);
595
596 out_stop:
597         /* Force commit to make the just-deleted blocks
598          * reusable. LU-456 */
599         if (rc == -ENOSPC)
600                 th->th_sync = 1;
601
602
603         if (rc == 0 && granted > 0) {
604                 if (tgt_grant_commit_cb_add(th, exp, granted) == 0)
605                         granted = 0;
606         }
607
608         th->th_result = rc;
609         dt_trans_stop(env, dt, th);
610         if (rc == -ENOSPC && retries++ < 3) {
611                 CDEBUG(D_INODE, "retry after force commit, retries:%d\n",
612                        retries);
613                 goto retry;
614         }
615
616 out:
617         dt_bufs_put(env, dob, lnb, niocount);
618         mdt_dom_read_unlock(mo);
619         if (granted > 0)
620                 tgt_grant_commit(exp, granted, old_rc);
621         RETURN(rc);
622 }
623
624 void mdt_dom_obj_lvb_update(const struct lu_env *env, struct mdt_object *mo,
625                             bool increase_only)
626 {
627         struct mdt_device *mdt = mdt_dev(mo->mot_obj.lo_dev);
628         struct ldlm_res_id resid;
629         struct ldlm_resource *res;
630
631         fid_build_reg_res_name(mdt_object_fid(mo), &resid);
632         res = ldlm_resource_get(mdt->mdt_namespace, NULL, &resid,
633                                 LDLM_IBITS, 1);
634         if (IS_ERR(res))
635                 return;
636
637         /* Update lvbo data if exists. */
638         if (mdt_dom_lvb_is_valid(res))
639                 mdt_dom_disk_lvbo_update(env, mo, res, increase_only);
640         ldlm_resource_putref(res);
641 }
642
643 int mdt_obd_commitrw(const struct lu_env *env, int cmd, struct obd_export *exp,
644                      struct obdo *oa, int objcount, struct obd_ioobj *obj,
645                      struct niobuf_remote *rnb, int npages,
646                      struct niobuf_local *lnb, int old_rc)
647 {
648         struct mdt_thread_info *info = mdt_th_info(env);
649         struct mdt_device *mdt = mdt_dev(exp->exp_obd->obd_lu_dev);
650         struct mdt_object *mo = info->mti_object;
651         struct lu_attr *la = &info->mti_attr.ma_attr;
652         __u64 valid;
653         int rc = 0;
654
655         if (npages == 0) {
656                 CERROR("%s: no pages to commit\n",
657                        exp->exp_obd->obd_name);
658                 rc = -EPROTO;
659         }
660
661         LASSERT(mo);
662
663         if (cmd == OBD_BRW_WRITE) {
664                 /* Don't update timestamps if this write is older than a
665                  * setattr which modifies the timestamps. b=10150 */
666
667                 /* XXX when we start having persistent reservations this needs
668                  * to be changed to ofd_fmd_get() to create the fmd if it
669                  * doesn't already exist so we can store the reservation handle
670                  * there. */
671                 valid = OBD_MD_FLUID | OBD_MD_FLGID;
672                 valid |= OBD_MD_FLATIME | OBD_MD_FLMTIME | OBD_MD_FLCTIME;
673
674                 la_from_obdo(la, oa, valid);
675
676                 rc = mdt_commitrw_write(env, exp, mdt, mo, la, objcount,
677                                         npages, lnb, oa->o_grant_used, old_rc);
678                 if (rc == 0)
679                         obdo_from_la(oa, la, VALID_FLAGS | LA_GID | LA_UID);
680                 else
681                         obdo_from_la(oa, la, LA_GID | LA_UID);
682
683                 mdt_dom_obj_lvb_update(env, mo, false);
684                 /* don't report overquota flag if we failed before reaching
685                  * commit */
686                 if (old_rc == 0 && (rc == 0 || rc == -EDQUOT)) {
687                         /* return the overquota flags to client */
688                         if (lnb[0].lnb_flags & OBD_BRW_OVER_USRQUOTA) {
689                                 if (oa->o_valid & OBD_MD_FLFLAGS)
690                                         oa->o_flags |= OBD_FL_NO_USRQUOTA;
691                                 else
692                                         oa->o_flags = OBD_FL_NO_USRQUOTA;
693                         }
694
695                         if (lnb[0].lnb_flags & OBD_BRW_OVER_GRPQUOTA) {
696                                 if (oa->o_valid & OBD_MD_FLFLAGS)
697                                         oa->o_flags |= OBD_FL_NO_GRPQUOTA;
698                                 else
699                                         oa->o_flags = OBD_FL_NO_GRPQUOTA;
700                         }
701
702                         oa->o_valid |= OBD_MD_FLFLAGS | OBD_MD_FLUSRQUOTA |
703                                        OBD_MD_FLGRPQUOTA;
704                 }
705         } else if (cmd == OBD_BRW_READ) {
706                 /* If oa != NULL then mdt_preprw_read updated the inode
707                  * atime and we should update the lvb so that other glimpses
708                  * will also get the updated value. bug 5972 */
709                 if (oa)
710                         mdt_dom_obj_lvb_update(env, mo, true);
711                 rc = mdt_commitrw_read(env, mdt, mo, objcount, npages, lnb);
712                 if (old_rc)
713                         rc = old_rc;
714         } else {
715                 rc = -EPROTO;
716         }
717         /* this put is pair to object_get in ofd_preprw_write */
718         mdt_thread_info_fini(info);
719         RETURN(rc);
720 }
721
722 int mdt_object_punch(const struct lu_env *env, struct dt_device *dt,
723                      struct dt_object *dob, __u64 start, __u64 end,
724                      struct lu_attr *la)
725 {
726         struct thandle *th;
727         int rc;
728
729         ENTRY;
730
731         /* we support truncate, not punch yet */
732         LASSERT(end == OBD_OBJECT_EOF);
733
734         if (!dt_object_exists(dob))
735                 RETURN(-ENOENT);
736
737         th = dt_trans_create(env, dt);
738         if (IS_ERR(th))
739                 RETURN(PTR_ERR(th));
740
741         rc = dt_declare_attr_set(env, dob, la, th);
742         if (rc)
743                 GOTO(stop, rc);
744
745         rc = dt_declare_punch(env, dob, start, OBD_OBJECT_EOF, th);
746         if (rc)
747                 GOTO(stop, rc);
748
749         tgt_vbr_obj_set(env, dob);
750         rc = dt_trans_start(env, dt, th);
751         if (rc)
752                 GOTO(stop, rc);
753
754         dt_write_lock(env, dob, 0);
755         rc = dt_punch(env, dob, start, OBD_OBJECT_EOF, th);
756         if (rc)
757                 GOTO(unlock, rc);
758         rc = dt_attr_set(env, dob, la, th);
759         if (rc)
760                 GOTO(unlock, rc);
761 unlock:
762         dt_write_unlock(env, dob);
763 stop:
764         th->th_result = rc;
765         dt_trans_stop(env, dt, th);
766         RETURN(rc);
767 }
768
769 int mdt_punch_hdl(struct tgt_session_info *tsi)
770 {
771         const struct obdo *oa = &tsi->tsi_ost_body->oa;
772         struct ost_body *repbody;
773         struct mdt_thread_info *info;
774         struct lu_attr *la;
775         struct ldlm_namespace *ns = tsi->tsi_tgt->lut_obd->obd_namespace;
776         struct obd_export *exp = tsi->tsi_exp;
777         struct mdt_device *mdt = mdt_dev(exp->exp_obd->obd_lu_dev);
778         struct mdt_object *mo;
779         struct dt_object *dob;
780         __u64 flags = 0;
781         struct lustre_handle lh = { 0, };
782         __u64 start, end;
783         int rc;
784         bool srvlock;
785
786         ENTRY;
787
788         /* check that we do support OBD_CONNECT_TRUNCLOCK. */
789         CLASSERT(OST_CONNECT_SUPPORTED & OBD_CONNECT_TRUNCLOCK);
790
791         if ((oa->o_valid & (OBD_MD_FLSIZE | OBD_MD_FLBLOCKS)) !=
792             (OBD_MD_FLSIZE | OBD_MD_FLBLOCKS))
793                 RETURN(err_serious(-EPROTO));
794
795         repbody = req_capsule_server_get(tsi->tsi_pill, &RMF_OST_BODY);
796         if (repbody == NULL)
797                 RETURN(err_serious(-ENOMEM));
798
799         /* punch start,end are passed in o_size,o_blocks throught wire */
800         start = oa->o_size;
801         end = oa->o_blocks;
802
803         if (end != OBD_OBJECT_EOF) /* Only truncate is supported */
804                 RETURN(-EPROTO);
805
806         info = tsi2mdt_info(tsi);
807         la = &info->mti_attr.ma_attr;
808         /* standard truncate optimization: if file body is completely
809          * destroyed, don't send data back to the server. */
810         if (start == 0)
811                 flags |= LDLM_FL_AST_DISCARD_DATA;
812
813         repbody->oa.o_oi = oa->o_oi;
814         repbody->oa.o_valid = OBD_MD_FLID;
815
816         srvlock = (exp_connect_flags(exp) & OBD_CONNECT_SRVLOCK) &&
817                   oa->o_valid & OBD_MD_FLFLAGS &&
818                   oa->o_flags & OBD_FL_SRVLOCK;
819
820         if (srvlock) {
821                 rc = tgt_mdt_data_lock(ns, &tsi->tsi_resid, &lh, LCK_PW,
822                                        &flags);
823                 if (rc != 0)
824                         GOTO(out, rc);
825         }
826
827         CDEBUG(D_INODE, "calling punch for object "DFID", valid = %#llx"
828                ", start = %lld, end = %lld\n", PFID(&tsi->tsi_fid),
829                oa->o_valid, start, end);
830
831         mo = mdt_object_find(tsi->tsi_env, mdt, &tsi->tsi_fid);
832         if (IS_ERR(mo))
833                 GOTO(out_unlock, rc = PTR_ERR(mo));
834
835         if (!mdt_object_exists(mo))
836                 GOTO(out_put, rc = -ENOENT);
837
838         /* Shouldn't happen on dirs */
839         if (S_ISDIR(lu_object_attr(&mo->mot_obj))) {
840                 rc = -EPERM;
841                 CERROR("%s: Truncate on dir "DFID": rc = %d\n",
842                        exp->exp_obd->obd_name, PFID(&tsi->tsi_fid), rc);
843                 GOTO(out_put, rc);
844         }
845
846         mdt_dom_write_lock(mo);
847         dob = mdt_obj2dt(mo);
848
849         la_from_obdo(la, oa, OBD_MD_FLMTIME | OBD_MD_FLATIME | OBD_MD_FLCTIME);
850         la->la_size = start;
851         la->la_valid |= LA_SIZE;
852
853         rc = mdt_object_punch(tsi->tsi_env, mdt->mdt_bottom, dob,
854                               start, end, la);
855         mdt_dom_write_unlock(mo);
856         if (rc)
857                 GOTO(out_put, rc);
858
859         mdt_dom_obj_lvb_update(tsi->tsi_env, mo, false);
860         mdt_io_counter_incr(tsi->tsi_exp, LPROC_MDT_IO_PUNCH,
861                             tsi->tsi_jobid, 1);
862         EXIT;
863 out_put:
864         lu_object_put(tsi->tsi_env, &mo->mot_obj);
865 out_unlock:
866         if (srvlock)
867                 tgt_extent_unlock(&lh, LCK_PW);
868 out:
869         mdt_thread_info_fini(info);
870         return rc;
871 }
872
873 /**
874  * MDT glimpse for Data-on-MDT
875  *
876  * If there is write lock on client then function issues glimpse_ast to get
877  * an actual size from that client.
878  *
879  */
880 int mdt_do_glimpse(const struct lu_env *env, struct ldlm_namespace *ns,
881                    struct ldlm_resource *res)
882 {
883         union ldlm_policy_data policy;
884         struct lustre_handle lockh;
885         enum ldlm_mode mode;
886         struct ldlm_lock *lock;
887         struct ldlm_glimpse_work *gl_work;
888         struct list_head gl_list;
889         int rc;
890
891         ENTRY;
892
893         /* There can be only one write lock covering data, try to match it. */
894         policy.l_inodebits.bits = MDS_INODELOCK_DOM;
895         mode = ldlm_lock_match(ns, LDLM_FL_TEST_LOCK,
896                                &res->lr_name, LDLM_IBITS, &policy,
897                                LCK_PW, &lockh, 0);
898
899         /* There is no PW lock on this object; finished. */
900         if (mode == 0)
901                 RETURN(0);
902
903         lock = ldlm_handle2lock(&lockh);
904         if (lock == NULL)
905                 RETURN(0);
906
907         /*
908          * This check is for lock taken in mdt_reint_unlink() that does
909          * not have l_glimpse_ast set. So the logic is: if there is a lock
910          * with no l_glimpse_ast set, this object is being destroyed already.
911          * Hence, if you are grabbing DLM locks on the server, always set
912          * non-NULL glimpse_ast (e.g., ldlm_request.c::ldlm_glimpse_ast()).
913          */
914         if (lock->l_glimpse_ast == NULL) {
915                 LDLM_DEBUG(lock, "no l_glimpse_ast");
916                 GOTO(out, rc = -ENOENT);
917         }
918
919         OBD_SLAB_ALLOC_PTR_GFP(gl_work, ldlm_glimpse_work_kmem, GFP_ATOMIC);
920         if (!gl_work)
921                 GOTO(out, rc = -ENOMEM);
922
923         /* Populate the gl_work structure.
924          * Grab additional reference on the lock which will be released in
925          * ldlm_work_gl_ast_lock() */
926         gl_work->gl_lock = LDLM_LOCK_GET(lock);
927         /* The glimpse callback is sent to one single IO lock. As a result,
928          * the gl_work list is just composed of one element */
929         INIT_LIST_HEAD(&gl_list);
930         list_add_tail(&gl_work->gl_list, &gl_list);
931         /* There is actually no need for a glimpse descriptor when glimpsing
932          * IO locks */
933         gl_work->gl_desc = NULL;
934         /* the ldlm_glimpse_work structure is allocated on the stack */
935         gl_work->gl_flags = LDLM_GL_WORK_SLAB_ALLOCATED;
936
937         ldlm_glimpse_locks(res, &gl_list); /* this will update the LVB */
938
939         /* If the list is not empty, we failed to glimpse a lock and
940          * must clean it up. Usually due to a race with unlink.*/
941         if (!list_empty(&gl_list)) {
942                 LDLM_LOCK_RELEASE(lock);
943                 OBD_SLAB_FREE_PTR(gl_work, ldlm_glimpse_work_kmem);
944         }
945         rc = 0;
946         EXIT;
947 out:
948         LDLM_LOCK_PUT(lock);
949         return rc;
950 }
951
952 static void mdt_lvb2body(struct ldlm_resource *res, struct mdt_body *mb)
953 {
954         struct ost_lvb *res_lvb;
955
956         lock_res(res);
957         res_lvb = res->lr_lvb_data;
958         mb->mbo_dom_size = res_lvb->lvb_size;
959         mb->mbo_dom_blocks = res_lvb->lvb_blocks;
960         mb->mbo_mtime = res_lvb->lvb_mtime;
961         mb->mbo_ctime = res_lvb->lvb_ctime;
962         mb->mbo_atime = res_lvb->lvb_atime;
963
964         CDEBUG(D_DLMTRACE, "size %llu\n", res_lvb->lvb_size);
965
966         mb->mbo_valid |= OBD_MD_FLATIME | OBD_MD_FLCTIME | OBD_MD_FLMTIME |
967                          OBD_MD_DOM_SIZE;
968         unlock_res(res);
969 }
970
971 /**
972  * MDT glimpse for Data-on-MDT
973  *
974  * This function is called when MDT get attributes for the DoM object.
975  * If there is write lock on client then function issues glimpse_ast to get
976  * an actual size from that client.
977  */
978 int mdt_dom_object_size(const struct lu_env *env, struct mdt_device *mdt,
979                         const struct lu_fid *fid, struct mdt_body *mb,
980                         bool dom_lock)
981 {
982         struct ldlm_res_id resid;
983         struct ldlm_resource *res;
984         int rc = 0;
985
986         ENTRY;
987
988         fid_build_reg_res_name(fid, &resid);
989         res = ldlm_resource_get(mdt->mdt_namespace, NULL, &resid,
990                                 LDLM_IBITS, 1);
991         if (IS_ERR(res))
992                 RETURN(-ENOENT);
993
994         /* Update lvbo data if DoM lock returned or if LVB is not yet valid. */
995         if (dom_lock || !mdt_dom_lvb_is_valid(res))
996                 mdt_dom_lvbo_update(res, NULL, NULL, false);
997
998         mdt_lvb2body(res, mb);
999         ldlm_resource_putref(res);
1000         RETURN(rc);
1001 }
1002
1003 /**
1004  * MDT DoM lock intent policy (glimpse)
1005  *
1006  * Intent policy is called when lock has an intent, for DoM file that
1007  * means glimpse lock and policy fills Lock Value Block (LVB).
1008  *
1009  * If already granted lock is found it will be placed in \a lockp and
1010  * returned back to caller function.
1011  *
1012  * \param[in] tsi        session info
1013  * \param[in,out] lockp  pointer to the lock
1014  * \param[in] flags      LDLM flags
1015  *
1016  * \retval              ELDLM_LOCK_REPLACED if already granted lock was found
1017  *                      and placed in \a lockp
1018  * \retval              ELDLM_LOCK_ABORTED in other cases except error
1019  * \retval              negative value on error
1020  */
1021 int mdt_glimpse_enqueue(struct mdt_thread_info *mti, struct ldlm_namespace *ns,
1022                         struct ldlm_lock **lockp, __u64 flags)
1023 {
1024         struct ldlm_lock *lock = *lockp;
1025         struct ldlm_resource *res = lock->l_resource;
1026         ldlm_processing_policy policy;
1027         struct ldlm_reply *rep;
1028         struct mdt_body *mbo;
1029         int rc;
1030
1031         ENTRY;
1032
1033         policy = ldlm_get_processing_policy(res);
1034         LASSERT(policy != NULL);
1035
1036         req_capsule_set_size(mti->mti_pill, &RMF_MDT_MD, RCL_SERVER, 0);
1037         req_capsule_set_size(mti->mti_pill, &RMF_ACL, RCL_SERVER, 0);
1038         rc = req_capsule_server_pack(mti->mti_pill);
1039         if (rc)
1040                 RETURN(err_serious(rc));
1041
1042         rep = req_capsule_server_get(mti->mti_pill, &RMF_DLM_REP);
1043         if (rep == NULL)
1044                 RETURN(-EPROTO);
1045
1046         mbo = req_capsule_server_get(mti->mti_pill, &RMF_MDT_BODY);
1047         if (mbo == NULL)
1048                 RETURN(-EPROTO);
1049
1050         lock_res(res);
1051         /* Check if this is a resend case (MSG_RESENT is set on RPC) and a
1052          * lock was found by ldlm_handle_enqueue(); if so no need to grant
1053          * it again. */
1054         if (flags & LDLM_FL_RESENT) {
1055                 rc = LDLM_ITER_CONTINUE;
1056         } else {
1057                 __u64 tmpflags = 0;
1058                 enum ldlm_error err;
1059
1060                 rc = policy(lock, &tmpflags, LDLM_PROCESS_RESCAN, &err, NULL);
1061                 check_res_locked(res);
1062         }
1063         unlock_res(res);
1064
1065         /* The lock met with no resistance; we're finished. */
1066         if (rc == LDLM_ITER_CONTINUE) {
1067                 GOTO(fill_mbo, rc = ELDLM_LOCK_REPLACED);
1068         } else if (flags & LDLM_FL_BLOCK_NOWAIT) {
1069                 /* LDLM_FL_BLOCK_NOWAIT means it is for AGL. Do not send glimpse
1070                  * callback for glimpse size. The real size user will trigger
1071                  * the glimpse callback when necessary. */
1072                 GOTO(fill_mbo, rc = ELDLM_LOCK_ABORTED);
1073         }
1074
1075         rc = mdt_do_glimpse(mti->mti_env, ns, res);
1076         if (rc == -ENOENT) {
1077                 /* We are racing with unlink(); just return -ENOENT */
1078                 rep->lock_policy_res2 = ptlrpc_status_hton(-ENOENT);
1079                 rc = 0;
1080         } else if (rc == -EINVAL) {
1081                 /* this is possible is client lock has been cancelled but
1082                  * still exists on server. If that lock was found on server
1083                  * as only conflicting lock then the client has already
1084                  * size authority and glimpse is not needed. */
1085                 CDEBUG(D_DLMTRACE, "Glimpse from the client owning lock\n");
1086                 rc = 0;
1087         } else if (rc < 0) {
1088                 RETURN(rc);
1089         }
1090         rc = ELDLM_LOCK_ABORTED;
1091 fill_mbo:
1092         /* LVB can be without valid data in case of DOM */
1093         if (!mdt_dom_lvb_is_valid(res))
1094                 mdt_dom_lvbo_update(res, lock, NULL, false);
1095         mdt_lvb2body(res, mbo);
1096         RETURN(rc);
1097 }
1098
1099 int mdt_brw_enqueue(struct mdt_thread_info *mti, struct ldlm_namespace *ns,
1100                     struct ldlm_lock **lockp, __u64 flags)
1101 {
1102         struct tgt_session_info *tsi = tgt_ses_info(mti->mti_env);
1103         struct lu_fid *fid = &tsi->tsi_fid;
1104         struct ldlm_lock *lock = *lockp;
1105         struct ldlm_resource *res = lock->l_resource;
1106         struct ldlm_reply *rep;
1107         struct mdt_body *mbo;
1108         struct mdt_lock_handle *lhc = &mti->mti_lh[MDT_LH_RMT];
1109         struct mdt_object *mo;
1110         int rc = 0;
1111
1112         ENTRY;
1113
1114         /* Get lock from request for possible resent case. */
1115         mdt_intent_fixup_resent(mti, *lockp, lhc, flags);
1116         req_capsule_set_size(mti->mti_pill, &RMF_MDT_MD, RCL_SERVER, 0);
1117         req_capsule_set_size(mti->mti_pill, &RMF_ACL, RCL_SERVER, 0);
1118         rc = req_capsule_server_pack(mti->mti_pill);
1119         if (rc)
1120                 RETURN(err_serious(rc));
1121
1122         rep = req_capsule_server_get(mti->mti_pill, &RMF_DLM_REP);
1123         if (rep == NULL)
1124                 RETURN(-EPROTO);
1125
1126         mbo = req_capsule_server_get(mti->mti_pill, &RMF_MDT_BODY);
1127         if (mbo == NULL)
1128                 RETURN(-EPROTO);
1129
1130         fid_extract_from_res_name(fid, &res->lr_name);
1131         mo = mdt_object_find(mti->mti_env, mti->mti_mdt, fid);
1132         if (unlikely(IS_ERR(mo)))
1133                 RETURN(PTR_ERR(mo));
1134
1135         if (!mdt_object_exists(mo))
1136                 GOTO(out, rc = -ENOENT);
1137
1138         if (mdt_object_remote(mo))
1139                 GOTO(out, rc = -EPROTO);
1140
1141         /* resent case */
1142         if (!lustre_handle_is_used(&lhc->mlh_reg_lh)) {
1143                 mdt_lock_handle_init(lhc);
1144                 mdt_lock_reg_init(lhc, (*lockp)->l_req_mode);
1145                 /* This will block MDT thread but it should be fine until
1146                  * client caches small amount of data for DoM, which should be
1147                  * smaller than one BRW RPC and should be able to be
1148                  * piggybacked by lock cancel RPC.
1149                  * If the client could hold the lock too long, this code can be
1150                  * revised to call mdt_object_lock_try(). And if fails, it will
1151                  * return ELDLM_OK here and fall back into normal lock enqueue
1152                  * process.
1153                  */
1154                 rc = mdt_object_lock(mti, mo, lhc, MDS_INODELOCK_DOM);
1155                 if (rc)
1156                         GOTO(out, rc);
1157         }
1158
1159         if (!mdt_dom_lvb_is_valid(res)) {
1160                 rc = mdt_dom_lvb_alloc(res);
1161                 if (rc)
1162                         GOTO(out_fail, rc);
1163                 mdt_dom_disk_lvbo_update(mti->mti_env, mo, res, false);
1164         }
1165         mdt_lvb2body(res, mbo);
1166 out_fail:
1167         rep->lock_policy_res2 = clear_serious(rc);
1168         if (rep->lock_policy_res2) {
1169                 lhc->mlh_reg_lh.cookie = 0ull;
1170                 GOTO(out, rc = ELDLM_LOCK_ABORTED);
1171         }
1172
1173         rc = mdt_intent_lock_replace(mti, lockp, lhc, flags, rc);
1174 out:
1175         mdt_object_put(mti->mti_env, mo);
1176         RETURN(rc);
1177 }
1178
1179 void mdt_dom_discard_data(struct mdt_thread_info *info,
1180                           const struct lu_fid *fid)
1181 {
1182         struct mdt_device *mdt = info->mti_mdt;
1183         union ldlm_policy_data *policy = &info->mti_policy;
1184         struct ldlm_res_id *res_id = &info->mti_res_id;
1185         struct lustre_handle dom_lh;
1186         __u64 flags = LDLM_FL_AST_DISCARD_DATA;
1187         int rc = 0;
1188
1189         policy->l_inodebits.bits = MDS_INODELOCK_DOM;
1190         policy->l_inodebits.try_bits = 0;
1191         fid_build_reg_res_name(fid, res_id);
1192
1193         /* Tell the clients that the object is gone now and that they should
1194          * throw away any cached pages. */
1195         rc = ldlm_cli_enqueue_local(mdt->mdt_namespace, res_id, LDLM_IBITS,
1196                                     policy, LCK_PW, &flags, ldlm_blocking_ast,
1197                                     ldlm_completion_ast, NULL, NULL, 0,
1198                                     LVB_T_NONE, NULL, &dom_lh);
1199
1200         /* We only care about the side-effects, just drop the lock. */
1201         if (rc == ELDLM_OK)
1202                 ldlm_lock_decref(&dom_lh, LCK_PW);
1203 }
1204
1205 /* check if client has already DoM lock for given resource */
1206 bool mdt_dom_client_has_lock(struct mdt_thread_info *info,
1207                              const struct lu_fid *fid)
1208 {
1209         struct mdt_device *mdt = info->mti_mdt;
1210         union ldlm_policy_data *policy = &info->mti_policy;
1211         struct ldlm_res_id *res_id = &info->mti_res_id;
1212         struct lustre_handle lockh;
1213         enum ldlm_mode mode;
1214         struct ldlm_lock *lock;
1215         bool rc;
1216
1217         policy->l_inodebits.bits = MDS_INODELOCK_DOM;
1218         fid_build_reg_res_name(fid, res_id);
1219
1220         mode = ldlm_lock_match(mdt->mdt_namespace, LDLM_FL_BLOCK_GRANTED |
1221                                LDLM_FL_TEST_LOCK, res_id, LDLM_IBITS, policy,
1222                                LCK_PW, &lockh, 0);
1223
1224         /* There is no other PW lock on this object; finished. */
1225         if (mode == 0)
1226                 return false;
1227
1228         lock = ldlm_handle2lock(&lockh);
1229         if (lock == 0)
1230                 return false;
1231
1232         /* check if lock from the same client */
1233         rc = (lock->l_export->exp_handle.h_cookie ==
1234               info->mti_exp->exp_handle.h_cookie);
1235         LDLM_LOCK_PUT(lock);
1236         return rc;
1237 }
1238