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 }