lustre/osd-zfs/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.gnu.org/licenses/gpl-2.0.html
  19  *
  20  * GPL HEADER END
  21  */
  22 /*
  23  * Copyright (c) 2009, 2010, Oracle and/or its affiliates. All rights reserved.
  24  * Use is subject to license terms.
  25  *
  26  * Copyright (c) 2012, 2016, 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/osd-zfs/osd_oi.c
  33  * OI functions to map fid to dnode
  34  *
  35  * Author: Alex Zhuravlev <bzzz@whamcloud.com>
  36  * Author: Mike Pershin <tappro@whamcloud.com>
  37  * Author: Di Wang <di.wang@intel.com>
  38  */
  39
  40 #define DEBUG_SUBSYSTEM S_OSD
  41
  42 #include <lustre_ver.h>
  43 #include <libcfs/libcfs.h>
  44 #include <obd_support.h>
  45 #include <lustre_net.h>
  46 #include <obd.h>
  47 #include <obd_class.h>
  48 #include <lustre_disk.h>
  49 #include <lustre_fid.h>
  50
  51 #include "osd_internal.h"
  52
  53 #include <sys/dnode.h>
  54 #include <sys/dbuf.h>
  55 #include <sys/spa.h>
  56 #include <sys/stat.h>
  57 #include <sys/zap.h>
  58 #include <sys/spa_impl.h>
  59 #include <sys/zfs_znode.h>
  60 #include <sys/dmu_tx.h>
  61 #include <sys/dmu_objset.h>
  62 #include <sys/dsl_prop.h>
  63 #include <sys/sa_impl.h>
  64 #include <sys/txg.h>
  65
  66 #define OSD_OI_FID_NR         (1UL << 7)
  67 #define OSD_OI_FID_NR_MAX     (1UL << OSD_OI_FID_OID_BITS_MAX)
  68 unsigned int osd_oi_count = OSD_OI_FID_NR;
  69
  70
  71 /*
  72  * zfs osd maintains names for known fids in the name hierarchy
  73  * so that one can mount filesystem with regular ZFS stack and
  74  * access files
  75  */
  76 struct named_oid {
  77         unsigned long    oid;
  78         char            *name;
  79 };
  80
  81 static const struct named_oid oids[] = {
  82         { .oid = LAST_RECV_OID,        .name = LAST_RCVD },
  83         { .oid = OFD_LAST_GROUP_OID,   .name = "LAST_GROUP" },
  84         { .oid = LLOG_CATALOGS_OID,    .name = "CATALOGS" },
  85         { .oid = MGS_CONFIGS_OID,      /*MOUNT_CONFIGS_DIR*/ },
  86         { .oid = FID_SEQ_SRV_OID,      .name = "seq_srv" },
  87         { .oid = FID_SEQ_CTL_OID,      .name = "seq_ctl" },
  88         { .oid = FLD_INDEX_OID,        .name = "fld" },
  89         { .oid = MDD_LOV_OBJ_OID,      .name = LOV_OBJID },
  90         { .oid = OFD_HEALTH_CHECK_OID, .name = HEALTH_CHECK },
  91         { .oid = ACCT_USER_OID,        .name = "acct_usr_inode" },
  92         { .oid = ACCT_GROUP_OID,       .name = "acct_grp_inode" },
  93         { .oid = REPLY_DATA_OID,       .name = REPLY_DATA },
  94         { .oid = 0 }
  95 };
  96
  97 static char *oid2name(const unsigned long oid)
  98 {
  99         int i = 0;
 100
 101         while (oids[i].oid) {
 102                 if (oids[i].oid == oid)
 103                         return oids[i].name;
 104                 i++;
 105         }
 106         return NULL;
 107 }
 108
 109 /**
 110  * Lookup an existing OI by the given name.
 111  */
 112 static int
 113 osd_oi_lookup(const struct lu_env *env, struct osd_device *o,
 114               uint64_t parent, const char *name, struct osd_oi *oi)
 115 {
 116         struct zpl_direntry     *zde = &osd_oti_get(env)->oti_zde.lzd_reg;
 117         int                      rc;
 118
 119         rc = -zap_lookup(o->od_os, parent, name, 8, 1, (void *)zde);
 120         if (rc)
 121                 return rc;
 122
 123         rc = strlcpy(oi->oi_name, name, sizeof(oi->oi_name));
 124         if (rc >= sizeof(oi->oi_name))
 125                 return -E2BIG;
 126
 127         oi->oi_zapid = zde->zde_dnode;
 128
 129         return 0;
 130 }
 131
 132 /**
 133  * Create a new OI with the given name.
 134  */
 135 static int
 136 osd_oi_create(const struct lu_env *env, struct osd_device *o,
 137               uint64_t parent, const char *name, uint64_t *child)
 138 {
 139         struct zpl_direntry     *zde = &osd_oti_get(env)->oti_zde.lzd_reg;
 140         struct lu_attr          *la = &osd_oti_get(env)->oti_la;
 141         sa_handle_t             *sa_hdl = NULL;
 142         dmu_tx_t                *tx;
 143         uint64_t                 oid;
 144         int                      rc;
 145
 146         /* verify it doesn't already exist */
 147         rc = -zap_lookup(o->od_os, parent, name, 8, 1, (void *)zde);
 148         if (rc == 0)
 149                 return -EEXIST;
 150
 151         if (o->od_dt_dev.dd_rdonly)
 152                 return -EROFS;
 153
 154         /* create fid-to-dnode index */
 155         tx = dmu_tx_create(o->od_os);
 156         if (tx == NULL)
 157                 return -ENOMEM;
 158
 159         dmu_tx_hold_zap(tx, DMU_NEW_OBJECT, 1, NULL);
 160         dmu_tx_hold_bonus(tx, parent);
 161         dmu_tx_hold_zap(tx, parent, TRUE, name);
 162         dmu_tx_hold_sa_create(tx, ZFS_SA_BASE_ATTR_SIZE);
 163
 164         rc = -dmu_tx_assign(tx, TXG_WAIT);
 165         if (rc) {
 166                 dmu_tx_abort(tx);
 167                 return rc;
 168         }
 169
 170         oid = zap_create_flags(o->od_os, 0, ZAP_FLAG_HASH64,
 171                                DMU_OT_DIRECTORY_CONTENTS,
 172                                14, /* == ZFS fzap_default_block_shift */
 173                                DN_MAX_INDBLKSHIFT, /* indirect block shift */
 174                                DMU_OT_SA, DN_MAX_BONUSLEN, tx);
 175
 176         rc = -sa_handle_get(o->od_os, oid, NULL, SA_HDL_PRIVATE, &sa_hdl);
 177         if (rc)
 178                 goto commit;
 179         la->la_valid = LA_MODE | LA_UID | LA_GID;
 180         la->la_mode = S_IFDIR | S_IRUGO | S_IWUSR | S_IXUGO;
 181         la->la_uid = la->la_gid = 0;
 182         rc = __osd_attr_init(env, o, sa_hdl, tx, la, parent);
 183         sa_handle_destroy(sa_hdl);
 184         if (rc)
 185                 goto commit;
 186
 187         zde->zde_dnode = oid;
 188         zde->zde_pad = 0;
 189         zde->zde_type = IFTODT(S_IFDIR);
 190
 191         rc = -zap_add(o->od_os, parent, name, 8, 1, (void *)zde, tx);
 192
 193 commit:
 194         if (rc)
 195                 dmu_object_free(o->od_os, oid, tx);
 196         dmu_tx_commit(tx);
 197
 198         if (rc == 0)
 199                 *child = oid;
 200
 201         return rc;
 202 }
 203
 204 static int
 205 osd_oi_find_or_create(const struct lu_env *env, struct osd_device *o,
 206                       uint64_t parent, const char *name, uint64_t *child)
 207 {
 208         struct osd_oi   oi;
 209         int             rc;
 210
 211         rc = osd_oi_lookup(env, o, parent, name, &oi);
 212         if (rc == 0)
 213                 *child = oi.oi_zapid;
 214         else if (rc == -ENOENT)
 215                 rc = osd_oi_create(env, o, parent, name, child);
 216
 217         return rc;
 218 }
 219
 220 /**
 221  * Lookup the target index/flags of the fid, so it will know where
 222  * the object is located (tgt index) and it is MDT or OST object.
 223  */
 224 int osd_fld_lookup(const struct lu_env *env, struct osd_device *osd,
 225                    u64 seq, struct lu_seq_range *range)
 226 {
 227         struct seq_server_site  *ss = osd_seq_site(osd);
 228
 229         if (fid_seq_is_idif(seq)) {
 230                 fld_range_set_ost(range);
 231                 range->lsr_index = idif_ost_idx(seq);
 232                 return 0;
 233         }
 234
 235         if (!fid_seq_in_fldb(seq)) {
 236                 fld_range_set_mdt(range);
 237                 if (ss != NULL)
 238                         /* FIXME: If ss is NULL, it suppose not get lsr_index
 239                          * at all */
 240                         range->lsr_index = ss->ss_node_id;
 241                 return 0;
 242         }
 243
 244         LASSERT(ss != NULL);
 245         fld_range_set_any(range);
 246         /* OSD will only do local fld lookup */
 247         return fld_local_lookup(env, ss->ss_server_fld, seq, range);
 248 }
 249
 250 int fid_is_on_ost(const struct lu_env *env, struct osd_device *osd,
 251                   const struct lu_fid *fid)
 252 {
 253         struct lu_seq_range     *range = &osd_oti_get(env)->oti_seq_range;
 254         int                     rc;
 255         ENTRY;
 256
 257         if (fid_is_idif(fid))
 258                 RETURN(1);
 259
 260         if (unlikely(fid_is_local_file(fid) || fid_is_llog(fid)) ||
 261                      fid_is_name_llog(fid) || fid_is_quota(fid))
 262                 RETURN(0);
 263
 264         rc = osd_fld_lookup(env, osd, fid_seq(fid), range);
 265         if (rc != 0) {
 266                 if (rc != -ENOENT)
 267                         CERROR("%s: "DFID" lookup failed: rc = %d\n",
 268                                osd_name(osd), PFID(fid), rc);
 269                 RETURN(0);
 270         }
 271
 272         if (fld_range_is_ost(range))
 273                 RETURN(1);
 274
 275         RETURN(0);
 276 }
 277
 278 static struct osd_seq *osd_seq_find_locked(struct osd_seq_list *seq_list,
 279                                            u64 seq)
 280 {
 281         struct osd_seq *osd_seq;
 282
 283         list_for_each_entry(osd_seq, &seq_list->osl_seq_list, os_seq_list) {
 284                 if (osd_seq->os_seq == seq)
 285                         return osd_seq;
 286         }
 287         return NULL;
 288 }
 289
 290 static struct osd_seq *osd_seq_find(struct osd_seq_list *seq_list, u64 seq)
 291 {
 292         struct osd_seq *osd_seq;
 293
 294         read_lock(&seq_list->osl_seq_list_lock);
 295         osd_seq = osd_seq_find_locked(seq_list, seq);
 296         read_unlock(&seq_list->osl_seq_list_lock);
 297
 298         return osd_seq;
 299 }
 300
 301 static struct osd_seq *osd_find_or_add_seq(const struct lu_env *env,
 302                                            struct osd_device *osd, u64 seq)
 303 {
 304         struct osd_seq_list     *seq_list = &osd->od_seq_list;
 305         struct osd_seq          *osd_seq;
 306         char                    *key = osd_oti_get(env)->oti_buf;
 307         char                    *seq_name = osd_oti_get(env)->oti_str;
 308         struct osd_oi           oi;
 309         uint64_t                sdb, odb;
 310         int                     i;
 311         int                     rc = 0;
 312         ENTRY;
 313
 314         osd_seq = osd_seq_find(seq_list, seq);
 315         if (osd_seq != NULL)
 316                 RETURN(osd_seq);
 317
 318         down(&seq_list->osl_seq_init_sem);
 319         /* Check again, in case some one else already add it
 320          * to the list */
 321         osd_seq = osd_seq_find(seq_list, seq);
 322         if (osd_seq != NULL)
 323                 GOTO(out, rc = 0);
 324
 325         OBD_ALLOC_PTR(osd_seq);
 326         if (osd_seq == NULL)
 327                 GOTO(out, rc = -ENOMEM);
 328
 329         INIT_LIST_HEAD(&osd_seq->os_seq_list);
 330         osd_seq->os_seq = seq;
 331
 332         /* Init subdir count to be 32, but each seq can have
 333          * different subdir count */
 334         osd_seq->os_subdir_count = OSD_OST_MAP_SIZE;
 335         OBD_ALLOC(osd_seq->os_compat_dirs,
 336                   sizeof(uint64_t) * osd_seq->os_subdir_count);
 337         if (osd_seq->os_compat_dirs == NULL)
 338                 GOTO(out, rc = -ENOMEM);
 339
 340         oi.oi_zapid = osd->od_O_id;
 341         sprintf(seq_name, (fid_seq_is_rsvd(seq) ||
 342                 fid_seq_is_mdt0(seq)) ?  "%llu" : "%llx",
 343                 fid_seq_is_idif(seq) ? 0 : seq);
 344
 345         rc = osd_oi_find_or_create(env, osd, oi.oi_zapid, seq_name, &odb);
 346         if (rc != 0) {
 347                 CERROR("%s: Can not create %s : rc = %d\n",
 348                        osd_name(osd), seq_name, rc);
 349                 GOTO(out, rc);
 350         }
 351
 352         for (i = 0; i < OSD_OST_MAP_SIZE; i++) {
 353                 sprintf(key, "d%d", i);
 354                 rc = osd_oi_find_or_create(env, osd, odb, key, &sdb);
 355                 if (rc)
 356                         GOTO(out, rc);
 357                 osd_seq->os_compat_dirs[i] = sdb;
 358         }
 359
 360         write_lock(&seq_list->osl_seq_list_lock);
 361         list_add(&osd_seq->os_seq_list, &seq_list->osl_seq_list);
 362         write_unlock(&seq_list->osl_seq_list_lock);
 363 out:
 364         up(&seq_list->osl_seq_init_sem);
 365         if (rc != 0) {
 366                 if (osd_seq != NULL && osd_seq->os_compat_dirs != NULL)
 367                         OBD_FREE(osd_seq->os_compat_dirs,
 368                                  sizeof(uint64_t) * osd_seq->os_subdir_count);
 369                 if (osd_seq != NULL)
 370                         OBD_FREE_PTR(osd_seq);
 371                 osd_seq = ERR_PTR(rc);
 372         }
 373         RETURN(osd_seq);
 374 }
 375
 376 /*
 377  * objects w/o a natural reference (unlike a file on a MDS)
 378  * are put under a special hierarchy /O/<seq>/d0..dXX
 379  * this function returns a directory specific fid belongs to
 380  */
 381 static uint64_t
 382 osd_get_idx_for_ost_obj(const struct lu_env *env, struct osd_device *osd,
 383                         const struct lu_fid *fid, char *buf, int bufsize)
 384 {
 385         struct osd_seq  *osd_seq;
 386         unsigned long   b;
 387         u64             id;
 388         int             rc;
 389
 390         osd_seq = osd_find_or_add_seq(env, osd, fid_seq(fid));
 391         if (IS_ERR(osd_seq)) {
 392                 CERROR("%s: Can not find seq group "DFID"\n", osd_name(osd),
 393                        PFID(fid));
 394                 return PTR_ERR(osd_seq);
 395         }
 396
 397         if (fid_is_last_id(fid)) {
 398                 id = 0;
 399         } else {
 400                 rc = fid_to_ostid(fid, &osd_oti_get(env)->oti_ostid);
 401                 LASSERT(rc == 0); /* we should not get here with IGIF */
 402                 id = ostid_id(&osd_oti_get(env)->oti_ostid);
 403         }
 404
 405         b = id % OSD_OST_MAP_SIZE;
 406         LASSERT(osd_seq->os_compat_dirs[b]);
 407
 408         if (buf)
 409                 snprintf(buf, bufsize, "%llu", id);
 410
 411         return osd_seq->os_compat_dirs[b];
 412 }
 413
 414 /* XXX: f_ver is not counted, but may differ too */
 415 static void osd_fid2str(char *buf, const struct lu_fid *fid)
 416 {
 417         sprintf(buf, DFID_NOBRACE, PFID(fid));
 418 }
 419
 420 /*
 421  * Determine the zap object id which is being used as the OI for the
 422  * given fid.  The lowest N bits in the sequence ID are used as the
 423  * index key.  On failure 0 is returned which zfs treats internally
 424  * as an invalid object id.
 425  */
 426 static uint64_t
 427 osd_get_idx_for_fid(struct osd_device *osd, const struct lu_fid *fid,
 428                     char *buf)
 429 {
 430         struct osd_oi *oi;
 431
 432         LASSERT(osd->od_oi_table != NULL);
 433         oi = osd->od_oi_table[fid_seq(fid) & (osd->od_oi_count - 1)];
 434         if (buf)
 435                 osd_fid2str(buf, fid);
 436
 437         return oi->oi_zapid;
 438 }
 439
 440 uint64_t osd_get_name_n_idx(const struct lu_env *env, struct osd_device *osd,
 441                             const struct lu_fid *fid, char *buf, int bufsize)
 442 {
 443         uint64_t zapid;
 444
 445         LASSERT(fid);
 446
 447         if (fid_is_on_ost(env, osd, fid) == 1 || fid_seq(fid) == FID_SEQ_ECHO) {
 448                 zapid = osd_get_idx_for_ost_obj(env, osd, fid, buf, bufsize);
 449         } else if (unlikely(fid_seq(fid) == FID_SEQ_LOCAL_FILE)) {
 450                 /* special objects with fixed known fids get their name */
 451                 char *name = oid2name(fid_oid(fid));
 452
 453                 if (name) {
 454                         zapid = osd->od_root;
 455                         if (buf)
 456                                 strncpy(buf, name, bufsize);
 457                         if (fid_is_acct(fid))
 458                                 zapid = MASTER_NODE_OBJ;
 459                 } else {
 460                         zapid = osd_get_idx_for_fid(osd, fid, buf);
 461                 }
 462         } else {
 463                 zapid = osd_get_idx_for_fid(osd, fid, buf);
 464         }
 465
 466         return zapid;
 467 }
 468
 469 static inline int fid_is_fs_root(const struct lu_fid *fid)
 470 {
 471         /* Map root inode to special local object FID */
 472         return fid_seq(fid) == FID_SEQ_LOCAL_FILE &&
 473                 fid_oid(fid) == OSD_FS_ROOT_OID;
 474 }
 475
 476 int osd_fid_lookup(const struct lu_env *env, struct osd_device *dev,
 477                    const struct lu_fid *fid, uint64_t *oid)
 478 {
 479         struct osd_thread_info  *info = osd_oti_get(env);
 480         char                    *buf = info->oti_buf;
 481         uint64_t                zapid;
 482         int                     rc = 0;
 483         ENTRY;
 484
 485         if (OBD_FAIL_CHECK(OBD_FAIL_SRV_ENOENT))
 486                 RETURN(-ENOENT);
 487
 488         if (unlikely(fid_is_acct(fid))) {
 489                 if (fid_oid(fid) == ACCT_USER_OID)
 490                         *oid = dev->od_iusr_oid;
 491                 else
 492                         *oid = dev->od_igrp_oid;
 493         } else if (unlikely(fid_is_fs_root(fid))) {
 494                 *oid = dev->od_root;
 495         } else {
 496                 zapid = osd_get_name_n_idx(env, dev, fid, buf,
 497                                            sizeof(info->oti_buf));
 498                 rc = -zap_lookup(dev->od_os, zapid, buf,
 499                                 8, 1, &info->oti_zde);
 500                 if (rc)
 501                         RETURN(rc);
 502                 *oid = info->oti_zde.lzd_reg.zde_dnode;
 503         }
 504
 505         if (rc == 0)
 506                 osd_dmu_prefetch(dev->od_os, *oid, 0, 0, 0,
 507                                  ZIO_PRIORITY_ASYNC_READ);
 508
 509         RETURN(rc);
 510 }
 511
 512 /**
 513  * Close an entry in a specific slot.
 514  */
 515 static void
 516 osd_oi_remove_table(const struct lu_env *env, struct osd_device *o, int key)
 517 {
 518         struct osd_oi *oi;
 519
 520         LASSERT(key < o->od_oi_count);
 521
 522         oi = o->od_oi_table[key];
 523         if (oi) {
 524                 if (oi->oi_dn)
 525                         osd_dnode_rele(oi->oi_dn);
 526                 OBD_FREE_PTR(oi);
 527                 o->od_oi_table[key] = NULL;
 528         }
 529 }
 530
 531 /**
 532  * Allocate and open a new entry in the specified unused slot.
 533  */
 534 static int
 535 osd_oi_add_table(const struct lu_env *env, struct osd_device *o,
 536                  char *name, int key)
 537 {
 538         struct osd_oi *oi;
 539         int rc;
 540
 541         LASSERT(key < o->od_oi_count);
 542         LASSERT(o->od_oi_table[key] == NULL);
 543
 544         OBD_ALLOC_PTR(oi);
 545         if (oi == NULL)
 546                 return -ENOMEM;
 547
 548         rc = osd_oi_lookup(env, o, o->od_root, name, oi);
 549         if (rc) {
 550                 OBD_FREE_PTR(oi);
 551                 return rc;
 552         }
 553
 554         o->od_oi_table[key] = oi;
 555         __osd_obj2dnode(env, o->od_os, oi->oi_zapid, &oi->oi_dn);
 556
 557         return 0;
 558 }
 559
 560 /**
 561  * Depopulate the OI table.
 562  */
 563 static void
 564 osd_oi_close_table(const struct lu_env *env, struct osd_device *o)
 565 {
 566         int i;
 567
 568         for (i = 0; i < o->od_oi_count; i++)
 569                 osd_oi_remove_table(env, o, i);
 570 }
 571
 572 /**
 573  * Populate the OI table based.
 574  */
 575 static int
 576 osd_oi_open_table(const struct lu_env *env, struct osd_device *o, int count)
 577 {
 578         char name[16];
 579         int  i, rc = 0;
 580         ENTRY;
 581
 582         for (i = 0; i < count; i++) {
 583                 sprintf(name, "%s.%d", DMU_OSD_OI_NAME_BASE, i);
 584                 rc = osd_oi_add_table(env, o, name, i);
 585                 if (rc) {
 586                         osd_oi_close_table(env, o);
 587                         break;
 588                 }
 589         }
 590
 591         RETURN(rc);
 592 }
 593
 594 /**
 595  * Determine if the type and number of OIs used by this file system.
 596  */
 597 static int
 598 osd_oi_probe(const struct lu_env *env, struct osd_device *o, int *count)
 599 {
 600         uint64_t        root_oid = o->od_root;
 601         struct osd_oi   oi;
 602         char            name[16];
 603         int             rc;
 604         ENTRY;
 605
 606         /*
 607          * Check for multiple OIs and determine the count.  There is no
 608          * gap handling, if an OI is missing the wrong size can be returned.
 609          * The only safeguard is that we know the number of OIs must be a
 610          * power of two and this is checked for basic sanity.
 611          */
 612         for (*count = 0; *count < OSD_OI_FID_NR_MAX; (*count)++) {
 613                 sprintf(name, "%s.%d", DMU_OSD_OI_NAME_BASE, *count);
 614                 rc = osd_oi_lookup(env, o, root_oid, name, &oi);
 615                 if (rc == 0)
 616                         continue;
 617
 618                 if (rc == -ENOENT) {
 619                         if (*count == 0)
 620                                 break;
 621
 622                         if ((*count & (*count - 1)) != 0)
 623                                 RETURN(-EDOM);
 624
 625                         RETURN(0);
 626                 }
 627
 628                 RETURN(rc);
 629         }
 630
 631         /*
 632          * No OIs exist, this must be a new filesystem.
 633          */
 634         *count = 0;
 635
 636         RETURN(0);
 637 }
 638
 639 static void osd_ost_seq_fini(const struct lu_env *env, struct osd_device *osd)
 640 {
 641         struct osd_seq_list     *osl = &osd->od_seq_list;
 642         struct osd_seq          *osd_seq, *tmp;
 643
 644         write_lock(&osl->osl_seq_list_lock);
 645         list_for_each_entry_safe(osd_seq, tmp, &osl->osl_seq_list,
 646                                  os_seq_list) {
 647                 list_del(&osd_seq->os_seq_list);
 648                 OBD_FREE(osd_seq->os_compat_dirs,
 649                          sizeof(uint64_t) * osd_seq->os_subdir_count);
 650                 OBD_FREE(osd_seq, sizeof(*osd_seq));
 651         }
 652         write_unlock(&osl->osl_seq_list_lock);
 653
 654         return;
 655 }
 656
 657 /**
 658  * Create /O subdirectory to map legacy OST objects for compatibility.
 659  */
 660 static int
 661 osd_oi_init_compat(const struct lu_env *env, struct osd_device *o)
 662 {
 663         uint64_t         odb, sdb;
 664         int              rc;
 665         ENTRY;
 666
 667         rc = osd_oi_find_or_create(env, o, o->od_root, "O", &sdb);
 668         if (rc)
 669                 RETURN(rc);
 670
 671         o->od_O_id = sdb;
 672
 673         /* Create on-disk indexes to maintain per-UID/GID inode usage.
 674          * Those new indexes are created in the top-level ZAP outside the
 675          * namespace in order not to confuse ZPL which might interpret those
 676          * indexes as directories and assume the values are object IDs */
 677         rc = osd_oi_find_or_create(env, o, MASTER_NODE_OBJ,
 678                         oid2name(ACCT_USER_OID), &odb);
 679         if (rc)
 680                 RETURN(rc);
 681         o->od_iusr_oid = odb;
 682
 683         rc = osd_oi_find_or_create(env, o, MASTER_NODE_OBJ,
 684                         oid2name(ACCT_GROUP_OID), &odb);
 685         if (rc)
 686                 RETURN(rc);
 687         o->od_igrp_oid = odb;
 688
 689         RETURN(rc);
 690 }
 691
 692 /**
 693  * Initialize the OIs by either opening or creating them as needed.
 694  */
 695 int osd_oi_init(const struct lu_env *env, struct osd_device *o)
 696 {
 697         char    *key = osd_oti_get(env)->oti_buf;
 698         int      i, rc, count = 0;
 699         ENTRY;
 700
 701         rc = osd_oi_probe(env, o, &count);
 702         if (rc)
 703                 RETURN(rc);
 704
 705         if (count == 0) {
 706                 uint64_t odb, sdb;
 707
 708                 count = osd_oi_count;
 709                 odb = o->od_root;
 710
 711                 for (i = 0; i < count; i++) {
 712                         sprintf(key, "%s.%d", DMU_OSD_OI_NAME_BASE, i);
 713                         rc = osd_oi_find_or_create(env, o, odb, key, &sdb);
 714                         if (rc)
 715                                 RETURN(rc);
 716                 }
 717         }
 718
 719         rc = osd_oi_init_compat(env, o);
 720         if (rc)
 721                 RETURN(rc);
 722
 723         LASSERT((count & (count - 1)) == 0);
 724         o->od_oi_count = count;
 725         OBD_ALLOC(o->od_oi_table, sizeof(struct osd_oi *) * count);
 726         if (o->od_oi_table == NULL)
 727                 RETURN(-ENOMEM);
 728
 729         rc = osd_oi_open_table(env, o, count);
 730         if (rc) {
 731                 OBD_FREE(o->od_oi_table, sizeof(struct osd_oi *) * count);
 732                 o->od_oi_table = NULL;
 733         }
 734
 735         RETURN(rc);
 736 }
 737
 738 void osd_oi_fini(const struct lu_env *env, struct osd_device *o)
 739 {
 740         ENTRY;
 741
 742         osd_ost_seq_fini(env, o);
 743
 744         if (o->od_oi_table != NULL) {
 745                 (void) osd_oi_close_table(env, o);
 746                 OBD_FREE(o->od_oi_table,
 747                          sizeof(struct osd_oi *) * o->od_oi_count);
 748                 o->od_oi_table = NULL;
 749                 o->od_oi_count = 0;
 750         }
 751
 752         EXIT;
 753 }
 754
 755 int osd_options_init(void)
 756 {
 757         /* osd_oi_count - Default number of OIs, 128 works well for ZFS */
 758         if (osd_oi_count == 0 || osd_oi_count > OSD_OI_FID_NR_MAX)
 759                 osd_oi_count = OSD_OI_FID_NR;
 760
 761         if ((osd_oi_count & (osd_oi_count - 1)) != 0) {
 762                 LCONSOLE_WARN("Round up osd_oi_count %d to power2 %d\n",
 763                         osd_oi_count, size_roundup_power2(osd_oi_count));
 764                 osd_oi_count = size_roundup_power2(osd_oi_count);
 765         }
 766
 767         return 0;
 768 }
 769
 770 /*
 771  * the following set of functions are used to maintain per-thread
 772  * cache of FID->ino mapping. this mechanism is used to avoid
 773  * expensive LU/OI lookups.
 774  */
 775 struct osd_idmap_cache *osd_idc_find(const struct lu_env *env,
 776                                      struct osd_device *osd,
 777                                      const struct lu_fid *fid)
 778 {
 779         struct osd_thread_info *oti = osd_oti_get(env);
 780         struct osd_idmap_cache *idc = oti->oti_ins_cache;
 781         int i;
 782
 783         for (i = 0; i < oti->oti_ins_cache_used; i++) {
 784                 if (!lu_fid_eq(&idc[i].oic_fid, fid))
 785                         continue;
 786                 if (idc[i].oic_dev != osd)
 787                         continue;
 788
 789                 return idc + i;
 790         }
 791
 792         return NULL;
 793 }
 794
 795 struct osd_idmap_cache *osd_idc_add(const struct lu_env *env,
 796                                     struct osd_device *osd,
 797                                     const struct lu_fid *fid)
 798 {
 799         struct osd_thread_info *oti = osd_oti_get(env);
 800         struct osd_idmap_cache *idc;
 801         int i;
 802
 803         if (unlikely(oti->oti_ins_cache_used >= oti->oti_ins_cache_size)) {
 804                 i = oti->oti_ins_cache_size * 2;
 805                 LASSERT(i < 1000);
 806                 if (i == 0)
 807                         i = OSD_INS_CACHE_SIZE;
 808                 OBD_ALLOC(idc, sizeof(*idc) * i);
 809                 if (idc == NULL)
 810                         return ERR_PTR(-ENOMEM);
 811                 if (oti->oti_ins_cache != NULL) {
 812                         memcpy(idc, oti->oti_ins_cache,
 813                                oti->oti_ins_cache_used * sizeof(*idc));
 814                         OBD_FREE(oti->oti_ins_cache,
 815                                  oti->oti_ins_cache_used * sizeof(*idc));
 816                 }
 817                 oti->oti_ins_cache = idc;
 818                 oti->oti_ins_cache_size = i;
 819         }
 820
 821         idc = &oti->oti_ins_cache[oti->oti_ins_cache_used++];
 822         idc->oic_fid = *fid;
 823         idc->oic_dev = osd;
 824         idc->oic_dnode = 0;
 825         idc->oic_remote = 0;
 826
 827         return idc;
 828 }
 829
 830 /**
 831  * Lookup mapping for the given fid in the cache
 832  *
 833  * Initialize a new one if not found. the initialization checks whether
 834  * the object is local or remote. for the local objects, OI is used to
 835  * learn dnode#. the function is used when the caller has no information
 836  * about the object, e.g. at dt_insert().
 837  */
 838 struct osd_idmap_cache *osd_idc_find_or_init(const struct lu_env *env,
 839                                              struct osd_device *osd,
 840                                              const struct lu_fid *fid)
 841 {
 842         struct osd_idmap_cache *idc;
 843         int rc;
 844
 845         idc = osd_idc_find(env, osd, fid);
 846         if (idc != NULL)
 847                 return idc;
 848
 849         /* new mapping is needed */
 850         idc = osd_idc_add(env, osd, fid);
 851         if (IS_ERR(idc))
 852                 return idc;
 853
 854         /* initialize it */
 855         rc = osd_remote_fid(env, osd, fid);
 856         if (unlikely(rc < 0))
 857                 return ERR_PTR(rc);
 858
 859         if (rc == 0) {
 860                 /* the object is local, lookup in OI */
 861                 uint64_t dnode;
 862
 863                 rc = osd_fid_lookup(env, osd, fid, &dnode);
 864                 if (unlikely(rc < 0)) {
 865                         CERROR("%s: can't lookup: rc = %d\n",
 866                                osd->od_svname, rc);
 867                         return ERR_PTR(rc);
 868                 }
 869                 LASSERT(dnode < (1ULL << DN_MAX_OBJECT_SHIFT));
 870                 idc->oic_dnode = dnode;
 871         } else {
 872                 /* the object is remote */
 873                 idc->oic_remote = 1;
 874         }
 875
 876         return idc;
 877 }
 878
 879 /*
 880  * lookup mapping for given FID and fill it from the given object.
 881  * the object is local by definition.
 882  */
 883 int osd_idc_find_and_init(const struct lu_env *env, struct osd_device *osd,
 884                           struct osd_object *obj)
 885 {
 886         const struct lu_fid *fid = lu_object_fid(&obj->oo_dt.do_lu);
 887         struct osd_idmap_cache *idc;
 888
 889         idc = osd_idc_find(env, osd, fid);
 890         if (idc != NULL) {
 891                 if (obj->oo_dn == NULL)
 892                         return 0;
 893                 idc->oic_dnode = obj->oo_dn->dn_object;
 894                 return 0;
 895         }
 896
 897         /* new mapping is needed */
 898         idc = osd_idc_add(env, osd, fid);
 899         if (IS_ERR(idc))
 900                 return PTR_ERR(idc);
 901
 902         if (obj->oo_dn)
 903                 idc->oic_dnode = obj->oo_dn->dn_object;
 904
 905         return 0;
 906 }