Whamcloud - gitweb
LU-14627 lnet: Ensure ref taken when queueing for discovery
[fs/lustre-release.git] / lustre / obdclass / cl_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, 2014, Intel Corporation.
27  */
28 /*
29  * This file is part of Lustre, http://www.lustre.org/
30  *
31  * Client Extent Lock.
32  *
33  *   Author: Nikita Danilov <nikita.danilov@sun.com>
34  *   Author: Jinshan Xiong <jinshan.xiong@intel.com>
35  */
36
37 #define DEBUG_SUBSYSTEM S_CLASS
38
39 #include <linux/list.h>
40 #include <libcfs/libcfs.h>
41 #include <obd_class.h>
42 #include <obd_support.h>
43 #include <lustre_fid.h>
44 #include <cl_object.h>
45 #include "cl_internal.h"
46
47 static void cl_lock_trace0(int level, const struct lu_env *env,
48                            const char *prefix, const struct cl_lock *lock,
49                            const char *func, const int line)
50 {
51         struct cl_object_header *h = cl_object_header(lock->cll_descr.cld_obj);
52         CDEBUG(level, "%s: %p (%p/%d) at %s():%d\n",
53                prefix, lock, env, h->coh_nesting, func, line);
54 }
55 #define cl_lock_trace(level, env, prefix, lock)                         \
56         cl_lock_trace0(level, env, prefix, lock, __FUNCTION__, __LINE__)
57
58 /**
59  * Adds lock slice to the compound lock.
60  *
61  * This is called by cl_object_operations::coo_lock_init() methods to add a
62  * per-layer state to the lock. New state is added at the end of
63  * cl_lock::cll_layers list, that is, it is at the bottom of the stack.
64  *
65  * \see cl_req_slice_add(), cl_page_slice_add(), cl_io_slice_add()
66  */
67 void cl_lock_slice_add(struct cl_lock *lock, struct cl_lock_slice *slice,
68                        struct cl_object *obj,
69                        const struct cl_lock_operations *ops)
70 {
71         ENTRY;
72         slice->cls_lock = lock;
73         list_add_tail(&slice->cls_linkage, &lock->cll_layers);
74         slice->cls_obj = obj;
75         slice->cls_ops = ops;
76         EXIT;
77 }
78 EXPORT_SYMBOL(cl_lock_slice_add);
79
80 void cl_lock_fini(const struct lu_env *env, struct cl_lock *lock)
81 {
82         ENTRY;
83
84         cl_lock_trace(D_DLMTRACE, env, "destroy lock", lock);
85
86         while (!list_empty(&lock->cll_layers)) {
87                 struct cl_lock_slice *slice;
88
89                 slice = list_entry(lock->cll_layers.next,
90                                 struct cl_lock_slice, cls_linkage);
91                 list_del_init(lock->cll_layers.next);
92                 slice->cls_ops->clo_fini(env, slice);
93         }
94         POISON(lock, 0x5a, sizeof(*lock));
95         EXIT;
96 }
97 EXPORT_SYMBOL(cl_lock_fini);
98
99 int cl_lock_init(const struct lu_env *env, struct cl_lock *lock,
100                  const struct cl_io *io)
101 {
102         struct cl_object *obj = lock->cll_descr.cld_obj;
103         struct cl_object *scan;
104         int result = 0;
105         ENTRY;
106
107         /* Make sure cl_lock::cll_descr is initialized. */
108         LASSERT(obj != NULL);
109
110         INIT_LIST_HEAD(&lock->cll_layers);
111         cl_object_for_each(scan, obj) {
112                 if (scan->co_ops->coo_lock_init != NULL)
113                         result = scan->co_ops->coo_lock_init(env, scan, lock,
114                                                              io);
115
116                 if (result != 0) {
117                         cl_lock_fini(env, lock);
118                         break;
119                 }
120         }
121         RETURN(result);
122 }
123 EXPORT_SYMBOL(cl_lock_init);
124
125 /**
126  * Returns a slice with a lock, corresponding to the given layer in the
127  * device stack.
128  *
129  * \see cl_page_at()
130  */
131 const struct cl_lock_slice *cl_lock_at(const struct cl_lock *lock,
132                                        const struct lu_device_type *dtype)
133 {
134         const struct cl_lock_slice *slice;
135
136         ENTRY;
137
138         list_for_each_entry(slice, &lock->cll_layers, cls_linkage) {
139                 if (slice->cls_obj->co_lu.lo_dev->ld_type == dtype)
140                         RETURN(slice);
141         }
142         RETURN(NULL);
143 }
144 EXPORT_SYMBOL(cl_lock_at);
145
146 void cl_lock_cancel(const struct lu_env *env, struct cl_lock *lock)
147 {
148         const struct cl_lock_slice *slice;
149         ENTRY;
150
151         cl_lock_trace(D_DLMTRACE, env, "cancel lock", lock);
152         list_for_each_entry_reverse(slice, &lock->cll_layers, cls_linkage) {
153                 if (slice->cls_ops->clo_cancel != NULL)
154                         slice->cls_ops->clo_cancel(env, slice);
155         }
156
157         EXIT;
158 }
159 EXPORT_SYMBOL(cl_lock_cancel);
160
161 /**
162  * Enqueue a lock.
163  * \param anchor: if we need to wait for resources before getting the lock,
164  *                use @anchor for the purpose.
165  * \retval 0  enqueue successfully
166  * \retval <0 error code
167  */
168 int cl_lock_enqueue(const struct lu_env *env, struct cl_io *io,
169                     struct cl_lock *lock, struct cl_sync_io *anchor)
170 {
171         const struct cl_lock_slice *slice;
172         int rc = 0;
173
174         ENTRY;
175
176         list_for_each_entry(slice, &lock->cll_layers, cls_linkage) {
177                 if (slice->cls_ops->clo_enqueue == NULL)
178                         continue;
179
180                 rc = slice->cls_ops->clo_enqueue(env, slice, io, anchor);
181                 if (rc != 0)
182                         break;
183         }
184         RETURN(rc);
185 }
186 EXPORT_SYMBOL(cl_lock_enqueue);
187
188 /**
189  * Main high-level entry point of cl_lock interface that finds existing or
190  * enqueues new lock matching given description.
191  */
192 int cl_lock_request(const struct lu_env *env, struct cl_io *io,
193                     struct cl_lock *lock)
194 {
195         struct cl_sync_io       *anchor = NULL;
196         __u32                   enq_flags = lock->cll_descr.cld_enq_flags;
197         int                     rc;
198         ENTRY;
199
200         rc = cl_lock_init(env, lock, io);
201         if (rc < 0)
202                 RETURN(rc);
203
204         if ((enq_flags & CEF_GLIMPSE) && !(enq_flags & CEF_SPECULATIVE)) {
205                 anchor = &cl_env_info(env)->clt_anchor;
206                 cl_sync_io_init(anchor, 1);
207         }
208
209         rc = cl_lock_enqueue(env, io, lock, anchor);
210
211         if (anchor != NULL) {
212                 int rc2;
213
214                 /* drop the reference count held at initialization time */
215                 cl_sync_io_note(env, anchor, 0);
216                 rc2 = cl_sync_io_wait(env, anchor, 0);
217                 if (rc2 < 0 && rc == 0)
218                         rc = rc2;
219         }
220
221         if (rc < 0)
222                 cl_lock_release(env, lock);
223         RETURN(rc);
224 }
225 EXPORT_SYMBOL(cl_lock_request);
226
227 /**
228  * Releases a hold and a reference on a lock, obtained by cl_lock_hold().
229  */
230 void cl_lock_release(const struct lu_env *env, struct cl_lock *lock)
231 {
232         ENTRY;
233
234         cl_lock_trace(D_DLMTRACE, env, "release lock", lock);
235         cl_lock_cancel(env, lock);
236         cl_lock_fini(env, lock);
237         EXIT;
238 }
239 EXPORT_SYMBOL(cl_lock_release);
240
241 const char *cl_lock_mode_name(const enum cl_lock_mode mode)
242 {
243         static const char * const names[] = {
244                 [CLM_READ]    = "R",
245                 [CLM_WRITE]   = "W",
246                 [CLM_GROUP]   = "G"
247         };
248         BUILD_BUG_ON(CLM_MAX != ARRAY_SIZE(names));
249         return names[mode];
250 }
251 EXPORT_SYMBOL(cl_lock_mode_name);
252
253 /**
254  * Prints human readable representation of a lock description.
255  */
256 void cl_lock_descr_print(const struct lu_env *env, void *cookie,
257                          lu_printer_t printer,
258                          const struct cl_lock_descr *descr)
259 {
260         const struct lu_fid  *fid;
261
262         fid = lu_object_fid(&descr->cld_obj->co_lu);
263         (*printer)(env, cookie, DDESCR"@"DFID, PDESCR(descr), PFID(fid));
264 }
265 EXPORT_SYMBOL(cl_lock_descr_print);
266
267 /**
268  * Prints human readable representation of \a lock to the \a f.
269  */
270 void cl_lock_print(const struct lu_env *env, void *cookie,
271                    lu_printer_t printer, const struct cl_lock *lock)
272 {
273         const struct cl_lock_slice *slice;
274
275         (*printer)(env, cookie, "lock@%p", lock);
276         cl_lock_descr_print(env, cookie, printer, &lock->cll_descr);
277         (*printer)(env, cookie, " {\n");
278
279         list_for_each_entry(slice, &lock->cll_layers, cls_linkage) {
280                 (*printer)(env, cookie, "    %s@%p: ",
281                            slice->cls_obj->co_lu.lo_dev->ld_type->ldt_name,
282                            slice);
283                 if (slice->cls_ops->clo_print != NULL)
284                         slice->cls_ops->clo_print(env, cookie, printer, slice);
285                 (*printer)(env, cookie, "\n");
286         }
287         (*printer)(env, cookie, "} lock@%p\n", lock);
288 }
289 EXPORT_SYMBOL(cl_lock_print);