Whamcloud - gitweb
b=19906
[fs/lustre-release.git] / lustre / lov / lov_lock.c
1 /* -*- mode: c; c-basic-offset: 8; indent-tabs-mode: nil; -*-
2  * vim:expandtab:shiftwidth=8:tabstop=8:
3  *
4  * GPL HEADER START
5  *
6  * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.
7  *
8  * This program is free software; you can redistribute it and/or modify
9  * it under the terms of the GNU General Public License version 2 only,
10  * as published by the Free Software Foundation.
11  *
12  * This program is distributed in the hope that it will be useful, but
13  * WITHOUT ANY WARRANTY; without even the implied warranty of
14  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
15  * General Public License version 2 for more details (a copy is included
16  * in the LICENSE file that accompanied this code).
17  *
18  * You should have received a copy of the GNU General Public License
19  * version 2 along with this program; If not, see
20  * http://www.sun.com/software/products/lustre/docs/GPLv2.pdf
21  *
22  * Please contact Sun Microsystems, Inc., 4150 Network Circle, Santa Clara,
23  * CA 95054 USA or visit www.sun.com if you need additional information or
24  * have any questions.
25  *
26  * GPL HEADER END
27  */
28 /*
29  * Copyright 2008 Sun Microsystems, Inc.  All rights reserved.
30  * Use is subject to license terms.
31  */
32 /*
33  * This file is part of Lustre, http://www.lustre.org/
34  * Lustre is a trademark of Sun Microsystems, Inc.
35  *
36  * Implementation of cl_lock for LOV layer.
37  *
38  *   Author: Nikita Danilov <nikita.danilov@sun.com>
39  */
40
41 #define DEBUG_SUBSYSTEM S_LOV
42
43 #include "lov_cl_internal.h"
44
45 /** \addtogroup lov
46  *  @{
47  */
48
49 static struct cl_lock_closure *lov_closure_get(const struct lu_env *env,
50                                                struct cl_lock *parent);
51
52 static int lov_lock_unuse(const struct lu_env *env,
53                           const struct cl_lock_slice *slice);
54 /*****************************************************************************
55  *
56  * Lov lock operations.
57  *
58  */
59
60 static struct lov_sublock_env *lov_sublock_env_get(const struct lu_env *env,
61                                                    struct cl_lock *parent,
62                                                    struct lov_lock_sub *lls)
63 {
64         struct lov_sublock_env *subenv;
65         struct lov_io          *lio    = lov_env_io(env);
66         struct cl_io           *io     = lio->lis_cl.cis_io;
67         struct lov_io_sub      *sub;
68
69         subenv = &lov_env_session(env)->ls_subenv;
70
71         /*
72          * FIXME: We tend to use the subio's env & io to call the sublock
73          * lock operations because osc lock sometimes stores some control
74          * variables in thread's IO infomation(Now only lockless information).
75          * However, if the lock's host(object) is different from the object
76          * for current IO, we have no way to get the subenv and subio because
77          * they are not initialized at all. As a temp fix, in this case,
78          * we still borrow the parent's env to call sublock operations.
79          */
80         if (!io || !cl_object_same(io->ci_obj, parent->cll_descr.cld_obj)) {
81                 subenv->lse_env = env;
82                 subenv->lse_io  = io;
83                 subenv->lse_sub = NULL;
84         } else {
85                 sub = lov_sub_get(env, lio, lls->sub_stripe);
86                 if (!IS_ERR(sub)) {
87                         subenv->lse_env = sub->sub_env;
88                         subenv->lse_io  = sub->sub_io;
89                         subenv->lse_sub = sub;
90                 } else {
91                         subenv = (void*)sub;
92                 }
93         }
94         return subenv;
95 }
96
97 static void lov_sublock_env_put(struct lov_sublock_env *subenv)
98 {
99         if (subenv && subenv->lse_sub)
100                 lov_sub_put(subenv->lse_sub);
101 }
102
103 static void lov_sublock_adopt(const struct lu_env *env, struct lov_lock *lck,
104                               struct cl_lock *sublock, int idx,
105                               struct lov_lock_link *link)
106 {
107         struct lovsub_lock *lsl;
108         struct cl_lock     *parent = lck->lls_cl.cls_lock;
109         int                 rc;
110
111         LASSERT(cl_lock_is_mutexed(parent));
112         LASSERT(cl_lock_is_mutexed(sublock));
113         ENTRY;
114
115         lsl = cl2sub_lock(sublock);
116         /*
117          * check that sub-lock doesn't have lock link to this top-lock.
118          */
119         LASSERT(lov_lock_link_find(env, lck, lsl) == NULL);
120         LASSERT(idx < lck->lls_nr);
121
122         lck->lls_sub[idx].sub_lock = lsl;
123         lck->lls_nr_filled++;
124         LASSERT(lck->lls_nr_filled <= lck->lls_nr);
125         list_add_tail(&link->lll_list, &lsl->lss_parents);
126         link->lll_idx = idx;
127         link->lll_super = lck;
128         cl_lock_get(parent);
129         lu_ref_add(&parent->cll_reference, "lov-child", sublock);
130         lck->lls_sub[idx].sub_flags |= LSF_HELD;
131         cl_lock_user_add(env, sublock);
132
133         rc = lov_sublock_modify(env, lck, lsl, &sublock->cll_descr, idx);
134         LASSERT(rc == 0); /* there is no way this can fail, currently */
135         EXIT;
136 }
137
138 static struct cl_lock *lov_sublock_alloc(const struct lu_env *env,
139                                          const struct cl_io *io,
140                                          struct lov_lock *lck,
141                                          int idx, struct lov_lock_link **out)
142 {
143         struct cl_lock       *sublock;
144         struct cl_lock       *parent;
145         struct lov_lock_link *link;
146
147         LASSERT(idx < lck->lls_nr);
148         ENTRY;
149
150         OBD_SLAB_ALLOC_PTR_GFP(link, lov_lock_link_kmem, CFS_ALLOC_IO);
151         if (link != NULL) {
152                 struct lov_sublock_env *subenv;
153                 struct lov_lock_sub  *lls;
154                 struct cl_lock_descr *descr;
155
156                 parent = lck->lls_cl.cls_lock;
157                 lls    = &lck->lls_sub[idx];
158                 descr  = &lls->sub_descr;
159
160                 subenv = lov_sublock_env_get(env, parent, lls);
161                 if (!IS_ERR(subenv)) {
162                         /* CAVEAT: Don't try to add a field in lov_lock_sub
163                          * to remember the subio. This is because lock is able
164                          * to be cached, but this is not true for IO. This
165                          * further means a sublock might be referenced in
166                          * different io context. -jay */
167
168                         sublock = cl_lock_hold(subenv->lse_env, subenv->lse_io,
169                                                descr, "lov-parent", parent);
170                         lov_sublock_env_put(subenv);
171                 } else {
172                         /* error occurs. */
173                         sublock = (void*)subenv;
174                 }
175
176                 if (!IS_ERR(sublock))
177                         *out = link;
178                 else
179                         OBD_SLAB_FREE_PTR(link, lov_lock_link_kmem);
180         } else
181                 sublock = ERR_PTR(-ENOMEM);
182         RETURN(sublock);
183 }
184
185 static void lov_sublock_unlock(const struct lu_env *env,
186                                struct lovsub_lock *lsl,
187                                struct cl_lock_closure *closure,
188                                struct lov_sublock_env *subenv)
189 {
190         ENTRY;
191         lov_sublock_env_put(subenv);
192         lsl->lss_active = NULL;
193         cl_lock_disclosure(env, closure);
194         EXIT;
195 }
196
197 static int lov_sublock_lock(const struct lu_env *env,
198                             struct lov_lock *lck,
199                             struct lov_lock_sub *lls,
200                             struct cl_lock_closure *closure,
201                             struct lov_sublock_env **lsep)
202 {
203         struct lovsub_lock *sublock;
204         struct cl_lock     *child;
205         int                 result = 0;
206         ENTRY;
207
208         LASSERT(list_empty(&closure->clc_list));
209
210         sublock = lls->sub_lock;
211         child = sublock->lss_cl.cls_lock;
212         result = cl_lock_closure_build(env, child, closure);
213         if (result == 0) {
214                 struct cl_lock *parent = closure->clc_origin;
215
216                 LASSERT(cl_lock_is_mutexed(child));
217                 sublock->lss_active = parent;
218
219                 if (unlikely((child->cll_state == CLS_FREEING) ||
220                              (child->cll_flags & CLF_CANCELLED))) {
221                         struct lov_lock_link *link;
222                         /*
223                          * we could race with lock deletion which temporarily
224                          * put the lock in freeing state, bug 19080.
225                          */
226                         LASSERT(!(lls->sub_flags & LSF_HELD));
227
228                         link = lov_lock_link_find(env, lck, sublock);
229                         LASSERT(link != NULL);
230                         lov_lock_unlink(env, link, sublock);
231                         lov_sublock_unlock(env, sublock, closure, NULL);
232                         lck->lls_cancel_race = 1;
233                         result = CLO_REPEAT;
234                 } else if (lsep) {
235                         struct lov_sublock_env *subenv;
236                         subenv = lov_sublock_env_get(env, parent, lls);
237                         if (IS_ERR(subenv)) {
238                                 lov_sublock_unlock(env, sublock,
239                                                    closure, NULL);
240                                 result = PTR_ERR(subenv);
241                         } else {
242                                 *lsep = subenv;
243                         }
244                 }
245         }
246         RETURN(result);
247 }
248
249 /**
250  * Updates the result of a top-lock operation from a result of sub-lock
251  * sub-operations. Top-operations like lov_lock_{enqueue,use,unuse}() iterate
252  * over sub-locks and lov_subresult() is used to calculate return value of a
253  * top-operation. To this end, possible return values of sub-operations are
254  * ordered as
255  *
256  *     - 0                  success
257  *     - CLO_WAIT           wait for event
258  *     - CLO_REPEAT         repeat top-operation
259  *     - -ne                fundamental error
260  *
261  * Top-level return code can only go down through this list. CLO_REPEAT
262  * overwrites CLO_WAIT, because lock mutex was released and sleeping condition
263  * has to be rechecked by the upper layer.
264  */
265 static int lov_subresult(int result, int rc)
266 {
267         int result_rank;
268         int rc_rank;
269
270         LASSERT(result <= 0 || result == CLO_REPEAT || result == CLO_WAIT);
271         LASSERT(rc <= 0 || rc == CLO_REPEAT || rc == CLO_WAIT);
272         CLASSERT(CLO_WAIT < CLO_REPEAT);
273
274         ENTRY;
275
276         /* calculate ranks in the ordering above */
277         result_rank = result < 0 ? 1 + CLO_REPEAT : result;
278         rc_rank = rc < 0 ? 1 + CLO_REPEAT : rc;
279
280         if (result_rank < rc_rank)
281                 result = rc;
282         RETURN(result);
283 }
284
285 /**
286  * Creates sub-locks for a given lov_lock for the first time.
287  *
288  * Goes through all sub-objects of top-object, and creates sub-locks on every
289  * sub-object intersecting with top-lock extent. This is complicated by the
290  * fact that top-lock (that is being created) can be accessed concurrently
291  * through already created sub-locks (possibly shared with other top-locks).
292  */
293 static int lov_lock_sub_init(const struct lu_env *env,
294                              struct lov_lock *lck, const struct cl_io *io)
295 {
296         int result = 0;
297         int i;
298         int nr;
299         obd_off start;
300         obd_off end;
301         obd_off file_start;
302         obd_off file_end;
303
304         struct lov_object       *loo    = cl2lov(lck->lls_cl.cls_obj);
305         struct lov_layout_raid0 *r0     = lov_r0(loo);
306         struct cl_lock          *parent = lck->lls_cl.cls_lock;
307
308         ENTRY;
309
310         lck->lls_orig = parent->cll_descr;
311         file_start = cl_offset(lov2cl(loo), parent->cll_descr.cld_start);
312         file_end   = cl_offset(lov2cl(loo), parent->cll_descr.cld_end + 1) - 1;
313
314         for (i = 0, nr = 0; i < r0->lo_nr; i++) {
315                 /*
316                  * XXX for wide striping smarter algorithm is desirable,
317                  * breaking out of the loop, early.
318                  */
319                 if (lov_stripe_intersects(r0->lo_lsm, i,
320                                           file_start, file_end, &start, &end))
321                         nr++;
322         }
323         LASSERT(nr > 0);
324         OBD_ALLOC(lck->lls_sub, nr * sizeof lck->lls_sub[0]);
325         if (lck->lls_sub == NULL)
326                 RETURN(-ENOMEM);
327
328         lck->lls_nr = nr;
329         /*
330          * First, fill in sub-lock descriptions in
331          * lck->lls_sub[].sub_descr. They are used by lov_sublock_alloc()
332          * (called below in this function, and by lov_lock_enqueue()) to
333          * create sub-locks. At this moment, no other thread can access
334          * top-lock.
335          */
336         for (i = 0, nr = 0; i < r0->lo_nr; ++i) {
337                 if (lov_stripe_intersects(r0->lo_lsm, i,
338                                           file_start, file_end, &start, &end)) {
339                         struct cl_lock_descr *descr;
340
341                         descr = &lck->lls_sub[nr].sub_descr;
342
343                         LASSERT(descr->cld_obj == NULL);
344                         descr->cld_obj   = lovsub2cl(r0->lo_sub[i]);
345                         descr->cld_start = cl_index(descr->cld_obj, start);
346                         descr->cld_end   = cl_index(descr->cld_obj, end);
347                         descr->cld_mode  = parent->cll_descr.cld_mode;
348                         descr->cld_gid   = parent->cll_descr.cld_gid;
349                         descr->cld_enq_flags   = parent->cll_descr.cld_enq_flags;
350                         /* XXX has no effect */
351                         lck->lls_sub[nr].sub_got = *descr;
352                         lck->lls_sub[nr].sub_stripe = i;
353                         nr++;
354                 }
355         }
356         LASSERT(nr == lck->lls_nr);
357         /*
358          * Then, create sub-locks. Once at least one sub-lock was created,
359          * top-lock can be reached by other threads.
360          */
361         for (i = 0; i < lck->lls_nr; ++i) {
362                 struct cl_lock       *sublock;
363                 struct lov_lock_link *link;
364
365                 if (lck->lls_sub[i].sub_lock == NULL) {
366                         sublock = lov_sublock_alloc(env, io, lck, i, &link);
367                         if (IS_ERR(sublock)) {
368                                 result = PTR_ERR(sublock);
369                                 break;
370                         }
371                         cl_lock_get_trust(sublock);
372                         cl_lock_mutex_get(env, sublock);
373                         cl_lock_mutex_get(env, parent);
374                         /*
375                          * recheck under mutex that sub-lock wasn't created
376                          * concurrently, and that top-lock is still alive.
377                          */
378                         if (lck->lls_sub[i].sub_lock == NULL &&
379                             parent->cll_state < CLS_FREEING) {
380                                 lov_sublock_adopt(env, lck, sublock, i, link);
381                                 cl_lock_mutex_put(env, parent);
382                         } else {
383                                 OBD_SLAB_FREE_PTR(link, lov_lock_link_kmem);
384                                 cl_lock_mutex_put(env, parent);
385                                 cl_lock_unhold(env, sublock,
386                                                "lov-parent", parent);
387                         }
388                         cl_lock_mutex_put(env, sublock);
389                         cl_lock_put(env, sublock);
390                 }
391         }
392         /*
393          * Some sub-locks can be missing at this point. This is not a problem,
394          * because enqueue will create them anyway. Main duty of this function
395          * is to fill in sub-lock descriptions in a race free manner.
396          */
397         RETURN(result);
398 }
399
400 static int lov_sublock_release(const struct lu_env *env, struct lov_lock *lck,
401                                int i, int deluser, int rc)
402 {
403         struct cl_lock *parent = lck->lls_cl.cls_lock;
404
405         LASSERT(cl_lock_is_mutexed(parent));
406         ENTRY;
407
408         if (lck->lls_sub[i].sub_flags & LSF_HELD) {
409                 struct cl_lock    *sublock;
410                 int dying;
411
412                 LASSERT(lck->lls_sub[i].sub_lock != NULL);
413                 sublock = lck->lls_sub[i].sub_lock->lss_cl.cls_lock;
414                 LASSERT(cl_lock_is_mutexed(sublock));
415
416                 lck->lls_sub[i].sub_flags &= ~LSF_HELD;
417                 if (deluser)
418                         cl_lock_user_del(env, sublock);
419                 /*
420                  * If the last hold is released, and cancellation is pending
421                  * for a sub-lock, release parent mutex, to avoid keeping it
422                  * while sub-lock is being paged out.
423                  */
424                 dying = (sublock->cll_descr.cld_mode == CLM_PHANTOM ||
425                          sublock->cll_descr.cld_mode == CLM_GROUP ||
426                          (sublock->cll_flags & (CLF_CANCELPEND|CLF_DOOMED))) &&
427                         sublock->cll_holds == 1;
428                 if (dying)
429                         cl_lock_mutex_put(env, parent);
430                 cl_lock_unhold(env, sublock, "lov-parent", parent);
431                 if (dying) {
432                         cl_lock_mutex_get(env, parent);
433                         rc = lov_subresult(rc, CLO_REPEAT);
434                 }
435                 /*
436                  * From now on lck->lls_sub[i].sub_lock is a "weak" pointer,
437                  * not backed by a reference on a
438                  * sub-lock. lovsub_lock_delete() will clear
439                  * lck->lls_sub[i].sub_lock under semaphores, just before
440                  * sub-lock is destroyed.
441                  */
442         }
443         RETURN(rc);
444 }
445
446 static void lov_sublock_hold(const struct lu_env *env, struct lov_lock *lck,
447                              int i)
448 {
449         struct cl_lock *parent = lck->lls_cl.cls_lock;
450
451         LASSERT(cl_lock_is_mutexed(parent));
452         ENTRY;
453
454         if (!(lck->lls_sub[i].sub_flags & LSF_HELD)) {
455                 struct cl_lock *sublock;
456
457                 LASSERT(lck->lls_sub[i].sub_lock != NULL);
458                 sublock = lck->lls_sub[i].sub_lock->lss_cl.cls_lock;
459                 LASSERT(cl_lock_is_mutexed(sublock));
460                 LASSERT(sublock->cll_state != CLS_FREEING);
461
462                 lck->lls_sub[i].sub_flags |= LSF_HELD;
463
464                 cl_lock_get_trust(sublock);
465                 cl_lock_hold_add(env, sublock, "lov-parent", parent);
466                 cl_lock_user_add(env, sublock);
467                 cl_lock_put(env, sublock);
468         }
469         EXIT;
470 }
471
472 static void lov_lock_fini(const struct lu_env *env,
473                           struct cl_lock_slice *slice)
474 {
475         struct lov_lock *lck;
476         int i;
477
478         ENTRY;
479         lck = cl2lov_lock(slice);
480         LASSERT(lck->lls_nr_filled == 0);
481         if (lck->lls_sub != NULL) {
482                 for (i = 0; i < lck->lls_nr; ++i)
483                         /*
484                          * No sub-locks exists at this point, as sub-lock has
485                          * a reference on its parent.
486                          */
487                         LASSERT(lck->lls_sub[i].sub_lock == NULL);
488                 OBD_FREE(lck->lls_sub, lck->lls_nr * sizeof lck->lls_sub[0]);
489         }
490         OBD_SLAB_FREE_PTR(lck, lov_lock_kmem);
491         EXIT;
492 }
493
494 /**
495  * Tries to advance a state machine of a given sub-lock toward enqueuing of
496  * the top-lock.
497  *
498  * \retval 0 if state-transition can proceed
499  * \retval -ve otherwise.
500  */
501 static int lov_lock_enqueue_one(const struct lu_env *env, struct lov_lock *lck,
502                                 struct cl_lock *sublock,
503                                 struct cl_io *io, __u32 enqflags, int last)
504 {
505         int result;
506         ENTRY;
507
508         /* first, try to enqueue a sub-lock ... */
509         result = cl_enqueue_try(env, sublock, io, enqflags);
510         if (sublock->cll_state == CLS_ENQUEUED)
511                 /* if it is enqueued, try to `wait' on it---maybe it's already
512                  * granted */
513                 result = cl_wait_try(env, sublock);
514         /*
515          * If CEF_ASYNC flag is set, then all sub-locks can be enqueued in
516          * parallel, otherwise---enqueue has to wait until sub-lock is granted
517          * before proceeding to the next one.
518          */
519         if (result == CLO_WAIT && sublock->cll_state <= CLS_HELD &&
520             enqflags & CEF_ASYNC && !last)
521                 result = 0;
522         RETURN(result);
523 }
524
525 /**
526  * Helper function for lov_lock_enqueue() that creates missing sub-lock.
527  */
528 static int lov_sublock_fill(const struct lu_env *env, struct cl_lock *parent,
529                             struct cl_io *io, struct lov_lock *lck, int idx)
530 {
531         struct lov_lock_link *link;
532         struct cl_lock       *sublock;
533         int                   result;
534
535         LASSERT(parent->cll_depth == 1);
536         cl_lock_mutex_put(env, parent);
537         sublock = lov_sublock_alloc(env, io, lck, idx, &link);
538         if (!IS_ERR(sublock))
539                 cl_lock_mutex_get(env, sublock);
540         cl_lock_mutex_get(env, parent);
541
542         if (!IS_ERR(sublock)) {
543                 cl_lock_get_trust(sublock);
544                 if (parent->cll_state == CLS_QUEUING &&
545                     lck->lls_sub[idx].sub_lock == NULL) {
546                         lov_sublock_adopt(env, lck, sublock, idx, link);
547                 } else {
548                         OBD_SLAB_FREE_PTR(link, lov_lock_link_kmem);
549                         /* other thread allocated sub-lock, or enqueue is no
550                          * longer going on */
551                         cl_lock_mutex_put(env, parent);
552                         cl_lock_unhold(env, sublock, "lov-parent", parent);
553                         cl_lock_mutex_get(env, parent);
554                 }
555                 cl_lock_mutex_put(env, sublock);
556                 cl_lock_put(env, sublock);
557                 result = CLO_REPEAT;
558         } else
559                 result = PTR_ERR(sublock);
560         return result;
561 }
562
563 /**
564  * Implementation of cl_lock_operations::clo_enqueue() for lov layer. This
565  * function is rather subtle, as it enqueues top-lock (i.e., advances top-lock
566  * state machine from CLS_QUEUING to CLS_ENQUEUED states) by juggling sub-lock
567  * state machines in the face of sub-locks sharing (by multiple top-locks),
568  * and concurrent sub-lock cancellations.
569  */
570 static int lov_lock_enqueue(const struct lu_env *env,
571                             const struct cl_lock_slice *slice,
572                             struct cl_io *io, __u32 enqflags)
573 {
574         struct cl_lock         *lock    = slice->cls_lock;
575         struct lov_lock        *lck     = cl2lov_lock(slice);
576         struct cl_lock_closure *closure = lov_closure_get(env, lock);
577         int i;
578         int result;
579         enum cl_lock_state minstate;
580
581         ENTRY;
582
583         for (result = 0, minstate = CLS_FREEING, i = 0; i < lck->lls_nr; ++i) {
584                 int rc;
585                 struct lovsub_lock     *sub;
586                 struct lov_lock_sub    *lls;
587                 struct cl_lock         *sublock;
588                 struct lov_sublock_env *subenv;
589
590                 if (lock->cll_state != CLS_QUEUING) {
591                         /*
592                          * Lock might have left QUEUING state if previous
593                          * iteration released its mutex. Stop enqueing in this
594                          * case and let the upper layer to decide what to do.
595                          */
596                         LASSERT(i > 0 && result != 0);
597                         break;
598                 }
599
600                 lls = &lck->lls_sub[i];
601                 sub = lls->sub_lock;
602                 /*
603                  * Sub-lock might have been canceled, while top-lock was
604                  * cached.
605                  */
606                 if (sub == NULL) {
607                         result = lov_sublock_fill(env, lock, io, lck, i);
608                         /* lov_sublock_fill() released @lock mutex,
609                          * restart. */
610                         break;
611                 }
612                 sublock = sub->lss_cl.cls_lock;
613                 rc = lov_sublock_lock(env, lck, lls, closure, &subenv);
614                 if (rc == 0) {
615                         lov_sublock_hold(env, lck, i);
616                         rc = lov_lock_enqueue_one(subenv->lse_env, lck, sublock,
617                                                   subenv->lse_io, enqflags,
618                                                   i == lck->lls_nr - 1);
619                         minstate = min(minstate, sublock->cll_state);
620                         /*
621                          * Don't hold a sub-lock in CLS_CACHED state, see
622                          * description for lov_lock::lls_sub.
623                          */
624                         if (sublock->cll_state > CLS_HELD)
625                                 rc = lov_sublock_release(env, lck, i, 1, rc);
626                         lov_sublock_unlock(env, sub, closure, subenv);
627                 }
628                 result = lov_subresult(result, rc);
629                 if (result != 0)
630                         break;
631         }
632         cl_lock_closure_fini(closure);
633         RETURN(result ?: minstate >= CLS_ENQUEUED ? 0 : CLO_WAIT);
634 }
635
636 static int lov_lock_unuse(const struct lu_env *env,
637                           const struct cl_lock_slice *slice)
638 {
639         struct lov_lock        *lck     = cl2lov_lock(slice);
640         struct cl_lock_closure *closure = lov_closure_get(env, slice->cls_lock);
641         int i;
642         int result;
643
644         ENTRY;
645
646         for (result = 0, i = 0; i < lck->lls_nr; ++i) {
647                 int rc;
648                 struct lovsub_lock     *sub;
649                 struct cl_lock         *sublock;
650                 struct lov_lock_sub    *lls;
651                 struct lov_sublock_env *subenv;
652
653                 /* top-lock state cannot change concurrently, because single
654                  * thread (one that released the last hold) carries unlocking
655                  * to the completion. */
656                 LASSERT(slice->cls_lock->cll_state == CLS_INTRANSIT);
657                 lls = &lck->lls_sub[i];
658                 sub = lls->sub_lock;
659                 if (sub == NULL)
660                         continue;
661
662                 sublock = sub->lss_cl.cls_lock;
663                 rc = lov_sublock_lock(env, lck, lls, closure, &subenv);
664                 if (rc == 0) {
665                         if (lls->sub_flags & LSF_HELD) {
666                                 LASSERT(sublock->cll_state == CLS_HELD);
667                                 rc = cl_unuse_try(subenv->lse_env, sublock);
668                                 rc = lov_sublock_release(env, lck, i, 0, rc);
669                         }
670                         lov_sublock_unlock(env, sub, closure, subenv);
671                 }
672                 result = lov_subresult(result, rc);
673         }
674
675         if (result == 0 && lck->lls_cancel_race) {
676                 lck->lls_cancel_race = 0;
677                 result = -ESTALE;
678         }
679         cl_lock_closure_fini(closure);
680         RETURN(result);
681 }
682
683
684 static void lov_lock_cancel(const struct lu_env *env,
685                            const struct cl_lock_slice *slice)
686 {
687         struct lov_lock        *lck     = cl2lov_lock(slice);
688         struct cl_lock_closure *closure = lov_closure_get(env, slice->cls_lock);
689         int i;
690         int result;
691
692         ENTRY;
693
694         for (result = 0, i = 0; i < lck->lls_nr; ++i) {
695                 int rc;
696                 struct lovsub_lock     *sub;
697                 struct cl_lock         *sublock;
698                 struct lov_lock_sub    *lls;
699                 struct lov_sublock_env *subenv;
700
701                 /* top-lock state cannot change concurrently, because single
702                  * thread (one that released the last hold) carries unlocking
703                  * to the completion. */
704                 lls = &lck->lls_sub[i];
705                 sub = lls->sub_lock;
706                 if (sub == NULL)
707                         continue;
708
709                 sublock = sub->lss_cl.cls_lock;
710                 rc = lov_sublock_lock(env, lck, lls, closure, &subenv);
711                 if (rc == 0) {
712                         if (!(lls->sub_flags & LSF_HELD)) {
713                                 lov_sublock_unlock(env, sub, closure, subenv);
714                                 continue;
715                         }
716
717                         switch(sublock->cll_state) {
718                         case CLS_HELD:
719                                 rc = cl_unuse_try(subenv->lse_env,
720                                                   sublock);
721                                 lov_sublock_release(env, lck, i, 0, 0);
722                                 break;
723                         case CLS_ENQUEUED:
724                                 /* TODO: it's not a good idea to cancel this
725                                  * lock because it's innocent. But it's
726                                  * acceptable. The better way would be to
727                                  * define a new lock method to unhold the
728                                  * dlm lock. */
729                                 cl_lock_cancel(env, sublock);
730                         default:
731                                 lov_sublock_release(env, lck, i, 1, 0);
732                                 break;
733                         }
734                         lov_sublock_unlock(env, sub, closure, subenv);
735                 }
736
737                 if (rc == CLO_REPEAT) {
738                         --i;
739                         continue;
740                 }
741
742                 result = lov_subresult(result, rc);
743         }
744
745         if (result)
746                 CL_LOCK_DEBUG(D_ERROR, env, slice->cls_lock,
747                               "lov_lock_cancel fails with %d.\n", result);
748
749         cl_lock_closure_fini(closure);
750 }
751
752 static int lov_lock_wait(const struct lu_env *env,
753                          const struct cl_lock_slice *slice)
754 {
755         struct lov_lock        *lck     = cl2lov_lock(slice);
756         struct cl_lock_closure *closure = lov_closure_get(env, slice->cls_lock);
757         enum cl_lock_state      minstate;
758         int                     result;
759         int                     i;
760
761         ENTRY;
762
763         for (result = 0, minstate = CLS_FREEING, i = 0; i < lck->lls_nr; ++i) {
764                 int rc;
765                 struct lovsub_lock     *sub;
766                 struct cl_lock         *sublock;
767                 struct lov_lock_sub    *lls;
768                 struct lov_sublock_env *subenv;
769
770                 lls = &lck->lls_sub[i];
771                 sub = lls->sub_lock;
772                 LASSERT(sub != NULL);
773                 sublock = sub->lss_cl.cls_lock;
774                 rc = lov_sublock_lock(env, lck, lls, closure, &subenv);
775                 if (rc == 0) {
776                         LASSERT(sublock->cll_state >= CLS_ENQUEUED);
777                         if (sublock->cll_state < CLS_HELD)
778                                 rc = cl_wait_try(env, sublock);
779
780                         minstate = min(minstate, sublock->cll_state);
781                         lov_sublock_unlock(env, sub, closure, subenv);
782                 }
783                 result = lov_subresult(result, rc);
784                 if (result != 0)
785                         break;
786         }
787         cl_lock_closure_fini(closure);
788         RETURN(result ?: minstate >= CLS_HELD ? 0 : CLO_WAIT);
789 }
790
791 static int lov_lock_use(const struct lu_env *env,
792                         const struct cl_lock_slice *slice)
793 {
794         struct lov_lock        *lck     = cl2lov_lock(slice);
795         struct cl_lock_closure *closure = lov_closure_get(env, slice->cls_lock);
796         int                     result;
797         int                     i;
798
799         LASSERT(slice->cls_lock->cll_state == CLS_INTRANSIT);
800         ENTRY;
801
802         for (result = 0, i = 0; i < lck->lls_nr; ++i) {
803                 int rc;
804                 struct lovsub_lock     *sub;
805                 struct cl_lock         *sublock;
806                 struct lov_lock_sub    *lls;
807                 struct lov_sublock_env *subenv;
808
809                 LASSERT(slice->cls_lock->cll_state == CLS_INTRANSIT);
810
811                 lls = &lck->lls_sub[i];
812                 sub = lls->sub_lock;
813                 if (sub == NULL) {
814                         /*
815                          * Sub-lock might have been canceled, while top-lock was
816                          * cached.
817                          */
818                         result = -ESTALE;
819                         break;
820                 }
821
822                 sublock = sub->lss_cl.cls_lock;
823                 rc = lov_sublock_lock(env, lck, lls, closure, &subenv);
824                 if (rc == 0) {
825                         LASSERT(sublock->cll_state != CLS_FREEING);
826                         lov_sublock_hold(env, lck, i);
827                         if (sublock->cll_state == CLS_CACHED) {
828                                 rc = cl_use_try(subenv->lse_env, sublock, 0);
829                                 if (rc != 0)
830                                         rc = lov_sublock_release(env, lck,
831                                                                  i, 1, rc);
832                         }
833                         lov_sublock_unlock(env, sub, closure, subenv);
834                 }
835                 result = lov_subresult(result, rc);
836                 if (result != 0)
837                         break;
838         }
839
840         if (lck->lls_cancel_race) {
841                 /*
842                  * If there is unlocking happened at the same time, then
843                  * sublock_lock state should be FREEING, and lov_sublock_lock
844                  * should return CLO_REPEAT. In this case, it should return
845                  * ESTALE, and up layer should reset the lock state to be NEW.
846                  */
847                 lck->lls_cancel_race = 0;
848                 LASSERT(result != 0);
849                 result = -ESTALE;
850         }
851         cl_lock_closure_fini(closure);
852         RETURN(result);
853 }
854
855 #if 0
856 static int lock_lock_multi_match()
857 {
858         struct cl_lock          *lock    = slice->cls_lock;
859         struct cl_lock_descr    *subneed = &lov_env_info(env)->lti_ldescr;
860         struct lov_object       *loo     = cl2lov(lov->lls_cl.cls_obj);
861         struct lov_layout_raid0 *r0      = lov_r0(loo);
862         struct lov_lock_sub     *sub;
863         struct cl_object        *subobj;
864         obd_off  fstart;
865         obd_off  fend;
866         obd_off  start;
867         obd_off  end;
868         int i;
869
870         fstart = cl_offset(need->cld_obj, need->cld_start);
871         fend   = cl_offset(need->cld_obj, need->cld_end + 1) - 1;
872         subneed->cld_mode = need->cld_mode;
873         cl_lock_mutex_get(env, lock);
874         for (i = 0; i < lov->lls_nr; ++i) {
875                 sub = &lov->lls_sub[i];
876                 if (sub->sub_lock == NULL)
877                         continue;
878                 subobj = sub->sub_descr.cld_obj;
879                 if (!lov_stripe_intersects(r0->lo_lsm, sub->sub_stripe,
880                                            fstart, fend, &start, &end))
881                         continue;
882                 subneed->cld_start = cl_index(subobj, start);
883                 subneed->cld_end   = cl_index(subobj, end);
884                 subneed->cld_obj   = subobj;
885                 if (!cl_lock_ext_match(&sub->sub_got, subneed)) {
886                         result = 0;
887                         break;
888                 }
889         }
890         cl_lock_mutex_put(env, lock);
891 }
892 #endif
893
894 /**
895  * Check if the extent region \a descr is covered by \a child against the
896  * specific \a stripe.
897  */
898 static int lov_lock_stripe_is_matching(const struct lu_env *env,
899                                        struct lov_object *lov, int stripe,
900                                        const struct cl_lock_descr *child,
901                                        const struct cl_lock_descr *descr)
902 {
903         struct lov_stripe_md *lsm = lov_r0(lov)->lo_lsm;
904         obd_off start;
905         obd_off end;
906         int result;
907
908         if (lov_r0(lov)->lo_nr == 1)
909                 return cl_lock_ext_match(child, descr);
910
911         /*
912          * For a multi-stripes object:
913          * - make sure the descr only covers child's stripe, and
914          * - check if extent is matching.
915          */
916         start = cl_offset(&lov->lo_cl, descr->cld_start);
917         end   = cl_offset(&lov->lo_cl, descr->cld_end + 1) - 1;
918         result = end - start <= lsm->lsm_stripe_size &&
919                  stripe == lov_stripe_number(lsm, start) &&
920                  stripe == lov_stripe_number(lsm, end);
921         if (result) {
922                 struct cl_lock_descr *subd = &lov_env_info(env)->lti_ldescr;
923                 obd_off sub_start;
924                 obd_off sub_end;
925
926                 subd->cld_obj  = NULL;   /* don't need sub object at all */
927                 subd->cld_mode = descr->cld_mode;
928                 subd->cld_gid  = descr->cld_gid;
929                 result = lov_stripe_intersects(lsm, stripe, start, end,
930                                                &sub_start, &sub_end);
931                 LASSERT(result);
932                 subd->cld_start = cl_index(child->cld_obj, sub_start);
933                 subd->cld_end   = cl_index(child->cld_obj, sub_end);
934                 result = cl_lock_ext_match(child, subd);
935         }
936         return result;
937 }
938
939 /**
940  * An implementation of cl_lock_operations::clo_fits_into() method.
941  *
942  * Checks whether a lock (given by \a slice) is suitable for \a
943  * io. Multi-stripe locks can be used only for "quick" io, like truncate, or
944  * O_APPEND write.
945  *
946  * \see ccc_lock_fits_into().
947  */
948 static int lov_lock_fits_into(const struct lu_env *env,
949                               const struct cl_lock_slice *slice,
950                               const struct cl_lock_descr *need,
951                               const struct cl_io *io)
952 {
953         struct lov_lock   *lov = cl2lov_lock(slice);
954         struct lov_object *obj = cl2lov(slice->cls_obj);
955         int result;
956
957         LASSERT(cl_object_same(need->cld_obj, slice->cls_obj));
958         LASSERT(lov->lls_nr > 0);
959
960         ENTRY;
961
962         if (need->cld_mode == CLM_GROUP)
963                 /*
964                  * always allow to match group lock.
965                  */
966                 result = cl_lock_ext_match(&lov->lls_orig, need);
967         else if (lov->lls_nr == 1) {
968                 struct cl_lock_descr *got = &lov->lls_sub[0].sub_got;
969                 result = lov_lock_stripe_is_matching(env,
970                                                      cl2lov(slice->cls_obj),
971                                                      lov->lls_sub[0].sub_stripe,
972                                                      got, need);
973         } else if (io->ci_type != CIT_TRUNC && io->ci_type != CIT_MISC &&
974                    !cl_io_is_append(io) && need->cld_mode != CLM_PHANTOM)
975                 /*
976                  * Multi-stripe locks are only suitable for `quick' IO and for
977                  * glimpse.
978                  */
979                 result = 0;
980         else
981                 /*
982                  * Most general case: multi-stripe existing lock, and
983                  * (potentially) multi-stripe @need lock. Check that @need is
984                  * covered by @lov's sub-locks.
985                  *
986                  * For now, ignore lock expansions made by the server, and
987                  * match against original lock extent.
988                  */
989                 result = cl_lock_ext_match(&lov->lls_orig, need);
990         CDEBUG(D_DLMTRACE, DDESCR"/"DDESCR" %i %i/%i: %i\n",
991                PDESCR(&lov->lls_orig), PDESCR(&lov->lls_sub[0].sub_got),
992                lov->lls_sub[0].sub_stripe, lov->lls_nr, lov_r0(obj)->lo_nr,
993                result);
994         RETURN(result);
995 }
996
997 void lov_lock_unlink(const struct lu_env *env,
998                      struct lov_lock_link *link, struct lovsub_lock *sub)
999 {
1000         struct lov_lock *lck    = link->lll_super;
1001         struct cl_lock  *parent = lck->lls_cl.cls_lock;
1002
1003         LASSERT(cl_lock_is_mutexed(parent));
1004         LASSERT(cl_lock_is_mutexed(sub->lss_cl.cls_lock));
1005         ENTRY;
1006
1007         list_del_init(&link->lll_list);
1008         LASSERT(lck->lls_sub[link->lll_idx].sub_lock == sub);
1009         /* yank this sub-lock from parent's array */
1010         lck->lls_sub[link->lll_idx].sub_lock = NULL;
1011         LASSERT(lck->lls_nr_filled > 0);
1012         lck->lls_nr_filled--;
1013         lu_ref_del(&parent->cll_reference, "lov-child", sub->lss_cl.cls_lock);
1014         cl_lock_put(env, parent);
1015         OBD_SLAB_FREE_PTR(link, lov_lock_link_kmem);
1016         EXIT;
1017 }
1018
1019 struct lov_lock_link *lov_lock_link_find(const struct lu_env *env,
1020                                          struct lov_lock *lck,
1021                                          struct lovsub_lock *sub)
1022 {
1023         struct lov_lock_link *scan;
1024
1025         LASSERT(cl_lock_is_mutexed(sub->lss_cl.cls_lock));
1026         ENTRY;
1027
1028         list_for_each_entry(scan, &sub->lss_parents, lll_list) {
1029                 if (scan->lll_super == lck)
1030                         RETURN(scan);
1031         }
1032         RETURN(NULL);
1033 }
1034
1035 /**
1036  * An implementation of cl_lock_operations::clo_delete() method. This is
1037  * invoked for "top-to-bottom" delete, when lock destruction starts from the
1038  * top-lock, e.g., as a result of inode destruction.
1039  *
1040  * Unlinks top-lock from all its sub-locks. Sub-locks are not deleted there:
1041  * this is done separately elsewhere:
1042  *
1043  *     - for inode destruction, lov_object_delete() calls cl_object_kill() for
1044  *       each sub-object, purging its locks;
1045  *
1046  *     - in other cases (e.g., a fatal error with a top-lock) sub-locks are
1047  *       left in the cache.
1048  */
1049 static void lov_lock_delete(const struct lu_env *env,
1050                             const struct cl_lock_slice *slice)
1051 {
1052         struct lov_lock        *lck     = cl2lov_lock(slice);
1053         struct cl_lock_closure *closure = lov_closure_get(env, slice->cls_lock);
1054         int i;
1055
1056         LASSERT(slice->cls_lock->cll_state == CLS_FREEING);
1057         ENTRY;
1058
1059         for (i = 0; i < lck->lls_nr; ++i) {
1060                 struct lov_lock_sub *lls;
1061                 struct lovsub_lock  *lsl;
1062                 struct cl_lock      *sublock;
1063                 int rc;
1064
1065                 lls = &lck->lls_sub[i];
1066                 lsl = lls->sub_lock;
1067                 if (lsl == NULL)
1068                         continue;
1069
1070                 sublock = lsl->lss_cl.cls_lock;
1071                 rc = lov_sublock_lock(env, lck, lls, closure, NULL);
1072                 if (rc == 0) {
1073                         if (lls->sub_flags & LSF_HELD)
1074                                 lov_sublock_release(env, lck, i, 1, 0);
1075                         if (sublock->cll_state < CLS_FREEING) {
1076                                 struct lov_lock_link *link;
1077
1078                                 link = lov_lock_link_find(env, lck, lsl);
1079                                 LASSERT(link != NULL);
1080                                 lov_lock_unlink(env, link, lsl);
1081                                 LASSERT(lck->lls_sub[i].sub_lock == NULL);
1082                         }
1083                         lov_sublock_unlock(env, lsl, closure, NULL);
1084                 } else if (rc == CLO_REPEAT) {
1085                         --i; /* repeat with this lock */
1086                 } else {
1087                         CL_LOCK_DEBUG(D_ERROR, env, sublock,
1088                                       "Cannot get sub-lock for delete: %i\n",
1089                                       rc);
1090                 }
1091         }
1092         cl_lock_closure_fini(closure);
1093         EXIT;
1094 }
1095
1096 static int lov_lock_print(const struct lu_env *env, void *cookie,
1097                           lu_printer_t p, const struct cl_lock_slice *slice)
1098 {
1099         struct lov_lock *lck = cl2lov_lock(slice);
1100         int              i;
1101
1102         (*p)(env, cookie, "%d\n", lck->lls_nr);
1103         for (i = 0; i < lck->lls_nr; ++i) {
1104                 struct lov_lock_sub *sub;
1105
1106                 sub = &lck->lls_sub[i];
1107                 (*p)(env, cookie, "    %d %x: ", i, sub->sub_flags);
1108                 if (sub->sub_lock != NULL)
1109                         cl_lock_print(env, cookie, p,
1110                                       sub->sub_lock->lss_cl.cls_lock);
1111                 else
1112                         (*p)(env, cookie, "---\n");
1113         }
1114         return 0;
1115 }
1116
1117 static const struct cl_lock_operations lov_lock_ops = {
1118         .clo_fini      = lov_lock_fini,
1119         .clo_enqueue   = lov_lock_enqueue,
1120         .clo_wait      = lov_lock_wait,
1121         .clo_use       = lov_lock_use,
1122         .clo_unuse     = lov_lock_unuse,
1123         .clo_cancel    = lov_lock_cancel,
1124         .clo_fits_into = lov_lock_fits_into,
1125         .clo_delete    = lov_lock_delete,
1126         .clo_print     = lov_lock_print
1127 };
1128
1129 int lov_lock_init_raid0(const struct lu_env *env, struct cl_object *obj,
1130                         struct cl_lock *lock, const struct cl_io *io)
1131 {
1132         struct lov_lock *lck;
1133         int result;
1134
1135         ENTRY;
1136         OBD_SLAB_ALLOC_PTR_GFP(lck, lov_lock_kmem, CFS_ALLOC_IO);
1137         if (lck != NULL) {
1138                 cl_lock_slice_add(lock, &lck->lls_cl, obj, &lov_lock_ops);
1139                 result = lov_lock_sub_init(env, lck, io);
1140         } else
1141                 result = -ENOMEM;
1142         RETURN(result);
1143 }
1144
1145 static struct cl_lock_closure *lov_closure_get(const struct lu_env *env,
1146                                                struct cl_lock *parent)
1147 {
1148         struct cl_lock_closure *closure;
1149
1150         closure = &lov_env_info(env)->lti_closure;
1151         LASSERT(list_empty(&closure->clc_list));
1152         cl_lock_closure_init(env, closure, parent, 1);
1153         return closure;
1154 }
1155
1156
1157 /** @} lov */