lustre/osd-ldiskfs/osd_compat.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) 2010, Oracle and/or its affiliates. All rights reserved.
  28  * Use is subject to license terms.
  29  *
  30  * Copyright (c) 2012, 2013, Intel Corporation.
  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_compat.c
  37  *
  38  * on-disk structure for managing /O
  39  *
  40  * Author: Alex Zhuravlev <bzzz@whamcloud.com>
  41  */
  42
  43 /* LUSTRE_VERSION_CODE */
  44 #include <lustre_ver.h>
  45 /* prerequisite for linux/xattr.h */
  46 #include <linux/types.h>
  47 /* prerequisite for linux/xattr.h */
  48 #include <linux/fs.h>
  49 /* XATTR_{REPLACE,CREATE} */
  50 #include <linux/xattr.h>
  51
  52 /*
  53  * struct OBD_{ALLOC,FREE}*()
  54  * OBD_FAIL_CHECK
  55  */
  56 #include <obd_support.h>
  57
  58 #include "osd_internal.h"
  59 #include "osd_oi.h"
  60
  61 static void osd_push_ctxt(const struct osd_device *dev,
  62                           struct lvfs_run_ctxt *newctxt,
  63                           struct lvfs_run_ctxt *save)
  64 {
  65         OBD_SET_CTXT_MAGIC(newctxt);
  66         newctxt->pwdmnt = dev->od_mnt;
  67         newctxt->pwd = dev->od_mnt->mnt_root;
  68         newctxt->fs = get_ds();
  69
  70         push_ctxt(save, newctxt, NULL);
  71 }
  72
  73 static void osd_pop_ctxt(const struct osd_device *dev,
  74                          struct lvfs_run_ctxt *new,
  75                          struct lvfs_run_ctxt *save)
  76 {
  77         pop_ctxt(save, new, NULL);
  78 }
  79
  80 /* utility to make a directory */
  81 static struct dentry *simple_mkdir(struct dentry *dir, struct vfsmount *mnt,
  82                                    const char *name, int mode, int fix)
  83 {
  84         struct dentry *dchild;
  85         int err = 0;
  86         ENTRY;
  87
  88         // ASSERT_KERNEL_CTXT("kernel doing mkdir outside kernel context\n");
  89         CDEBUG(D_INODE, "creating directory %.*s\n", (int)strlen(name), name);
  90         dchild = ll_lookup_one_len(name, dir, strlen(name));
  91         if (IS_ERR(dchild))
  92                 GOTO(out_up, dchild);
  93
  94         if (dchild->d_inode) {
  95                 int old_mode = dchild->d_inode->i_mode;
  96                 if (!S_ISDIR(old_mode)) {
  97                         CERROR("found %s (%lu/%u) is mode %o\n", name,
  98                                dchild->d_inode->i_ino,
  99                                dchild->d_inode->i_generation, old_mode);
 100                         GOTO(out_err, err = -ENOTDIR);
 101                 }
 102
 103                 /* Fixup directory permissions if necessary */
 104                 if (fix && (old_mode & S_IALLUGO) != (mode & S_IALLUGO)) {
 105                         CDEBUG(D_CONFIG,
 106                                "fixing permissions on %s from %o to %o\n",
 107                                name, old_mode, mode);
 108                         dchild->d_inode->i_mode = (mode & S_IALLUGO) |
 109                                                   (old_mode & ~S_IALLUGO);
 110                         mark_inode_dirty(dchild->d_inode);
 111                 }
 112                 GOTO(out_up, dchild);
 113         }
 114
 115         err = ll_vfs_mkdir(dir->d_inode, dchild, mnt, mode);
 116         if (err)
 117                 GOTO(out_err, err);
 118
 119         RETURN(dchild);
 120
 121 out_err:
 122         dput(dchild);
 123         dchild = ERR_PTR(err);
 124 out_up:
 125         return dchild;
 126 }
 127
 128 int osd_last_rcvd_subdir_count(struct osd_device *osd)
 129 {
 130         struct lr_server_data lsd;
 131         struct dentry        *dlast;
 132         loff_t                off;
 133         int                   rc = 0;
 134         int                   count = FILTER_SUBDIR_COUNT;
 135
 136         ENTRY;
 137
 138         dlast = ll_lookup_one_len(LAST_RCVD, osd_sb(osd)->s_root,
 139                                   strlen(LAST_RCVD));
 140         if (IS_ERR(dlast))
 141                 return PTR_ERR(dlast);
 142         else if (dlast->d_inode == NULL)
 143                 goto out;
 144
 145         off = 0;
 146         rc = osd_ldiskfs_read(dlast->d_inode, &lsd, sizeof(lsd), &off);
 147         if (rc == sizeof(lsd)) {
 148                 CDEBUG(D_INFO, "read last_rcvd header, uuid = %s, "
 149                        "subdir count = %d\n", lsd.lsd_uuid,
 150                        lsd.lsd_subdir_count);
 151                 if (le16_to_cpu(lsd.lsd_subdir_count) > 0)
 152                         count = le16_to_cpu(lsd.lsd_subdir_count);
 153         } else if (rc != 0) {
 154                 CERROR("Can't read last_rcvd file, rc = %d\n", rc);
 155                 if (rc > 0)
 156                         rc = -EFAULT;
 157                 dput(dlast);
 158                 return rc;
 159         }
 160 out:
 161         dput(dlast);
 162         LASSERT(count > 0);
 163         return count;
 164 }
 165
 166 static const char remote_parent_dir[] = "REMOTE_PARENT_DIR";
 167 static int osd_mdt_init(const struct lu_env *env, struct osd_device *dev)
 168 {
 169         struct lvfs_run_ctxt    new;
 170         struct lvfs_run_ctxt    save;
 171         struct dentry           *parent;
 172         struct osd_mdobj_map    *omm;
 173         struct dentry           *d;
 174         struct osd_thread_info  *info = osd_oti_get(env);
 175         struct lu_fid           *fid = &info->oti_fid;
 176         int                     rc = 0;
 177         ENTRY;
 178
 179         OBD_ALLOC_PTR(dev->od_mdt_map);
 180         if (dev->od_mdt_map == NULL)
 181                 RETURN(-ENOMEM);
 182
 183         omm = dev->od_mdt_map;
 184
 185         LASSERT(dev->od_fsops);
 186
 187         parent = osd_sb(dev)->s_root;
 188         osd_push_ctxt(dev, &new, &save);
 189
 190         d = simple_mkdir(parent, dev->od_mnt, remote_parent_dir,
 191                          0755, 1);
 192         if (IS_ERR(d))
 193                 GOTO(cleanup, rc = PTR_ERR(d));
 194
 195         omm->omm_remote_parent = d;
 196
 197         /* Set LMA for remote parent inode */
 198         lu_local_obj_fid(fid, REMOTE_PARENT_DIR_OID);
 199         rc = osd_ea_fid_set(info, d->d_inode, fid, 0);
 200         if (rc != 0)
 201                 GOTO(cleanup, rc);
 202 cleanup:
 203         pop_ctxt(&save, &new, NULL);
 204         if (rc) {
 205                 if (omm->omm_remote_parent != NULL)
 206                         dput(omm->omm_remote_parent);
 207                 OBD_FREE_PTR(omm);
 208                 dev->od_mdt_map = NULL;
 209         }
 210         RETURN(rc);
 211 }
 212
 213 static void osd_mdt_fini(struct osd_device *osd)
 214 {
 215         struct osd_mdobj_map *omm = osd->od_mdt_map;
 216
 217         if (omm == NULL)
 218                 return;
 219
 220         if (omm->omm_remote_parent)
 221                 dput(omm->omm_remote_parent);
 222
 223         OBD_FREE_PTR(omm);
 224         osd->od_ost_map = NULL;
 225 }
 226
 227 int osd_add_to_remote_parent(const struct lu_env *env, struct osd_device *osd,
 228                              struct osd_object *obj, struct osd_thandle *oh)
 229 {
 230         struct osd_mdobj_map    *omm = osd->od_mdt_map;
 231         struct osd_thread_info  *oti = osd_oti_get(env);
 232         struct lustre_mdt_attrs *lma = &oti->oti_mdt_attrs;
 233         char                    *name = oti->oti_name;
 234         struct dentry           *dentry;
 235         struct dentry           *parent;
 236         int                     rc;
 237
 238         /* Set REMOTE_PARENT in lma, so other process like unlink or lfsck
 239          * can identify this object quickly */
 240         rc = osd_get_lma(oti, obj->oo_inode, &oti->oti_obj_dentry, lma);
 241         if (rc != 0)
 242                 RETURN(rc);
 243
 244         lma->lma_incompat |= LMAI_REMOTE_PARENT;
 245         lustre_lma_swab(lma);
 246         rc = __osd_xattr_set(oti, obj->oo_inode, XATTR_NAME_LMA, lma,
 247                              sizeof(*lma), XATTR_REPLACE);
 248         if (rc != 0)
 249                 RETURN(rc);
 250
 251         parent = omm->omm_remote_parent;
 252         sprintf(name, DFID_NOBRACE, PFID(lu_object_fid(&obj->oo_dt.do_lu)));
 253         dentry = osd_child_dentry_by_inode(env, parent->d_inode,
 254                                            name, strlen(name));
 255         mutex_lock(&parent->d_inode->i_mutex);
 256         rc = osd_ldiskfs_add_entry(oh->ot_handle, dentry, obj->oo_inode,
 257                                    NULL);
 258         CDEBUG(D_INODE, "%s: add %s:%lu to remote parent %lu.\n", osd_name(osd),
 259                name, obj->oo_inode->i_ino, parent->d_inode->i_ino);
 260         LASSERTF(parent->d_inode->i_nlink > 1, "%s: %lu nlink %d",
 261                  osd_name(osd), parent->d_inode->i_ino,
 262                  parent->d_inode->i_nlink);
 263         parent->d_inode->i_nlink++;
 264         mark_inode_dirty(parent->d_inode);
 265         mutex_unlock(&parent->d_inode->i_mutex);
 266         RETURN(rc);
 267 }
 268
 269 int osd_delete_from_remote_parent(const struct lu_env *env,
 270                                   struct osd_device *osd,
 271                                   struct osd_object *obj,
 272                                   struct osd_thandle *oh)
 273 {
 274         struct osd_mdobj_map       *omm = osd->od_mdt_map;
 275         struct osd_thread_info     *oti = osd_oti_get(env);
 276         struct lustre_mdt_attrs    *lma = &oti->oti_mdt_attrs;
 277         char                       *name = oti->oti_name;
 278         struct dentry              *dentry;
 279         struct dentry              *parent;
 280         struct ldiskfs_dir_entry_2 *de;
 281         struct buffer_head         *bh;
 282         int                        rc;
 283
 284         /* Check lma to see whether it is remote object */
 285         rc = osd_get_lma(oti, obj->oo_inode, &oti->oti_obj_dentry, lma);
 286         if (rc != 0)
 287                 RETURN(rc);
 288
 289         if (likely(!(lma->lma_incompat & LMAI_REMOTE_PARENT)))
 290                 RETURN(0);
 291
 292         parent = omm->omm_remote_parent;
 293         sprintf(name, DFID_NOBRACE, PFID(lu_object_fid(&obj->oo_dt.do_lu)));
 294         dentry = osd_child_dentry_by_inode(env, parent->d_inode,
 295                                            name, strlen(name));
 296         mutex_lock(&parent->d_inode->i_mutex);
 297         bh = osd_ldiskfs_find_entry(parent->d_inode, dentry, &de, NULL);
 298         if (bh == NULL) {
 299                 mutex_unlock(&parent->d_inode->i_mutex);
 300                 RETURN(-ENOENT);
 301         }
 302         CDEBUG(D_INODE, "%s: el %s:%lu to remote parent %lu.\n", osd_name(osd),
 303                name, obj->oo_inode->i_ino, parent->d_inode->i_ino);
 304         rc = ldiskfs_delete_entry(oh->ot_handle, parent->d_inode, de, bh);
 305         LASSERTF(parent->d_inode->i_nlink > 1, "%s: %lu nlink %d",
 306                  osd_name(osd), parent->d_inode->i_ino,
 307                  parent->d_inode->i_nlink);
 308         parent->d_inode->i_nlink--;
 309         mark_inode_dirty(parent->d_inode);
 310         mutex_unlock(&parent->d_inode->i_mutex);
 311         brelse(bh);
 312
 313         /* Get rid of REMOTE_PARENT flag from incompat */
 314         lma->lma_incompat &= ~LMAI_REMOTE_PARENT;
 315         lustre_lma_swab(lma);
 316         rc = __osd_xattr_set(oti, obj->oo_inode, XATTR_NAME_LMA, lma,
 317                              sizeof(*lma), XATTR_REPLACE);
 318         RETURN(rc);
 319 }
 320
 321 /*
 322  * directory structure on legacy OST:
 323  *
 324  * O/<seq>/d0-31/<objid>
 325  * O/<seq>/LAST_ID
 326  * last_rcvd
 327  * LAST_GROUP
 328  * CONFIGS
 329  *
 330  */
 331 static int osd_ost_init(struct osd_device *dev)
 332 {
 333         struct lvfs_run_ctxt  new;
 334         struct lvfs_run_ctxt  save;
 335         struct dentry        *rootd = osd_sb(dev)->s_root;
 336         struct dentry        *d;
 337         int                   rc;
 338         ENTRY;
 339
 340         OBD_ALLOC_PTR(dev->od_ost_map);
 341         if (dev->od_ost_map == NULL)
 342                 RETURN(-ENOMEM);
 343
 344         /* to get subdir count from last_rcvd */
 345         rc = osd_last_rcvd_subdir_count(dev);
 346         if (rc < 0) {
 347                 OBD_FREE_PTR(dev->od_ost_map);
 348                 RETURN(rc);
 349         }
 350
 351         dev->od_ost_map->om_subdir_count = rc;
 352         rc = 0;
 353
 354         CFS_INIT_LIST_HEAD(&dev->od_ost_map->om_seq_list);
 355         rwlock_init(&dev->od_ost_map->om_seq_list_lock);
 356         sema_init(&dev->od_ost_map->om_dir_init_sem, 1);
 357
 358         LASSERT(dev->od_fsops);
 359         osd_push_ctxt(dev, &new, &save);
 360
 361         d = simple_mkdir(rootd, dev->od_mnt, "O", 0755, 1);
 362         if (IS_ERR(d))
 363                 GOTO(cleanup, rc = PTR_ERR(d));
 364
 365         ldiskfs_set_inode_state(d->d_inode, LDISKFS_STATE_LUSTRE_NO_OI);
 366         dev->od_ost_map->om_root = d;
 367
 368 cleanup:
 369         osd_pop_ctxt(dev, &new, &save);
 370         if (IS_ERR(d)) {
 371                 OBD_FREE_PTR(dev->od_ost_map);
 372                 RETURN(PTR_ERR(d));
 373         }
 374
 375         RETURN(rc);
 376 }
 377
 378 static void osd_seq_free(struct osd_obj_map *map,
 379                          struct osd_obj_seq *osd_seq)
 380 {
 381         int j;
 382
 383         cfs_list_del_init(&osd_seq->oos_seq_list);
 384
 385         if (osd_seq->oos_dirs) {
 386                 for (j = 0; j < osd_seq->oos_subdir_count; j++) {
 387                         if (osd_seq->oos_dirs[j])
 388                                 dput(osd_seq->oos_dirs[j]);
 389                 }
 390                 OBD_FREE(osd_seq->oos_dirs,
 391                          sizeof(struct dentry *) * osd_seq->oos_subdir_count);
 392         }
 393
 394         if (osd_seq->oos_root)
 395                 dput(osd_seq->oos_root);
 396
 397         OBD_FREE_PTR(osd_seq);
 398 }
 399
 400 static void osd_ost_fini(struct osd_device *osd)
 401 {
 402         struct osd_obj_seq    *osd_seq;
 403         struct osd_obj_seq    *tmp;
 404         struct osd_obj_map    *map = osd->od_ost_map;
 405         ENTRY;
 406
 407         if (map == NULL)
 408                 return;
 409
 410         write_lock(&map->om_seq_list_lock);
 411         cfs_list_for_each_entry_safe(osd_seq, tmp,
 412                                      &map->om_seq_list,
 413                                      oos_seq_list) {
 414                 osd_seq_free(map, osd_seq);
 415         }
 416         write_unlock(&map->om_seq_list_lock);
 417         if (map->om_root)
 418                 dput(map->om_root);
 419         OBD_FREE_PTR(map);
 420         osd->od_ost_map = NULL;
 421         EXIT;
 422 }
 423
 424 int osd_obj_map_init(const struct lu_env *env, struct osd_device *dev)
 425 {
 426         int rc;
 427         ENTRY;
 428
 429         /* prepare structures for OST */
 430         rc = osd_ost_init(dev);
 431         if (rc)
 432                 RETURN(rc);
 433
 434         /* prepare structures for MDS */
 435         rc = osd_mdt_init(env, dev);
 436
 437         RETURN(rc);
 438 }
 439
 440 struct osd_obj_seq *osd_seq_find_locked(struct osd_obj_map *map, obd_seq seq)
 441 {
 442         struct osd_obj_seq *osd_seq;
 443
 444         cfs_list_for_each_entry(osd_seq, &map->om_seq_list, oos_seq_list) {
 445                 if (osd_seq->oos_seq == seq)
 446                         return osd_seq;
 447         }
 448         return NULL;
 449 }
 450
 451 struct osd_obj_seq *osd_seq_find(struct osd_obj_map *map, obd_seq seq)
 452 {
 453         struct osd_obj_seq *osd_seq;
 454
 455         read_lock(&map->om_seq_list_lock);
 456         osd_seq = osd_seq_find_locked(map, seq);
 457         read_unlock(&map->om_seq_list_lock);
 458         return osd_seq;
 459 }
 460
 461 void osd_obj_map_fini(struct osd_device *dev)
 462 {
 463         osd_ost_fini(dev);
 464         osd_mdt_fini(dev);
 465 }
 466
 467 static int osd_obj_del_entry(struct osd_thread_info *info,
 468                              struct osd_device *osd,
 469                              struct dentry *dird, char *name,
 470                              struct thandle *th)
 471 {
 472         struct ldiskfs_dir_entry_2 *de;
 473         struct buffer_head         *bh;
 474         struct osd_thandle         *oh;
 475         struct dentry              *child;
 476         struct inode               *dir = dird->d_inode;
 477         int                         rc;
 478         ENTRY;
 479
 480         oh = container_of(th, struct osd_thandle, ot_super);
 481         LASSERT(oh->ot_handle != NULL);
 482         LASSERT(oh->ot_handle->h_transaction != NULL);
 483
 484
 485         child = &info->oti_child_dentry;
 486         child->d_name.hash = 0;
 487         child->d_name.name = name;
 488         child->d_name.len = strlen(name);
 489         child->d_parent = dird;
 490         child->d_inode = NULL;
 491
 492         ll_vfs_dq_init(dir);
 493         mutex_lock(&dir->i_mutex);
 494         rc = -ENOENT;
 495         bh = osd_ldiskfs_find_entry(dir, child, &de, NULL);
 496         if (bh) {
 497                 rc = ldiskfs_delete_entry(oh->ot_handle, dir, de, bh);
 498                 brelse(bh);
 499         }
 500         mutex_unlock(&dir->i_mutex);
 501
 502         RETURN(rc);
 503 }
 504
 505 int osd_obj_add_entry(struct osd_thread_info *info,
 506                       struct osd_device *osd,
 507                       struct dentry *dir, char *name,
 508                       const struct osd_inode_id *id,
 509                       struct thandle *th)
 510 {
 511         struct osd_thandle *oh;
 512         struct dentry *child;
 513         struct inode *inode;
 514         int rc;
 515
 516         ENTRY;
 517
 518         oh = container_of(th, struct osd_thandle, ot_super);
 519         LASSERT(oh->ot_handle != NULL);
 520         LASSERT(oh->ot_handle->h_transaction != NULL);
 521
 522         inode = &info->oti_inode;
 523         inode->i_sb = osd_sb(osd);
 524         osd_id_to_inode(inode, id);
 525         inode->i_mode = S_IFREG; /* for type in ldiskfs dir entry */
 526
 527         child = &info->oti_child_dentry;
 528         child->d_name.hash = 0;
 529         child->d_name.name = name;
 530         child->d_name.len = strlen(name);
 531         child->d_parent = dir;
 532         child->d_inode = inode;
 533
 534         ll_vfs_dq_init(dir->d_inode);
 535         mutex_lock(&dir->d_inode->i_mutex);
 536         rc = osd_ldiskfs_add_entry(oh->ot_handle, child, inode, NULL);
 537         mutex_unlock(&dir->d_inode->i_mutex);
 538
 539         RETURN(rc);
 540 }
 541
 542 /**
 543  * Use LPU64 for legacy OST sequences, but use LPX64i for new
 544  * sequences names, so that the O/{seq}/dN/{oid} more closely
 545  * follows the DFID/PFID format. This makes it easier to map from
 546  * debug messages to objects in the future, and the legacy space
 547  * of FID_SEQ_OST_MDT0 will be unused in the future.
 548  **/
 549 static inline void osd_seq_name(char *seq_name, obd_seq seq)
 550 {
 551         sprintf(seq_name, (fid_seq_is_rsvd(seq) ||
 552                            fid_seq_is_mdt0(seq)) ? LPU64 : LPX64i,
 553                 fid_seq_is_idif(seq) ? 0 : seq);
 554 }
 555
 556 static inline void osd_oid_name(char *name, const struct lu_fid *fid, obd_id id)
 557 {
 558         sprintf(name, (fid_seq_is_rsvd(fid_seq(fid)) ||
 559                 fid_seq_is_mdt0(fid_seq(fid)) ||
 560                 fid_seq_is_idif(fid_seq(fid))) ? LPU64 : LPX64i, id);
 561 }
 562
 563 /* external locking is required */
 564 static int osd_seq_load_locked(struct osd_device *osd,
 565                                struct osd_obj_seq *osd_seq)
 566 {
 567         struct osd_obj_map  *map = osd->od_ost_map;
 568         struct dentry       *seq_dir;
 569         int                 rc = 0;
 570         int                 i;
 571         char                seq_name[32];
 572         ENTRY;
 573
 574         if (osd_seq->oos_root != NULL)
 575                 RETURN(0);
 576
 577         LASSERT(map);
 578         LASSERT(map->om_root);
 579
 580         osd_seq_name(seq_name, osd_seq->oos_seq);
 581
 582         seq_dir = simple_mkdir(map->om_root, osd->od_mnt, seq_name, 0755, 1);
 583         if (IS_ERR(seq_dir))
 584                 GOTO(out_err, rc = PTR_ERR(seq_dir));
 585         else if (seq_dir->d_inode == NULL)
 586                 GOTO(out_put, rc = -EFAULT);
 587
 588         ldiskfs_set_inode_state(seq_dir->d_inode, LDISKFS_STATE_LUSTRE_NO_OI);
 589         osd_seq->oos_root = seq_dir;
 590
 591         LASSERT(osd_seq->oos_dirs == NULL);
 592         OBD_ALLOC(osd_seq->oos_dirs,
 593                   sizeof(seq_dir) * osd_seq->oos_subdir_count);
 594         if (osd_seq->oos_dirs == NULL)
 595                 GOTO(out_put, rc = -ENOMEM);
 596
 597         for (i = 0; i < osd_seq->oos_subdir_count; i++) {
 598                 struct dentry   *dir;
 599                 char         name[32];
 600
 601                 sprintf(name, "d%u", i);
 602                 dir = simple_mkdir(osd_seq->oos_root, osd->od_mnt, name,
 603                                    0700, 1);
 604                 if (IS_ERR(dir)) {
 605                         rc = PTR_ERR(dir);
 606                 } else if (dir->d_inode) {
 607                         ldiskfs_set_inode_state(dir->d_inode,
 608                                                 LDISKFS_STATE_LUSTRE_NO_OI);
 609                         osd_seq->oos_dirs[i] = dir;
 610                         rc = 0;
 611                 } else {
 612                         LBUG();
 613                 }
 614         }
 615
 616         if (rc != 0)
 617                 osd_seq_free(map, osd_seq);
 618 out_put:
 619         if (rc != 0) {
 620                 dput(seq_dir);
 621                 osd_seq->oos_root = NULL;
 622         }
 623 out_err:
 624         RETURN(rc);
 625 }
 626
 627 struct osd_obj_seq *osd_seq_load(struct osd_device *osd, obd_seq seq)
 628 {
 629         struct osd_obj_map      *map;
 630         struct osd_obj_seq      *osd_seq;
 631         int                     rc = 0;
 632         ENTRY;
 633
 634         map = osd->od_ost_map;
 635         LASSERT(map);
 636         LASSERT(map->om_root);
 637
 638         osd_seq = osd_seq_find(map, seq);
 639         if (likely(osd_seq != NULL))
 640                 RETURN(osd_seq);
 641
 642         /* Serializing init process */
 643         down(&map->om_dir_init_sem);
 644
 645         /* Check whether the seq has been added */
 646         read_lock(&map->om_seq_list_lock);
 647         osd_seq = osd_seq_find_locked(map, seq);
 648         if (osd_seq != NULL) {
 649                 read_unlock(&map->om_seq_list_lock);
 650                 GOTO(cleanup, rc = 0);
 651         }
 652         read_unlock(&map->om_seq_list_lock);
 653
 654         OBD_ALLOC_PTR(osd_seq);
 655         if (osd_seq == NULL)
 656                 GOTO(cleanup, rc = -ENOMEM);
 657
 658         CFS_INIT_LIST_HEAD(&osd_seq->oos_seq_list);
 659         osd_seq->oos_seq = seq;
 660         /* Init subdir count to be 32, but each seq can have
 661          * different subdir count */
 662         osd_seq->oos_subdir_count = map->om_subdir_count;
 663         rc = osd_seq_load_locked(osd, osd_seq);
 664         if (rc != 0)
 665                 GOTO(cleanup, rc);
 666
 667         write_lock(&map->om_seq_list_lock);
 668         cfs_list_add(&osd_seq->oos_seq_list, &map->om_seq_list);
 669         write_unlock(&map->om_seq_list_lock);
 670
 671 cleanup:
 672         up(&map->om_dir_init_sem);
 673         if (rc != 0) {
 674                 if (osd_seq != NULL)
 675                         OBD_FREE_PTR(osd_seq);
 676                 RETURN(ERR_PTR(rc));
 677         }
 678
 679         RETURN(osd_seq);
 680 }
 681
 682 int osd_obj_map_lookup(struct osd_thread_info *info, struct osd_device *dev,
 683                        const struct lu_fid *fid, struct osd_inode_id *id)
 684 {
 685         struct osd_obj_map              *map;
 686         struct osd_obj_seq              *osd_seq;
 687         struct dentry                   *d_seq;
 688         struct dentry                   *child;
 689         struct ost_id                   *ostid = &info->oti_ostid;
 690         int                             dirn;
 691         char                            name[32];
 692         struct ldiskfs_dir_entry_2      *de;
 693         struct buffer_head              *bh;
 694         struct inode                    *dir;
 695         struct inode                    *inode;
 696         ENTRY;
 697
 698         /* on the very first lookup we find and open directories */
 699
 700         map = dev->od_ost_map;
 701         LASSERT(map);
 702         LASSERT(map->om_root);
 703
 704         fid_to_ostid(fid, ostid);
 705         osd_seq = osd_seq_load(dev, ostid_seq(ostid));
 706         if (IS_ERR(osd_seq))
 707                 RETURN(PTR_ERR(osd_seq));
 708
 709         dirn = ostid_id(ostid) & (osd_seq->oos_subdir_count - 1);
 710         d_seq = osd_seq->oos_dirs[dirn];
 711         LASSERT(d_seq);
 712
 713         osd_oid_name(name, fid, ostid_id(ostid));
 714
 715         child = &info->oti_child_dentry;
 716         child->d_parent = d_seq;
 717         child->d_name.hash = 0;
 718         child->d_name.name = name;
 719         /* XXX: we can use rc from sprintf() instead of strlen() */
 720         child->d_name.len = strlen(name);
 721
 722         dir = d_seq->d_inode;
 723         mutex_lock(&dir->i_mutex);
 724         bh = osd_ldiskfs_find_entry(dir, child, &de, NULL);
 725         mutex_unlock(&dir->i_mutex);
 726
 727         if (bh == NULL)
 728                 RETURN(-ENOENT);
 729
 730         osd_id_gen(id, le32_to_cpu(de->inode), OSD_OII_NOGEN);
 731         brelse(bh);
 732
 733         inode = osd_iget(info, dev, id);
 734         if (IS_ERR(inode))
 735                 RETURN(PTR_ERR(inode));
 736
 737         iput(inode);
 738         RETURN(0);
 739 }
 740
 741 int osd_obj_map_insert(struct osd_thread_info *info,
 742                        struct osd_device *osd,
 743                        const struct lu_fid *fid,
 744                        const struct osd_inode_id *id,
 745                        struct thandle *th)
 746 {
 747         struct osd_obj_map      *map;
 748         struct osd_obj_seq      *osd_seq;
 749         struct dentry           *d;
 750         struct ost_id           *ostid = &info->oti_ostid;
 751         int                     dirn, rc = 0;
 752         char                    name[32];
 753         ENTRY;
 754
 755         map = osd->od_ost_map;
 756         LASSERT(map);
 757
 758         /* map fid to seq:objid */
 759         fid_to_ostid(fid, ostid);
 760
 761         osd_seq = osd_seq_load(osd, ostid_seq(ostid));
 762         if (IS_ERR(osd_seq))
 763                 RETURN(PTR_ERR(osd_seq));
 764
 765         dirn = ostid_id(ostid) & (osd_seq->oos_subdir_count - 1);
 766         d = osd_seq->oos_dirs[dirn];
 767         LASSERT(d);
 768
 769         osd_oid_name(name, fid, ostid_id(ostid));
 770         rc = osd_obj_add_entry(info, osd, d, name, id, th);
 771
 772         RETURN(rc);
 773 }
 774
 775 int osd_obj_map_delete(struct osd_thread_info *info, struct osd_device *osd,
 776                        const struct lu_fid *fid, struct thandle *th)
 777 {
 778         struct osd_obj_map      *map;
 779         struct osd_obj_seq      *osd_seq;
 780         struct dentry           *d;
 781         struct ost_id           *ostid = &info->oti_ostid;
 782         int                     dirn, rc = 0;
 783         char                    name[32];
 784         ENTRY;
 785
 786         map = osd->od_ost_map;
 787         LASSERT(map);
 788
 789         /* map fid to seq:objid */
 790         fid_to_ostid(fid, ostid);
 791
 792         osd_seq = osd_seq_load(osd, ostid_seq(ostid));
 793         if (IS_ERR(osd_seq))
 794                 GOTO(cleanup, rc = PTR_ERR(osd_seq));
 795
 796         dirn = ostid_id(ostid) & (osd_seq->oos_subdir_count - 1);
 797         d = osd_seq->oos_dirs[dirn];
 798         LASSERT(d);
 799
 800         osd_oid_name(name, fid, ostid_id(ostid));
 801         rc = osd_obj_del_entry(info, osd, d, name, th);
 802 cleanup:
 803         RETURN(rc);
 804 }
 805
 806 int osd_obj_spec_insert(struct osd_thread_info *info, struct osd_device *osd,
 807                         const struct lu_fid *fid,
 808                         const struct osd_inode_id *id,
 809                         struct thandle *th)
 810 {
 811         struct osd_obj_map      *map = osd->od_ost_map;
 812         struct dentry           *root = osd_sb(osd)->s_root;
 813         char                    *name;
 814         int                     rc = 0;
 815         ENTRY;
 816
 817         if (fid_is_last_id(fid)) {
 818                 struct osd_obj_seq      *osd_seq;
 819
 820                 /* on creation of LAST_ID we create O/<seq> hierarchy */
 821                 LASSERT(map);
 822                 osd_seq = osd_seq_load(osd, fid_seq(fid));
 823                 if (IS_ERR(osd_seq))
 824                         RETURN(PTR_ERR(osd_seq));
 825                 rc = osd_obj_add_entry(info, osd, osd_seq->oos_root,
 826                                        "LAST_ID", id, th);
 827         } else {
 828                 name = osd_lf_fid2name(fid);
 829                 if (name == NULL)
 830                         CWARN("UNKNOWN COMPAT FID "DFID"\n", PFID(fid));
 831                 else if (name[0])
 832                         rc = osd_obj_add_entry(info, osd, root, name, id, th);
 833         }
 834
 835         RETURN(rc);
 836 }
 837
 838 int osd_obj_spec_lookup(struct osd_thread_info *info, struct osd_device *osd,
 839                         const struct lu_fid *fid, struct osd_inode_id *id)
 840 {
 841         struct dentry   *root;
 842         struct dentry   *dentry;
 843         struct inode    *inode;
 844         char            *name;
 845         int             rc = -ENOENT;
 846         ENTRY;
 847
 848         if (fid_is_last_id(fid)) {
 849                 struct osd_obj_seq *osd_seq;
 850
 851                 osd_seq = osd_seq_load(osd, fid_seq(fid));
 852                 if (IS_ERR(osd_seq))
 853                         RETURN(PTR_ERR(osd_seq));
 854                 root = osd_seq->oos_root;
 855                 name = "LAST_ID";
 856         } else {
 857                 root = osd_sb(osd)->s_root;
 858                 name = osd_lf_fid2name(fid);
 859                 if (name == NULL || strlen(name) == 0)
 860                         RETURN(-ENOENT);
 861         }
 862
 863         dentry = ll_lookup_one_len(name, root, strlen(name));
 864         if (!IS_ERR(dentry)) {
 865                 inode = dentry->d_inode;
 866                 if (inode) {
 867                         if (is_bad_inode(inode)) {
 868                                 rc = -EIO;
 869                         } else {
 870                                 osd_id_gen(id, inode->i_ino,
 871                                            inode->i_generation);
 872                                 rc = 0;
 873                         }
 874                 }
 875                 /* if dentry is accessible after osd_compat_spec_insert it
 876                  * will still contain NULL inode, so don't keep it in cache */
 877                 d_invalidate(dentry);
 878                 dput(dentry);
 879         }
 880
 881         RETURN(rc);
 882 }