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
 281         ENTRY;
 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
 324         ENTRY;
 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          */
 338         if (strcmp(name, XATTR_NAME_FID) == 0 && obj->oo_pfid_in_lma)
 339                 rc = osd_get_pfid_from_lma(env, obj, buf, &size);
 340         else
 341                 rc = osd_xattr_get_internal(env, obj, buf, name, &size);
 342
 343         if (rc == -ENOENT)
 344                 rc = -ENODATA;
 345         else if (rc == 0)
 346                 rc = size;
 347 out:
 348         up_read(&obj->oo_guard);
 349         RETURN(rc);
 350 }
 351
 352 /* the function is used to declare EAs when SA is not supported */
 353 static void __osd_xattr_declare_legacy(const struct lu_env *env,
 354                                        struct osd_object *obj,
 355                                        int vallen, const char *name,
 356                                        struct osd_thandle *oh)
 357 {
 358         struct osd_device *osd = osd_obj2dev(obj);
 359         dmu_tx_t *tx = oh->ot_tx;
 360         uint64_t xa_data_obj;
 361         int rc;
 362
 363         if (obj->oo_xattr == ZFS_NO_OBJECT) {
 364                 /* xattr zap + entry */
 365                 dmu_tx_hold_zap(tx, DMU_NEW_OBJECT, TRUE, (char *) name);
 366                 /* xattr value obj */
 367                 dmu_tx_hold_sa_create(tx, ZFS_SA_BASE_ATTR_SIZE);
 368                 dmu_tx_hold_write(tx, DMU_NEW_OBJECT, 0, vallen);
 369                 return;
 370         }
 371
 372         rc = -zap_lookup(osd->od_os, obj->oo_xattr, name, sizeof(uint64_t), 1,
 373                         &xa_data_obj);
 374         if (rc == 0) {
 375                 /*
 376                  * Entry already exists.
 377                  * We'll truncate the existing object.
 378                  */
 379                 dmu_tx_hold_bonus(tx, xa_data_obj);
 380                 dmu_tx_hold_free(tx, xa_data_obj, vallen, DMU_OBJECT_END);
 381                 dmu_tx_hold_write(tx, xa_data_obj, 0, vallen);
 382         } else if (rc == -ENOENT) {
 383                 /*
 384                  * Entry doesn't exist, we need to create a new one and a new
 385                  * object to store the value.
 386                  */
 387                 dmu_tx_hold_bonus(tx, obj->oo_xattr);
 388                 dmu_tx_hold_zap(tx, obj->oo_xattr, TRUE, (char *) name);
 389                 dmu_tx_hold_sa_create(tx, ZFS_SA_BASE_ATTR_SIZE);
 390                 dmu_tx_hold_write(tx, DMU_NEW_OBJECT, 0, vallen);
 391         }
 392 }
 393
 394 void __osd_xattr_declare_set(const struct lu_env *env, struct osd_object *obj,
 395                              int vallen, const char *name,
 396                              struct osd_thandle *oh)
 397 {
 398         struct osd_device *osd = osd_obj2dev(obj);
 399         dmu_tx_t *tx = oh->ot_tx;
 400         int bonuslen;
 401
 402         if (unlikely(obj->oo_destroyed))
 403                 return;
 404
 405         if (strcmp(name, XATTR_NAME_LINK) == 0 &&
 406             osd->od_remote_parent_dir != ZFS_NO_OBJECT) {
 407                 /* If some name entry resides on remote MDT, then will create
 408                  * agent entry under remote parent. On the other hand, if the
 409                  * remote entry will be removed, then related agent entry may
 410                  * need to be removed from the remote parent. So there may be
 411                  * kinds of cases, let's declare enough credits. The credits
 412                  * for create agent entry is enough for remove case.
 413                  */
 414                 osd_tx_hold_zap(tx, osd->od_remote_parent_dir,
 415                                 NULL, TRUE, NULL);
 416         }
 417
 418         if (unlikely(!osd_obj2dev(obj)->od_xattr_in_sa)) {
 419                 __osd_xattr_declare_legacy(env, obj, vallen, name, oh);
 420                 return;
 421         }
 422
 423         /* declare EA in SA */
 424         if (dt_object_exists(&obj->oo_dt)) {
 425                 LASSERT(obj->oo_sa_hdl);
 426                 /* XXX: it should be possible to skip spill declaration if
 427                  * specific EA is part of bonus and doesn't grow
 428                  */
 429                 dmu_tx_hold_spill(tx, obj->oo_dn->dn_object);
 430                 return;
 431         }
 432
 433         bonuslen = osd_obj_bonuslen(obj);
 434
 435         /* the object doesn't exist, but we've declared bonus
 436          * in osd_declare_object_create() yet
 437          */
 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
 454         ENTRY;
 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                 rc = -ENOMEM;
 537                 CERROR("%s: can't allocate buffer for xattr update: rc = %d\n",
 538                        osd->od_svname, rc);
 539                 return rc;
 540         }
 541
 542         rc = -nvlist_pack(obj->oo_sa_xattr, (char **)&lb->lb_buf, &size,
 543                           NV_ENCODE_XDR, KM_SLEEP);
 544         if (rc)
 545                 return rc;
 546
 547         SA_ADD_BULK_ATTR(bulk, cnt, SA_ZPL_DXATTR(osd), NULL, lb->lb_buf, size);
 548
 549         rc = -sa_replace_all_by_template(obj->oo_sa_hdl, bulk, cnt, oh->ot_tx);
 550
 551         return rc;
 552 }
 553
 554 int __osd_sa_xattr_update(const struct lu_env *env, struct osd_object *obj,
 555                            struct osd_thandle *oh)
 556 {
 557         struct lu_buf *lb = &osd_oti_get(env)->oti_xattr_lbuf;
 558         struct osd_device *osd = osd_obj2dev(obj);
 559         char *dxattr;
 560         size_t size;
 561         int rc;
 562
 563         obj->oo_late_xattr = 0;
 564
 565         /* Update the SA for additions, modifications, and removals. */
 566         rc = -nvlist_size(obj->oo_sa_xattr, &size, NV_ENCODE_XDR);
 567         if (rc)
 568                 return rc;
 569
 570         lu_buf_check_and_alloc(lb, size);
 571         if (lb->lb_buf == NULL) {
 572                 rc = -ENOMEM;
 573                 CERROR("%s: can't allocate buffer for xattr update: rc = %d\n",
 574                        osd->od_svname, rc);
 575                 return rc;
 576         }
 577
 578         dxattr = lb->lb_buf;
 579         rc = -nvlist_pack(obj->oo_sa_xattr, &dxattr, &size,
 580                         NV_ENCODE_XDR, KM_SLEEP);
 581         if (rc)
 582                 return rc;
 583         LASSERT(dxattr == lb->lb_buf);
 584
 585         sa_update(obj->oo_sa_hdl, SA_ZPL_DXATTR(osd), dxattr, size, oh->ot_tx);
 586
 587         return 0;
 588 }
 589
 590 /*
 591  * Set an extended attribute.
 592  * This transaction must have called udmu_xattr_declare_set() first.
 593  *
 594  * Returns 0 on success or a negative error number on failure.
 595  *
 596  * No locking is done here.
 597  */
 598 int __osd_sa_xattr_schedule_update(const struct lu_env *env,
 599                                    struct osd_object *obj,
 600                                    struct osd_thandle *oh)
 601 {
 602         ENTRY;
 603         LASSERT(obj->oo_sa_hdl);
 604         LASSERT(obj->oo_sa_xattr);
 605
 606         /* schedule batched SA update in osd_object_sa_dirty_rele() */
 607         obj->oo_late_xattr = 1;
 608         osd_object_sa_dirty_add(obj, oh);
 609
 610         RETURN(0);
 611
 612 }
 613
 614 int __osd_sa_xattr_set(const struct lu_env *env, struct osd_object *obj,
 615                        const struct lu_buf *buf, const char *name, int fl,
 616                        struct osd_thandle *oh)
 617 {
 618         uchar_t *nv_value;
 619         size_t size;
 620         int nv_size;
 621         int rc;
 622         int too_big = 0;
 623
 624         rc = __osd_xattr_cache(obj);
 625         if (rc)
 626                 return rc;
 627
 628         LASSERT(obj->oo_sa_xattr);
 629         if (buf->lb_len > OBD_MAX_EA_SIZE) {
 630                 too_big = 1;
 631         } else {
 632                 /* Prevent the DXATTR SA from consuming the entire SA region */
 633                 rc = -nvlist_size(obj->oo_sa_xattr, &size, NV_ENCODE_XDR);
 634                 if (rc)
 635                         return rc;
 636
 637                 if (size + buf->lb_len > DXATTR_MAX_SA_SIZE)
 638                         too_big = 1;
 639         }
 640
 641         /* even in case of -EFBIG we must lookup xattr and check can we
 642          * rewrite it then delete from SA
 643          */
 644         rc = -nvlist_lookup_byte_array(obj->oo_sa_xattr, name, &nv_value,
 645                                         &nv_size);
 646         if (rc == 0) {
 647                 if (fl & LU_XATTR_CREATE) {
 648                         return -EEXIST;
 649                 } else if (too_big) {
 650                         rc = -nvlist_remove(obj->oo_sa_xattr, name,
 651                                                 DATA_TYPE_BYTE_ARRAY);
 652                         if (rc < 0)
 653                                 return rc;
 654                         rc = __osd_sa_xattr_schedule_update(env, obj, oh);
 655                         return rc == 0 ? -EFBIG : rc;
 656                 }
 657         } else if (rc == -ENOENT) {
 658                 if (fl & LU_XATTR_REPLACE)
 659                         return -ENODATA;
 660                 else if (too_big)
 661                         return -EFBIG;
 662         } else {
 663                 return rc;
 664         }
 665
 666         /* Ensure xattr doesn't exist in ZAP */
 667         if (obj->oo_xattr != ZFS_NO_OBJECT) {
 668                 struct osd_device *osd = osd_obj2dev(obj);
 669                 uint64_t           objid;
 670
 671                 rc = -zap_lookup(osd->od_os, obj->oo_xattr,
 672                                  name, 8, 1, &objid);
 673                 if (rc == 0) {
 674                         rc = -dmu_object_free(osd->od_os, objid, oh->ot_tx);
 675                         if (rc == 0)
 676                                 zap_remove(osd->od_os, obj->oo_xattr,
 677                                            name, oh->ot_tx);
 678                 }
 679         }
 680
 681         rc = -nvlist_add_byte_array(obj->oo_sa_xattr, name,
 682                                     (uchar_t *)buf->lb_buf, buf->lb_len);
 683         if (rc)
 684                 return rc;
 685
 686         /* batch updates only for just created dnodes where we
 687          * used to set number of EAs in a single transaction
 688          */
 689         if (obj->oo_dn->dn_allocated_txg == oh->ot_tx->tx_txg)
 690                 rc = __osd_sa_xattr_schedule_update(env, obj, oh);
 691         else
 692                 rc = __osd_sa_xattr_update(env, obj, oh);
 693
 694         return rc;
 695 }
 696
 697 int
 698 __osd_xattr_set(const struct lu_env *env, struct osd_object *obj,
 699                 const struct lu_buf *buf, const char *name, int fl,
 700                 struct osd_thandle *oh)
 701 {
 702         struct osd_device *osd = osd_obj2dev(obj);
 703         dnode_t *xa_zap_dn = NULL;
 704         dnode_t *xa_data_dn = NULL;
 705         uint64_t xa_data_obj;
 706         sa_handle_t *sa_hdl = NULL;
 707         dmu_tx_t *tx = oh->ot_tx;
 708         uint64_t size;
 709         int rc;
 710
 711         LASSERT(obj->oo_sa_hdl);
 712
 713         if (obj->oo_xattr == ZFS_NO_OBJECT) {
 714                 struct lu_attr *la = &osd_oti_get(env)->oti_la;
 715
 716                 la->la_valid = LA_MODE;
 717                 la->la_mode = S_IFDIR | 0755;
 718                 rc = __osd_zap_create(env, osd, &xa_zap_dn, tx, la, 0, 0);
 719                 if (rc)
 720                         return rc;
 721
 722                 obj->oo_xattr = xa_zap_dn->dn_object;
 723                 rc = osd_object_sa_update(obj, SA_ZPL_XATTR(osd),
 724                                 &obj->oo_xattr, 8, oh);
 725                 if (rc)
 726                         goto out;
 727         }
 728
 729         rc = -zap_lookup(osd->od_os, obj->oo_xattr, name, sizeof(uint64_t), 1,
 730                          &xa_data_obj);
 731         if (rc == 0) {
 732                 if (fl & LU_XATTR_CREATE) {
 733                         rc = -EEXIST;
 734                         goto out;
 735                 }
 736                 /* Entry already exists. We'll truncate the existing object. */
 737                 rc = __osd_obj2dnode(osd->od_os, xa_data_obj, &xa_data_dn);
 738                 if (rc)
 739                         goto out;
 740
 741                 rc = -sa_handle_get(osd->od_os, xa_data_obj, NULL,
 742                                         SA_HDL_PRIVATE, &sa_hdl);
 743                 if (rc)
 744                         goto out;
 745
 746                 rc = -sa_lookup(sa_hdl, SA_ZPL_SIZE(osd), &size, 8);
 747                 if (rc)
 748                         goto out_sa;
 749
 750                 rc = -dmu_free_range(osd->od_os, xa_data_dn->dn_object,
 751                                      0, DMU_OBJECT_END, tx);
 752                 if (rc)
 753                         goto out_sa;
 754         } else if (rc == -ENOENT) {
 755                 struct lu_attr *la = &osd_oti_get(env)->oti_la;
 756                 /*
 757                  * Entry doesn't exist, we need to create a new one and a new
 758                  * object to store the value.
 759                  */
 760                 if (fl & LU_XATTR_REPLACE) {
 761                         /* should be ENOATTR according to the
 762                          * man, but that is undefined here
 763                          */
 764                         rc = -ENODATA;
 765                         goto out;
 766                 }
 767
 768                 la->la_valid = LA_MODE;
 769                 la->la_mode = S_IFREG | 0644;
 770                 rc = __osd_object_create(env, osd, obj,
 771                                          lu_object_fid(&obj->oo_dt.do_lu),
 772                                          &xa_data_dn, tx, la);
 773                 if (rc)
 774                         goto out;
 775                 xa_data_obj = xa_data_dn->dn_object;
 776
 777                 rc = -sa_handle_get(osd->od_os, xa_data_obj, NULL,
 778                                         SA_HDL_PRIVATE, &sa_hdl);
 779                 if (rc)
 780                         goto out;
 781
 782                 rc = -zap_add(osd->od_os, obj->oo_xattr, name, sizeof(uint64_t),
 783                                 1, &xa_data_obj, tx);
 784                 if (rc)
 785                         goto out_sa;
 786         } else {
 787                 /* There was an error looking up the xattr name */
 788                 goto out;
 789         }
 790
 791         /* Finally write the xattr value */
 792         dmu_write(osd->od_os, xa_data_obj, 0, buf->lb_len, buf->lb_buf, tx);
 793
 794         size = buf->lb_len;
 795         rc = -sa_update(sa_hdl, SA_ZPL_SIZE(osd), &size, 8, tx);
 796
 797 out_sa:
 798         sa_handle_destroy(sa_hdl);
 799 out:
 800         if (xa_data_dn != NULL)
 801                 osd_dnode_rele(xa_data_dn);
 802         if (xa_zap_dn != NULL)
 803                 osd_dnode_rele(xa_zap_dn);
 804
 805         return rc;
 806 }
 807
 808 static int osd_xattr_split_pfid(const struct lu_env *env,
 809                                 struct osd_object *obj, struct osd_thandle *oh)
 810 {
 811         struct osd_thread_info *info = osd_oti_get(env);
 812         struct lustre_ost_attrs *loa =
 813                 (struct lustre_ost_attrs *)&info->oti_buf;
 814         struct lustre_mdt_attrs *lma = &loa->loa_lma;
 815         struct lu_buf buf = {
 816                 .lb_buf = loa,
 817                 .lb_len = sizeof(info->oti_buf),
 818         };
 819         int size;
 820         int rc;
 821
 822         ENTRY;
 823         BUILD_BUG_ON(sizeof(info->oti_buf) < sizeof(*loa));
 824         rc = osd_xattr_get_internal(env, obj, &buf, XATTR_NAME_LMA, &size);
 825         if (rc)
 826                 RETURN(rc);
 827
 828         lustre_loa_swab(loa, true);
 829         LASSERT(lma->lma_compat & LMAC_STRIPE_INFO);
 830
 831         lma->lma_compat &= ~(LMAC_STRIPE_INFO | LMAC_COMP_INFO);
 832         lustre_lma_swab(lma);
 833         buf.lb_buf = lma;
 834         buf.lb_len = sizeof(*lma);
 835         rc = osd_xattr_set_internal(env, obj, &buf, XATTR_NAME_LMA,
 836                                     LU_XATTR_REPLACE, oh);
 837         if (!rc)
 838                 obj->oo_pfid_in_lma = 0;
 839
 840         RETURN(rc);
 841 }
 842
 843 /*
 844  * In DNE environment, the object (in spite of regular file or directory)
 845  * and its name entry may reside on different MDTs. Under such case, we will
 846  * create an agent entry on the MDT where the object resides. The agent entry
 847  * references the object locally, that makes the object to be visible to the
 848  * userspace when mounted as 'zfs' directly. Then the userspace tools, such
 849  * as 'tar' can handle the object properly.
 850  *
 851  * We handle the agent entry during set linkEA that is the common interface
 852  * for both regular file and directroy, can handle kinds of cases, such as
 853  * create/link/unlink/rename, and so on.
 854  *
 855  * NOTE: we need to do that for both directory and regular file, so we can NOT
 856  *       do that when ea_{insert,delete} that are directory based operations.
 857  */
 858 static int osd_xattr_handle_linkea(const struct lu_env *env,
 859                                    struct osd_device *osd,
 860                                    struct osd_object *obj,
 861                                    const struct lu_buf *buf,
 862                                    struct osd_thandle *oh)
 863 {
 864         const struct lu_fid *fid = lu_object_fid(&obj->oo_dt.do_lu);
 865         struct lu_fid *tfid = &osd_oti_get(env)->oti_fid;
 866         struct linkea_data ldata = { .ld_buf = (struct lu_buf *)buf };
 867         struct lu_name tmpname;
 868         int rc;
 869         bool remote = false;
 870
 871         ENTRY;
 872         rc = linkea_init_with_rec(&ldata);
 873         if (!rc) {
 874                 linkea_first_entry(&ldata);
 875                 while (ldata.ld_lee != NULL && !remote) {
 876                         linkea_entry_unpack(ldata.ld_lee, &ldata.ld_reclen,
 877                                             &tmpname, tfid);
 878                         if (osd_remote_fid(env, osd, tfid) > 0)
 879                                 remote = true;
 880                         else
 881                                 linkea_next_entry(&ldata);
 882                 }
 883         } else if (rc == -ENODATA) {
 884                 rc = 0;
 885         } else {
 886                 RETURN(rc);
 887         }
 888
 889         if (lu_object_has_agent_entry(&obj->oo_dt.do_lu) && !remote) {
 890                 rc = osd_delete_from_remote_parent(env, osd, obj, oh, false);
 891                 if (rc)
 892                         CERROR("%s: failed to remove agent entry for "DFID": rc = %d\n",
 893                                osd_name(osd), PFID(fid), rc);
 894         } else if (!lu_object_has_agent_entry(&obj->oo_dt.do_lu) && remote) {
 895                 rc = osd_add_to_remote_parent(env, osd, obj, oh);
 896                 if (rc)
 897                         CWARN("%s: failed to create agent entry for "DFID": rc = %d\n",
 898                               osd_name(osd), PFID(fid), rc);
 899         }
 900
 901         RETURN(rc);
 902 }
 903
 904 int osd_xattr_set(const struct lu_env *env, struct dt_object *dt,
 905                   const struct lu_buf *buf, const char *name, int fl,
 906                   struct thandle *handle)
 907 {
 908         struct osd_object *obj = osd_dt_obj(dt);
 909         struct osd_device *osd = osd_obj2dev(obj);
 910         struct osd_thandle *oh;
 911         int rc = 0;
 912
 913         ENTRY;
 914         LASSERT(handle != NULL);
 915         LASSERT(osd_invariant(obj));
 916
 917         if (!osd_obj2dev(obj)->od_posix_acl &&
 918             (strcmp(name, XATTR_NAME_POSIX_ACL_ACCESS) == 0 ||
 919              strcmp(name, XATTR_NAME_POSIX_ACL_DEFAULT) == 0))
 920                 RETURN(-EOPNOTSUPP);
 921
 922         oh = container_of(handle, struct osd_thandle, ot_super);
 923
 924         down_write(&obj->oo_guard);
 925         CDEBUG(D_INODE, "Setting xattr %s with size %d\n",
 926                 name, (int)buf->lb_len);
 927         /* For the OST migrated from ldiskfs, the PFID EA may
 928          * be stored in LMA because of ldiskfs inode size.
 929          */
 930         if (unlikely(strcmp(name, XATTR_NAME_FID) == 0 &&
 931                      obj->oo_pfid_in_lma)) {
 932                 rc = osd_xattr_split_pfid(env, obj, oh);
 933                 if (!rc)
 934                         fl = LU_XATTR_CREATE;
 935         } else if (strcmp(name, XATTR_NAME_LINK) == 0 &&
 936                    osd->od_remote_parent_dir != ZFS_NO_OBJECT) {
 937                 rc = osd_xattr_handle_linkea(env, osd, obj, buf, oh);
 938         }
 939
 940         if (!rc)
 941                 rc = osd_xattr_set_internal(env, obj, buf, name, fl, oh);
 942         up_write(&obj->oo_guard);
 943
 944         RETURN(rc);
 945 }
 946
 947 static void
 948 __osd_xattr_declare_del(const struct lu_env *env, struct osd_object *obj,
 949                         const char *name, struct osd_thandle *oh)
 950 {
 951         struct osd_device *osd = osd_obj2dev(obj);
 952         dmu_tx_t *tx = oh->ot_tx;
 953         uint64_t xa_data_obj;
 954         int rc;
 955
 956         /* update SA_ZPL_DXATTR if xattr was in SA */
 957         dmu_tx_hold_sa(tx, obj->oo_sa_hdl, 0);
 958
 959         if (obj->oo_xattr == ZFS_NO_OBJECT)
 960                 return;
 961
 962         rc = -zap_lookup(osd->od_os, obj->oo_xattr, name, 8, 1, &xa_data_obj);
 963         if (rc == 0) {
 964                 /* Entry exists. Will delete the existing obj and ZAP entry */
 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
 986         ENTRY;
 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
 993         down_read(&obj->oo_guard);
 994         if (likely(dt_object_exists(&obj->oo_dt) && !obj->oo_destroyed)) {
 995                 LASSERT(obj->oo_dn != NULL);
 996                 __osd_xattr_declare_del(env, obj, name, oh);
 997         }
 998         up_read(&obj->oo_guard);
 999
1000         RETURN(0);
1001 }
1002
1003 static int __osd_sa_xattr_del(const struct lu_env *env, struct osd_object *obj,
1004                               const char *name, struct osd_thandle *oh)
1005 {
1006         int rc;
1007
1008         rc = __osd_xattr_cache(obj);
1009         if (rc)
1010                 return rc;
1011
1012         rc = -nvlist_remove(obj->oo_sa_xattr, name, DATA_TYPE_BYTE_ARRAY);
1013         if (rc)
1014                 return rc;
1015
1016         /*
1017          * only migrate delete LMV, and it needs to be done immediately, because
1018          * it's used in deleting sub stripes, and if this is delayed, later when
1019          * destroying the master object, it will delete sub stripes again.
1020          */
1021         if (!strcmp(name, XATTR_NAME_LMV))
1022                 rc = __osd_sa_xattr_update(env, obj, oh);
1023         else
1024                 rc = __osd_sa_xattr_schedule_update(env, obj, oh);
1025         return rc;
1026 }
1027
1028 static int __osd_xattr_del(const struct lu_env *env, struct osd_object *obj,
1029                            const char *name, struct osd_thandle *oh)
1030 {
1031         struct osd_device *osd = osd_obj2dev(obj);
1032         uint64_t xa_data_obj;
1033         int rc;
1034
1035         if (unlikely(!dt_object_exists(&obj->oo_dt) || obj->oo_destroyed))
1036                 return -ENOENT;
1037
1038         /* try remove xattr from SA at first */
1039         rc = __osd_sa_xattr_del(env, obj, name, oh);
1040         if (rc != -ENOENT)
1041                 return rc;
1042
1043         if (obj->oo_xattr == ZFS_NO_OBJECT)
1044                 return 0;
1045
1046         rc = -zap_lookup(osd->od_os, obj->oo_xattr, name, sizeof(uint64_t), 1,
1047                         &xa_data_obj);
1048         if (rc == -ENOENT) {
1049                 rc = 0;
1050         } else if (rc == 0) {
1051                 /* Entry exists. We'll delete the existing obj and ZAP entry */
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
1069         ENTRY;
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          */
1086         if (unlikely(strcmp(name, XATTR_NAME_FID) == 0 && obj->oo_pfid_in_lma))
1087                 rc = osd_xattr_split_pfid(env, obj, oh);
1088         else
1089                 rc = __osd_xattr_del(env, obj, name, oh);
1090
1091 out:
1092         up_write(&obj->oo_guard);
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
1227         ENTRY;
1228         down_read(&obj->oo_guard);
1229         if (unlikely(!dt_object_exists(dt) || obj->oo_destroyed))
1230                 GOTO(out, rc = -ENOENT);
1231         LASSERT(obj->oo_dn != NULL);
1232
1233         rc = osd_sa_xattr_list(env, obj, lb);
1234         if (rc < 0)
1235                 GOTO(out, rc);
1236
1237         counted = rc;
1238
1239         /* continue with dnode xattr if any */
1240         if (obj->oo_xattr == ZFS_NO_OBJECT)
1241                 GOTO(out, rc = counted);
1242
1243         rc = osd_zap_cursor_init(&zc, osd->od_os, obj->oo_xattr, 0);
1244         if (rc)
1245                 GOTO(out, rc);
1246
1247         while ((rc = -zap_cursor_retrieve(zc, za)) == 0) {
1248                 if (!osd_obj2dev(obj)->od_posix_acl &&
1249                     (strcmp(za->za_name, XATTR_NAME_POSIX_ACL_ACCESS) == 0 ||
1250                      strcmp(za->za_name, XATTR_NAME_POSIX_ACL_DEFAULT) == 0)) {
1251                         zap_cursor_advance(zc);
1252                         continue;
1253                 }
1254
1255                 rc = strlen(za->za_name);
1256                 if (lb->lb_buf != NULL) {
1257                         if (counted + rc + 1 > lb->lb_len)
1258                                 GOTO(out_fini, rc = -ERANGE);
1259
1260                         memcpy(lb->lb_buf + counted, za->za_name, rc + 1);
1261                 }
1262                 counted += rc + 1;
1263
1264                 zap_cursor_advance(zc);
1265         }
1266         if (rc == -ENOENT) /* no more kes in the index */
1267                 rc = 0;
1268         else if (unlikely(rc < 0))
1269                 GOTO(out_fini, rc);
1270         rc = counted;
1271
1272 out_fini:
1273         osd_zap_cursor_fini(zc);
1274 out:
1275         up_read(&obj->oo_guard);
1276         RETURN(rc);
1277
1278 }