LU-13974 llog: check stale osp object

[fs/lustre-release.git] / lustre / osp / osp_md_object.c

diff --git a/lustre/osp/osp_md_object.c b/lustre/osp/osp_md_object.c
index fb1c055..2abaf95 100644 (file)
--- a/lustre/osp/osp_md_object.c
+++ b/lustre/osp/osp_md_object.c
@@ -363,7 +363,7 @@ int osp_md_attr_set(const struct lu_env *env, struct dt_object *dt,
  *
  * \param[in] env      execution environment
  * \param[in] dt       object to be locked
- * \param[in] role     lock role from MDD layer, see mdd_object_role().
+ * \param[in] role     lock role from MDD layer, see dt_object_role().
  */
 static void osp_md_read_lock(const struct lu_env *env, struct dt_object *dt,
                             unsigned role)
@@ -383,7 +383,7 @@ static void osp_md_read_lock(const struct lu_env *env, struct dt_object *dt,
  *
  * \param[in] env      execution environment
  * \param[in] dt       object to be locked
- * \param[in] role     lock role from MDD layer, see mdd_object_role().
+ * \param[in] role     lock role from MDD layer, see dt_object_role().
  */
 static void osp_md_write_lock(const struct lu_env *env, struct dt_object *dt,
                              unsigned role)
@@ -873,7 +873,7 @@ static int osp_md_xattr_list(const struct lu_env *env, struct dt_object *dt,
                                             OUT_UPDATE_REPLY_SIZE);
        if (reply->ourp_magic != UPDATE_REPLY_MAGIC) {
                DEBUG_REQ(D_ERROR, req,
-                         "%s: Wrong version %x expected %x "DFID": rc = %d\n",
+                         "%s: Wrong version %x expected %x "DFID": rc = %d",
                          dname, reply->ourp_magic, UPDATE_REPLY_MAGIC,
                          PFID(lu_object_fid(&dt->do_lu)), -EPROTO);
 
@@ -1089,6 +1089,7 @@ struct dt_object_operations osp_md_obj_ops = {
        .do_object_lock       = osp_md_object_lock,
        .do_object_unlock     = osp_md_object_unlock,
        .do_invalidate        = osp_invalidate,
+       .do_check_stale       = osp_check_stale,
 };
 
 /**
@@ -1127,6 +1128,23 @@ static ssize_t osp_md_declare_write(const struct lu_env *env,
        return 0;
 }
 
+static int osp_write_interpreter(const struct lu_env *env,
+                                 struct object_update_reply *reply,
+                                 struct ptlrpc_request *req,
+                                 struct osp_object *obj,
+                                 void *data, int index, int rc)
+{
+       if (rc) {
+               CDEBUG(D_HA, "error "DFID": rc = %d\n",
+                      PFID(lu_object_fid(&obj->opo_obj.do_lu)), rc);
+               spin_lock(&obj->opo_lock);
+               obj->opo_attr.la_valid = 0;
+               obj->opo_stale = 1;
+               spin_unlock(&obj->opo_lock);
+       }
+       return 0;
+}
+
 /**
  * Implementation of dt_body_operations::dbo_write
  *
@@ -1167,6 +1185,12 @@ static ssize_t osp_md_write(const struct lu_env *env, struct dt_object *dt,
        if (rc < 0)
                RETURN(rc);
 
+       /* to be able to invalidate object's state in case of an error */
+       rc = osp_insert_update_callback(env, update, obj, NULL,
+                       osp_write_interpreter);
+       if (rc < 0)
+               RETURN(rc);
+
        /* XXX: how about the write error happened later? */
        *pos += buf->lb_len;
 
@@ -1207,9 +1231,7 @@ static ssize_t osp_md_read(const struct lu_env *env, struct dt_object *dt,
        struct out_read_reply *orr;
        struct ptlrpc_bulk_desc *desc;
        struct object_update_reply *reply;
-       __u32 left_size;
-       int nbufs;
-       int i;
+       int pages;
        int rc;
        ENTRY;
 
@@ -1237,26 +1259,18 @@ static ssize_t osp_md_read(const struct lu_env *env, struct dt_object *dt,
        if (rc != 0)
                GOTO(out_update, rc);
 
-       nbufs = (rbuf->lb_len + OUT_BULK_BUFFER_SIZE - 1) /
-                                       OUT_BULK_BUFFER_SIZE;
+       /* First *and* last might be partial pages, hence +1 */
+       pages = DIV_ROUND_UP(rbuf->lb_len, PAGE_SIZE) + 1;
+
        /* allocate bulk descriptor */
-       desc = ptlrpc_prep_bulk_imp(req, nbufs, 1,
-                                   PTLRPC_BULK_PUT_SINK | PTLRPC_BULK_BUF_KVEC,
-                                   MDS_BULK_PORTAL, &ptlrpc_bulk_kvec_ops);
+       desc = ptlrpc_prep_bulk_imp(req, pages, 1,
+                                   PTLRPC_BULK_PUT_SINK,
+                                   MDS_BULK_PORTAL,
+                                   &ptlrpc_bulk_kiov_nopin_ops);
        if (desc == NULL)
                GOTO(out, rc = -ENOMEM);
 
-       /* split the buffer into small chunk size */
-       left_size = rbuf->lb_len;
-       for (i = 0; i < nbufs; i++) {
-               int read_size;
-
-               read_size = left_size > OUT_BULK_BUFFER_SIZE ?
-                               OUT_BULK_BUFFER_SIZE : left_size;
-               desc->bd_frag_ops->add_iov_frag(desc, ptr, read_size);
-
-               ptr += read_size;
-       }
+       desc->bd_frag_ops->add_iov_frag(desc, ptr, rbuf->lb_len);
 
        osp_set_req_replay(osp, req);
        req->rq_bulk_read = 1;