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