lustre/osd-ldiskfs/osd_oi.c

   1 /* -*- mode: c; c-basic-offset: 8; indent-tabs-mode: nil; -*-
   2  * vim:expandtab:shiftwidth=8:tabstop=8:
   3  *
   4  * GPL HEADER START
   5  *
   6  * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.
   7  *
   8  * This program is free software; you can redistribute it and/or modify
   9  * it under the terms of the GNU General Public License version 2 only,
  10  * as published by the Free Software Foundation.
  11  *
  12  * This program is distributed in the hope that it will be useful, but
  13  * WITHOUT ANY WARRANTY; without even the implied warranty of
  14  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
  15  * General Public License version 2 for more details (a copy is included
  16  * in the LICENSE file that accompanied this code).
  17  *
  18  * You should have received a copy of the GNU General Public License
  19  * version 2 along with this program; If not, see
  20  * http://www.sun.com/software/products/lustre/docs/GPLv2.pdf
  21  *
  22  * Please contact Sun Microsystems, Inc., 4150 Network Circle, Santa Clara,
  23  * CA 95054 USA or visit www.sun.com if you need additional information or
  24  * have any questions.
  25  *
  26  * GPL HEADER END
  27  */
  28 /*
  29  * Copyright (c) 2007, 2010, Oracle and/or its affiliates. All rights reserved.
  30  * Use is subject to license terms.
  31  *
  32  * Copyright (c) 2012, Whamcloud, Inc.
  33  */
  34 /*
  35  * This file is part of Lustre, http://www.lustre.org/
  36  * Lustre is a trademark of Sun Microsystems, Inc.
  37  *
  38  * lustre/osd/osd_oi.c
  39  *
  40  * Object Index.
  41  *
  42  * Author: Nikita Danilov <nikita@clusterfs.com>
  43  */
  44
  45 /*
  46  * oi uses two mechanisms to implement fid->cookie mapping:
  47  *
  48  *     - persistent index, where cookie is a record and fid is a key, and
  49  *
  50  *     - algorithmic mapping for "igif" fids.
  51  *
  52  */
  53
  54 #ifndef EXPORT_SYMTAB
  55 # define EXPORT_SYMTAB
  56 #endif
  57 #define DEBUG_SUBSYSTEM S_MDS
  58
  59 #include <linux/module.h>
  60
  61 /* LUSTRE_VERSION_CODE */
  62 #include <lustre_ver.h>
  63 /*
  64  * struct OBD_{ALLOC,FREE}*()
  65  * OBD_FAIL_CHECK
  66  */
  67 #include <obd.h>
  68 #include <obd_support.h>
  69
  70 /* fid_cpu_to_be() */
  71 #include <lustre_fid.h>
  72
  73 #include "osd_oi.h"
  74 /* osd_lookup(), struct osd_thread_info */
  75 #include "osd_internal.h"
  76 #include "osd_igif.h"
  77 #include "dt_object.h"
  78
  79 #define OSD_OI_FID_NR         (1UL << OSD_OI_FID_OID_BITS)
  80 #define OSD_OI_FID_NR_MAX     (1UL << OSD_OI_FID_OID_BITS_MAX)
  81
  82 static unsigned int osd_oi_count = OSD_OI_FID_NR;
  83 CFS_MODULE_PARM(osd_oi_count, "i", int, 0444,
  84                 "Number of Object Index containers to be created, "
  85                 "it's only valid for new filesystem.");
  86
  87 /** to serialize concurrent OI index initialization */
  88 static cfs_mutex_t oi_init_lock;
  89
  90 static struct dt_index_features oi_feat = {
  91         .dif_flags       = DT_IND_UPDATE,
  92         .dif_recsize_min = sizeof(struct osd_inode_id),
  93         .dif_recsize_max = sizeof(struct osd_inode_id),
  94         .dif_ptrsize     = 4
  95 };
  96
  97 #define OSD_OI_NAME_BASE        "oi.16"
  98
  99 static void osd_oi_table_put(struct osd_thread_info *info,
 100                              struct osd_oi **oi_table, unsigned oi_count)
 101 {
 102         struct iam_container *bag;
 103         int                   i;
 104
 105         for (i = 0; i < oi_count; i++) {
 106                 LASSERT(oi_table[i] != NULL);
 107                 LASSERT(oi_table[i]->oi_inode != NULL);
 108
 109                 bag = &(oi_table[i]->oi_dir.od_container);
 110                 if (bag->ic_object == oi_table[i]->oi_inode)
 111                         iam_container_fini(bag);
 112                 iput(oi_table[i]->oi_inode);
 113                 oi_table[i]->oi_inode = NULL;
 114                 OBD_FREE_PTR(oi_table[i]);
 115         }
 116 }
 117
 118 static int osd_oi_index_create_one(struct osd_thread_info *info,
 119                                    struct osd_device *osd, const char *name,
 120                                    struct dt_index_features *feat)
 121 {
 122         const struct lu_env             *env = info->oti_env;
 123         struct osd_inode_id             *id  = &info->oti_id;
 124         struct buffer_head              *bh;
 125         struct inode                    *inode;
 126         struct ldiskfs_dir_entry_2      *de;
 127         struct dentry                   *dentry;
 128         struct inode                    *dir;
 129         handle_t                        *jh;
 130         int                              rc;
 131
 132         dentry = osd_child_dentry_by_inode(env, osd_sb(osd)->s_root->d_inode,
 133                                            name, strlen(name));
 134         dir = osd_sb(osd)->s_root->d_inode;
 135         bh = osd_ldiskfs_find_entry(dir, dentry, &de, NULL);
 136         if (bh) {
 137                 brelse(bh);
 138
 139                 id->oii_ino = le32_to_cpu(de->inode);
 140                 id->oii_gen = OSD_OII_NOGEN;
 141
 142                 inode = osd_iget(info, osd, id);
 143                 if (!IS_ERR(inode)) {
 144                         iput(inode);
 145                         RETURN(-EEXIST);
 146                 }
 147                 RETURN(PTR_ERR(inode));
 148         }
 149
 150         jh = ldiskfs_journal_start_sb(osd_sb(osd), 100);
 151         LASSERT(!IS_ERR(jh));
 152
 153         inode = ldiskfs_create_inode(jh, osd_sb(osd)->s_root->d_inode,
 154                                      (S_IFREG | S_IRUGO | S_IWUSR));
 155         LASSERT(!IS_ERR(inode));
 156
 157         if (feat->dif_flags & DT_IND_VARKEY)
 158                 rc = iam_lvar_create(inode, feat->dif_keysize_max,
 159                                      feat->dif_ptrsize, feat->dif_recsize_max,
 160                                      jh);
 161         else
 162                 rc = iam_lfix_create(inode, feat->dif_keysize_max,
 163                                      feat->dif_ptrsize, feat->dif_recsize_max,
 164                                      jh);
 165
 166         dentry = osd_child_dentry_by_inode(env, osd_sb(osd)->s_root->d_inode,
 167                                            name, strlen(name));
 168         rc = osd_ldiskfs_add_entry(jh, dentry, inode, NULL);
 169         LASSERT(rc == 0);
 170
 171         ldiskfs_journal_stop(jh);
 172         iput(inode);
 173
 174         return rc;
 175 }
 176
 177 static struct inode *osd_oi_index_open(struct osd_thread_info *info,
 178                                        struct osd_device *osd,
 179                                        const char *name,
 180                                        struct dt_index_features *f,
 181                                        bool create)
 182 {
 183         struct dentry *dentry;
 184         struct inode  *inode;
 185         int            rc;
 186
 187         dentry = ll_lookup_one_len(name, osd_sb(osd)->s_root, strlen(name));
 188         if (IS_ERR(dentry))
 189                 return (void *) dentry;
 190
 191         if (dentry->d_inode) {
 192                 LASSERT(!is_bad_inode(dentry->d_inode));
 193                 inode = dentry->d_inode;
 194                 atomic_inc(&inode->i_count);
 195                 dput(dentry);
 196                 return inode;
 197         }
 198
 199         /* create */
 200         dput(dentry);
 201         shrink_dcache_parent(osd_sb(osd)->s_root);
 202         if (!create)
 203                 return ERR_PTR(-ENOENT);
 204
 205         rc = osd_oi_index_create_one(info, osd, name, f);
 206         if (rc)
 207                 RETURN(ERR_PTR(rc));
 208
 209         dentry = ll_lookup_one_len(name, osd_sb(osd)->s_root, strlen(name));
 210         if (IS_ERR(dentry))
 211                 return (void *) dentry;
 212
 213         if (dentry->d_inode) {
 214                 LASSERT(!is_bad_inode(dentry->d_inode));
 215                 inode = dentry->d_inode;
 216                 atomic_inc(&inode->i_count);
 217                 dput(dentry);
 218                 return inode;
 219         }
 220
 221         return ERR_PTR(-ENOENT);
 222 }
 223
 224 /**
 225  * Open an OI(Ojbect Index) container.
 226  *
 227  * \param       name    Name of OI container
 228  * \param       objp    Pointer of returned OI
 229  *
 230  * \retval      0       success
 231  * \retval      -ve     failure
 232  */
 233 static int osd_oi_open(struct osd_thread_info *info, struct osd_device *osd,
 234                        char *name, struct osd_oi **oi_slot, bool create)
 235 {
 236         struct osd_directory *dir;
 237         struct iam_container *bag;
 238         struct inode         *inode;
 239         struct osd_oi        *oi;
 240         int                   rc;
 241
 242         ENTRY;
 243
 244         oi_feat.dif_keysize_min = sizeof(struct lu_fid);
 245         oi_feat.dif_keysize_max = sizeof(struct lu_fid);
 246
 247         inode = osd_oi_index_open(info, osd, name, &oi_feat, create);
 248         if (IS_ERR(inode))
 249                 RETURN(PTR_ERR(inode));
 250
 251         OBD_ALLOC_PTR(oi);
 252         if (oi == NULL)
 253                 GOTO(out_inode, rc = -ENOMEM);
 254
 255         oi->oi_inode = inode;
 256         dir = &oi->oi_dir;
 257
 258         bag = &dir->od_container;
 259         rc = iam_container_init(bag, &dir->od_descr, inode);
 260         if (rc < 0)
 261                 GOTO(out_free, rc);
 262
 263         rc = iam_container_setup(bag);
 264         if (rc < 0)
 265                 GOTO(out_container, rc);
 266
 267         *oi_slot = oi;
 268         RETURN(0);
 269
 270 out_container:
 271         iam_container_fini(bag);
 272 out_free:
 273         OBD_FREE_PTR(oi);
 274 out_inode:
 275         iput(inode);
 276         return rc;
 277 }
 278
 279 /**
 280  * Open OI(Object Index) table.
 281  * If \a oi_count is zero, which means caller doesn't know how many OIs there
 282  * will be, this function can either return 0 for new filesystem, or number
 283  * of OIs on existed filesystem.
 284  *
 285  * If \a oi_count is non-zero, which means caller does know number of OIs on
 286  * filesystem, this function should return the exactly same number on
 287  * success, or error code in failure.
 288  *
 289  * \param     oi_count  Number of expected OI containers
 290  * \param     create    Create OIs if doesn't exist
 291  *
 292  * \retval    +ve       number of opened OI containers
 293  * \retval      0       no OI containers found
 294  * \retval    -ve       failure
 295  */
 296 static int
 297 osd_oi_table_open(struct osd_thread_info *info, struct osd_device *osd,
 298                   struct osd_oi **oi_table, unsigned oi_count, bool create)
 299 {
 300         struct dt_device *dev = &osd->od_dt_dev;
 301         int               count = 0;
 302         int               rc = 0;
 303         int               i;
 304
 305         /* NB: oi_count != 0 means that we have already created/known all OIs
 306          * and have known exact number of OIs. */
 307         LASSERT(oi_count <= OSD_OI_FID_NR_MAX);
 308
 309         for (i = 0; i < (oi_count != 0 ? oi_count : OSD_OI_FID_NR_MAX); i++) {
 310                 char name[12];
 311
 312                 sprintf(name, "%s.%d", OSD_OI_NAME_BASE, i);
 313                 rc = osd_oi_open(info, osd, name, &oi_table[i], create);
 314                 if (rc == 0) {
 315                         count++;
 316                         continue;
 317                 }
 318
 319                 if (rc == -ENOENT && oi_count == 0)
 320                         return count;
 321
 322                 CERROR("%s: can't open %s: rc = %d\n",
 323                        dev->dd_lu_dev.ld_obd->obd_name, name, rc);
 324                 if (oi_count > 0) {
 325                         CERROR("%s: expect to open total %d OI files.\n",
 326                                dev->dd_lu_dev.ld_obd->obd_name, oi_count);
 327                 }
 328                 break;
 329         }
 330
 331         if (rc < 0) {
 332                 osd_oi_table_put(info, oi_table, count);
 333                 return rc;
 334         }
 335
 336         return count;
 337 }
 338
 339 int osd_oi_init(struct osd_thread_info *info, struct osd_device *osd)
 340 {
 341         struct dt_device *dev = &osd->od_dt_dev;
 342         struct osd_oi   **oi;
 343         int               rc;
 344
 345         OBD_ALLOC(oi, sizeof(*oi) * OSD_OI_FID_NR_MAX);
 346         if (oi == NULL)
 347                 return -ENOMEM;
 348
 349         cfs_mutex_lock(&oi_init_lock);
 350         /* try to open existing multiple OIs first */
 351         rc = osd_oi_table_open(info, osd, oi, 0, false);
 352         if (rc != 0)
 353                 goto out;
 354
 355         /* if previous failed then try found single OI from old filesystem */
 356         rc = osd_oi_open(info, osd, OSD_OI_NAME_BASE, &oi[0], false);
 357         if (rc == 0) { /* found single OI from old filesystem */
 358                 rc = 1;
 359                 goto out;
 360         } else if (rc != -ENOENT) {
 361                 CERROR("%s: can't open %s: rc = %d\n",
 362                        dev->dd_lu_dev.ld_obd->obd_name, OSD_OI_NAME_BASE, rc);
 363                 goto out;
 364         }
 365
 366         /* No OIs exist, new filesystem, create OI objects */
 367         rc = osd_oi_table_open(info, osd, oi, osd_oi_count, true);
 368         LASSERT(ergo(rc >= 0, rc == osd_oi_count));
 369 out:
 370         if (rc < 0) {
 371                 OBD_FREE(oi, sizeof(*oi) * OSD_OI_FID_NR_MAX);
 372         } else {
 373                 LASSERT((rc & (rc - 1)) == 0);
 374                 osd->od_oi_table = oi;
 375                 osd->od_oi_count = rc;
 376                 rc = 0;
 377         }
 378
 379         cfs_mutex_unlock(&oi_init_lock);
 380         return rc;
 381 }
 382
 383 void osd_oi_fini(struct osd_thread_info *info, struct osd_device *osd)
 384 {
 385         osd_oi_table_put(info, osd->od_oi_table, osd->od_oi_count);
 386
 387         OBD_FREE(osd->od_oi_table,
 388                  sizeof(*(osd->od_oi_table)) * OSD_OI_FID_NR_MAX);
 389         osd->od_oi_table = NULL;
 390 }
 391
 392 static inline int fid_is_fs_root(const struct lu_fid *fid)
 393 {
 394         /* Map root inode to special local object FID */
 395         return (unlikely(fid_seq(fid) == FID_SEQ_LOCAL_FILE &&
 396                          fid_oid(fid) == OSD_FS_ROOT_OID));
 397 }
 398
 399 static int osd_oi_iam_lookup(struct osd_thread_info *oti,
 400                              struct osd_oi *oi, struct dt_rec *rec,
 401                              const struct dt_key *key)
 402 {
 403         struct iam_container  *bag;
 404         struct iam_iterator   *it = &oti->oti_idx_it;
 405         struct iam_rec        *iam_rec;
 406         struct iam_path_descr *ipd;
 407         int                    rc;
 408         ENTRY;
 409
 410         LASSERT(oi);
 411         LASSERT(oi->oi_inode);
 412
 413         bag = &oi->oi_dir.od_container;
 414         ipd = osd_idx_ipd_get(oti->oti_env, bag);
 415         if (IS_ERR(ipd))
 416                 RETURN(-ENOMEM);
 417
 418         /* got ipd now we can start iterator. */
 419         iam_it_init(it, bag, 0, ipd);
 420
 421         rc = iam_it_get(it, (struct iam_key *)key);
 422         if (rc >= 0) {
 423                 if (S_ISDIR(oi->oi_inode->i_mode))
 424                         iam_rec = (struct iam_rec *)oti->oti_ldp;
 425                 else
 426                         iam_rec = (struct iam_rec *)rec;
 427
 428                 iam_reccpy(&it->ii_path.ip_leaf, (struct iam_rec *)iam_rec);
 429                 if (S_ISDIR(oi->oi_inode->i_mode))
 430                         osd_fid_unpack((struct lu_fid *)rec,
 431                                        (struct osd_fid_pack *)iam_rec);
 432         }
 433         iam_it_put(it);
 434         iam_it_fini(it);
 435         osd_ipd_put(oti->oti_env, bag, ipd);
 436
 437         LINVRNT(osd_invariant(obj));
 438
 439         RETURN(rc);
 440 }
 441
 442 int osd_oi_lookup(struct osd_thread_info *info, struct osd_device *osd,
 443                   const struct lu_fid *fid, struct osd_inode_id *id)
 444 {
 445         struct lu_fid       *oi_fid = &info->oti_fid;
 446         const struct dt_key *key;
 447         int                  rc     = 0;
 448
 449         if (osd_fid_is_igif(fid)) {
 450                 lu_igif_to_id(fid, id);
 451                 rc = 0;
 452         } else {
 453                 if (!fid_is_norm(fid))
 454                         return -ENOENT;
 455
 456                 fid_cpu_to_be(oi_fid, fid);
 457                 key = (struct dt_key *) oi_fid;
 458
 459                 rc = osd_oi_iam_lookup(info, osd_fid2oi(osd, fid),
 460                                        (struct dt_rec *)id, key);
 461
 462                 if (rc > 0) {
 463                         id->oii_ino = be32_to_cpu(id->oii_ino);
 464                         id->oii_gen = be32_to_cpu(id->oii_gen);
 465                         rc = 0;
 466                 } else if (rc == 0) {
 467                         rc = -ENOENT;
 468                 }
 469         }
 470         return rc;
 471 }
 472
 473 static int osd_oi_iam_insert(struct osd_thread_info *oti, struct osd_oi *oi,
 474                              const struct dt_rec *rec, const struct dt_key *key,
 475                              struct thandle *th, int ignore_quota)
 476 {
 477         struct iam_container  *bag;
 478         struct iam_rec        *iam_rec = (struct iam_rec *)oti->oti_ldp;
 479         struct iam_path_descr *ipd;
 480         struct osd_thandle    *oh;
 481         int                    rc;
 482 #ifdef HAVE_QUOTA_SUPPORT
 483         cfs_cap_t              save    = cfs_curproc_cap_pack();
 484 #endif
 485         ENTRY;
 486
 487         LASSERT(oi);
 488         LASSERT(oi->oi_inode);
 489
 490         bag = &oi->oi_dir.od_container;
 491         ipd = osd_idx_ipd_get(oti->oti_env, bag);
 492         if (unlikely(ipd == NULL))
 493                 RETURN(-ENOMEM);
 494
 495         oh = container_of0(th, struct osd_thandle, ot_super);
 496         LASSERT(oh->ot_handle != NULL);
 497         LASSERT(oh->ot_handle->h_transaction != NULL);
 498 #ifdef HAVE_QUOTA_SUPPORT
 499         if (ignore_quota)
 500                 cfs_cap_raise(CFS_CAP_SYS_RESOURCE);
 501         else
 502                 cfs_cap_lower(CFS_CAP_SYS_RESOURCE);
 503 #endif
 504         if (S_ISDIR(oi->oi_inode->i_mode))
 505                 osd_fid_pack((struct osd_fid_pack *)iam_rec, rec,
 506                              &oti->oti_fid);
 507         else
 508                 iam_rec = (struct iam_rec *) rec;
 509         rc = iam_insert(oh->ot_handle, bag, (const struct iam_key *)key,
 510                         iam_rec, ipd);
 511 #ifdef HAVE_QUOTA_SUPPORT
 512         cfs_curproc_cap_unpack(save);
 513 #endif
 514         osd_ipd_put(oti->oti_env, bag, ipd);
 515         LINVRNT(osd_invariant(obj));
 516         RETURN(rc);
 517 }
 518
 519 int osd_oi_insert(struct osd_thread_info *info, struct osd_device *osd,
 520                   const struct lu_fid *fid, const struct osd_inode_id *id0,
 521                   struct thandle *th, int ignore_quota)
 522 {
 523         struct lu_fid       *oi_fid = &info->oti_fid;
 524         struct osd_inode_id *id;
 525         const struct dt_key *key;
 526
 527         if (!fid_is_norm(fid))
 528                 return 0;
 529
 530         fid_cpu_to_be(oi_fid, fid);
 531         key = (struct dt_key *)oi_fid;
 532
 533         id  = &info->oti_id;
 534         id->oii_ino = cpu_to_be32(id0->oii_ino);
 535         id->oii_gen = cpu_to_be32(id0->oii_gen);
 536
 537         return osd_oi_iam_insert(info, osd_fid2oi(osd, fid),
 538                                  (struct dt_rec *)id, key, th, ignore_quota);
 539 }
 540
 541 static int osd_oi_iam_delete(struct osd_thread_info *oti, struct osd_oi *oi,
 542                              const struct dt_key *key, struct thandle *handle)
 543 {
 544         struct iam_container  *bag;
 545         struct iam_path_descr *ipd;
 546         struct osd_thandle    *oh;
 547         int                    rc;
 548         ENTRY;
 549
 550         LASSERT(oi);
 551
 552         bag = &oi->oi_dir.od_container;
 553         ipd = osd_idx_ipd_get(oti->oti_env, bag);
 554         if (unlikely(ipd == NULL))
 555                 RETURN(-ENOMEM);
 556
 557         oh = container_of0(handle, struct osd_thandle, ot_super);
 558         LASSERT(oh->ot_handle != NULL);
 559         LASSERT(oh->ot_handle->h_transaction != NULL);
 560
 561         rc = iam_delete(oh->ot_handle, bag, (const struct iam_key *)key, ipd);
 562         osd_ipd_put(oti->oti_env, bag, ipd);
 563         LINVRNT(osd_invariant(obj));
 564         RETURN(rc);
 565 }
 566
 567 int osd_oi_delete(struct osd_thread_info *info,
 568                   struct osd_device *osd, const struct lu_fid *fid,
 569                   struct thandle *th)
 570 {
 571         struct lu_fid       *oi_fid = &info->oti_fid;
 572         const struct dt_key *key;
 573
 574         if (!fid_is_norm(fid))
 575                 return 0;
 576
 577         fid_cpu_to_be(oi_fid, fid);
 578         key = (struct dt_key *)oi_fid;
 579
 580         return osd_oi_iam_delete(info, osd_fid2oi(osd, fid), key, th);
 581 }
 582
 583 int osd_oi_mod_init()
 584 {
 585         if (osd_oi_count == 0 || osd_oi_count > OSD_OI_FID_NR_MAX)
 586                 osd_oi_count = OSD_OI_FID_NR;
 587
 588         if ((osd_oi_count & (osd_oi_count - 1)) != 0) {
 589                 LCONSOLE_WARN("Round up oi_count %d to power2 %d\n",
 590                               osd_oi_count, size_roundup_power2(osd_oi_count));
 591                 osd_oi_count = size_roundup_power2(osd_oi_count);
 592         }
 593
 594         cfs_mutex_init(&oi_init_lock);
 595         return 0;
 596 }