LU-10166 mdc: invalid free in changelog reader

Use kthread_stop() to instruct the producer thread to exit
when the device is closed, and only then: release the CRS
structure.

The previous implementation left small time windows open,
during which the producer threads could free a structure
before the consumer thread was completely done with it.

Change-Id: Id2038aa9b7fcfd2c3347f628e749f9d2c265ac6e
Signed-off-by: Henri Doreau <henri.doreau@cea.fr>
Reviewed-on: https://review.whamcloud.com/29818
Reviewed-by: John L. Hammond <john.hammond@intel.com>
Reviewed-by: Quentin Bouget <quentin.bouget@cea.fr>
Tested-by: Jenkins
Tested-by: Maloo <hpdd-maloo@intel.com>
Reviewed-by: Oleg Drokin <oleg.drokin@intel.com>

diff --git a/lustre/mdc/mdc_changelog.c b/lustre/mdc/mdc_changelog.c
index 3ae27f9..8431b1c 100644 (file)
--- a/lustre/mdc/mdc_changelog.c
+++ b/lustre/mdc/mdc_changelog.c
@@ -69,12 +69,12 @@ struct chlg_registered_dev {
 struct chlg_reader_state {
        /* Shortcut to the corresponding OBD device */
        struct obd_device       *crs_obd;
 struct chlg_reader_state {
        /* Shortcut to the corresponding OBD device */
        struct obd_device       *crs_obd;

+       /* Producer thread (if any) */
+       struct task_struct      *crs_prod_task;

        /* An error occurred that prevents from reading further */
        bool                     crs_err;
        /* EOF, no more records available */
        bool                     crs_eof;
        /* An error occurred that prevents from reading further */
        bool                     crs_err;
        /* EOF, no more records available */
        bool                     crs_eof;

-       /* Userland reader closed connection */
-       bool                     crs_closed;

        /* Desired start position */
        __u64                    crs_start_offset;
        /* Wait queue for the catalog processing thread */
        /* Desired start position */
        __u64                    crs_start_offset;
        /* Wait queue for the catalog processing thread */


@@ -154,9 +154,9 @@ static int chlg_read_cat_process_cb(const struct lu_env *env,
 
        l_wait_event(crs->crs_waitq_prod,
                     (crs->crs_rec_count < CDEV_CHLG_MAX_PREFETCH ||
 
        l_wait_event(crs->crs_waitq_prod,
                     (crs->crs_rec_count < CDEV_CHLG_MAX_PREFETCH ||

-                     crs->crs_closed), &lwi);
+                     kthread_should_stop()), &lwi);

 
 

-       if (crs->crs_closed)
+       if (kthread_should_stop())

                RETURN(LLOG_PROC_BREAK);
 
        len = changelog_rec_size(&rec->cr) + rec->cr.cr_namelen;
                RETURN(LLOG_PROC_BREAK);
 
        len = changelog_rec_size(&rec->cr) + rec->cr.cr_namelen;


@@ -188,22 +188,6 @@ static void enq_record_delete(struct chlg_rec_entry *rec)
 }
 
 /**
 }
 
 /**

- * Release resources associated to a changelog_reader_state instance.
- *
- * @param  crs  CRS instance to release.
- */
-static void crs_free(struct chlg_reader_state *crs)
-{
-       struct chlg_rec_entry *rec;
-       struct chlg_rec_entry *tmp;
-
-       list_for_each_entry_safe(rec, tmp, &crs->crs_rec_queue, enq_linkage)
-               enq_record_delete(rec);
-
-       OBD_FREE_PTR(crs);
-}
-
-/**

  * Record prefetch thread entry point. Opens the changelog catalog and starts
  * reading records.
  *
  * Record prefetch thread entry point. Opens the changelog catalog and starts
  * reading records.
  *


@@ -245,8 +229,12 @@ static int chlg_load(void *args)
                GOTO(err_out, rc);
        }
 
                GOTO(err_out, rc);
        }
 

+       crs->crs_eof = true;
+

 err_out:
 err_out:

-       crs->crs_err = true;
+       if (rc < 0)
+               crs->crs_err = true;
+

        wake_up_all(&crs->crs_waitq_cons);
 
        if (llh != NULL)
        wake_up_all(&crs->crs_waitq_cons);
 
        if (llh != NULL)


@@ -255,8 +243,8 @@ err_out:
        if (ctx != NULL)
                llog_ctxt_put(ctx);
 
        if (ctx != NULL)
                llog_ctxt_put(ctx);
 

-       l_wait_event(crs->crs_waitq_prod, crs->crs_closed, &lwi);
-       crs_free(crs);
+       l_wait_event(crs->crs_waitq_prod, kthread_should_stop(), &lwi);
+

        RETURN(rc);
 }
 
        RETURN(rc);
 }
 


@@ -511,7 +499,6 @@ static int chlg_open(struct inode *inode, struct file *file)
        crs->crs_obd = obd;
        crs->crs_err = false;
        crs->crs_eof = false;
        crs->crs_obd = obd;
        crs->crs_err = false;
        crs->crs_eof = false;

-       crs->crs_closed = false;

 
        mutex_init(&crs->crs_lock);
        INIT_LIST_HEAD(&crs->crs_rec_queue);
 
        mutex_init(&crs->crs_lock);
        INIT_LIST_HEAD(&crs->crs_rec_queue);


@@ -526,6 +513,7 @@ static int chlg_open(struct inode *inode, struct file *file)
                               obd->obd_name, rc);
                        GOTO(err_crs, rc);
                }
                               obd->obd_name, rc);
                        GOTO(err_crs, rc);
                }

+               crs->crs_prod_task = task;

        }
 
        file->private_data = crs;
        }
 
        file->private_data = crs;


@@ -546,14 +534,16 @@ err_crs:
 static int chlg_release(struct inode *inode, struct file *file)
 {
        struct chlg_reader_state *crs = file->private_data;
 static int chlg_release(struct inode *inode, struct file *file)
 {
        struct chlg_reader_state *crs = file->private_data;

+       struct chlg_rec_entry *rec;
+       struct chlg_rec_entry *tmp;

 
 

-       if (file->f_mode & FMODE_READ) {
-               crs->crs_closed = true;
-               wake_up_all(&crs->crs_waitq_prod);
-       } else {
-               /* No producer thread, release resource ourselves */
-               crs_free(crs);
-       }
+       if (crs->crs_prod_task)
+               kthread_stop(crs->crs_prod_task);
+
+       list_for_each_entry_safe(rec, tmp, &crs->crs_rec_queue, enq_linkage)
+               enq_record_delete(rec);
+
+       OBD_FREE_PTR(crs);

        return 0;
 }
 
        return 0;
 }
 


@@ -567,7 +557,7 @@ static int chlg_release(struct inode *inode, struct file *file)
  */
 static unsigned int chlg_poll(struct file *file, poll_table *wait)
 {
  */
 static unsigned int chlg_poll(struct file *file, poll_table *wait)
 {

-       struct chlg_reader_state *crs  = file->private_data;
+       struct chlg_reader_state *crs = file->private_data;

        unsigned int mask = 0;
 
        mutex_lock(&crs->crs_lock);
        unsigned int mask = 0;
 
        mutex_lock(&crs->crs_lock);