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         /* Check for merge case, since the fld entry can only be increamental,
 113          * so we will only check whether it can be merged from the left. */
 114         if (new_range->lsr_start == range->lsr_end && range->lsr_end != 0 &&
 115             lu_seq_range_compare_loc(new_range, range) == 0) {
 116                 range_cpu_to_be(tmp, range);
 117                 rc = dt_declare_delete(env, fld->lsf_obj,
 118                                        (struct dt_key *)&tmp->lsr_start, th);
 119                 if (rc) {
 120                         CERROR("%s: declare record "DRANGE" failed: rc = %d\n",
 121                                fld->lsf_name, PRANGE(range), rc);
 122                         GOTO(out, rc);
 123                 }
 124                 *tmp = *new_range;
 125                 tmp->lsr_start = range->lsr_start;
 126         } else {
 127                 *tmp = *new_range;
 128         }
 129
 130         range_cpu_to_be(tmp, tmp);
 131         rc = dt_declare_insert(env, fld->lsf_obj, (struct dt_rec *)tmp,
 132                                (struct dt_key *)&tmp->lsr_start, th);
 133 out:
 134         RETURN(rc);
 135 }
 136
 137 /**
 138  * insert range in fld store.
 139  *
 140  *      \param  range  range to be inserted
 141  *      \param  th     transaction for this operation as it could compound
 142  *                     transaction.
 143  *
 144  *      \retval  0  success
 145  *      \retval  -ve error
 146  *
 147  * The whole fld index insertion is protected by seq->lss_mutex (see
 148  * seq_server_alloc_super), i.e. only one thread will access fldb each
 149  * time, so we do not need worry the fld file and cache will being
 150  * changed between declare and create.
 151  * Because the fld entry can only be increamental, so we will only check
 152  * whether it can be merged from the left.
 153  *
 154  * Caller must hold fld->lsf_lock
 155  **/
 156 int fld_index_create(const struct lu_env *env, struct lu_server_fld *fld,
 157                      const struct lu_seq_range *new_range, struct thandle *th)
 158 {
 159         struct lu_seq_range     *range;
 160         struct lu_seq_range     *tmp;
 161         struct fld_thread_info  *info;
 162         int                     rc = 0;
 163         int                     deleted = 0;
 164         struct fld_cache_entry  *flde;
 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, 1);
 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         ENTRY;
 277
 278         LASSERT(mutex_is_locked(&fld->lsf_lock));
 279
 280         if (dt->dd_rdonly)
 281                 RETURN(0);
 282
 283         th = dt_trans_create(env, dt);
 284         if (IS_ERR(th))
 285                 RETURN(PTR_ERR(th));
 286
 287         rc = fld_declare_index_create(env, fld, range, th);
 288         if (rc != 0) {
 289                 if (rc == -EEXIST)
 290                         rc = 0;
 291                 GOTO(out, rc);
 292         }
 293
 294         rc = dt_trans_start_local(env, dt, th);
 295         if (rc)
 296                 GOTO(out, rc);
 297
 298         rc = fld_index_create(env, fld, range, th);
 299         if (rc == -EEXIST)
 300                 rc = 0;
 301 out:
 302         dt_trans_stop(env, dt, th);
 303         RETURN(rc);
 304 }
 305 EXPORT_SYMBOL(fld_insert_entry);
 306
 307 static int fld_insert_special_entries(const struct lu_env *env,
 308                                       struct lu_server_fld *fld)
 309 {
 310         int rc;
 311
 312         rc = fld_insert_entry(env, fld, &IGIF_FLD_RANGE);
 313         if (rc != 0)
 314                 RETURN(rc);
 315
 316         rc = fld_insert_entry(env, fld, &DOT_LUSTRE_FLD_RANGE);
 317         if (rc != 0)
 318                 RETURN(rc);
 319
 320         rc = fld_insert_entry(env, fld, &ROOT_FLD_RANGE);
 321
 322         RETURN(rc);
 323 }
 324
 325 int fld_index_init(const struct lu_env *env, struct lu_server_fld *fld,
 326                    struct dt_device *dt, int type)
 327 {
 328         struct dt_object        *dt_obj = NULL;
 329         struct lu_fid           fid;
 330         struct lu_attr          *attr = NULL;
 331         struct lu_seq_range     *range = NULL;
 332         struct fld_thread_info  *info;
 333         struct dt_object_format dof;
 334         struct dt_it            *it;
 335         const struct dt_it_ops  *iops;
 336         int                     rc;
 337         __u32                   index;
 338         int range_count = 0;
 339         ENTRY;
 340
 341         info = lu_context_key_get(&env->le_ctx, &fld_thread_key);
 342         LASSERT(info != NULL);
 343
 344         lu_local_obj_fid(&fid, FLD_INDEX_OID);
 345         OBD_ALLOC_PTR(attr);
 346         if (attr == NULL)
 347                 RETURN(-ENOMEM);
 348
 349         memset(attr, 0, sizeof(*attr));
 350         attr->la_valid = LA_MODE;
 351         attr->la_mode = S_IFREG | 0666;
 352         dof.dof_type = DFT_INDEX;
 353         dof.u.dof_idx.di_feat = &fld_index_features;
 354
 355         dt_obj = dt_locate(env, dt, &fid);
 356         if (IS_ERR(dt_obj)) {
 357                 rc = PTR_ERR(dt_obj);
 358                 dt_obj = NULL;
 359                 GOTO(out, rc);
 360         }
 361
 362         LASSERT(dt_obj != NULL);
 363         if (!dt_object_exists(dt_obj)) {
 364                 dt_object_put(env, dt_obj);
 365                 dt_obj = dt_find_or_create(env, dt, &fid, &dof, attr);
 366                 fld->lsf_new = 1;
 367                 if (IS_ERR(dt_obj)) {
 368                         rc = PTR_ERR(dt_obj);
 369                         CERROR("%s: Can't find \"%s\" obj %d\n", fld->lsf_name,
 370                                 fld_index_name, rc);
 371                         dt_obj = NULL;
 372                         GOTO(out, rc);
 373                 }
 374         }
 375
 376         fld->lsf_obj = dt_obj;
 377         rc = dt_obj->do_ops->do_index_try(env, dt_obj, &fld_index_features);
 378         if (rc != 0) {
 379                 CERROR("%s: File \"%s\" is not an index: rc = %d!\n",
 380                        fld->lsf_name, fld_index_name, rc);
 381                 GOTO(out, rc);
 382         }
 383
 384         range = &info->fti_rec;
 385         /* Load fld entry to cache */
 386         iops = &dt_obj->do_index_ops->dio_it;
 387         it = iops->init(env, dt_obj, 0);
 388         if (IS_ERR(it))
 389                 GOTO(out, rc = PTR_ERR(it));
 390
 391         rc = iops->load(env, it, 0);
 392         if (rc > 0)
 393                 rc = 0;
 394         else if (rc == 0)
 395                 rc = iops->next(env, it);
 396
 397         if (rc < 0)
 398                 GOTO(out_it_fini, rc);
 399
 400         while (rc == 0) {
 401                 rc = iops->rec(env, it, (struct dt_rec *)range, 0);
 402                 if (rc != 0)
 403                         GOTO(out_it_put, rc);
 404
 405                 range_be_to_cpu(range, range);
 406
 407                 /* Newly created ldiskfs IAM indexes may include a
 408                  * zeroed-out key and record. Ignore it here. */
 409                 if (range->lsr_start < range->lsr_end) {
 410                         rc = fld_cache_insert(fld->lsf_cache, range);
 411                         if (rc != 0)
 412                                 GOTO(out_it_put, rc);
 413
 414                         range_count++;
 415                 }
 416
 417                 rc = iops->next(env, it);
 418                 if (rc < 0)
 419                         GOTO(out_it_fini, rc);
 420         }
 421
 422         if (range_count == 0)
 423                 fld->lsf_new = 1;
 424
 425         rc = fld_name_to_index(fld->lsf_name, &index);
 426         if (rc < 0)
 427                 GOTO(out_it_put, rc);
 428         else
 429                 rc = 0;
 430
 431         if (index == 0 && type == LU_SEQ_RANGE_MDT) {
 432                 /* Note: fld_insert_entry will detect whether these
 433                  * special entries already exist inside FLDB */
 434                 mutex_lock(&fld->lsf_lock);
 435                 rc = fld_insert_special_entries(env, fld);
 436                 mutex_unlock(&fld->lsf_lock);
 437                 if (rc != 0) {
 438                         CERROR("%s: insert special entries failed!: rc = %d\n",
 439                                fld->lsf_name, rc);
 440                         GOTO(out_it_put, rc);
 441                 }
 442         }
 443 out_it_put:
 444         iops->put(env, it);
 445 out_it_fini:
 446         iops->fini(env, it);
 447 out:
 448         if (attr != NULL)
 449                 OBD_FREE_PTR(attr);
 450
 451         if (rc < 0) {
 452                 if (dt_obj != NULL)
 453                         dt_object_put(env, dt_obj);
 454                 fld->lsf_obj = NULL;
 455         }
 456         RETURN(rc);
 457 }
 458
 459 void fld_index_fini(const struct lu_env *env, struct lu_server_fld *fld)
 460 {
 461         ENTRY;
 462         if (fld->lsf_obj != NULL) {
 463                 if (!IS_ERR(fld->lsf_obj))
 464                         dt_object_put(env, fld->lsf_obj);
 465                 fld->lsf_obj = NULL;
 466         }
 467         EXIT;
 468 }
 469
 470 int fld_server_read(const struct lu_env *env, struct lu_server_fld *fld,
 471                     struct lu_seq_range *range, void *data, int data_len)
 472 {
 473         struct lu_seq_range_array *lsra = data;
 474         struct fld_thread_info    *info;
 475         struct dt_object          *dt_obj = fld->lsf_obj;
 476         struct lu_seq_range       *entry;
 477         struct dt_it              *it;
 478         const struct dt_it_ops    *iops;
 479         int                       rc;
 480
 481         ENTRY;
 482
 483         lsra->lsra_count = 0;
 484         iops = &dt_obj->do_index_ops->dio_it;
 485         it = iops->init(env, dt_obj, 0);
 486         if (IS_ERR(it))
 487                 RETURN(PTR_ERR(it));
 488
 489         rc = iops->load(env, it, range->lsr_end);
 490         if (rc <= 0)
 491                 GOTO(out_it_fini, rc);
 492
 493         info = lu_context_key_get(&env->le_ctx, &fld_thread_key);
 494         LASSERT(info != NULL);
 495         entry = &info->fti_rec;
 496         do {
 497                 rc = iops->rec(env, it, (struct dt_rec *)entry, 0);
 498                 if (rc != 0)
 499                         GOTO(out_it_put, rc);
 500
 501                 if (offsetof(typeof(*lsra), lsra_lsr[lsra->lsra_count + 1]) >
 502                     data_len)
 503                         GOTO(out, rc = -EAGAIN);
 504
 505                 range_be_to_cpu(entry, entry);
 506                 if (entry->lsr_index == range->lsr_index &&
 507                     entry->lsr_flags == range->lsr_flags &&
 508                     entry->lsr_start > range->lsr_start) {
 509                         lsra->lsra_lsr[lsra->lsra_count] = *entry;
 510                         lsra->lsra_count++;
 511                 }
 512
 513                 rc = iops->next(env, it);
 514         } while (rc == 0);
 515         if (rc > 0)
 516                 rc = 0;
 517 out:
 518         range_array_cpu_to_le(lsra, lsra);
 519 out_it_put:
 520         iops->put(env, it);
 521 out_it_fini:
 522         iops->fini(env, it);
 523
 524         RETURN(rc);
 525 }