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 int fld_declare_index_create(const struct lu_env *env,
  81                              struct lu_server_fld *fld,
  82                              const struct lu_seq_range *new_range,
  83                              struct thandle *th)
  84 {
  85         struct lu_seq_range *tmp;
  86         struct lu_seq_range *range;
  87         struct fld_thread_info *info;
  88         int rc = 0;
  89
  90         ENTRY;
  91
  92         info = lu_context_key_get(&env->le_ctx, &fld_thread_key);
  93         range = &info->fti_lrange;
  94         tmp = &info->fti_irange;
  95         memset(range, 0, sizeof(*range));
  96
  97         rc = fld_index_lookup(env, fld, new_range->lsr_start, range);
  98         if (rc == 0) {
  99                 /* In case of duplicate entry, the location must be same */
 100                 LASSERT((lu_seq_range_compare_loc(new_range, range) == 0));
 101                 GOTO(out, rc = -EEXIST);
 102         }
 103
 104         if (rc != -ENOENT) {
 105                 CERROR("%s: lookup range "DRANGE" error: rc = %d\n",
 106                         fld->lsf_name, PRANGE(range), rc);
 107                 GOTO(out, rc);
 108         }
 109
 110         /*
 111          * Check for merge case, since the fld entry can only be increamental,
 112          * so we will only check whether it can be merged from the left.
 113          */
 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
 166         ENTRY;
 167
 168         info = lu_context_key_get(&env->le_ctx, &fld_thread_key);
 169
 170         LASSERT(mutex_is_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             lu_seq_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                 if (rc != 0)
 187                         GOTO(out, rc);
 188                 *tmp = *new_range;
 189                 tmp->lsr_start = range->lsr_start;
 190                 deleted = 1;
 191         } else {
 192                 *tmp = *new_range;
 193         }
 194
 195         range_cpu_to_be(tmp, tmp);
 196         rc = dt_insert(env, fld->lsf_obj, (struct dt_rec *)tmp,
 197                        (struct dt_key *)&tmp->lsr_start, th);
 198         if (rc != 0) {
 199                 CERROR("%s: insert range "DRANGE" failed: rc = %d\n",
 200                        fld->lsf_name, PRANGE(new_range), rc);
 201                 GOTO(out, rc);
 202         }
 203
 204         flde = fld_cache_entry_create(new_range);
 205         if (IS_ERR(flde))
 206                 GOTO(out, rc = PTR_ERR(flde));
 207
 208         write_lock(&fld->lsf_cache->fci_lock);
 209         if (deleted)
 210                 fld_cache_delete_nolock(fld->lsf_cache, new_range);
 211         rc = fld_cache_insert_nolock(fld->lsf_cache, flde);
 212         write_unlock(&fld->lsf_cache->fci_lock);
 213         if (rc)
 214                 OBD_FREE_PTR(flde);
 215 out:
 216         RETURN(rc);
 217 }
 218
 219 /**
 220  * lookup range for a seq passed. note here we only care about the start/end,
 221  * caller should handle the attached location data (flags, index).
 222  *
 223  * \param  seq     seq for lookup.
 224  * \param  range   result of lookup.
 225  *
 226  * \retval  0           found, \a range is the matched range;
 227  * \retval -ENOENT      not found, \a range is the left-side range;
 228  * \retval  -ve         other error;
 229  */
 230 int fld_index_lookup(const struct lu_env *env, struct lu_server_fld *fld,
 231                      u64 seq, struct lu_seq_range *range)
 232 {
 233         struct lu_seq_range *fld_rec;
 234         struct fld_thread_info *info;
 235         int rc;
 236
 237         ENTRY;
 238
 239         info = lu_context_key_get(&env->le_ctx, &fld_thread_key);
 240         fld_rec = &info->fti_rec;
 241
 242         rc = fld_cache_lookup(fld->lsf_cache, seq, fld_rec);
 243         if (rc == 0) {
 244                 *range = *fld_rec;
 245                 if (lu_seq_range_within(range, seq))
 246                         rc = 0;
 247                 else
 248                         rc = -ENOENT;
 249         }
 250
 251         CDEBUG(D_INFO, "%s: lookup seq = %#llx range : "DRANGE" rc = %d\n",
 252                fld->lsf_name, seq, PRANGE(range), rc);
 253
 254         RETURN(rc);
 255 }
 256
 257 /**
 258  * insert entry in fld store.
 259  *
 260  * \param  env    relevant lu_env
 261  * \param  fld    fld store
 262  * \param  range  range to be inserted
 263  *
 264  * \retval  0  success
 265  * \retval  -ve error
 266  *
 267  * Caller must hold fld->lsf_lock
 268  **/
 269
 270 int fld_insert_entry(const struct lu_env *env,
 271                      struct lu_server_fld *fld,
 272                      const struct lu_seq_range *range)
 273 {
 274         struct thandle *th;
 275         struct dt_device *dt = lu2dt_dev(fld->lsf_obj->do_lu.lo_dev);
 276         int rc;
 277
 278         ENTRY;
 279
 280         LASSERT(mutex_is_locked(&fld->lsf_lock));
 281
 282         if (dt->dd_rdonly)
 283                 RETURN(0);
 284
 285         th = dt_trans_create(env, dt);
 286         if (IS_ERR(th))
 287                 RETURN(PTR_ERR(th));
 288
 289         rc = fld_declare_index_create(env, fld, range, th);
 290         if (rc != 0) {
 291                 if (rc == -EEXIST)
 292                         rc = 0;
 293                 GOTO(out, rc);
 294         }
 295
 296         rc = dt_trans_start_local(env, dt, th);
 297         if (rc)
 298                 GOTO(out, rc);
 299
 300         rc = fld_index_create(env, fld, range, th);
 301         if (rc == -EEXIST)
 302                 rc = 0;
 303 out:
 304         dt_trans_stop(env, dt, th);
 305         RETURN(rc);
 306 }
 307 EXPORT_SYMBOL(fld_insert_entry);
 308
 309 static int fld_insert_special_entries(const struct lu_env *env,
 310                                       struct lu_server_fld *fld)
 311 {
 312         int rc;
 313
 314         rc = fld_insert_entry(env, fld, &IGIF_FLD_RANGE);
 315         if (rc != 0)
 316                 RETURN(rc);
 317
 318         rc = fld_insert_entry(env, fld, &DOT_LUSTRE_FLD_RANGE);
 319         if (rc != 0)
 320                 RETURN(rc);
 321
 322         rc = fld_insert_entry(env, fld, &ROOT_FLD_RANGE);
 323
 324         RETURN(rc);
 325 }
 326
 327 int fld_index_init(const struct lu_env *env, struct lu_server_fld *fld,
 328                    struct dt_device *dt, int type)
 329 {
 330         struct dt_object *dt_obj = NULL;
 331         struct lu_fid fid;
 332         struct lu_attr *attr = NULL;
 333         struct lu_seq_range *range = NULL;
 334         struct fld_thread_info *info;
 335         struct dt_object_format dof;
 336         struct dt_it *it;
 337         const struct dt_it_ops *iops;
 338         int rc;
 339         u32 index;
 340         int range_count = 0;
 341
 342         ENTRY;
 343
 344         info = lu_context_key_get(&env->le_ctx, &fld_thread_key);
 345         LASSERT(info != NULL);
 346
 347         lu_local_obj_fid(&fid, FLD_INDEX_OID);
 348         OBD_ALLOC_PTR(attr);
 349         if (!attr)
 350                 RETURN(-ENOMEM);
 351
 352         memset(attr, 0, sizeof(*attr));
 353         attr->la_valid = LA_MODE;
 354         attr->la_mode = S_IFREG | 0666;
 355         dof.dof_type = DFT_INDEX;
 356         dof.u.dof_idx.di_feat = &fld_index_features;
 357
 358         dt_obj = dt_locate(env, dt, &fid);
 359         if (IS_ERR(dt_obj)) {
 360                 rc = PTR_ERR(dt_obj);
 361                 dt_obj = NULL;
 362                 GOTO(out, rc);
 363         }
 364
 365         LASSERT(dt_obj != NULL);
 366         if (!dt_object_exists(dt_obj)) {
 367                 dt_object_put(env, dt_obj);
 368                 dt_obj = dt_find_or_create(env, dt, &fid, &dof, attr);
 369                 fld->lsf_new = 1;
 370                 if (IS_ERR(dt_obj)) {
 371                         rc = PTR_ERR(dt_obj);
 372                         CERROR("%s: Can't find \"%s\" obj %d\n", fld->lsf_name,
 373                                 fld_index_name, rc);
 374                         dt_obj = NULL;
 375                         GOTO(out, rc);
 376                 }
 377         }
 378
 379         fld->lsf_obj = dt_obj;
 380         rc = dt_obj->do_ops->do_index_try(env, dt_obj, &fld_index_features);
 381         if (rc != 0) {
 382                 CERROR("%s: File \"%s\" is not an index: rc = %d!\n",
 383                        fld->lsf_name, fld_index_name, rc);
 384                 GOTO(out, rc);
 385         }
 386
 387         range = &info->fti_rec;
 388         /* Load fld entry to cache */
 389         iops = &dt_obj->do_index_ops->dio_it;
 390         it = iops->init(env, dt_obj, 0);
 391         if (IS_ERR(it))
 392                 GOTO(out, rc = PTR_ERR(it));
 393
 394         rc = iops->load(env, it, 0);
 395         if (rc > 0)
 396                 rc = 0;
 397         else if (rc == 0)
 398                 rc = iops->next(env, it);
 399
 400         if (rc < 0)
 401                 GOTO(out_it_fini, rc);
 402
 403         while (rc == 0) {
 404                 rc = iops->rec(env, it, (struct dt_rec *)range, 0);
 405                 if (rc != 0)
 406                         GOTO(out_it_put, rc);
 407
 408                 range_be_to_cpu(range, range);
 409
 410                 /*
 411                  * Newly created ldiskfs IAM indexes may include a
 412                  * zeroed-out key and record. Ignore it here.
 413                  */
 414                 if (range->lsr_start < range->lsr_end) {
 415                         rc = fld_cache_insert(fld->lsf_cache, range);
 416                         if (rc != 0)
 417                                 GOTO(out_it_put, rc);
 418
 419                         range_count++;
 420                 }
 421
 422                 rc = iops->next(env, it);
 423                 if (rc < 0)
 424                         GOTO(out_it_fini, rc);
 425         }
 426
 427         if (range_count == 0)
 428                 fld->lsf_new = 1;
 429
 430         rc = fld_name_to_index(fld->lsf_name, &index);
 431         if (rc < 0)
 432                 GOTO(out_it_put, rc);
 433         else
 434                 rc = 0;
 435
 436         if (index == 0 && type == LU_SEQ_RANGE_MDT) {
 437                 /*
 438                  * Note: fld_insert_entry will detect whether these
 439                  * special entries already exist inside FLDB
 440                  */
 441                 mutex_lock(&fld->lsf_lock);
 442                 rc = fld_insert_special_entries(env, fld);
 443                 mutex_unlock(&fld->lsf_lock);
 444                 if (rc != 0) {
 445                         CERROR("%s: insert special entries failed!: rc = %d\n",
 446                                fld->lsf_name, rc);
 447                         GOTO(out_it_put, rc);
 448                 }
 449         }
 450 out_it_put:
 451         iops->put(env, it);
 452 out_it_fini:
 453         iops->fini(env, it);
 454 out:
 455         if (attr)
 456                 OBD_FREE_PTR(attr);
 457
 458         if (rc < 0) {
 459                 if (dt_obj)
 460                         dt_object_put(env, dt_obj);
 461                 fld->lsf_obj = NULL;
 462         }
 463         RETURN(rc);
 464 }
 465
 466 void fld_index_fini(const struct lu_env *env, struct lu_server_fld *fld)
 467 {
 468         ENTRY;
 469         if (fld->lsf_obj) {
 470                 if (!IS_ERR(fld->lsf_obj))
 471                         dt_object_put(env, fld->lsf_obj);
 472                 fld->lsf_obj = NULL;
 473         }
 474         EXIT;
 475 }
 476
 477 int fld_server_read(const struct lu_env *env, struct lu_server_fld *fld,
 478                     struct lu_seq_range *range, void *data, int data_len)
 479 {
 480         struct lu_seq_range_array *lsra = data;
 481         struct fld_thread_info *info;
 482         struct dt_object *dt_obj = fld->lsf_obj;
 483         struct lu_seq_range *entry;
 484         struct dt_it *it;
 485         const struct dt_it_ops *iops;
 486         int rc;
 487
 488         ENTRY;
 489
 490         lsra->lsra_count = 0;
 491         iops = &dt_obj->do_index_ops->dio_it;
 492         it = iops->init(env, dt_obj, 0);
 493         if (IS_ERR(it))
 494                 RETURN(PTR_ERR(it));
 495
 496         rc = iops->load(env, it, range->lsr_end);
 497         if (rc <= 0)
 498                 GOTO(out_it_fini, rc);
 499
 500         info = lu_context_key_get(&env->le_ctx, &fld_thread_key);
 501         LASSERT(info != NULL);
 502         entry = &info->fti_rec;
 503         do {
 504                 rc = iops->rec(env, it, (struct dt_rec *)entry, 0);
 505                 if (rc != 0)
 506                         GOTO(out_it_put, rc);
 507
 508                 if (offsetof(typeof(*lsra), lsra_lsr[lsra->lsra_count + 1]) >
 509                     data_len)
 510                         GOTO(out, rc = -EAGAIN);
 511
 512                 range_be_to_cpu(entry, entry);
 513                 if (entry->lsr_index == range->lsr_index &&
 514                     entry->lsr_flags == range->lsr_flags &&
 515                     entry->lsr_start > range->lsr_start) {
 516                         lsra->lsra_lsr[lsra->lsra_count] = *entry;
 517                         lsra->lsra_count++;
 518                 }
 519
 520                 rc = iops->next(env, it);
 521         } while (rc == 0);
 522         if (rc > 0)
 523                 rc = 0;
 524 out:
 525         range_array_cpu_to_le(lsra, lsra);
 526 out_it_put:
 527         iops->put(env, it);
 528 out_it_fini:
 529         iops->fini(env, it);
 530
 531         RETURN(rc);
 532 }