Whamcloud - gitweb
ab1d9d9b0e858ed782beca1bc7fdcbaaac503ac0
[fs/lustre-release.git] / lustre / lov / lovsub_lock.c
1 /*
2  * GPL HEADER START
3  *
4  * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.
5  *
6  * This program is free software; you can redistribute it and/or modify
7  * it under the terms of the GNU General Public License version 2 only,
8  * as published by the Free Software Foundation.
9  *
10  * This program is distributed in the hope that it will be useful, but
11  * WITHOUT ANY WARRANTY; without even the implied warranty of
12  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
13  * General Public License version 2 for more details (a copy is included
14  * in the LICENSE file that accompanied this code).
15  *
16  * You should have received a copy of the GNU General Public License
17  * version 2 along with this program; If not, see
18  * http://www.sun.com/software/products/lustre/docs/GPLv2.pdf
19  *
20  * Please contact Sun Microsystems, Inc., 4150 Network Circle, Santa Clara,
21  * CA 95054 USA or visit www.sun.com if you need additional information or
22  * have any questions.
23  *
24  * GPL HEADER END
25  */
26 /*
27  * Copyright (c) 2008, 2010, Oracle and/or its affiliates. All rights reserved.
28  * Use is subject to license terms.
29  *
30  * Copyright (c) 2011, Whamcloud, Inc.
31  */
32 /*
33  * This file is part of Lustre, http://www.lustre.org/
34  * Lustre is a trademark of Sun Microsystems, Inc.
35  *
36  * Implementation of cl_lock for LOVSUB 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 /*****************************************************************************
50  *
51  * Lovsub lock operations.
52  *
53  */
54
55 static void lovsub_lock_fini(const struct lu_env *env,
56                              struct cl_lock_slice *slice)
57 {
58         struct lovsub_lock   *lsl;
59
60         ENTRY;
61         lsl = cl2lovsub_lock(slice);
62         LASSERT(cfs_list_empty(&lsl->lss_parents));
63         OBD_SLAB_FREE_PTR(lsl, lovsub_lock_kmem);
64         EXIT;
65 }
66
67 static void lovsub_parent_lock(const struct lu_env *env, struct lov_lock *lov)
68 {
69         struct cl_lock *parent;
70
71         ENTRY;
72         parent = lov->lls_cl.cls_lock;
73         cl_lock_get(parent);
74         lu_ref_add(&parent->cll_reference, "lovsub-parent", cfs_current());
75         cl_lock_mutex_get(env, parent);
76         EXIT;
77 }
78
79 static void lovsub_parent_unlock(const struct lu_env *env, struct lov_lock *lov)
80 {
81         struct cl_lock *parent;
82
83         ENTRY;
84         parent = lov->lls_cl.cls_lock;
85         cl_lock_mutex_put(env, lov->lls_cl.cls_lock);
86         lu_ref_del(&parent->cll_reference, "lovsub-parent", cfs_current());
87         cl_lock_put(env, parent);
88         EXIT;
89 }
90
91 /**
92  * Implements cl_lock_operations::clo_state() method for lovsub layer, which
93  * method is called whenever sub-lock state changes. Propagates state change
94  * to the top-locks.
95  */
96 static void lovsub_lock_state(const struct lu_env *env,
97                               const struct cl_lock_slice *slice,
98                               enum cl_lock_state state)
99 {
100         struct lovsub_lock   *sub = cl2lovsub_lock(slice);
101         struct lov_lock_link *scan;
102
103         LASSERT(cl_lock_is_mutexed(slice->cls_lock));
104         ENTRY;
105
106         cfs_list_for_each_entry(scan, &sub->lss_parents, lll_list) {
107                 struct lov_lock *lov    = scan->lll_super;
108                 struct cl_lock  *parent = lov->lls_cl.cls_lock;
109
110                 if (sub->lss_active != parent) {
111                         lovsub_parent_lock(env, lov);
112                         cl_lock_signal(env, parent);
113                         lovsub_parent_unlock(env, lov);
114                 }
115         }
116         EXIT;
117 }
118
119 /**
120  * Implementation of cl_lock_operation::clo_weigh() estimating lock weight by
121  * asking parent lock.
122  */
123 static unsigned long lovsub_lock_weigh(const struct lu_env *env,
124                                        const struct cl_lock_slice *slice)
125 {
126         struct lovsub_lock *lock = cl2lovsub_lock(slice);
127         struct lov_lock    *lov;
128         unsigned long       dumbbell;
129
130         ENTRY;
131
132         LASSERT(cl_lock_is_mutexed(slice->cls_lock));
133
134         if (!cfs_list_empty(&lock->lss_parents)) {
135                 /*
136                  * It is not clear whether all parents have to be asked and
137                  * their estimations summed, or it is enough to ask one. For
138                  * the current usages, one is always enough.
139                  */
140                 lov = container_of(lock->lss_parents.next,
141                                    struct lov_lock_link, lll_list)->lll_super;
142
143                 lovsub_parent_lock(env, lov);
144                 dumbbell = cl_lock_weigh(env, lov->lls_cl.cls_lock);
145                 lovsub_parent_unlock(env, lov);
146         } else
147                 dumbbell = 0;
148
149         RETURN(dumbbell);
150 }
151
152 /**
153  * Maps start/end offsets within a stripe, to offsets within a file.
154  */
155 static void lovsub_lock_descr_map(const struct cl_lock_descr *in,
156                                   struct lov_object *obj,
157                                   int stripe, struct cl_lock_descr *out)
158 {
159         struct lov_stripe_md *lsm = lov_r0(obj)->lo_lsm;
160         pgoff_t size; /* stripe size in pages */
161         pgoff_t skip; /* how many pages in every stripe are occupied by
162                        * "other" stripes */
163         pgoff_t start;
164         pgoff_t end;
165
166         ENTRY;
167         start = in->cld_start;
168         end   = in->cld_end;
169
170         if (lsm->lsm_stripe_count > 1) {
171                 size = cl_index(lov2cl(obj), lsm->lsm_stripe_size);
172                 skip = (lsm->lsm_stripe_count - 1) * size;
173
174                 /* XXX overflow check here? */
175                 start += start/size * skip + stripe * size;
176
177                 if (end != CL_PAGE_EOF) {
178                         end += end/size * skip + stripe * size;
179                         /*
180                          * And check for overflow...
181                          */
182                         if (end < in->cld_end)
183                                 end = CL_PAGE_EOF;
184                 }
185         }
186         out->cld_start = start;
187         out->cld_end   = end;
188         EXIT;
189 }
190
191 /**
192  * Adjusts parent lock extent when a sub-lock is attached to a parent. This is
193  * called in two ways:
194  *
195  *     - as part of receive call-back, when server returns granted extent to
196  *       the client, and
197  *
198  *     - when top-lock finds existing sub-lock in the cache.
199  *
200  * Note, that lock mode is not propagated to the parent: i.e., if CLM_READ
201  * top-lock matches CLM_WRITE sub-lock, top-lock is still CLM_READ.
202  */
203 int lov_sublock_modify(const struct lu_env *env, struct lov_lock *lov,
204                        struct lovsub_lock *sublock,
205                        const struct cl_lock_descr *d, int idx)
206 {
207         struct cl_lock       *parent;
208         struct lovsub_object *subobj;
209         struct cl_lock_descr *pd;
210         struct cl_lock_descr *parent_descr;
211         int                   result;
212
213         parent       = lov->lls_cl.cls_lock;
214         parent_descr = &parent->cll_descr;
215         LASSERT(cl_lock_mode_match(d->cld_mode, parent_descr->cld_mode));
216
217         subobj = cl2lovsub(sublock->lss_cl.cls_obj);
218         pd     = &lov_env_info(env)->lti_ldescr;
219
220         pd->cld_obj  = parent_descr->cld_obj;
221         pd->cld_mode = parent_descr->cld_mode;
222         pd->cld_gid  = parent_descr->cld_gid;
223         lovsub_lock_descr_map(d, subobj->lso_super, subobj->lso_index, pd);
224         lov->lls_sub[idx].sub_got = *d;
225         /*
226          * Notify top-lock about modification, if lock description changes
227          * materially.
228          */
229         if (!cl_lock_ext_match(parent_descr, pd))
230                 result = cl_lock_modify(env, parent, pd);
231         else
232                 result = 0;
233         return result;
234 }
235
236 static int lovsub_lock_modify(const struct lu_env *env,
237                               const struct cl_lock_slice *s,
238                               const struct cl_lock_descr *d)
239 {
240         struct lovsub_lock   *lock   = cl2lovsub_lock(s);
241         struct lov_lock_link *scan;
242         struct lov_lock      *lov;
243         int result                   = 0;
244
245         ENTRY;
246
247         LASSERT(cl_lock_mode_match(d->cld_mode,
248                                    s->cls_lock->cll_descr.cld_mode));
249         cfs_list_for_each_entry(scan, &lock->lss_parents, lll_list) {
250                 int rc;
251
252                 lov = scan->lll_super;
253                 lovsub_parent_lock(env, lov);
254                 rc = lov_sublock_modify(env, lov, lock, d, scan->lll_idx);
255                 lovsub_parent_unlock(env, lov);
256                 result = result ?: rc;
257         }
258         RETURN(result);
259 }
260
261 static int lovsub_lock_closure(const struct lu_env *env,
262                                const struct cl_lock_slice *slice,
263                                struct cl_lock_closure *closure)
264 {
265         struct lovsub_lock   *sub;
266         struct cl_lock       *parent;
267         struct lov_lock_link *scan;
268         int                   result;
269
270         LASSERT(cl_lock_is_mutexed(slice->cls_lock));
271         ENTRY;
272
273         sub    = cl2lovsub_lock(slice);
274         result = 0;
275
276         cfs_list_for_each_entry(scan, &sub->lss_parents, lll_list) {
277                 parent = scan->lll_super->lls_cl.cls_lock;
278                 result = cl_lock_closure_build(env, parent, closure);
279                 if (result != 0)
280                         break;
281         }
282         RETURN(result);
283 }
284
285 /**
286  * A helper function for lovsub_lock_delete() that deals with a given parent
287  * top-lock.
288  */
289 static int lovsub_lock_delete_one(const struct lu_env *env,
290                                   struct cl_lock *child, struct lov_lock *lov)
291 {
292         struct cl_lock *parent;
293         int             result;
294         ENTRY;
295
296         parent = lov->lls_cl.cls_lock;
297         if (parent->cll_error)
298                 RETURN(0);
299
300         result = 0;
301         switch (parent->cll_state) {
302         case CLS_ENQUEUED:
303                 /* See LU-1355 for the case that a glimpse lock is
304                  * interrupted by signal */
305                 LASSERT(parent->cll_flags & CLF_CANCELLED);
306                 break;
307         case CLS_QUEUING:
308         case CLS_FREEING:
309                 cl_lock_signal(env, parent);
310                 break;
311         case CLS_INTRANSIT:
312                 /*
313                  * Here lies a problem: a sub-lock is canceled while top-lock
314                  * is being unlocked. Top-lock cannot be moved into CLS_NEW
315                  * state, because unlocking has to succeed eventually by
316                  * placing lock into CLS_CACHED (or failing it), see
317                  * cl_unuse_try(). Nor can top-lock be left in CLS_CACHED
318                  * state, because lov maintains an invariant that all
319                  * sub-locks exist in CLS_CACHED (this allows cached top-lock
320                  * to be reused immediately). Nor can we wait for top-lock
321                  * state to change, because this can be synchronous to the
322                  * current thread.
323                  *
324                  * We know for sure that lov_lock_unuse() will be called at
325                  * least one more time to finish un-using, so leave a mark on
326                  * the top-lock, that will be seen by the next call to
327                  * lov_lock_unuse().
328                  */
329                 if (cl_lock_is_intransit(parent))
330                         lov->lls_cancel_race = 1;
331                 break;
332         case CLS_CACHED:
333                 /*
334                  * if a sub-lock is canceled move its top-lock into CLS_NEW
335                  * state to preserve an invariant that a top-lock in
336                  * CLS_CACHED is immediately ready for re-use (i.e., has all
337                  * sub-locks), and so that next attempt to re-use the top-lock
338                  * enqueues missing sub-lock.
339                  */
340                 cl_lock_state_set(env, parent, CLS_NEW);
341                 /* fall through */
342         case CLS_NEW:
343                 /*
344                  * if last sub-lock is canceled, destroy the top-lock (which
345                  * is now `empty') proactively.
346                  */
347                 if (lov->lls_nr_filled == 0) {
348                         /* ... but unfortunately, this cannot be done easily,
349                          * as cancellation of a top-lock might acquire mutices
350                          * of its other sub-locks, violating lock ordering,
351                          * see cl_lock_{cancel,delete}() preconditions.
352                          *
353                          * To work around this, the mutex of this sub-lock is
354                          * released, top-lock is destroyed, and sub-lock mutex
355                          * acquired again. The list of parents has to be
356                          * re-scanned from the beginning after this.
357                          *
358                          * Only do this if no mutices other than on @child and
359                          * @parent are held by the current thread.
360                          *
361                          * TODO: The lock modal here is too complex, because
362                          * the lock may be canceled and deleted by voluntarily:
363                          *    cl_lock_request
364                          *      -> osc_lock_enqueue_wait
365                          *        -> osc_lock_cancel_wait
366                          *          -> cl_lock_delete
367                          *            -> lovsub_lock_delete
368                          *              -> cl_lock_cancel/delete
369                          *                -> ...
370                          *
371                          * The better choice is to spawn a kernel thread for
372                          * this purpose. -jay
373                          */
374                         if (cl_lock_nr_mutexed(env) == 2) {
375                                 cl_lock_mutex_put(env, child);
376                                 cl_lock_cancel(env, parent);
377                                 cl_lock_delete(env, parent);
378                                 result = 1;
379                         }
380                 }
381                 break;
382         case CLS_HELD:
383                 CL_LOCK_DEBUG(D_ERROR, env, parent, "Delete CLS_HELD lock\n");
384         default:
385                 CERROR("Impossible state: %d\n", parent->cll_state);
386                 LBUG();
387                 break;
388         }
389
390         RETURN(result);
391 }
392
393 /**
394  * An implementation of cl_lock_operations::clo_delete() method. This is
395  * invoked in "bottom-to-top" delete, when lock destruction starts from the
396  * sub-lock (e.g, as a result of ldlm lock LRU policy).
397  */
398 static void lovsub_lock_delete(const struct lu_env *env,
399                                const struct cl_lock_slice *slice)
400 {
401         struct cl_lock     *child = slice->cls_lock;
402         struct lovsub_lock *sub   = cl2lovsub_lock(slice);
403         int restart;
404
405         LASSERT(cl_lock_is_mutexed(child));
406
407         ENTRY;
408         /*
409          * Destruction of a sub-lock might take multiple iterations, because
410          * when the last sub-lock of a given top-lock is deleted, top-lock is
411          * canceled proactively, and this requires to release sub-lock
412          * mutex. Once sub-lock mutex has been released, list of its parents
413          * has to be re-scanned from the beginning.
414          */
415         do {
416                 struct lov_lock      *lov;
417                 struct lov_lock_link *scan;
418                 struct lov_lock_link *temp;
419                 struct lov_lock_sub  *subdata;
420
421                 restart = 0;
422                 cfs_list_for_each_entry_safe(scan, temp,
423                                              &sub->lss_parents, lll_list) {
424                         lov     = scan->lll_super;
425                         subdata = &lov->lls_sub[scan->lll_idx];
426                         lovsub_parent_lock(env, lov);
427                         subdata->sub_got = subdata->sub_descr;
428                         lov_lock_unlink(env, scan, sub);
429                         restart = lovsub_lock_delete_one(env, child, lov);
430                         lovsub_parent_unlock(env, lov);
431
432                         if (restart) {
433                                 cl_lock_mutex_get(env, child);
434                                 break;
435                         }
436                }
437         } while (restart);
438         EXIT;
439 }
440
441 static int lovsub_lock_print(const struct lu_env *env, void *cookie,
442                              lu_printer_t p, const struct cl_lock_slice *slice)
443 {
444         struct lovsub_lock   *sub = cl2lovsub_lock(slice);
445         struct lov_lock      *lov;
446         struct lov_lock_link *scan;
447
448         cfs_list_for_each_entry(scan, &sub->lss_parents, lll_list) {
449                 lov = scan->lll_super;
450                 (*p)(env, cookie, "[%d %p ", scan->lll_idx, lov);
451                 if (lov != NULL)
452                         cl_lock_descr_print(env, cookie, p,
453                                             &lov->lls_cl.cls_lock->cll_descr);
454                 (*p)(env, cookie, "] ");
455         }
456         return 0;
457 }
458
459 static const struct cl_lock_operations lovsub_lock_ops = {
460         .clo_fini    = lovsub_lock_fini,
461         .clo_state   = lovsub_lock_state,
462         .clo_delete  = lovsub_lock_delete,
463         .clo_modify  = lovsub_lock_modify,
464         .clo_closure = lovsub_lock_closure,
465         .clo_weigh   = lovsub_lock_weigh,
466         .clo_print   = lovsub_lock_print
467 };
468
469 int lovsub_lock_init(const struct lu_env *env, struct cl_object *obj,
470                      struct cl_lock *lock, const struct cl_io *io)
471 {
472         struct lovsub_lock *lsk;
473         int result;
474
475         ENTRY;
476         OBD_SLAB_ALLOC_PTR_GFP(lsk, lovsub_lock_kmem, CFS_ALLOC_IO);
477         if (lsk != NULL) {
478                 CFS_INIT_LIST_HEAD(&lsk->lss_parents);
479                 cl_lock_slice_add(lock, &lsk->lss_cl, obj, &lovsub_lock_ops);
480                 result = 0;
481         } else
482                 result = -ENOMEM;
483         RETURN(result);
484 }
485
486 /** @} lov */