Whamcloud - gitweb
LU-3285 lvbo: pass lock as parameter to lvbo_update()
[fs/lustre-release.git] / lustre / ofd / ofd_lvb.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) 2009, 2010, Oracle and/or its affiliates. All rights reserved.
24  * Use is subject to license terms.
25  *
26  * Copyright (c) 2012, 2014, 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  * lustre/ofd/ofd_lvb.c
33  *
34  * This file contains methods for OBD Filter Device (OFD)
35  * Lock Value Block (LVB) operations.
36  *
37  * LVB is special opaque (to LDLM) data that is associated with an LDLM lock
38  * and transferred from client to server and back. OFD LVBs are used to
39  * maintain current object size/times.
40  *
41  * Author: Andreas Dilger <andreas.dilger@intel.com>
42  * Author: Mikhail Pershin <mike.pershin@intel.com>
43  * Author: Alexey Zhuravlev <alexey.zhuravlev@intel.com>
44  */
45
46 #define DEBUG_SUBSYSTEM S_FILTER
47
48 #include <lustre_swab.h>
49 #include "ofd_internal.h"
50
51 /**
52  * Implementation of ldlm_valblock_ops::lvbo_free for OFD.
53  *
54  * This function frees allocated LVB data if it associated with the given
55  * LDLM resource.
56  *
57  * \param[in] res       LDLM resource
58  *
59  * \retval              0 on successful setup
60  * \retval              negative value on error
61  */
62 static int ofd_lvbo_free(struct ldlm_resource *res)
63 {
64         if (res->lr_lvb_data)
65                 OBD_FREE(res->lr_lvb_data, res->lr_lvb_len);
66
67         return 0;
68 }
69
70 /**
71  * Implementation of ldlm_valblock_ops::lvbo_init for OFD.
72  *
73  * This function allocates and initializes new LVB data for the given
74  * LDLM resource if it is not allocated yet. New LVB is filled with attributes
75  * of the object associated with that resource. Function does nothing if LVB
76  * for the given LDLM resource is allocated already.
77  *
78  * Called with res->lr_lvb_sem held.
79  *
80  * \param[in] lock      LDLM lock on resource
81  *
82  * \retval              0 on successful setup
83  * \retval              negative value on error
84  */
85 static int ofd_lvbo_init(struct ldlm_resource *res)
86 {
87         struct ost_lvb          *lvb;
88         struct ofd_device       *ofd;
89         struct ofd_object       *fo;
90         struct ofd_thread_info  *info;
91         struct lu_env            env;
92         int                      rc = 0;
93
94         ENTRY;
95
96         LASSERT(res);
97         LASSERT(mutex_is_locked(&res->lr_lvb_mutex));
98
99         if (res->lr_lvb_data != NULL)
100                 RETURN(0);
101
102         ofd = ldlm_res_to_ns(res)->ns_lvbp;
103         LASSERT(ofd != NULL);
104
105         if (OBD_FAIL_CHECK(OBD_FAIL_LDLM_OST_LVB))
106                 RETURN(-ENOMEM);
107
108         rc = lu_env_init(&env, LCT_DT_THREAD);
109         if (rc)
110                 RETURN(rc);
111
112         OBD_ALLOC_PTR(lvb);
113         if (lvb == NULL)
114                 GOTO(out_env, rc = -ENOMEM);
115
116         res->lr_lvb_data = lvb;
117         res->lr_lvb_len = sizeof(*lvb);
118
119         info = ofd_info_init(&env, NULL);
120         ost_fid_from_resid(&info->fti_fid, &res->lr_name,
121                            ofd->ofd_lut.lut_lsd.lsd_osd_index);
122         fo = ofd_object_find(&env, ofd, &info->fti_fid);
123         if (IS_ERR(fo))
124                 GOTO(out_lvb, rc = PTR_ERR(fo));
125
126         rc = ofd_attr_get(&env, fo, &info->fti_attr);
127         if (rc)
128                 GOTO(out_obj, rc);
129
130         lvb->lvb_size = info->fti_attr.la_size;
131         lvb->lvb_blocks = info->fti_attr.la_blocks;
132         lvb->lvb_mtime = info->fti_attr.la_mtime;
133         lvb->lvb_atime = info->fti_attr.la_atime;
134         lvb->lvb_ctime = info->fti_attr.la_ctime;
135
136         CDEBUG(D_DLMTRACE, "res: "DFID" initial lvb size: %llu, "
137                "mtime: %#llx, blocks: %#llx\n",
138                PFID(&info->fti_fid), lvb->lvb_size,
139                lvb->lvb_mtime, lvb->lvb_blocks);
140
141         EXIT;
142 out_obj:
143         ofd_object_put(&env, fo);
144 out_lvb:
145         if (rc != 0)
146                 OST_LVB_SET_ERR(lvb->lvb_blocks, rc);
147 out_env:
148         lu_env_fini(&env);
149         /* Don't free lvb data on lookup error */
150         return rc;
151 }
152
153 /**
154  * Implementation of ldlm_valblock_ops::lvbo_update for OFD.
155  *
156  * When a client generates a glimpse enqueue, it wants to get the current
157  * file size and updated attributes for a stat() type operation, but these
158  * attributes may be writeback cached on another client. The client with
159  * the DLM extent lock at the highest offset is asked for its current
160  * attributes via a glimpse callback on its extent lock, on the assumption
161  * that it has the highest file size and the newest timestamps. The timestamps
162  * are guaranteed to be correct if there is only a single writer on the file,
163  * but may be slightly inaccurate if there are multiple concurrent writers on
164  * the same object. In order to avoid race conditions between the glimpse AST
165  * and the client cancelling the lock, ofd_lvbo_update() also updates
166  * the attributes from the local object. If the last client hasn't done any
167  * writes yet, or has already written its data and cancelled its lock before
168  * it processed the glimpse, then the local inode will have more uptodate
169  * information.
170  *
171  * This is called in two ways:
172  *  \a req != NULL : called by the DLM itself after a glimpse callback
173  *  \a req == NULL : called by the OFD after a disk write
174  *
175  * \param[in] lock              LDLM lock
176  * \param[in] req               PTLRPC request
177  * \param[in] increase_only     don't allow LVB values to decrease
178  *
179  * \retval              0 on successful setup
180  * \retval              negative value on error
181  */
182 static int ofd_lvbo_update(struct ldlm_resource *res, struct ldlm_lock *lock,
183                            struct ptlrpc_request *req, int increase_only)
184 {
185         struct ofd_device       *ofd;
186         struct ofd_object       *fo;
187         struct ofd_thread_info  *info;
188         struct ost_lvb          *lvb;
189         struct lu_env            env;
190         int                      rc = 0;
191
192         ENTRY;
193
194         LASSERT(res != NULL);
195
196         ofd = ldlm_res_to_ns(res)->ns_lvbp;
197         LASSERT(ofd != NULL);
198
199         rc = lu_env_init(&env, LCT_DT_THREAD);
200         if (rc)
201                 RETURN(rc);
202
203         info = ofd_info_init(&env, NULL);
204         fid_extract_from_res_name(&info->fti_fid, &res->lr_name);
205
206         lvb = res->lr_lvb_data;
207         if (lvb == NULL) {
208                 CERROR("%s: no LVB data for "DFID"\n",
209                        ofd_name(ofd), PFID(&info->fti_fid));
210                 GOTO(out_env, rc = 0);
211         }
212
213         /* Update the LVB from the network message */
214         if (req != NULL) {
215                 struct ost_lvb *rpc_lvb;
216                 bool lvb_type;
217
218                 if (req->rq_import != NULL)
219                         lvb_type = imp_connect_lvb_type(req->rq_import);
220                 else
221                         lvb_type = exp_connect_lvb_type(req->rq_export);
222
223                 if (!lvb_type) {
224                         struct ost_lvb_v1 *lvb_v1;
225
226                         lvb_v1 = req_capsule_server_swab_get(&req->rq_pill,
227                                         &RMF_DLM_LVB, lustre_swab_ost_lvb_v1);
228                         if (lvb_v1 == NULL)
229                                 goto disk_update;
230
231                         rpc_lvb = &info->fti_lvb;
232                         memcpy(rpc_lvb, lvb_v1, sizeof *lvb_v1);
233                         rpc_lvb->lvb_mtime_ns = 0;
234                         rpc_lvb->lvb_atime_ns = 0;
235                         rpc_lvb->lvb_ctime_ns = 0;
236                 } else {
237                         rpc_lvb = req_capsule_server_swab_get(&req->rq_pill,
238                                                               &RMF_DLM_LVB,
239                                                         lustre_swab_ost_lvb);
240                         if (rpc_lvb == NULL)
241                                 goto disk_update;
242                 }
243
244                 lock_res(res);
245                 if (rpc_lvb->lvb_size > lvb->lvb_size || !increase_only) {
246                         CDEBUG(D_DLMTRACE, "res: "DFID" updating lvb size: "
247                                "%llu -> %llu\n", PFID(&info->fti_fid),
248                                lvb->lvb_size, rpc_lvb->lvb_size);
249                         lvb->lvb_size = rpc_lvb->lvb_size;
250                 }
251                 if (rpc_lvb->lvb_mtime > lvb->lvb_mtime || !increase_only) {
252                         CDEBUG(D_DLMTRACE, "res: "DFID" updating lvb mtime: "
253                                "%llu -> %llu\n", PFID(&info->fti_fid),
254                                lvb->lvb_mtime, rpc_lvb->lvb_mtime);
255                         lvb->lvb_mtime = rpc_lvb->lvb_mtime;
256                 }
257                 if (rpc_lvb->lvb_atime > lvb->lvb_atime || !increase_only) {
258                         CDEBUG(D_DLMTRACE, "res: "DFID" updating lvb atime: "
259                                "%llu -> %llu\n", PFID(&info->fti_fid),
260                                lvb->lvb_atime, rpc_lvb->lvb_atime);
261                         lvb->lvb_atime = rpc_lvb->lvb_atime;
262                 }
263                 if (rpc_lvb->lvb_ctime > lvb->lvb_ctime || !increase_only) {
264                         CDEBUG(D_DLMTRACE, "res: "DFID" updating lvb ctime: "
265                                "%llu -> %llu\n", PFID(&info->fti_fid),
266                                lvb->lvb_ctime, rpc_lvb->lvb_ctime);
267                         lvb->lvb_ctime = rpc_lvb->lvb_ctime;
268                 }
269                 if (rpc_lvb->lvb_blocks > lvb->lvb_blocks || !increase_only) {
270                         CDEBUG(D_DLMTRACE, "res: "DFID" updating lvb blocks: "
271                                "%llu -> %llu\n", PFID(&info->fti_fid),
272                                lvb->lvb_blocks, rpc_lvb->lvb_blocks);
273                         lvb->lvb_blocks = rpc_lvb->lvb_blocks;
274                 }
275                 unlock_res(res);
276         }
277
278 disk_update:
279         /* Update the LVB from the disk inode */
280         ost_fid_from_resid(&info->fti_fid, &res->lr_name,
281                            ofd->ofd_lut.lut_lsd.lsd_osd_index);
282         fo = ofd_object_find(&env, ofd, &info->fti_fid);
283         if (IS_ERR(fo))
284                 GOTO(out_env, rc = PTR_ERR(fo));
285
286         rc = ofd_attr_get(&env, fo, &info->fti_attr);
287         if (rc)
288                 GOTO(out_obj, rc);
289
290         lock_res(res);
291         if (info->fti_attr.la_size > lvb->lvb_size || !increase_only) {
292                 CDEBUG(D_DLMTRACE, "res: "DFID" updating lvb size from disk: "
293                        "%llu -> %llu\n", PFID(&info->fti_fid),
294                        lvb->lvb_size, info->fti_attr.la_size);
295                 lvb->lvb_size = info->fti_attr.la_size;
296         }
297
298         if (info->fti_attr.la_mtime >lvb->lvb_mtime || !increase_only) {
299                 CDEBUG(D_DLMTRACE, "res: "DFID" updating lvb mtime from disk: "
300                        "%llu -> %llu\n", PFID(&info->fti_fid),
301                        lvb->lvb_mtime, info->fti_attr.la_mtime);
302                 lvb->lvb_mtime = info->fti_attr.la_mtime;
303         }
304         if (info->fti_attr.la_atime >lvb->lvb_atime || !increase_only) {
305                 CDEBUG(D_DLMTRACE, "res: "DFID" updating lvb atime from disk: "
306                        "%llu -> %llu\n", PFID(&info->fti_fid),
307                        lvb->lvb_atime, info->fti_attr.la_atime);
308                 lvb->lvb_atime = info->fti_attr.la_atime;
309         }
310         if (info->fti_attr.la_ctime >lvb->lvb_ctime || !increase_only) {
311                 CDEBUG(D_DLMTRACE, "res: "DFID" updating lvb ctime from disk: "
312                        "%llu -> %llu\n", PFID(&info->fti_fid),
313                        lvb->lvb_ctime, info->fti_attr.la_ctime);
314                 lvb->lvb_ctime = info->fti_attr.la_ctime;
315         }
316         if (info->fti_attr.la_blocks > lvb->lvb_blocks || !increase_only) {
317                 CDEBUG(D_DLMTRACE, "res: "DFID" updating lvb blocks from disk: "
318                        "%llu -> %llu\n", PFID(&info->fti_fid), lvb->lvb_blocks,
319                        (unsigned long long)info->fti_attr.la_blocks);
320                 lvb->lvb_blocks = info->fti_attr.la_blocks;
321         }
322         unlock_res(res);
323
324 out_obj:
325         ofd_object_put(&env, fo);
326 out_env:
327         lu_env_fini(&env);
328         return rc;
329 }
330
331 /**
332  * Implementation of ldlm_valblock_ops::lvbo_size for OFD.
333  *
334  * This function returns size of LVB data so appropriate RPC size will be
335  * reserved. This is used for compatibility needs between server and client
336  * of different Lustre versions.
337  *
338  * \param[in] lock      LDLM lock
339  *
340  * \retval              size of LVB data
341  */
342 static int ofd_lvbo_size(struct ldlm_lock *lock)
343 {
344         if (lock->l_export != NULL && exp_connect_lvb_type(lock->l_export))
345                 return sizeof(struct ost_lvb);
346         else
347                 return sizeof(struct ost_lvb_v1);
348 }
349
350 /**
351  * Implementation of ldlm_valblock_ops::lvbo_fill for OFD.
352  *
353  * This function is called to fill the given RPC buffer \a buf with LVB data
354  *
355  * \param[in] lock      LDLM lock
356  * \param[in] buf       RPC buffer to fill
357  * \param[in] buflen    buffer length
358  *
359  * \retval              size of LVB data written into \a buf buffer
360  */
361 static int ofd_lvbo_fill(struct ldlm_lock *lock, void *buf, int buflen)
362 {
363         struct ldlm_resource *res = lock->l_resource;
364         int lvb_len;
365
366         /* Former lvbo_init not allocate the "LVB". */
367         if (unlikely(res->lr_lvb_len == 0))
368                 return 0;
369
370         lvb_len = ofd_lvbo_size(lock);
371         LASSERT(lvb_len <= res->lr_lvb_len);
372
373         if (lvb_len > buflen)
374                 lvb_len = buflen;
375
376         lock_res(res);
377         memcpy(buf, res->lr_lvb_data, lvb_len);
378         unlock_res(res);
379
380         return lvb_len;
381 }
382
383 struct ldlm_valblock_ops ofd_lvbo = {
384         .lvbo_init      = ofd_lvbo_init,
385         .lvbo_update    = ofd_lvbo_update,
386         .lvbo_free      = ofd_lvbo_free,
387         .lvbo_size      = ofd_lvbo_size,
388         .lvbo_fill      = ofd_lvbo_fill
389 };