LU-9714 llog: fix llog_process_thread race

lgh_cur_offset is used at llog_write_osd_rec,
and modified at llog_process_thread.
When two threads of llog_process_thread are processing the
same llog, it could happen that one thread do llog_write,
and second modify lgh_cur_offset. This situation drives to
wrong modification of llog.

Signed-off-by: Alexander Boyko <alexander.boyko@seagate.com>
Seagate-bug-id: MRP-4455
Change-Id: I7a63850c876146b14118a7d395cf3cfb3a40dd67
Reviewed-on: https://review.whamcloud.com/27838
Reviewed-by: Sergey Cheremencev <c17829@cray.com>
Tested-by: Jenkins
Tested-by: Maloo <hpdd-maloo@intel.com>
Reviewed-by: Mike Pershin <mike.pershin@intel.com>
Reviewed-by: Oleg Drokin <oleg.drokin@intel.com>

diff --git a/lustre/include/lustre_log.h b/lustre/include/lustre_log.h
index e259ed3..a0386db 100644 (file)
--- a/lustre/include/lustre_log.h
+++ b/lustre/include/lustre_log.h
@@ -272,8 +272,7 @@ struct llog_handle {
         * case, after it will have reached LLOG_HDR_BITMAP_SIZE, llh_cat_idx
         * will become its upper limit */
        int                      lgh_last_idx;
-       int                      lgh_cur_idx; /* used during llog_process */
-       __u64                    lgh_cur_offset; /* used during llog_process */
+       __u64                    lgh_cur_offset; /* used for test only */
        struct llog_ctxt        *lgh_ctxt;
        union {
                struct plain_handle_data         phd;

diff --git a/lustre/include/uapi/linux/lustre/lustre_idl.h b/lustre/include/uapi/linux/lustre/lustre_idl.h
index 64814fe..9e199d2 100644 (file)
--- a/lustre/include/uapi/linux/lustre/lustre_idl.h
+++ b/lustre/include/uapi/linux/lustre/lustre_idl.h
@@ -2887,9 +2887,13 @@ struct llog_log_hdr {
                                                 llh->llh_hdr.lrh_len - \
                                                 sizeof(llh->llh_tail)))
 
-/** log cookies are used to reference a specific log file and a record therein */
+/** log cookies are used to reference a specific log file and a record therein,
+    and pass record offset from llog_process_thread to llog_write */
 struct llog_cookie {
-        struct llog_logid       lgc_lgl;
+       union {
+               struct llog_logid       lgc_lgl;
+               __u64                   lgc_offset;
+       };
         __u32                   lgc_subsys;
         __u32                   lgc_index;
         __u32                   lgc_padding;

diff --git a/lustre/obdclass/llog.c b/lustre/obdclass/llog.c
index d3b1e04..816f46f 100644 (file)
--- a/lustre/obdclass/llog.c
+++ b/lustre/obdclass/llog.c
@@ -419,6 +419,7 @@ static int llog_process_thread(void *arg)
        struct llog_handle              *loghandle = lpi->lpi_loghandle;
        struct llog_log_hdr             *llh = loghandle->lgh_hdr;
        struct llog_process_cat_data    *cd  = lpi->lpi_catdata;
+       struct llog_thread_info         *lti;
        char                            *buf;
        size_t                           chunk_size;
        __u64                            cur_offset;
@@ -432,6 +433,8 @@ static int llog_process_thread(void *arg)
        if (llh == NULL)
                RETURN(-EINVAL);
 
+       lti = lpi->lpi_env == NULL ? NULL : llog_info(lpi->lpi_env);
+
        cur_offset = chunk_size = llh->llh_hdr.lrh_len;
        /* expect chunk_size to be power of two */
        LASSERT(is_power_of_2(chunk_size));
@@ -594,15 +597,37 @@ repeat:
                               rec->lrh_index, rec->lrh_len,
                               (int)(buf + chunk_size - (char *)rec));
 
-                       loghandle->lgh_cur_idx = rec->lrh_index;
+                       /* lgh_cur_offset is used only at llog_test_3 */
                        loghandle->lgh_cur_offset = (char *)rec - (char *)buf +
                                                    chunk_offset;
 
                        /* if set, process the callback on this record */
                        if (ext2_test_bit(index, LLOG_HDR_BITMAP(llh))) {
+                               struct llog_cookie *lgc;
+                               __u64   tmp_off;
+                               int     tmp_idx;
+
+                               if (lti != NULL) {
+                                       lgc = &lti->lgi_cookie;
+                                       /* store lu_env for recursive calls */
+                                       tmp_off = lgc->lgc_offset;
+                                       tmp_idx = lgc->lgc_index;
+
+                                       lgc->lgc_offset = (char *)rec -
+                                               (char *)buf + chunk_offset;
+                                       lgc->lgc_index = rec->lrh_index;
+                               }
+                               /* using lu_env for passing record offset to
+                                * llog_write through various callbacks */
                                rc = lpi->lpi_cb(lpi->lpi_env, loghandle, rec,
                                                 lpi->lpi_cbdata);
                                last_called_index = index;
+
+                               if (lti != NULL) {
+                                       lgc->lgc_offset = tmp_off;
+                                       lgc->lgc_index = tmp_idx;
+                               }
+
                                if (rc == LLOG_PROC_BREAK) {
                                        GOTO(out, rc);
                                } else if (rc == LLOG_DEL_RECORD) {
@@ -1140,7 +1165,8 @@ int llog_write(const struct lu_env *env, struct llog_handle *loghandle,
 {
        struct dt_device        *dt;
        struct thandle          *th;
-       int                      rc;
+       bool                    need_cookie;
+       int                     rc;
 
        ENTRY;
 
@@ -1163,8 +1189,21 @@ int llog_write(const struct lu_env *env, struct llog_handle *loghandle,
        if (rc)
                GOTO(out_trans, rc);
 
+       need_cookie = !(idx == LLOG_HEADER_IDX || idx == LLOG_NEXT_IDX);
+
        down_write(&loghandle->lgh_lock);
-       rc = llog_write_rec(env, loghandle, rec, NULL, idx, th);
+       if (need_cookie) {
+               struct llog_thread_info *lti = llog_info(env);
+
+               /* cookie comes from llog_process_thread */
+               rc = llog_write_rec(env, loghandle, rec, &lti->lgi_cookie,
+                                   rec->lrh_index, th);
+               /* upper layer didn`t pass cookie so change rc */
+               rc = (rc == 1 ? 0 : rc);
+       } else {
+               rc = llog_write_rec(env, loghandle, rec, NULL, idx, th);
+       }
+
        up_write(&loghandle->lgh_lock);
 out_trans:
        dt_trans_stop(env, dt, th);

diff --git a/lustre/obdclass/llog_osd.c b/lustre/obdclass/llog_osd.c
index f1c4b22..ffa19bd 100644 (file)
--- a/lustre/obdclass/llog_osd.c
+++ b/lustre/obdclass/llog_osd.c
@@ -362,7 +362,7 @@ static int llog_osd_declare_write_rec(const struct lu_env *env,
  *                             the full llog record to write. This is
  *                             the beginning of buffer to write, the length
  *                             of buffer is stored in \a rec::lrh_len
- * \param[out] reccookie       pointer to the cookie to return back if needed.
+ * \param[in,out] reccookie    pointer to the cookie to return back if needed.
  *                             It is used for further cancel of this llog
  *                             record.
  * \param[in]  idx             index of the llog record. If \a idx == -1 then
@@ -490,26 +490,26 @@ static int llog_osd_write_rec(const struct lu_env *env,
                                             &lgi->lgi_off, th);
 
                        RETURN(rc);
-               } else if (loghandle->lgh_cur_idx > 0) {
+               } else if (llh->llh_flags & LLOG_F_IS_FIXSIZE) {
+                       lgi->lgi_off = llh->llh_hdr.lrh_len +
+                                      (idx - 1) * reclen;
+               } else if (reccookie != NULL && reccookie->lgc_index > 0) {
                        /**
-                        * The lgh_cur_offset can be used only if index is
+                        * The lgc_offset can be used only if index is
                         * the same.
                         */
-                       if (idx != loghandle->lgh_cur_idx) {
+                       if (idx != reccookie->lgc_index) {
                                CERROR("%s: modify index mismatch %d %d\n",
                                       o->do_lu.lo_dev->ld_obd->obd_name, idx,
-                                      loghandle->lgh_cur_idx);
+                                      reccookie->lgc_index);
                                RETURN(-EFAULT);
                        }
 
-                       lgi->lgi_off = loghandle->lgh_cur_offset;
+                       lgi->lgi_off = reccookie->lgc_offset;
                        CDEBUG(D_OTHER, "modify record "DFID": idx:%u, "
                               "len:%u offset %llu\n",
                               PFID(&loghandle->lgh_id.lgl_oi.oi_fid), idx,
                               rec->lrh_len, (long long)lgi->lgi_off);
-               } else if (llh->llh_flags & LLOG_F_IS_FIXSIZE) {
-                       lgi->lgi_off = llh->llh_hdr.lrh_len +
-                                      (idx - 1) * reclen;
                } else {
                        /* This can be result of lgh_cur_idx is not set during
                         * llog processing or llh_size is not set to proper