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