X-Git-Url: https://git.whamcloud.com/?p=fs%2Flustre-release.git;a=blobdiff_plain;f=lustre%2Fmdc%2Fmdc_changelog.c;h=fb9fa54adc38ccc3f9fe5980ee750bbd93ee9cca;hp=252592c74eb8c432bb613959799766b5239440c1;hb=5c883ea2748ae9e430a9cd863a9b630b2a74440a;hpb=d2629cfcabfa3a22ddf4a6c474364d0012d80390 diff --git a/lustre/mdc/mdc_changelog.c b/lustre/mdc/mdc_changelog.c index 252592c..fb9fa54 100644 --- a/lustre/mdc/mdc_changelog.c +++ b/lustre/mdc/mdc_changelog.c @@ -36,6 +36,7 @@ #include #include +#include #include "mdc_internal.h" @@ -70,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 { @@ -106,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. @@ -133,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", @@ -198,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); @@ -216,12 +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_NID, + 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", @@ -229,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; @@ -249,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); @@ -406,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; + + obd = chlg_obd_get(crs->crs_ced); + if (obd == NULL) + return -ENODEV; - return obd_set_info_async(NULL, obd->obd_self_export, - strlen(KEY_CHANGELOG_CLEAR), - KEY_CHANGELOG_CLEAR, sizeof(cs), &cs, NULL); + 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 */ @@ -464,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. @@ -498,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; @@ -524,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; @@ -534,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; } @@ -558,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; @@ -588,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, @@ -596,6 +669,7 @@ static const struct file_operations chlg_fops = { .open = chlg_open, .release = chlg_release, .poll = chlg_poll, + .unlocked_ioctl = chlg_ioctl, }; /** @@ -628,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; @@ -646,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) @@ -691,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 */ @@ -709,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);