Whamcloud - gitweb
LU-12930 various: use schedule_timeout_*interruptible
[fs/lustre-release.git] / lustre / mdc / mdc_changelog.c
index 483d24d..fb9fa54 100644 (file)
@@ -23,6 +23,8 @@
  * Copyright (c) 2017, Commissariat a l'Energie Atomique et aux Energies
  *                     Alternatives.
  *
+ * Copyright (c) 2017, Intel Corporation.
+ *
  * Author: Henri Doreau <henri.doreau@cea.fr>
  */
 
@@ -34,6 +36,7 @@
 #include <linux/miscdevice.h>
 
 #include <lustre_log.h>
+#include <uapi/linux/lustre/lustre_ioctl.h>
 
 #include "mdc_internal.h"
 
@@ -68,25 +71,30 @@ struct chlg_registered_dev {
 
 struct chlg_reader_state {
        /* Shortcut to the corresponding OBD device */
-       struct obd_device       *crs_obd;
+       struct obd_device          *crs_obd;
+       /* the corresponding chlg_registered_dev */
+       struct chlg_registered_dev *crs_ced;
        /* Producer thread (if any) */
-       struct task_struct      *crs_prod_task;
+       struct task_struct         *crs_prod_task;
        /* An error occurred that prevents from reading further */
-       int                      crs_err;
+       int                         crs_err;
        /* EOF, no more records available */
-       bool                     crs_eof;
+       bool                        crs_eof;
        /* Desired start position */
-       __u64                    crs_start_offset;
+       __u64                       crs_start_offset;
        /* Wait queue for the catalog processing thread */
-       wait_queue_head_t        crs_waitq_prod;
+       wait_queue_head_t           crs_waitq_prod;
        /* Wait queue for the record copy threads */
-       wait_queue_head_t        crs_waitq_cons;
+       wait_queue_head_t           crs_waitq_cons;
        /* Mutex protecting crs_rec_count and crs_rec_queue */
-       struct mutex             crs_lock;
+       struct mutex                crs_lock;
        /* Number of item in the list */
-       __u64                    crs_rec_count;
+       __u64                       crs_rec_count;
        /* List of prefetched enqueued_record::enq_linkage_items */
-       struct list_head         crs_rec_queue;
+       struct list_head            crs_rec_queue;
+       unsigned int                crs_last_catidx;
+       unsigned int                crs_last_idx;
+       bool                        crs_poll;
 };
 
 struct chlg_rec_entry {
@@ -104,6 +112,43 @@ enum {
 };
 
 /**
+ * Deregister a changelog character device whose refcount has reached zero.
+ */
+static void chlg_dev_clear(struct kref *kref)
+{
+       struct chlg_registered_dev *entry = container_of(kref,
+                                               struct chlg_registered_dev,
+                                               ced_refs);
+       ENTRY;
+
+       list_del(&entry->ced_link);
+       misc_deregister(&entry->ced_misc);
+       OBD_FREE_PTR(entry);
+       EXIT;
+}
+
+static inline struct obd_device* chlg_obd_get(struct chlg_registered_dev *dev)
+{
+       struct obd_device *obd;
+
+       mutex_lock(&chlg_registered_dev_lock);
+       if (list_empty(&dev->ced_obds))
+               return NULL;
+
+       obd = list_first_entry(&dev->ced_obds, struct obd_device,
+                              u.cli.cl_chg_dev_linkage);
+       class_incref(obd, "changelog", dev);
+       mutex_unlock(&chlg_registered_dev_lock);
+       return obd;
+}
+
+static inline void chlg_obd_put(struct chlg_registered_dev *dev,
+                        struct obd_device *obd)
+{
+       class_decref(obd, "changelog", dev);
+}
+
+/**
  * ChangeLog catalog processing callback invoked on each record.
  * If the current record is eligible to userland delivery, push
  * it into the crs_rec_queue where the consumer code will fetch it.
@@ -131,6 +176,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",
@@ -196,12 +244,23 @@ static void enq_record_delete(struct chlg_rec_entry *rec)
 static int chlg_load(void *args)
 {
        struct chlg_reader_state *crs = args;
-       struct obd_device *obd = crs->crs_obd;
+       struct chlg_registered_dev *ced = crs->crs_ced;
+       struct obd_device *obd = NULL;
        struct llog_ctxt *ctx = NULL;
        struct llog_handle *llh = NULL;
        int rc;
        ENTRY;
 
+       crs->crs_last_catidx = -1;
+       crs->crs_last_idx = 0;
+
+again:
+       obd = chlg_obd_get(ced);
+       if (obd == NULL)
+               RETURN(-ENODEV);
+
+       crs->crs_obd = obd;
+
        ctx = llog_get_context(obd, LLOG_CHANGELOG_REPL_CTXT);
        if (ctx == NULL)
                GOTO(err_out, rc = -ENOENT);
@@ -214,11 +273,15 @@ static int chlg_load(void *args)
                GOTO(err_out, rc);
        }
 
+
        rc = llog_init_handle(NULL, llh,
                              LLOG_F_IS_CAT |
                              LLOG_F_EXT_JOBID |
                              LLOG_F_EXT_EXTRA_FLAGS |
-                             LLOG_F_EXT_X_UIDGID,
+                             LLOG_F_EXT_X_UIDGID |
+                             LLOG_F_EXT_X_NID |
+                             LLOG_F_EXT_X_OMODE |
+                             LLOG_F_EXT_X_XATTR,
                              NULL);
        if (rc) {
                CERROR("%s: fail to init llog handle: rc = %d\n",
@@ -226,11 +289,19 @@ 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);
+               llog_ctxt_put(ctx);
+               class_decref(obd, "changelog", crs);
+               schedule_timeout_interruptible(cfs_time_seconds(1));
+               goto again;
+       }
 
        crs->crs_eof = true;
 
@@ -246,6 +317,8 @@ err_out:
        if (ctx != NULL)
                llog_ctxt_put(ctx);
 
+       crs->crs_obd = NULL;
+       chlg_obd_put(ced, obd);
        wait_event_interruptible(crs->crs_waitq_prod, kthread_should_stop());
 
        RETURN(rc);
@@ -403,15 +476,23 @@ static loff_t chlg_llseek(struct file *file, loff_t off, int whence)
  */
 static int chlg_clear(struct chlg_reader_state *crs, __u32 reader, __u64 record)
 {
-       struct obd_device *obd = crs->crs_obd;
+       struct obd_device *obd = NULL;
        struct changelog_setinfo cs  = {
                .cs_recno = record,
                .cs_id    = reader
        };
+       int rc;
 
-       return obd_set_info_async(NULL, obd->obd_self_export,
-                                 strlen(KEY_CHANGELOG_CLEAR),
-                                 KEY_CHANGELOG_CLEAR, sizeof(cs), &cs, NULL);
+       obd = chlg_obd_get(crs->crs_ced);
+       if (obd == NULL)
+               return -ENODEV;
+
+       rc = obd_set_info_async(NULL, obd->obd_self_export,
+                               strlen(KEY_CHANGELOG_CLEAR),
+                               KEY_CHANGELOG_CLEAR, sizeof(cs), &cs, NULL);
+
+       chlg_obd_put(crs->crs_ced, obd);
+       return rc;
 }
 
 /** Maximum changelog control command size */
@@ -461,31 +542,6 @@ out_kbuf:
 }
 
 /**
- * Find the OBD device associated to a changelog character device.
- * @param[in]  cdev  character device instance descriptor
- * @return corresponding OBD device or NULL if none was found.
- */
-static struct obd_device *chlg_obd_get(dev_t cdev)
-{
-       int minor = MINOR(cdev);
-       struct obd_device *obd = NULL;
-       struct chlg_registered_dev *curr;
-
-       mutex_lock(&chlg_registered_dev_lock);
-       list_for_each_entry(curr, &chlg_registered_devices, ced_link) {
-               if (curr->ced_misc.minor == minor) {
-                       /* take the first available OBD device attached */
-                       obd = list_first_entry(&curr->ced_obds,
-                                              struct obd_device,
-                                              u.cli.cl_chg_dev_linkage);
-                       break;
-               }
-       }
-       mutex_unlock(&chlg_registered_dev_lock);
-       return obd;
-}
-
-/**
  * Open handler, initialize internal CRS state and spawn prefetch thread if
  * needed.
  * @param[in]  inode  Inode struct for the open character device.
@@ -495,19 +551,20 @@ static struct obd_device *chlg_obd_get(dev_t cdev)
 static int chlg_open(struct inode *inode, struct file *file)
 {
        struct chlg_reader_state *crs;
-       struct obd_device *obd = chlg_obd_get(inode->i_rdev);
+       struct miscdevice *misc = file->private_data;
+       struct chlg_registered_dev *dev;
        struct task_struct *task;
        int rc;
        ENTRY;
 
-       if (!obd)
-               RETURN(-ENODEV);
+       dev = container_of(misc, struct chlg_registered_dev, ced_misc);
 
        OBD_ALLOC_PTR(crs);
        if (!crs)
                RETURN(-ENOMEM);
 
-       crs->crs_obd = obd;
+       kref_get(&dev->ced_refs);
+       crs->crs_ced = dev;
        crs->crs_err = false;
        crs->crs_eof = false;
 
@@ -521,7 +578,7 @@ static int chlg_open(struct inode *inode, struct file *file)
                if (IS_ERR(task)) {
                        rc = PTR_ERR(task);
                        CERROR("%s: cannot start changelog thread: rc = %d\n",
-                              obd->obd_name, rc);
+                              dev->ced_name, rc);
                        GOTO(err_crs, rc);
                }
                crs->crs_prod_task = task;
@@ -531,6 +588,7 @@ static int chlg_open(struct inode *inode, struct file *file)
        RETURN(0);
 
 err_crs:
+       kref_put(&dev->ced_refs, chlg_dev_clear);
        OBD_FREE_PTR(crs);
        return rc;
 }
@@ -555,6 +613,7 @@ static int chlg_release(struct inode *inode, struct file *file)
        list_for_each_entry_safe(rec, tmp, &crs->crs_rec_queue, enq_linkage)
                enq_record_delete(rec);
 
+       kref_put(&crs->crs_ced->ced_refs, chlg_dev_clear);
        OBD_FREE_PTR(crs);
 
        return rc;
@@ -585,6 +644,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,
@@ -593,6 +669,7 @@ static const struct file_operations chlg_fops = {
        .open           = chlg_open,
        .release        = chlg_release,
        .poll           = chlg_poll,
+       .unlocked_ioctl = chlg_ioctl,
 };
 
 /**
@@ -625,6 +702,7 @@ chlg_registered_dev_find_by_name(const char *name)
 {
        struct chlg_registered_dev *dit;
 
+       LASSERT(mutex_is_locked(&chlg_registered_dev_lock));
        list_for_each_entry(dit, &chlg_registered_devices, ced_link)
                if (strcmp(name, dit->ced_name) == 0)
                        return dit;
@@ -643,6 +721,7 @@ chlg_registered_dev_find_by_obd(const struct obd_device *obd)
        struct chlg_registered_dev *dit;
        struct obd_device *oit;
 
+       LASSERT(mutex_is_locked(&chlg_registered_dev_lock));
        list_for_each_entry(dit, &chlg_registered_devices, ced_link)
                list_for_each_entry(oit, &dit->ced_obds,
                                    u.cli.cl_chg_dev_linkage)
@@ -688,13 +767,16 @@ int mdc_changelog_cdev_init(struct obd_device *obd)
                GOTO(out_unlock, rc = 0);
        }
 
+       list_add_tail(&obd->u.cli.cl_chg_dev_linkage, &entry->ced_obds);
+       list_add_tail(&entry->ced_link, &chlg_registered_devices);
+
        /* Register new character device */
        rc = misc_register(&entry->ced_misc);
-       if (rc != 0)
+       if (rc != 0) {
+               list_del_init(&obd->u.cli.cl_chg_dev_linkage);
+               list_del(&entry->ced_link);
                GOTO(out_unlock, rc);
-
-       list_add_tail(&obd->u.cli.cl_chg_dev_linkage, &entry->ced_obds);
-       list_add_tail(&entry->ced_link, &chlg_registered_devices);
+       }
 
        entry = NULL;   /* prevent it from being freed below */
 
@@ -706,30 +788,15 @@ out_unlock:
 }
 
 /**
- * Deregister a changelog character device whose refcount has reached zero.
- */
-static void chlg_dev_clear(struct kref *kref)
-{
-       struct chlg_registered_dev *entry = container_of(kref,
-                                                     struct chlg_registered_dev,
-                                                     ced_refs);
-       ENTRY;
-
-       list_del(&entry->ced_link);
-       misc_deregister(&entry->ced_misc);
-       OBD_FREE_PTR(entry);
-       EXIT;
-}
-
-/**
  * Release OBD, decrease reference count of the corresponding changelog device.
  */
 void mdc_changelog_cdev_finish(struct obd_device *obd)
 {
-       struct chlg_registered_dev *dev = chlg_registered_dev_find_by_obd(obd);
-       ENTRY;
+       struct chlg_registered_dev *dev;
 
+       ENTRY;
        mutex_lock(&chlg_registered_dev_lock);
+       dev = chlg_registered_dev_find_by_obd(obd);
        list_del_init(&obd->u.cli.cl_chg_dev_linkage);
        kref_put(&dev->ced_refs, chlg_dev_clear);
        mutex_unlock(&chlg_registered_dev_lock);