LU-14052 ofd: support for multiple access readers

[fs/lustre-release.git] / lustre / ofd / ofd_io.c

diff --git a/lustre/ofd/ofd_io.c b/lustre/ofd/ofd_io.c
index f836194..dc5230a 100644 (file)
--- a/lustre/ofd/ofd_io.c
+++ b/lustre/ofd/ofd_io.c
@@ -572,6 +572,7 @@ static int ofd_preprw_read(const struct lu_env *env, struct obd_export *exp,
        enum dt_bufs_type dbt = DT_BUFS_TYPE_READ;
        int maxlnb = *nr_local;
        __u64 begin, end;
+       ktime_t kstart = ktime_get();
 
        ENTRY;
        LASSERT(env != NULL);
@@ -627,7 +628,7 @@ static int ofd_preprw_read(const struct lu_env *env, struct obd_export *exp,
        if (unlikely(rc))
                GOTO(buf_put, rc);
 
-       ofd_access(ofd,
+       ofd_access(env, ofd,
                &(struct lu_fid) {
                        .f_seq = oa->o_parent_seq,
                        .f_oid = oa->o_parent_oid,
@@ -638,7 +639,9 @@ static int ofd_preprw_read(const struct lu_env *env, struct obd_export *exp,
                niocount,
                READ);
 
-       ofd_counter_incr(exp, LPROC_OFD_STATS_READ, jobid, tot_bytes);
+       ofd_counter_incr(exp, LPROC_OFD_STATS_READ_BYTES, jobid, tot_bytes);
+       ofd_counter_incr(exp, LPROC_OFD_STATS_READ, jobid,
+                        ktime_us_delta(ktime_get(), kstart));
        RETURN(0);
 
 buf_put:
@@ -684,6 +687,7 @@ static int ofd_preprw_write(const struct lu_env *env, struct obd_export *exp,
        enum dt_bufs_type dbt = DT_BUFS_TYPE_WRITE;
        int maxlnb = *nr_local;
        __u64 begin, end;
+       ktime_t kstart = ktime_get();
 
        ENTRY;
        LASSERT(env != NULL);
@@ -827,7 +831,7 @@ static int ofd_preprw_write(const struct lu_env *env, struct obd_export *exp,
 
        ofd_read_unlock(env, fo);
 
-       ofd_access(ofd,
+       ofd_access(env, ofd,
                &(struct lu_fid) {
                        .f_seq = oa->o_parent_seq,
                        .f_oid = oa->o_parent_oid,
@@ -838,7 +842,9 @@ static int ofd_preprw_write(const struct lu_env *env, struct obd_export *exp,
                obj->ioo_bufcnt,
                WRITE);
 
-       ofd_counter_incr(exp, LPROC_OFD_STATS_WRITE, jobid, tot_bytes);
+       ofd_counter_incr(exp, LPROC_OFD_STATS_WRITE_BYTES, jobid, tot_bytes);
+       ofd_counter_incr(exp, LPROC_OFD_STATS_WRITE, jobid,
+                        ktime_us_delta(ktime_get(), kstart));
        RETURN(0);
 err:
        dt_bufs_put(env, ofd_object_child(fo), lnb, *nr_local);
@@ -1310,6 +1316,8 @@ retry:
                GOTO(out_unlock, rc = -ENOENT);
 
        if (likely(!fake_write)) {
+               OBD_FAIL_TIMEOUT_ORSET(OBD_FAIL_OST_WR_ATTR_DELAY,
+                                      OBD_FAIL_ONCE, cfs_fail_val);
                rc = dt_write_commit(env, o, lnb, niocount, th, oa->o_size);
                if (rc)
                        GOTO(out_unlock, rc);