return osp->opd_storage->dd_lu_dev.ld_site->ld_seq_site->ss_node_id;
}
-static inline const char *osp_dto2name(struct osp_object *obj)
-{
- return obj->opo_obj.do_lu.lo_dev->ld_obd->obd_name;
-}
-
static inline bool is_ost_obj(struct lu_object *lo)
{
return !lu2osp_dev(lo->lo_dev)->opd_connect_mdt;
}
-static inline void __osp_oac_xattr_assignment(struct osp_object *obj,
- struct osp_xattr_entry *oxe,
- const struct lu_buf *buf)
-{
- if (buf->lb_len > 0)
- memcpy(oxe->oxe_value, buf->lb_buf, buf->lb_len);
-
- oxe->oxe_vallen = buf->lb_len;
- oxe->oxe_exist = 1;
- oxe->oxe_ready = 1;
-}
-
/**
* Assign FID to the OST object.
*
lu_object_assign_fid(env, &o->opo_obj.do_lu, &osi->osi_fid);
}
-#define OXE_DEFAULT_LEN 16
-
-/**
- * Release reference from the OSP object extended attribute entry.
- *
- * If it is the last reference, then free the entry.
- *
- * \param[in] oxe pointer to the OSP object extended attribute entry.
- */
-static inline void osp_oac_xattr_put(struct osp_xattr_entry *oxe)
-{
- if (atomic_dec_and_test(&oxe->oxe_ref)) {
- LASSERT(list_empty(&oxe->oxe_list));
-
- OBD_FREE(oxe, oxe->oxe_buflen);
- }
-}
-
/**
* Find the named extended attribute in the OSP object attributes cache.
*
osp_oac_xattr_find_or_add(struct osp_object *obj, const char *name, size_t len)
{
struct osp_xattr_entry *oxe;
- struct osp_xattr_entry *tmp = NULL;
- size_t namelen = strlen(name);
- size_t size = sizeof(*oxe) + namelen + 1 +
- (len ? len : OXE_DEFAULT_LEN);
+ struct osp_xattr_entry *tmp = NULL;
+ size_t namelen = strlen(name);
+ size_t size = sizeof(*oxe) + namelen + 1 + len;
oxe = osp_oac_xattr_find(obj, name, false);
if (oxe != NULL)
}
/**
- * Assign the cached OST-object's EA with the given value.
+ * Add the given extended attribute to the OSP object attributes cache.
*
- * If the current EA entry in cache has not enough space to hold the new
- * value, remove it, create a new one, then assign with the given value.
+ * If there is an old extended attributed entry with the same name,
+ * remove it from the cache and return it via the parameter \a poxe.
*
* \param[in] obj pointer to the OSP object
- * \param[in] oxe pointer to the cached EA entry to be assigned
- * \param[in] buf pointer to the buffer with new EA value
- *
- * \retval pointer to the new created EA entry in cache if
- * current entry is not big enough; otherwise, the
- * input 'oxe' will be returned.
+ * \param[in,out] poxe double pointer to the OSP object extended attribute
+ * entry: the new extended attribute entry is transferred
+ * via such pointer target, and if old the extended
+ * attribute entry exists, then it will be returned back
+ * via such pointer target.
+ * \param[in] len the length of the (new) extended attribute value
+ *
+ * \retval pointer to the new extended attribute entry
+ * \retval NULL for failure cases.
*/
static struct osp_xattr_entry *
-osp_oac_xattr_assignment(struct osp_object *obj, struct osp_xattr_entry *oxe,
- const struct lu_buf *buf)
+osp_oac_xattr_replace(struct osp_object *obj,
+ struct osp_xattr_entry **poxe, size_t len)
{
- struct osp_xattr_entry *new = NULL;
- struct osp_xattr_entry *old = NULL;
- int namelen = oxe->oxe_namelen;
- bool unlink_only = false;
+ struct osp_xattr_entry *oxe;
+ size_t namelen = (*poxe)->oxe_namelen;
+ size_t size = sizeof(*oxe) + namelen + 1 + len;
- if (oxe->oxe_buflen - oxe->oxe_namelen - 1 < buf->lb_len) {
- size_t size = sizeof(*oxe) + namelen + 1 + buf->lb_len;
-
- OBD_ALLOC(new, size);
- if (likely(new)) {
- INIT_LIST_HEAD(&new->oxe_list);
- new->oxe_buflen = size;
- new->oxe_namelen = namelen;
- memcpy(new->oxe_buf, oxe->oxe_buf, namelen);
- new->oxe_value = new->oxe_buf + namelen + 1;
- /* One ref is for the caller,
- * the other is for the entry on the list. */
- atomic_set(&new->oxe_ref, 2);
- __osp_oac_xattr_assignment(obj, new, buf);
- } else {
- unlink_only = true;
- CWARN("%s: cannot update cached xattr %.*s of "DFID"\n",
- osp_dto2name(obj), namelen, oxe->oxe_buf,
- PFID(lu_object_fid(&obj->opo_obj.do_lu)));
- }
- }
+ OBD_ALLOC(oxe, size);
+ if (unlikely(oxe == NULL))
+ return NULL;
- spin_lock(&obj->opo_lock);
- old = osp_oac_xattr_find_locked(obj, oxe->oxe_buf, namelen);
- if (likely(old)) {
- if (new) {
- /* Unlink the 'old'. */
- list_del_init(&old->oxe_list);
+ INIT_LIST_HEAD(&oxe->oxe_list);
+ oxe->oxe_buflen = size;
+ oxe->oxe_namelen = namelen;
+ memcpy(oxe->oxe_buf, (*poxe)->oxe_buf, namelen);
+ oxe->oxe_value = oxe->oxe_buf + namelen + 1;
+ /* One ref is for the caller, the other is for the entry on the list. */
+ atomic_set(&oxe->oxe_ref, 2);
- /* Drop the ref for 'old' on list. */
- osp_oac_xattr_put(old);
+ spin_lock(&obj->opo_lock);
+ *poxe = osp_oac_xattr_find_locked(obj, oxe->oxe_buf, namelen);
+ LASSERT(*poxe != NULL);
- /* Insert 'new' into list. */
- list_add_tail(&new->oxe_list, &obj->opo_xattr_list);
+ list_del_init(&(*poxe)->oxe_list);
+ list_add_tail(&oxe->oxe_list, &obj->opo_xattr_list);
+ spin_unlock(&obj->opo_lock);
- /* Drop the ref for current using. */
- osp_oac_xattr_put(oxe);
- oxe = new;
- } else if (unlink_only) {
- /* Unlink the 'old'. */
- list_del_init(&old->oxe_list);
+ return oxe;
+}
- /* Drop the ref for 'old' on list. */
- osp_oac_xattr_put(old);
- } else {
- __osp_oac_xattr_assignment(obj, oxe, buf);
- }
- } else if (new) {
- /* Someone unlinked the 'old' by race, need NOT to assign
- * for unlinked 'oxe', just add the 'new' one. */
- list_add_tail(&new->oxe_list, &obj->opo_xattr_list);
+/**
+ * Release reference from the OSP object extended attribute entry.
+ *
+ * If it is the last reference, then free the entry.
+ *
+ * \param[in] oxe pointer to the OSP object extended attribute entry.
+ */
+static inline void osp_oac_xattr_put(struct osp_xattr_entry *oxe)
+{
+ if (atomic_dec_and_test(&oxe->oxe_ref)) {
+ LASSERT(list_empty(&oxe->oxe_list));
- /* Drop the ref for current using. */
- osp_oac_xattr_put(oxe);
- oxe = new;
+ OBD_FREE(oxe, oxe->oxe_buflen);
}
- spin_unlock(&obj->opo_lock);
-
- return oxe;
}
/**
struct osp_xattr_entry *oxe = data;
struct lu_buf *rbuf = &osp_env_info(env)->osi_lb2;
- if (!rc) {
+ if (rc == 0) {
size_t len = sizeof(*oxe) + oxe->oxe_namelen + 1;
rc = object_update_result_data_get(reply, rbuf, index);
if (rc < 0 || rbuf->lb_len > (oxe->oxe_buflen - len)) {
spin_lock(&obj->opo_lock);
- if (unlikely(rc == -ENODATA)) {
- oxe->oxe_exist = 0;
- oxe->oxe_ready = 1;
- } else {
- oxe->oxe_ready = 0;
- }
+ oxe->oxe_ready = 0;
spin_unlock(&obj->opo_lock);
osp_oac_xattr_put(oxe);
}
spin_lock(&obj->opo_lock);
- __osp_oac_xattr_assignment(obj, oxe, rbuf);
+ oxe->oxe_vallen = rbuf->lb_len;
+ memcpy(oxe->oxe_value, rbuf->lb_buf, rbuf->lb_len);
+ oxe->oxe_exist = 1;
+ oxe->oxe_ready = 1;
spin_unlock(&obj->opo_lock);
} else if (rc == -ENOENT || rc == -ENODATA) {
spin_lock(&obj->opo_lock);
LASSERT(buf != NULL);
LASSERT(name != NULL);
+ namelen = strlen(name);
+
+ /* If only for xattr size, return directly. */
if (unlikely(buf->lb_len == 0))
- return -EINVAL;
+ return 0;
oxe = osp_oac_xattr_find_or_add(obj, name, buf->lb_len);
if (oxe == NULL)
return -ENOMEM;
- namelen = strlen(name);
mutex_lock(&osp->opd_async_requests_mutex);
rc = osp_insert_async_request(env, OUT_XATTR_GET, obj, 1,
&namelen, (const void **)&name,
struct ptlrpc_request *req = NULL;
struct object_update_reply *reply;
struct osp_xattr_entry *oxe = NULL;
- const char *dname = osp_dto2name(obj);
- int rc = 0;
+ const char *dname = dt->do_lu.lo_dev->ld_obd->obd_name;
+ int rc = 0;
ENTRY;
LASSERT(buf != NULL);
if (rc < 0)
GOTO(out, rc);
- /* The peer should have converted '0' as '-ENODATA'. */
- if (unlikely(!rc))
- GOTO(out, rc = -EINVAL);
-
- /* For detecting EA size. */
- if (!buf->lb_buf)
+ if (buf->lb_buf == NULL)
GOTO(out, rc);
- if (!oxe) {
+ if (unlikely(buf->lb_len < rbuf->lb_len))
+ GOTO(out, rc = -ERANGE);
+
+ memcpy(buf->lb_buf, rbuf->lb_buf, rbuf->lb_len);
+
+ if (oxe == NULL) {
oxe = osp_oac_xattr_find_or_add(obj, name, rbuf->lb_len);
if (oxe == NULL) {
CWARN("%s: Fail to add xattr (%s) to "
}
}
- oxe = osp_oac_xattr_assignment(obj, oxe, rbuf);
+ if (oxe->oxe_buflen - oxe->oxe_namelen - 1 < rbuf->lb_len) {
+ struct osp_xattr_entry *old = oxe;
+ struct osp_xattr_entry *tmp;
- GOTO(out, rc);
+ tmp = osp_oac_xattr_replace(obj, &old, rbuf->lb_len);
+ osp_oac_xattr_put(oxe);
+ oxe = tmp;
+ if (tmp == NULL) {
+ CWARN("%s: Fail to update xattr (%s) to "
+ "cache for "DFID": rc = %d\n",
+ dname, name, PFID(lu_object_fid(&dt->do_lu)), rc);
+ spin_lock(&obj->opo_lock);
+ old->oxe_ready = 0;
+ spin_unlock(&obj->opo_lock);
-out:
- if (rc > 0 && buf->lb_buf) {
- if (unlikely(buf->lb_len < rbuf->lb_len))
- rc = -ERANGE;
- else
- memcpy(buf->lb_buf, rbuf->lb_buf, rbuf->lb_len);
+ GOTO(out, rc);
+ }
+
+ /* Drop the ref for entry on list. */
+ osp_oac_xattr_put(old);
}
+ spin_lock(&obj->opo_lock);
+ oxe->oxe_vallen = rbuf->lb_len;
+ memcpy(oxe->oxe_value, rbuf->lb_buf, rbuf->lb_len);
+ oxe->oxe_exist = 1;
+ oxe->oxe_ready = 1;
+ spin_unlock(&obj->opo_lock);
+
+ GOTO(out, rc);
+
+out:
if (req != NULL)
ptlrpc_req_finished(req);
oxe = osp_oac_xattr_find_or_add(o, name, buf->lb_len);
if (oxe == NULL) {
CWARN("%s: cannot cache xattr '%s' of "DFID"\n",
- osp_dto2name(o), name, PFID(lu_object_fid(&dt->do_lu)));
+ dt->do_lu.lo_dev->ld_obd->obd_name,
+ name, PFID(lu_object_fid(&dt->do_lu)));
RETURN(0);
}
- oxe = osp_oac_xattr_assignment(o, oxe, buf);
- if (oxe)
+ if (oxe->oxe_buflen - oxe->oxe_namelen - 1 < buf->lb_len) {
+ struct osp_xattr_entry *old = oxe;
+ struct osp_xattr_entry *tmp;
+
+ tmp = osp_oac_xattr_replace(o, &old, buf->lb_len);
osp_oac_xattr_put(oxe);
+ oxe = tmp;
+ if (tmp == NULL) {
+ CWARN("%s: cannot update cached xattr '%s' of "DFID"\n",
+ dt->do_lu.lo_dev->ld_obd->obd_name,
+ name, PFID(lu_object_fid(&dt->do_lu)));
+ spin_lock(&o->opo_lock);
+ old->oxe_ready = 0;
+ spin_unlock(&o->opo_lock);
+
+ RETURN(0);
+ }
+
+ /* Drop the ref for entry on list. */
+ osp_oac_xattr_put(old);
+ }
+
+ spin_lock(&o->opo_lock);
+ oxe->oxe_vallen = buf->lb_len;
+ memcpy(oxe->oxe_value, buf->lb_buf, buf->lb_len);
+ oxe->oxe_exist = 1;
+ oxe->oxe_ready = 1;
+ spin_unlock(&o->opo_lock);
+ osp_oac_xattr_put(oxe);
RETURN(0);
}
git://git.whamcloud.com / fs / lustre-release.git / commitdiff