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 = osd_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         osd_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         LASSERT(obj->oo_dn != NULL);
 326         LASSERT(osd_invariant(obj));
 327
 328         if (!osd_obj2dev(obj)->od_posix_acl &&
 329             (strcmp(name, XATTR_NAME_POSIX_ACL_ACCESS) == 0 ||
 330              strcmp(name, XATTR_NAME_POSIX_ACL_DEFAULT) == 0))
 331                 RETURN(-EOPNOTSUPP);
 332
 333         down_read(&obj->oo_guard);
 334         if (unlikely(!dt_object_exists(dt) || obj->oo_destroyed)) {
 335                 up_read(&obj->oo_guard);
 336                 RETURN(-ENOENT);
 337         }
 338
 339         /* For the OST migrated from ldiskfs, the PFID EA may
 340          * be stored in LMA because of ldiskfs inode size. */
 341         if (strcmp(name, XATTR_NAME_FID) == 0 && obj->oo_pfid_in_lma)
 342                 rc = osd_get_pfid_from_lma(env, obj, buf, &size);
 343         else
 344                 rc = osd_xattr_get_internal(env, obj, buf, name, &size);
 345         up_read(&obj->oo_guard);
 346
 347         if (rc == -ENOENT)
 348                 rc = -ENODATA;
 349         else if (rc == 0)
 350                 rc = size;
 351         RETURN(rc);
 352 }
 353
 354 /* the function is used to declare EAs when SA is not supported */
 355 void __osd_xattr_declare_legacy(const struct lu_env *env,
 356                                 struct osd_object *obj,
 357                                 int vallen, const char *name,
 358                                 struct osd_thandle *oh)
 359 {
 360         struct osd_device *osd = osd_obj2dev(obj);
 361         dmu_tx_t *tx = oh->ot_tx;
 362         uint64_t xa_data_obj;
 363         int rc;
 364
 365         if (obj->oo_xattr == ZFS_NO_OBJECT) {
 366                 /* xattr zap + entry */
 367                 dmu_tx_hold_zap(tx, DMU_NEW_OBJECT, TRUE, (char *) name);
 368                 /* xattr value obj */
 369                 dmu_tx_hold_sa_create(tx, ZFS_SA_BASE_ATTR_SIZE);
 370                 dmu_tx_hold_write(tx, DMU_NEW_OBJECT, 0, vallen);
 371                 return;
 372         }
 373
 374         rc = -zap_lookup(osd->od_os, obj->oo_xattr, name, sizeof(uint64_t), 1,
 375                         &xa_data_obj);
 376         if (rc == 0) {
 377                 /*
 378                  * Entry already exists.
 379                  * We'll truncate the existing object.
 380                  */
 381                 dmu_tx_hold_bonus(tx, xa_data_obj);
 382                 dmu_tx_hold_free(tx, xa_data_obj, vallen, DMU_OBJECT_END);
 383                 dmu_tx_hold_write(tx, xa_data_obj, 0, vallen);
 384         } else if (rc == -ENOENT) {
 385                 /*
 386                  * Entry doesn't exist, we need to create a new one and a new
 387                  * object to store the value.
 388                  */
 389                 dmu_tx_hold_bonus(tx, obj->oo_xattr);
 390                 dmu_tx_hold_zap(tx, obj->oo_xattr, TRUE, (char *) name);
 391                 dmu_tx_hold_sa_create(tx, ZFS_SA_BASE_ATTR_SIZE);
 392                 dmu_tx_hold_write(tx, DMU_NEW_OBJECT, 0, vallen);
 393         }
 394 }
 395
 396 void __osd_xattr_declare_set(const struct lu_env *env, struct osd_object *obj,
 397                              int vallen, const char *name,
 398                              struct osd_thandle *oh)
 399 {
 400         struct osd_device *osd = osd_obj2dev(obj);
 401         dmu_tx_t *tx = oh->ot_tx;
 402         int bonuslen;
 403
 404         if (unlikely(obj->oo_destroyed))
 405                 return;
 406
 407         if (strcmp(name, XATTR_NAME_LINK) == 0 &&
 408             osd->od_remote_parent_dir != ZFS_NO_OBJECT) {
 409                 /* If some name entry resides on remote MDT, then will create
 410                  * agent entry under remote parent. On the other hand, if the
 411                  * remote entry will be removed, then related agent entry may
 412                  * need to be removed from the remote parent. So there may be
 413                  * kinds of cases, let's declare enough credits. The credits
 414                  * for create agent entry is enough for remove case. */
 415                 osd_tx_hold_zap(tx, osd->od_remote_parent_dir,
 416                                 NULL, TRUE, NULL);
 417         }
 418
 419         if (unlikely(!osd_obj2dev(obj)->od_xattr_in_sa)) {
 420                 __osd_xattr_declare_legacy(env, obj, vallen, name, oh);
 421                 return;
 422         }
 423
 424         /* declare EA in SA */
 425         if (dt_object_exists(&obj->oo_dt)) {
 426                 LASSERT(obj->oo_sa_hdl);
 427                 /* XXX: it should be possible to skip spill
 428                  * declaration if specific EA is part of
 429                  * bonus and doesn't grow */
 430                 dmu_tx_hold_spill(tx, obj->oo_dn->dn_object);
 431                 return;
 432         }
 433
 434         bonuslen = osd_obj_bonuslen(obj);
 435
 436         /* the object doesn't exist, but we've declared bonus
 437          * in osd_declare_object_create() yet */
 438         if (obj->oo_ea_in_bonus > bonuslen) {
 439                 /* spill has been declared already */
 440         } else if (obj->oo_ea_in_bonus + vallen > bonuslen) {
 441                 /* we're about to exceed bonus, let's declare spill */
 442                 dmu_tx_hold_spill(tx, DMU_NEW_OBJECT);
 443         }
 444         obj->oo_ea_in_bonus += vallen;
 445 }
 446
 447 int osd_declare_xattr_set(const struct lu_env *env, struct dt_object *dt,
 448                           const struct lu_buf *buf, const char *name,
 449                           int fl, struct thandle *handle)
 450 {
 451         struct osd_object  *obj = osd_dt_obj(dt);
 452         struct osd_thandle *oh;
 453         ENTRY;
 454
 455         LASSERT(handle != NULL);
 456         oh = container_of(handle, struct osd_thandle, ot_super);
 457
 458         down_read(&obj->oo_guard);
 459         __osd_xattr_declare_set(env, obj, buf->lb_len, name, oh);
 460         up_read(&obj->oo_guard);
 461
 462         RETURN(0);
 463 }
 464
 465 int __osd_sa_attr_init(const struct lu_env *env, struct osd_object *obj,
 466                        struct osd_thandle *oh)
 467 {
 468         sa_bulk_attr_t *bulk = osd_oti_get(env)->oti_attr_bulk;
 469         struct osa_attr *osa = &osd_oti_get(env)->oti_osa;
 470         struct lu_buf *lb = &osd_oti_get(env)->oti_xattr_lbuf;
 471         struct osd_device *osd = osd_obj2dev(obj);
 472         uint64_t gen;
 473         inode_timespec_t now;
 474         size_t size;
 475         int rc, cnt;
 476
 477         obj->oo_late_xattr = 0;
 478         obj->oo_late_attr_set = 0;
 479
 480         gen = dmu_tx_get_txg(oh->ot_tx);
 481         gethrestime(&now);
 482         ZFS_TIME_ENCODE(&now, osa->btime);
 483
 484         obj->oo_attr.la_valid |= LA_BTIME;
 485         obj->oo_attr.la_btime = osa->btime[0];
 486         osa->atime[0] = obj->oo_attr.la_atime;
 487         osa->ctime[0] = obj->oo_attr.la_ctime;
 488         osa->mtime[0] = obj->oo_attr.la_mtime;
 489         osa->mode = obj->oo_attr.la_mode;
 490         osa->uid = obj->oo_attr.la_uid;
 491         osa->gid = obj->oo_attr.la_gid;
 492         osa->rdev = obj->oo_attr.la_rdev;
 493         osa->nlink = obj->oo_attr.la_nlink;
 494         osa->flags = attrs_fs2zfs(obj->oo_attr.la_flags);
 495         osa->size  = obj->oo_attr.la_size;
 496 #ifdef ZFS_PROJINHERIT
 497         if (osd->od_projectused_dn) {
 498                 if (obj->oo_attr.la_valid & LA_PROJID)
 499                         osa->projid = obj->oo_attr.la_projid;
 500                 else
 501                         osa->projid = ZFS_DEFAULT_PROJID;
 502                 osa->flags |= ZFS_PROJID;
 503                 obj->oo_with_projid = 1;
 504         }
 505 #endif
 506
 507         cnt = 0;
 508         SA_ADD_BULK_ATTR(bulk, cnt, SA_ZPL_MODE(osd), NULL, &osa->mode, 8);
 509         SA_ADD_BULK_ATTR(bulk, cnt, SA_ZPL_SIZE(osd), NULL, &osa->size, 8);
 510         SA_ADD_BULK_ATTR(bulk, cnt, SA_ZPL_GEN(osd), NULL, &gen, 8);
 511         SA_ADD_BULK_ATTR(bulk, cnt, SA_ZPL_UID(osd), NULL, &osa->uid, 8);
 512         SA_ADD_BULK_ATTR(bulk, cnt, SA_ZPL_GID(osd), NULL, &osa->gid, 8);
 513         SA_ADD_BULK_ATTR(bulk, cnt, SA_ZPL_PARENT(osd), NULL,
 514                          &obj->oo_parent, 8);
 515         SA_ADD_BULK_ATTR(bulk, cnt, SA_ZPL_FLAGS(osd), NULL, &osa->flags, 8);
 516         SA_ADD_BULK_ATTR(bulk, cnt, SA_ZPL_ATIME(osd), NULL, osa->atime, 16);
 517         SA_ADD_BULK_ATTR(bulk, cnt, SA_ZPL_MTIME(osd), NULL, osa->mtime, 16);
 518         SA_ADD_BULK_ATTR(bulk, cnt, SA_ZPL_CTIME(osd), NULL, osa->ctime, 16);
 519         SA_ADD_BULK_ATTR(bulk, cnt, SA_ZPL_CRTIME(osd), NULL, osa->btime, 16);
 520         SA_ADD_BULK_ATTR(bulk, cnt, SA_ZPL_LINKS(osd), NULL, &osa->nlink, 8);
 521 #ifdef ZFS_PROJINHERIT
 522         if (osd->od_projectused_dn)
 523                 SA_ADD_BULK_ATTR(bulk, cnt, SA_ZPL_PROJID(osd), NULL,
 524                                  &osa->projid, 8);
 525 #endif
 526         SA_ADD_BULK_ATTR(bulk, cnt, SA_ZPL_RDEV(osd), NULL, &osa->rdev, 8);
 527         LASSERT(cnt <= ARRAY_SIZE(osd_oti_get(env)->oti_attr_bulk));
 528
 529         /* Update the SA for additions, modifications, and removals. */
 530         rc = -nvlist_size(obj->oo_sa_xattr, &size, NV_ENCODE_XDR);
 531         if (rc)
 532                 return rc;
 533
 534         lu_buf_check_and_alloc(lb, size);
 535         if (lb->lb_buf == NULL) {
 536                 CERROR("%s: can't allocate buffer for xattr update\n",
 537                                 osd->od_svname);
 538                 return -ENOMEM;
 539         }
 540
 541         rc = -nvlist_pack(obj->oo_sa_xattr, (char **)&lb->lb_buf, &size,
 542                           NV_ENCODE_XDR, KM_SLEEP);
 543         if (rc)
 544                 return rc;
 545
 546         SA_ADD_BULK_ATTR(bulk, cnt, SA_ZPL_DXATTR(osd), NULL, lb->lb_buf, size);
 547
 548         rc = -sa_replace_all_by_template(obj->oo_sa_hdl, bulk, cnt, oh->ot_tx);
 549
 550         return rc;
 551 }
 552
 553 int __osd_sa_xattr_update(const struct lu_env *env, struct osd_object *obj,
 554                            struct osd_thandle *oh)
 555 {
 556         struct lu_buf     *lb = &osd_oti_get(env)->oti_xattr_lbuf;
 557         struct osd_device *osd = osd_obj2dev(obj);
 558         char              *dxattr;
 559         size_t             size;
 560         int                rc;
 561
 562         obj->oo_late_xattr = 0;
 563
 564         /* Update the SA for additions, modifications, and removals. */
 565         rc = -nvlist_size(obj->oo_sa_xattr, &size, NV_ENCODE_XDR);
 566         if (rc)
 567                 return rc;
 568
 569         lu_buf_check_and_alloc(lb, size);
 570         if (lb->lb_buf == NULL) {
 571                 CERROR("%s: can't allocate buffer for xattr update\n",
 572                                 osd->od_svname);
 573                 return -ENOMEM;
 574         }
 575
 576         dxattr = lb->lb_buf;
 577         rc = -nvlist_pack(obj->oo_sa_xattr, &dxattr, &size,
 578                         NV_ENCODE_XDR, KM_SLEEP);
 579         if (rc)
 580                 return rc;
 581         LASSERT(dxattr == lb->lb_buf);
 582
 583         sa_update(obj->oo_sa_hdl, SA_ZPL_DXATTR(osd), dxattr, size, oh->ot_tx);
 584
 585         return 0;
 586 }
 587
 588 /*
 589  * Set an extended attribute.
 590  * This transaction must have called udmu_xattr_declare_set() first.
 591  *
 592  * Returns 0 on success or a negative error number on failure.
 593  *
 594  * No locking is done here.
 595  */
 596 int __osd_sa_xattr_schedule_update(const struct lu_env *env,
 597                                    struct osd_object *obj,
 598                                    struct osd_thandle *oh)
 599 {
 600         ENTRY;
 601         LASSERT(obj->oo_sa_hdl);
 602         LASSERT(obj->oo_sa_xattr);
 603
 604         /* schedule batched SA update in osd_object_sa_dirty_rele() */
 605         obj->oo_late_xattr = 1;
 606         osd_object_sa_dirty_add(obj, oh);
 607
 608         RETURN(0);
 609
 610 }
 611
 612 int __osd_sa_xattr_set(const struct lu_env *env, struct osd_object *obj,
 613                        const struct lu_buf *buf, const char *name, int fl,
 614                        struct osd_thandle *oh)
 615 {
 616         uchar_t *nv_value;
 617         size_t  size;
 618         int     nv_size;
 619         int     rc;
 620         int     too_big = 0;
 621
 622         rc = __osd_xattr_cache(obj);
 623         if (rc)
 624                 return rc;
 625
 626         LASSERT(obj->oo_sa_xattr);
 627         if (buf->lb_len > OBD_MAX_EA_SIZE) {
 628                 too_big = 1;
 629         } else {
 630                 /* Prevent the DXATTR SA from consuming the entire SA
 631                  * region */
 632                 rc = -nvlist_size(obj->oo_sa_xattr, &size, NV_ENCODE_XDR);
 633                 if (rc)
 634                         return rc;
 635
 636                 if (size + buf->lb_len > DXATTR_MAX_SA_SIZE)
 637                         too_big = 1;
 638         }
 639
 640         /* even in case of -EFBIG we must lookup xattr and check can we
 641          * rewrite it then delete from SA */
 642         rc = -nvlist_lookup_byte_array(obj->oo_sa_xattr, name, &nv_value,
 643                                         &nv_size);
 644         if (rc == 0) {
 645                 if (fl & LU_XATTR_CREATE) {
 646                         return -EEXIST;
 647                 } else if (too_big) {
 648                         rc = -nvlist_remove(obj->oo_sa_xattr, name,
 649                                                 DATA_TYPE_BYTE_ARRAY);
 650                         if (rc < 0)
 651                                 return rc;
 652                         rc = __osd_sa_xattr_schedule_update(env, obj, oh);
 653                         return rc == 0 ? -EFBIG : rc;
 654                 }
 655         } else if (rc == -ENOENT) {
 656                 if (fl & LU_XATTR_REPLACE)
 657                         return -ENODATA;
 658                 else if (too_big)
 659                         return -EFBIG;
 660         } else {
 661                 return rc;
 662         }
 663
 664         /* Ensure xattr doesn't exist in ZAP */
 665         if (obj->oo_xattr != ZFS_NO_OBJECT) {
 666                 struct osd_device *osd = osd_obj2dev(obj);
 667                 uint64_t           objid;
 668                 rc = -zap_lookup(osd->od_os, obj->oo_xattr,
 669                                  name, 8, 1, &objid);
 670                 if (rc == 0) {
 671                         rc = -dmu_object_free(osd->od_os, objid, oh->ot_tx);
 672                         if (rc == 0)
 673                                 zap_remove(osd->od_os, obj->oo_xattr,
 674                                            name, oh->ot_tx);
 675                 }
 676         }
 677
 678         rc = -nvlist_add_byte_array(obj->oo_sa_xattr, name,
 679                                     (uchar_t *)buf->lb_buf, buf->lb_len);
 680         if (rc)
 681                 return rc;
 682
 683         /* batch updates only for just created dnodes where we
 684          * used to set number of EAs in a single transaction */
 685         if (obj->oo_dn->dn_allocated_txg == oh->ot_tx->tx_txg)
 686                 rc = __osd_sa_xattr_schedule_update(env, obj, oh);
 687         else
 688                 rc = __osd_sa_xattr_update(env, obj, oh);
 689
 690         return rc;
 691 }
 692
 693 int
 694 __osd_xattr_set(const struct lu_env *env, struct osd_object *obj,
 695                 const struct lu_buf *buf, const char *name, int fl,
 696                 struct osd_thandle *oh)
 697 {
 698         struct osd_device *osd = osd_obj2dev(obj);
 699         dnode_t *xa_zap_dn = NULL;
 700         dnode_t *xa_data_dn = NULL;
 701         uint64_t           xa_data_obj;
 702         sa_handle_t       *sa_hdl = NULL;
 703         dmu_tx_t          *tx = oh->ot_tx;
 704         uint64_t           size;
 705         int                rc;
 706
 707         LASSERT(obj->oo_sa_hdl);
 708
 709         if (obj->oo_xattr == ZFS_NO_OBJECT) {
 710                 struct lu_attr *la = &osd_oti_get(env)->oti_la;
 711
 712                 la->la_valid = LA_MODE;
 713                 la->la_mode = S_IFDIR | S_IRUGO | S_IWUSR | S_IXUGO;
 714                 rc = __osd_zap_create(env, osd, &xa_zap_dn, tx, la, 0, 0);
 715                 if (rc)
 716                         return rc;
 717
 718                 obj->oo_xattr = xa_zap_dn->dn_object;
 719                 rc = osd_object_sa_update(obj, SA_ZPL_XATTR(osd),
 720                                 &obj->oo_xattr, 8, oh);
 721                 if (rc)
 722                         goto out;
 723         }
 724
 725         rc = -zap_lookup(osd->od_os, obj->oo_xattr, name, sizeof(uint64_t), 1,
 726                          &xa_data_obj);
 727         if (rc == 0) {
 728                 if (fl & LU_XATTR_CREATE) {
 729                         rc = -EEXIST;
 730                         goto out;
 731                 }
 732                 /*
 733                  * Entry already exists.
 734                  * We'll truncate the existing object.
 735                  */
 736                 rc = __osd_obj2dnode(osd->od_os, xa_data_obj, &xa_data_dn);
 737                 if (rc)
 738                         goto out;
 739
 740                 rc = -sa_handle_get(osd->od_os, xa_data_obj, NULL,
 741                                         SA_HDL_PRIVATE, &sa_hdl);
 742                 if (rc)
 743                         goto out;
 744
 745                 rc = -sa_lookup(sa_hdl, SA_ZPL_SIZE(osd), &size, 8);
 746                 if (rc)
 747                         goto out_sa;
 748
 749                 rc = -dmu_free_range(osd->od_os, xa_data_dn->dn_object,
 750                                      0, DMU_OBJECT_END, tx);
 751                 if (rc)
 752                         goto out_sa;
 753         } else if (rc == -ENOENT) {
 754                 struct lu_attr *la = &osd_oti_get(env)->oti_la;
 755                 /*
 756                  * Entry doesn't exist, we need to create a new one and a new
 757                  * object to store the value.
 758                  */
 759                 if (fl & LU_XATTR_REPLACE) {
 760                         /* should be ENOATTR according to the
 761                          * man, but that is undefined here */
 762                         rc = -ENODATA;
 763                         goto out;
 764                 }
 765
 766                 la->la_valid = LA_MODE;
 767                 la->la_mode = S_IFREG | S_IRUGO | S_IWUSR;
 768                 rc = __osd_object_create(env, osd, obj,
 769                                          lu_object_fid(&obj->oo_dt.do_lu),
 770                                          &xa_data_dn, tx, la);
 771                 if (rc)
 772                         goto out;
 773                 xa_data_obj = xa_data_dn->dn_object;
 774
 775                 rc = -sa_handle_get(osd->od_os, xa_data_obj, NULL,
 776                                         SA_HDL_PRIVATE, &sa_hdl);
 777                 if (rc)
 778                         goto out;
 779
 780                 rc = -zap_add(osd->od_os, obj->oo_xattr, name, sizeof(uint64_t),
 781                                 1, &xa_data_obj, tx);
 782                 if (rc)
 783                         goto out_sa;
 784         } else {
 785                 /* There was an error looking up the xattr name */
 786                 goto out;
 787         }
 788
 789         /* Finally write the xattr value */
 790         dmu_write(osd->od_os, xa_data_obj, 0, buf->lb_len, buf->lb_buf, tx);
 791
 792         size = buf->lb_len;
 793         rc = -sa_update(sa_hdl, SA_ZPL_SIZE(osd), &size, 8, tx);
 794
 795 out_sa:
 796         sa_handle_destroy(sa_hdl);
 797 out:
 798         if (xa_data_dn != NULL)
 799                 osd_dnode_rele(xa_data_dn);
 800         if (xa_zap_dn != NULL)
 801                 osd_dnode_rele(xa_zap_dn);
 802
 803         return rc;
 804 }
 805
 806 static int osd_xattr_split_pfid(const struct lu_env *env,
 807                                 struct osd_object *obj, struct osd_thandle *oh)
 808 {
 809         struct osd_thread_info *info = osd_oti_get(env);
 810         struct lustre_ost_attrs *loa =
 811                 (struct lustre_ost_attrs *)&info->oti_buf;
 812         struct lustre_mdt_attrs *lma = &loa->loa_lma;
 813         struct lu_buf buf = {
 814                 .lb_buf = loa,
 815                 .lb_len = sizeof(info->oti_buf),
 816         };
 817         int size;
 818         int rc;
 819         ENTRY;
 820
 821         BUILD_BUG_ON(sizeof(info->oti_buf) < sizeof(*loa));
 822         rc = osd_xattr_get_internal(env, obj, &buf, XATTR_NAME_LMA, &size);
 823         if (rc)
 824                 RETURN(rc);
 825
 826         lustre_loa_swab(loa, true);
 827         LASSERT(lma->lma_compat & LMAC_STRIPE_INFO);
 828
 829         lma->lma_compat &= ~(LMAC_STRIPE_INFO | LMAC_COMP_INFO);
 830         lustre_lma_swab(lma);
 831         buf.lb_buf = lma;
 832         buf.lb_len = sizeof(*lma);
 833         rc = osd_xattr_set_internal(env, obj, &buf, XATTR_NAME_LMA,
 834                                     LU_XATTR_REPLACE, oh);
 835         if (!rc)
 836                 obj->oo_pfid_in_lma = 0;
 837
 838         RETURN(rc);
 839 }
 840
 841 /*
 842  * In DNE environment, the object (in spite of regular file or directory)
 843  * and its name entry may reside on different MDTs. Under such case, we will
 844  * create an agent entry on the MDT where the object resides. The agent entry
 845  * references the object locally, that makes the object to be visible to the
 846  * userspace when mounted as 'zfs' directly. Then the userspace tools, such
 847  * as 'tar' can handle the object properly.
 848  *
 849  * We handle the agent entry during set linkEA that is the common interface
 850  * for both regular file and directroy, can handle kinds of cases, such as
 851  * create/link/unlink/rename, and so on.
 852  *
 853  * NOTE: we need to do that for both directory and regular file, so we can NOT
 854  *       do that when ea_{insert,delete} that are directory based operations.
 855  */
 856 static int osd_xattr_handle_linkea(const struct lu_env *env,
 857                                    struct osd_device *osd,
 858                                    struct osd_object *obj,
 859                                    const struct lu_buf *buf,
 860                                    struct osd_thandle *oh)
 861 {
 862         const struct lu_fid *fid = lu_object_fid(&obj->oo_dt.do_lu);
 863         struct lu_fid *tfid = &osd_oti_get(env)->oti_fid;
 864         struct linkea_data ldata = { .ld_buf = (struct lu_buf *)buf };
 865         struct lu_name tmpname;
 866         int rc;
 867         bool remote = false;
 868         ENTRY;
 869
 870         rc = linkea_init_with_rec(&ldata);
 871         if (!rc) {
 872                 linkea_first_entry(&ldata);
 873                 while (ldata.ld_lee != NULL && !remote) {
 874                         linkea_entry_unpack(ldata.ld_lee, &ldata.ld_reclen,
 875                                             &tmpname, tfid);
 876                         if (osd_remote_fid(env, osd, tfid) > 0)
 877                                 remote = true;
 878                         else
 879                                 linkea_next_entry(&ldata);
 880                 }
 881         } else if (rc == -ENODATA) {
 882                 rc = 0;
 883         } else {
 884                 RETURN(rc);
 885         }
 886
 887         if (lu_object_has_agent_entry(&obj->oo_dt.do_lu) && !remote) {
 888                 rc = osd_delete_from_remote_parent(env, osd, obj, oh, false);
 889                 if (rc)
 890                         CERROR("%s: failed to remove agent entry for "DFID
 891                                ": rc = %d\n", osd_name(osd), PFID(fid), rc);
 892         } else if (!lu_object_has_agent_entry(&obj->oo_dt.do_lu) && remote) {
 893                 rc = osd_add_to_remote_parent(env, osd, obj, oh);
 894                 if (rc)
 895                         CWARN("%s: failed to create agent entry for "DFID
 896                               ": rc = %d\n", osd_name(osd), PFID(fid), rc);
 897         }
 898
 899         RETURN(rc);
 900 }
 901
 902 int osd_xattr_set(const struct lu_env *env, struct dt_object *dt,
 903                   const struct lu_buf *buf, const char *name, int fl,
 904                   struct thandle *handle)
 905 {
 906         struct osd_object *obj = osd_dt_obj(dt);
 907         struct osd_device *osd = osd_obj2dev(obj);
 908         struct osd_thandle *oh;
 909         int rc = 0;
 910         ENTRY;
 911
 912         LASSERT(handle != NULL);
 913         LASSERT(osd_invariant(obj));
 914
 915         if (!osd_obj2dev(obj)->od_posix_acl &&
 916             (strcmp(name, XATTR_NAME_POSIX_ACL_ACCESS) == 0 ||
 917              strcmp(name, XATTR_NAME_POSIX_ACL_DEFAULT) == 0))
 918                 RETURN(-EOPNOTSUPP);
 919
 920         oh = container_of(handle, struct osd_thandle, ot_super);
 921
 922         down_write(&obj->oo_guard);
 923         CDEBUG(D_INODE, "Setting xattr %s with size %d\n",
 924                 name, (int)buf->lb_len);
 925         /* For the OST migrated from ldiskfs, the PFID EA may
 926          * be stored in LMA because of ldiskfs inode size. */
 927         if (unlikely(strcmp(name, XATTR_NAME_FID) == 0 &&
 928                      obj->oo_pfid_in_lma)) {
 929                 rc = osd_xattr_split_pfid(env, obj, oh);
 930                 if (!rc)
 931                         fl = LU_XATTR_CREATE;
 932         } else if (strcmp(name, XATTR_NAME_LINK) == 0 &&
 933                    osd->od_remote_parent_dir != ZFS_NO_OBJECT) {
 934                 rc = osd_xattr_handle_linkea(env, osd, obj, buf, oh);
 935         }
 936
 937         if (!rc)
 938                 rc = osd_xattr_set_internal(env, obj, buf, name, fl, oh);
 939         up_write(&obj->oo_guard);
 940
 941         RETURN(rc);
 942 }
 943
 944 static void
 945 __osd_xattr_declare_del(const struct lu_env *env, struct osd_object *obj,
 946                         const char *name, struct osd_thandle *oh)
 947 {
 948         struct osd_device *osd = osd_obj2dev(obj);
 949         dmu_tx_t          *tx = oh->ot_tx;
 950         uint64_t           xa_data_obj;
 951         int                rc;
 952
 953         /* update SA_ZPL_DXATTR if xattr was in SA */
 954         dmu_tx_hold_sa(tx, obj->oo_sa_hdl, 0);
 955
 956         if (obj->oo_xattr == ZFS_NO_OBJECT)
 957                 return;
 958
 959         rc = -zap_lookup(osd->od_os, obj->oo_xattr, name, 8, 1, &xa_data_obj);
 960         if (rc == 0) {
 961                 /*
 962                  * Entry exists.
 963                  * We'll delete the existing object and ZAP entry.
 964                  */
 965                 dmu_tx_hold_bonus(tx, xa_data_obj);
 966                 dmu_tx_hold_free(tx, xa_data_obj, 0, DMU_OBJECT_END);
 967                 dmu_tx_hold_zap(tx, obj->oo_xattr, FALSE, (char *) name);
 968                 return;
 969         } else if (rc == -ENOENT) {
 970                 /*
 971                  * Entry doesn't exist, nothing to be changed.
 972                  */
 973                 return;
 974         }
 975
 976         /* An error happened */
 977         tx->tx_err = -rc;
 978 }
 979
 980 int osd_declare_xattr_del(const struct lu_env *env, struct dt_object *dt,
 981                           const char *name, struct thandle *handle)
 982 {
 983         struct osd_object  *obj = osd_dt_obj(dt);
 984         struct osd_thandle *oh;
 985         ENTRY;
 986
 987         LASSERT(handle != NULL);
 988         LASSERT(osd_invariant(obj));
 989
 990         oh = container_of(handle, struct osd_thandle, ot_super);
 991         LASSERT(oh->ot_tx != NULL);
 992         LASSERT(obj->oo_dn != NULL);
 993
 994         down_read(&obj->oo_guard);
 995         if (likely(dt_object_exists(&obj->oo_dt) && !obj->oo_destroyed))
 996                 __osd_xattr_declare_del(env, obj, name, oh);
 997         up_read(&obj->oo_guard);
 998
 999         RETURN(0);
1000 }
1001
1002 static int __osd_sa_xattr_del(const struct lu_env *env, struct osd_object *obj,
1003                               const char *name, struct osd_thandle *oh)
1004 {
1005         int rc;
1006
1007         rc = __osd_xattr_cache(obj);
1008         if (rc)
1009                 return rc;
1010
1011         rc = -nvlist_remove(obj->oo_sa_xattr, name, DATA_TYPE_BYTE_ARRAY);
1012         if (rc)
1013                 return rc;
1014
1015         /*
1016          * only migrate delete LMV, and it needs to be done immediately, because
1017          * it's used in deleting sub stripes, and if this is delayed, later when
1018          * destroying the master object, it will delete sub stripes again.
1019          */
1020         if (!strcmp(name, XATTR_NAME_LMV))
1021                 rc = __osd_sa_xattr_update(env, obj, oh);
1022         else
1023                 rc = __osd_sa_xattr_schedule_update(env, obj, oh);
1024         return rc;
1025 }
1026
1027 static int __osd_xattr_del(const struct lu_env *env, struct osd_object *obj,
1028                            const char *name, struct osd_thandle *oh)
1029 {
1030         struct osd_device *osd = osd_obj2dev(obj);
1031         uint64_t           xa_data_obj;
1032         int                rc;
1033
1034         if (unlikely(!dt_object_exists(&obj->oo_dt) || obj->oo_destroyed))
1035                 return -ENOENT;
1036
1037         /* try remove xattr from SA at first */
1038         rc = __osd_sa_xattr_del(env, obj, name, oh);
1039         if (rc != -ENOENT)
1040                 return rc;
1041
1042         if (obj->oo_xattr == ZFS_NO_OBJECT)
1043                 return 0;
1044
1045         rc = -zap_lookup(osd->od_os, obj->oo_xattr, name, sizeof(uint64_t), 1,
1046                         &xa_data_obj);
1047         if (rc == -ENOENT) {
1048                 rc = 0;
1049         } else if (rc == 0) {
1050                 /*
1051                  * Entry exists.
1052                  * We'll delete the existing object and ZAP entry.
1053                  */
1054                 rc = -dmu_object_free(osd->od_os, xa_data_obj, oh->ot_tx);
1055                 if (rc)
1056                         return rc;
1057
1058                 rc = -zap_remove(osd->od_os, obj->oo_xattr, name, oh->ot_tx);
1059         }
1060
1061         return rc;
1062 }
1063
1064 int osd_xattr_del(const struct lu_env *env, struct dt_object *dt,
1065                   const char *name, struct thandle *handle)
1066 {
1067         struct osd_object  *obj = osd_dt_obj(dt);
1068         struct osd_thandle *oh;
1069         int                 rc;
1070         ENTRY;
1071
1072         LASSERT(handle != NULL);
1073         LASSERT(obj->oo_dn != NULL);
1074         LASSERT(osd_invariant(obj));
1075         LASSERT(dt_object_exists(dt));
1076         oh = container_of(handle, struct osd_thandle, ot_super);
1077         LASSERT(oh->ot_tx != NULL);
1078
1079         if (!osd_obj2dev(obj)->od_posix_acl &&
1080             (strcmp(name, XATTR_NAME_POSIX_ACL_ACCESS) == 0 ||
1081              strcmp(name, XATTR_NAME_POSIX_ACL_DEFAULT) == 0))
1082                 RETURN(-EOPNOTSUPP);
1083
1084         down_write(&obj->oo_guard);
1085         /* For the OST migrated from ldiskfs, the PFID EA may
1086          * be stored in LMA because of ldiskfs inode size. */
1087         if (unlikely(strcmp(name, XATTR_NAME_FID) == 0 && obj->oo_pfid_in_lma))
1088                 rc = osd_xattr_split_pfid(env, obj, oh);
1089         else
1090                 rc = __osd_xattr_del(env, obj, name, oh);
1091         up_write(&obj->oo_guard);
1092
1093         RETURN(rc);
1094 }
1095
1096 void osd_declare_xattrs_destroy(const struct lu_env *env,
1097                                 struct osd_object *obj, struct osd_thandle *oh)
1098 {
1099         struct osd_device *osd = osd_obj2dev(obj);
1100         zap_attribute_t   *za = &osd_oti_get(env)->oti_za;
1101         uint64_t           oid = obj->oo_xattr, xid;
1102         dmu_tx_t          *tx = oh->ot_tx;
1103         zap_cursor_t      *zc;
1104         int                rc;
1105
1106         if (oid == ZFS_NO_OBJECT)
1107                 return; /* Nothing to do for SA xattrs */
1108
1109         /* Declare to free the ZAP holding xattrs */
1110         dmu_tx_hold_free(tx, oid, 0, DMU_OBJECT_END);
1111
1112         rc = osd_zap_cursor_init(&zc, osd->od_os, oid, 0);
1113         if (rc)
1114                 goto out;
1115
1116         while (zap_cursor_retrieve(zc, za) == 0) {
1117                 LASSERT(za->za_num_integers == 1);
1118                 LASSERT(za->za_integer_length == sizeof(uint64_t));
1119
1120                 rc = -zap_lookup(osd->od_os, oid, za->za_name,
1121                                  sizeof(uint64_t), 1, &xid);
1122                 if (rc) {
1123                         CERROR("%s: xattr %s lookup failed: rc = %d\n",
1124                                osd->od_svname, za->za_name, rc);
1125                         break;
1126                 }
1127                 dmu_tx_hold_free(tx, xid, 0, DMU_OBJECT_END);
1128
1129                 zap_cursor_advance(zc);
1130         }
1131
1132         osd_zap_cursor_fini(zc);
1133 out:
1134         if (rc && tx->tx_err == 0)
1135                 tx->tx_err = -rc;
1136 }
1137
1138 int osd_xattrs_destroy(const struct lu_env *env,
1139                        struct osd_object *obj, struct osd_thandle *oh)
1140 {
1141         struct osd_device *osd = osd_obj2dev(obj);
1142         dmu_tx_t          *tx = oh->ot_tx;
1143         zap_attribute_t   *za = &osd_oti_get(env)->oti_za;
1144         zap_cursor_t      *zc;
1145         uint64_t           xid;
1146         int                rc;
1147
1148         /* The transaction must have been assigned to a transaction group. */
1149         LASSERT(tx->tx_txg != 0);
1150
1151         if (obj->oo_xattr == ZFS_NO_OBJECT)
1152                 return 0; /* Nothing to do for SA xattrs */
1153
1154         /* Free the ZAP holding the xattrs */
1155         rc = osd_zap_cursor_init(&zc, osd->od_os, obj->oo_xattr, 0);
1156         if (rc)
1157                 return rc;
1158
1159         while (zap_cursor_retrieve(zc, za) == 0) {
1160                 LASSERT(za->za_num_integers == 1);
1161                 LASSERT(za->za_integer_length == sizeof(uint64_t));
1162
1163                 rc = -zap_lookup(osd->od_os, obj->oo_xattr, za->za_name,
1164                                  sizeof(uint64_t), 1, &xid);
1165                 if (rc) {
1166                         CERROR("%s: lookup xattr %s failed: rc = %d\n",
1167                                osd->od_svname, za->za_name, rc);
1168                 } else {
1169                         rc = -dmu_object_free(osd->od_os, xid, tx);
1170                         if (rc)
1171                                 CERROR("%s: free xattr %s failed: rc = %d\n",
1172                                        osd->od_svname, za->za_name, rc);
1173                 }
1174                 zap_cursor_advance(zc);
1175         }
1176         osd_zap_cursor_fini(zc);
1177
1178         rc = -dmu_object_free(osd->od_os, obj->oo_xattr, tx);
1179         if (rc)
1180                 CERROR("%s: free xattr %llu failed: rc = %d\n",
1181                        osd->od_svname, obj->oo_xattr, rc);
1182
1183         return rc;
1184 }
1185
1186 static int
1187 osd_sa_xattr_list(const struct lu_env *env, struct osd_object *obj,
1188                   const struct lu_buf *lb)
1189 {
1190         nvpair_t *nvp = NULL;
1191         int       len, counted = 0;
1192         int       rc = 0;
1193
1194         rc = __osd_xattr_cache(obj);
1195         if (rc)
1196                 return rc;
1197
1198         while ((nvp = nvlist_next_nvpair(obj->oo_sa_xattr, nvp)) != NULL) {
1199                 const char *name = nvpair_name(nvp);
1200
1201                 if (!osd_obj2dev(obj)->od_posix_acl &&
1202                     (strcmp(name, XATTR_NAME_POSIX_ACL_ACCESS) == 0 ||
1203                      strcmp(name, XATTR_NAME_POSIX_ACL_DEFAULT) == 0))
1204                         continue;
1205
1206                 len = strlen(name);
1207                 if (lb->lb_buf != NULL) {
1208                         if (counted + len + 1 > lb->lb_len)
1209                                 return -ERANGE;
1210
1211                         memcpy(lb->lb_buf + counted, name, len + 1);
1212                 }
1213                 counted += len + 1;
1214         }
1215         return counted;
1216 }
1217
1218 int osd_xattr_list(const struct lu_env *env, struct dt_object *dt,
1219                    const struct lu_buf *lb)
1220 {
1221         struct osd_object      *obj = osd_dt_obj(dt);
1222         struct osd_device      *osd = osd_obj2dev(obj);
1223         zap_attribute_t        *za = &osd_oti_get(env)->oti_za;
1224         zap_cursor_t           *zc;
1225         int                    rc, counted;
1226         ENTRY;
1227
1228         LASSERT(obj->oo_dn != NULL);
1229         LASSERT(osd_invariant(obj));
1230         LASSERT(dt_object_exists(dt));
1231
1232         down_read(&obj->oo_guard);
1233
1234         rc = osd_sa_xattr_list(env, obj, lb);
1235         if (rc < 0)
1236                 GOTO(out, rc);
1237
1238         counted = rc;
1239
1240         /* continue with dnode xattr if any */
1241         if (obj->oo_xattr == ZFS_NO_OBJECT)
1242                 GOTO(out, rc = counted);
1243
1244         rc = osd_zap_cursor_init(&zc, osd->od_os, obj->oo_xattr, 0);
1245         if (rc)
1246                 GOTO(out, rc);
1247
1248         while ((rc = -zap_cursor_retrieve(zc, za)) == 0) {
1249                 if (!osd_obj2dev(obj)->od_posix_acl &&
1250                     (strcmp(za->za_name, XATTR_NAME_POSIX_ACL_ACCESS) == 0 ||
1251                      strcmp(za->za_name, XATTR_NAME_POSIX_ACL_DEFAULT) == 0)) {
1252                         zap_cursor_advance(zc);
1253                         continue;
1254                 }
1255
1256                 rc = strlen(za->za_name);
1257                 if (lb->lb_buf != NULL) {
1258                         if (counted + rc + 1 > lb->lb_len)
1259                                 GOTO(out_fini, rc = -ERANGE);
1260
1261                         memcpy(lb->lb_buf + counted, za->za_name, rc + 1);
1262                 }
1263                 counted += rc + 1;
1264
1265                 zap_cursor_advance(zc);
1266         }
1267         if (rc == -ENOENT) /* no more kes in the index */
1268                 rc = 0;
1269         else if (unlikely(rc < 0))
1270                 GOTO(out_fini, rc);
1271         rc = counted;
1272
1273 out_fini:
1274         osd_zap_cursor_fini(zc);
1275 out:
1276         up_read(&obj->oo_guard);
1277         RETURN(rc);
1278
1279 }