LU-911 osd: osd_ldiskfs_read returns short reads

osd_ldiskfs_read returned always the requested size even if short
read occurs hiding possible problems. Now it returns real size.

dt_read() helper returns just the same result and can be used to read
any data.

dt_record_read() must be used only for fixed length records

Fix mdt_getattr_internal() code related to readlink, it always tried
to read link of size + 1, so now we need to take this into account.

Signed-off-by: Mikhail Pershin <tappro@whamcloud.com>
Change-Id: Ib603c5cbe5b06f3f6a9aef74c52bb78cdbd4c3eb
Reviewed-on: http://review.whamcloud.com/2263
Tested-by: Hudson
Tested-by: Maloo <whamcloud.maloo@gmail.com>
Reviewed-by: Alex Zhuravlev <bzzz@whamcloud.com>
Reviewed-by: Oleg Drokin <green@whamcloud.com>

diff --git a/lustre/include/dt_object.h b/lustre/include/dt_object.h
index ea09e46..f07dd5a 100644 (file)
--- a/lustre/include/dt_object.h
+++ b/lustre/include/dt_object.h
@@ -744,12 +744,13 @@ void dt_version_set(const struct lu_env *env, struct dt_object *o,
 dt_obj_version_t dt_version_get(const struct lu_env *env, struct dt_object *o);
 
 
+int dt_read(const struct lu_env *env, struct dt_object *dt,
+            struct lu_buf *buf, loff_t *pos);
 int dt_record_read(const struct lu_env *env, struct dt_object *dt,
                    struct lu_buf *buf, loff_t *pos);
 int dt_record_write(const struct lu_env *env, struct dt_object *dt,
                     const struct lu_buf *buf, loff_t *pos, struct thandle *th);
 
-
 static inline struct thandle *dt_trans_create(const struct lu_env *env,
                                               struct dt_device *d)
 {
@@ -798,6 +799,8 @@ static inline int dt_declare_record_write(const struct lu_env *env,
 
         LASSERTF(dt != NULL, "dt is NULL when we want to write record\n");
         LASSERT(th != NULL);
+        LASSERT(dt->do_body_ops);
+        LASSERT(dt->do_body_ops->dbo_declare_write);
         rc = dt->do_body_ops->dbo_declare_write(env, dt, size, pos, th);
         return rc;
 }

diff --git a/lustre/mdt/mdt_handler.c b/lustre/mdt/mdt_handler.c
index e2f4c3c..63444a1 100644 (file)
--- a/lustre/mdt/mdt_handler.c
+++ b/lustre/mdt/mdt_handler.c
@@ -578,7 +578,9 @@ static int mdt_getattr_internal(struct mdt_thread_info *info,
         } else if (S_ISLNK(la->la_mode) &&
                    reqbody->valid & OBD_MD_LINKNAME) {
                 buffer->lb_buf = ma->ma_lmm;
-                buffer->lb_len = reqbody->eadatasize;
+                /* eadatasize from client includes NULL-terminator, so
+                 * there is no need to read it */
+                buffer->lb_len = reqbody->eadatasize - 1;
                 rc = mo_readlink(env, next, buffer);
                 if (unlikely(rc <= 0)) {
                         CERROR("readlink failed: %d\n", rc);
@@ -587,9 +589,14 @@ static int mdt_getattr_internal(struct mdt_thread_info *info,
                         if (OBD_FAIL_CHECK(OBD_FAIL_MDS_READLINK_EPROTO))
                                  rc -= 2;
                         repbody->valid |= OBD_MD_LINKNAME;
-                        repbody->eadatasize = rc;
+                        /* we need to report back size with NULL-terminator
+                         * because client expects that */
+                        repbody->eadatasize = rc + 1;
+                        if (repbody->eadatasize != reqbody->eadatasize)
+                                CERROR("Read shorter link %d than expected "
+                                       "%d\n", rc, reqbody->eadatasize - 1);
                         /* NULL terminate */
-                        ((char*)ma->ma_lmm)[rc - 1] = 0;
+                        ((char*)ma->ma_lmm)[rc] = 0;
                         CDEBUG(D_INODE, "symlink dest %s, len = %d\n",
                                (char*)ma->ma_lmm, rc);
                         rc = 0;

diff --git a/lustre/obdclass/dt_object.c b/lustre/obdclass/dt_object.c
index e486f06..becb572 100644 (file)
--- a/lustre/obdclass/dt_object.c
+++ b/lustre/obdclass/dt_object.c
@@ -391,6 +391,38 @@ void dt_global_fini(void)
         lu_context_key_degister(&dt_key);
 }
 
+/**
+ * Generic read helper. May return an error for partial reads.
+ *
+ * \param env  lustre environment
+ * \param dt   object to be read
+ * \param buf  lu_buf to be filled, with buffer pointer and length
+ * \param pos position to start reading, updated as data is read
+ *
+ * \retval real size of data read
+ * \retval -ve errno on failure
+ */
+int dt_read(const struct lu_env *env, struct dt_object *dt,
+            struct lu_buf *buf, loff_t *pos)
+{
+        LASSERTF(dt != NULL, "dt is NULL when we want to read record\n");
+        return dt->do_body_ops->dbo_read(env, dt, buf, pos, BYPASS_CAPA);
+}
+EXPORT_SYMBOL(dt_read);
+
+/**
+ * Read structures of fixed size from storage.  Unlike dt_read(), using
+ * dt_record_read() will return an error for partial reads.
+ *
+ * \param env  lustre environment
+ * \param dt   object to be read
+ * \param buf  lu_buf to be filled, with buffer pointer and length
+ * \param pos position to start reading, updated as data is read
+ *
+ * \retval 0 on successfully reading full buffer
+ * \retval -EFAULT on short read
+ * \retval -ve errno on failure
+ */
 int dt_record_read(const struct lu_env *env, struct dt_object *dt,
                    struct lu_buf *buf, loff_t *pos)
 {
@@ -452,7 +484,7 @@ void dt_version_set(const struct lu_env *env, struct dt_object *o,
         vbuf.lb_len = sizeof(version);
 
         rc = dt_xattr_set(env, o, &vbuf, xname, 0, th, BYPASS_CAPA);
-        if (rc != 0)
+        if (rc < 0)
                 CDEBUG(D_INODE, "Can't set version, rc %d\n", rc);
         return;
 }

diff --git a/lustre/osd-ldiskfs/osd_io.c b/lustre/osd-ldiskfs/osd_io.c
index d2b10e6..7b47961 100644 (file)
--- a/lustre/osd-ldiskfs/osd_io.c
+++ b/lustre/osd-ldiskfs/osd_io.c
@@ -855,7 +855,7 @@ int osd_ldiskfs_read(struct inode *inode, void *buf, int size, loff_t *offs)
 {
         struct buffer_head *bh;
         unsigned long block;
-        int osize = size;
+        int osize;
         int blocksize;
         int csize;
         int boffs;
@@ -878,15 +878,16 @@ int osd_ldiskfs_read(struct inode *inode, void *buf, int size, loff_t *offs)
         }
 
         blocksize = 1 << inode->i_blkbits;
+        osize = size;
         while (size > 0) {
                 block = *offs >> inode->i_blkbits;
                 boffs = *offs & (blocksize - 1);
                 csize = min(blocksize - boffs, size);
                 bh = ldiskfs_bread(NULL, inode, block, 0, &err);
                 if (!bh) {
-                        CERROR("%s: error reading offset %llu (block %lu): "
-                               "rc = %d\n",
-                               inode->i_sb->s_id, *offs, block, err);
+                        CERROR("%s: can't read %u@%llu on ino %lu: rc = %d\n",
+                               LDISKFS_SB(inode->i_sb)->s_es->s_volume_name,
+                               csize, *offs, inode->i_ino, err);
                         return err;
                 }