Whamcloud - gitweb
LU-1445 osd: add fld look up in osd.
[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, 2013, Intel Corporation.
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,
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 int fld_declare_index_create(const struct lu_env *env,
86                              struct lu_server_fld *fld,
87                              const struct lu_seq_range *new_range,
88                              struct thandle *th)
89 {
90         struct lu_seq_range     *tmp;
91         struct lu_seq_range     *range;
92         struct fld_thread_info  *info;
93         int                     rc = 0;
94
95         ENTRY;
96
97         info = lu_context_key_get(&env->le_ctx, &fld_thread_key);
98         range = &info->fti_lrange;
99         tmp = &info->fti_irange;
100         memset(range, 0, sizeof(*range));
101
102         rc = fld_index_lookup(env, fld, new_range->lsr_start, range);
103         if (rc == 0) {
104                 /* In case of duplicate entry, the location must be same */
105                 LASSERT((range_compare_loc(new_range, range) == 0));
106                 GOTO(out, rc = -EEXIST);
107         }
108
109         if (rc != -ENOENT) {
110                 CERROR("%s: lookup range "DRANGE" error: rc = %d\n",
111                         fld->lsf_name, PRANGE(range), rc);
112                 GOTO(out, rc);
113         }
114
115         /* Check for merge case, since the fld entry can only be increamental,
116          * so we will only check whether it can be merged from the left. */
117         if (new_range->lsr_start == range->lsr_end && range->lsr_end != 0 &&
118             range_compare_loc(new_range, range) == 0) {
119                 range_cpu_to_be(tmp, range);
120                 rc = dt_declare_delete(env, fld->lsf_obj,
121                                        (struct dt_key *)&tmp->lsr_start, th);
122                 if (rc) {
123                         CERROR("%s: declare record "DRANGE" failed: rc = %d\n",
124                                fld->lsf_name, PRANGE(range), rc);
125                         GOTO(out, rc);
126                 }
127                 memcpy(tmp, new_range, sizeof(*new_range));
128                 tmp->lsr_start = range->lsr_start;
129         } else {
130                 memcpy(tmp, new_range, sizeof(*new_range));
131         }
132
133         range_cpu_to_be(tmp, tmp);
134         rc = dt_declare_insert(env, fld->lsf_obj, (struct dt_rec *)tmp,
135                                (struct dt_key *)&tmp->lsr_start, th);
136 out:
137         RETURN(rc);
138 }
139
140 /**
141  * insert range in fld store.
142  *
143  *      \param  range  range to be inserted
144  *      \param  th     transaction for this operation as it could compound
145  *                     transaction.
146  *
147  *      \retval  0  success
148  *      \retval  -ve error
149  *
150  * The whole fld index insertion is protected by seq->lss_mutex (see
151  * seq_server_alloc_super), i.e. only one thread will access fldb each
152  * time, so we do not need worry the fld file and cache will being
153  * changed between declare and create.
154  * Because the fld entry can only be increamental, so we will only check
155  * whether it can be merged from the left.
156  **/
157 int fld_index_create(const struct lu_env *env, struct lu_server_fld *fld,
158                      const struct lu_seq_range *new_range, struct thandle *th)
159 {
160         struct lu_seq_range     *range;
161         struct lu_seq_range     *tmp;
162         struct fld_thread_info  *info;
163         int                     rc = 0;
164         int                     deleted = 0;
165         struct fld_cache_entry  *flde;
166         ENTRY;
167
168         info = lu_context_key_get(&env->le_ctx, &fld_thread_key);
169
170         LASSERT_MUTEX_LOCKED(&fld->lsf_lock);
171
172         range = &info->fti_lrange;
173         memset(range, 0, sizeof(*range));
174         tmp = &info->fti_irange;
175         rc = fld_index_lookup(env, fld, new_range->lsr_start, range);
176         if (rc != -ENOENT) {
177                 rc = rc == 0 ? -EEXIST : rc;
178                 GOTO(out, rc);
179         }
180
181         if (new_range->lsr_start == range->lsr_end && range->lsr_end != 0 &&
182             range_compare_loc(new_range, range) == 0) {
183                 range_cpu_to_be(tmp, range);
184                 rc = dt_delete(env, fld->lsf_obj,
185                                (struct dt_key *)&tmp->lsr_start, th,
186                                 BYPASS_CAPA);
187                 if (rc != 0)
188                         GOTO(out, rc);
189                 memcpy(tmp, new_range, sizeof(*new_range));
190                 tmp->lsr_start = range->lsr_start;
191                 deleted = 1;
192         } else {
193                 memcpy(tmp, new_range, sizeof(*new_range));
194         }
195
196         range_cpu_to_be(tmp, tmp);
197         rc = dt_insert(env, fld->lsf_obj, (struct dt_rec *)tmp,
198                        (struct dt_key *)&tmp->lsr_start, th, BYPASS_CAPA, 1);
199         if (rc != 0) {
200                 CERROR("%s: insert range "DRANGE" failed: rc = %d\n",
201                        fld->lsf_name, PRANGE(new_range), rc);
202                 GOTO(out, rc);
203         }
204
205         flde = fld_cache_entry_create(new_range);
206         if (IS_ERR(flde))
207                 GOTO(out, rc = PTR_ERR(flde));
208
209         write_lock(&fld->lsf_cache->fci_lock);
210         if (deleted)
211                 fld_cache_delete_nolock(fld->lsf_cache, new_range);
212         rc = fld_cache_insert_nolock(fld->lsf_cache, flde);
213         write_unlock(&fld->lsf_cache->fci_lock);
214         if (rc)
215                 OBD_FREE_PTR(flde);
216 out:
217         RETURN(rc);
218 }
219
220 /**
221  * lookup range for a seq passed. note here we only care about the start/end,
222  * caller should handle the attached location data (flags, index).
223  *
224  * \param  seq     seq for lookup.
225  * \param  range   result of lookup.
226  *
227  * \retval  0           found, \a range is the matched range;
228  * \retval -ENOENT      not found, \a range is the left-side range;
229  * \retval  -ve         other error;
230  */
231 int fld_index_lookup(const struct lu_env *env, struct lu_server_fld *fld,
232                      seqno_t seq, struct lu_seq_range *range)
233 {
234         struct lu_seq_range     *fld_rec;
235         struct fld_thread_info  *info;
236         int rc;
237
238         ENTRY;
239
240         info = lu_context_key_get(&env->le_ctx, &fld_thread_key);
241         fld_rec = &info->fti_rec;
242
243         rc = fld_cache_lookup(fld->lsf_cache, seq, fld_rec);
244         if (rc == 0) {
245                 *range = *fld_rec;
246                 if (range_within(range, seq))
247                         rc = 0;
248                 else
249                         rc = -ENOENT;
250         }
251
252         CDEBUG(D_INFO, "%s: lookup seq = "LPX64" range : "DRANGE" rc = %d\n",
253                fld->lsf_name, seq, PRANGE(range), rc);
254
255         RETURN(rc);
256 }
257
258 static int fld_insert_igif_fld(struct lu_server_fld *fld,
259                                const struct lu_env *env)
260 {
261         struct thandle *th;
262         int rc;
263         ENTRY;
264
265         th = dt_trans_create(env, lu2dt_dev(fld->lsf_obj->do_lu.lo_dev));
266         if (IS_ERR(th))
267                 RETURN(PTR_ERR(th));
268
269         rc = fld_declare_index_create(env, fld, &IGIF_FLD_RANGE, th);
270         if (rc != 0) {
271                 if (rc == -EEXIST)
272                         rc = 0;
273                 GOTO(out, rc);
274         }
275
276         rc = dt_trans_start_local(env, lu2dt_dev(fld->lsf_obj->do_lu.lo_dev),
277                                   th);
278         if (rc)
279                 GOTO(out, rc);
280
281         rc = fld_index_create(env, fld, &IGIF_FLD_RANGE, th);
282         if (rc == -EEXIST)
283                 rc = 0;
284 out:
285         dt_trans_stop(env, lu2dt_dev(fld->lsf_obj->do_lu.lo_dev), th);
286         RETURN(rc);
287 }
288
289 int fld_index_init(const struct lu_env *env, struct lu_server_fld *fld,
290                    struct dt_device *dt)
291 {
292         struct dt_object        *dt_obj = NULL;
293         struct lu_fid           fid;
294         struct lu_attr          *attr = NULL;
295         struct lu_seq_range     *range = NULL;
296         struct fld_thread_info  *info;
297         struct dt_object_format dof;
298         struct dt_it            *it;
299         const struct dt_it_ops  *iops;
300         int                     rc;
301         ENTRY;
302
303         info = lu_context_key_get(&env->le_ctx, &fld_thread_key);
304         LASSERT(info != NULL);
305
306         lu_local_obj_fid(&fid, FLD_INDEX_OID);
307         OBD_ALLOC_PTR(attr);
308         if (attr == NULL)
309                 RETURN(-ENOMEM);
310
311         memset(attr, 0, sizeof(attr));
312         attr->la_valid = LA_MODE;
313         attr->la_mode = S_IFREG | 0666;
314         dof.dof_type = DFT_INDEX;
315         dof.u.dof_idx.di_feat = &fld_index_features;
316
317         dt_obj = dt_find_or_create(env, dt, &fid, &dof, attr);
318         if (IS_ERR(dt_obj)) {
319                 rc = PTR_ERR(dt_obj);
320                 CERROR("%s: Can't find \"%s\" obj %d\n", fld->lsf_name,
321                         fld_index_name, rc);
322                 dt_obj = NULL;
323                 GOTO(out, rc);
324         }
325
326         fld->lsf_obj = dt_obj;
327         rc = dt_obj->do_ops->do_index_try(env, dt_obj, &fld_index_features);
328         if (rc == 0) {
329                 LASSERT(dt_obj->do_index_ops != NULL);
330                 mutex_lock(&fld->lsf_lock);
331                 rc = fld_insert_igif_fld(fld, env);
332                 mutex_unlock(&fld->lsf_lock);
333                 if (rc != 0) {
334                         CERROR("insert igif in fld! = %d\n", rc);
335                         GOTO(out, rc);
336                 }
337         } else {
338                 CERROR("%s: File \"%s\" is not an index: rc = %d!\n",
339                        fld->lsf_name, fld_index_name, rc);
340                 GOTO(out, rc);
341         }
342
343         range = &info->fti_rec;
344         /* Load fld entry to cache */
345         iops = &dt_obj->do_index_ops->dio_it;
346         it = iops->init(env, dt_obj, 0, NULL);
347         if (IS_ERR(it))
348                 GOTO(out, rc = PTR_ERR(it));
349
350         rc = iops->load(env, it, 0);
351         if (rc < 0)
352                 GOTO(out_it_fini, rc);
353
354         do {
355                 rc = iops->rec(env, it, (struct dt_rec *)range, 0);
356                 if (rc != 0)
357                         GOTO(out_it_fini, rc);
358
359                 LASSERT(range != NULL);
360                 range_be_to_cpu(range, range);
361                 rc = fld_cache_insert(fld->lsf_cache, range);
362                 if (rc != 0)
363                         GOTO(out_it_fini, rc);
364                 rc = iops->next(env, it);
365
366         } while (rc == 0);
367         rc = 0;
368
369 out_it_fini:
370         iops->fini(env, it);
371 out:
372         if (attr != NULL)
373                 OBD_FREE_PTR(attr);
374
375         if (rc != 0) {
376                 if (dt_obj != NULL)
377                         lu_object_put(env, &dt_obj->do_lu);
378                 fld->lsf_obj = NULL;
379         }
380         RETURN(rc);
381 }
382
383 void fld_index_fini(const struct lu_env *env, struct lu_server_fld *fld)
384 {
385         ENTRY;
386         if (fld->lsf_obj != NULL) {
387                 if (!IS_ERR(fld->lsf_obj))
388                         lu_object_put(env, &fld->lsf_obj->do_lu);
389                 fld->lsf_obj = NULL;
390         }
391         EXIT;
392 }