lustre/osd-zfs/osd_index.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) 2009, 2010, Oracle and/or its affiliates. All rights reserved.
  28  * Use is subject to license terms.
  29  *
  30  * Copyright (c) 2012, 2014, 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-zfs/osd_index.c
  37  *
  38  * Author: Alex Zhuravlev <bzzz@whamcloud.com>
  39  * Author: Mike Pershin <tappro@whamcloud.com>
  40  */
  41
  42 #define DEBUG_SUBSYSTEM S_OSD
  43
  44 #include <lustre_ver.h>
  45 #include <libcfs/libcfs.h>
  46 #include <obd_support.h>
  47 #include <lustre_net.h>
  48 #include <obd.h>
  49 #include <obd_class.h>
  50 #include <lustre_disk.h>
  51 #include <lustre_fid.h>
  52
  53 #include "osd_internal.h"
  54
  55 #include <sys/dnode.h>
  56 #include <sys/dbuf.h>
  57 #include <sys/spa.h>
  58 #include <sys/stat.h>
  59 #include <sys/zap.h>
  60 #include <sys/spa_impl.h>
  61 #include <sys/zfs_znode.h>
  62 #include <sys/dmu_tx.h>
  63 #include <sys/dmu_objset.h>
  64 #include <sys/dsl_prop.h>
  65 #include <sys/sa_impl.h>
  66 #include <sys/txg.h>
  67
  68 static inline int osd_object_is_zap(dmu_buf_t *db)
  69 {
  70         dmu_buf_impl_t *dbi = (dmu_buf_impl_t *) db;
  71         dnode_t *dn;
  72         int rc;
  73
  74         DB_DNODE_ENTER(dbi);
  75         dn = DB_DNODE(dbi);
  76         rc = (dn->dn_type == DMU_OT_DIRECTORY_CONTENTS ||
  77                         dn->dn_type == DMU_OT_USERGROUP_USED);
  78         DB_DNODE_EXIT(dbi);
  79
  80         return rc;
  81 }
  82
  83 /* We don't actually have direct access to the zap_hashbits() function
  84  * so just pretend like we do for now.  If this ever breaks we can look at
  85  * it at that time. */
  86 #define zap_hashbits(zc) 48
  87 /*
  88  * ZFS hash format:
  89  * | cd (16 bits) | hash (48 bits) |
  90  * we need it in other form:
  91  * |0| hash (48 bit) | cd (15 bit) |
  92  * to be a full 64-bit ordered hash so that Lustre readdir can use it to merge
  93  * the readdir hashes from multiple directory stripes uniformly on the client.
  94  * Another point is sign bit, the hash range should be in [0, 2^63-1] because
  95  * loff_t (for llseek) needs to be a positive value.  This means the "cd" field
  96  * should only be the low 15 bits.
  97  */
  98 uint64_t osd_zap_cursor_serialize(zap_cursor_t *zc)
  99 {
 100         uint64_t zfs_hash = zap_cursor_serialize(zc) & (~0ULL >> 1);
 101
 102         return (zfs_hash >> zap_hashbits(zc)) |
 103                 (zfs_hash << (63 - zap_hashbits(zc)));
 104 }
 105
 106 void osd_zap_cursor_init_serialized(zap_cursor_t *zc, struct objset *os,
 107                                     uint64_t id, uint64_t dirhash)
 108 {
 109         uint64_t zfs_hash = ((dirhash << zap_hashbits(zc)) & (~0ULL >> 1)) |
 110                 (dirhash >> (63 - zap_hashbits(zc)));
 111
 112         zap_cursor_init_serialized(zc, os, id, zfs_hash);
 113 }
 114
 115 int osd_zap_cursor_init(zap_cursor_t **zc, struct objset *os,
 116                         uint64_t id, uint64_t dirhash)
 117 {
 118         zap_cursor_t *t;
 119
 120         OBD_ALLOC_PTR(t);
 121         if (unlikely(t == NULL))
 122                 return -ENOMEM;
 123
 124         osd_zap_cursor_init_serialized(t, os, id, dirhash);
 125         *zc = t;
 126
 127         return 0;
 128 }
 129
 130 void osd_zap_cursor_fini(zap_cursor_t *zc)
 131 {
 132         zap_cursor_fini(zc);
 133         OBD_FREE_PTR(zc);
 134 }
 135
 136 static inline void osd_obj_cursor_init_serialized(zap_cursor_t *zc,
 137                                                  struct osd_object *o,
 138                                                  uint64_t dirhash)
 139 {
 140         struct osd_device *d = osd_obj2dev(o);
 141         osd_zap_cursor_init_serialized(zc, d->od_os,
 142                                        o->oo_db->db_object, dirhash);
 143 }
 144
 145 static inline int osd_obj_cursor_init(zap_cursor_t **zc, struct osd_object *o,
 146                         uint64_t dirhash)
 147 {
 148         struct osd_device *d = osd_obj2dev(o);
 149         return osd_zap_cursor_init(zc, d->od_os, o->oo_db->db_object, dirhash);
 150 }
 151
 152 static struct dt_it *osd_index_it_init(const struct lu_env *env,
 153                                        struct dt_object *dt,
 154                                        __u32 unused,
 155                                        struct lustre_capa *capa)
 156 {
 157         struct osd_thread_info  *info = osd_oti_get(env);
 158         struct osd_zap_it       *it;
 159         struct osd_object       *obj = osd_dt_obj(dt);
 160         struct lu_object        *lo  = &dt->do_lu;
 161         int                      rc;
 162         ENTRY;
 163
 164         /* XXX: check capa ? */
 165
 166         LASSERT(lu_object_exists(lo));
 167         LASSERT(obj->oo_db);
 168         LASSERT(osd_object_is_zap(obj->oo_db));
 169         LASSERT(info);
 170
 171         if (info->oti_it_inline) {
 172                 OBD_ALLOC_PTR(it);
 173                 if (it == NULL)
 174                         RETURN(ERR_PTR(-ENOMEM));
 175         } else {
 176                 it = &info->oti_it_zap;
 177                 info->oti_it_inline = 1;
 178         }
 179
 180         rc = osd_obj_cursor_init(&it->ozi_zc, obj, 0);
 181         if (rc != 0) {
 182                 if (it != &info->oti_it_zap)
 183                         OBD_FREE_PTR(it);
 184                 else
 185                         info->oti_it_inline = 0;
 186
 187                 RETURN(ERR_PTR(rc));
 188         }
 189
 190         it->ozi_obj   = obj;
 191         it->ozi_capa  = capa;
 192         it->ozi_reset = 1;
 193         lu_object_get(lo);
 194
 195         RETURN((struct dt_it *)it);
 196 }
 197
 198 static void osd_index_it_fini(const struct lu_env *env, struct dt_it *di)
 199 {
 200         struct osd_thread_info  *info   = osd_oti_get(env);
 201         struct osd_zap_it       *it     = (struct osd_zap_it *)di;
 202         struct osd_object       *obj;
 203         ENTRY;
 204
 205         LASSERT(it);
 206         LASSERT(it->ozi_obj);
 207
 208         obj = it->ozi_obj;
 209
 210         osd_zap_cursor_fini(it->ozi_zc);
 211         lu_object_put(env, &obj->oo_dt.do_lu);
 212         if (it != &info->oti_it_zap)
 213                 OBD_FREE_PTR(it);
 214         else
 215                 info->oti_it_inline = 0;
 216
 217         EXIT;
 218 }
 219
 220
 221 static void osd_index_it_put(const struct lu_env *env, struct dt_it *di)
 222 {
 223         /* PBS: do nothing : ref are incremented at retrive and decreamented
 224          *      next/finish. */
 225 }
 226
 227 static inline void osd_it_append_attrs(struct lu_dirent *ent, __u32 attr,
 228                                        int len, __u16 type)
 229 {
 230         const unsigned    align = sizeof(struct luda_type) - 1;
 231         struct luda_type *lt;
 232
 233         /* check if file type is required */
 234         if (attr & LUDA_TYPE) {
 235                 len = (len + align) & ~align;
 236
 237                 lt = (void *)ent->lde_name + len;
 238                 lt->lt_type = cpu_to_le16(DTTOIF(type));
 239                 ent->lde_attrs |= LUDA_TYPE;
 240         }
 241
 242         ent->lde_attrs = cpu_to_le32(ent->lde_attrs);
 243 }
 244
 245 /*
 246  * as we don't know FID, we can't use LU object, so this function
 247  * partially duplicate __osd_xattr_get() which is built around
 248  * LU-object and uses it to cache data like regular EA dnode, etc
 249  */
 250 static int osd_find_parent_by_dnode(const struct lu_env *env,
 251                                     struct dt_object *o,
 252                                     struct lu_fid *fid)
 253 {
 254         struct osd_device       *osd = osd_obj2dev(osd_dt_obj(o));
 255         struct lustre_mdt_attrs *lma;
 256         struct lu_buf            buf;
 257         sa_handle_t             *sa_hdl;
 258         nvlist_t                *nvbuf = NULL;
 259         uchar_t                 *value;
 260         uint64_t                 dnode;
 261         int                      rc, size;
 262         ENTRY;
 263
 264         /* first of all, get parent dnode from own attributes */
 265         LASSERT(osd_dt_obj(o)->oo_db);
 266         rc = -sa_handle_get(osd->od_os, osd_dt_obj(o)->oo_db->db_object,
 267                             NULL, SA_HDL_PRIVATE, &sa_hdl);
 268         if (rc)
 269                 RETURN(rc);
 270
 271         dnode = ZFS_NO_OBJECT;
 272         rc = -sa_lookup(sa_hdl, SA_ZPL_PARENT(osd), &dnode, 8);
 273         sa_handle_destroy(sa_hdl);
 274         if (rc)
 275                 RETURN(rc);
 276
 277         /* now get EA buffer */
 278         rc = __osd_xattr_load(osd, dnode, &nvbuf);
 279         if (rc)
 280                 GOTO(regular, rc);
 281
 282         /* XXX: if we get that far.. should we cache the result? */
 283
 284         /* try to find LMA attribute */
 285         LASSERT(nvbuf != NULL);
 286         rc = -nvlist_lookup_byte_array(nvbuf, XATTR_NAME_LMA, &value, &size);
 287         if (rc == 0 && size >= sizeof(*lma)) {
 288                 lma = (struct lustre_mdt_attrs *)value;
 289                 lustre_lma_swab(lma);
 290                 *fid = lma->lma_self_fid;
 291                 GOTO(out, rc = 0);
 292         }
 293
 294 regular:
 295         /* no LMA attribute in SA, let's try regular EA */
 296
 297         /* first of all, get parent dnode storing regular EA */
 298         rc = -sa_handle_get(osd->od_os, dnode, NULL, SA_HDL_PRIVATE, &sa_hdl);
 299         if (rc)
 300                 GOTO(out, rc);
 301
 302         dnode = ZFS_NO_OBJECT;
 303         rc = -sa_lookup(sa_hdl, SA_ZPL_XATTR(osd), &dnode, 8);
 304         sa_handle_destroy(sa_hdl);
 305         if (rc)
 306                 GOTO(out, rc);
 307
 308         CLASSERT(sizeof(*lma) <= sizeof(osd_oti_get(env)->oti_buf));
 309         buf.lb_buf = osd_oti_get(env)->oti_buf;
 310         buf.lb_len = sizeof(osd_oti_get(env)->oti_buf);
 311
 312         /* now try to find LMA */
 313         rc = __osd_xattr_get_large(env, osd, dnode, &buf,
 314                                    XATTR_NAME_LMA, &size);
 315         if (rc == 0 && size >= sizeof(*lma)) {
 316                 lma = buf.lb_buf;
 317                 lustre_lma_swab(lma);
 318                 *fid = lma->lma_self_fid;
 319                 GOTO(out, rc = 0);
 320         } else if (rc < 0) {
 321                 GOTO(out, rc);
 322         } else {
 323                 GOTO(out, rc = -EIO);
 324         }
 325
 326 out:
 327         if (nvbuf != NULL)
 328                 nvlist_free(nvbuf);
 329         RETURN(rc);
 330 }
 331
 332 static int osd_find_parent_fid(const struct lu_env *env, struct dt_object *o,
 333                                struct lu_fid *fid)
 334 {
 335         struct link_ea_header  *leh;
 336         struct link_ea_entry   *lee;
 337         struct lu_buf           buf;
 338         int                     rc;
 339         ENTRY;
 340
 341         buf.lb_buf = osd_oti_get(env)->oti_buf;
 342         buf.lb_len = sizeof(osd_oti_get(env)->oti_buf);
 343
 344         rc = osd_xattr_get(env, o, &buf, XATTR_NAME_LINK, BYPASS_CAPA);
 345         if (rc == -ERANGE) {
 346                 rc = osd_xattr_get(env, o, &LU_BUF_NULL,
 347                                    XATTR_NAME_LINK, BYPASS_CAPA);
 348                 if (rc < 0)
 349                         RETURN(rc);
 350                 LASSERT(rc > 0);
 351                 OBD_ALLOC(buf.lb_buf, rc);
 352                 if (buf.lb_buf == NULL)
 353                         RETURN(-ENOMEM);
 354                 buf.lb_len = rc;
 355                 rc = osd_xattr_get(env, o, &buf, XATTR_NAME_LINK, BYPASS_CAPA);
 356         }
 357         if (rc < 0)
 358                 GOTO(out, rc);
 359         if (rc < sizeof(*leh) + sizeof(*lee))
 360                 GOTO(out, rc = -EINVAL);
 361
 362         leh = buf.lb_buf;
 363         if (leh->leh_magic == __swab32(LINK_EA_MAGIC)) {
 364                 leh->leh_magic = LINK_EA_MAGIC;
 365                 leh->leh_reccount = __swab32(leh->leh_reccount);
 366                 leh->leh_len = __swab64(leh->leh_len);
 367         }
 368         if (leh->leh_magic != LINK_EA_MAGIC)
 369                 GOTO(out, rc = -EINVAL);
 370         if (leh->leh_reccount == 0)
 371                 GOTO(out, rc = -ENODATA);
 372
 373         lee = (struct link_ea_entry *)(leh + 1);
 374         fid_be_to_cpu(fid, (const struct lu_fid *)&lee->lee_parent_fid);
 375         rc = 0;
 376
 377 out:
 378         if (buf.lb_buf != osd_oti_get(env)->oti_buf)
 379                 OBD_FREE(buf.lb_buf, buf.lb_len);
 380
 381 #if 0
 382         /* this block can be enabled for additional verification
 383          * it's trying to match FID from LinkEA vs. FID from LMA */
 384         if (rc == 0) {
 385                 struct lu_fid fid2;
 386                 int rc2;
 387                 rc2 = osd_find_parent_by_dnode(env, o, &fid2);
 388                 if (rc2 == 0)
 389                         if (lu_fid_eq(fid, &fid2) == 0)
 390                                 CERROR("wrong parent: "DFID" != "DFID"\n",
 391                                        PFID(fid), PFID(&fid2));
 392         }
 393 #endif
 394
 395         /* no LinkEA is found, let's try to find the fid in parent's LMA */
 396         if (unlikely(rc != 0))
 397                 rc = osd_find_parent_by_dnode(env, o, fid);
 398
 399         RETURN(rc);
 400 }
 401
 402 static int osd_dir_lookup(const struct lu_env *env, struct dt_object *dt,
 403                           struct dt_rec *rec, const struct dt_key *key,
 404                           struct lustre_capa *capa)
 405 {
 406         struct osd_thread_info *oti = osd_oti_get(env);
 407         struct osd_object  *obj = osd_dt_obj(dt);
 408         struct osd_device  *osd = osd_obj2dev(obj);
 409         char               *name = (char *)key;
 410         int                 rc;
 411         ENTRY;
 412
 413         LASSERT(osd_object_is_zap(obj->oo_db));
 414
 415         if (name[0] == '.') {
 416                 if (name[1] == 0) {
 417                         const struct lu_fid *f = lu_object_fid(&dt->do_lu);
 418                         memcpy(rec, f, sizeof(*f));
 419                         RETURN(1);
 420                 } else if (name[1] == '.' && name[2] == 0) {
 421                         rc = osd_find_parent_fid(env, dt, (struct lu_fid *)rec);
 422                         RETURN(rc == 0 ? 1 : rc);
 423                 }
 424         }
 425
 426         rc = -zap_lookup(osd->od_os, obj->oo_db->db_object,
 427                          (char *)key, 8, sizeof(oti->oti_zde) / 8,
 428                          (void *)&oti->oti_zde);
 429         memcpy(rec, &oti->oti_zde.lzd_fid, sizeof(struct lu_fid));
 430
 431         RETURN(rc == 0 ? 1 : rc);
 432 }
 433
 434 static int osd_declare_dir_insert(const struct lu_env *env,
 435                                   struct dt_object *dt,
 436                                   const struct dt_rec *rec,
 437                                   const struct dt_key *key,
 438                                   struct thandle *th)
 439 {
 440         struct osd_object  *obj = osd_dt_obj(dt);
 441         struct osd_thandle *oh;
 442         uint64_t object;
 443         ENTRY;
 444
 445         LASSERT(th != NULL);
 446         oh = container_of0(th, struct osd_thandle, ot_super);
 447
 448         /* This is for inserting dot/dotdot for new created dir. */
 449         if (obj->oo_db == NULL)
 450                 object = DMU_NEW_OBJECT;
 451         else
 452                 object = obj->oo_db->db_object;
 453
 454         dmu_tx_hold_bonus(oh->ot_tx, object);
 455         dmu_tx_hold_zap(oh->ot_tx, object, TRUE, (char *)key);
 456
 457         RETURN(0);
 458 }
 459
 460 /**
 461  * Find the osd object for given fid.
 462  *
 463  * \param fid need to find the osd object having this fid
 464  *
 465  * \retval osd_object on success
 466  * \retval        -ve on error
 467  */
 468 struct osd_object *osd_object_find(const struct lu_env *env,
 469                                    struct dt_object *dt,
 470                                    const struct lu_fid *fid)
 471 {
 472         struct lu_device         *ludev = dt->do_lu.lo_dev;
 473         struct osd_object        *child = NULL;
 474         struct lu_object         *luch;
 475         struct lu_object         *lo;
 476
 477         /*
 478          * at this point topdev might not exist yet
 479          * (i.e. MGS is preparing profiles). so we can
 480          * not rely on topdev and instead lookup with
 481          * our device passed as topdev. this can't work
 482          * if the object isn't cached yet (as osd doesn't
 483          * allocate lu_header). IOW, the object must be
 484          * in the cache, otherwise lu_object_alloc() crashes
 485          * -bzzz
 486          */
 487         luch = lu_object_find_at(env, ludev, fid, NULL);
 488         if (IS_ERR(luch))
 489                 return (void *)luch;
 490
 491         if (lu_object_exists(luch)) {
 492                 lo = lu_object_locate(luch->lo_header, ludev->ld_type);
 493                 if (lo != NULL)
 494                         child = osd_obj(lo);
 495                 else
 496                         LU_OBJECT_DEBUG(D_ERROR, env, luch,
 497                                         "%s: object can't be located "DFID"\n",
 498                                         osd_dev(ludev)->od_svname, PFID(fid));
 499
 500                 if (child == NULL) {
 501                         lu_object_put(env, luch);
 502                         CERROR("%s: Unable to get osd_object "DFID"\n",
 503                                osd_dev(ludev)->od_svname, PFID(fid));
 504                         child = ERR_PTR(-ENOENT);
 505                 }
 506         } else {
 507                 LU_OBJECT_DEBUG(D_ERROR, env, luch,
 508                                 "%s: lu_object does not exists "DFID"\n",
 509                                 osd_dev(ludev)->od_svname, PFID(fid));
 510                 lu_object_put(env, luch);
 511                 child = ERR_PTR(-ENOENT);
 512         }
 513
 514         return child;
 515 }
 516
 517 /**
 518  * Put the osd object once done with it.
 519  *
 520  * \param obj osd object that needs to be put
 521  */
 522 static inline void osd_object_put(const struct lu_env *env,
 523                                   struct osd_object *obj)
 524 {
 525         lu_object_put(env, &obj->oo_dt.do_lu);
 526 }
 527
 528 static int osd_seq_exists(const struct lu_env *env, struct osd_device *osd,
 529                           u64 seq)
 530 {
 531         struct lu_seq_range     *range = &osd_oti_get(env)->oti_seq_range;
 532         struct seq_server_site  *ss = osd_seq_site(osd);
 533         int                     rc;
 534         ENTRY;
 535
 536         LASSERT(ss != NULL);
 537         LASSERT(ss->ss_server_fld != NULL);
 538
 539         rc = osd_fld_lookup(env, osd, seq, range);
 540         if (rc != 0) {
 541                 CERROR("%s: Can not lookup fld for "LPX64"\n",
 542                        osd_name(osd), seq);
 543                 RETURN(0);
 544         }
 545
 546         RETURN(ss->ss_node_id == range->lsr_index);
 547 }
 548
 549 static int osd_remote_fid(const struct lu_env *env, struct osd_device *osd,
 550                           const struct lu_fid *fid)
 551 {
 552         struct seq_server_site  *ss = osd_seq_site(osd);
 553         ENTRY;
 554
 555         /* FID seqs not in FLDB, must be local seq */
 556         if (unlikely(!fid_seq_in_fldb(fid_seq(fid))))
 557                 RETURN(0);
 558
 559         /* If FLD is not being initialized yet, it only happens during the
 560          * initialization, likely during mgs initialization, and we assume
 561          * this is local FID. */
 562         if (ss == NULL || ss->ss_server_fld == NULL)
 563                 RETURN(0);
 564
 565         /* Only check the local FLDB here */
 566         if (osd_seq_exists(env, osd, fid_seq(fid)))
 567                 RETURN(0);
 568
 569         RETURN(1);
 570 }
 571
 572 /**
 573  *      Inserts (key, value) pair in \a directory object.
 574  *
 575  *      \param  dt      osd index object
 576  *      \param  key     key for index
 577  *      \param  rec     record reference
 578  *      \param  th      transaction handler
 579  *      \param  capa    capability descriptor
 580  *      \param  ignore_quota update should not affect quota
 581  *
 582  *      \retval  0  success
 583  *      \retval -ve failure
 584  */
 585 static int osd_dir_insert(const struct lu_env *env, struct dt_object *dt,
 586                           const struct dt_rec *rec, const struct dt_key *key,
 587                           struct thandle *th, struct lustre_capa *capa,
 588                           int ignore_quota)
 589 {
 590         struct osd_thread_info *oti = osd_oti_get(env);
 591         struct osd_object   *parent = osd_dt_obj(dt);
 592         struct osd_device   *osd = osd_obj2dev(parent);
 593         struct dt_insert_rec *rec1 = (struct dt_insert_rec *)rec;
 594         const struct lu_fid *fid = rec1->rec_fid;
 595         struct osd_thandle  *oh;
 596         struct osd_object   *child = NULL;
 597         __u32                attr;
 598         char                *name = (char *)key;
 599         int                  rc;
 600         ENTRY;
 601
 602         LASSERT(parent->oo_db);
 603         LASSERT(osd_object_is_zap(parent->oo_db));
 604
 605         LASSERT(dt_object_exists(dt));
 606         LASSERT(osd_invariant(parent));
 607
 608         LASSERT(th != NULL);
 609         oh = container_of0(th, struct osd_thandle, ot_super);
 610
 611         rc = osd_remote_fid(env, osd, fid);
 612         if (rc < 0) {
 613                 CERROR("%s: Can not find object "DFID": rc = %d\n",
 614                        osd->od_svname, PFID(fid), rc);
 615                 RETURN(rc);
 616         }
 617
 618         if (unlikely(rc == 1)) {
 619                 /* Insert remote entry */
 620                 memset(&oti->oti_zde.lzd_reg, 0, sizeof(oti->oti_zde.lzd_reg));
 621                 oti->oti_zde.lzd_reg.zde_type = IFTODT(rec1->rec_type & S_IFMT);
 622         } else {
 623                 /*
 624                  * To simulate old Orion setups with ./..  stored in the
 625                  * directories
 626                  */
 627                 /* Insert local entry */
 628                 child = osd_object_find(env, dt, fid);
 629                 if (IS_ERR(child))
 630                         RETURN(PTR_ERR(child));
 631
 632                 LASSERT(child->oo_db);
 633                 if (name[0] == '.') {
 634                         if (name[1] == 0) {
 635                                 /* do not store ".", instead generate it
 636                                  * during iteration */
 637                                 GOTO(out, rc = 0);
 638                         } else if (name[1] == '.' && name[2] == 0) {
 639                                 if (OBD_FAIL_CHECK(OBD_FAIL_LFSCK_BAD_PARENT)) {
 640                                         struct lu_fid tfid = *fid;
 641
 642                                         osd_object_put(env, child);
 643                                         tfid.f_oid--;
 644                                         child = osd_object_find(env, dt, &tfid);
 645                                         if (IS_ERR(child))
 646                                                 RETURN(PTR_ERR(child));
 647
 648                                         LASSERT(child->oo_db);
 649                                 }
 650
 651                                 /* update parent dnode in the child.
 652                                  * later it will be used to generate ".." */
 653                                 rc = osd_object_sa_update(parent,
 654                                                  SA_ZPL_PARENT(osd),
 655                                                  &child->oo_db->db_object,
 656                                                  8, oh);
 657
 658                                 GOTO(out, rc);
 659                         }
 660                 }
 661                 CLASSERT(sizeof(oti->oti_zde.lzd_reg) == 8);
 662                 CLASSERT(sizeof(oti->oti_zde) % 8 == 0);
 663                 attr = child->oo_dt.do_lu.lo_header ->loh_attr;
 664                 oti->oti_zde.lzd_reg.zde_type = IFTODT(attr & S_IFMT);
 665                 oti->oti_zde.lzd_reg.zde_dnode = child->oo_db->db_object;
 666         }
 667
 668         oti->oti_zde.lzd_fid = *fid;
 669         /* Insert (key,oid) into ZAP */
 670         rc = -zap_add(osd->od_os, parent->oo_db->db_object,
 671                       (char *)key, 8, sizeof(oti->oti_zde) / 8,
 672                       (void *)&oti->oti_zde, oh->ot_tx);
 673         if (unlikely(rc == -EEXIST &&
 674                      name[0] == '.' && name[1] == '.' && name[2] == 0))
 675                 /* Update (key,oid) in ZAP */
 676                 rc = -zap_update(osd->od_os, parent->oo_db->db_object,
 677                                 (char *)key, 8, sizeof(oti->oti_zde) / 8,
 678                                 (void *)&oti->oti_zde, oh->ot_tx);
 679
 680 out:
 681         if (child != NULL)
 682                 osd_object_put(env, child);
 683
 684         RETURN(rc);
 685 }
 686
 687 static int osd_declare_dir_delete(const struct lu_env *env,
 688                                   struct dt_object *dt,
 689                                   const struct dt_key *key,
 690                                   struct thandle *th)
 691 {
 692         struct osd_object *obj = osd_dt_obj(dt);
 693         struct osd_thandle *oh;
 694         ENTRY;
 695
 696         LASSERT(dt_object_exists(dt));
 697         LASSERT(osd_invariant(obj));
 698
 699         LASSERT(th != NULL);
 700         oh = container_of0(th, struct osd_thandle, ot_super);
 701
 702         LASSERT(obj->oo_db);
 703         LASSERT(osd_object_is_zap(obj->oo_db));
 704
 705         dmu_tx_hold_zap(oh->ot_tx, obj->oo_db->db_object, TRUE, (char *)key);
 706
 707         RETURN(0);
 708 }
 709
 710 static int osd_dir_delete(const struct lu_env *env, struct dt_object *dt,
 711                           const struct dt_key *key, struct thandle *th,
 712                           struct lustre_capa *capa)
 713 {
 714         struct osd_object *obj = osd_dt_obj(dt);
 715         struct osd_device *osd = osd_obj2dev(obj);
 716         struct osd_thandle *oh;
 717         dmu_buf_t *zap_db = obj->oo_db;
 718         char      *name = (char *)key;
 719         int rc;
 720         ENTRY;
 721
 722         LASSERT(obj->oo_db);
 723         LASSERT(osd_object_is_zap(obj->oo_db));
 724
 725         LASSERT(th != NULL);
 726         oh = container_of0(th, struct osd_thandle, ot_super);
 727
 728         /*
 729          * In Orion . and .. were stored in the directory (not generated upon
 730          * request as now). we preserve them for backward compatibility
 731          */
 732         if (name[0] == '.') {
 733                 if (name[1] == 0) {
 734                         RETURN(0);
 735                 } else if (name[1] == '.' && name[2] == 0) {
 736                         RETURN(0);
 737                 }
 738         }
 739
 740         /* Remove key from the ZAP */
 741         rc = -zap_remove(osd->od_os, zap_db->db_object,
 742                          (char *) key, oh->ot_tx);
 743
 744         if (unlikely(rc && rc != -ENOENT))
 745                 CERROR("%s: zap_remove failed: rc = %d\n", osd->od_svname, rc);
 746
 747         RETURN(rc);
 748 }
 749
 750 static struct dt_it *osd_dir_it_init(const struct lu_env *env,
 751                                      struct dt_object *dt,
 752                                      __u32 unused,
 753                                      struct lustre_capa *capa)
 754 {
 755         struct osd_zap_it *it;
 756
 757         it = (struct osd_zap_it *)osd_index_it_init(env, dt, unused, capa);
 758         if (!IS_ERR(it))
 759                 it->ozi_pos = 0;
 760
 761         RETURN((struct dt_it *)it);
 762 }
 763
 764 /**
 765  *  Move Iterator to record specified by \a key
 766  *
 767  *  \param  di      osd iterator
 768  *  \param  key     key for index
 769  *
 770  *  \retval +ve  di points to record with least key not larger than key
 771  *  \retval  0   di points to exact matched key
 772  *  \retval -ve  failure
 773  */
 774 static int osd_dir_it_get(const struct lu_env *env,
 775                           struct dt_it *di, const struct dt_key *key)
 776 {
 777         struct osd_zap_it *it = (struct osd_zap_it *)di;
 778         struct osd_object *obj = it->ozi_obj;
 779         char              *name = (char *)key;
 780         int                rc;
 781         ENTRY;
 782
 783         LASSERT(it);
 784         LASSERT(it->ozi_zc);
 785
 786         /* reset the cursor */
 787         zap_cursor_fini(it->ozi_zc);
 788         osd_obj_cursor_init_serialized(it->ozi_zc, obj, 0);
 789
 790         /* XXX: implementation of the API is broken at the moment */
 791         LASSERT(((const char *)key)[0] == 0);
 792
 793         if (name[0] == 0) {
 794                 it->ozi_pos = 0;
 795                 RETURN(1);
 796         }
 797
 798         if (name[0] == '.') {
 799                 if (name[1] == 0) {
 800                         it->ozi_pos = 1;
 801                         GOTO(out, rc = 1);
 802                 } else if (name[1] == '.' && name[2] == 0) {
 803                         it->ozi_pos = 2;
 804                         GOTO(out, rc = 1);
 805                 }
 806         }
 807
 808         /* neither . nor .. - some real record */
 809         it->ozi_pos = 3;
 810         rc = +1;
 811
 812 out:
 813         RETURN(rc);
 814 }
 815
 816 static void osd_dir_it_put(const struct lu_env *env, struct dt_it *di)
 817 {
 818         /* PBS: do nothing : ref are incremented at retrive and decreamented
 819          *      next/finish. */
 820 }
 821
 822 /*
 823  * in Orion . and .. were stored in the directory, while ZPL
 824  * and current osd-zfs generate them up on request. so, we
 825  * need to ignore previously stored . and ..
 826  */
 827 static int osd_index_retrieve_skip_dots(struct osd_zap_it *it,
 828                                         zap_attribute_t *za)
 829 {
 830         int rc, isdot;
 831
 832         do {
 833                 rc = -zap_cursor_retrieve(it->ozi_zc, za);
 834
 835                 isdot = 0;
 836                 if (unlikely(rc == 0 && za->za_name[0] == '.')) {
 837                         if (za->za_name[1] == 0) {
 838                                 isdot = 1;
 839                         } else if (za->za_name[1] == '.' &&
 840                                    za->za_name[2] == 0) {
 841                                 isdot = 1;
 842                         }
 843                         if (unlikely(isdot))
 844                                 zap_cursor_advance(it->ozi_zc);
 845                 }
 846         } while (unlikely(rc == 0 && isdot));
 847
 848         return rc;
 849 }
 850
 851 /**
 852  * to load a directory entry at a time and stored it in
 853  * iterator's in-memory data structure.
 854  *
 855  * \param di, struct osd_it_ea, iterator's in memory structure
 856  *
 857  * \retval +ve, iterator reached to end
 858  * \retval   0, iterator not reached to end
 859  * \retval -ve, on error
 860  */
 861 static int osd_dir_it_next(const struct lu_env *env, struct dt_it *di)
 862 {
 863         struct osd_zap_it *it = (struct osd_zap_it *)di;
 864         zap_attribute_t   *za = &osd_oti_get(env)->oti_za;
 865         int                rc;
 866
 867         ENTRY;
 868
 869         /* temp. storage should be enough for any key supported by ZFS */
 870         CLASSERT(sizeof(za->za_name) <= sizeof(it->ozi_name));
 871
 872         /*
 873          * the first ->next() moves the cursor to .
 874          * the second ->next() moves the cursor to ..
 875          * then we get to the real records and have to verify any exist
 876          */
 877         if (it->ozi_pos <= 2) {
 878                 it->ozi_pos++;
 879                 if (it->ozi_pos <=2)
 880                         RETURN(0);
 881
 882         } else {
 883                 zap_cursor_advance(it->ozi_zc);
 884         }
 885
 886         /*
 887          * According to current API we need to return error if its last entry.
 888          * zap_cursor_advance() does not return any value. So we need to call
 889          * retrieve to check if there is any record.  We should make
 890          * changes to Iterator API to not return status for this API
 891          */
 892         rc = osd_index_retrieve_skip_dots(it, za);
 893
 894         if (rc == -ENOENT) /* end of dir */
 895                 RETURN(+1);
 896
 897         RETURN(rc);
 898 }
 899
 900 static struct dt_key *osd_dir_it_key(const struct lu_env *env,
 901                                      const struct dt_it *di)
 902 {
 903         struct osd_zap_it *it = (struct osd_zap_it *)di;
 904         zap_attribute_t   *za = &osd_oti_get(env)->oti_za;
 905         int                rc = 0;
 906         ENTRY;
 907
 908         if (it->ozi_pos <= 1) {
 909                 it->ozi_pos = 1;
 910                 RETURN((struct dt_key *)".");
 911         } else if (it->ozi_pos == 2) {
 912                 RETURN((struct dt_key *)"..");
 913         }
 914
 915         if ((rc = -zap_cursor_retrieve(it->ozi_zc, za)))
 916                 RETURN(ERR_PTR(rc));
 917
 918         strcpy(it->ozi_name, za->za_name);
 919
 920         RETURN((struct dt_key *)it->ozi_name);
 921 }
 922
 923 static int osd_dir_it_key_size(const struct lu_env *env, const struct dt_it *di)
 924 {
 925         struct osd_zap_it *it = (struct osd_zap_it *)di;
 926         zap_attribute_t   *za = &osd_oti_get(env)->oti_za;
 927         int                rc;
 928         ENTRY;
 929
 930         if (it->ozi_pos <= 1) {
 931                 it->ozi_pos = 1;
 932                 RETURN(2);
 933         } else if (it->ozi_pos == 2) {
 934                 RETURN(3);
 935         }
 936
 937         if ((rc = -zap_cursor_retrieve(it->ozi_zc, za)) == 0)
 938                 rc = strlen(za->za_name);
 939
 940         RETURN(rc);
 941 }
 942
 943 static int osd_dir_it_rec(const struct lu_env *env, const struct dt_it *di,
 944                           struct dt_rec *dtrec, __u32 attr)
 945 {
 946         struct osd_zap_it   *it = (struct osd_zap_it *)di;
 947         struct lu_dirent    *lde = (struct lu_dirent *)dtrec;
 948         struct luz_direntry *zde = &osd_oti_get(env)->oti_zde;
 949         zap_attribute_t     *za = &osd_oti_get(env)->oti_za;
 950         int                  rc, namelen;
 951         ENTRY;
 952
 953         if (it->ozi_pos <= 1) {
 954                 lde->lde_hash = cpu_to_le64(1);
 955                 strcpy(lde->lde_name, ".");
 956                 lde->lde_namelen = cpu_to_le16(1);
 957                 lde->lde_fid = *lu_object_fid(&it->ozi_obj->oo_dt.do_lu);
 958                 lde->lde_attrs = LUDA_FID;
 959                 /* append lustre attributes */
 960                 osd_it_append_attrs(lde, attr, 1, IFTODT(S_IFDIR));
 961                 lde->lde_reclen = cpu_to_le16(lu_dirent_calc_size(1, attr));
 962                 it->ozi_pos = 1;
 963                 GOTO(out, rc = 0);
 964
 965         } else if (it->ozi_pos == 2) {
 966                 lde->lde_hash = cpu_to_le64(2);
 967                 strcpy(lde->lde_name, "..");
 968                 lde->lde_namelen = cpu_to_le16(2);
 969                 lde->lde_attrs = LUDA_FID;
 970                 /* append lustre attributes */
 971                 osd_it_append_attrs(lde, attr, 2, IFTODT(S_IFDIR));
 972                 lde->lde_reclen = cpu_to_le16(lu_dirent_calc_size(2, attr));
 973                 rc = osd_find_parent_fid(env, &it->ozi_obj->oo_dt, &lde->lde_fid);
 974
 975                 /* ENOENT happens at the root of filesystem so ignore it */
 976                 if (rc == -ENOENT)
 977                         rc = 0;
 978                 GOTO(out, rc);
 979         }
 980
 981         LASSERT(lde);
 982
 983         rc = -zap_cursor_retrieve(it->ozi_zc, za);
 984         if (unlikely(rc != 0))
 985                 GOTO(out, rc);
 986
 987         lde->lde_hash = cpu_to_le64(osd_zap_cursor_serialize(it->ozi_zc));
 988         namelen = strlen(za->za_name);
 989         if (namelen > NAME_MAX)
 990                 GOTO(out, rc = -EOVERFLOW);
 991         strcpy(lde->lde_name, za->za_name);
 992         lde->lde_namelen = cpu_to_le16(namelen);
 993
 994         if (za->za_integer_length != 8 || za->za_num_integers < 3) {
 995                 CERROR("%s: unsupported direntry format: %d %d\n",
 996                        osd_obj2dev(it->ozi_obj)->od_svname,
 997                        za->za_integer_length, (int)za->za_num_integers);
 998
 999                 GOTO(out, rc = -EIO);
1000         }
1001
1002         rc = -zap_lookup(it->ozi_zc->zc_objset, it->ozi_zc->zc_zapobj,
1003                          za->za_name, za->za_integer_length, 3, zde);
1004         if (rc)
1005                 GOTO(out, rc);
1006
1007         lde->lde_fid = zde->lzd_fid;
1008         lde->lde_attrs = LUDA_FID;
1009
1010         /* append lustre attributes */
1011         osd_it_append_attrs(lde, attr, namelen, zde->lzd_reg.zde_type);
1012
1013         lde->lde_reclen = cpu_to_le16(lu_dirent_calc_size(namelen, attr));
1014
1015 out:
1016         RETURN(rc);
1017 }
1018
1019 static int osd_dir_it_rec_size(const struct lu_env *env, const struct dt_it *di,
1020                                __u32 attr)
1021 {
1022         struct osd_zap_it   *it = (struct osd_zap_it *)di;
1023         zap_attribute_t     *za = &osd_oti_get(env)->oti_za;
1024         size_t               namelen = 0;
1025         int                  rc;
1026         ENTRY;
1027
1028         if (it->ozi_pos <= 1)
1029                 namelen = 1;
1030         else if (it->ozi_pos == 2)
1031                 namelen = 2;
1032
1033         if (namelen > 0) {
1034                 rc = lu_dirent_calc_size(namelen, attr);
1035                 RETURN(rc);
1036         }
1037
1038         rc = -zap_cursor_retrieve(it->ozi_zc, za);
1039         if (unlikely(rc != 0))
1040                 RETURN(rc);
1041
1042         if (za->za_integer_length != 8 || za->za_num_integers < 3) {
1043                 CERROR("%s: unsupported direntry format: %d %d\n",
1044                        osd_obj2dev(it->ozi_obj)->od_svname,
1045                        za->za_integer_length, (int)za->za_num_integers);
1046                 RETURN(-EIO);
1047         }
1048
1049         namelen = strlen(za->za_name);
1050         if (namelen > NAME_MAX)
1051                 RETURN(-EOVERFLOW);
1052
1053         rc = lu_dirent_calc_size(namelen, attr);
1054
1055         RETURN(rc);
1056 }
1057
1058 static __u64 osd_dir_it_store(const struct lu_env *env, const struct dt_it *di)
1059 {
1060         struct osd_zap_it *it = (struct osd_zap_it *)di;
1061         __u64              pos;
1062         ENTRY;
1063
1064         if (it->ozi_pos <= 2)
1065                 pos = it->ozi_pos;
1066         else
1067                 pos = osd_zap_cursor_serialize(it->ozi_zc);
1068
1069         RETURN(pos);
1070 }
1071
1072 /*
1073  * return status :
1074  *  rc == 0 -> end of directory.
1075  *  rc >  0 -> ok, proceed.
1076  *  rc <  0 -> error.  ( EOVERFLOW  can be masked.)
1077  */
1078 static int osd_dir_it_load(const struct lu_env *env,
1079                         const struct dt_it *di, __u64 hash)
1080 {
1081         struct osd_zap_it *it = (struct osd_zap_it *)di;
1082         struct osd_object *obj = it->ozi_obj;
1083         zap_attribute_t   *za = &osd_oti_get(env)->oti_za;
1084         int                rc;
1085         ENTRY;
1086
1087         /* reset the cursor */
1088         zap_cursor_fini(it->ozi_zc);
1089         osd_obj_cursor_init_serialized(it->ozi_zc, obj, hash);
1090
1091         if (hash <= 2) {
1092                 it->ozi_pos = hash;
1093                 rc = +1;
1094         } else {
1095                 it->ozi_pos = 3;
1096                 /* to return whether the end has been reached */
1097                 rc = osd_index_retrieve_skip_dots(it, za);
1098                 if (rc == 0)
1099                         rc = +1;
1100                 else if (rc == -ENOENT)
1101                         rc = 0;
1102         }
1103
1104         RETURN(rc);
1105 }
1106
1107 struct dt_index_operations osd_dir_ops = {
1108         .dio_lookup         = osd_dir_lookup,
1109         .dio_declare_insert = osd_declare_dir_insert,
1110         .dio_insert         = osd_dir_insert,
1111         .dio_declare_delete = osd_declare_dir_delete,
1112         .dio_delete         = osd_dir_delete,
1113         .dio_it     = {
1114                 .init     = osd_dir_it_init,
1115                 .fini     = osd_index_it_fini,
1116                 .get      = osd_dir_it_get,
1117                 .put      = osd_dir_it_put,
1118                 .next     = osd_dir_it_next,
1119                 .key      = osd_dir_it_key,
1120                 .key_size = osd_dir_it_key_size,
1121                 .rec      = osd_dir_it_rec,
1122                 .rec_size = osd_dir_it_rec_size,
1123                 .store    = osd_dir_it_store,
1124                 .load     = osd_dir_it_load
1125         }
1126 };
1127
1128 /*
1129  * Primitives for index files using binary keys.
1130  */
1131
1132 /* key integer_size is 8 */
1133 static int osd_prepare_key_uint64(struct osd_object *o, __u64 *dst,
1134                                   const struct dt_key *src)
1135 {
1136         int size;
1137
1138         LASSERT(dst);
1139         LASSERT(src);
1140
1141         /* align keysize to 64bit */
1142         size = (o->oo_keysize + sizeof(__u64) - 1) / sizeof(__u64);
1143         size *= sizeof(__u64);
1144
1145         LASSERT(size <= MAXNAMELEN);
1146
1147         if (unlikely(size > o->oo_keysize))
1148                 memset(dst + o->oo_keysize, 0, size - o->oo_keysize);
1149         memcpy(dst, (const char *)src, o->oo_keysize);
1150
1151         return (size/sizeof(__u64));
1152 }
1153
1154 static int osd_index_lookup(const struct lu_env *env, struct dt_object *dt,
1155                         struct dt_rec *rec, const struct dt_key *key,
1156                         struct lustre_capa *capa)
1157 {
1158         struct osd_object *obj = osd_dt_obj(dt);
1159         struct osd_device *osd = osd_obj2dev(obj);
1160         __u64             *k = osd_oti_get(env)->oti_key64;
1161         int                rc;
1162         ENTRY;
1163
1164         rc = osd_prepare_key_uint64(obj, k, key);
1165
1166         rc = -zap_lookup_uint64(osd->od_os, obj->oo_db->db_object,
1167                                 k, rc, obj->oo_recusize, obj->oo_recsize,
1168                                 (void *)rec);
1169         RETURN(rc == 0 ? 1 : rc);
1170 }
1171
1172 static int osd_declare_index_insert(const struct lu_env *env,
1173                                     struct dt_object *dt,
1174                                     const struct dt_rec *rec,
1175                                     const struct dt_key *key,
1176                                     struct thandle *th)
1177 {
1178         struct osd_object  *obj = osd_dt_obj(dt);
1179         struct osd_thandle *oh;
1180         ENTRY;
1181
1182         LASSERT(th != NULL);
1183         oh = container_of0(th, struct osd_thandle, ot_super);
1184
1185         LASSERT(obj->oo_db);
1186
1187         dmu_tx_hold_bonus(oh->ot_tx, obj->oo_db->db_object);
1188
1189         /* It is not clear what API should be used for binary keys, so we pass
1190          * a null name which has the side effect of over-reserving space,
1191          * accounting for the worst case. See zap_count_write() */
1192         dmu_tx_hold_zap(oh->ot_tx, obj->oo_db->db_object, TRUE, NULL);
1193
1194         RETURN(0);
1195 }
1196
1197 static int osd_index_insert(const struct lu_env *env, struct dt_object *dt,
1198                             const struct dt_rec *rec, const struct dt_key *key,
1199                             struct thandle *th, struct lustre_capa *capa,
1200                             int ignore_quota)
1201 {
1202         struct osd_object  *obj = osd_dt_obj(dt);
1203         struct osd_device  *osd = osd_obj2dev(obj);
1204         struct osd_thandle *oh;
1205         __u64              *k = osd_oti_get(env)->oti_key64;
1206         int                 rc;
1207         ENTRY;
1208
1209         LASSERT(obj->oo_db);
1210         LASSERT(dt_object_exists(dt));
1211         LASSERT(osd_invariant(obj));
1212         LASSERT(th != NULL);
1213
1214         oh = container_of0(th, struct osd_thandle, ot_super);
1215
1216         rc = osd_prepare_key_uint64(obj, k, key);
1217
1218         /* Insert (key,oid) into ZAP */
1219         rc = -zap_add_uint64(osd->od_os, obj->oo_db->db_object,
1220                              k, rc, obj->oo_recusize, obj->oo_recsize,
1221                              (void *)rec, oh->ot_tx);
1222         RETURN(rc);
1223 }
1224
1225 static int osd_declare_index_delete(const struct lu_env *env,
1226                                     struct dt_object *dt,
1227                                     const struct dt_key *key,
1228                                     struct thandle *th)
1229 {
1230         struct osd_object  *obj = osd_dt_obj(dt);
1231         struct osd_thandle *oh;
1232         ENTRY;
1233
1234         LASSERT(dt_object_exists(dt));
1235         LASSERT(osd_invariant(obj));
1236         LASSERT(th != NULL);
1237         LASSERT(obj->oo_db);
1238
1239         oh = container_of0(th, struct osd_thandle, ot_super);
1240         dmu_tx_hold_zap(oh->ot_tx, obj->oo_db->db_object, TRUE, NULL);
1241
1242         RETURN(0);
1243 }
1244
1245 static int osd_index_delete(const struct lu_env *env, struct dt_object *dt,
1246                             const struct dt_key *key, struct thandle *th,
1247                             struct lustre_capa *capa)
1248 {
1249         struct osd_object  *obj = osd_dt_obj(dt);
1250         struct osd_device  *osd = osd_obj2dev(obj);
1251         struct osd_thandle *oh;
1252         __u64              *k = osd_oti_get(env)->oti_key64;
1253         int                 rc;
1254         ENTRY;
1255
1256         LASSERT(obj->oo_db);
1257         LASSERT(th != NULL);
1258         oh = container_of0(th, struct osd_thandle, ot_super);
1259
1260         rc = osd_prepare_key_uint64(obj, k, key);
1261
1262         /* Remove binary key from the ZAP */
1263         rc = -zap_remove_uint64(osd->od_os, obj->oo_db->db_object,
1264                                 k, rc, oh->ot_tx);
1265         RETURN(rc);
1266 }
1267
1268 static int osd_index_it_get(const struct lu_env *env, struct dt_it *di,
1269                             const struct dt_key *key)
1270 {
1271         struct osd_zap_it *it = (struct osd_zap_it *)di;
1272         struct osd_object *obj = it->ozi_obj;
1273         struct osd_device *osd = osd_obj2dev(obj);
1274         ENTRY;
1275
1276         LASSERT(it);
1277         LASSERT(it->ozi_zc);
1278
1279         /*
1280          * XXX: we need a binary version of zap_cursor_move_to_key()
1281          *      to implement this API */
1282         if (*((const __u64 *)key) != 0)
1283                 CERROR("NOT IMPLEMETED YET (move to "LPX64")\n",
1284                        *((__u64 *)key));
1285
1286         zap_cursor_fini(it->ozi_zc);
1287         zap_cursor_init(it->ozi_zc, osd->od_os, obj->oo_db->db_object);
1288         it->ozi_reset = 1;
1289
1290         RETURN(+1);
1291 }
1292
1293 static int osd_index_it_next(const struct lu_env *env, struct dt_it *di)
1294 {
1295         struct osd_zap_it *it = (struct osd_zap_it *)di;
1296         zap_attribute_t   *za = &osd_oti_get(env)->oti_za;
1297         int                rc;
1298         ENTRY;
1299
1300         if (it->ozi_reset == 0)
1301                 zap_cursor_advance(it->ozi_zc);
1302         it->ozi_reset = 0;
1303
1304         /*
1305          * According to current API we need to return error if it's last entry.
1306          * zap_cursor_advance() does not return any value. So we need to call
1307          * retrieve to check if there is any record.  We should make
1308          * changes to Iterator API to not return status for this API
1309          */
1310         rc = -zap_cursor_retrieve(it->ozi_zc, za);
1311         if (rc == -ENOENT)
1312                 RETURN(+1);
1313
1314         RETURN((rc));
1315 }
1316
1317 static struct dt_key *osd_index_it_key(const struct lu_env *env,
1318                                        const struct dt_it *di)
1319 {
1320         struct osd_zap_it *it = (struct osd_zap_it *)di;
1321         struct osd_object *obj = it->ozi_obj;
1322         zap_attribute_t   *za = &osd_oti_get(env)->oti_za;
1323         int                rc = 0;
1324         ENTRY;
1325
1326         it->ozi_reset = 0;
1327         rc = -zap_cursor_retrieve(it->ozi_zc, za);
1328         if (rc)
1329                 RETURN(ERR_PTR(rc));
1330
1331         /* the binary key is stored in the name */
1332         memcpy(&it->ozi_key, za->za_name, obj->oo_keysize);
1333
1334         RETURN((struct dt_key *)&it->ozi_key);
1335 }
1336
1337 static int osd_index_it_key_size(const struct lu_env *env,
1338                                 const struct dt_it *di)
1339 {
1340         struct osd_zap_it *it = (struct osd_zap_it *)di;
1341         struct osd_object *obj = it->ozi_obj;
1342         RETURN(obj->oo_keysize);
1343 }
1344
1345 static int osd_index_it_rec(const struct lu_env *env, const struct dt_it *di,
1346                             struct dt_rec *rec, __u32 attr)
1347 {
1348         zap_attribute_t   *za = &osd_oti_get(env)->oti_za;
1349         struct osd_zap_it *it = (struct osd_zap_it *)di;
1350         struct osd_object *obj = it->ozi_obj;
1351         struct osd_device *osd = osd_obj2dev(obj);
1352         __u64             *k = osd_oti_get(env)->oti_key64;
1353         int                rc;
1354         ENTRY;
1355
1356         it->ozi_reset = 0;
1357         rc = -zap_cursor_retrieve(it->ozi_zc, za);
1358         if (rc)
1359                 RETURN(rc);
1360
1361         rc = osd_prepare_key_uint64(obj, k, (const struct dt_key *)za->za_name);
1362
1363         rc = -zap_lookup_uint64(osd->od_os, obj->oo_db->db_object,
1364                                 k, rc, obj->oo_recusize, obj->oo_recsize,
1365                                 (void *)rec);
1366         RETURN(rc);
1367 }
1368
1369 static __u64 osd_index_it_store(const struct lu_env *env,
1370                                 const struct dt_it *di)
1371 {
1372         struct osd_zap_it *it = (struct osd_zap_it *)di;
1373
1374         it->ozi_reset = 0;
1375         RETURN((__u64)zap_cursor_serialize(it->ozi_zc));
1376 }
1377
1378 static int osd_index_it_load(const struct lu_env *env, const struct dt_it *di,
1379                              __u64 hash)
1380 {
1381         struct osd_zap_it *it = (struct osd_zap_it *)di;
1382         struct osd_object *obj = it->ozi_obj;
1383         struct osd_device *osd = osd_obj2dev(obj);
1384         zap_attribute_t   *za = &osd_oti_get(env)->oti_za;
1385         int                rc;
1386         ENTRY;
1387
1388         /* reset the cursor */
1389         zap_cursor_fini(it->ozi_zc);
1390         zap_cursor_init_serialized(it->ozi_zc, osd->od_os,
1391                                    obj->oo_db->db_object, hash);
1392         it->ozi_reset = 0;
1393
1394         rc = -zap_cursor_retrieve(it->ozi_zc, za);
1395         if (rc == 0)
1396                 RETURN(+1);
1397         else if (rc == -ENOENT)
1398                 RETURN(0);
1399
1400         RETURN(rc);
1401 }
1402
1403 static struct dt_index_operations osd_index_ops = {
1404         .dio_lookup             = osd_index_lookup,
1405         .dio_declare_insert     = osd_declare_index_insert,
1406         .dio_insert             = osd_index_insert,
1407         .dio_declare_delete     = osd_declare_index_delete,
1408         .dio_delete             = osd_index_delete,
1409         .dio_it = {
1410                 .init           = osd_index_it_init,
1411                 .fini           = osd_index_it_fini,
1412                 .get            = osd_index_it_get,
1413                 .put            = osd_index_it_put,
1414                 .next           = osd_index_it_next,
1415                 .key            = osd_index_it_key,
1416                 .key_size       = osd_index_it_key_size,
1417                 .rec            = osd_index_it_rec,
1418                 .store          = osd_index_it_store,
1419                 .load           = osd_index_it_load
1420         }
1421 };
1422
1423 struct osd_metadnode_it {
1424         struct osd_device       *mit_dev;
1425         __u64                    mit_pos;
1426         struct lu_fid            mit_fid;
1427         int                      mit_prefetched;
1428         __u64                    mit_prefetched_dnode;
1429 };
1430
1431 static struct dt_it *osd_zfs_otable_it_init(const struct lu_env *env,
1432                                             struct dt_object *dt, __u32 attr,
1433                                             struct lustre_capa *capa)
1434 {
1435         struct osd_device       *dev   = osd_dev(dt->do_lu.lo_dev);
1436         struct osd_metadnode_it *it;
1437         ENTRY;
1438
1439         OBD_ALLOC_PTR(it);
1440         if (unlikely(it == NULL))
1441                 RETURN(ERR_PTR(-ENOMEM));
1442
1443         it->mit_dev = dev;
1444
1445         /* XXX: dmu_object_next() does NOT find dnodes allocated
1446          *      in the current non-committed txg, so we force txg
1447          *      commit to find all existing dnodes ... */
1448         txg_wait_synced(dmu_objset_pool(dev->od_os), 0ULL);
1449
1450         RETURN((struct dt_it *)it);
1451 }
1452
1453 static void osd_zfs_otable_it_fini(const struct lu_env *env, struct dt_it *di)
1454 {
1455         struct osd_metadnode_it *it  = (struct osd_metadnode_it *)di;
1456
1457         OBD_FREE_PTR(it);
1458 }
1459
1460 static int osd_zfs_otable_it_get(const struct lu_env *env,
1461                                  struct dt_it *di, const struct dt_key *key)
1462 {
1463         return 0;
1464 }
1465
1466 static void osd_zfs_otable_it_put(const struct lu_env *env, struct dt_it *di)
1467 {
1468 }
1469
1470 #define OTABLE_PREFETCH         256
1471
1472 static void osd_zfs_otable_prefetch(const struct lu_env *env,
1473                                     struct osd_metadnode_it *it)
1474 {
1475         struct osd_device       *dev = it->mit_dev;
1476         int                      rc;
1477
1478         /* can go negative on the very first access to the iterator
1479          * or if some non-Lustre objects were found */
1480         if (unlikely(it->mit_prefetched < 0))
1481                 it->mit_prefetched = 0;
1482
1483         if (it->mit_prefetched >= (OTABLE_PREFETCH >> 1))
1484                 return;
1485
1486         if (it->mit_prefetched_dnode == 0)
1487                 it->mit_prefetched_dnode = it->mit_pos;
1488
1489         while (it->mit_prefetched < OTABLE_PREFETCH) {
1490                 rc = -dmu_object_next(dev->od_os, &it->mit_prefetched_dnode,
1491                                       B_FALSE, 0);
1492                 if (unlikely(rc != 0))
1493                         break;
1494
1495                 /* dmu_prefetch() was exported in 0.6.2, if you use with
1496                  * an older release, just comment it out - this is an
1497                  * optimization */
1498                 dmu_prefetch(dev->od_os, it->mit_prefetched_dnode, 0, 0);
1499
1500                 it->mit_prefetched++;
1501         }
1502 }
1503
1504 static int osd_zfs_otable_it_next(const struct lu_env *env, struct dt_it *di)
1505 {
1506         struct osd_metadnode_it *it  = (struct osd_metadnode_it *)di;
1507         struct lustre_mdt_attrs *lma;
1508         struct osd_device       *dev = it->mit_dev;
1509         nvlist_t                *nvbuf = NULL;
1510         uchar_t                 *v;
1511         __u64                    dnode;
1512         int                      rc, s;
1513
1514         memset(&it->mit_fid, 0, sizeof(it->mit_fid));
1515
1516         dnode = it->mit_pos;
1517         do {
1518                 rc = -dmu_object_next(dev->od_os, &it->mit_pos, B_FALSE, 0);
1519                 if (unlikely(rc != 0))
1520                         GOTO(out, rc = 1);
1521                 it->mit_prefetched--;
1522
1523                 /* LMA is required for this to be a Lustre object.
1524                  * If there is no xattr skip it. */
1525                 rc = __osd_xattr_load(dev, it->mit_pos, &nvbuf);
1526                 if (unlikely(rc != 0))
1527                         continue;
1528
1529                 LASSERT(nvbuf != NULL);
1530                 rc = -nvlist_lookup_byte_array(nvbuf, XATTR_NAME_LMA, &v, &s);
1531                 if (likely(rc == 0)) {
1532                         /* Lustre object */
1533                         lma = (struct lustre_mdt_attrs *)v;
1534                         lustre_lma_swab(lma);
1535                         it->mit_fid = lma->lma_self_fid;
1536                         nvlist_free(nvbuf);
1537                         break;
1538                 } else {
1539                         /* not a Lustre object, try next one */
1540                         nvlist_free(nvbuf);
1541                 }
1542
1543         } while (1);
1544
1545
1546         /* we aren't prefetching in the above loop because the number of
1547          * non-Lustre objects is very small and we will be repeating very
1548          * rare. in case we want to use this to iterate over non-Lustre
1549          * objects (i.e. when we convert regular ZFS in Lustre) it makes
1550          * sense to initiate prefetching in the loop */
1551
1552         /* 0 - there are more items, +1 - the end */
1553         if (likely(rc == 0))
1554                 osd_zfs_otable_prefetch(env, it);
1555
1556         CDEBUG(D_OTHER, "advance: %llu -> %llu "DFID": %d\n", dnode,
1557                it->mit_pos, PFID(&it->mit_fid), rc);
1558
1559 out:
1560         return rc;
1561 }
1562
1563 static struct dt_key *osd_zfs_otable_it_key(const struct lu_env *env,
1564                                             const struct dt_it *di)
1565 {
1566         return NULL;
1567 }
1568
1569 static int osd_zfs_otable_it_key_size(const struct lu_env *env,
1570                                       const struct dt_it *di)
1571 {
1572         return sizeof(__u64);
1573 }
1574
1575 static int osd_zfs_otable_it_rec(const struct lu_env *env,
1576                                  const struct dt_it *di,
1577                                  struct dt_rec *rec, __u32 attr)
1578 {
1579         struct osd_metadnode_it *it  = (struct osd_metadnode_it *)di;
1580         struct lu_fid *fid = (struct lu_fid *)rec;
1581         ENTRY;
1582
1583         *fid = it->mit_fid;
1584
1585         RETURN(0);
1586 }
1587
1588
1589 static __u64 osd_zfs_otable_it_store(const struct lu_env *env,
1590                                      const struct dt_it *di)
1591 {
1592         struct osd_metadnode_it *it  = (struct osd_metadnode_it *)di;
1593
1594         return it->mit_pos;
1595 }
1596
1597 static int osd_zfs_otable_it_load(const struct lu_env *env,
1598                                   const struct dt_it *di, __u64 hash)
1599 {
1600         struct osd_metadnode_it *it  = (struct osd_metadnode_it *)di;
1601
1602         it->mit_pos = hash;
1603         it->mit_prefetched = 0;
1604         it->mit_prefetched_dnode = 0;
1605
1606         return osd_zfs_otable_it_next(env, (struct dt_it *)di);
1607 }
1608
1609 static int osd_zfs_otable_it_key_rec(const struct lu_env *env,
1610                                      const struct dt_it *di, void *key_rec)
1611 {
1612         return 0;
1613 }
1614
1615 const struct dt_index_operations osd_zfs_otable_ops = {
1616         .dio_it = {
1617                 .init     = osd_zfs_otable_it_init,
1618                 .fini     = osd_zfs_otable_it_fini,
1619                 .get      = osd_zfs_otable_it_get,
1620                 .put      = osd_zfs_otable_it_put,
1621                 .next     = osd_zfs_otable_it_next,
1622                 .key      = osd_zfs_otable_it_key,
1623                 .key_size = osd_zfs_otable_it_key_size,
1624                 .rec      = osd_zfs_otable_it_rec,
1625                 .store    = osd_zfs_otable_it_store,
1626                 .load     = osd_zfs_otable_it_load,
1627                 .key_rec  = osd_zfs_otable_it_key_rec,
1628         }
1629 };
1630
1631 int osd_index_try(const struct lu_env *env, struct dt_object *dt,
1632                 const struct dt_index_features *feat)
1633 {
1634         struct osd_object *obj = osd_dt_obj(dt);
1635         ENTRY;
1636
1637         LASSERT(dt_object_exists(dt));
1638
1639         /*
1640          * XXX: implement support for fixed-size keys sorted with natural
1641          *      numerical way (not using internal hash value)
1642          */
1643         if (feat->dif_flags & DT_IND_RANGE)
1644                 RETURN(-ERANGE);
1645
1646         if (unlikely(feat == &dt_otable_features)) {
1647                 dt->do_index_ops = &osd_zfs_otable_ops;
1648                 RETURN(0);
1649         }
1650
1651         LASSERT(obj->oo_db != NULL);
1652         if (likely(feat == &dt_directory_features)) {
1653                 if (osd_object_is_zap(obj->oo_db))
1654                         dt->do_index_ops = &osd_dir_ops;
1655                 else
1656                         RETURN(-ENOTDIR);
1657         } else if (unlikely(feat == &dt_acct_features)) {
1658                 LASSERT(fid_is_acct(lu_object_fid(&dt->do_lu)));
1659                 dt->do_index_ops = &osd_acct_index_ops;
1660         } else if (osd_object_is_zap(obj->oo_db) &&
1661                    dt->do_index_ops == NULL) {
1662                 /* For index file, we don't support variable key & record sizes
1663                  * and the key has to be unique */
1664                 if ((feat->dif_flags & ~DT_IND_UPDATE) != 0)
1665                         RETURN(-EINVAL);
1666
1667                 if (feat->dif_keysize_max > ZAP_MAXNAMELEN)
1668                         RETURN(-E2BIG);
1669                 if (feat->dif_keysize_max != feat->dif_keysize_min)
1670                         RETURN(-EINVAL);
1671
1672                 /* As for the record size, it should be a multiple of 8 bytes
1673                  * and smaller than the maximum value length supported by ZAP.
1674                  */
1675                 if (feat->dif_recsize_max > ZAP_MAXVALUELEN)
1676                         RETURN(-E2BIG);
1677                 if (feat->dif_recsize_max != feat->dif_recsize_min)
1678                         RETURN(-EINVAL);
1679
1680                 obj->oo_keysize = feat->dif_keysize_max;
1681                 obj->oo_recsize = feat->dif_recsize_max;
1682                 obj->oo_recusize = 1;
1683
1684                 /* ZFS prefers to work with array of 64bits */
1685                 if ((obj->oo_recsize & 7) == 0) {
1686                         obj->oo_recsize >>= 3;
1687                         obj->oo_recusize = 8;
1688                 }
1689                 dt->do_index_ops = &osd_index_ops;
1690         }
1691
1692         RETURN(0);
1693 }