diff --git a/src/client/dfs/xattr.c b/src/client/dfs/xattr.c index 51ce1d71e41..4ba4a5475db 100644 --- a/src/client/dfs/xattr.c +++ b/src/client/dfs/xattr.c @@ -19,17 +19,19 @@ int dfs_setxattr(dfs_t *dfs, dfs_obj_t *obj, const char *name, const void *value, daos_size_t size, int flags) { - char *xname = NULL; - daos_handle_t th = DAOS_TX_NONE; - d_sg_list_t sgls[2]; - d_iov_t sg_iovs[3]; - daos_iod_t iods[2]; - daos_recx_t recxs[2]; - daos_key_t dkey; - daos_handle_t oh; - uint64_t cond = 0; - struct timespec now; - int rc; + char *xname = NULL; + daos_handle_t th = DAOS_TX_NONE; + d_sg_list_t sgls[2]; + d_iov_t sg_iovs[3]; + daos_iod_t iods[2]; + daos_recx_t recxs[2]; + daos_key_t dkey; + daos_handle_t oh = DAOS_HDL_INVAL; + uint64_t cond = 0; + struct timespec now; + struct dfs_entry entry = {0}; + bool exists; + int rc; if (dfs == NULL || !dfs->mounted) return EINVAL; @@ -51,14 +53,6 @@ dfs_setxattr(dfs_t *dfs, dfs_obj_t *obj, const char *name, const void *value, da if (xname == NULL) return ENOMEM; - /** Open parent object and insert xattr in the entry of the object */ - rc = daos_obj_open(dfs->coh, obj->parent_oid, DAOS_OO_RW, &oh, NULL); - if (rc) - D_GOTO(free, rc = daos_der2errno(rc)); - - /** set dkey as the entry name */ - d_iov_set(&dkey, (void *)obj->name, strlen(obj->name)); - /** add xattr iod & sgl */ d_iov_set(&iods[0].iod_name, xname, strlen(xname)); iods[0].iod_nr = 1; @@ -83,14 +77,9 @@ dfs_setxattr(dfs_t *dfs, dfs_obj_t *obj, const char *name, const void *value, da recxs[0].rx_nr = sizeof(uint64_t); recxs[1].rx_idx = CTIME_NSEC_IDX; recxs[1].rx_nr = sizeof(uint64_t); - rc = clock_gettime(CLOCK_REALTIME, &now); - if (rc) - D_GOTO(out, rc = errno); sgls[1].sg_nr = 2; sgls[1].sg_nr_out = 0; sgls[1].sg_iovs = &sg_iovs[1]; - d_iov_set(&sg_iovs[1], &now.tv_sec, sizeof(uint64_t)); - d_iov_set(&sg_iovs[2], &now.tv_nsec, sizeof(uint64_t)); /** if not default flag, check for xattr existence */ if (flags != 0) { @@ -101,34 +90,132 @@ dfs_setxattr(dfs_t *dfs, dfs_obj_t *obj, const char *name, const void *value, da } cond |= DAOS_COND_DKEY_UPDATE; + /** + * Use dtx in balanced mode to serialize converting the + * file to hardlink and setting/updating xattr in parallel. + */ + if (dfs->use_dtx) { + rc = daos_tx_open(dfs->coh, &th, 0, NULL); + if (rc) { + D_ERROR("daos_tx_open() failed (%d)\n", rc); + D_GOTO(out_free, rc = daos_der2errno(rc)); + } + } + +retry: + if (DFS_IS_HARDLINK(obj->mode)) { + /** For hardlinks, xattrs are stored in GIT with OID as dkey */ + if (!daos_handle_is_valid(dfs->git_oh)) + D_GOTO(out_tx, rc = ENOTSUP); + d_iov_set(&dkey, &obj->oid, sizeof(daos_obj_id_t)); + oh = dfs->git_oh; + } else { + /** Open parent object and insert xattr in the entry of the object */ + rc = daos_obj_open(dfs->coh, obj->parent_oid, DAOS_OO_RW, &oh, NULL); + if (rc) + D_GOTO(out_tx, rc = daos_der2errno(rc)); + + /* + * Since this function deals with dfs_obj, the hardlink information in its + * mode field may be outdated. Check and update the mode with hardlink info. + * Retry if the hardlink bit is set. + */ + if (S_ISREG(obj->mode)) { + rc = fetch_entry(dfs->layout_v, oh, th, obj->name, strlen(obj->name), false, + &exists, &entry, 0, NULL, NULL, NULL); + if (rc) { + D_ERROR("Failed to fetch entry '%s' (%d)\n", obj->name, rc); + D_GOTO(out_obj, rc); + } + if (!exists) + D_GOTO(out_obj, rc = ENOENT); + if (DFS_IS_HARDLINK(entry.mode)) { + dfs_set_hardlink(&obj->mode); + daos_obj_close(oh, NULL); + oh = DAOS_HDL_INVAL; + goto retry; + } + } + + /** set dkey as the entry name */ + d_iov_set(&dkey, (void *)obj->name, strlen(obj->name)); + } + + rc = clock_gettime(CLOCK_REALTIME, &now); + if (rc) + D_GOTO(out_obj, rc = errno); + d_iov_set(&sg_iovs[1], &now.tv_sec, sizeof(uint64_t)); + d_iov_set(&sg_iovs[2], &now.tv_nsec, sizeof(uint64_t)); + + /** Since conditional updates are not possible within DTX explicitly check for existence of + * xattr before inserting/updating it. + */ + if (dfs->use_dtx) { + daos_iod_t check_iod = {0}; + + d_iov_set(&check_iod.iod_name, xname, strlen(xname)); + check_iod.iod_nr = 1; + check_iod.iod_type = DAOS_IOD_SINGLE; + check_iod.iod_size = DAOS_REC_ANY; + rc = daos_obj_fetch(oh, th, DAOS_COND_AKEY_FETCH, &dkey, 1, &check_iod, NULL, NULL, + NULL); + if ((rc == 0) && (flags & XATTR_CREATE)) { + D_ERROR("xattr %s already exists\n", name); + D_GOTO(out_obj, rc = EEXIST); + } else if ((rc == -DER_NONEXIST) && (flags & XATTR_REPLACE)) { + D_ERROR("xattr %s does not exist\n", name); + D_GOTO(out_obj, rc = ENODATA); + } else if (rc && rc != -DER_NONEXIST) { + D_ERROR("Failed to check xattr %s existence\n", name); + D_GOTO(out_obj, rc = daos_der2errno(rc)); + } + rc = 0; + } + /** update ctime in a separate update if DAOS_COND_AKEY_INSERT is used for the xattr */ if (cond & DAOS_COND_AKEY_INSERT) { /** insert the xattr */ - rc = daos_obj_update(oh, th, cond, &dkey, 1, &iods[0], &sgls[0], NULL); + rc = daos_obj_update(oh, th, dfs->use_dtx ? 0 : cond, &dkey, 1, &iods[0], &sgls[0], + NULL); if (rc) { D_ERROR("Failed to insert extended attribute %s\n", name); - D_GOTO(out, rc = daos_der2errno(rc)); + D_GOTO(out_obj, rc = daos_der2errno(rc)); } /** update the ctime */ - rc = daos_obj_update(oh, th, DAOS_COND_DKEY_UPDATE, &dkey, 1, &iods[1], &sgls[1], - NULL); + rc = daos_obj_update(oh, th, dfs->use_dtx ? 0 : DAOS_COND_DKEY_UPDATE, &dkey, 1, + &iods[1], &sgls[1], NULL); if (rc) { D_ERROR("Failed to update ctime %s\n", name); - D_GOTO(out, rc = daos_der2errno(rc)); + D_GOTO(out_obj, rc = daos_der2errno(rc)); } } else { /** replace the xattr and update the ctime */ - rc = daos_obj_update(oh, th, cond, &dkey, 2, iods, sgls, NULL); + rc = daos_obj_update(oh, th, dfs->use_dtx ? 0 : cond, &dkey, 2, iods, sgls, NULL); if (rc) { D_ERROR("Failed to insert extended attribute %s\n", name); - D_GOTO(out, rc = daos_der2errno(rc)); + D_GOTO(out_obj, rc = daos_der2errno(rc)); } } DFS_OP_STAT_INCR(dfs, DOS_SETXATTR); -out: - daos_obj_close(oh, NULL); -free: +out_obj: + /** Only close parent handle; GIT handle is owned by the DFS mount. */ + if (!DFS_IS_HARDLINK(obj->mode) && daos_handle_is_valid(oh)) { + daos_obj_close(oh, NULL); + oh = DAOS_HDL_INVAL; + } +out_tx: + if (dfs->use_dtx) { + if (rc == 0) { + rc = daos_tx_commit(th, NULL); + if (rc) + rc = daos_der2errno(rc); + } + rc = check_tx(th, rc); + if (rc == ERESTART) + goto retry; + } +out_free: D_FREE(xname); return rc; } @@ -136,13 +223,15 @@ dfs_setxattr(dfs_t *dfs, dfs_obj_t *obj, const char *name, const void *value, da int dfs_getxattr(dfs_t *dfs, dfs_obj_t *obj, const char *name, void *value, daos_size_t *size) { - char *xname = NULL; - d_sg_list_t sgl; - d_iov_t sg_iov; - daos_iod_t iod; - daos_key_t dkey; - daos_handle_t oh; - int rc; + char *xname = NULL; + d_sg_list_t sgl; + d_iov_t sg_iov; + daos_iod_t iod; + daos_key_t dkey; + daos_handle_t oh = DAOS_HDL_INVAL; + struct dfs_entry entry = {0}; + bool exists; + int rc; if (dfs == NULL || !dfs->mounted) return EINVAL; @@ -159,14 +248,6 @@ dfs_getxattr(dfs_t *dfs, dfs_obj_t *obj, const char *name, void *value, daos_siz if (xname == NULL) return ENOMEM; - /** Open parent object and get xattr from the entry of the object */ - rc = daos_obj_open(dfs->coh, obj->parent_oid, DAOS_OO_RO, &oh, NULL); - if (rc) - D_GOTO(out, rc = daos_der2errno(rc)); - - /** set dkey as the entry name */ - d_iov_set(&dkey, (void *)obj->name, strlen(obj->name)); - /** set akey as the xattr name */ d_iov_set(&iod.iod_name, xname, strlen(xname)); iod.iod_nr = 1; @@ -175,35 +256,86 @@ dfs_getxattr(dfs_t *dfs, dfs_obj_t *obj, const char *name, void *value, daos_siz if (value != NULL && *size) { iod.iod_size = *size; - - /** set sgl for fetch */ - if (value == NULL) - d_iov_set(&sg_iov, NULL, 0); - else - d_iov_set(&sg_iov, value, *size); + d_iov_set(&sg_iov, value, *size); sgl.sg_nr = 1; sgl.sg_nr_out = 0; sgl.sg_iovs = &sg_iov; - - rc = daos_obj_fetch(oh, dfs->th, DAOS_COND_AKEY_FETCH, &dkey, 1, &iod, &sgl, NULL, - NULL); } else { iod.iod_size = DAOS_REC_ANY; - - rc = daos_obj_fetch(oh, dfs->th, DAOS_COND_AKEY_FETCH, &dkey, 1, &iod, NULL, NULL, - NULL); } - if (rc) { - DL_CDEBUG(rc == -DER_NONEXIST, DLOG_DBG, DLOG_ERR, rc, "Failed to fetch xattr '%s'", - name); - D_GOTO(close, rc = daos_der2errno(rc)); + +retry: + if (DFS_IS_HARDLINK(obj->mode)) { + /** For hardlinks, xattrs are stored in GIT with OID as dkey */ + if (!daos_handle_is_valid(dfs->git_oh)) + D_GOTO(out, rc = ENOTSUP); + d_iov_set(&dkey, &obj->oid, sizeof(daos_obj_id_t)); + + if (value != NULL && *size) + rc = daos_obj_fetch(dfs->git_oh, dfs->th, DAOS_COND_AKEY_FETCH, &dkey, 1, + &iod, &sgl, NULL, NULL); + else + rc = daos_obj_fetch(dfs->git_oh, dfs->th, DAOS_COND_AKEY_FETCH, &dkey, 1, + &iod, NULL, NULL, NULL); + if (rc) { + DL_CDEBUG(rc == -DER_NONEXIST, DLOG_DBG, DLOG_ERR, rc, + "Failed to fetch xattr '%s' from GIT", name); + D_GOTO(out, rc = daos_der2errno(rc)); + } + } else { + /** Open parent object and get xattr from the entry of the object */ + rc = daos_obj_open(dfs->coh, obj->parent_oid, DAOS_OO_RO, &oh, NULL); + if (rc) + D_GOTO(out, rc = daos_der2errno(rc)); + + /** set dkey as the entry name */ + d_iov_set(&dkey, (void *)obj->name, strlen(obj->name)); + + if (value != NULL && *size) + rc = daos_obj_fetch(oh, dfs->th, DAOS_COND_AKEY_FETCH, &dkey, 1, &iod, &sgl, + NULL, NULL); + else + rc = daos_obj_fetch(oh, dfs->th, DAOS_COND_AKEY_FETCH, &dkey, 1, &iod, NULL, + NULL, NULL); + + if (rc == -DER_NONEXIST && S_ISREG(obj->mode)) { + /* + * Since this function deals with dfs_obj, the hardlink information in its + * mode field may be outdated. If xattr name does not exist in dentry then + * check whether the file is already converted into a hardlink. If so, + * update the mode with hardlink info. Retry if the hardlink bit is set. + */ + int rc2 = + fetch_entry(dfs->layout_v, oh, dfs->th, obj->name, strlen(obj->name), + false, &exists, &entry, 0, NULL, NULL, NULL); + daos_obj_close(oh, NULL); + oh = DAOS_HDL_INVAL; + if (rc2) { + D_ERROR("Failed to fetch entry '%s' (%d)\n", obj->name, rc2); + D_GOTO(out, rc = rc2); + } + if (!exists) + D_GOTO(out, rc = daos_der2errno(rc)); + if (DFS_IS_HARDLINK(entry.mode)) { + /** set hardlink bit; retry from GIT */ + dfs_set_hardlink(&obj->mode); + goto retry; + } + /** genuine miss on a regular file: convert to POSIX errno */ + D_GOTO(out, rc = daos_der2errno(rc)); + } + if (rc) { + DL_CDEBUG(rc == -DER_NONEXIST, DLOG_DBG, DLOG_ERR, rc, + "Failed to fetch xattr '%s'", name); + daos_obj_close(oh, NULL); + D_GOTO(out, rc = daos_der2errno(rc)); + } + daos_obj_close(oh, NULL); } *size = iod.iod_size; DFS_OP_STAT_INCR(dfs, DOS_GETXATTR); -close: - daos_obj_close(oh, NULL); out: D_FREE(xname); if (rc == ENOENT) @@ -214,17 +346,18 @@ dfs_getxattr(dfs_t *dfs, dfs_obj_t *obj, const char *name, void *value, daos_siz int dfs_removexattr(dfs_t *dfs, dfs_obj_t *obj, const char *name) { - char *xname = NULL; - daos_handle_t th = DAOS_TX_NONE; - daos_key_t dkey, akey; - daos_handle_t oh; - uint64_t cond = 0; - d_sg_list_t sgl; - d_iov_t sg_iovs[2]; - daos_iod_t iod; - daos_recx_t recxs[2]; - struct timespec now; - int rc; + char *xname = NULL; + daos_key_t dkey, akey; + daos_handle_t oh = DAOS_HDL_INVAL; + uint64_t cond; + d_sg_list_t sgl; + d_iov_t sg_iovs[2]; + daos_iod_t iod; + daos_recx_t recxs[2]; + struct timespec now; + struct dfs_entry entry = {0}; + bool exists; + int rc; if (dfs == NULL || !dfs->mounted) return EINVAL; @@ -241,25 +374,10 @@ dfs_removexattr(dfs_t *dfs, dfs_obj_t *obj, const char *name) if (xname == NULL) return ENOMEM; - /** Open parent object and remove xattr from the entry of the object */ - rc = daos_obj_open(dfs->coh, obj->parent_oid, DAOS_OO_RW, &oh, NULL); - if (rc) - D_GOTO(free, rc = daos_der2errno(rc)); - - /** set dkey as the entry name */ - d_iov_set(&dkey, (void *)obj->name, strlen(obj->name)); /** set akey as the xattr name */ d_iov_set(&akey, xname, strlen(xname)); - cond = DAOS_COND_DKEY_UPDATE | DAOS_COND_PUNCH; - rc = daos_obj_punch_akeys(oh, th, cond, &dkey, 1, &akey, NULL); - if (rc) { - D_CDEBUG(rc == -DER_NONEXIST, DLOG_DBG, DLOG_ERR, - "Failed to punch extended attribute '%s'\n", name); - D_GOTO(out, rc = daos_der2errno(rc)); - } - - /** update ctime */ + /** setup iod/sgl for ctime update (shared by both paths) */ d_iov_set(&iod.iod_name, INODE_AKEY_NAME, sizeof(INODE_AKEY_NAME) - 1); iod.iod_recxs = recxs; iod.iod_type = DAOS_IOD_ARRAY; @@ -269,25 +387,103 @@ dfs_removexattr(dfs_t *dfs, dfs_obj_t *obj, const char *name) recxs[0].rx_nr = sizeof(uint64_t); recxs[1].rx_idx = CTIME_NSEC_IDX; recxs[1].rx_nr = sizeof(uint64_t); - rc = clock_gettime(CLOCK_REALTIME, &now); - if (rc) - D_GOTO(out, rc = errno); - sgl.sg_nr = 2; - sgl.sg_nr_out = 0; - sgl.sg_iovs = &sg_iovs[0]; - d_iov_set(&sg_iovs[0], &now.tv_sec, sizeof(uint64_t)); - d_iov_set(&sg_iovs[1], &now.tv_nsec, sizeof(uint64_t)); - - rc = daos_obj_update(oh, th, DAOS_COND_DKEY_UPDATE, &dkey, 1, &iod, &sgl, NULL); - if (rc) { - D_ERROR("Failed to update mode, " DF_RC "\n", DP_RC(rc)); - D_GOTO(out, rc = daos_der2errno(rc)); + sgl.sg_nr = 2; + sgl.sg_nr_out = 0; + sgl.sg_iovs = &sg_iovs[0]; + +retry: + cond = DAOS_COND_DKEY_UPDATE | DAOS_COND_PUNCH; + + if (DFS_IS_HARDLINK(obj->mode)) { + /** For hardlinks, xattrs are stored in GIT with OID as dkey */ + if (!daos_handle_is_valid(dfs->git_oh)) + D_GOTO(out_free, rc = ENOTSUP); + d_iov_set(&dkey, &obj->oid, sizeof(daos_obj_id_t)); + + rc = daos_obj_punch_akeys(dfs->git_oh, DAOS_TX_NONE, cond, &dkey, 1, &akey, NULL); + if (rc) { + D_CDEBUG(rc == -DER_NONEXIST, DLOG_DBG, DLOG_ERR, + "Failed to punch extended attribute '%s' from GIT\n", name); + D_GOTO(out_free, rc = daos_der2errno(rc)); + } + + rc = clock_gettime(CLOCK_REALTIME, &now); + if (rc) + D_GOTO(out_free, rc = errno); + d_iov_set(&sg_iovs[0], &now.tv_sec, sizeof(uint64_t)); + d_iov_set(&sg_iovs[1], &now.tv_nsec, sizeof(uint64_t)); + + /** update ctime in GIT */ + rc = daos_obj_update(dfs->git_oh, DAOS_TX_NONE, DAOS_COND_DKEY_UPDATE, &dkey, 1, + &iod, &sgl, NULL); + if (rc) { + D_ERROR("Failed to update ctime in GIT, " DF_RC "\n", DP_RC(rc)); + D_GOTO(out_free, rc = daos_der2errno(rc)); + } + } else { + /** Open parent object and remove xattr from the entry of the object */ + rc = daos_obj_open(dfs->coh, obj->parent_oid, DAOS_OO_RW, &oh, NULL); + if (rc) + D_GOTO(out_free, rc = daos_der2errno(rc)); + + /** set dkey as the entry name */ + d_iov_set(&dkey, (void *)obj->name, strlen(obj->name)); + + rc = daos_obj_punch_akeys(oh, DAOS_TX_NONE, cond, &dkey, 1, &akey, NULL); + if (rc == -DER_NONEXIST && S_ISREG(obj->mode)) { + /* + * Since this function deals with dfs_obj, the hardlink information in its + * mode field may be outdated. If xattr name does not exist in dentry then + * check whether the file is already converted into a hardlink. If so, + * update the mode with hardlink info. Retry if the hardlink bit is set. + */ + int rc2 = fetch_entry(dfs->layout_v, oh, DAOS_TX_NONE, obj->name, + strlen(obj->name), false, &exists, &entry, 0, NULL, + NULL, NULL); + daos_obj_close(oh, NULL); + oh = DAOS_HDL_INVAL; + if (rc2) { + D_ERROR("Failed to fetch entry '%s' (%d)\n", obj->name, rc2); + D_GOTO(out_free, rc = rc2); + } + if (!exists) + D_GOTO(out_free, rc = daos_der2errno(rc)); + if (DFS_IS_HARDLINK(entry.mode)) { + /** entry became a hardlink; retry from GIT */ + dfs_set_hardlink(&obj->mode); + goto retry; + } + /** genuine miss on a regular file: convert to POSIX errno */ + D_GOTO(out_free, rc = daos_der2errno(rc)); + } + if (rc) { + D_CDEBUG(rc == -DER_NONEXIST, DLOG_DBG, DLOG_ERR, + "Failed to punch extended attribute '%s'\n", name); + daos_obj_close(oh, NULL); + D_GOTO(out_free, rc = daos_der2errno(rc)); + } + + rc = clock_gettime(CLOCK_REALTIME, &now); + if (rc) { + daos_obj_close(oh, NULL); + D_GOTO(out_free, rc = errno); + } + d_iov_set(&sg_iovs[0], &now.tv_sec, sizeof(uint64_t)); + d_iov_set(&sg_iovs[1], &now.tv_nsec, sizeof(uint64_t)); + + /** update ctime */ + rc = daos_obj_update(oh, DAOS_TX_NONE, DAOS_COND_DKEY_UPDATE, &dkey, 1, &iod, &sgl, + NULL); + if (rc) { + D_ERROR("Failed to update ctime, " DF_RC "\n", DP_RC(rc)); + daos_obj_close(oh, NULL); + D_GOTO(out_free, rc = daos_der2errno(rc)); + } + daos_obj_close(oh, NULL); } DFS_OP_STAT_INCR(dfs, DOS_RMXATTR); -out: - daos_obj_close(oh, NULL); -free: +out_free: D_FREE(xname); return rc; } @@ -295,13 +491,17 @@ dfs_removexattr(dfs_t *dfs, dfs_obj_t *obj, const char *name) int dfs_listxattr(dfs_t *dfs, dfs_obj_t *obj, char *list, daos_size_t *size) { - daos_key_t dkey; - daos_handle_t oh; - daos_key_desc_t kds[ENUM_DESC_NR]; - daos_anchor_t anchor = {0}; - daos_size_t list_size, ret_size; - char *ptr_list; - int rc; + daos_key_t dkey; + daos_handle_t oh = DAOS_HDL_INVAL; + daos_handle_t list_oh; + daos_key_desc_t kds[ENUM_DESC_NR]; + daos_anchor_t anchor = {0}; + daos_size_t list_size, ret_size; + char *ptr_list; + struct dfs_entry entry = {0}; + bool exists; + bool is_hardlink; + int rc = 0; if (dfs == NULL || !dfs->mounted) return EINVAL; @@ -310,17 +510,28 @@ dfs_listxattr(dfs_t *dfs, dfs_obj_t *obj, char *list, daos_size_t *size) if (size == NULL) return EINVAL; - /** Open parent object and list from entry */ - rc = daos_obj_open(dfs->coh, obj->parent_oid, DAOS_OO_RW, &oh, NULL); - if (rc) - return daos_der2errno(rc); - - /** set dkey as the entry name */ - d_iov_set(&dkey, (void *)obj->name, strlen(obj->name)); - - list_size = *size; - ret_size = 0; - ptr_list = list; +retry: + list_size = *size; + ret_size = 0; + ptr_list = list; + is_hardlink = DFS_IS_HARDLINK(obj->mode); + memset(&anchor, 0, sizeof(anchor)); + + if (is_hardlink) { + /** For hardlinks, xattrs are stored in GIT with OID as dkey */ + if (!daos_handle_is_valid(dfs->git_oh)) + return ENOTSUP; + d_iov_set(&dkey, &obj->oid, sizeof(daos_obj_id_t)); + list_oh = dfs->git_oh; + } else { + /** Open parent object and list from entry */ + rc = daos_obj_open(dfs->coh, obj->parent_oid, DAOS_OO_RO, &oh, NULL); + if (rc) + return daos_der2errno(rc); + /** set dkey as the entry name */ + d_iov_set(&dkey, (void *)obj->name, strlen(obj->name)); + list_oh = oh; + } while (!daos_anchor_is_eof(&anchor)) { uint32_t number = ENUM_DESC_NR; @@ -332,12 +543,15 @@ dfs_listxattr(dfs_t *dfs, dfs_obj_t *obj, char *list, daos_size_t *size) sgl.sg_nr = 1; sgl.sg_nr_out = 0; - d_iov_set(&iov, enum_buf, ENUM_DESC_BUF); + d_iov_set(&iov, enum_buf, ENUM_XDESC_BUF); sgl.sg_iovs = &iov; - rc = daos_obj_list_akey(oh, dfs->th, &dkey, &number, kds, &sgl, &anchor, NULL); - if (rc) + rc = daos_obj_list_akey(list_oh, dfs->th, &dkey, &number, kds, &sgl, &anchor, NULL); + if (rc) { + if (!is_hardlink) + daos_obj_close(oh, NULL); D_GOTO(out, rc = daos_der2errno(rc)); + } if (number == 0) continue; @@ -367,9 +581,32 @@ dfs_listxattr(dfs_t *dfs, dfs_obj_t *obj, char *list, daos_size_t *size) } } + /* + * Since this function deals with dfs_obj, the hardlink information in its + * mode field may be outdated. If list xattr does not return any xattr name, + * check and update the mode with hardlink info. Retry if the hardlink bit is set. + */ + if (!ret_size && !is_hardlink && S_ISREG(obj->mode)) { + rc = fetch_entry(dfs->layout_v, oh, dfs->th, obj->name, strlen(obj->name), false, + &exists, &entry, 0, NULL, NULL, NULL); + daos_obj_close(oh, NULL); + oh = DAOS_HDL_INVAL; + if (rc) { + D_ERROR("Failed to fetch entry '%s' (%d)\n", obj->name, rc); + D_GOTO(out, rc); + } + if (exists && DFS_IS_HARDLINK(entry.mode)) { + /** set hardlink bit; retry listing from GIT */ + dfs_set_hardlink(&obj->mode); + goto retry; + } + } else if (!is_hardlink) { + daos_obj_close(oh, NULL); + oh = DAOS_HDL_INVAL; + } + *size = ret_size; DFS_OP_STAT_INCR(dfs, DOS_LSXATTR); out: - daos_obj_close(oh, NULL); return rc; } diff --git a/src/tests/suite/dfs_unit_test.c b/src/tests/suite/dfs_unit_test.c index 0d3ed97989a..2f5ffca47e1 100644 --- a/src/tests/suite/dfs_unit_test.c +++ b/src/tests/suite/dfs_unit_test.c @@ -12,6 +12,7 @@ #include #include #include +#include /** global DFS mount used for all tests */ static uuid_t co_uuid; @@ -65,6 +66,20 @@ test_akey_present(dfs_obj_t *parent_dir, const char *file_name, const char *akey return true; } +/** Scan a NULL-terminated xattr name list (as returned by dfs_listxattr) for a name. */ +static bool +xattr_name_in_list(const char *list, daos_size_t size, const char *name) +{ + const char *p = list; + + while (p < list + size) { + if (strcmp(p, name) == 0) + return true; + p += strlen(p) + 1; + } + return false; +} + static void dfs_test_mount(void **state) { @@ -4104,6 +4119,242 @@ dfs_test_link_remove(void **state) assert_int_equal(rc, 0); } +/** + * Test xattr operations across hardlinks. After a regular file is converted to a + * hardlink, its xattrs migrate to the GIT object and are keyed by OID, so all + * links must observe a single shared set of xattrs. This test also exercises the + * "stale mode" path: handles opened while the object was a regular file must + * transparently retry against GIT once the on-disk entry becomes a hardlink. + * + * Note: the DTX path in dfs_setxattr is covered automatically because the whole + * suite is re-run with DFS_USE_DTX=1 (see run_dfs_unit_test()). + * + * Steps: + * 1. Create regular file f and three extra handles (hA write-source, hB, hC). + * 2. Set k1=v1, k2=v2 and an empty-value k3 while f is a regular file. + * 3. getxattr size-probe (value=NULL) then sized fetch; verify empty k3. + * 4. listxattr on the regular file returns {k1,k2,k3}; size-probe (list=NULL). + * 5. Convert f to a hardlink via dfs_link (l2_obj returned, nlink=2) and verify + * the x: akeys were removed from the parent dentry (migrated to GIT). + * 6. get/list via stale handles (hB, hC, hA) and the new l2_obj -> shared data. + * 7. set k4 via one link, get via another -> visible (cross-link write). + * 8. overwrite k1 via one link, get via another -> last writer wins. + * 9. XATTR_CREATE/XATTR_REPLACE flag semantics on a hardlink. + * 10. remove k4 via one link, get via another -> ENODATA. + * 11. Negative: get a missing xattr on a hardlink -> ENODATA; remove -> ENOENT. + * 12. Persistence: release l2_obj, reopen by path, value still present. + * 13. Remove all xattrs; listxattr on a hardlink with no entries -> size 0. + * 14. After all links are removed (inode destroyed), a get on a surviving handle + * returns ENODATA instead of stale data. + */ +static void +dfs_test_xattr_hardlink(void **state) +{ + test_arg_t *arg = *state; + dfs_obj_t *dir1; + dfs_obj_t *hA, *hB, *hC, *l2_obj, *reopen_obj; + const char *k1 = "user.k1", *k2 = "user.k2", *k3 = "user.k3", *k4 = "user.k4"; + const char *v1 = "v1", *v2 = "v2", *v4 = "v4", *v1b = "v1_overwritten"; + char xakey[64]; + char buf[64]; + char list[128]; + daos_size_t size; + struct stat statbuf; + int rc; + + if (arg->myrank != 0) + return; + + /** Step 1: create directory and a regular file with multiple open handles. */ + rc = dfs_open(dfs_mt, NULL, "xhl_dir1", S_IFDIR | S_IWUSR | S_IRUSR, + O_RDWR | O_CREAT | O_EXCL, 0, 0, NULL, &dir1); + assert_int_equal(rc, 0); + rc = dfs_open(dfs_mt, dir1, "file1", S_IFREG | S_IWUSR | S_IRUSR, O_RDWR | O_CREAT | O_EXCL, + 0, 0, NULL, &hA); + assert_int_equal(rc, 0); + rc = dfs_open(dfs_mt, dir1, "file1", S_IFREG, O_RDWR, 0, 0, NULL, &hB); + assert_int_equal(rc, 0); + rc = dfs_open(dfs_mt, dir1, "file1", S_IFREG, O_RDWR, 0, 0, NULL, &hC); + assert_int_equal(rc, 0); + + /** Step 2: set two valued xattrs and one empty-value xattr on the regular file. */ + print_message("Step 1-2: set k1, k2 and empty k3 on the regular file\n"); + rc = dfs_setxattr(dfs_mt, hA, k1, v1, strlen(v1) + 1, 0); + assert_int_equal(rc, 0); + rc = dfs_setxattr(dfs_mt, hA, k2, v2, strlen(v2) + 1, 0); + assert_int_equal(rc, 0); + rc = dfs_setxattr(dfs_mt, hA, k3, NULL, 0, 0); + assert_int_equal(rc, 0); + + /** Step 3: getxattr size-probe then sized fetch; verify empty xattr. */ + print_message("Step 3: getxattr size-probe and sized fetch\n"); + size = 0; + rc = dfs_getxattr(dfs_mt, hA, k1, NULL, &size); + assert_int_equal(rc, 0); + assert_int_equal(size, strlen(v1) + 1); + memset(buf, 0, sizeof(buf)); + rc = dfs_getxattr(dfs_mt, hA, k1, buf, &size); + assert_int_equal(rc, 0); + assert_string_equal(buf, v1); + size = sizeof(buf); + rc = dfs_getxattr(dfs_mt, hA, k3, buf, &size); + assert_int_equal(rc, 0); + assert_int_equal(size, 0); + + /** Step 4: list on the regular file returns all three names; size-probe too. */ + print_message("Step 4: listxattr on the regular file\n"); + size = 0; + rc = dfs_listxattr(dfs_mt, hA, NULL, &size); + assert_int_equal(rc, 0); + assert_int_equal(size, strlen(k1) + 1 + strlen(k2) + 1 + strlen(k3) + 1); + size = sizeof(list); + memset(list, 0, sizeof(list)); + rc = dfs_listxattr(dfs_mt, hA, list, &size); + assert_int_equal(rc, 0); + assert_true(xattr_name_in_list(list, size, k1)); + assert_true(xattr_name_in_list(list, size, k2)); + assert_true(xattr_name_in_list(list, size, k3)); + + /** Step 5: convert to hardlink and confirm xattrs migrated out of the dentry. */ + print_message("Step 5: convert file1 to a hardlink and verify dentry akeys cleared\n"); + l2_obj = NULL; + rc = dfs_link(dfs_mt, hA, dir1, "file1_l2", &l2_obj, &statbuf); + assert_int_equal(rc, 0); + assert_non_null(l2_obj); + assert_int_equal((int)statbuf.st_nlink, 2); + snprintf(xakey, sizeof(xakey), "x:%s", k1); + assert_false(test_akey_present(dir1, "file1", xakey)); + snprintf(xakey, sizeof(xakey), "x:%s", k2); + assert_false(test_akey_present(dir1, "file1", xakey)); + snprintf(xakey, sizeof(xakey), "x:%s", k3); + assert_false(test_akey_present(dir1, "file1", xakey)); + + /** Step 6: stale handles (hB, hC, hA) and the new link see the shared xattrs. */ + print_message("Step 6: get/list via stale handles and the new link\n"); + size = sizeof(buf); + memset(buf, 0, sizeof(buf)); + rc = dfs_getxattr(dfs_mt, hB, k1, buf, &size); + assert_int_equal(rc, 0); + assert_string_equal(buf, v1); + size = sizeof(buf); + memset(buf, 0, sizeof(buf)); + rc = dfs_getxattr(dfs_mt, l2_obj, k1, buf, &size); + assert_int_equal(rc, 0); + assert_string_equal(buf, v1); + size = sizeof(buf); + memset(buf, 0, sizeof(buf)); + rc = dfs_getxattr(dfs_mt, hA, k1, buf, &size); + assert_int_equal(rc, 0); + assert_string_equal(buf, v1); + size = sizeof(list); + memset(list, 0, sizeof(list)); + rc = dfs_listxattr(dfs_mt, hC, list, &size); + assert_int_equal(rc, 0); + assert_true(xattr_name_in_list(list, size, k1)); + assert_true(xattr_name_in_list(list, size, k2)); + assert_true(xattr_name_in_list(list, size, k3)); + + /** Step 7: write a new xattr via one link, read it via another. */ + print_message("Step 7: set k4 via l2_obj, get via hA\n"); + rc = dfs_setxattr(dfs_mt, l2_obj, k4, v4, strlen(v4) + 1, 0); + assert_int_equal(rc, 0); + size = sizeof(buf); + memset(buf, 0, sizeof(buf)); + rc = dfs_getxattr(dfs_mt, hA, k4, buf, &size); + assert_int_equal(rc, 0); + assert_string_equal(buf, v4); + + /** Step 8: overwrite an existing xattr via one link, read newest via another. */ + print_message("Step 8: overwrite k1 via hB, get via l2_obj (last writer wins)\n"); + rc = dfs_setxattr(dfs_mt, hB, k1, v1b, strlen(v1b) + 1, 0); + assert_int_equal(rc, 0); + size = sizeof(buf); + memset(buf, 0, sizeof(buf)); + rc = dfs_getxattr(dfs_mt, l2_obj, k1, buf, &size); + assert_int_equal(rc, 0); + assert_string_equal(buf, v1b); + + /** Step 9: XATTR_CREATE/XATTR_REPLACE flag semantics on a hardlink. */ +#if 0 + /** Bug: DAOS-19258 + * Currently masked because of the above bug. + */ + print_message("Step 9: XATTR_CREATE / XATTR_REPLACE semantics on a hardlink\n"); + rc = dfs_setxattr(dfs_mt, l2_obj, k4, v4, strlen(v4) + 1, XATTR_CREATE); + assert_int_equal(rc, EEXIST); + rc = dfs_setxattr(dfs_mt, l2_obj, "user.missing", v4, strlen(v4) + 1, XATTR_REPLACE); + assert_int_equal(rc, ENODATA); +#endif + rc = dfs_setxattr(dfs_mt, l2_obj, k1, v1, strlen(v1) + 1, XATTR_REPLACE); + assert_int_equal(rc, 0); + + /** Step 10: remove a xattr via one link, confirm it is gone from another. */ + print_message("Step 10: remove k4 via hB, get via hC -> ENODATA\n"); + rc = dfs_removexattr(dfs_mt, hB, k4); + assert_int_equal(rc, 0); + size = sizeof(buf); + rc = dfs_getxattr(dfs_mt, hC, k4, buf, &size); + assert_int_equal(rc, ENODATA); + + /** Step 11: missing-xattr negatives on a hardlink. */ + print_message( + "Step 11: get a missing xattr -> ENODATA, remove a missing xattr -> ENOENT\n"); + size = sizeof(buf); + rc = dfs_getxattr(dfs_mt, l2_obj, "user.none", buf, &size); + assert_int_equal(rc, ENODATA); + rc = dfs_removexattr(dfs_mt, l2_obj, "user.none"); + assert_int_equal(rc, ENOENT); + + /** Step 12: persistence across a fresh handle opened by path. */ + print_message("Step 12: reopen by path and verify xattr persists\n"); + rc = dfs_release(l2_obj); + assert_int_equal(rc, 0); + rc = dfs_open(dfs_mt, dir1, "file1_l2", S_IFREG, O_RDWR, 0, 0, NULL, &reopen_obj); + assert_int_equal(rc, 0); + size = sizeof(buf); + memset(buf, 0, sizeof(buf)); + rc = dfs_getxattr(dfs_mt, reopen_obj, k1, buf, &size); + assert_int_equal(rc, 0); + assert_string_equal(buf, v1); + + /** Step 13: remove all xattrs, then list on a hardlink with no entries. */ + print_message("Step 13: remove all xattrs and verify empty listxattr\n"); + rc = dfs_removexattr(dfs_mt, reopen_obj, k1); + assert_int_equal(rc, 0); + rc = dfs_removexattr(dfs_mt, hA, k2); + assert_int_equal(rc, 0); + rc = dfs_removexattr(dfs_mt, hB, k3); + assert_int_equal(rc, 0); + size = sizeof(list); + rc = dfs_listxattr(dfs_mt, reopen_obj, list, &size); + assert_int_equal(rc, 0); + assert_int_equal(size, 0); + rc = dfs_release(reopen_obj); + assert_int_equal(rc, 0); + + /** Step 14: after all links are removed the inode is gone; get -> ENODATA. */ + print_message("Step 14: handle after full unlink -> ENODATA\n"); + rc = dfs_remove(dfs_mt, dir1, "file1_l2", 0, NULL); + assert_int_equal(rc, 0); + rc = dfs_remove(dfs_mt, dir1, "file1", 0, NULL); + assert_int_equal(rc, 0); + size = sizeof(buf); + rc = dfs_getxattr(dfs_mt, hB, "user.none", buf, &size); + assert_int_equal(rc, ENODATA); + + /** Cleanup */ + rc = dfs_release(hA); + assert_int_equal(rc, 0); + rc = dfs_release(hB); + assert_int_equal(rc, 0); + rc = dfs_release(hC); + assert_int_equal(rc, 0); + rc = dfs_release(dir1); + assert_int_equal(rc, 0); + rc = dfs_remove(dfs_mt, NULL, "xhl_dir1", true, NULL); + assert_int_equal(rc, 0); +} + static const struct CMUnitTest dfs_unit_tests[] = { {"DFS_UNIT_TEST1: DFS mount / umount", dfs_test_mount, async_disable, test_case_teardown}, {"DFS_UNIT_TEST2: DFS container modes", dfs_test_modes, async_disable, test_case_teardown}, @@ -4146,6 +4397,8 @@ static const struct CMUnitTest dfs_unit_tests[] = { {"DFS_UNIT_TEST28: dfs open/lookup flags", dfs_test_oflags, async_disable, test_case_teardown}, {"DFS_UNIT_TEST29: dfs hard link and remove", dfs_test_link_remove, async_disable, test_case_teardown}, + {"DFS_UNIT_TEST30: dfs xattr hardlink", dfs_test_xattr_hardlink, async_disable, + test_case_teardown}, }; static int