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 /*
  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 #ifndef EXPORT_SYMTAB
  53 # define EXPORT_SYMTAB
  54 #endif
  55 #define DEBUG_SUBSYSTEM S_MDS
  56
  57 #include <linux/module.h>
  58
  59 /* LUSTRE_VERSION_CODE */
  60 #include <lustre_ver.h>
  61 /*
  62  * struct OBD_{ALLOC,FREE}*()
  63  * OBD_FAIL_CHECK
  64  */
  65 #include <obd.h>
  66 #include <obd_support.h>
  67
  68 /* fid_cpu_to_be() */
  69 #include <lustre_fid.h>
  70
  71 #include "osd_oi.h"
  72 /* osd_lookup(), struct osd_thread_info */
  73 #include "osd_internal.h"
  74 #include "osd_igif.h"
  75 #include "dt_object.h"
  76
  77 #define OSD_OI_FID_NR         (1UL << OSD_OI_FID_OID_BITS)
  78 #define OSD_OI_FID_NR_MAX     (1UL << OSD_OI_FID_OID_BITS_MAX)
  79
  80 static unsigned int osd_oi_count = OSD_OI_FID_NR;
  81 CFS_MODULE_PARM(osd_oi_count, "i", int, 0444,
  82                 "Number of Object Index containers to be created, "
  83                 "it's only valid for new filesystem.");
  84
  85 /** to serialize concurrent OI index initialization */
  86 static cfs_mutex_t oi_init_lock;
  87
  88 static struct dt_index_features oi_feat = {
  89         .dif_flags       = DT_IND_UPDATE,
  90         .dif_recsize_min = sizeof(struct osd_inode_id),
  91         .dif_recsize_max = sizeof(struct osd_inode_id),
  92         .dif_ptrsize     = 4
  93 };
  94
  95 #define OSD_OI_NAME_BASE        "oi.16"
  96
  97 /**
  98  * Open an OI(Ojbect Index) container.
  99  *
 100  * \param       name    Name of OI container
 101  * \param       objp    Pointer of returned OI
 102  *
 103  * \retval      0       success
 104  * \retval      -ve     failure
 105  */
 106 static int
 107 osd_oi_open(struct osd_thread_info *info,
 108             struct dt_device *dev, char *name, struct dt_object **objp)
 109 {
 110         const struct lu_env *env = info->oti_env;
 111         struct dt_object    *obj;
 112         int                  rc;
 113
 114         obj = dt_store_open(env, dev, "", name, &info->oti_fid);
 115         if (IS_ERR(obj))
 116                 return PTR_ERR(obj);
 117
 118         oi_feat.dif_keysize_min = sizeof(info->oti_fid);
 119         oi_feat.dif_keysize_max = sizeof(info->oti_fid);
 120
 121         rc = obj->do_ops->do_index_try(env, obj, &oi_feat);
 122         if (rc != 0) {
 123                 lu_object_put(info->oti_env, &obj->do_lu);
 124                 CERROR("%s: wrong index %s: rc = %d\n",
 125                        dev->dd_lu_dev.ld_obd->obd_name, name, rc);
 126                 return rc;
 127         }
 128
 129         *objp = obj;
 130         return 0;
 131 }
 132
 133
 134 static void
 135 osd_oi_table_put(struct osd_thread_info *info,
 136                  struct osd_oi *oi_table, unsigned oi_count)
 137 {
 138         int     i;
 139
 140         for (i = 0; i < oi_count; i++) {
 141                 LASSERT(oi_table[i].oi_dir != NULL);
 142
 143                 lu_object_put(info->oti_env, &oi_table[i].oi_dir->do_lu);
 144                 oi_table[i].oi_dir = NULL;
 145         }
 146 }
 147
 148 /**
 149  * Open OI(Object Index) table.
 150  * If \a oi_count is zero, which means caller doesn't know how many OIs there
 151  * will be, this function can either return 0 for new filesystem, or number
 152  * of OIs on existed filesystem.
 153  *
 154  * If \a oi_count is non-zero, which means caller does know number of OIs on
 155  * filesystem, this function should return the exactly same number on
 156  * success, or error code in failure.
 157  *
 158  * \param     oi_count  Number of expected OI containers
 159  * \param     try_all   Try to open all OIs even see failures
 160  *
 161  * \retval    +ve       number of opened OI containers
 162  * \retval      0       no OI containers found
 163  * \retval    -ve       failure
 164  */
 165 static int
 166 osd_oi_table_open(struct osd_thread_info *info, struct dt_device *dev,
 167                   struct osd_oi *oi_table, unsigned oi_count, int try_all)
 168 {
 169         int     count = 0;
 170         int     rc = 0;
 171         int     i;
 172
 173         /* NB: oi_count != 0 means that we have already created/known all OIs
 174          * and have known exact number of OIs. */
 175         LASSERT(oi_count <= OSD_OI_FID_NR_MAX);
 176
 177         for (i = 0; i < (oi_count != 0 ? oi_count : OSD_OI_FID_NR_MAX); i++) {
 178                 char name[12];
 179
 180                 sprintf(name, "%s.%d", OSD_OI_NAME_BASE, i);
 181                 rc = osd_oi_open(info, dev, name, &oi_table[i].oi_dir);
 182                 if (rc == 0) {
 183                         count++;
 184                         continue;
 185                 }
 186
 187                 if (try_all)
 188                         continue;
 189
 190                 if (rc == -ENOENT && oi_count == 0)
 191                         return count;
 192
 193                 CERROR("%s: can't open %s: rc = %d\n",
 194                        dev->dd_lu_dev.ld_obd->obd_name, name, rc);
 195
 196                 if (oi_count > 0) {
 197                         CERROR("%s: expect to open total %d OI files.\n",
 198                                dev->dd_lu_dev.ld_obd->obd_name, oi_count);
 199                 }
 200
 201                 break;
 202         }
 203
 204         if (try_all)
 205                 return count;
 206
 207         if (rc < 0) {
 208                 osd_oi_table_put(info, oi_table, count);
 209                 return rc;
 210         }
 211
 212         return count;
 213 }
 214
 215 static int osd_oi_table_create(struct osd_thread_info *info,
 216                                struct dt_device *dev,
 217                                struct md_device *mdev, int oi_count)
 218 {
 219         const struct lu_env *env;
 220         struct md_object *mdo;
 221         int i;
 222
 223         env = info->oti_env;
 224         for (i = 0; i < oi_count; ++i) {
 225                 char name[12];
 226
 227                 sprintf(name, "%s.%d", OSD_OI_NAME_BASE, i);
 228
 229                 lu_local_obj_fid(&info->oti_fid, OSD_OI_FID_OID_FIRST + i);
 230                 oi_feat.dif_keysize_min = sizeof(info->oti_fid);
 231                 oi_feat.dif_keysize_max = sizeof(info->oti_fid);
 232
 233                 mdo = llo_store_create_index(env, mdev, dev, "", name,
 234                                              &info->oti_fid, &oi_feat);
 235                 if (IS_ERR(mdo)) {
 236                         CERROR("Failed to create OI[%d] on %s: %d\n",
 237                                i, dev->dd_lu_dev.ld_obd->obd_name,
 238                                (int)PTR_ERR(mdo));
 239                         RETURN(PTR_ERR(mdo));
 240                 }
 241
 242                 lu_object_put(env, &mdo->mo_lu);
 243         }
 244         return 0;
 245 }
 246
 247 int osd_oi_init(struct osd_thread_info *info,
 248                 struct osd_oi **oi_table,
 249                 struct dt_device *dev,
 250                 struct md_device *mdev)
 251 {
 252         struct osd_oi *oi;
 253         int rc;
 254
 255         OBD_ALLOC(oi, sizeof(*oi) * OSD_OI_FID_NR_MAX);
 256         if (oi == NULL)
 257                 return -ENOMEM;
 258
 259         cfs_mutex_lock(&oi_init_lock);
 260
 261         rc = osd_oi_table_open(info, dev, oi, 0, 0);
 262         if (rc != 0)
 263                 goto out;
 264
 265         rc = osd_oi_open(info, dev, OSD_OI_NAME_BASE, &oi[0].oi_dir);
 266         if (rc == 0) { /* found single OI from old filesystem */
 267                 rc = 1;
 268                 goto out;
 269         }
 270
 271         if (rc != -ENOENT) {
 272                 CERROR("%s: can't open %s: rc = %d\n",
 273                        dev->dd_lu_dev.ld_obd->obd_name, OSD_OI_NAME_BASE, rc);
 274                 goto out;
 275         }
 276
 277         /* create OI objects */
 278         rc = osd_oi_table_create(info, dev, mdev, osd_oi_count);
 279         if (rc != 0)
 280                 goto out;
 281
 282         rc = osd_oi_table_open(info, dev, oi, osd_oi_count, 0);
 283         LASSERT(rc == osd_oi_count || rc < 0);
 284
 285  out:
 286         if (rc < 0) {
 287                 OBD_FREE(oi, sizeof(*oi) * OSD_OI_FID_NR_MAX);
 288         } else {
 289                 LASSERT((rc & (rc - 1)) == 0);
 290                 *oi_table = oi;
 291         }
 292
 293         cfs_mutex_unlock(&oi_init_lock);
 294         return rc;
 295 }
 296
 297 void osd_oi_fini(struct osd_thread_info *info,
 298                  struct osd_oi **oi_table, unsigned oi_count)
 299 {
 300         struct osd_oi *oi = *oi_table;
 301
 302         osd_oi_table_put(info, oi, oi_count);
 303
 304         OBD_FREE(oi, sizeof(*oi) * OSD_OI_FID_NR_MAX);
 305         *oi_table = NULL;
 306 }
 307
 308 int osd_oi_lookup(struct osd_thread_info *info, struct osd_oi *oi,
 309                   const struct lu_fid *fid, struct osd_inode_id *id)
 310 {
 311         struct lu_fid *oi_fid = &info->oti_fid;
 312         int rc;
 313
 314         if (osd_fid_is_igif(fid)) {
 315                 lu_igif_to_id(fid, id);
 316                 rc = 0;
 317         } else {
 318                 struct dt_object    *idx;
 319                 const struct dt_key *key;
 320
 321                 if (!fid_is_norm(fid))
 322                         return -ENOENT;
 323
 324                 idx = oi->oi_dir;
 325                 fid_cpu_to_be(oi_fid, fid);
 326                 key = (struct dt_key *) oi_fid;
 327                 rc = idx->do_index_ops->dio_lookup(info->oti_env, idx,
 328                                                    (struct dt_rec *)id, key,
 329                                                    BYPASS_CAPA);
 330                 if (rc > 0) {
 331                         id->oii_ino = be32_to_cpu(id->oii_ino);
 332                         id->oii_gen = be32_to_cpu(id->oii_gen);
 333                         rc = 0;
 334                 } else if (rc == 0)
 335                         rc = -ENOENT;
 336         }
 337         return rc;
 338 }
 339
 340 int osd_oi_insert(struct osd_thread_info *info, struct osd_oi *oi,
 341                   const struct lu_fid *fid, const struct osd_inode_id *id0,
 342                   struct thandle *th, int ignore_quota)
 343 {
 344         struct lu_fid *oi_fid = &info->oti_fid;
 345         struct dt_object    *idx;
 346         struct osd_inode_id *id;
 347         const struct dt_key *key;
 348
 349         if (!fid_is_norm(fid))
 350                 return 0;
 351
 352         idx = oi->oi_dir;
 353         fid_cpu_to_be(oi_fid, fid);
 354         key = (struct dt_key *) oi_fid;
 355
 356         id  = &info->oti_id;
 357         id->oii_ino = cpu_to_be32(id0->oii_ino);
 358         id->oii_gen = cpu_to_be32(id0->oii_gen);
 359         return idx->do_index_ops->dio_insert(info->oti_env, idx,
 360                                              (struct dt_rec *)id,
 361                                              key, th, BYPASS_CAPA,
 362                                              ignore_quota);
 363 }
 364
 365 int osd_oi_delete(struct osd_thread_info *info,
 366                   struct osd_oi *oi, const struct lu_fid *fid,
 367                   struct thandle *th)
 368 {
 369         struct lu_fid *oi_fid = &info->oti_fid;
 370         struct dt_object    *idx;
 371         const struct dt_key *key;
 372
 373         if (!fid_is_norm(fid))
 374                 return 0;
 375
 376         idx = oi->oi_dir;
 377         fid_cpu_to_be(oi_fid, fid);
 378         key = (struct dt_key *) oi_fid;
 379         return idx->do_index_ops->dio_delete(info->oti_env, idx,
 380                                              key, th, BYPASS_CAPA);
 381 }
 382
 383 int osd_oi_mod_init()
 384 {
 385         if (osd_oi_count == 0 || osd_oi_count > OSD_OI_FID_NR_MAX)
 386                 osd_oi_count = OSD_OI_FID_NR;
 387
 388         if ((osd_oi_count & (osd_oi_count - 1)) != 0) {
 389                 LCONSOLE_WARN("Round up oi_count %d to power2 %d\n",
 390                               osd_oi_count, size_roundup_power2(osd_oi_count));
 391                 osd_oi_count = size_roundup_power2(osd_oi_count);
 392         }
 393
 394         cfs_mutex_init(&oi_init_lock);
 395         return 0;
 396 }