lustre/osd-zfs/osd_xattr.c

   1 /*
   2  * GPL HEADER START
   3  *
   4  * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.
   5  *
   6  * This program is free software; you can redistribute it and/or modify
   7  * it under the terms of the GNU General Public License version 2 only,
   8  * as published by the Free Software Foundation.
   9  *
  10  * This program is distributed in the hope that it will be useful, but
  11  * WITHOUT ANY WARRANTY; without even the implied warranty of
  12  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
  13  * General Public License version 2 for more details (a copy is included
  14  * in the LICENSE file that accompanied this code).
  15  *
  16  * You should have received a copy of the GNU General Public License
  17  * version 2 along with this program; If not, see
  18  * http://www.gnu.org/licenses/gpl-2.0.html
  19  *
  20  * GPL HEADER END
  21  */
  22 /*
  23  * Copyright (c) 2009, 2010, Oracle and/or its affiliates. All rights reserved.
  24  * Use is subject to license terms.
  25  *
  26  * Copyright (c) 2012, 2017, Intel Corporation.
  27  */
  28 /*
  29  * This file is part of Lustre, http://www.lustre.org/
  30  *
  31  * lustre/osd-zfs/osd_xattr.c
  32  * functions to manipulate extended attributes and system attributes
  33  *
  34  * Author: Alex Zhuravlev <bzzz@whamcloud.com>
  35  * Author: Mike Pershin <tappro@whamcloud.com>
  36  */
  37
  38 #define DEBUG_SUBSYSTEM S_OSD
  39
  40 #include <libcfs/libcfs.h>
  41 #include <obd_support.h>
  42 #include <lustre_net.h>
  43 #include <obd.h>
  44 #include <obd_class.h>
  45 #include <lustre_disk.h>
  46 #include <lustre_fid.h>
  47 #include <lustre_linkea.h>
  48
  49 #include "osd_internal.h"
  50
  51 #include <sys/dnode.h>
  52 #include <sys/dbuf.h>
  53 #include <sys/spa.h>
  54 #include <sys/stat.h>
  55 #include <sys/zap.h>
  56 #include <sys/spa_impl.h>
  57 #include <sys/zfs_znode.h>
  58 #include <sys/dmu_tx.h>
  59 #include <sys/dmu_objset.h>
  60 #include <sys/dsl_prop.h>
  61 #include <sys/sa_impl.h>
  62 #include <sys/txg.h>
  63
  64 #include <linux/posix_acl_xattr.h>
  65 #include <lustre_scrub.h>
  66
  67 int __osd_xattr_load(struct osd_device *osd, sa_handle_t *hdl, nvlist_t **sa)
  68 {
  69         char        *buf;
  70         int          rc, size;
  71
  72         rc = -sa_size(hdl, SA_ZPL_DXATTR(osd), &size);
  73         if (rc) {
  74                 if (rc == -ENOENT)
  75                         rc = -nvlist_alloc(sa, NV_UNIQUE_NAME, KM_SLEEP);
  76                 goto out_sa;
  77         }
  78
  79         buf = zio_buf_alloc(size);
  80         if (buf == NULL) {
  81                 rc = -ENOMEM;
  82                 goto out_sa;
  83         }
  84         rc = -sa_lookup(hdl, SA_ZPL_DXATTR(osd), buf, size);
  85         if (rc == 0)
  86                 rc = -nvlist_unpack(buf, size, sa, KM_SLEEP);
  87         zio_buf_free(buf, size);
  88 out_sa:
  89
  90         return rc;
  91 }
  92
  93 static inline int __osd_xattr_cache(struct osd_object *obj)
  94 {
  95         LASSERT(obj->oo_sa_hdl);
  96         if (obj->oo_sa_xattr != NULL)
  97                 return 0;
  98         return __osd_xattr_load(osd_obj2dev(obj),
  99                                 obj->oo_sa_hdl, &obj->oo_sa_xattr);
 100 }
 101
 102 static int
 103 __osd_sa_xattr_get(const struct lu_env *env, struct osd_object *obj,
 104                    const struct lu_buf *buf, const char *name, int *sizep)
 105 {
 106         uchar_t *nv_value;
 107         int      rc = 0;
 108
 109         rc = __osd_xattr_cache(obj);
 110         if (rc)
 111                 return rc;
 112
 113         LASSERT(obj->oo_sa_xattr);
 114         rc = -nvlist_lookup_byte_array(obj->oo_sa_xattr, name,
 115                                        &nv_value, sizep);
 116         if (rc)
 117                 return rc;
 118
 119         if (buf == NULL || buf->lb_buf == NULL) {
 120                 /* return the required size by *sizep */
 121                 return 0;
 122         }
 123
 124         if (*sizep > buf->lb_len)
 125                 return -ERANGE; /* match ldiskfs error */
 126
 127         memcpy(buf->lb_buf, nv_value, *sizep);
 128         return 0;
 129 }
 130
 131 int __osd_xattr_get_large(const struct lu_env *env, struct osd_device *osd,
 132                           uint64_t xattr, struct lu_buf *buf,
 133                           const char *name, int *sizep)
 134 {
 135         dnode_t         *xa_data_dn;
 136         sa_handle_t *sa_hdl = NULL;
 137         uint64_t         xa_data_obj, size;
 138         int              rc;
 139
 140         /* are there any extended attributes? */
 141         if (xattr == ZFS_NO_OBJECT)
 142                 return -ENOENT;
 143
 144         /* Lookup the object number containing the xattr data */
 145         rc = -zap_lookup(osd->od_os, xattr, name, sizeof(uint64_t), 1,
 146                         &xa_data_obj);
 147         if (rc)
 148                 return rc;
 149
 150         rc = __osd_obj2dnode(osd->od_os, xa_data_obj, &xa_data_dn);
 151         if (rc)
 152                 return rc;
 153
 154         rc = -sa_handle_get(osd->od_os, xa_data_obj, NULL, SA_HDL_PRIVATE,
 155                         &sa_hdl);
 156         if (rc)
 157                 goto out_rele;
 158
 159         /* Get the xattr value length / object size */
 160         rc = -sa_lookup(sa_hdl, SA_ZPL_SIZE(osd), &size, 8);
 161         if (rc)
 162                 goto out;
 163
 164         if (size > INT_MAX) {
 165                 rc = -EOVERFLOW;
 166                 goto out;
 167         }
 168
 169         *sizep = (int)size;
 170
 171         if (buf == NULL || buf->lb_buf == NULL) {
 172                 /* We only need to return the required size */
 173                 goto out;
 174         }
 175         if (*sizep > buf->lb_len) {
 176                 rc = -ERANGE; /* match ldiskfs error */
 177                 goto out;
 178         }
 179
 180         rc = -dmu_read(osd->od_os, xa_data_dn->dn_object, 0,
 181                         size, buf->lb_buf, DMU_READ_PREFETCH);
 182
 183 out:
 184         sa_handle_destroy(sa_hdl);
 185 out_rele:
 186         osd_dnode_rele(xa_data_dn);
 187
 188         return rc;
 189 }
 190
 191 /**
 192  * Copy an extended attribute into the buffer provided, or compute
 193  * the required buffer size if \a buf is NULL.
 194  *
 195  * On success, the number of bytes used or required is stored in \a sizep.
 196  *
 197  * Note that no locking is done here.
 198  *
 199  * \param[in] env      execution environment
 200  * \param[in] obj      object for which to retrieve xattr
 201  * \param[out] buf     buffer to store xattr value in
 202  * \param[in] name     name of xattr to copy
 203  * \param[out] sizep   bytes used or required to store xattr
 204  *
 205  * \retval 0           on success
 206  * \retval negative    negated errno on failure
 207  */
 208 int osd_xattr_get_internal(const struct lu_env *env, struct osd_object *obj,
 209                            struct lu_buf *buf, const char *name, int *sizep)
 210 {
 211         int rc;
 212
 213         if (unlikely(!dt_object_exists(&obj->oo_dt) || obj->oo_destroyed))
 214                 return -ENOENT;
 215
 216         /* check SA_ZPL_DXATTR first then fallback to directory xattr */
 217         rc = __osd_sa_xattr_get(env, obj, buf, name, sizep);
 218         if (rc != -ENOENT)
 219                 return rc;
 220
 221         return __osd_xattr_get_large(env, osd_obj2dev(obj), obj->oo_xattr,
 222                                      buf, name, sizep);
 223 }
 224
 225 /**
 226  * Copy LMA extended attribute into provided buffer
 227  *
 228  * Note that no locking is done here.
 229  *
 230  * \param[in] env      execution environment
 231  * \param[in] obj      object for which to retrieve xattr
 232  * \param[out] buf     buffer to store xattr value in
 233  *
 234  * \retval 0           on success
 235  * \retval negative    negated errno on failure
 236  */
 237 int osd_xattr_get_lma(const struct lu_env *env, struct osd_object *obj,
 238                       struct lu_buf *buf)
 239 {
 240         int size = 0;
 241         int rc = -ENOENT;
 242
 243         if (!buf)
 244                 return 0;
 245
 246         if (unlikely(obj->oo_destroyed))
 247                 goto out_lma;
 248
 249         /* check SA_ZPL_DXATTR first then fallback to directory xattr */
 250         rc = __osd_sa_xattr_get(env, obj, buf, XATTR_NAME_LMA, &size);
 251         if (!rc && unlikely(size < sizeof(struct lustre_mdt_attrs)))
 252                 rc = -EINVAL;
 253         if (rc != -ENOENT)
 254                 goto out_lma;
 255
 256         rc = __osd_xattr_get_large(env, osd_obj2dev(obj), obj->oo_xattr,
 257                                      buf, XATTR_NAME_LMA, &size);
 258         if (!rc && unlikely(size < sizeof(struct lustre_mdt_attrs)))
 259                 rc = -EINVAL;
 260
 261 out_lma:
 262         return rc;
 263 }
 264
 265 static int osd_get_pfid_from_lma(const struct lu_env *env,
 266                                  struct osd_object *obj,
 267                                  struct lu_buf *buf, int *sizep)
 268 {
 269         struct osd_thread_info *info = osd_oti_get(env);
 270         struct lustre_ost_attrs *loa =
 271                 (struct lustre_ost_attrs *)&info->oti_buf;
 272         struct lustre_mdt_attrs *lma = &loa->loa_lma;
 273         struct filter_fid *ff;
 274         struct ost_layout *ol;
 275         struct lu_buf tbuf = {
 276                 .lb_buf = loa,
 277                 .lb_len = sizeof(info->oti_buf),
 278         };
 279         int rc;
 280         ENTRY;
 281
 282         BUILD_BUG_ON(sizeof(info->oti_buf) < sizeof(*loa));
 283         rc = osd_xattr_get_internal(env, obj, &tbuf,
 284                                     XATTR_NAME_LMA, sizep);
 285         if (rc)
 286                 RETURN(rc);
 287
 288         lustre_loa_swab(loa, true);
 289         LASSERT(lma->lma_compat & LMAC_STRIPE_INFO);
 290
 291         *sizep = sizeof(*ff);
 292         if (buf->lb_len == 0 || !buf->lb_buf)
 293                 RETURN(0);
 294
 295         if (buf->lb_len < *sizep)
 296                 RETURN(-ERANGE);
 297
 298         ff = buf->lb_buf;
 299         ol = &ff->ff_layout;
 300         ol->ol_stripe_count = cpu_to_le32(loa->loa_parent_fid.f_ver >>
 301                                           PFID_STRIPE_IDX_BITS);
 302         ol->ol_stripe_size = cpu_to_le32(loa->loa_stripe_size);
 303         loa->loa_parent_fid.f_ver &= PFID_STRIPE_COUNT_MASK;
 304         fid_cpu_to_le(&ff->ff_parent, &loa->loa_parent_fid);
 305         if (lma->lma_compat & LMAC_COMP_INFO) {
 306                 ol->ol_comp_start = cpu_to_le64(loa->loa_comp_start);
 307                 ol->ol_comp_end = cpu_to_le64(loa->loa_comp_end);
 308                 ol->ol_comp_id = cpu_to_le32(loa->loa_comp_id);
 309         } else {
 310                 ol->ol_comp_start = 0;
 311                 ol->ol_comp_end = 0;
 312                 ol->ol_comp_id = 0;
 313         }
 314
 315         RETURN(0);
 316 }
 317
 318 int osd_xattr_get(const struct lu_env *env, struct dt_object *dt,
 319                   struct lu_buf *buf, const char *name)
 320 {
 321         struct osd_object  *obj  = osd_dt_obj(dt);
 322         int                 rc, size = 0;
 323         ENTRY;
 324
 325         if (!osd_obj2dev(obj)->od_posix_acl &&
 326             (strcmp(name, XATTR_NAME_POSIX_ACL_ACCESS) == 0 ||
 327              strcmp(name, XATTR_NAME_POSIX_ACL_DEFAULT) == 0))
 328                 RETURN(-EOPNOTSUPP);
 329
 330         down_read(&obj->oo_guard);
 331         if (unlikely(!dt_object_exists(dt) || obj->oo_destroyed))
 332                 GOTO(out, rc = -ENOENT);
 333         LASSERT(obj->oo_dn != NULL);
 334
 335         /* For the OST migrated from ldiskfs, the PFID EA may
 336          * be stored in LMA because of ldiskfs inode size. */
 337         if (strcmp(name, XATTR_NAME_FID) == 0 && obj->oo_pfid_in_lma)
 338                 rc = osd_get_pfid_from_lma(env, obj, buf, &size);
 339         else
 340                 rc = osd_xattr_get_internal(env, obj, buf, name, &size);
 341
 342         if (rc == -ENOENT)
 343                 rc = -ENODATA;
 344         else if (rc == 0)
 345                 rc = size;
 346 out:
 347         up_read(&obj->oo_guard);
 348         RETURN(rc);
 349 }
 350
 351 /* the function is used to declare EAs when SA is not supported */
 352 static void __osd_xattr_declare_legacy(const struct lu_env *env,
 353                                        struct osd_object *obj,
 354                                        int vallen, const char *name,
 355                                        struct osd_thandle *oh)
 356 {
 357         struct osd_device *osd = osd_obj2dev(obj);
 358         dmu_tx_t *tx = oh->ot_tx;
 359         uint64_t xa_data_obj;
 360         int rc;
 361
 362         if (obj->oo_xattr == ZFS_NO_OBJECT) {
 363                 /* xattr zap + entry */
 364                 dmu_tx_hold_zap(tx, DMU_NEW_OBJECT, TRUE, (char *) name);
 365                 /* xattr value obj */
 366                 dmu_tx_hold_sa_create(tx, ZFS_SA_BASE_ATTR_SIZE);
 367                 dmu_tx_hold_write(tx, DMU_NEW_OBJECT, 0, vallen);
 368                 return;
 369         }
 370
 371         rc = -zap_lookup(osd->od_os, obj->oo_xattr, name, sizeof(uint64_t), 1,
 372                         &xa_data_obj);
 373         if (rc == 0) {
 374                 /*
 375                  * Entry already exists.
 376                  * We'll truncate the existing object.
 377                  */
 378                 dmu_tx_hold_bonus(tx, xa_data_obj);
 379                 dmu_tx_hold_free(tx, xa_data_obj, vallen, DMU_OBJECT_END);
 380                 dmu_tx_hold_write(tx, xa_data_obj, 0, vallen);
 381         } else if (rc == -ENOENT) {
 382                 /*
 383                  * Entry doesn't exist, we need to create a new one and a new
 384                  * object to store the value.
 385                  */
 386                 dmu_tx_hold_bonus(tx, obj->oo_xattr);
 387                 dmu_tx_hold_zap(tx, obj->oo_xattr, TRUE, (char *) name);
 388                 dmu_tx_hold_sa_create(tx, ZFS_SA_BASE_ATTR_SIZE);
 389                 dmu_tx_hold_write(tx, DMU_NEW_OBJECT, 0, vallen);
 390         }
 391 }
 392
 393 void __osd_xattr_declare_set(const struct lu_env *env, struct osd_object *obj,
 394                              int vallen, const char *name,
 395                              struct osd_thandle *oh)
 396 {
 397         struct osd_device *osd = osd_obj2dev(obj);
 398         dmu_tx_t *tx = oh->ot_tx;
 399         int bonuslen;
 400
 401         if (unlikely(obj->oo_destroyed))
 402                 return;
 403
 404         if (strcmp(name, XATTR_NAME_LINK) == 0 &&
 405             osd->od_remote_parent_dir != ZFS_NO_OBJECT) {
 406                 /* If some name entry resides on remote MDT, then will create
 407                  * agent entry under remote parent. On the other hand, if the
 408                  * remote entry will be removed, then related agent entry may
 409                  * need to be removed from the remote parent. So there may be
 410                  * kinds of cases, let's declare enough credits. The credits
 411                  * for create agent entry is enough for remove case. */
 412                 osd_tx_hold_zap(tx, osd->od_remote_parent_dir,
 413                                 NULL, TRUE, NULL);
 414         }
 415
 416         if (unlikely(!osd_obj2dev(obj)->od_xattr_in_sa)) {
 417                 __osd_xattr_declare_legacy(env, obj, vallen, name, oh);
 418                 return;
 419         }
 420
 421         /* declare EA in SA */
 422         if (dt_object_exists(&obj->oo_dt)) {
 423                 LASSERT(obj->oo_sa_hdl);
 424                 /* XXX: it should be possible to skip spill
 425                  * declaration if specific EA is part of
 426                  * bonus and doesn't grow */
 427                 dmu_tx_hold_spill(tx, obj->oo_dn->dn_object);
 428                 return;
 429         }
 430
 431         bonuslen = osd_obj_bonuslen(obj);
 432
 433         /* the object doesn't exist, but we've declared bonus
 434          * in osd_declare_object_create() yet */
 435         if (obj->oo_ea_in_bonus > bonuslen) {
 436                 /* spill has been declared already */
 437         } else if (obj->oo_ea_in_bonus + vallen > bonuslen) {
 438                 /* we're about to exceed bonus, let's declare spill */
 439                 dmu_tx_hold_spill(tx, DMU_NEW_OBJECT);
 440         }
 441         obj->oo_ea_in_bonus += vallen;
 442 }
 443
 444 int osd_declare_xattr_set(const struct lu_env *env, struct dt_object *dt,
 445                           const struct lu_buf *buf, const char *name,
 446                           int fl, struct thandle *handle)
 447 {
 448         struct osd_object  *obj = osd_dt_obj(dt);
 449         struct osd_thandle *oh;
 450         ENTRY;
 451
 452         LASSERT(handle != NULL);
 453         oh = container_of(handle, struct osd_thandle, ot_super);
 454
 455         down_read(&obj->oo_guard);
 456         __osd_xattr_declare_set(env, obj, buf->lb_len, name, oh);
 457         up_read(&obj->oo_guard);
 458
 459         RETURN(0);
 460 }
 461
 462 int __osd_sa_attr_init(const struct lu_env *env, struct osd_object *obj,
 463                        struct osd_thandle *oh)
 464 {
 465         sa_bulk_attr_t *bulk = osd_oti_get(env)->oti_attr_bulk;
 466         struct osa_attr *osa = &osd_oti_get(env)->oti_osa;
 467         struct lu_buf *lb = &osd_oti_get(env)->oti_xattr_lbuf;
 468         struct osd_device *osd = osd_obj2dev(obj);
 469         uint64_t gen;
 470         inode_timespec_t now;
 471         size_t size;
 472         int rc, cnt;
 473
 474         obj->oo_late_xattr = 0;
 475         obj->oo_late_attr_set = 0;
 476
 477         gen = dmu_tx_get_txg(oh->ot_tx);
 478         gethrestime(&now);
 479         ZFS_TIME_ENCODE(&now, osa->btime);
 480
 481         obj->oo_attr.la_valid |= LA_BTIME;
 482         obj->oo_attr.la_btime = osa->btime[0];
 483         osa->atime[0] = obj->oo_attr.la_atime;
 484         osa->ctime[0] = obj->oo_attr.la_ctime;
 485         osa->mtime[0] = obj->oo_attr.la_mtime;
 486         osa->mode = obj->oo_attr.la_mode;
 487         osa->uid = obj->oo_attr.la_uid;
 488         osa->gid = obj->oo_attr.la_gid;
 489         osa->rdev = obj->oo_attr.la_rdev;
 490         osa->nlink = obj->oo_attr.la_nlink;
 491         osa->flags = attrs_fs2zfs(obj->oo_attr.la_flags);
 492         osa->size  = obj->oo_attr.la_size;
 493 #ifdef ZFS_PROJINHERIT
 494         if (osd->od_projectused_dn) {
 495                 if (obj->oo_attr.la_valid & LA_PROJID)
 496                         osa->projid = obj->oo_attr.la_projid;
 497                 else
 498                         osa->projid = ZFS_DEFAULT_PROJID;
 499                 osa->flags |= ZFS_PROJID;
 500                 obj->oo_with_projid = 1;
 501         }
 502 #endif
 503
 504         cnt = 0;
 505         SA_ADD_BULK_ATTR(bulk, cnt, SA_ZPL_MODE(osd), NULL, &osa->mode, 8);
 506         SA_ADD_BULK_ATTR(bulk, cnt, SA_ZPL_SIZE(osd), NULL, &osa->size, 8);
 507         SA_ADD_BULK_ATTR(bulk, cnt, SA_ZPL_GEN(osd), NULL, &gen, 8);
 508         SA_ADD_BULK_ATTR(bulk, cnt, SA_ZPL_UID(osd), NULL, &osa->uid, 8);
 509         SA_ADD_BULK_ATTR(bulk, cnt, SA_ZPL_GID(osd), NULL, &osa->gid, 8);
 510         SA_ADD_BULK_ATTR(bulk, cnt, SA_ZPL_PARENT(osd), NULL,
 511                          &obj->oo_parent, 8);
 512         SA_ADD_BULK_ATTR(bulk, cnt, SA_ZPL_FLAGS(osd), NULL, &osa->flags, 8);
 513         SA_ADD_BULK_ATTR(bulk, cnt, SA_ZPL_ATIME(osd), NULL, osa->atime, 16);
 514         SA_ADD_BULK_ATTR(bulk, cnt, SA_ZPL_MTIME(osd), NULL, osa->mtime, 16);
 515         SA_ADD_BULK_ATTR(bulk, cnt, SA_ZPL_CTIME(osd), NULL, osa->ctime, 16);
 516         SA_ADD_BULK_ATTR(bulk, cnt, SA_ZPL_CRTIME(osd), NULL, osa->btime, 16);
 517         SA_ADD_BULK_ATTR(bulk, cnt, SA_ZPL_LINKS(osd), NULL, &osa->nlink, 8);
 518 #ifdef ZFS_PROJINHERIT
 519         if (osd->od_projectused_dn)
 520                 SA_ADD_BULK_ATTR(bulk, cnt, SA_ZPL_PROJID(osd), NULL,
 521                                  &osa->projid, 8);
 522 #endif
 523         SA_ADD_BULK_ATTR(bulk, cnt, SA_ZPL_RDEV(osd), NULL, &osa->rdev, 8);
 524         LASSERT(cnt <= ARRAY_SIZE(osd_oti_get(env)->oti_attr_bulk));
 525
 526         /* Update the SA for additions, modifications, and removals. */
 527         rc = -nvlist_size(obj->oo_sa_xattr, &size, NV_ENCODE_XDR);
 528         if (rc)
 529                 return rc;
 530
 531         lu_buf_check_and_alloc(lb, size);
 532         if (lb->lb_buf == NULL) {
 533                 CERROR("%s: can't allocate buffer for xattr update\n",
 534                                 osd->od_svname);
 535                 return -ENOMEM;
 536         }
 537
 538         rc = -nvlist_pack(obj->oo_sa_xattr, (char **)&lb->lb_buf, &size,
 539                           NV_ENCODE_XDR, KM_SLEEP);
 540         if (rc)
 541                 return rc;
 542
 543         SA_ADD_BULK_ATTR(bulk, cnt, SA_ZPL_DXATTR(osd), NULL, lb->lb_buf, size);
 544
 545         rc = -sa_replace_all_by_template(obj->oo_sa_hdl, bulk, cnt, oh->ot_tx);
 546
 547         return rc;
 548 }
 549
 550 int __osd_sa_xattr_update(const struct lu_env *env, struct osd_object *obj,
 551                            struct osd_thandle *oh)
 552 {
 553         struct lu_buf     *lb = &osd_oti_get(env)->oti_xattr_lbuf;
 554         struct osd_device *osd = osd_obj2dev(obj);
 555         char              *dxattr;
 556         size_t             size;
 557         int                rc;
 558
 559         obj->oo_late_xattr = 0;
 560
 561         /* Update the SA for additions, modifications, and removals. */
 562         rc = -nvlist_size(obj->oo_sa_xattr, &size, NV_ENCODE_XDR);
 563         if (rc)
 564                 return rc;
 565
 566         lu_buf_check_and_alloc(lb, size);
 567         if (lb->lb_buf == NULL) {
 568                 CERROR("%s: can't allocate buffer for xattr update\n",
 569                                 osd->od_svname);
 570                 return -ENOMEM;
 571         }
 572
 573         dxattr = lb->lb_buf;
 574         rc = -nvlist_pack(obj->oo_sa_xattr, &dxattr, &size,
 575                         NV_ENCODE_XDR, KM_SLEEP);
 576         if (rc)
 577                 return rc;
 578         LASSERT(dxattr == lb->lb_buf);
 579
 580         sa_update(obj->oo_sa_hdl, SA_ZPL_DXATTR(osd), dxattr, size, oh->ot_tx);
 581
 582         return 0;
 583 }
 584
 585 /*
 586  * Set an extended attribute.
 587  * This transaction must have called udmu_xattr_declare_set() first.
 588  *
 589  * Returns 0 on success or a negative error number on failure.
 590  *
 591  * No locking is done here.
 592  */
 593 int __osd_sa_xattr_schedule_update(const struct lu_env *env,
 594                                    struct osd_object *obj,
 595                                    struct osd_thandle *oh)
 596 {
 597         ENTRY;
 598         LASSERT(obj->oo_sa_hdl);
 599         LASSERT(obj->oo_sa_xattr);
 600
 601         /* schedule batched SA update in osd_object_sa_dirty_rele() */
 602         obj->oo_late_xattr = 1;
 603         osd_object_sa_dirty_add(obj, oh);
 604
 605         RETURN(0);
 606
 607 }
 608
 609 int __osd_sa_xattr_set(const struct lu_env *env, struct osd_object *obj,
 610                        const struct lu_buf *buf, const char *name, int fl,
 611                        struct osd_thandle *oh)
 612 {
 613         uchar_t *nv_value;
 614         size_t  size;
 615         int     nv_size;
 616         int     rc;
 617         int     too_big = 0;
 618
 619         rc = __osd_xattr_cache(obj);
 620         if (rc)
 621                 return rc;
 622
 623         LASSERT(obj->oo_sa_xattr);
 624         if (buf->lb_len > OBD_MAX_EA_SIZE) {
 625                 too_big = 1;
 626         } else {
 627                 /* Prevent the DXATTR SA from consuming the entire SA
 628                  * region */
 629                 rc = -nvlist_size(obj->oo_sa_xattr, &size, NV_ENCODE_XDR);
 630                 if (rc)
 631                         return rc;
 632
 633                 if (size + buf->lb_len > DXATTR_MAX_SA_SIZE)
 634                         too_big = 1;
 635         }
 636
 637         /* even in case of -EFBIG we must lookup xattr and check can we
 638          * rewrite it then delete from SA */
 639         rc = -nvlist_lookup_byte_array(obj->oo_sa_xattr, name, &nv_value,
 640                                         &nv_size);
 641         if (rc == 0) {
 642                 if (fl & LU_XATTR_CREATE) {
 643                         return -EEXIST;
 644                 } else if (too_big) {
 645                         rc = -nvlist_remove(obj->oo_sa_xattr, name,
 646                                                 DATA_TYPE_BYTE_ARRAY);
 647                         if (rc < 0)
 648                                 return rc;
 649                         rc = __osd_sa_xattr_schedule_update(env, obj, oh);
 650                         return rc == 0 ? -EFBIG : rc;
 651                 }
 652         } else if (rc == -ENOENT) {
 653                 if (fl & LU_XATTR_REPLACE)
 654                         return -ENODATA;
 655                 else if (too_big)
 656                         return -EFBIG;
 657         } else {
 658                 return rc;
 659         }
 660
 661         /* Ensure xattr doesn't exist in ZAP */
 662         if (obj->oo_xattr != ZFS_NO_OBJECT) {
 663                 struct osd_device *osd = osd_obj2dev(obj);
 664                 uint64_t           objid;
 665                 rc = -zap_lookup(osd->od_os, obj->oo_xattr,
 666                                  name, 8, 1, &objid);
 667                 if (rc == 0) {
 668                         rc = -dmu_object_free(osd->od_os, objid, oh->ot_tx);
 669                         if (rc == 0)
 670                                 zap_remove(osd->od_os, obj->oo_xattr,
 671                                            name, oh->ot_tx);
 672                 }
 673         }
 674
 675         rc = -nvlist_add_byte_array(obj->oo_sa_xattr, name,
 676                                     (uchar_t *)buf->lb_buf, buf->lb_len);
 677         if (rc)
 678                 return rc;
 679
 680         /* batch updates only for just created dnodes where we
 681          * used to set number of EAs in a single transaction */
 682         if (obj->oo_dn->dn_allocated_txg == oh->ot_tx->tx_txg)
 683                 rc = __osd_sa_xattr_schedule_update(env, obj, oh);
 684         else
 685                 rc = __osd_sa_xattr_update(env, obj, oh);
 686
 687         return rc;
 688 }
 689
 690 int
 691 __osd_xattr_set(const struct lu_env *env, struct osd_object *obj,
 692                 const struct lu_buf *buf, const char *name, int fl,
 693                 struct osd_thandle *oh)
 694 {
 695         struct osd_device *osd = osd_obj2dev(obj);
 696         dnode_t *xa_zap_dn = NULL;
 697         dnode_t *xa_data_dn = NULL;
 698         uint64_t           xa_data_obj;
 699         sa_handle_t       *sa_hdl = NULL;
 700         dmu_tx_t          *tx = oh->ot_tx;
 701         uint64_t           size;
 702         int                rc;
 703
 704         LASSERT(obj->oo_sa_hdl);
 705
 706         if (obj->oo_xattr == ZFS_NO_OBJECT) {
 707                 struct lu_attr *la = &osd_oti_get(env)->oti_la;
 708
 709                 la->la_valid = LA_MODE;
 710                 la->la_mode = S_IFDIR | S_IRUGO | S_IWUSR | S_IXUGO;
 711                 rc = __osd_zap_create(env, osd, &xa_zap_dn, tx, la, 0, 0);
 712                 if (rc)
 713                         return rc;
 714
 715                 obj->oo_xattr = xa_zap_dn->dn_object;
 716                 rc = osd_object_sa_update(obj, SA_ZPL_XATTR(osd),
 717                                 &obj->oo_xattr, 8, oh);
 718                 if (rc)
 719                         goto out;
 720         }
 721
 722         rc = -zap_lookup(osd->od_os, obj->oo_xattr, name, sizeof(uint64_t), 1,
 723                          &xa_data_obj);
 724         if (rc == 0) {
 725                 if (fl & LU_XATTR_CREATE) {
 726                         rc = -EEXIST;
 727                         goto out;
 728                 }
 729                 /*
 730                  * Entry already exists.
 731                  * We'll truncate the existing object.
 732                  */
 733                 rc = __osd_obj2dnode(osd->od_os, xa_data_obj, &xa_data_dn);
 734                 if (rc)
 735                         goto out;
 736
 737                 rc = -sa_handle_get(osd->od_os, xa_data_obj, NULL,
 738                                         SA_HDL_PRIVATE, &sa_hdl);
 739                 if (rc)
 740                         goto out;
 741
 742                 rc = -sa_lookup(sa_hdl, SA_ZPL_SIZE(osd), &size, 8);
 743                 if (rc)
 744                         goto out_sa;
 745
 746                 rc = -dmu_free_range(osd->od_os, xa_data_dn->dn_object,
 747                                      0, DMU_OBJECT_END, tx);
 748                 if (rc)
 749                         goto out_sa;
 750         } else if (rc == -ENOENT) {
 751                 struct lu_attr *la = &osd_oti_get(env)->oti_la;
 752                 /*
 753                  * Entry doesn't exist, we need to create a new one and a new
 754                  * object to store the value.
 755                  */
 756                 if (fl & LU_XATTR_REPLACE) {
 757                         /* should be ENOATTR according to the
 758                          * man, but that is undefined here */
 759                         rc = -ENODATA;
 760                         goto out;
 761                 }
 762
 763                 la->la_valid = LA_MODE;
 764                 la->la_mode = S_IFREG | S_IRUGO | S_IWUSR;
 765                 rc = __osd_object_create(env, osd, obj,
 766                                          lu_object_fid(&obj->oo_dt.do_lu),
 767                                          &xa_data_dn, tx, la);
 768                 if (rc)
 769                         goto out;
 770                 xa_data_obj = xa_data_dn->dn_object;
 771
 772                 rc = -sa_handle_get(osd->od_os, xa_data_obj, NULL,
 773                                         SA_HDL_PRIVATE, &sa_hdl);
 774                 if (rc)
 775                         goto out;
 776
 777                 rc = -zap_add(osd->od_os, obj->oo_xattr, name, sizeof(uint64_t),
 778                                 1, &xa_data_obj, tx);
 779                 if (rc)
 780                         goto out_sa;
 781         } else {
 782                 /* There was an error looking up the xattr name */
 783                 goto out;
 784         }
 785
 786         /* Finally write the xattr value */
 787         dmu_write(osd->od_os, xa_data_obj, 0, buf->lb_len, buf->lb_buf, tx);
 788
 789         size = buf->lb_len;
 790         rc = -sa_update(sa_hdl, SA_ZPL_SIZE(osd), &size, 8, tx);
 791
 792 out_sa:
 793         sa_handle_destroy(sa_hdl);
 794 out:
 795         if (xa_data_dn != NULL)
 796                 osd_dnode_rele(xa_data_dn);
 797         if (xa_zap_dn != NULL)
 798                 osd_dnode_rele(xa_zap_dn);
 799
 800         return rc;
 801 }
 802
 803 static int osd_xattr_split_pfid(const struct lu_env *env,
 804                                 struct osd_object *obj, struct osd_thandle *oh)
 805 {
 806         struct osd_thread_info *info = osd_oti_get(env);
 807         struct lustre_ost_attrs *loa =
 808                 (struct lustre_ost_attrs *)&info->oti_buf;
 809         struct lustre_mdt_attrs *lma = &loa->loa_lma;
 810         struct lu_buf buf = {
 811                 .lb_buf = loa,
 812                 .lb_len = sizeof(info->oti_buf),
 813         };
 814         int size;
 815         int rc;
 816         ENTRY;
 817
 818         BUILD_BUG_ON(sizeof(info->oti_buf) < sizeof(*loa));
 819         rc = osd_xattr_get_internal(env, obj, &buf, XATTR_NAME_LMA, &size);
 820         if (rc)
 821                 RETURN(rc);
 822
 823         lustre_loa_swab(loa, true);
 824         LASSERT(lma->lma_compat & LMAC_STRIPE_INFO);
 825
 826         lma->lma_compat &= ~(LMAC_STRIPE_INFO | LMAC_COMP_INFO);
 827         lustre_lma_swab(lma);
 828         buf.lb_buf = lma;
 829         buf.lb_len = sizeof(*lma);
 830         rc = osd_xattr_set_internal(env, obj, &buf, XATTR_NAME_LMA,
 831                                     LU_XATTR_REPLACE, oh);
 832         if (!rc)
 833                 obj->oo_pfid_in_lma = 0;
 834
 835         RETURN(rc);
 836 }
 837
 838 /*
 839  * In DNE environment, the object (in spite of regular file or directory)
 840  * and its name entry may reside on different MDTs. Under such case, we will
 841  * create an agent entry on the MDT where the object resides. The agent entry
 842  * references the object locally, that makes the object to be visible to the
 843  * userspace when mounted as 'zfs' directly. Then the userspace tools, such
 844  * as 'tar' can handle the object properly.
 845  *
 846  * We handle the agent entry during set linkEA that is the common interface
 847  * for both regular file and directroy, can handle kinds of cases, such as
 848  * create/link/unlink/rename, and so on.
 849  *
 850  * NOTE: we need to do that for both directory and regular file, so we can NOT
 851  *       do that when ea_{insert,delete} that are directory based operations.
 852  */
 853 static int osd_xattr_handle_linkea(const struct lu_env *env,
 854                                    struct osd_device *osd,
 855                                    struct osd_object *obj,
 856                                    const struct lu_buf *buf,
 857                                    struct osd_thandle *oh)
 858 {
 859         const struct lu_fid *fid = lu_object_fid(&obj->oo_dt.do_lu);
 860         struct lu_fid *tfid = &osd_oti_get(env)->oti_fid;
 861         struct linkea_data ldata = { .ld_buf = (struct lu_buf *)buf };
 862         struct lu_name tmpname;
 863         int rc;
 864         bool remote = false;
 865         ENTRY;
 866
 867         rc = linkea_init_with_rec(&ldata);
 868         if (!rc) {
 869                 linkea_first_entry(&ldata);
 870                 while (ldata.ld_lee != NULL && !remote) {
 871                         linkea_entry_unpack(ldata.ld_lee, &ldata.ld_reclen,
 872                                             &tmpname, tfid);
 873                         if (osd_remote_fid(env, osd, tfid) > 0)
 874                                 remote = true;
 875                         else
 876                                 linkea_next_entry(&ldata);
 877                 }
 878         } else if (rc == -ENODATA) {
 879                 rc = 0;
 880         } else {
 881                 RETURN(rc);
 882         }
 883
 884         if (lu_object_has_agent_entry(&obj->oo_dt.do_lu) && !remote) {
 885                 rc = osd_delete_from_remote_parent(env, osd, obj, oh, false);
 886                 if (rc)
 887                         CERROR("%s: failed to remove agent entry for "DFID
 888                                ": rc = %d\n", osd_name(osd), PFID(fid), rc);
 889         } else if (!lu_object_has_agent_entry(&obj->oo_dt.do_lu) && remote) {
 890                 rc = osd_add_to_remote_parent(env, osd, obj, oh);
 891                 if (rc)
 892                         CWARN("%s: failed to create agent entry for "DFID
 893                               ": rc = %d\n", osd_name(osd), PFID(fid), rc);
 894         }
 895
 896         RETURN(rc);
 897 }
 898
 899 int osd_xattr_set(const struct lu_env *env, struct dt_object *dt,
 900                   const struct lu_buf *buf, const char *name, int fl,
 901                   struct thandle *handle)
 902 {
 903         struct osd_object *obj = osd_dt_obj(dt);
 904         struct osd_device *osd = osd_obj2dev(obj);
 905         struct osd_thandle *oh;
 906         int rc = 0;
 907         ENTRY;
 908
 909         LASSERT(handle != NULL);
 910         LASSERT(osd_invariant(obj));
 911
 912         if (!osd_obj2dev(obj)->od_posix_acl &&
 913             (strcmp(name, XATTR_NAME_POSIX_ACL_ACCESS) == 0 ||
 914              strcmp(name, XATTR_NAME_POSIX_ACL_DEFAULT) == 0))
 915                 RETURN(-EOPNOTSUPP);
 916
 917         oh = container_of(handle, struct osd_thandle, ot_super);
 918
 919         down_write(&obj->oo_guard);
 920         CDEBUG(D_INODE, "Setting xattr %s with size %d\n",
 921                 name, (int)buf->lb_len);
 922         /* For the OST migrated from ldiskfs, the PFID EA may
 923          * be stored in LMA because of ldiskfs inode size. */
 924         if (unlikely(strcmp(name, XATTR_NAME_FID) == 0 &&
 925                      obj->oo_pfid_in_lma)) {
 926                 rc = osd_xattr_split_pfid(env, obj, oh);
 927                 if (!rc)
 928                         fl = LU_XATTR_CREATE;
 929         } else if (strcmp(name, XATTR_NAME_LINK) == 0 &&
 930                    osd->od_remote_parent_dir != ZFS_NO_OBJECT) {
 931                 rc = osd_xattr_handle_linkea(env, osd, obj, buf, oh);
 932         }
 933
 934         if (!rc)
 935                 rc = osd_xattr_set_internal(env, obj, buf, name, fl, oh);
 936         up_write(&obj->oo_guard);
 937
 938         RETURN(rc);
 939 }
 940
 941 static void
 942 __osd_xattr_declare_del(const struct lu_env *env, struct osd_object *obj,
 943                         const char *name, struct osd_thandle *oh)
 944 {
 945         struct osd_device *osd = osd_obj2dev(obj);
 946         dmu_tx_t          *tx = oh->ot_tx;
 947         uint64_t           xa_data_obj;
 948         int                rc;
 949
 950         /* update SA_ZPL_DXATTR if xattr was in SA */
 951         dmu_tx_hold_sa(tx, obj->oo_sa_hdl, 0);
 952
 953         if (obj->oo_xattr == ZFS_NO_OBJECT)
 954                 return;
 955
 956         rc = -zap_lookup(osd->od_os, obj->oo_xattr, name, 8, 1, &xa_data_obj);
 957         if (rc == 0) {
 958                 /*
 959                  * Entry exists.
 960                  * We'll delete the existing object and ZAP entry.
 961                  */
 962                 dmu_tx_hold_bonus(tx, xa_data_obj);
 963                 dmu_tx_hold_free(tx, xa_data_obj, 0, DMU_OBJECT_END);
 964                 dmu_tx_hold_zap(tx, obj->oo_xattr, FALSE, (char *) name);
 965                 return;
 966         } else if (rc == -ENOENT) {
 967                 /*
 968                  * Entry doesn't exist, nothing to be changed.
 969                  */
 970                 return;
 971         }
 972
 973         /* An error happened */
 974         tx->tx_err = -rc;
 975 }
 976
 977 int osd_declare_xattr_del(const struct lu_env *env, struct dt_object *dt,
 978                           const char *name, struct thandle *handle)
 979 {
 980         struct osd_object  *obj = osd_dt_obj(dt);
 981         struct osd_thandle *oh;
 982         ENTRY;
 983
 984         LASSERT(handle != NULL);
 985         LASSERT(osd_invariant(obj));
 986
 987         oh = container_of(handle, struct osd_thandle, ot_super);
 988         LASSERT(oh->ot_tx != NULL);
 989
 990         down_read(&obj->oo_guard);
 991         if (likely(dt_object_exists(&obj->oo_dt) && !obj->oo_destroyed)) {
 992                 LASSERT(obj->oo_dn != NULL);
 993                 __osd_xattr_declare_del(env, obj, name, oh);
 994         }
 995         up_read(&obj->oo_guard);
 996
 997         RETURN(0);
 998 }
 999
1000 static int __osd_sa_xattr_del(const struct lu_env *env, struct osd_object *obj,
1001                               const char *name, struct osd_thandle *oh)
1002 {
1003         int rc;
1004
1005         rc = __osd_xattr_cache(obj);
1006         if (rc)
1007                 return rc;
1008
1009         rc = -nvlist_remove(obj->oo_sa_xattr, name, DATA_TYPE_BYTE_ARRAY);
1010         if (rc)
1011                 return rc;
1012
1013         /*
1014          * only migrate delete LMV, and it needs to be done immediately, because
1015          * it's used in deleting sub stripes, and if this is delayed, later when
1016          * destroying the master object, it will delete sub stripes again.
1017          */
1018         if (!strcmp(name, XATTR_NAME_LMV))
1019                 rc = __osd_sa_xattr_update(env, obj, oh);
1020         else
1021                 rc = __osd_sa_xattr_schedule_update(env, obj, oh);
1022         return rc;
1023 }
1024
1025 static int __osd_xattr_del(const struct lu_env *env, struct osd_object *obj,
1026                            const char *name, struct osd_thandle *oh)
1027 {
1028         struct osd_device *osd = osd_obj2dev(obj);
1029         uint64_t           xa_data_obj;
1030         int                rc;
1031
1032         if (unlikely(!dt_object_exists(&obj->oo_dt) || obj->oo_destroyed))
1033                 return -ENOENT;
1034
1035         /* try remove xattr from SA at first */
1036         rc = __osd_sa_xattr_del(env, obj, name, oh);
1037         if (rc != -ENOENT)
1038                 return rc;
1039
1040         if (obj->oo_xattr == ZFS_NO_OBJECT)
1041                 return 0;
1042
1043         rc = -zap_lookup(osd->od_os, obj->oo_xattr, name, sizeof(uint64_t), 1,
1044                         &xa_data_obj);
1045         if (rc == -ENOENT) {
1046                 rc = 0;
1047         } else if (rc == 0) {
1048                 /*
1049                  * Entry exists.
1050                  * We'll delete the existing object and ZAP entry.
1051                  */
1052                 rc = -dmu_object_free(osd->od_os, xa_data_obj, oh->ot_tx);
1053                 if (rc)
1054                         return rc;
1055
1056                 rc = -zap_remove(osd->od_os, obj->oo_xattr, name, oh->ot_tx);
1057         }
1058
1059         return rc;
1060 }
1061
1062 int osd_xattr_del(const struct lu_env *env, struct dt_object *dt,
1063                   const char *name, struct thandle *handle)
1064 {
1065         struct osd_object  *obj = osd_dt_obj(dt);
1066         struct osd_thandle *oh;
1067         int                 rc;
1068         ENTRY;
1069
1070         LASSERT(handle != NULL);
1071         oh = container_of(handle, struct osd_thandle, ot_super);
1072         LASSERT(oh->ot_tx != NULL);
1073
1074         if (!osd_obj2dev(obj)->od_posix_acl &&
1075             (strcmp(name, XATTR_NAME_POSIX_ACL_ACCESS) == 0 ||
1076              strcmp(name, XATTR_NAME_POSIX_ACL_DEFAULT) == 0))
1077                 RETURN(-EOPNOTSUPP);
1078
1079         down_write(&obj->oo_guard);
1080         if (unlikely(!dt_object_exists(dt) || obj->oo_destroyed))
1081                 GOTO(out, rc = -ENOENT);
1082         LASSERT(obj->oo_dn != NULL);
1083         /* For the OST migrated from ldiskfs, the PFID EA may
1084          * be stored in LMA because of ldiskfs inode size. */
1085         if (unlikely(strcmp(name, XATTR_NAME_FID) == 0 && obj->oo_pfid_in_lma))
1086                 rc = osd_xattr_split_pfid(env, obj, oh);
1087         else
1088                 rc = __osd_xattr_del(env, obj, name, oh);
1089
1090 out:
1091         up_write(&obj->oo_guard);
1092         RETURN(rc);
1093 }
1094
1095 void osd_declare_xattrs_destroy(const struct lu_env *env,
1096                                 struct osd_object *obj, struct osd_thandle *oh)
1097 {
1098         struct osd_device *osd = osd_obj2dev(obj);
1099         zap_attribute_t   *za = &osd_oti_get(env)->oti_za;
1100         uint64_t           oid = obj->oo_xattr, xid;
1101         dmu_tx_t          *tx = oh->ot_tx;
1102         zap_cursor_t      *zc;
1103         int                rc;
1104
1105         if (oid == ZFS_NO_OBJECT)
1106                 return; /* Nothing to do for SA xattrs */
1107
1108         /* Declare to free the ZAP holding xattrs */
1109         dmu_tx_hold_free(tx, oid, 0, DMU_OBJECT_END);
1110
1111         rc = osd_zap_cursor_init(&zc, osd->od_os, oid, 0);
1112         if (rc)
1113                 goto out;
1114
1115         while (zap_cursor_retrieve(zc, za) == 0) {
1116                 LASSERT(za->za_num_integers == 1);
1117                 LASSERT(za->za_integer_length == sizeof(uint64_t));
1118
1119                 rc = -zap_lookup(osd->od_os, oid, za->za_name,
1120                                  sizeof(uint64_t), 1, &xid);
1121                 if (rc) {
1122                         CERROR("%s: xattr %s lookup failed: rc = %d\n",
1123                                osd->od_svname, za->za_name, rc);
1124                         break;
1125                 }
1126                 dmu_tx_hold_free(tx, xid, 0, DMU_OBJECT_END);
1127
1128                 zap_cursor_advance(zc);
1129         }
1130
1131         osd_zap_cursor_fini(zc);
1132 out:
1133         if (rc && tx->tx_err == 0)
1134                 tx->tx_err = -rc;
1135 }
1136
1137 int osd_xattrs_destroy(const struct lu_env *env,
1138                        struct osd_object *obj, struct osd_thandle *oh)
1139 {
1140         struct osd_device *osd = osd_obj2dev(obj);
1141         dmu_tx_t          *tx = oh->ot_tx;
1142         zap_attribute_t   *za = &osd_oti_get(env)->oti_za;
1143         zap_cursor_t      *zc;
1144         uint64_t           xid;
1145         int                rc;
1146
1147         /* The transaction must have been assigned to a transaction group. */
1148         LASSERT(tx->tx_txg != 0);
1149
1150         if (obj->oo_xattr == ZFS_NO_OBJECT)
1151                 return 0; /* Nothing to do for SA xattrs */
1152
1153         /* Free the ZAP holding the xattrs */
1154         rc = osd_zap_cursor_init(&zc, osd->od_os, obj->oo_xattr, 0);
1155         if (rc)
1156                 return rc;
1157
1158         while (zap_cursor_retrieve(zc, za) == 0) {
1159                 LASSERT(za->za_num_integers == 1);
1160                 LASSERT(za->za_integer_length == sizeof(uint64_t));
1161
1162                 rc = -zap_lookup(osd->od_os, obj->oo_xattr, za->za_name,
1163                                  sizeof(uint64_t), 1, &xid);
1164                 if (rc) {
1165                         CERROR("%s: lookup xattr %s failed: rc = %d\n",
1166                                osd->od_svname, za->za_name, rc);
1167                 } else {
1168                         rc = -dmu_object_free(osd->od_os, xid, tx);
1169                         if (rc)
1170                                 CERROR("%s: free xattr %s failed: rc = %d\n",
1171                                        osd->od_svname, za->za_name, rc);
1172                 }
1173                 zap_cursor_advance(zc);
1174         }
1175         osd_zap_cursor_fini(zc);
1176
1177         rc = -dmu_object_free(osd->od_os, obj->oo_xattr, tx);
1178         if (rc)
1179                 CERROR("%s: free xattr %llu failed: rc = %d\n",
1180                        osd->od_svname, obj->oo_xattr, rc);
1181
1182         return rc;
1183 }
1184
1185 static int
1186 osd_sa_xattr_list(const struct lu_env *env, struct osd_object *obj,
1187                   const struct lu_buf *lb)
1188 {
1189         nvpair_t *nvp = NULL;
1190         int       len, counted = 0;
1191         int       rc = 0;
1192
1193         rc = __osd_xattr_cache(obj);
1194         if (rc)
1195                 return rc;
1196
1197         while ((nvp = nvlist_next_nvpair(obj->oo_sa_xattr, nvp)) != NULL) {
1198                 const char *name = nvpair_name(nvp);
1199
1200                 if (!osd_obj2dev(obj)->od_posix_acl &&
1201                     (strcmp(name, XATTR_NAME_POSIX_ACL_ACCESS) == 0 ||
1202                      strcmp(name, XATTR_NAME_POSIX_ACL_DEFAULT) == 0))
1203                         continue;
1204
1205                 len = strlen(name);
1206                 if (lb->lb_buf != NULL) {
1207                         if (counted + len + 1 > lb->lb_len)
1208                                 return -ERANGE;
1209
1210                         memcpy(lb->lb_buf + counted, name, len + 1);
1211                 }
1212                 counted += len + 1;
1213         }
1214         return counted;
1215 }
1216
1217 int osd_xattr_list(const struct lu_env *env, struct dt_object *dt,
1218                    const struct lu_buf *lb)
1219 {
1220         struct osd_object      *obj = osd_dt_obj(dt);
1221         struct osd_device      *osd = osd_obj2dev(obj);
1222         zap_attribute_t        *za = &osd_oti_get(env)->oti_za;
1223         zap_cursor_t           *zc;
1224         int                    rc, counted;
1225         ENTRY;
1226
1227         down_read(&obj->oo_guard);
1228         if (unlikely(!dt_object_exists(dt) || obj->oo_destroyed))
1229                 GOTO(out, rc = -ENOENT);
1230         LASSERT(obj->oo_dn != NULL);
1231
1232         rc = osd_sa_xattr_list(env, obj, lb);
1233         if (rc < 0)
1234                 GOTO(out, rc);
1235
1236         counted = rc;
1237
1238         /* continue with dnode xattr if any */
1239         if (obj->oo_xattr == ZFS_NO_OBJECT)
1240                 GOTO(out, rc = counted);
1241
1242         rc = osd_zap_cursor_init(&zc, osd->od_os, obj->oo_xattr, 0);
1243         if (rc)
1244                 GOTO(out, rc);
1245
1246         while ((rc = -zap_cursor_retrieve(zc, za)) == 0) {
1247                 if (!osd_obj2dev(obj)->od_posix_acl &&
1248                     (strcmp(za->za_name, XATTR_NAME_POSIX_ACL_ACCESS) == 0 ||
1249                      strcmp(za->za_name, XATTR_NAME_POSIX_ACL_DEFAULT) == 0)) {
1250                         zap_cursor_advance(zc);
1251                         continue;
1252                 }
1253
1254                 rc = strlen(za->za_name);
1255                 if (lb->lb_buf != NULL) {
1256                         if (counted + rc + 1 > lb->lb_len)
1257                                 GOTO(out_fini, rc = -ERANGE);
1258
1259                         memcpy(lb->lb_buf + counted, za->za_name, rc + 1);
1260                 }
1261                 counted += rc + 1;
1262
1263                 zap_cursor_advance(zc);
1264         }
1265         if (rc == -ENOENT) /* no more kes in the index */
1266                 rc = 0;
1267         else if (unlikely(rc < 0))
1268                 GOTO(out_fini, rc);
1269         rc = counted;
1270
1271 out_fini:
1272         osd_zap_cursor_fini(zc);
1273 out:
1274         up_read(&obj->oo_guard);
1275         RETURN(rc);
1276
1277 }