Whamcloud - gitweb
LU-10658 utils: check mount_lustre for allocation failure
[fs/lustre-release.git] / lustre / lov / lov_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.gnu.org/licenses/gpl-2.0.html
19  *
20  * GPL HEADER END
21  */
22 /*
23  * Copyright (c) 2008, 2010, Oracle and/or its affiliates. All rights reserved.
24  * Use is subject to license terms.
25  *
26  * Copyright (c) 2011, 2017, Intel Corporation.
27  */
28 /*
29  * This file is part of Lustre, http://www.lustre.org/
30  * Lustre is a trademark of Sun Microsystems, Inc.
31  *
32  * Implementation of cl_lock for LOV layer.
33  *
34  *   Author: Nikita Danilov <nikita.danilov@sun.com>
35  */
36
37 #define DEBUG_SUBSYSTEM S_LOV
38
39 #include "lov_cl_internal.h"
40
41 /** \addtogroup lov
42  *  @{
43  */
44
45 /*****************************************************************************
46  *
47  * Lov lock operations.
48  *
49  */
50
51 static struct lov_sublock_env *lov_sublock_env_get(const struct lu_env *env,
52                                                    const struct cl_lock *parent,
53                                                    struct lov_lock_sub *lls)
54 {
55         struct lov_sublock_env *subenv;
56         struct lov_io          *lio    = lov_env_io(env);
57         struct cl_io           *io     = lio->lis_cl.cis_io;
58         struct lov_io_sub      *sub;
59
60         subenv = &lov_env_session(env)->ls_subenv;
61
62         /*
63          * FIXME: We tend to use the subio's env & io to call the sublock
64          * lock operations because osc lock sometimes stores some control
65          * variables in thread's IO infomation(Now only lockless information).
66          * However, if the lock's host(object) is different from the object
67          * for current IO, we have no way to get the subenv and subio because
68          * they are not initialized at all. As a temp fix, in this case,
69          * we still borrow the parent's env to call sublock operations.
70          */
71         if (!io || !cl_object_same(io->ci_obj, parent->cll_descr.cld_obj)) {
72                 subenv->lse_env = env;
73                 subenv->lse_io = io;
74         } else {
75                 sub = lov_sub_get(env, lio, lls->sub_index);
76                 if (!IS_ERR(sub)) {
77                         subenv->lse_env = sub->sub_env;
78                         subenv->lse_io  = &sub->sub_io;
79                 } else {
80                         subenv = (void *)sub;
81                 }
82         }
83         return subenv;
84 }
85
86 static int lov_sublock_init(const struct lu_env *env,
87                             const struct cl_lock *parent,
88                             struct lov_lock_sub *lls)
89 {
90         struct lov_sublock_env *subenv;
91         int result;
92         ENTRY;
93
94         subenv = lov_sublock_env_get(env, parent, lls);
95         if (!IS_ERR(subenv)) {
96                 result = cl_lock_init(subenv->lse_env, &lls->sub_lock,
97                                       subenv->lse_io);
98         } else {
99                 /* error occurs. */
100                 result = PTR_ERR(subenv);
101         }
102         RETURN(result);
103 }
104
105 /**
106  * Creates sub-locks for a given lov_lock for the first time.
107  *
108  * Goes through all sub-objects of top-object, and creates sub-locks on every
109  * sub-object intersecting with top-lock extent. This is complicated by the
110  * fact that top-lock (that is being created) can be accessed concurrently
111  * through already created sub-locks (possibly shared with other top-locks).
112  */
113 static struct lov_lock *lov_lock_sub_init(const struct lu_env *env,
114                                           const struct cl_object *obj,
115                                           struct cl_lock *lock)
116 {
117         struct lov_object *lov = cl2lov(obj);
118         struct lov_lock *lovlck;
119         struct lu_extent ext;
120         loff_t start;
121         loff_t end;
122         int result = 0;
123         int i;
124         int index;
125         int nr;
126
127         ENTRY;
128
129         ext.e_start = cl_offset(obj, lock->cll_descr.cld_start);
130         if (lock->cll_descr.cld_end == CL_PAGE_EOF)
131                 ext.e_end = OBD_OBJECT_EOF;
132         else
133                 ext.e_end  = cl_offset(obj, lock->cll_descr.cld_end + 1);
134
135         nr = 0;
136         lov_foreach_io_layout(index, lov_env_io(env), &ext) {
137                 struct lov_layout_raid0 *r0 = lov_r0(lov, index);
138
139                 for (i = 0; i < r0->lo_nr; i++) {
140                         if (likely(r0->lo_sub[i] != NULL) && /* spare layout */
141                             lov_stripe_intersects(lov->lo_lsm, index, i,
142                                                   &ext, &start, &end))
143                                 nr++;
144                 }
145         }
146         /**
147          * Aggressive lock request (from cl_setattr_ost) which asks for
148          * [eof, -1) lock, could come across uninstantiated layout extent,
149          * hence a 0 nr is possible.
150          */
151
152         OBD_ALLOC_LARGE(lovlck, offsetof(struct lov_lock, lls_sub[nr]));
153         if (lovlck == NULL)
154                 RETURN(ERR_PTR(-ENOMEM));
155
156         lovlck->lls_nr = nr;
157         nr = 0;
158         lov_foreach_io_layout(index, lov_env_io(env), &ext) {
159                 struct lov_layout_raid0 *r0 = lov_r0(lov, index);
160
161                 for (i = 0; i < r0->lo_nr; ++i) {
162                         struct lov_lock_sub *lls = &lovlck->lls_sub[nr];
163                         struct cl_lock_descr *descr = &lls->sub_lock.cll_descr;
164
165                         if (unlikely(r0->lo_sub[i] == NULL) ||
166                             !lov_stripe_intersects(lov->lo_lsm, index, i,
167                                                    &ext, &start, &end))
168                                 continue;
169
170                         LASSERT(descr->cld_obj == NULL);
171                         descr->cld_obj   = lovsub2cl(r0->lo_sub[i]);
172                         descr->cld_start = cl_index(descr->cld_obj, start);
173                         descr->cld_end   = cl_index(descr->cld_obj, end);
174                         descr->cld_mode  = lock->cll_descr.cld_mode;
175                         descr->cld_gid   = lock->cll_descr.cld_gid;
176                         descr->cld_enq_flags = lock->cll_descr.cld_enq_flags;
177
178                         lls->sub_index = lov_comp_index(index, i);
179
180                         /* initialize sub lock */
181                         result = lov_sublock_init(env, lock, lls);
182                         if (result < 0)
183                                 break;
184
185                         lls->sub_initialized = 1;
186                         nr++;
187                 }
188         }
189         LASSERT(ergo(result == 0, nr == lovlck->lls_nr));
190
191         if (result != 0) {
192                 for (i = 0; i < nr; ++i) {
193                         if (!lovlck->lls_sub[i].sub_initialized)
194                                 break;
195
196                         cl_lock_fini(env, &lovlck->lls_sub[i].sub_lock);
197                 }
198
199                 OBD_FREE_LARGE(lovlck,
200                                 offsetof(struct lov_lock, lls_sub[nr]));
201                 lovlck = ERR_PTR(result);
202         }
203
204         RETURN(lovlck);
205 }
206
207 static void lov_lock_fini(const struct lu_env *env,
208                           struct cl_lock_slice *slice)
209 {
210         struct lov_lock *lovlck;
211         int i;
212
213         ENTRY;
214         lovlck = cl2lov_lock(slice);
215         for (i = 0; i < lovlck->lls_nr; ++i) {
216                 LASSERT(!lovlck->lls_sub[i].sub_is_enqueued);
217                 if (lovlck->lls_sub[i].sub_initialized)
218                         cl_lock_fini(env, &lovlck->lls_sub[i].sub_lock);
219         }
220         OBD_FREE_LARGE(lovlck,
221                        offsetof(struct lov_lock, lls_sub[lovlck->lls_nr]));
222         EXIT;
223 }
224
225 /**
226  * Implementation of cl_lock_operations::clo_enqueue() for lov layer. This
227  * function is rather subtle, as it enqueues top-lock (i.e., advances top-lock
228  * state machine from CLS_QUEUING to CLS_ENQUEUED states) by juggling sub-lock
229  * state machines in the face of sub-locks sharing (by multiple top-locks),
230  * and concurrent sub-lock cancellations.
231  */
232 static int lov_lock_enqueue(const struct lu_env *env,
233                             const struct cl_lock_slice *slice,
234                             struct cl_io *io, struct cl_sync_io *anchor)
235 {
236         struct cl_lock          *lock   = slice->cls_lock;
237         struct lov_lock         *lovlck = cl2lov_lock(slice);
238         int                     i;
239         int                     rc      = 0;
240
241         ENTRY;
242
243         for (i = 0; i < lovlck->lls_nr; ++i) {
244                 struct lov_lock_sub     *lls = &lovlck->lls_sub[i];
245                 struct lov_sublock_env  *subenv;
246
247                 subenv = lov_sublock_env_get(env, lock, lls);
248                 if (IS_ERR(subenv)) {
249                         rc = PTR_ERR(subenv);
250                         break;
251                 }
252
253                 rc = cl_lock_enqueue(subenv->lse_env, subenv->lse_io,
254                                      &lls->sub_lock, anchor);
255                 if (rc != 0)
256                         break;
257
258                 lls->sub_is_enqueued = 1;
259         }
260         RETURN(rc);
261 }
262
263 static void lov_lock_cancel(const struct lu_env *env,
264                             const struct cl_lock_slice *slice)
265 {
266         struct cl_lock  *lock   = slice->cls_lock;
267         struct lov_lock *lovlck = cl2lov_lock(slice);
268         int i;
269
270         ENTRY;
271
272         for (i = 0; i < lovlck->lls_nr; ++i) {
273                 struct lov_lock_sub     *lls = &lovlck->lls_sub[i];
274                 struct cl_lock          *sublock = &lls->sub_lock;
275                 struct lov_sublock_env  *subenv;
276
277                 if (!lls->sub_is_enqueued)
278                         continue;
279
280                 lls->sub_is_enqueued = 0;
281                 subenv = lov_sublock_env_get(env, lock, lls);
282                 if (!IS_ERR(subenv)) {
283                         cl_lock_cancel(subenv->lse_env, sublock);
284                 } else {
285                         CL_LOCK_DEBUG(D_ERROR, env, slice->cls_lock,
286                                       "lov_lock_cancel fails with %ld.\n",
287                                       PTR_ERR(subenv));
288                 }
289         }
290 }
291
292 static int lov_lock_print(const struct lu_env *env, void *cookie,
293                           lu_printer_t p, const struct cl_lock_slice *slice)
294 {
295         struct lov_lock *lck = cl2lov_lock(slice);
296         int              i;
297
298         (*p)(env, cookie, "%d\n", lck->lls_nr);
299         for (i = 0; i < lck->lls_nr; ++i) {
300                 struct lov_lock_sub *sub;
301
302                 sub = &lck->lls_sub[i];
303                 (*p)(env, cookie, "    %d %x: ", i, sub->sub_is_enqueued);
304                 cl_lock_print(env, cookie, p, &sub->sub_lock);
305         }
306         return 0;
307 }
308
309 static const struct cl_lock_operations lov_lock_ops = {
310         .clo_fini      = lov_lock_fini,
311         .clo_enqueue   = lov_lock_enqueue,
312         .clo_cancel    = lov_lock_cancel,
313         .clo_print     = lov_lock_print
314 };
315
316 int lov_lock_init_composite(const struct lu_env *env, struct cl_object *obj,
317                             struct cl_lock *lock, const struct cl_io *io)
318 {
319         struct lov_lock *lck;
320         int result = 0;
321
322         ENTRY;
323         lck = lov_lock_sub_init(env, obj, lock);
324         if (!IS_ERR(lck))
325                 cl_lock_slice_add(lock, &lck->lls_cl, obj, &lov_lock_ops);
326         else
327                 result = PTR_ERR(lck);
328         RETURN(result);
329 }
330
331 static void lov_empty_lock_fini(const struct lu_env *env,
332                                 struct cl_lock_slice *slice)
333 {
334         struct lov_lock *lck = cl2lov_lock(slice);
335         OBD_SLAB_FREE_PTR(lck, lov_lock_kmem);
336 }
337
338 static int lov_empty_lock_print(const struct lu_env *env, void *cookie,
339                         lu_printer_t p, const struct cl_lock_slice *slice)
340 {
341         (*p)(env, cookie, "empty\n");
342         return 0;
343 }
344
345 /* XXX: more methods will be added later. */
346 static const struct cl_lock_operations lov_empty_lock_ops = {
347         .clo_fini  = lov_empty_lock_fini,
348         .clo_print = lov_empty_lock_print
349 };
350
351 int lov_lock_init_empty(const struct lu_env *env, struct cl_object *obj,
352                         struct cl_lock *lock, const struct cl_io *io)
353 {
354         struct lov_lock *lck;
355         int result = -ENOMEM;
356
357         ENTRY;
358         OBD_SLAB_ALLOC_PTR_GFP(lck, lov_lock_kmem, GFP_NOFS);
359         if (lck != NULL) {
360                 cl_lock_slice_add(lock, &lck->lls_cl, obj, &lov_empty_lock_ops);
361                 result = 0;
362         }
363         RETURN(result);
364 }
365
366 /** @} lov */