LU-12553 mdc: polling mode for changelog reader

this allows the user (like lsom_sync and similar) to follow
the changelog and don't rescan getting duplicates.

Change-Id: I78dc163838c1b88f9447a4731ad4bfe00fec7eff
Signed-off-by: Alex Zhuravlev <bzzz@whamcloud.com>
Reviewed-on: https://review.whamcloud.com/35262
Reviewed-by: Patrick Farrell <pfarrell@whamcloud.com>
Tested-by: jenkins <devops@whamcloud.com>
Tested-by: Maloo <maloo@whamcloud.com>
Reviewed-by: Andreas Dilger <adilger@whamcloud.com>

diff --git a/lustre/include/lustre/lustreapi.h b/lustre/include/lustre/lustreapi.h
index 0b6e81c..4839fca 100644 (file)
--- a/lustre/include/lustre/lustreapi.h
+++ b/lustre/include/lustre/lustreapi.h
@@ -462,6 +462,7 @@ int llapi_changelog_start(void **priv, enum changelog_send_flag flags,
                          const char *mdtname, long long startrec);
 int llapi_changelog_fini(void **priv);
 int llapi_changelog_recv(void *priv, struct changelog_rec **rech);
+int llapi_changelog_in_buf(void *priv);
 int llapi_changelog_free(struct changelog_rec **rech);
 int llapi_changelog_get_fd(void *priv);
 /* Allow records up to endrec to be destroyed; requires registered id. */

diff --git a/lustre/include/uapi/linux/lustre/lustre_ioctl.h b/lustre/include/uapi/linux/lustre/lustre_ioctl.h
index 3c97d63..5a68cfb 100644 (file)
--- a/lustre/include/uapi/linux/lustre/lustre_ioctl.h
+++ b/lustre/include/uapi/linux/lustre/lustre_ioctl.h
@@ -239,6 +239,7 @@ static inline __u32 obd_ioctl_packlen(struct obd_ioctl_data *data)
 #define OBD_IOC_START_LFSCK    _IOWR('f', 230, OBD_IOC_DATA_TYPE)
 #define OBD_IOC_STOP_LFSCK     _IOW('f', 231, OBD_IOC_DATA_TYPE)
 #define OBD_IOC_QUERY_LFSCK    _IOR('f', 232, struct obd_ioctl_data)
+#define OBD_IOC_CHLG_POLL      _IOR('f', 233, long)
 /*     lustre/lustre_user.h    240-249 */
 /* was LIBCFS_IOC_DEBUG_MASK   _IOWR('f', 250, long) until 2.11 */
 

diff --git a/lustre/mdc/mdc_changelog.c b/lustre/mdc/mdc_changelog.c
index e29e705..4eeac84 100644 (file)
--- a/lustre/mdc/mdc_changelog.c
+++ b/lustre/mdc/mdc_changelog.c
@@ -36,6 +36,7 @@
 #include <linux/miscdevice.h>
 
 #include <lustre_log.h>
+#include <uapi/linux/lustre/lustre_ioctl.h>
 
 #include "mdc_internal.h"
 
@@ -89,6 +90,9 @@ struct chlg_reader_state {
        __u64                    crs_rec_count;
        /* List of prefetched enqueued_record::enq_linkage_items */
        struct list_head         crs_rec_queue;
+       unsigned int             crs_last_catidx;
+       unsigned int             crs_last_idx;
+       bool                     crs_poll;
 };
 
 struct chlg_rec_entry {
@@ -133,6 +137,9 @@ static int chlg_read_cat_process_cb(const struct lu_env *env,
 
        rec = container_of(hdr, struct llog_changelog_rec, cr_hdr);
 
+       crs->crs_last_catidx = llh->lgh_hdr->llh_cat_idx;
+       crs->crs_last_idx = hdr->lrh_index;
+
        if (rec->cr_hdr.lrh_type != CHANGELOG_REC) {
                rc = -EINVAL;
                CERROR("%s: not a changelog rec %x/%d in llog "DFID" rc = %d\n",
@@ -208,6 +215,10 @@ static int chlg_load(void *args)
        if (ctx == NULL)
                GOTO(err_out, rc = -ENOENT);
 
+       crs->crs_last_catidx = -1;
+       crs->crs_last_idx = 0;
+
+again:
        rc = llog_open(NULL, ctx, &llh, NULL, CHANGELOG_CATALOG,
                       LLOG_OPEN_EXISTS);
        if (rc) {
@@ -216,6 +227,7 @@ static int chlg_load(void *args)
                GOTO(err_out, rc);
        }
 
+
        rc = llog_init_handle(NULL, llh,
                              LLOG_F_IS_CAT |
                              LLOG_F_EXT_JOBID |
@@ -231,11 +243,17 @@ static int chlg_load(void *args)
                GOTO(err_out, rc);
        }
 
-       rc = llog_cat_process(NULL, llh, chlg_read_cat_process_cb, crs, 0, 0);
+       rc = llog_cat_process(NULL, llh, chlg_read_cat_process_cb, crs,
+                               crs->crs_last_catidx, crs->crs_last_idx);
        if (rc < 0) {
                CERROR("%s: fail to process llog: rc = %d\n", obd->obd_name, rc);
                GOTO(err_out, rc);
        }
+       if (!kthread_should_stop() && crs->crs_poll) {
+               llog_cat_close(NULL, llh);
+               schedule_timeout_interruptible(HZ);
+               goto again;
+       }
 
        crs->crs_eof = true;
 
@@ -590,6 +608,23 @@ static unsigned int chlg_poll(struct file *file, poll_table *wait)
        return mask;
 }
 
+static long chlg_ioctl(struct file *file, unsigned int cmd, unsigned long arg)
+{
+       int rc;
+
+       struct chlg_reader_state *crs = file->private_data;
+       switch (cmd) {
+       case OBD_IOC_CHLG_POLL:
+               crs->crs_poll = !!arg;
+               rc = 0;
+               break;
+       default:
+               rc = -EINVAL;
+               break;
+       }
+       return rc;
+}
+
 static const struct file_operations chlg_fops = {
        .owner          = THIS_MODULE,
        .llseek         = chlg_llseek,
@@ -598,6 +633,7 @@ static const struct file_operations chlg_fops = {
        .open           = chlg_open,
        .release        = chlg_release,
        .poll           = chlg_poll,
+       .unlocked_ioctl = chlg_ioctl,
 };
 
 /**

diff --git a/lustre/utils/liblustreapi_chlg.c b/lustre/utils/liblustreapi_chlg.c
index 5735fc1..fc83fa4 100644 (file)
--- a/lustre/utils/liblustreapi_chlg.c
+++ b/lustre/utils/liblustreapi_chlg.c
@@ -36,8 +36,10 @@
 #include <poll.h>
 #include <sys/stat.h>
 #include <sys/types.h>
+#include <sys/ioctl.h>
 
 #include <lustre/lustreapi.h>
+#include <linux/lustre/lustre_ioctl.h>
 
 
 static int chlg_dev_path(char *path, size_t path_len, const char *device)
@@ -93,7 +95,6 @@ int llapi_changelog_start(void **priv, enum changelog_send_flag flags,
        struct changelog_private *cp;
        static bool warned_extra_flags;
        static bool warned_jobid;
-       static bool warned_follow;
        char cdev_path[PATH_MAX];
        int rc;
 
@@ -150,12 +151,12 @@ int llapi_changelog_start(void **priv, enum changelog_send_flag flags,
                warned_jobid = true;
        }
 
-       /* Behavior expected by CHANGELOG_FLAG_FOLLOW is not implemented, warn
-        * the user and ignore it. */
-       if (flags & CHANGELOG_FLAG_FOLLOW && !warned_follow) {
-               llapi_err_noerrno(LLAPI_MSG_WARN, "warning: %s() called with "
-                                 "CHANGELOG_FLAG_FOLLOW (ignored)", __func__);
-               warned_follow = true;
+       if (flags & CHANGELOG_FLAG_FOLLOW) {
+               int rc;
+               rc = ioctl(cp->clp_fd, OBD_IOC_CHLG_POLL, 1);
+               if (rc < 0)
+                       llapi_err_noerrno(LLAPI_MSG_ERROR, "can't enable "
+                                         "CHANGELOG_FLAG_FOLLOW");
        }
 
        return 0;
@@ -294,6 +295,14 @@ int llapi_changelog_free(struct changelog_rec **rech)
        return 0;
 }
 
+int llapi_changelog_in_buf(void *priv)
+{
+       struct changelog_private *cp = priv;
+       if (cp->clp_buf + cp->clp_buf_len > cp->clp_buf_pos)
+               return 1;
+       return 0;
+}
+
 int llapi_changelog_clear(const char *mdtname, const char *idstr,
                          long long endrec)
 {