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 /*
  31  * Copyright (c) 2012, 2013, Intel Corporation.
  32  * Use is subject to license terms.
  33  */
  34 /*
  35  * This file is part of Lustre, http://www.lustre.org/
  36  * Lustre is a trademark of Sun Microsystems, Inc.
  37  *
  38  * lustre/osd-zfs/osd_index.c
  39  *
  40  * Author: Alex Zhuravlev <bzzz@whamcloud.com>
  41  * Author: Mike Pershin <tappro@whamcloud.com>
  42  */
  43
  44 #ifndef EXPORT_SYMTAB
  45 # define EXPORT_SYMTAB
  46 #endif
  47 #define DEBUG_SUBSYSTEM S_OSD
  48
  49 #include <lustre_ver.h>
  50 #include <libcfs/libcfs.h>
  51 #include <obd_support.h>
  52 #include <lustre_net.h>
  53 #include <obd.h>
  54 #include <obd_class.h>
  55 #include <lustre_disk.h>
  56 #include <lustre_fid.h>
  57
  58 #include "osd_internal.h"
  59
  60 #include <sys/dnode.h>
  61 #include <sys/dbuf.h>
  62 #include <sys/spa.h>
  63 #include <sys/stat.h>
  64 #include <sys/zap.h>
  65 #include <sys/spa_impl.h>
  66 #include <sys/zfs_znode.h>
  67 #include <sys/dmu_tx.h>
  68 #include <sys/dmu_objset.h>
  69 #include <sys/dsl_prop.h>
  70 #include <sys/sa_impl.h>
  71 #include <sys/txg.h>
  72
  73 static struct dt_it *osd_index_it_init(const struct lu_env *env,
  74                                        struct dt_object *dt,
  75                                        __u32 unused,
  76                                        struct lustre_capa *capa)
  77 {
  78         struct osd_thread_info  *info = osd_oti_get(env);
  79         struct osd_zap_it       *it;
  80         struct osd_object       *obj = osd_dt_obj(dt);
  81         struct osd_device       *osd = osd_obj2dev(obj);
  82         struct lu_object        *lo  = &dt->do_lu;
  83         ENTRY;
  84
  85         /* XXX: check capa ? */
  86
  87         LASSERT(lu_object_exists(lo));
  88         LASSERT(obj->oo_db);
  89         LASSERT(udmu_object_is_zap(obj->oo_db));
  90         LASSERT(info);
  91
  92         it = &info->oti_it_zap;
  93
  94         if (udmu_zap_cursor_init(&it->ozi_zc, &osd->od_objset,
  95                                  obj->oo_db->db_object, 0))
  96                 RETURN(ERR_PTR(-ENOMEM));
  97
  98         it->ozi_obj   = obj;
  99         it->ozi_capa  = capa;
 100         it->ozi_reset = 1;
 101         lu_object_get(lo);
 102
 103         RETURN((struct dt_it *)it);
 104 }
 105
 106 static void osd_index_it_fini(const struct lu_env *env, struct dt_it *di)
 107 {
 108         struct osd_zap_it *it = (struct osd_zap_it *)di;
 109         struct osd_object *obj;
 110         ENTRY;
 111
 112         LASSERT(it);
 113         LASSERT(it->ozi_obj);
 114
 115         obj = it->ozi_obj;
 116
 117         udmu_zap_cursor_fini(it->ozi_zc);
 118         lu_object_put(env, &obj->oo_dt.do_lu);
 119
 120         EXIT;
 121 }
 122
 123
 124 static void osd_index_it_put(const struct lu_env *env, struct dt_it *di)
 125 {
 126         /* PBS: do nothing : ref are incremented at retrive and decreamented
 127          *      next/finish. */
 128 }
 129
 130 int udmu_zap_cursor_retrieve_key(const struct lu_env *env,
 131                                  zap_cursor_t *zc, char *key, int max)
 132 {
 133         zap_attribute_t *za = &osd_oti_get(env)->oti_za;
 134         int              err;
 135
 136         if ((err = zap_cursor_retrieve(zc, za)))
 137                 return err;
 138
 139         if (key)
 140                 strcpy(key, za->za_name);
 141
 142         return 0;
 143 }
 144
 145 /*
 146  * zap_cursor_retrieve read from current record.
 147  * to read bytes we need to call zap_lookup explicitly.
 148  */
 149 int udmu_zap_cursor_retrieve_value(const struct lu_env *env,
 150                 zap_cursor_t *zc,  char *buf,
 151                 int buf_size, int *bytes_read)
 152 {
 153         zap_attribute_t *za = &osd_oti_get(env)->oti_za;
 154         int err, actual_size;
 155
 156         if ((err = zap_cursor_retrieve(zc, za)))
 157                 return err;
 158
 159         if (za->za_integer_length <= 0)
 160                 return (ERANGE);
 161
 162         actual_size = za->za_integer_length * za->za_num_integers;
 163
 164         if (actual_size > buf_size) {
 165                 actual_size = buf_size;
 166                 buf_size = actual_size / za->za_integer_length;
 167         } else {
 168                 buf_size = za->za_num_integers;
 169         }
 170
 171         err = -zap_lookup(zc->zc_objset, zc->zc_zapobj,
 172                         za->za_name, za->za_integer_length,
 173                         buf_size, buf);
 174
 175         if (!err)
 176                 *bytes_read = actual_size;
 177
 178         return err;
 179 }
 180
 181 static inline void osd_it_append_attrs(struct lu_dirent *ent, __u32 attr,
 182                                        int len, __u16 type)
 183 {
 184         const unsigned    align = sizeof(struct luda_type) - 1;
 185         struct luda_type *lt;
 186
 187         /* check if file type is required */
 188         if (attr & LUDA_TYPE) {
 189                 len = (len + align) & ~align;
 190
 191                 lt = (void *)ent->lde_name + len;
 192                 lt->lt_type = cpu_to_le16(DTTOIF(type));
 193                 ent->lde_attrs |= LUDA_TYPE;
 194         }
 195
 196         ent->lde_attrs = cpu_to_le32(ent->lde_attrs);
 197 }
 198
 199 /*
 200  * as we don't know FID, we can't use LU object, so this function
 201  * partially duplicate __osd_xattr_get() which is built around
 202  * LU-object and uses it to cache data like regular EA dnode, etc
 203  */
 204 static int osd_find_parent_by_dnode(const struct lu_env *env,
 205                                     struct dt_object *o,
 206                                     struct lu_fid *fid)
 207 {
 208         struct lustre_mdt_attrs *lma;
 209         udmu_objset_t           *uos = &osd_obj2dev(osd_dt_obj(o))->od_objset;
 210         struct lu_buf            buf;
 211         sa_handle_t             *sa_hdl;
 212         nvlist_t                *nvbuf = NULL;
 213         uchar_t                 *value;
 214         uint64_t                 dnode;
 215         int                      rc, size;
 216         ENTRY;
 217
 218         /* first of all, get parent dnode from own attributes */
 219         LASSERT(osd_dt_obj(o)->oo_db);
 220         rc = -sa_handle_get(uos->os, osd_dt_obj(o)->oo_db->db_object,
 221                             NULL, SA_HDL_PRIVATE, &sa_hdl);
 222         if (rc)
 223                 RETURN(rc);
 224
 225         dnode = ZFS_NO_OBJECT;
 226         rc = -sa_lookup(sa_hdl, SA_ZPL_PARENT(uos), &dnode, 8);
 227         sa_handle_destroy(sa_hdl);
 228         if (rc)
 229                 RETURN(rc);
 230
 231         /* now get EA buffer */
 232         rc = __osd_xattr_load(uos, dnode, &nvbuf);
 233         if (rc)
 234                 GOTO(regular, rc);
 235
 236         /* XXX: if we get that far.. should we cache the result? */
 237
 238         /* try to find LMA attribute */
 239         LASSERT(nvbuf != NULL);
 240         rc = -nvlist_lookup_byte_array(nvbuf, XATTR_NAME_LMA, &value, &size);
 241         if (rc == 0 && size >= sizeof(*lma)) {
 242                 lma = (struct lustre_mdt_attrs *)value;
 243                 lustre_lma_swab(lma);
 244                 *fid = lma->lma_self_fid;
 245                 GOTO(out, rc = 0);
 246         }
 247
 248 regular:
 249         /* no LMA attribute in SA, let's try regular EA */
 250
 251         /* first of all, get parent dnode storing regular EA */
 252         rc = -sa_handle_get(uos->os, dnode, NULL, SA_HDL_PRIVATE, &sa_hdl);
 253         if (rc)
 254                 GOTO(out, rc);
 255
 256         dnode = ZFS_NO_OBJECT;
 257         rc = -sa_lookup(sa_hdl, SA_ZPL_XATTR(uos), &dnode, 8);
 258         sa_handle_destroy(sa_hdl);
 259         if (rc)
 260                 GOTO(out, rc);
 261
 262         CLASSERT(sizeof(*lma) <= sizeof(osd_oti_get(env)->oti_buf));
 263         buf.lb_buf = osd_oti_get(env)->oti_buf;
 264         buf.lb_len = sizeof(osd_oti_get(env)->oti_buf);
 265
 266         /* now try to find LMA */
 267         rc = __osd_xattr_get_large(env, uos, dnode, &buf,
 268                                    XATTR_NAME_LMA, &size);
 269         if (rc == 0 && size >= sizeof(*lma)) {
 270                 lma = buf.lb_buf;
 271                 lustre_lma_swab(lma);
 272                 *fid = lma->lma_self_fid;
 273                 GOTO(out, rc = 0);
 274         } else if (rc < 0) {
 275                 GOTO(out, rc);
 276         } else {
 277                 GOTO(out, rc = -EIO);
 278         }
 279
 280 out:
 281         if (nvbuf != NULL)
 282                 nvlist_free(nvbuf);
 283         RETURN(rc);
 284 }
 285
 286 static int osd_find_parent_fid(const struct lu_env *env, struct dt_object *o,
 287                                struct lu_fid *fid)
 288 {
 289         struct link_ea_header  *leh;
 290         struct link_ea_entry   *lee;
 291         struct lu_buf           buf;
 292         int                     rc;
 293         ENTRY;
 294
 295         buf.lb_buf = osd_oti_get(env)->oti_buf;
 296         buf.lb_len = sizeof(osd_oti_get(env)->oti_buf);
 297
 298         rc = osd_xattr_get(env, o, &buf, XATTR_NAME_LINK, BYPASS_CAPA);
 299         if (rc == -ERANGE) {
 300                 rc = osd_xattr_get(env, o, &LU_BUF_NULL,
 301                                    XATTR_NAME_LINK, BYPASS_CAPA);
 302                 if (rc < 0)
 303                         RETURN(rc);
 304                 LASSERT(rc > 0);
 305                 OBD_ALLOC(buf.lb_buf, rc);
 306                 if (buf.lb_buf == NULL)
 307                         RETURN(-ENOMEM);
 308                 buf.lb_len = rc;
 309                 rc = osd_xattr_get(env, o, &buf, XATTR_NAME_LINK, BYPASS_CAPA);
 310         }
 311         if (rc < 0)
 312                 GOTO(out, rc);
 313         if (rc < sizeof(*leh) + sizeof(*lee))
 314                 GOTO(out, rc = -EINVAL);
 315
 316         leh = buf.lb_buf;
 317         if (leh->leh_magic == __swab32(LINK_EA_MAGIC)) {
 318                 leh->leh_magic = LINK_EA_MAGIC;
 319                 leh->leh_reccount = __swab32(leh->leh_reccount);
 320                 leh->leh_len = __swab64(leh->leh_len);
 321         }
 322         if (leh->leh_magic != LINK_EA_MAGIC)
 323                 GOTO(out, rc = -EINVAL);
 324         if (leh->leh_reccount == 0)
 325                 GOTO(out, rc = -ENODATA);
 326
 327         lee = (struct link_ea_entry *)(leh + 1);
 328         fid_be_to_cpu(fid, (const struct lu_fid *)&lee->lee_parent_fid);
 329         rc = 0;
 330
 331 out:
 332         if (buf.lb_buf != osd_oti_get(env)->oti_buf)
 333                 OBD_FREE(buf.lb_buf, buf.lb_len);
 334
 335 #if 0
 336         /* this block can be enabled for additional verification
 337          * it's trying to match FID from LinkEA vs. FID from LMA */
 338         if (rc == 0) {
 339                 struct lu_fid fid2;
 340                 int rc2;
 341                 rc2 = osd_find_parent_by_dnode(env, o, &fid2);
 342                 if (rc2 == 0)
 343                         if (lu_fid_eq(fid, &fid2) == 0)
 344                                 CERROR("wrong parent: "DFID" != "DFID"\n",
 345                                        PFID(fid), PFID(&fid2));
 346         }
 347 #endif
 348
 349         /* no LinkEA is found, let's try to find the fid in parent's LMA */
 350         if (unlikely(rc != 0))
 351                 rc = osd_find_parent_by_dnode(env, o, fid);
 352
 353         RETURN(rc);
 354 }
 355
 356 static int osd_dir_lookup(const struct lu_env *env, struct dt_object *dt,
 357                           struct dt_rec *rec, const struct dt_key *key,
 358                           struct lustre_capa *capa)
 359 {
 360         struct osd_thread_info *oti = osd_oti_get(env);
 361         struct osd_object  *obj = osd_dt_obj(dt);
 362         struct osd_device  *osd = osd_obj2dev(obj);
 363         char               *name = (char *)key;
 364         int                 rc;
 365         ENTRY;
 366
 367         LASSERT(udmu_object_is_zap(obj->oo_db));
 368
 369         if (name[0] == '.') {
 370                 if (name[1] == 0) {
 371                         const struct lu_fid *f = lu_object_fid(&dt->do_lu);
 372                         memcpy(rec, f, sizeof(*f));
 373                         RETURN(1);
 374                 } else if (name[1] == '.' && name[2] == 0) {
 375                         rc = osd_find_parent_fid(env, dt, (struct lu_fid *)rec);
 376                         RETURN(rc == 0 ? 1 : rc);
 377                 }
 378         }
 379
 380         rc = -zap_lookup(osd->od_objset.os, obj->oo_db->db_object,
 381                          (char *)key, 8, sizeof(oti->oti_zde) / 8,
 382                          (void *)&oti->oti_zde);
 383         memcpy(rec, &oti->oti_zde.lzd_fid, sizeof(struct lu_fid));
 384
 385         RETURN(rc == 0 ? 1 : rc);
 386 }
 387
 388 static int osd_declare_dir_insert(const struct lu_env *env,
 389                                   struct dt_object *dt,
 390                                   const struct dt_rec *rec,
 391                                   const struct dt_key *key,
 392                                   struct thandle *th)
 393 {
 394         struct osd_object  *obj = osd_dt_obj(dt);
 395         struct osd_thandle *oh;
 396         ENTRY;
 397
 398         LASSERT(th != NULL);
 399         oh = container_of0(th, struct osd_thandle, ot_super);
 400
 401         LASSERT(obj->oo_db);
 402         LASSERT(udmu_object_is_zap(obj->oo_db));
 403
 404         dmu_tx_hold_bonus(oh->ot_tx, obj->oo_db->db_object);
 405         dmu_tx_hold_zap(oh->ot_tx, obj->oo_db->db_object, TRUE, (char *)key);
 406
 407         RETURN(0);
 408 }
 409
 410 /**
 411  * Find the osd object for given fid.
 412  *
 413  * \param fid need to find the osd object having this fid
 414  *
 415  * \retval osd_object on success
 416  * \retval        -ve on error
 417  */
 418 struct osd_object *osd_object_find(const struct lu_env *env,
 419                                    struct dt_object *dt,
 420                                    const struct lu_fid *fid)
 421 {
 422         struct lu_device         *ludev = dt->do_lu.lo_dev;
 423         struct osd_object        *child = NULL;
 424         struct lu_object         *luch;
 425         struct lu_object         *lo;
 426
 427         /*
 428          * at this point topdev might not exist yet
 429          * (i.e. MGS is preparing profiles). so we can
 430          * not rely on topdev and instead lookup with
 431          * our device passed as topdev. this can't work
 432          * if the object isn't cached yet (as osd doesn't
 433          * allocate lu_header). IOW, the object must be
 434          * in the cache, otherwise lu_object_alloc() crashes
 435          * -bzzz
 436          */
 437         luch = lu_object_find_at(env, ludev, fid, NULL);
 438         if (IS_ERR(luch))
 439                 return (void *)luch;
 440
 441         if (lu_object_exists(luch)) {
 442                 lo = lu_object_locate(luch->lo_header, ludev->ld_type);
 443                 if (lo != NULL)
 444                         child = osd_obj(lo);
 445                 else
 446                         LU_OBJECT_DEBUG(D_ERROR, env, luch,
 447                                         "%s: object can't be located "DFID"\n",
 448                                         osd_dev(ludev)->od_svname, PFID(fid));
 449
 450                 if (child == NULL) {
 451                         lu_object_put(env, luch);
 452                         CERROR("%s: Unable to get osd_object "DFID"\n",
 453                                osd_dev(ludev)->od_svname, PFID(fid));
 454                         child = ERR_PTR(-ENOENT);
 455                 }
 456         } else {
 457                 LU_OBJECT_DEBUG(D_ERROR, env, luch,
 458                                 "%s: lu_object does not exists "DFID"\n",
 459                                 osd_dev(ludev)->od_svname, PFID(fid));
 460                 lu_object_put(env, luch);
 461                 child = ERR_PTR(-ENOENT);
 462         }
 463
 464         return child;
 465 }
 466
 467 /**
 468  * Put the osd object once done with it.
 469  *
 470  * \param obj osd object that needs to be put
 471  */
 472 static inline void osd_object_put(const struct lu_env *env,
 473                                   struct osd_object *obj)
 474 {
 475         lu_object_put(env, &obj->oo_dt.do_lu);
 476 }
 477
 478 static int osd_remote_fid(const struct lu_env *env, struct osd_device *osd,
 479                           struct lu_fid *fid)
 480 {
 481         struct lu_seq_range     *range = &osd_oti_get(env)->oti_seq_range;
 482         struct seq_server_site  *ss = osd_seq_site(osd);
 483         int                     rc;
 484         ENTRY;
 485
 486         if (!fid_is_norm(fid) && !fid_is_root(fid))
 487                 RETURN(0);
 488
 489         rc = osd_fld_lookup(env, osd, fid, range);
 490         if (rc != 0) {
 491                 CERROR("%s: Can not lookup fld for "DFID"\n",
 492                        osd_name(osd), PFID(fid));
 493                 RETURN(rc);
 494         }
 495
 496         RETURN(ss->ss_node_id != range->lsr_index);
 497 }
 498
 499 /**
 500  *      Inserts (key, value) pair in \a directory object.
 501  *
 502  *      \param  dt      osd index object
 503  *      \param  key     key for index
 504  *      \param  rec     record reference
 505  *      \param  th      transaction handler
 506  *      \param  capa    capability descriptor
 507  *      \param  ignore_quota update should not affect quota
 508  *
 509  *      \retval  0  success
 510  *      \retval -ve failure
 511  */
 512 static int osd_dir_insert(const struct lu_env *env, struct dt_object *dt,
 513                           const struct dt_rec *rec, const struct dt_key *key,
 514                           struct thandle *th, struct lustre_capa *capa,
 515                           int ignore_quota)
 516 {
 517         struct osd_thread_info *oti = osd_oti_get(env);
 518         struct osd_object   *parent = osd_dt_obj(dt);
 519         struct osd_device   *osd = osd_obj2dev(parent);
 520         struct lu_fid       *fid = (struct lu_fid *)rec;
 521         struct osd_thandle  *oh;
 522         struct osd_object   *child = NULL;
 523         __u32                attr;
 524         char                *name = (char *)key;
 525         int                  rc;
 526         ENTRY;
 527
 528         LASSERT(parent->oo_db);
 529         LASSERT(udmu_object_is_zap(parent->oo_db));
 530
 531         LASSERT(dt_object_exists(dt));
 532         LASSERT(osd_invariant(parent));
 533
 534         LASSERT(th != NULL);
 535         oh = container_of0(th, struct osd_thandle, ot_super);
 536
 537         rc = osd_remote_fid(env, osd, fid);
 538         if (rc < 0) {
 539                 CERROR("%s: Can not find object "DFID": rc = %d\n",
 540                        osd->od_svname, PFID(fid), rc);
 541                 RETURN(rc);
 542         }
 543
 544         if (unlikely(rc == 1)) {
 545                 /* Insert remote entry */
 546                 memset(&oti->oti_zde.lzd_reg, 0, sizeof(oti->oti_zde.lzd_reg));
 547                 oti->oti_zde.lzd_reg.zde_type = IFTODT(S_IFDIR & S_IFMT);
 548         } else {
 549                 /*
 550                  * To simulate old Orion setups with ./..  stored in the
 551                  * directories
 552                  */
 553                 /* Insert local entry */
 554                 child = osd_object_find(env, dt, fid);
 555                 if (IS_ERR(child))
 556                         RETURN(PTR_ERR(child));
 557
 558                 LASSERT(child->oo_db);
 559                 if (name[0] == '.') {
 560                         if (name[1] == 0) {
 561                                 /* do not store ".", instead generate it
 562                                  * during iteration */
 563                                 GOTO(out, rc = 0);
 564                         } else if (name[1] == '.' && name[2] == 0) {
 565                                 /* update parent dnode in the child.
 566                                  * later it will be used to generate ".." */
 567                                 udmu_objset_t *uos = &osd->od_objset;
 568                                 rc = osd_object_sa_update(parent,
 569                                                  SA_ZPL_PARENT(uos),
 570                                                  &child->oo_db->db_object,
 571                                                  8, oh);
 572                                 GOTO(out, rc);
 573                         }
 574                 }
 575                 CLASSERT(sizeof(oti->oti_zde.lzd_reg) == 8);
 576                 CLASSERT(sizeof(oti->oti_zde) % 8 == 0);
 577                 attr = child->oo_dt.do_lu.lo_header ->loh_attr;
 578                 oti->oti_zde.lzd_reg.zde_type = IFTODT(attr & S_IFMT);
 579                 oti->oti_zde.lzd_reg.zde_dnode = child->oo_db->db_object;
 580         }
 581
 582         oti->oti_zde.lzd_fid = *fid;
 583         /* Insert (key,oid) into ZAP */
 584         rc = -zap_add(osd->od_objset.os, parent->oo_db->db_object,
 585                       (char *)key, 8, sizeof(oti->oti_zde) / 8,
 586                       (void *)&oti->oti_zde, oh->ot_tx);
 587
 588 out:
 589         if (child != NULL)
 590                 osd_object_put(env, child);
 591
 592         RETURN(rc);
 593 }
 594
 595 static int osd_declare_dir_delete(const struct lu_env *env,
 596                                   struct dt_object *dt,
 597                                   const struct dt_key *key,
 598                                   struct thandle *th)
 599 {
 600         struct osd_object *obj = osd_dt_obj(dt);
 601         struct osd_thandle *oh;
 602         ENTRY;
 603
 604         LASSERT(dt_object_exists(dt));
 605         LASSERT(osd_invariant(obj));
 606
 607         LASSERT(th != NULL);
 608         oh = container_of0(th, struct osd_thandle, ot_super);
 609
 610         LASSERT(obj->oo_db);
 611         LASSERT(udmu_object_is_zap(obj->oo_db));
 612
 613         dmu_tx_hold_zap(oh->ot_tx, obj->oo_db->db_object, TRUE, (char *)key);
 614
 615         RETURN(0);
 616 }
 617
 618 static int osd_dir_delete(const struct lu_env *env, struct dt_object *dt,
 619                           const struct dt_key *key, struct thandle *th,
 620                           struct lustre_capa *capa)
 621 {
 622         struct osd_object *obj = osd_dt_obj(dt);
 623         struct osd_device *osd = osd_obj2dev(obj);
 624         struct osd_thandle *oh;
 625         dmu_buf_t *zap_db = obj->oo_db;
 626         char      *name = (char *)key;
 627         int rc;
 628         ENTRY;
 629
 630         LASSERT(obj->oo_db);
 631         LASSERT(udmu_object_is_zap(obj->oo_db));
 632
 633         LASSERT(th != NULL);
 634         oh = container_of0(th, struct osd_thandle, ot_super);
 635
 636         /*
 637          * In Orion . and .. were stored in the directory (not generated upon
 638          * request as now). we preserve them for backward compatibility
 639          */
 640         if (name[0] == '.') {
 641                 if (name[1] == 0) {
 642                         RETURN(0);
 643                 } else if (name[1] == '.' && name[2] == 0) {
 644                         RETURN(0);
 645                 }
 646         }
 647
 648         /* Remove key from the ZAP */
 649         rc = -zap_remove(osd->od_objset.os, zap_db->db_object,
 650                          (char *) key, oh->ot_tx);
 651
 652 #if LUSTRE_VERSION_CODE <= OBD_OCD_VERSION(2, 4, 53, 0)
 653         if (unlikely(rc == -ENOENT && name[0] == '.' &&
 654             (name[1] == 0 || (name[1] == '.' && name[2] == 0))))
 655                 rc = 0;
 656 #endif
 657         if (unlikely(rc && rc != -ENOENT))
 658                 CERROR("%s: zap_remove failed: rc = %d\n", osd->od_svname, rc);
 659
 660         RETURN(rc);
 661 }
 662
 663 static struct dt_it *osd_dir_it_init(const struct lu_env *env,
 664                                      struct dt_object *dt,
 665                                      __u32 unused,
 666                                      struct lustre_capa *capa)
 667 {
 668         struct osd_zap_it *it;
 669
 670         it = (struct osd_zap_it *)osd_index_it_init(env, dt, unused, capa);
 671         if (!IS_ERR(it))
 672                 it->ozi_pos = 0;
 673
 674         RETURN((struct dt_it *)it);
 675 }
 676
 677 /**
 678  *  Move Iterator to record specified by \a key
 679  *
 680  *  \param  di      osd iterator
 681  *  \param  key     key for index
 682  *
 683  *  \retval +ve  di points to record with least key not larger than key
 684  *  \retval  0   di points to exact matched key
 685  *  \retval -ve  failure
 686  */
 687 static int osd_dir_it_get(const struct lu_env *env,
 688                           struct dt_it *di, const struct dt_key *key)
 689 {
 690         struct osd_zap_it *it = (struct osd_zap_it *)di;
 691         struct osd_object *obj = it->ozi_obj;
 692         struct osd_device *osd = osd_obj2dev(obj);
 693         char              *name = (char *)key;
 694         int                rc;
 695         ENTRY;
 696
 697         LASSERT(it);
 698         LASSERT(it->ozi_zc);
 699
 700         udmu_zap_cursor_fini(it->ozi_zc);
 701
 702         if (udmu_zap_cursor_init(&it->ozi_zc, &osd->od_objset,
 703                                  obj->oo_db->db_object, 0))
 704                 RETURN(-ENOMEM);
 705
 706         /* XXX: implementation of the API is broken at the moment */
 707         LASSERT(((const char *)key)[0] == 0);
 708
 709         if (name[0] == 0) {
 710                 it->ozi_pos = 0;
 711                 RETURN(1);
 712         }
 713
 714         if (name[0] == '.') {
 715                 if (name[1] == 0) {
 716                         it->ozi_pos = 1;
 717                         GOTO(out, rc = 1);
 718                 } else if (name[1] == '.' && name[2] == 0) {
 719                         it->ozi_pos = 2;
 720                         GOTO(out, rc = 1);
 721                 }
 722         }
 723
 724         /* neither . nor .. - some real record */
 725         it->ozi_pos = 3;
 726         rc = +1;
 727
 728 out:
 729         RETURN(rc);
 730 }
 731
 732 static void osd_dir_it_put(const struct lu_env *env, struct dt_it *di)
 733 {
 734         /* PBS: do nothing : ref are incremented at retrive and decreamented
 735          *      next/finish. */
 736 }
 737
 738 /*
 739  * in Orion . and .. were stored in the directory, while ZPL
 740  * and current osd-zfs generate them up on request. so, we
 741  * need to ignore previously stored . and ..
 742  */
 743 static int osd_index_retrieve_skip_dots(struct osd_zap_it *it,
 744                                         zap_attribute_t *za)
 745 {
 746         int rc, isdot;
 747
 748         do {
 749                 rc = -zap_cursor_retrieve(it->ozi_zc, za);
 750
 751                 isdot = 0;
 752                 if (unlikely(rc == 0 && za->za_name[0] == '.')) {
 753                         if (za->za_name[1] == 0) {
 754                                 isdot = 1;
 755                         } else if (za->za_name[1] == '.' &&
 756                                    za->za_name[2] == 0) {
 757                                 isdot = 1;
 758                         }
 759                         if (unlikely(isdot))
 760                                 zap_cursor_advance(it->ozi_zc);
 761                 }
 762         } while (unlikely(rc == 0 && isdot));
 763
 764         return rc;
 765 }
 766
 767 /**
 768  * to load a directory entry at a time and stored it in
 769  * iterator's in-memory data structure.
 770  *
 771  * \param di, struct osd_it_ea, iterator's in memory structure
 772  *
 773  * \retval +ve, iterator reached to end
 774  * \retval   0, iterator not reached to end
 775  * \retval -ve, on error
 776  */
 777 static int osd_dir_it_next(const struct lu_env *env, struct dt_it *di)
 778 {
 779         struct osd_zap_it *it = (struct osd_zap_it *)di;
 780         zap_attribute_t   *za = &osd_oti_get(env)->oti_za;
 781         int                rc;
 782
 783         /* temp. storage should be enough for any key supported by ZFS */
 784         CLASSERT(sizeof(za->za_name) <= sizeof(it->ozi_name));
 785
 786         /*
 787          * the first ->next() moves the cursor to .
 788          * the second ->next() moves the cursor to ..
 789          * then we get to the real records and have to verify any exist
 790          */
 791         if (it->ozi_pos <= 2) {
 792                 it->ozi_pos++;
 793                 if (it->ozi_pos <=2)
 794                         RETURN(0);
 795         }
 796
 797         zap_cursor_advance(it->ozi_zc);
 798
 799         /*
 800          * According to current API we need to return error if its last entry.
 801          * zap_cursor_advance() does not return any value. So we need to call
 802          * retrieve to check if there is any record.  We should make
 803          * changes to Iterator API to not return status for this API
 804          */
 805         rc = osd_index_retrieve_skip_dots(it, za);
 806
 807         if (rc == -ENOENT) /* end of dir */
 808                 RETURN(+1);
 809
 810         RETURN(rc);
 811 }
 812
 813 static struct dt_key *osd_dir_it_key(const struct lu_env *env,
 814                                      const struct dt_it *di)
 815 {
 816         struct osd_zap_it *it = (struct osd_zap_it *)di;
 817         zap_attribute_t   *za = &osd_oti_get(env)->oti_za;
 818         int                rc = 0;
 819         ENTRY;
 820
 821         if (it->ozi_pos <= 1) {
 822                 it->ozi_pos = 1;
 823                 RETURN((struct dt_key *)".");
 824         } else if (it->ozi_pos == 2) {
 825                 RETURN((struct dt_key *)"..");
 826         }
 827
 828         if ((rc = -zap_cursor_retrieve(it->ozi_zc, za)))
 829                 RETURN(ERR_PTR(rc));
 830
 831         strcpy(it->ozi_name, za->za_name);
 832
 833 #if LUSTRE_VERSION_CODE < OBD_OCD_VERSION(2, 3, 90, 0)
 834         if (za->za_name[0] == '.') {
 835                 if (za->za_name[1] == 0 || (za->za_name[1] == '.' &&
 836                     za->za_name[2] == 0)) {
 837                         /* we should not get onto . and ..
 838                          * stored in the directory. ->next() and
 839                          * other methods should prevent this
 840                          */
 841                         LBUG();
 842                 }
 843         }
 844 #endif
 845
 846         RETURN((struct dt_key *)it->ozi_name);
 847 }
 848
 849 static int osd_dir_it_key_size(const struct lu_env *env, const struct dt_it *di)
 850 {
 851         struct osd_zap_it *it = (struct osd_zap_it *)di;
 852         zap_attribute_t   *za = &osd_oti_get(env)->oti_za;
 853         int                rc;
 854         ENTRY;
 855
 856         if (it->ozi_pos <= 1) {
 857                 it->ozi_pos = 1;
 858                 RETURN(2);
 859         } else if (it->ozi_pos == 2) {
 860                 RETURN(3);
 861         }
 862
 863         if ((rc = -zap_cursor_retrieve(it->ozi_zc, za)) == 0)
 864                 rc = strlen(za->za_name);
 865
 866 #if LUSTRE_VERSION_CODE < OBD_OCD_VERSION(2, 3, 90, 0)
 867         if (rc == 0 && za->za_name[0] == '.') {
 868                 if (za->za_name[1] == 0 || (za->za_name[1] == '.' &&
 869                     za->za_name[2] == 0)) {
 870                         /* we should not get onto . and ..
 871                          * stored in the directory. ->next() and
 872                          * other methods should prevent this
 873                          */
 874                         LBUG();
 875                 }
 876         }
 877 #endif
 878         RETURN(rc);
 879 }
 880
 881 static int osd_dir_it_rec(const struct lu_env *env, const struct dt_it *di,
 882                           struct dt_rec *dtrec, __u32 attr)
 883 {
 884         struct osd_zap_it   *it = (struct osd_zap_it *)di;
 885         struct lu_dirent    *lde = (struct lu_dirent *)dtrec;
 886         struct luz_direntry *zde = &osd_oti_get(env)->oti_zde;
 887         zap_attribute_t     *za = &osd_oti_get(env)->oti_za;
 888         int                  rc, namelen;
 889         ENTRY;
 890
 891         if (it->ozi_pos <= 1) {
 892                 lde->lde_hash = cpu_to_le64(1);
 893                 strcpy(lde->lde_name, ".");
 894                 lde->lde_namelen = cpu_to_le16(1);
 895                 lde->lde_fid = *lu_object_fid(&it->ozi_obj->oo_dt.do_lu);
 896                 lde->lde_attrs = LUDA_FID;
 897                 /* append lustre attributes */
 898                 osd_it_append_attrs(lde, attr, 1, IFTODT(S_IFDIR));
 899                 lde->lde_reclen = cpu_to_le16(lu_dirent_calc_size(1, attr));
 900                 it->ozi_pos = 1;
 901                 GOTO(out, rc = 0);
 902
 903         } else if (it->ozi_pos == 2) {
 904                 lde->lde_hash = cpu_to_le64(2);
 905                 strcpy(lde->lde_name, "..");
 906                 lde->lde_namelen = cpu_to_le16(2);
 907                 lde->lde_attrs = LUDA_FID;
 908                 /* append lustre attributes */
 909                 osd_it_append_attrs(lde, attr, 2, IFTODT(S_IFDIR));
 910                 lde->lde_reclen = cpu_to_le16(lu_dirent_calc_size(2, attr));
 911                 rc = osd_find_parent_fid(env, &it->ozi_obj->oo_dt, &lde->lde_fid);
 912                 /*
 913                  * early Orion code was not setting LinkEA, so it's possible
 914                  * some setups still have objects with no LinkEA set.
 915                  * but at that time .. was a real record in the directory
 916                  * so we should try to lookup .. in ZAP
 917                  */
 918                 if (rc != -ENOENT)
 919                         GOTO(out, rc);
 920         }
 921
 922         LASSERT(lde);
 923
 924         lde->lde_hash = cpu_to_le64(udmu_zap_cursor_serialize(it->ozi_zc));
 925
 926         if ((rc = -zap_cursor_retrieve(it->ozi_zc, za)))
 927                 GOTO(out, rc);
 928
 929         namelen = strlen(za->za_name);
 930         if (namelen > NAME_MAX)
 931                 GOTO(out, rc = -EOVERFLOW);
 932         strcpy(lde->lde_name, za->za_name);
 933         lde->lde_namelen = cpu_to_le16(namelen);
 934
 935         if (za->za_integer_length != 8 || za->za_num_integers < 3) {
 936                 CERROR("%s: unsupported direntry format: %d %d\n",
 937                        osd_obj2dev(it->ozi_obj)->od_svname,
 938                        za->za_integer_length, (int)za->za_num_integers);
 939
 940                 GOTO(out, rc = -EIO);
 941         }
 942
 943         rc = -zap_lookup(it->ozi_zc->zc_objset, it->ozi_zc->zc_zapobj,
 944                          za->za_name, za->za_integer_length, 3, zde);
 945         if (rc)
 946                 GOTO(out, rc);
 947
 948         lde->lde_fid = zde->lzd_fid;
 949         lde->lde_attrs = LUDA_FID;
 950
 951         /* append lustre attributes */
 952         osd_it_append_attrs(lde, attr, namelen, zde->lzd_reg.zde_type);
 953
 954         lde->lde_reclen = cpu_to_le16(lu_dirent_calc_size(namelen, attr));
 955
 956 out:
 957         RETURN(rc);
 958 }
 959
 960 static __u64 osd_dir_it_store(const struct lu_env *env, const struct dt_it *di)
 961 {
 962         struct osd_zap_it *it = (struct osd_zap_it *)di;
 963         __u64              pos;
 964         ENTRY;
 965
 966         if (it->ozi_pos <= 2)
 967                 pos = it->ozi_pos;
 968         else
 969                 pos = udmu_zap_cursor_serialize(it->ozi_zc);
 970
 971         RETURN(pos);
 972 }
 973
 974 /*
 975  * return status :
 976  *  rc == 0 -> end of directory.
 977  *  rc >  0 -> ok, proceed.
 978  *  rc <  0 -> error.  ( EOVERFLOW  can be masked.)
 979  */
 980 static int osd_dir_it_load(const struct lu_env *env,
 981                         const struct dt_it *di, __u64 hash)
 982 {
 983         struct osd_zap_it *it = (struct osd_zap_it *)di;
 984         struct osd_object *obj = it->ozi_obj;
 985         struct osd_device *osd = osd_obj2dev(obj);
 986         zap_attribute_t   *za = &osd_oti_get(env)->oti_za;
 987         int                rc;
 988         ENTRY;
 989
 990         udmu_zap_cursor_fini(it->ozi_zc);
 991         if (udmu_zap_cursor_init(&it->ozi_zc, &osd->od_objset,
 992                                  obj->oo_db->db_object, hash))
 993                 RETURN(-ENOMEM);
 994
 995         if (hash <= 2) {
 996                 it->ozi_pos = hash;
 997                 rc = +1;
 998         } else {
 999                 it->ozi_pos = 3;
1000                 /* to return whether the end has been reached */
1001                 rc = osd_index_retrieve_skip_dots(it, za);
1002                 if (rc == 0)
1003                         rc = +1;
1004                 else if (rc == -ENOENT)
1005                         rc = 0;
1006         }
1007
1008         RETURN(rc);
1009 }
1010
1011 static struct dt_index_operations osd_dir_ops = {
1012         .dio_lookup         = osd_dir_lookup,
1013         .dio_declare_insert = osd_declare_dir_insert,
1014         .dio_insert         = osd_dir_insert,
1015         .dio_declare_delete = osd_declare_dir_delete,
1016         .dio_delete         = osd_dir_delete,
1017         .dio_it     = {
1018                 .init     = osd_dir_it_init,
1019                 .fini     = osd_index_it_fini,
1020                 .get      = osd_dir_it_get,
1021                 .put      = osd_dir_it_put,
1022                 .next     = osd_dir_it_next,
1023                 .key      = osd_dir_it_key,
1024                 .key_size = osd_dir_it_key_size,
1025                 .rec      = osd_dir_it_rec,
1026                 .store    = osd_dir_it_store,
1027                 .load     = osd_dir_it_load
1028         }
1029 };
1030
1031 /*
1032  * Primitives for index files using binary keys.
1033  * XXX: only 64-bit keys are supported for now.
1034  */
1035
1036 static int osd_index_lookup(const struct lu_env *env, struct dt_object *dt,
1037                         struct dt_rec *rec, const struct dt_key *key,
1038                         struct lustre_capa *capa)
1039 {
1040         struct osd_object *obj = osd_dt_obj(dt);
1041         struct osd_device *osd = osd_obj2dev(obj);
1042         int                rc;
1043         ENTRY;
1044
1045         rc = -zap_lookup_uint64(osd->od_objset.os, obj->oo_db->db_object,
1046                                 (const __u64 *)key, 1, 8, obj->oo_recsize,
1047                                 (void *)rec);
1048         RETURN(rc == 0 ? 1 : rc);
1049 }
1050
1051 static int osd_declare_index_insert(const struct lu_env *env,
1052                                     struct dt_object *dt,
1053                                     const struct dt_rec *rec,
1054                                     const struct dt_key *key,
1055                                     struct thandle *th)
1056 {
1057         struct osd_object  *obj = osd_dt_obj(dt);
1058         struct osd_thandle *oh;
1059         ENTRY;
1060
1061         LASSERT(th != NULL);
1062         oh = container_of0(th, struct osd_thandle, ot_super);
1063
1064         LASSERT(obj->oo_db);
1065
1066         dmu_tx_hold_bonus(oh->ot_tx, obj->oo_db->db_object);
1067
1068         /* It is not clear what API should be used for binary keys, so we pass
1069          * a null name which has the side effect of over-reserving space,
1070          * accounting for the worst case. See zap_count_write() */
1071         dmu_tx_hold_zap(oh->ot_tx, obj->oo_db->db_object, TRUE, NULL);
1072
1073         RETURN(0);
1074 }
1075
1076 static int osd_index_insert(const struct lu_env *env, struct dt_object *dt,
1077                             const struct dt_rec *rec, const struct dt_key *key,
1078                             struct thandle *th, struct lustre_capa *capa,
1079                             int ignore_quota)
1080 {
1081         struct osd_object  *obj = osd_dt_obj(dt);
1082         struct osd_device  *osd = osd_obj2dev(obj);
1083         struct osd_thandle *oh;
1084         int                 rc;
1085         ENTRY;
1086
1087         LASSERT(obj->oo_db);
1088         LASSERT(dt_object_exists(dt));
1089         LASSERT(osd_invariant(obj));
1090         LASSERT(th != NULL);
1091
1092         oh = container_of0(th, struct osd_thandle, ot_super);
1093
1094         /* Insert (key,oid) into ZAP */
1095         rc = -zap_add_uint64(osd->od_objset.os, obj->oo_db->db_object,
1096                              (const __u64 *)key, 1, 8, obj->oo_recsize,
1097                              (void *)rec, oh->ot_tx);
1098         RETURN(rc);
1099 }
1100
1101 static int osd_declare_index_delete(const struct lu_env *env,
1102                                     struct dt_object *dt,
1103                                     const struct dt_key *key,
1104                                     struct thandle *th)
1105 {
1106         struct osd_object  *obj = osd_dt_obj(dt);
1107         struct osd_thandle *oh;
1108         ENTRY;
1109
1110         LASSERT(dt_object_exists(dt));
1111         LASSERT(osd_invariant(obj));
1112         LASSERT(th != NULL);
1113         LASSERT(obj->oo_db);
1114
1115         oh = container_of0(th, struct osd_thandle, ot_super);
1116         dmu_tx_hold_zap(oh->ot_tx, obj->oo_db->db_object, TRUE, NULL);
1117
1118         RETURN(0);
1119 }
1120
1121 static int osd_index_delete(const struct lu_env *env, struct dt_object *dt,
1122                             const struct dt_key *key, struct thandle *th,
1123                             struct lustre_capa *capa)
1124 {
1125         struct osd_object  *obj = osd_dt_obj(dt);
1126         struct osd_device  *osd = osd_obj2dev(obj);
1127         struct osd_thandle *oh;
1128         int                 rc;
1129         ENTRY;
1130
1131         LASSERT(obj->oo_db);
1132         LASSERT(th != NULL);
1133         oh = container_of0(th, struct osd_thandle, ot_super);
1134
1135         /* Remove binary key from the ZAP */
1136         rc = -zap_remove_uint64(osd->od_objset.os, obj->oo_db->db_object,
1137                                 (const __u64 *)key, 1, oh->ot_tx);
1138         RETURN(rc);
1139 }
1140
1141 static int osd_index_it_get(const struct lu_env *env, struct dt_it *di,
1142                             const struct dt_key *key)
1143 {
1144         struct osd_zap_it *it = (struct osd_zap_it *)di;
1145         struct osd_object *obj = it->ozi_obj;
1146         struct osd_device *osd = osd_obj2dev(obj);
1147         ENTRY;
1148
1149         LASSERT(it);
1150         LASSERT(it->ozi_zc);
1151
1152         /* XXX: API is broken at the moment */
1153         LASSERT(*((const __u64 *)key) == 0);
1154
1155         zap_cursor_fini(it->ozi_zc);
1156         memset(it->ozi_zc, 0, sizeof(*it->ozi_zc));
1157         zap_cursor_init(it->ozi_zc, osd->od_objset.os, obj->oo_db->db_object);
1158         it->ozi_reset = 1;
1159
1160         RETURN(+1);
1161 }
1162
1163 static int osd_index_it_next(const struct lu_env *env, struct dt_it *di)
1164 {
1165         struct osd_zap_it *it = (struct osd_zap_it *)di;
1166         zap_attribute_t   *za = &osd_oti_get(env)->oti_za;
1167         int                rc;
1168         ENTRY;
1169
1170         if (it->ozi_reset == 0)
1171                 zap_cursor_advance(it->ozi_zc);
1172         it->ozi_reset = 0;
1173
1174         /*
1175          * According to current API we need to return error if it's last entry.
1176          * zap_cursor_advance() does not return any value. So we need to call
1177          * retrieve to check if there is any record.  We should make
1178          * changes to Iterator API to not return status for this API
1179          */
1180         rc = -zap_cursor_retrieve(it->ozi_zc, za);
1181         if (rc == -ENOENT)
1182                 RETURN(+1);
1183
1184         RETURN((rc));
1185 }
1186
1187 static struct dt_key *osd_index_it_key(const struct lu_env *env,
1188                                        const struct dt_it *di)
1189 {
1190         struct osd_zap_it *it = (struct osd_zap_it *)di;
1191         zap_attribute_t   *za = &osd_oti_get(env)->oti_za;
1192         int                rc = 0;
1193         ENTRY;
1194
1195         it->ozi_reset = 0;
1196         rc = -zap_cursor_retrieve(it->ozi_zc, za);
1197         if (rc)
1198                 RETURN(ERR_PTR(rc));
1199
1200         /* the binary key is stored in the name */
1201         it->ozi_key = *((__u64 *)za->za_name);
1202
1203         RETURN((struct dt_key *)&it->ozi_key);
1204 }
1205
1206 static int osd_index_it_key_size(const struct lu_env *env,
1207                                 const struct dt_it *di)
1208 {
1209         /* we only support 64-bit binary keys for the time being */
1210         RETURN(sizeof(__u64));
1211 }
1212
1213 static int osd_index_it_rec(const struct lu_env *env, const struct dt_it *di,
1214                             struct dt_rec *rec, __u32 attr)
1215 {
1216         zap_attribute_t   *za = &osd_oti_get(env)->oti_za;
1217         struct osd_zap_it *it = (struct osd_zap_it *)di;
1218         struct osd_object *obj = it->ozi_obj;
1219         struct osd_device *osd = osd_obj2dev(obj);
1220         int                rc;
1221         ENTRY;
1222
1223         it->ozi_reset = 0;
1224         rc = -zap_cursor_retrieve(it->ozi_zc, za);
1225         if (rc)
1226                 RETURN(rc);
1227
1228         rc = -zap_lookup_uint64(osd->od_objset.os, obj->oo_db->db_object,
1229                                 (const __u64 *)za->za_name, 1, 8,
1230                                 obj->oo_recsize, (void *)rec);
1231         RETURN(rc);
1232 }
1233
1234 static __u64 osd_index_it_store(const struct lu_env *env,
1235                                 const struct dt_it *di)
1236 {
1237         struct osd_zap_it *it = (struct osd_zap_it *)di;
1238
1239         it->ozi_reset = 0;
1240         RETURN((__u64)zap_cursor_serialize(it->ozi_zc));
1241 }
1242
1243 static int osd_index_it_load(const struct lu_env *env, const struct dt_it *di,
1244                              __u64 hash)
1245 {
1246         struct osd_zap_it *it = (struct osd_zap_it *)di;
1247         struct osd_object *obj = it->ozi_obj;
1248         struct osd_device *osd = osd_obj2dev(obj);
1249         zap_attribute_t   *za = &osd_oti_get(env)->oti_za;
1250         int                rc;
1251         ENTRY;
1252
1253         /* close the current cursor */
1254         zap_cursor_fini(it->ozi_zc);
1255
1256         /* create a new one starting at hash */
1257         memset(it->ozi_zc, 0, sizeof(*it->ozi_zc));
1258         zap_cursor_init_serialized(it->ozi_zc, osd->od_objset.os,
1259                                    obj->oo_db->db_object, hash);
1260         it->ozi_reset = 0;
1261
1262         rc = -zap_cursor_retrieve(it->ozi_zc, za);
1263         if (rc == 0)
1264                 RETURN(+1);
1265         else if (rc == -ENOENT)
1266                 RETURN(0);
1267
1268         RETURN(rc);
1269 }
1270
1271 static struct dt_index_operations osd_index_ops = {
1272         .dio_lookup             = osd_index_lookup,
1273         .dio_declare_insert     = osd_declare_index_insert,
1274         .dio_insert             = osd_index_insert,
1275         .dio_declare_delete     = osd_declare_index_delete,
1276         .dio_delete             = osd_index_delete,
1277         .dio_it = {
1278                 .init           = osd_index_it_init,
1279                 .fini           = osd_index_it_fini,
1280                 .get            = osd_index_it_get,
1281                 .put            = osd_index_it_put,
1282                 .next           = osd_index_it_next,
1283                 .key            = osd_index_it_key,
1284                 .key_size       = osd_index_it_key_size,
1285                 .rec            = osd_index_it_rec,
1286                 .store          = osd_index_it_store,
1287                 .load           = osd_index_it_load
1288         }
1289 };
1290
1291 int osd_index_try(const struct lu_env *env, struct dt_object *dt,
1292                 const struct dt_index_features *feat)
1293 {
1294         struct osd_object *obj = osd_dt_obj(dt);
1295         ENTRY;
1296
1297         LASSERT(dt_object_exists(dt));
1298
1299         /*
1300          * XXX: implement support for fixed-size keys sorted with natural
1301          *      numerical way (not using internal hash value)
1302          */
1303         if (feat->dif_flags & DT_IND_RANGE)
1304                 RETURN(-ERANGE);
1305
1306         if (unlikely(feat == &dt_otable_features))
1307                 /* do not support oi scrub yet. */
1308                 RETURN(-ENOTSUPP);
1309
1310         LASSERT(obj->oo_db != NULL);
1311         if (likely(feat == &dt_directory_features)) {
1312                 if (udmu_object_is_zap(obj->oo_db))
1313                         dt->do_index_ops = &osd_dir_ops;
1314                 else
1315                         RETURN(-ENOTDIR);
1316         } else if (unlikely(feat == &dt_acct_features)) {
1317                 LASSERT(fid_is_acct(lu_object_fid(&dt->do_lu)));
1318                 dt->do_index_ops = &osd_acct_index_ops;
1319         } else if (udmu_object_is_zap(obj->oo_db) &&
1320                    dt->do_index_ops == NULL) {
1321                 /* For index file, we don't support variable key & record sizes
1322                  * and the key has to be unique */
1323                 if ((feat->dif_flags & ~DT_IND_UPDATE) != 0)
1324                         RETURN(-EINVAL);
1325
1326                 /* Although the zap_*_uint64() primitives support large keys, we
1327                  * limit ourselves to 64-bit keys for now */
1328                 if (feat->dif_keysize_max != sizeof(__u64) ||
1329                     feat->dif_keysize_min != sizeof(__u64))
1330                         RETURN(-EINVAL);
1331
1332                 /* As for the record size, it should be a multiple of 8 bytes
1333                  * and smaller than the maximum value length supported by ZAP.
1334                  */
1335                 if (feat->dif_recsize_max > ZAP_MAXVALUELEN)
1336                         RETURN(-E2BIG);
1337                 if (feat->dif_recsize_max != feat->dif_recsize_min ||
1338                     (feat->dif_recsize_max & (sizeof(__u64) - 1)))
1339                         RETURN(-EINVAL);
1340
1341                 obj->oo_recsize = feat->dif_recsize_max / sizeof(__u64);
1342                 dt->do_index_ops = &osd_index_ops;
1343         }
1344
1345         RETURN(0);
1346 }
1347