Whamcloud - gitweb
LU-1943 fld: Simplify transaction handling in FID/FLD
[fs/lustre-release.git] / lustre / fld / fld_index.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) 2007, 2010, Oracle and/or its affiliates. All rights reserved.
28  * Use is subject to license terms.
29  *
30  * Copyright (c) 2011, 2012, 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  * lustre/fld/fld_index.c
37  *
38  * Author: WangDi <wangdi@clusterfs.com>
39  * Author: Yury Umanets <umka@clusterfs.com>
40  */
41
42 #define DEBUG_SUBSYSTEM S_FLD
43
44 #ifdef __KERNEL__
45 # include <libcfs/libcfs.h>
46 # include <linux/module.h>
47 # include <linux/jbd.h>
48 #else /* __KERNEL__ */
49 # include <liblustre.h>
50 #endif
51
52 #include <obd.h>
53 #include <obd_class.h>
54 #include <lustre_ver.h>
55 #include <obd_support.h>
56 #include <lprocfs_status.h>
57
58 #include <dt_object.h>
59 #include <md_object.h>
60 #include <lustre_mdc.h>
61 #include <lustre_fid.h>
62 #include <lustre_fld.h>
63 #include "fld_internal.h"
64
65 const char fld_index_name[] = "fld";
66
67 static const struct lu_seq_range IGIF_FLD_RANGE = {
68         .lsr_start = FID_SEQ_IGIF,
69         .lsr_end   = FID_SEQ_IGIF_MAX + 1,
70         .lsr_index = 0,
71         .lsr_flags = LU_SEQ_RANGE_MDT
72 };
73
74 const struct dt_index_features fld_index_features = {
75         .dif_flags       = DT_IND_UPDATE | DT_IND_RANGE,
76         .dif_keysize_min = sizeof(seqno_t),
77         .dif_keysize_max = sizeof(seqno_t),
78         .dif_recsize_min = sizeof(struct lu_seq_range),
79         .dif_recsize_max = sizeof(struct lu_seq_range),
80         .dif_ptrsize     = 4
81 };
82
83 extern struct lu_context_key fld_thread_key;
84
85 static struct dt_key *fld_key(const struct lu_env *env, const seqno_t seq)
86 {
87         struct fld_thread_info *info;
88         ENTRY;
89
90         info = lu_context_key_get(&env->le_ctx, &fld_thread_key);
91         LASSERT(info != NULL);
92
93         info->fti_key = cpu_to_be64(seq);
94         RETURN((void *)&info->fti_key);
95 }
96
97 static struct dt_rec *fld_rec(const struct lu_env *env,
98                               const struct lu_seq_range *range)
99 {
100         struct fld_thread_info *info;
101         struct lu_seq_range *rec;
102         ENTRY;
103
104         info = lu_context_key_get(&env->le_ctx, &fld_thread_key);
105         LASSERT(info != NULL);
106         rec = &info->fti_rec;
107
108         range_cpu_to_be(rec, range);
109         RETURN((void *)rec);
110 }
111
112 int fld_declare_index_create(struct lu_server_fld *fld,
113                              const struct lu_env *env,
114                              const struct lu_seq_range *range,
115                              struct thandle *th)
116 {
117         int rc;
118
119         ENTRY;
120
121         if (fld->lsf_no_range_lookup) {
122                 /* Stub for underlying FS which can't lookup ranges */
123                 return 0;
124         }
125
126         LASSERT(range_is_sane(range));
127
128         rc = dt_declare_insert(env, fld->lsf_obj, fld_rec(env, range),
129                               fld_key(env, range->lsr_start), th);
130         RETURN(rc);
131 }
132
133 /**
134  * insert range in fld store.
135  *
136  *      \param  range  range to be inserted
137  *      \param  th     transaction for this operation as it could compound
138  *                     transaction.
139  *
140  *      \retval  0  success
141  *      \retval  -ve error
142  */
143 int fld_index_create(struct lu_server_fld *fld,
144                      const struct lu_env *env,
145                      const struct lu_seq_range *range,
146                      struct thandle *th)
147 {
148         int rc;
149
150         ENTRY;
151
152         if (fld->lsf_no_range_lookup) {
153                 /* Stub for underlying FS which can't lookup ranges */
154                 if (range->lsr_index != 0) {
155                         CERROR("%s: FLD backend does not support range"
156                                "lookups, so DNE and FIDs-on-OST are not"
157                                "supported in this configuration\n",
158                                fld->lsf_name);
159                         return -EINVAL;
160                 }
161         }
162
163         LASSERT(range_is_sane(range));
164
165         rc = dt_insert(env, fld->lsf_obj, fld_rec(env, range),
166                        fld_key(env, range->lsr_start), th, BYPASS_CAPA, 1);
167         CDEBUG(D_INFO, "%s: insert given range : "DRANGE" rc = %d\n",
168                fld->lsf_name, PRANGE(range), rc);
169         RETURN(rc);
170 }
171
172 /**
173  * delete range in fld store.
174  *
175  *      \param  range range to be deleted
176  *      \param  th     transaction
177  *
178  *      \retval  0  success
179  *      \retval  -ve error
180  */
181 int fld_index_delete(struct lu_server_fld *fld,
182                      const struct lu_env *env,
183                      struct lu_seq_range *range,
184                      struct thandle   *th)
185 {
186         int rc;
187
188         ENTRY;
189
190         rc = dt_delete(env, fld->lsf_obj, fld_key(env, range->lsr_start), th,
191                        BYPASS_CAPA);
192         CDEBUG(D_INFO, "%s: delete given range : "DRANGE" rc = %d\n",
193                fld->lsf_name, PRANGE(range), rc);
194         RETURN(rc);
195 }
196
197 /**
198  * lookup range for a seq passed. note here we only care about the start/end,
199  * caller should handle the attached location data (flags, index).
200  *
201  * \param  seq     seq for lookup.
202  * \param  range   result of lookup.
203  *
204  * \retval  0           found, \a range is the matched range;
205  * \retval -ENOENT      not found, \a range is the left-side range;
206  * \retval  -ve         other error;
207  */
208
209 int fld_index_lookup(struct lu_server_fld *fld,
210                      const struct lu_env *env,
211                      seqno_t seq,
212                      struct lu_seq_range *range)
213 {
214         struct dt_object        *dt_obj = fld->lsf_obj;
215         struct lu_seq_range     *fld_rec;
216         struct dt_key           *key = fld_key(env, seq);
217         struct fld_thread_info  *info;
218         int rc;
219
220         ENTRY;
221
222         if (fld->lsf_no_range_lookup) {
223                 /* Stub for underlying FS which can't lookup ranges */
224                 range->lsr_start = 0;
225                 range->lsr_end = ~0;
226                 range->lsr_index = 0;
227                 range->lsr_flags = LU_SEQ_RANGE_MDT;
228
229                 range_cpu_to_be(range, range);
230                 return 0;
231         }
232
233         info = lu_context_key_get(&env->le_ctx, &fld_thread_key);
234         fld_rec = &info->fti_rec;
235
236         rc = dt_obj->do_index_ops->dio_lookup(env, dt_obj,
237                                               (struct dt_rec*) fld_rec,
238                                               key, BYPASS_CAPA);
239
240         if (rc >= 0) {
241                 range_be_to_cpu(fld_rec, fld_rec);
242                 *range = *fld_rec;
243                 if (range_within(range, seq))
244                         rc = 0;
245                 else
246                         rc = -ENOENT;
247         }
248
249         CDEBUG(D_INFO, "%s: lookup seq = "LPX64" range : "DRANGE" rc = %d\n",
250                fld->lsf_name, seq, PRANGE(range), rc);
251
252         RETURN(rc);
253 }
254
255 static int fld_insert_igif_fld(struct lu_server_fld *fld,
256                                const struct lu_env *env)
257 {
258         struct thandle *th;
259         int rc;
260         ENTRY;
261
262         th = dt_trans_create(env, lu2dt_dev(fld->lsf_obj->do_lu.lo_dev));
263         if (IS_ERR(th))
264                 RETURN(PTR_ERR(th));
265         rc = fld_declare_index_create(fld, env, &IGIF_FLD_RANGE, th);
266         if (rc)
267                 GOTO(out, rc);
268
269         rc = dt_trans_start_local(env, lu2dt_dev(fld->lsf_obj->do_lu.lo_dev),
270                                   th);
271         if (rc)
272                 GOTO(out, rc);
273
274         rc = fld_index_create(fld, env, &IGIF_FLD_RANGE, th);
275         if (rc == -EEXIST)
276                 rc = 0;
277 out:
278         dt_trans_stop(env, lu2dt_dev(fld->lsf_obj->do_lu.lo_dev), th);
279         RETURN(rc);
280 }
281
282 int fld_index_init(struct lu_server_fld *fld,
283                    const struct lu_env *env,
284                    struct dt_device *dt)
285 {
286         struct dt_object *dt_obj;
287         struct lu_fid fid;
288         struct lu_attr attr;
289         struct dt_object_format dof;
290         int rc;
291         ENTRY;
292
293         lu_local_obj_fid(&fid, FLD_INDEX_OID);
294
295         memset(&attr, 0, sizeof(attr));
296         attr.la_valid = LA_MODE;
297         attr.la_mode = S_IFREG | 0666;
298         dof.dof_type = DFT_INDEX;
299         dof.u.dof_idx.di_feat = &fld_index_features;
300
301         dt_obj = dt_find_or_create(env, dt, &fid, &dof, &attr);
302         if (!IS_ERR(dt_obj)) {
303                 fld->lsf_obj = dt_obj;
304                 rc = dt_obj->do_ops->do_index_try(env, dt_obj,
305                                                   &fld_index_features);
306                 if (rc == 0) {
307                         LASSERT(dt_obj->do_index_ops != NULL);
308                         rc = fld_insert_igif_fld(fld, env);
309
310                         if (rc != 0) {
311                                 CERROR("insert igif in fld! = %d\n", rc);
312                                 lu_object_put(env, &dt_obj->do_lu);
313                                 fld->lsf_obj = NULL;
314                         }
315                 } else if (rc == -ERANGE) {
316                         CWARN("%s: File \"%s\" doesn't support range lookup, "
317                               "using stub. DNE and FIDs on OST will not work "
318                               "with this backend\n",
319                               fld->lsf_name, fld_index_name);
320
321                         LASSERT(dt_obj->do_index_ops == NULL);
322                         fld->lsf_no_range_lookup = 1;
323                         rc = 0;
324                 } else {
325                         CERROR("%s: File \"%s\" is not index, rc %d!\n",
326                                fld->lsf_name, fld_index_name, rc);
327                         lu_object_put(env, &fld->lsf_obj->do_lu);
328                         fld->lsf_obj = NULL;
329                 }
330
331
332         } else {
333                 CERROR("%s: Can't find \"%s\" obj %d\n",
334                        fld->lsf_name, fld_index_name, (int)PTR_ERR(dt_obj));
335                 rc = PTR_ERR(dt_obj);
336         }
337
338         RETURN(rc);
339 }
340
341 void fld_index_fini(struct lu_server_fld *fld,
342                     const struct lu_env *env)
343 {
344         ENTRY;
345         if (fld->lsf_obj != NULL) {
346                 if (!IS_ERR(fld->lsf_obj))
347                         lu_object_put(env, &fld->lsf_obj->do_lu);
348                 fld->lsf_obj = NULL;
349         }
350         EXIT;
351 }