* Copyright (c) 2017, Commissariat a l'Energie Atomique et aux Energies
* Alternatives.
*
+ * Copyright (c) 2017, Intel Corporation.
+ *
* Author: Henri Doreau <henri.doreau@cea.fr>
*/
#include <linux/miscdevice.h>
#include <lustre_log.h>
+#include <uapi/linux/lustre/lustre_ioctl.h>
#include "mdc_internal.h"
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;
+ int 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 */
__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 {
struct llog_changelog_rec *rec;
struct chlg_reader_state *crs = data;
struct chlg_rec_entry *enq;
- struct l_wait_info lwi = { 0 };
size_t len;
int rc;
ENTRY;
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",
PFID(&rec->cr.cr_tfid), PFID(&rec->cr.cr_pfid),
rec->cr.cr_namelen, changelog_rec_name(&rec->cr));
- l_wait_event(crs->crs_waitq_prod,
- (crs->crs_rec_count < CDEV_CHLG_MAX_PREFETCH ||
- crs->crs_closed), &lwi);
+ wait_event_interruptible(crs->crs_waitq_prod,
+ crs->crs_rec_count < CDEV_CHLG_MAX_PREFETCH ||
+ kthread_should_stop());
- if (crs->crs_closed)
+ if (kthread_should_stop())
RETURN(LLOG_PROC_BREAK);
len = changelog_rec_size(&rec->cr) + rec->cr.cr_namelen;
}
/**
- * 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.
*
struct obd_device *obd = crs->crs_obd;
struct llog_ctxt *ctx = NULL;
struct llog_handle *llh = NULL;
- struct l_wait_info lwi = { 0 };
int rc;
ENTRY;
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) {
GOTO(err_out, rc);
}
- rc = llog_init_handle(NULL, llh, LLOG_F_IS_CAT|LLOG_F_EXT_JOBID, NULL);
+
+ 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_NID |
+ LLOG_F_EXT_X_OMODE |
+ LLOG_F_EXT_X_XATTR,
+ NULL);
if (rc) {
CERROR("%s: fail to init llog handle: rc = %d\n",
obd->obd_name, rc);
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;
err_out:
- crs->crs_err = true;
+ if (rc < 0)
+ crs->crs_err = rc;
+
wake_up_all(&crs->crs_waitq_cons);
if (llh != NULL)
if (ctx != NULL)
llog_ctxt_put(ctx);
- l_wait_event(crs->crs_waitq_prod, crs->crs_closed, &lwi);
- crs_free(crs);
+ wait_event_interruptible(crs->crs_waitq_prod, kthread_should_stop());
+
RETURN(rc);
}
struct chlg_reader_state *crs = file->private_data;
struct chlg_rec_entry *rec;
struct chlg_rec_entry *tmp;
- struct l_wait_info lwi = { 0 };
- ssize_t written_total = 0;
+ size_t written_total = 0;
+ ssize_t rc;
LIST_HEAD(consumed);
ENTRY;
- if (file->f_flags & O_NONBLOCK && crs->crs_rec_count == 0)
- RETURN(-EAGAIN);
+ if (file->f_flags & O_NONBLOCK && crs->crs_rec_count == 0) {
+ if (crs->crs_err < 0)
+ RETURN(crs->crs_err);
+ else if (crs->crs_eof)
+ RETURN(0);
+ else
+ RETURN(-EAGAIN);
+ }
- l_wait_event(crs->crs_waitq_cons,
- crs->crs_rec_count > 0 || crs->crs_eof || crs->crs_err,
- &lwi);
+ rc = wait_event_interruptible(crs->crs_waitq_cons,
+ crs->crs_rec_count > 0 || crs->crs_eof || crs->crs_err);
mutex_lock(&crs->crs_lock);
list_for_each_entry_safe(rec, tmp, &crs->crs_rec_queue, enq_linkage) {
break;
if (copy_to_user(buff, rec->enq_record, rec->enq_length)) {
- if (written_total == 0)
- written_total = -EFAULT;
+ rc = -EFAULT;
break;
}
}
mutex_unlock(&crs->crs_lock);
- if (written_total > 0)
+ if (written_total > 0) {
+ rc = written_total;
wake_up_all(&crs->crs_waitq_prod);
+ } else if (rc == 0) {
+ rc = crs->crs_err;
+ }
list_for_each_entry_safe(rec, tmp, &consumed, enq_linkage)
enq_record_delete(rec);
*ppos = crs->crs_start_offset;
- RETURN(written_total);
+ RETURN(rc);
}
/**
}
/**
- * 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.
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 obd_device *obd;
struct task_struct *task;
int rc;
ENTRY;
- if (!obd)
- RETURN(-ENODEV);
+ dev = container_of(misc, struct chlg_registered_dev, ced_misc);
+ obd = list_first_entry(&dev->ced_obds,
+ struct obd_device,
+ u.cli.cl_chg_dev_linkage);
OBD_ALLOC_PTR(crs);
if (!crs)
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);
obd->obd_name, rc);
GOTO(err_crs, rc);
}
+ crs->crs_prod_task = task;
}
file->private_data = crs;
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;
+ int rc = 0;
- 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);
- }
- return 0;
+ if (crs->crs_prod_task)
+ rc = 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 rc;
}
/**
*/
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);
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,
.open = chlg_open,
.release = chlg_release,
.poll = chlg_poll,
+ .unlocked_ioctl = chlg_ioctl,
};
/**
{
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;
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)
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 */
ced_refs);
ENTRY;
+ LASSERT(mutex_is_locked(&chlg_registered_dev_lock));
list_del(&entry->ced_link);
misc_deregister(&entry->ced_misc);
OBD_FREE_PTR(entry);
*/
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);
git://git.whamcloud.com / fs / lustre-release.git / blobdiff