Whamcloud - gitweb
LU-8840 osp: handle EA cache properly
[fs/lustre-release.git] / lustre / osp / osp_object.c
index 1fcbac7..0781eb9 100644 (file)
@@ -126,11 +126,28 @@ static inline __u32 osp_dev2node(struct osp_device *osp)
        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.
  *
@@ -155,6 +172,24 @@ static void osp_object_assign_fid(const struct lu_env *env,
        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.
  *
@@ -235,9 +270,10 @@ static struct osp_xattr_entry *
 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;
+       struct osp_xattr_entry *tmp = NULL;
+       size_t namelen = strlen(name);
+       size_t size = sizeof(*oxe) + namelen + 1 +
+                     (len ? len : OXE_DEFAULT_LEN);
 
        oxe = osp_oac_xattr_find(obj, name, false);
        if (oxe != NULL)
@@ -272,67 +308,87 @@ osp_oac_xattr_find_or_add(struct osp_object *obj, const char *name, size_t len)
 }
 
 /**
- * Add the given extended attribute to the OSP object attributes cache.
+ * Assign the cached OST-object's EA 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.
+ * 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.
  *
  * \param[in] obj      pointer to the OSP object
- * \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.
+ * \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.
  */
 static struct osp_xattr_entry *
-osp_oac_xattr_replace(struct osp_object *obj,
-                     struct osp_xattr_entry **poxe, size_t len)
+osp_oac_xattr_assignment(struct osp_object *obj, struct osp_xattr_entry *oxe,
+                        const struct lu_buf *buf)
 {
-       struct osp_xattr_entry *oxe;
-       size_t                  namelen = (*poxe)->oxe_namelen;
-       size_t                  size    = sizeof(*oxe) + namelen + 1 + len;
+       struct osp_xattr_entry *new = NULL;
+       struct osp_xattr_entry *old = NULL;
+       int namelen = oxe->oxe_namelen;
+       bool unlink_only = false;
 
-       OBD_ALLOC(oxe, size);
-       if (unlikely(oxe == NULL))
-               return NULL;
-
-       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);
+       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)));
+               }
+       }
 
        spin_lock(&obj->opo_lock);
-       *poxe = osp_oac_xattr_find_locked(obj, oxe->oxe_buf, namelen);
-       LASSERT(*poxe != NULL);
+       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);
 
-       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 'old' on list. */
+                       osp_oac_xattr_put(old);
 
-       return oxe;
-}
+                       /* Insert 'new' into list. */
+                       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;
+               } else if (unlink_only) {
+                       /* Unlink the 'old'. */
+                       list_del_init(&old->oxe_list);
 
-               OBD_FREE(oxe, oxe->oxe_buflen);
+                       /* 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);
+
+               /* Drop the ref for current using. */
+               osp_oac_xattr_put(oxe);
+               oxe = new;
        }
+       spin_unlock(&obj->opo_lock);
+
+       return oxe;
 }
 
 /**
@@ -712,13 +768,18 @@ static int osp_xattr_get_interpterer(const struct lu_env *env,
        struct osp_xattr_entry  *oxe  = data;
        struct lu_buf           *rbuf = &osp_env_info(env)->osi_lb2;
 
-       if (rc == 0) {
+       if (!rc) {
                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);
-                       oxe->oxe_ready = 0;
+                       if (unlikely(rc == -ENODATA)) {
+                               oxe->oxe_exist = 0;
+                               oxe->oxe_ready = 1;
+                       } else {
+                               oxe->oxe_ready = 0;
+                       }
                        spin_unlock(&obj->opo_lock);
                        osp_oac_xattr_put(oxe);
 
@@ -726,10 +787,7 @@ static int osp_xattr_get_interpterer(const struct lu_env *env,
                }
 
                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;
+               __osp_oac_xattr_assignment(obj, oxe, rbuf);
                spin_unlock(&obj->opo_lock);
        } else if (rc == -ENOENT || rc == -ENODATA) {
                spin_lock(&obj->opo_lock);
@@ -777,16 +835,14 @@ static int osp_declare_xattr_get(const struct lu_env *env, struct dt_object *dt,
        LASSERT(buf != NULL);
        LASSERT(name != NULL);
 
-       namelen = strlen(name);
-
-       /* If only for xattr size, return directly. */
        if (unlikely(buf->lb_len == 0))
-               return 0;
+               return -EINVAL;
 
        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,
@@ -853,8 +909,8 @@ int osp_xattr_get(const struct lu_env *env, struct dt_object *dt,
        struct ptlrpc_request   *req    = NULL;
        struct object_update_reply *reply;
        struct osp_xattr_entry  *oxe    = NULL;
-       const char              *dname  = dt->do_lu.lo_dev->ld_obd->obd_name;
-       int                      rc     = 0;
+       const char *dname = osp_dto2name(obj);
+       int rc = 0;
        ENTRY;
 
        LASSERT(buf != NULL);
@@ -957,15 +1013,15 @@ unlock:
        if (rc < 0)
                GOTO(out, rc);
 
-       if (buf->lb_buf == NULL)
-               GOTO(out, rc);
-
-       if (unlikely(buf->lb_len < rbuf->lb_len))
-               GOTO(out, rc = -ERANGE);
+       /* The peer should have converted '0' as '-ENODATA'. */
+       if (unlikely(!rc))
+               GOTO(out, rc = -EINVAL);
 
-       memcpy(buf->lb_buf, rbuf->lb_buf, rbuf->lb_len);
+       /* For detecting EA size. */
+       if (!buf->lb_buf)
+               GOTO(out, rc);
 
-       if (oxe == NULL) {
+       if (!oxe) {
                oxe = osp_oac_xattr_find_or_add(obj, name, rbuf->lb_len);
                if (oxe == NULL) {
                        CWARN("%s: Fail to add xattr (%s) to "
@@ -976,38 +1032,18 @@ unlock:
                }
        }
 
-       if (oxe->oxe_buflen - oxe->oxe_namelen - 1 < rbuf->lb_len) {
-               struct osp_xattr_entry *old = oxe;
-               struct osp_xattr_entry *tmp;
-
-               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);
-
-                       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);
+       oxe = osp_oac_xattr_assignment(obj, oxe, rbuf);
 
        GOTO(out, rc);
 
 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);
+       }
+
        if (req != NULL)
                ptlrpc_req_finished(req);
 
@@ -1105,41 +1141,14 @@ int osp_xattr_set(const struct lu_env *env, struct dt_object *dt,
        oxe = osp_oac_xattr_find_or_add(o, name, buf->lb_len);
        if (oxe == NULL) {
                CWARN("%s: cannot cache xattr '%s' of "DFID"\n",
-                     dt->do_lu.lo_dev->ld_obd->obd_name,
-                     name, PFID(lu_object_fid(&dt->do_lu)));
+                     osp_dto2name(o), name, PFID(lu_object_fid(&dt->do_lu)));
 
                RETURN(0);
        }
 
-       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);
+       oxe = osp_oac_xattr_assignment(o, oxe, buf);
+       if (oxe)
                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);
 }