Whamcloud - gitweb
LU-12616 obclass: fix MDS start/stop race
[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.gnu.org/licenses/gpl-2.0.html
19  *
20  * GPL HEADER END
21  */
22 /*
23  * Copyright (c) 2007, 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  * lustre/fld/fld_index.c
33  *
34  * Author: WangDi <wangdi@clusterfs.com>
35  * Author: Yury Umanets <umka@clusterfs.com>
36  */
37
38 #define DEBUG_SUBSYSTEM S_FLD
39
40 #include <libcfs/libcfs.h>
41 #include <linux/module.h>
42 #include <obd_support.h>
43 #include <dt_object.h>
44 #include <lustre_fid.h>
45 #include <lustre_fld.h>
46 #include "fld_internal.h"
47
48 static const char fld_index_name[] = "fld";
49
50 static const struct lu_seq_range IGIF_FLD_RANGE = {
51         .lsr_start = FID_SEQ_IGIF,
52         .lsr_end   = FID_SEQ_IGIF_MAX + 1,
53         .lsr_index = 0,
54         .lsr_flags = LU_SEQ_RANGE_MDT
55 };
56
57 static const struct lu_seq_range DOT_LUSTRE_FLD_RANGE = {
58         .lsr_start = FID_SEQ_DOT_LUSTRE,
59         .lsr_end   = FID_SEQ_DOT_LUSTRE + 1,
60         .lsr_index = 0,
61         .lsr_flags = LU_SEQ_RANGE_MDT
62 };
63
64 static const struct lu_seq_range ROOT_FLD_RANGE = {
65         .lsr_start = FID_SEQ_ROOT,
66         .lsr_end   = FID_SEQ_ROOT + 1,
67         .lsr_index = 0,
68         .lsr_flags = LU_SEQ_RANGE_MDT
69 };
70
71 static const struct dt_index_features fld_index_features = {
72         .dif_flags       = DT_IND_UPDATE,
73         .dif_keysize_min = sizeof(u64),
74         .dif_keysize_max = sizeof(u64),
75         .dif_recsize_min = sizeof(struct lu_seq_range),
76         .dif_recsize_max = sizeof(struct lu_seq_range),
77         .dif_ptrsize     = 4
78 };
79
80 extern struct lu_context_key fld_thread_key;
81
82 int fld_declare_index_create(const struct lu_env *env,
83                              struct lu_server_fld *fld,
84                              const struct lu_seq_range *new_range,
85                              struct thandle *th)
86 {
87         struct lu_seq_range *tmp;
88         struct lu_seq_range *range;
89         struct fld_thread_info *info;
90         int rc = 0;
91
92         ENTRY;
93
94         info = lu_context_key_get(&env->le_ctx, &fld_thread_key);
95         range = &info->fti_lrange;
96         tmp = &info->fti_irange;
97         memset(range, 0, sizeof(*range));
98
99         rc = fld_index_lookup(env, fld, new_range->lsr_start, range);
100         if (rc == 0) {
101                 /* In case of duplicate entry, the location must be same */
102                 LASSERT((lu_seq_range_compare_loc(new_range, range) == 0));
103                 GOTO(out, rc = -EEXIST);
104         }
105
106         if (rc != -ENOENT) {
107                 CERROR("%s: lookup range "DRANGE" error: rc = %d\n",
108                         fld->lsf_name, PRANGE(range), rc);
109                 GOTO(out, rc);
110         }
111
112         /*
113          * Check for merge case, since the fld entry can only be increamental,
114          * so we will only check whether it can be merged from the left.
115          */
116         if (new_range->lsr_start == range->lsr_end && range->lsr_end != 0 &&
117             lu_seq_range_compare_loc(new_range, range) == 0) {
118                 range_cpu_to_be(tmp, range);
119                 rc = dt_declare_delete(env, fld->lsf_obj,
120                                        (struct dt_key *)&tmp->lsr_start, th);
121                 if (rc) {
122                         CERROR("%s: declare record "DRANGE" failed: rc = %d\n",
123                                fld->lsf_name, PRANGE(range), rc);
124                         GOTO(out, rc);
125                 }
126                 *tmp = *new_range;
127                 tmp->lsr_start = range->lsr_start;
128         } else {
129                 *tmp = *new_range;
130         }
131
132         range_cpu_to_be(tmp, tmp);
133         rc = dt_declare_insert(env, fld->lsf_obj, (struct dt_rec *)tmp,
134                                (struct dt_key *)&tmp->lsr_start, th);
135 out:
136         RETURN(rc);
137 }
138
139 /**
140  * insert range in fld store.
141  *
142  *      \param  range  range to be inserted
143  *      \param  th     transaction for this operation as it could compound
144  *                     transaction.
145  *
146  *      \retval  0  success
147  *      \retval  -ve error
148  *
149  * The whole fld index insertion is protected by seq->lss_mutex (see
150  * seq_server_alloc_super), i.e. only one thread will access fldb each
151  * time, so we do not need worry the fld file and cache will being
152  * changed between declare and create.
153  * Because the fld entry can only be increamental, so we will only check
154  * whether it can be merged from the left.
155  *
156  * Caller must hold fld->lsf_lock
157  **/
158 int fld_index_create(const struct lu_env *env, struct lu_server_fld *fld,
159                      const struct lu_seq_range *new_range, struct thandle *th)
160 {
161         struct lu_seq_range *range;
162         struct lu_seq_range *tmp;
163         struct fld_thread_info *info;
164         int rc = 0;
165         int deleted = 0;
166         struct fld_cache_entry *flde;
167
168         ENTRY;
169
170         info = lu_context_key_get(&env->le_ctx, &fld_thread_key);
171
172         LASSERT(mutex_is_locked(&fld->lsf_lock));
173
174         range = &info->fti_lrange;
175         memset(range, 0, sizeof(*range));
176         tmp = &info->fti_irange;
177         rc = fld_index_lookup(env, fld, new_range->lsr_start, range);
178         if (rc != -ENOENT) {
179                 rc = rc == 0 ? -EEXIST : rc;
180                 GOTO(out, rc);
181         }
182
183         if (new_range->lsr_start == range->lsr_end && range->lsr_end != 0 &&
184             lu_seq_range_compare_loc(new_range, range) == 0) {
185                 range_cpu_to_be(tmp, range);
186                 rc = dt_delete(env, fld->lsf_obj,
187                                (struct dt_key *)&tmp->lsr_start, th);
188                 if (rc != 0)
189                         GOTO(out, rc);
190                 *tmp = *new_range;
191                 tmp->lsr_start = range->lsr_start;
192                 deleted = 1;
193         } else {
194                 *tmp = *new_range;
195         }
196
197         range_cpu_to_be(tmp, tmp);
198         rc = dt_insert(env, fld->lsf_obj, (struct dt_rec *)tmp,
199                        (struct dt_key *)&tmp->lsr_start, th);
200         if (rc != 0) {
201                 CERROR("%s: insert range "DRANGE" failed: rc = %d\n",
202                        fld->lsf_name, PRANGE(new_range), rc);
203                 GOTO(out, rc);
204         }
205
206         flde = fld_cache_entry_create(new_range);
207         if (IS_ERR(flde))
208                 GOTO(out, rc = PTR_ERR(flde));
209
210         write_lock(&fld->lsf_cache->fci_lock);
211         if (deleted)
212                 fld_cache_delete_nolock(fld->lsf_cache, new_range);
213         rc = fld_cache_insert_nolock(fld->lsf_cache, flde);
214         write_unlock(&fld->lsf_cache->fci_lock);
215         if (rc)
216                 OBD_FREE_PTR(flde);
217 out:
218         RETURN(rc);
219 }
220
221 /**
222  * lookup range for a seq passed. note here we only care about the start/end,
223  * caller should handle the attached location data (flags, index).
224  *
225  * \param  seq     seq for lookup.
226  * \param  range   result of lookup.
227  *
228  * \retval  0           found, \a range is the matched range;
229  * \retval -ENOENT      not found, \a range is the left-side range;
230  * \retval  -ve         other error;
231  */
232 int fld_index_lookup(const struct lu_env *env, struct lu_server_fld *fld,
233                      u64 seq, struct lu_seq_range *range)
234 {
235         struct lu_seq_range *fld_rec;
236         struct fld_thread_info *info;
237         int rc;
238
239         ENTRY;
240
241         info = lu_context_key_get(&env->le_ctx, &fld_thread_key);
242         fld_rec = &info->fti_rec;
243
244         rc = fld_cache_lookup(fld->lsf_cache, seq, fld_rec);
245         if (rc == 0) {
246                 *range = *fld_rec;
247                 if (lu_seq_range_within(range, seq))
248                         rc = 0;
249                 else
250                         rc = -ENOENT;
251         }
252
253         CDEBUG(D_INFO, "%s: lookup seq = %#llx range : "DRANGE" rc = %d\n",
254                fld->lsf_name, seq, PRANGE(range), rc);
255
256         RETURN(rc);
257 }
258
259 /**
260  * insert entry in fld store.
261  *
262  * \param  env    relevant lu_env
263  * \param  fld    fld store
264  * \param  range  range to be inserted
265  *
266  * \retval  0  success
267  * \retval  -ve error
268  *
269  * Caller must hold fld->lsf_lock
270  **/
271
272 int fld_insert_entry(const struct lu_env *env,
273                      struct lu_server_fld *fld,
274                      const struct lu_seq_range *range)
275 {
276         struct thandle *th;
277         struct dt_device *dt = lu2dt_dev(fld->lsf_obj->do_lu.lo_dev);
278         int rc;
279
280         ENTRY;
281
282         LASSERT(mutex_is_locked(&fld->lsf_lock));
283
284         if (dt->dd_rdonly)
285                 RETURN(0);
286
287         th = dt_trans_create(env, dt);
288         if (IS_ERR(th))
289                 RETURN(PTR_ERR(th));
290
291         rc = fld_declare_index_create(env, fld, range, th);
292         if (rc != 0) {
293                 if (rc == -EEXIST)
294                         rc = 0;
295                 GOTO(out, rc);
296         }
297
298         rc = dt_trans_start_local(env, dt, th);
299         if (rc)
300                 GOTO(out, rc);
301
302         rc = fld_index_create(env, fld, range, th);
303         if (rc == -EEXIST)
304                 rc = 0;
305 out:
306         dt_trans_stop(env, dt, th);
307         RETURN(rc);
308 }
309 EXPORT_SYMBOL(fld_insert_entry);
310
311 static int fld_insert_special_entries(const struct lu_env *env,
312                                       struct lu_server_fld *fld)
313 {
314         int rc;
315
316         rc = fld_insert_entry(env, fld, &IGIF_FLD_RANGE);
317         if (rc != 0)
318                 RETURN(rc);
319
320         rc = fld_insert_entry(env, fld, &DOT_LUSTRE_FLD_RANGE);
321         if (rc != 0)
322                 RETURN(rc);
323
324         rc = fld_insert_entry(env, fld, &ROOT_FLD_RANGE);
325
326         RETURN(rc);
327 }
328
329 int fld_index_init(const struct lu_env *env, struct lu_server_fld *fld,
330                    struct dt_device *dt, int type)
331 {
332         struct dt_object *dt_obj = NULL;
333         struct lu_fid fid;
334         struct lu_attr *attr = NULL;
335         struct lu_seq_range *range = NULL;
336         struct fld_thread_info *info;
337         struct dt_object_format dof;
338         struct dt_it *it;
339         const struct dt_it_ops *iops;
340         int rc;
341         u32 index;
342         int range_count = 0;
343
344         ENTRY;
345
346         info = lu_context_key_get(&env->le_ctx, &fld_thread_key);
347         LASSERT(info != NULL);
348
349         lu_local_obj_fid(&fid, FLD_INDEX_OID);
350         OBD_ALLOC_PTR(attr);
351         if (!attr)
352                 RETURN(-ENOMEM);
353
354         memset(attr, 0, sizeof(*attr));
355         attr->la_valid = LA_MODE;
356         attr->la_mode = S_IFREG | 0666;
357         dof.dof_type = DFT_INDEX;
358         dof.u.dof_idx.di_feat = &fld_index_features;
359
360         dt_obj = dt_locate(env, dt, &fid);
361         if (IS_ERR(dt_obj)) {
362                 rc = PTR_ERR(dt_obj);
363                 dt_obj = NULL;
364                 GOTO(out, rc);
365         }
366
367         LASSERT(dt_obj != NULL);
368         if (!dt_object_exists(dt_obj)) {
369                 dt_object_put(env, dt_obj);
370                 dt_obj = dt_find_or_create(env, dt, &fid, &dof, attr);
371                 fld->lsf_new = 1;
372                 if (IS_ERR(dt_obj)) {
373                         rc = PTR_ERR(dt_obj);
374                         CERROR("%s: Can't find \"%s\" obj %d\n", fld->lsf_name,
375                                 fld_index_name, rc);
376                         dt_obj = NULL;
377                         GOTO(out, rc);
378                 }
379         }
380
381         fld->lsf_obj = dt_obj;
382         rc = dt_obj->do_ops->do_index_try(env, dt_obj, &fld_index_features);
383         if (rc != 0) {
384                 CERROR("%s: File \"%s\" is not an index: rc = %d!\n",
385                        fld->lsf_name, fld_index_name, rc);
386                 GOTO(out, rc);
387         }
388
389         range = &info->fti_rec;
390         /* Load fld entry to cache */
391         iops = &dt_obj->do_index_ops->dio_it;
392         it = iops->init(env, dt_obj, 0);
393         if (IS_ERR(it))
394                 GOTO(out, rc = PTR_ERR(it));
395
396         rc = iops->load(env, it, 0);
397         if (rc > 0)
398                 rc = 0;
399         else if (rc == 0)
400                 rc = iops->next(env, it);
401
402         if (rc < 0)
403                 GOTO(out_it_fini, rc);
404
405         while (rc == 0) {
406                 rc = iops->rec(env, it, (struct dt_rec *)range, 0);
407                 if (rc != 0)
408                         GOTO(out_it_put, rc);
409
410                 range_be_to_cpu(range, range);
411
412                 /*
413                  * Newly created ldiskfs IAM indexes may include a
414                  * zeroed-out key and record. Ignore it here.
415                  */
416                 if (range->lsr_start < range->lsr_end) {
417                         rc = fld_cache_insert(fld->lsf_cache, range);
418                         if (rc != 0)
419                                 GOTO(out_it_put, rc);
420
421                         range_count++;
422                 }
423
424                 rc = iops->next(env, it);
425                 if (rc < 0)
426                         GOTO(out_it_fini, rc);
427         }
428
429         if (range_count == 0)
430                 fld->lsf_new = 1;
431
432         rc = fld_name_to_index(fld->lsf_name, &index);
433         if (rc < 0)
434                 GOTO(out_it_put, rc);
435         else
436                 rc = 0;
437
438         if (index == 0 && type == LU_SEQ_RANGE_MDT) {
439                 /*
440                  * Note: fld_insert_entry will detect whether these
441                  * special entries already exist inside FLDB
442                  */
443                 mutex_lock(&fld->lsf_lock);
444                 rc = fld_insert_special_entries(env, fld);
445                 mutex_unlock(&fld->lsf_lock);
446                 if (rc != 0) {
447                         CERROR("%s: insert special entries failed!: rc = %d\n",
448                                fld->lsf_name, rc);
449                         GOTO(out_it_put, rc);
450                 }
451         }
452 out_it_put:
453         iops->put(env, it);
454 out_it_fini:
455         iops->fini(env, it);
456 out:
457         if (attr)
458                 OBD_FREE_PTR(attr);
459
460         if (rc < 0) {
461                 if (dt_obj)
462                         dt_object_put(env, dt_obj);
463                 fld->lsf_obj = NULL;
464         }
465         RETURN(rc);
466 }
467
468 void fld_index_fini(const struct lu_env *env, struct lu_server_fld *fld)
469 {
470         ENTRY;
471         if (fld->lsf_obj) {
472                 if (!IS_ERR(fld->lsf_obj))
473                         dt_object_put(env, fld->lsf_obj);
474                 fld->lsf_obj = NULL;
475         }
476         EXIT;
477 }
478
479 int fld_server_read(const struct lu_env *env, struct lu_server_fld *fld,
480                     struct lu_seq_range *range, void *data, int data_len)
481 {
482         struct lu_seq_range_array *lsra = data;
483         struct fld_thread_info *info;
484         struct dt_object *dt_obj = fld->lsf_obj;
485         struct lu_seq_range *entry;
486         struct dt_it *it;
487         const struct dt_it_ops *iops;
488         int rc;
489
490         ENTRY;
491
492         lsra->lsra_count = 0;
493         iops = &dt_obj->do_index_ops->dio_it;
494         it = iops->init(env, dt_obj, 0);
495         if (IS_ERR(it))
496                 RETURN(PTR_ERR(it));
497
498         rc = iops->load(env, it, range->lsr_end);
499         if (rc <= 0)
500                 GOTO(out_it_fini, rc);
501
502         info = lu_context_key_get(&env->le_ctx, &fld_thread_key);
503         LASSERT(info != NULL);
504         entry = &info->fti_rec;
505         do {
506                 rc = iops->rec(env, it, (struct dt_rec *)entry, 0);
507                 if (rc != 0)
508                         GOTO(out_it_put, rc);
509
510                 if (offsetof(typeof(*lsra), lsra_lsr[lsra->lsra_count + 1]) >
511                     data_len)
512                         GOTO(out, rc = -EAGAIN);
513
514                 range_be_to_cpu(entry, entry);
515                 if (entry->lsr_index == range->lsr_index &&
516                     entry->lsr_flags == range->lsr_flags &&
517                     entry->lsr_start > range->lsr_start) {
518                         lsra->lsra_lsr[lsra->lsra_count] = *entry;
519                         lsra->lsra_count++;
520                 }
521
522                 rc = iops->next(env, it);
523         } while (rc == 0);
524         if (rc > 0)
525                 rc = 0;
526 out:
527         range_array_cpu_to_le(lsra, lsra);
528 out_it_put:
529         iops->put(env, it);
530 out_it_fini:
531         iops->fini(env, it);
532
533         RETURN(rc);
534 }