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 d86bb5c..dc5230a 100644 (file)
--- a/lustre/ofd/ofd_io.c
+++ b/lustre/ofd/ofd_io.c
@@ -475,6 +475,71 @@ out:
 
 }
 
+/*
+ * Lazy ATIME update to refresh atime every ofd_atime_diff
+ * seconds so that external scanning tool can see it actual
+ * within that period and be able to identify accessed files
+ */
+static void ofd_handle_atime(const struct lu_env *env, struct ofd_device *ofd,
+                            struct ofd_object *fo, time64_t atime)
+{
+       struct lu_attr *la;
+       struct dt_object *o;
+       struct thandle *th;
+       int rc;
+
+       if (ofd->ofd_atime_diff == 0)
+               return;
+
+       la = &ofd_info(env)->fti_attr2;
+       o = ofd_object_child(fo);
+
+       if (unlikely(fo->ofo_atime_ondisk == 0)) {
+               rc = dt_attr_get(env, o, la);
+               if (unlikely(rc))
+                       return;
+               LASSERT(la->la_valid & LA_ATIME);
+               if (la->la_atime == 0)
+                       la->la_atime = la->la_mtime;
+               fo->ofo_atime_ondisk = la->la_atime;
+       }
+       if (atime - fo->ofo_atime_ondisk < ofd->ofd_atime_diff)
+               return;
+
+       /* atime hasn't been updated too long, update it */
+       fo->ofo_atime_ondisk = atime;
+
+       th = ofd_trans_create(env, ofd);
+       if (IS_ERR(th)) {
+               CERROR("%s: cannot create transaction: rc = %d\n",
+                      ofd_name(ofd), (int)PTR_ERR(th));
+               return;
+       }
+
+       la->la_valid = LA_ATIME;
+       rc = dt_declare_attr_set(env, o, la, th);
+       if (rc)
+               GOTO(out_tx, rc);
+
+       rc = dt_trans_start_local(env, ofd->ofd_osd , th);
+       if (rc) {
+               CERROR("%s: cannot start transaction: rc = %d\n",
+                      ofd_name(ofd), rc);
+               GOTO(out_tx, rc);
+       }
+
+       ofd_read_lock(env, fo);
+       if (ofd_object_exists(fo)) {
+               la->la_atime = fo->ofo_atime_ondisk;
+               rc = dt_attr_set(env, o, la, th);
+       }
+
+       ofd_read_unlock(env, fo);
+
+out_tx:
+       ofd_trans_stop(env, ofd, th, rc);
+}
+
 /**
  * Prepare buffers for read request processing.
  *
@@ -506,6 +571,8 @@ static int ofd_preprw_read(const struct lu_env *env, struct obd_export *exp,
        int i, j, rc, tot_bytes = 0;
        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);
@@ -517,6 +584,9 @@ static int ofd_preprw_read(const struct lu_env *env, struct obd_export *exp,
 
        ofd_info(env)->fti_obj = fo;
 
+       if (oa->o_valid & OBD_MD_FLATIME)
+               ofd_handle_atime(env, ofd, fo, oa->o_atime);
+
        ofd_read_lock(env, fo);
        if (!ofd_object_exists(fo))
                GOTO(unlock, rc = -ENOENT);
@@ -530,7 +600,12 @@ static int ofd_preprw_read(const struct lu_env *env, struct obd_export *exp,
        if (ptlrpc_connection_is_local(exp->exp_connection))
                dbt |= DT_BUFS_TYPE_LOCAL;
 
+       begin = -1;
+       end = 0;
+
        for (*nr_local = 0, i = 0, j = 0; i < niocount; i++) {
+               begin = min_t(__u64, begin, rnb[i].rnb_offset);
+               end = max_t(__u64, end, rnb[i].rnb_offset + rnb[i].rnb_len);
 
                if (OBD_FAIL_CHECK(OBD_FAIL_OST_2BIG_NIOBUF))
                        rnb[i].rnb_len = 100 * 1024 * 1024;
@@ -553,7 +628,20 @@ static int ofd_preprw_read(const struct lu_env *env, struct obd_export *exp,
        if (unlikely(rc))
                GOTO(buf_put, rc);
 
-       ofd_counter_incr(exp, LPROC_OFD_STATS_READ, jobid, tot_bytes);
+       ofd_access(env, ofd,
+               &(struct lu_fid) {
+                       .f_seq = oa->o_parent_seq,
+                       .f_oid = oa->o_parent_oid,
+                       .f_ver = oa->o_stripe_idx,
+               },
+               begin, end,
+               tot_bytes,
+               niocount,
+               READ);
+
+       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:
@@ -598,6 +686,8 @@ static int ofd_preprw_write(const struct lu_env *env, struct obd_export *exp,
        int i, j, k, rc = 0, tot_bytes = 0;
        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);
@@ -705,8 +795,14 @@ static int ofd_preprw_write(const struct lu_env *env, struct obd_export *exp,
        if (ptlrpc_connection_is_local(exp->exp_connection))
                dbt |= DT_BUFS_TYPE_LOCAL;
 
+       begin = -1;
+       end = 0;
+
        /* parse remote buffers to local buffers and prepare the latter */
        for (*nr_local = 0, i = 0, j = 0; i < obj->ioo_bufcnt; i++) {
+               begin = min_t(__u64, begin, rnb[i].rnb_offset);
+               end = max_t(__u64, end, rnb[i].rnb_offset + rnb[i].rnb_len);
+
                if (OBD_FAIL_CHECK(OBD_FAIL_OST_2BIG_NIOBUF))
                        rnb[i].rnb_len += PAGE_SIZE;
                rc = dt_bufs_get(env, ofd_object_child(fo),
@@ -734,7 +830,21 @@ static int ofd_preprw_write(const struct lu_env *env, struct obd_export *exp,
                GOTO(err, rc);
 
        ofd_read_unlock(env, fo);
-       ofd_counter_incr(exp, LPROC_OFD_STATS_WRITE, jobid, tot_bytes);
+
+       ofd_access(env, ofd,
+               &(struct lu_fid) {
+                       .f_seq = oa->o_parent_seq,
+                       .f_oid = oa->o_parent_oid,
+                       .f_ver = oa->o_stripe_idx,
+               },
+               begin, end,
+               tot_bytes,
+               obj->ioo_bufcnt,
+               WRITE);
+
+       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);
@@ -1186,6 +1296,10 @@ retry:
                        GOTO(out_stop, rc);
        }
 
+       /* don't update atime on disk if it is older */
+       if (la->la_valid & LA_ATIME && la->la_atime <= fo->ofo_atime_ondisk)
+               la->la_valid &= ~LA_ATIME;
+
        if (la->la_valid) {
                /* update [mac]time if needed */
                rc = dt_declare_attr_set(env, o, la, th);
@@ -1202,7 +1316,9 @@ retry:
                GOTO(out_unlock, rc = -ENOENT);
 
        if (likely(!fake_write)) {
-               rc = dt_write_commit(env, o, lnb, niocount, th);
+               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);
        }
@@ -1213,6 +1329,8 @@ retry:
                rc = dt_attr_set(env, o, la, th);
                if (rc)
                        GOTO(out_unlock, rc);
+               if (la->la_valid & LA_ATIME)
+                       fo->ofo_atime_ondisk = la->la_atime;
        }
 
        /* get attr to return */