From e650c53d3b734199c202f1c6cd868c812e0db2ed Mon Sep 17 00:00:00 2001 From: Jinshan Xiong Date: Fri, 23 Mar 2018 05:57:40 +0000 Subject: [PATCH 3/4] cleanup and bugfix --- libcfs/include/libcfs/libcfs_debug.h | 8 +- lustre/include/lustre_compat.h | 2 +- lustre/llite/lloop.c | 294 +++++++++++++++++------------------ lustre/osc/osc_io.c | 2 + lustre/utils/obd.c | 2 +- 5 files changed, 148 insertions(+), 160 deletions(-) diff --git a/libcfs/include/libcfs/libcfs_debug.h b/libcfs/include/libcfs/libcfs_debug.h index bf8ea34..8fa8138 100644 --- a/libcfs/include/libcfs/libcfs_debug.h +++ b/libcfs/include/libcfs/libcfs_debug.h @@ -100,6 +100,7 @@ struct ptldebug_header { #define S_LND 0x00000800 /* ALL LNDs */ #define S_PINGER 0x00001000 #define S_FILTER 0x00002000 +#define S_LLOOP 0x00004000 /* unused */ #define S_ECHO 0x00008000 #define S_LDLM 0x00010000 @@ -121,9 +122,10 @@ struct ptldebug_header { #define LIBCFS_DEBUG_SUBSYS_NAMES { \ "undefined", "mdc", "mds", "osc", "ost", "class", "log", \ - "llite", "rpc", "mgmt", "lnet", "lnd", "pinger", "filter", "", \ - "echo", "ldlm", "lov", "lquota", "osd", "lfsck", "snapshot", "",\ - "lmv", "", "sec", "gss", "", "mgc", "mgs", "fid", "fld", NULL } + "llite", "rpc", "mgmt", "lnet", "lnd", "pinger", "filter", \ + "lloop", "echo", "ldlm", "lov", "lquota", "osd", "lfsck", \ + "snapshot", "", "lmv", "", "sec", "gss", "", "mgc", "mgs", \ + "fid", "fld", NULL } /* Debugging masks (32 bits, non-overlapping) */ #define D_TRACE 0x00000001 /* ENTRY/EXIT markers */ diff --git a/lustre/include/lustre_compat.h b/lustre/include/lustre_compat.h index 5ab62b8..8b2f25e 100644 --- a/lustre/include/lustre_compat.h +++ b/lustre/include/lustre_compat.h @@ -322,7 +322,7 @@ static inline int ll_namei_to_lookup_intent_flag(int flag) #else # ifdef HAVE_VOID_MAKE_REQUEST_FN # define ll_mrf_ret void -# define LL_MRF_RETURN(rc) +# define LL_MRF_RETURN(rc) RETURN_EXIT # else # define ll_mrf_ret int # define LL_MRF_RETURN(rc) RETURN(rc) diff --git a/lustre/llite/lloop.c b/lustre/llite/lloop.c index b7b1db6..83efcaf 100644 --- a/lustre/llite/lloop.c +++ b/lustre/llite/lloop.c @@ -79,6 +79,7 @@ * */ +#define DEBUG_SUBSYSTEM S_LLOOP #include #include @@ -107,46 +108,53 @@ #include #include "llite_internal.h" -#define LLOOP_MAX_SEGMENTS LNET_MAX_IOV +#define LLOOP_MAX_SEGMENTS LNET_MAX_IOV +#define LLOOP_MAX_PAGES (LLOOP_MAX_SEGMENTS * 4) /* Possible states of device */ -enum { +enum lloop_state { LLOOP_UNBOUND, LLOOP_BOUND, LLOOP_RUNDOWN, }; -struct lloop_device { - int lo_number; - int lo_refcnt; - loff_t lo_offset; - loff_t lo_sizelimit; - int lo_flags; - struct file *lo_backing_file; - struct block_device *lo_device; - unsigned lo_blocksize; +struct lloop_bio_list { + spinlock_t bl_lock; + atomic_t bl_count; + struct bio *bl_head; + struct bio **bl_tail; +}; - gfp_t old_gfp_mask; +struct lloop_device { + int lo_number; + atomic_t lo_refcnt; + loff_t lo_offset; + loff_t lo_sizelimit; + struct file *lo_backing_file; + struct block_device *lo_device; + unsigned lo_blocksize; + unsigned lo_flags; + + gfp_t lo_gfp; + + enum lloop_state lo_state; + atomic_t lo_pending; spinlock_t lo_lock; - struct bio *lo_bio; - struct bio *lo_biotail; - int lo_state; - struct semaphore lo_sem; - struct mutex lo_ctl_mutex; - atomic_t lo_pending; + struct lloop_bio_list lo_reqs; + struct completion lo_completion; wait_queue_head_t lo_bh_wait; - struct request_queue *lo_queue; + struct request_queue *lo_queue; - const struct lu_env *lo_env; - struct cl_io lo_io; - struct ll_dio_pages lo_pvec; + const struct lu_env *lo_env; + struct cl_io lo_io; + struct ll_dio_pages lo_pvec; /* data to handle bio for lustre. */ struct lo_request_data { - struct page *lrd_pages[LLOOP_MAX_SEGMENTS]; - loff_t lrd_offsets[LLOOP_MAX_SEGMENTS]; + struct page *lrd_pages[LLOOP_MAX_PAGES]; + loff_t lrd_offsets[LLOOP_MAX_PAGES]; } lo_requests[1]; }; @@ -165,6 +173,24 @@ static struct gendisk **disks; static struct mutex lloop_mutex; static void *ll_iocontrol_magic = NULL; +static inline void lloop_bio_list_init(struct lloop_bio_list *bl) +{ + memset(bl, 0, sizeof(*bl)); + + atomic_set(&bl->bl_count, 0); + spin_lock_init(&bl->bl_lock); + bl->bl_tail = &bl->bl_head; +} + +static inline sector_t bio_sector(struct bio *bio) +{ +#ifdef HAVE_BVEC_ITER + return bio->bi_iter.bi_sector; +#else + return bio->bi_sector; +#endif +} + static loff_t get_loop_size(struct lloop_device *lo, struct file *file) { loff_t size, offset, loopsize; @@ -183,14 +209,18 @@ static loff_t get_loop_size(struct lloop_device *lo, struct file *file) return loopsize >> 9; } +static int loop_bio_rw(struct bio *bio) +{ + return bio_rw(bio) == WRITE ? WRITE : READ; +} + static int do_bio_lustrebacked(struct lloop_device *lo, struct bio *head) { const struct lu_env *env = lo->lo_env; struct cl_io *io = &lo->lo_io; - struct dentry *de = lo->lo_backing_file->f_path.dentry; - struct inode *inode = de->d_inode; + struct inode *inode = file_inode(lo->lo_backing_file); struct cl_object *obj = ll_i2info(inode)->lli_clob; - pgoff_t offset; + loff_t offset; int ret; int iter; struct bio_vec *bvec; @@ -200,9 +230,8 @@ static int do_bio_lustrebacked(struct lloop_device *lo, struct bio *head) ssize_t bytes; struct ll_dio_pages *pvec = &lo->lo_pvec; - struct page **pages = pvec->ldp_pages; - loff_t *offsets = pvec->ldp_offsets; + vfs_fsync(lo->lo_backing_file, 0); truncate_inode_pages(inode->i_mapping, 0); /* initialize the IO */ @@ -214,25 +243,21 @@ static int do_bio_lustrebacked(struct lloop_device *lo, struct bio *head) io->ci_lockreq = CILR_NEVER; LASSERT(head != NULL); - rw = head->bi_rw; + rw = loop_bio_rw(head); for (bio = head; bio != NULL; bio = bio->bi_next) { - LASSERT(rw == bio->bi_rw); + LASSERT(rw == loop_bio_rw(bio)); -#ifdef HAVE_BVEC_ITER - offset = (pgoff_t)(bio->bi_iter.bi_sector << 9) + lo->lo_offset; -#else - offset = (pgoff_t)(bio->bi_sector << 9) + lo->lo_offset; -#endif + offset = (bio_sector(bio) << 9) + lo->lo_offset; bio_for_each_segment_all(bvec, bio, iter) { BUG_ON(bvec->bv_offset != 0); BUG_ON(bvec->bv_len != PAGE_CACHE_SIZE); - pages[page_count] = bvec->bv_page; - offsets[page_count] = offset; + pvec->ldp_pages[page_count] = bvec->bv_page; + pvec->ldp_offsets[page_count] = offset; page_count++; offset += bvec->bv_len; } - LASSERT(page_count <= LLOOP_MAX_SEGMENTS); + LASSERT(page_count <= LLOOP_MAX_PAGES); } ll_stats_ops_tally(ll_i2sbi(inode), @@ -242,48 +267,28 @@ static int do_bio_lustrebacked(struct lloop_device *lo, struct bio *head) pvec->ldp_size = page_count << PAGE_CACHE_SHIFT; pvec->ldp_nr = page_count; - /* FIXME: in ll_direct_rw_pages, it has to allocate many cl_page{}s to - * write those pages into OST. Even worse case is that more pages - * would be asked to write out to swap space, and then finally get here - * again. - * Unfortunately this is NOT easy to fix. - * Thoughts on solution: - * 0. Define a reserved pool for cl_pages, which could be a list of - * pre-allocated cl_pages; - * 1. Define a new operation in cl_object_operations{}, says clo_depth, - * which measures how many layers for this lustre object. Generally - * speaking, the depth would be 2, one for llite, and one for lovsub. - * However, for SNS, there will be more since we need additional page - * to store parity; - * 2. Reserve the # of (page_count * depth) cl_pages from the reserved - * pool. Afterwards, the clio would allocate the pages from reserved - * pool, this guarantees we neeedn't allocate the cl_pages from - * generic cl_page slab cache. - * Of course, if there is NOT enough pages in the pool, we might - * be asked to write less pages once, this purely depends on - * implementation. Anyway, we should be careful to avoid deadlocking. - */ + CDEBUG(D_INFO, "%s pvec pages: %zd\n", rw ? "Writing" : "Reading", page_count); mutex_lock(&inode->i_mutex); bytes = ll_direct_rw_pages(env, io, rw, inode, pvec); mutex_unlock(&inode->i_mutex); cl_io_fini(env, io); + CDEBUG(D_INFO, "I/O to pvec done, bytes: %zd\n", bytes); return (bytes == pvec->ldp_size) ? 0 : (int)bytes; } /* * Add bio to back of pending list */ -static void loop_add_bio(struct lloop_device *lo, struct bio *bio) +static void loop_add_bio(struct lloop_device *lo, struct lloop_bio_list *bl, + struct bio *bio) { unsigned long flags; - spin_lock_irqsave(&lo->lo_lock, flags); - if (lo->lo_biotail) { - lo->lo_biotail->bi_next = bio; - lo->lo_biotail = bio; - } else - lo->lo_bio = lo->lo_biotail = bio; - spin_unlock_irqrestore(&lo->lo_lock, flags); + spin_lock_irqsave(&bl->bl_lock, flags); + *bl->bl_tail = bio; + bl->bl_tail = &bio->bi_next; + atomic_inc(&bl->bl_count); + spin_unlock_irqrestore(&bl->bl_lock, flags); atomic_inc(&lo->lo_pending); if (waitqueue_active(&lo->lo_bh_wait)) @@ -293,75 +298,59 @@ static void loop_add_bio(struct lloop_device *lo, struct bio *bio) /* * Grab first pending buffer */ -static unsigned int loop_get_bio(struct lloop_device *lo, struct bio **req) +static unsigned int loop_get_bio(struct lloop_bio_list *bl, struct bio **req) { - struct bio *first; struct bio **bio; unsigned int count = 0; unsigned int page_count = 0; int rw; - spin_lock_irq(&lo->lo_lock); - first = lo->lo_bio; - if (unlikely(first == NULL)) { - spin_unlock_irq(&lo->lo_lock); - return 0; - } + spin_lock_irq(&bl->bl_lock); + bio = &bl->bl_head; + while (*bio) { + CDEBUG(D_INFO, "bio sector %lu count %u vcnt %u, rw: %d\n", + bio_sector(*bio), page_count, (*bio)->bi_vcnt, + bio_rw(*bio)); + + if (!*req) { /* first request */ + *req = *bio; + rw = loop_bio_rw(*bio); + } - /* TODO: need to split the bio, too bad. */ - LASSERT(first->bi_vcnt <= LLOOP_MAX_SEGMENTS); + if (loop_bio_rw(*bio) != rw) + break; - rw = first->bi_rw; - bio = &lo->lo_bio; - while (*bio && (*bio)->bi_rw == rw) { -#ifdef HAVE_BVEC_ITER - CDEBUG(D_INFO, "bio sector %llu size %u count %u vcnt%u \n", - (unsigned long long)(*bio)->bi_iter.bi_sector, - (*bio)->bi_iter.bi_size, page_count, (*bio)->bi_vcnt); -#else - CDEBUG(D_INFO, "bio sector %llu size %u count %u vcnt%u \n", - (unsigned long long)(*bio)->bi_sector, (*bio)->bi_size, - page_count, (*bio)->bi_vcnt); -#endif - if (page_count + (*bio)->bi_vcnt > LLOOP_MAX_SEGMENTS) + LASSERT((*bio)->bi_vcnt <= LLOOP_MAX_PAGES); + if (page_count + (*bio)->bi_vcnt > LLOOP_MAX_PAGES) break; page_count += (*bio)->bi_vcnt; count++; bio = &(*bio)->bi_next; } - if (*bio) { - /* Some of bios can't be mergable. */ - lo->lo_bio = *bio; - *bio = NULL; - } else { - /* Hit the end of queue */ - lo->lo_biotail = NULL; - lo->lo_bio = NULL; - } - *req = first; - spin_unlock_irq(&lo->lo_lock); + + atomic_sub(count, &bl->bl_count); + bl->bl_head = *bio; + if (!*bio) /* Hit the end of queue */ + bl->bl_tail = &bl->bl_head; + *bio = NULL; + spin_unlock_irq(&bl->bl_lock); + return count; } static ll_mrf_ret -loop_make_request(struct request_queue *q, struct bio *old_bio) +loop_make_request(struct request_queue *q, struct bio *bio) { struct lloop_device *lo = q->queuedata; - int rw = bio_rw(old_bio); + int rw = bio_rw(bio); int inactive; if (!lo) goto err; -#ifdef HAVE_BVEC_ITER - CDEBUG(D_INFO, "submit bio sector %llu size %u\n", - (unsigned long long)old_bio->bi_iter.bi_sector, - old_bio->bi_iter.bi_size); -#else - CDEBUG(D_INFO, "submit bio sector %llu size %u\n", - (unsigned long long)old_bio->bi_sector, old_bio->bi_size); -#endif + CDEBUG(D_INFO, "submit bio sector %lu pages: %u, rw: %d\n", + bio_sector(bio), bio->bi_vcnt, rw); spin_lock_irq(&lo->lo_lock); inactive = (lo->lo_state != LLOOP_BOUND); @@ -369,19 +358,13 @@ loop_make_request(struct request_queue *q, struct bio *old_bio) if (inactive) goto err; - if (rw == WRITE) { - if (lo->lo_flags & LO_FLAGS_READ_ONLY) - goto err; - } else if (rw == READA) { - rw = READ; - } else if (rw != READ) { - CERROR("lloop: unknown command (%x)\n", rw); - goto err; - } - loop_add_bio(lo, old_bio); + if (rw == WRITE && lo->lo_flags & LO_FLAGS_READ_ONLY) + goto err; + + loop_add_bio(lo, &lo->lo_reqs, bio); LL_MRF_RETURN(0); err: - bio_io_error(old_bio); + bio_io_error(bio); LL_MRF_RETURN(0); } @@ -402,7 +385,7 @@ static inline void loop_handle_bio(struct lloop_device *lo, struct bio *bio) { int ret; - ret = do_bio_lustrebacked(lo, bio); + ret = do_bio_lustrebacked(lo, bio); while (bio) { struct bio *tmp = bio->bi_next; @@ -419,8 +402,7 @@ static inline void loop_handle_bio(struct lloop_device *lo, struct bio *bio) static inline int loop_active(struct lloop_device *lo) { - return atomic_read(&lo->lo_pending) || - (lo->lo_state == LLOOP_RUNDOWN); + return atomic_read(&lo->lo_pending) || lo->lo_state == LLOOP_RUNDOWN; } /* @@ -441,8 +423,6 @@ static int loop_thread(void *data) set_user_nice(current, -20); - lo->lo_state = LLOOP_BOUND; - env = cl_env_get(&refcheck); if (IS_ERR(env)) GOTO(out, ret = PTR_ERR(env)); @@ -452,15 +432,14 @@ static int loop_thread(void *data) lo->lo_pvec.ldp_pages = lo->lo_requests[0].lrd_pages; lo->lo_pvec.ldp_offsets = lo->lo_requests[0].lrd_offsets; - /* - * up sem, we are running - */ - up(&lo->lo_sem); + lo->lo_state = LLOOP_BOUND; + complete(&lo->lo_completion); for (;;) { wait_event(lo->lo_bh_wait, loop_active(lo)); if (!atomic_read(&lo->lo_pending)) { - int exiting = 0; + bool exiting; + spin_lock_irq(&lo->lo_lock); exiting = (lo->lo_state == LLOOP_RUNDOWN); spin_unlock_irq(&lo->lo_lock); @@ -469,7 +448,7 @@ static int loop_thread(void *data) } bio = NULL; - count = loop_get_bio(lo, &bio); + count = loop_get_bio(&lo->lo_reqs, &bio); if (!count) { CWARN("lloop(minor: %d): missing bio\n", lo->lo_number); continue; @@ -491,11 +470,14 @@ static int loop_thread(void *data) LASSERT(count <= atomic_read(&lo->lo_pending)); loop_handle_bio(lo, bio); atomic_sub(count, &lo->lo_pending); + + if (need_resched()) + cond_resched(); } cl_env_put(env, &refcheck); out: - up(&lo->lo_sem); + complete(&lo->lo_completion); return ret; } @@ -542,10 +524,10 @@ static int loop_set_fd(struct lloop_device *lo, struct file *unused, lo->lo_flags = lo_flags; lo->lo_backing_file = file; lo->lo_sizelimit = 0; - lo->old_gfp_mask = mapping_gfp_mask(mapping); - mapping_set_gfp_mask(mapping, lo->old_gfp_mask & ~(__GFP_IO|__GFP_FS)); + lo->lo_gfp = mapping_gfp_mask(mapping); + mapping_set_gfp_mask(mapping, lo->lo_gfp & ~(__GFP_IO|__GFP_FS)); - lo->lo_bio = lo->lo_biotail = NULL; + lloop_bio_list_init(&lo->lo_reqs); /* * set queue make_request_fn, and add limits based on lower level @@ -558,8 +540,9 @@ static int loop_set_fd(struct lloop_device *lo, struct file *unused, #endif /* queue parameters */ + blk_queue_logical_block_size(lo->lo_queue, PAGE_CACHE_SIZE); blk_queue_max_hw_sectors(lo->lo_queue, - LLOOP_MAX_SEGMENTS << (PAGE_CACHE_SHIFT - 9)); + LLOOP_MAX_PAGES << (PAGE_CACHE_SHIFT - 9)); blk_queue_max_segments(lo->lo_queue, LLOOP_MAX_SEGMENTS); set_capacity(disks[lo->lo_number], size); @@ -568,7 +551,7 @@ static int loop_set_fd(struct lloop_device *lo, struct file *unused, set_blocksize(bdev, lo->lo_blocksize); kthread_run(loop_thread, lo, "lloop%d", lo->lo_number); - down(&lo->lo_sem); + wait_for_completion(&lo->lo_completion); return 0; out: @@ -581,23 +564,28 @@ static int loop_clr_fd(struct lloop_device *lo, struct block_device *bdev, int count) { struct file *filp = lo->lo_backing_file; - gfp_t gfp = lo->old_gfp_mask; + gfp_t gfp = lo->lo_gfp; if (lo->lo_state != LLOOP_BOUND) return -ENXIO; - if (lo->lo_refcnt > count) /* we needed one fd for the ioctl */ + /* we needed one fd for the ioctl */ + if (atomic_read(&lo->lo_refcnt) > count) { + CERROR("refcnt = %d, expect = %d\n", atomic_read(&lo->lo_refcnt), count); return -EBUSY; + } - if (filp == NULL) + if (!filp) return -EINVAL; + reinit_completion(&lo->lo_completion); + spin_lock_irq(&lo->lo_lock); lo->lo_state = LLOOP_RUNDOWN; spin_unlock_irq(&lo->lo_lock); wake_up(&lo->lo_bh_wait); - down(&lo->lo_sem); + wait_for_completion(&lo->lo_completion); lo->lo_backing_file = NULL; lo->lo_device = NULL; lo->lo_offset = 0; @@ -618,9 +606,7 @@ static int lo_open(struct block_device *bdev, fmode_t mode) { struct lloop_device *lo = bdev->bd_disk->private_data; - mutex_lock(&lo->lo_ctl_mutex); - lo->lo_refcnt++; - mutex_unlock(&lo->lo_ctl_mutex); + atomic_inc(&lo->lo_refcnt); return 0; } @@ -634,9 +620,8 @@ lo_release(struct gendisk *disk, fmode_t mode) { struct lloop_device *lo = disk->private_data; - mutex_lock(&lo->lo_ctl_mutex); - --lo->lo_refcnt; - mutex_unlock(&lo->lo_ctl_mutex); + atomic_dec(&lo->lo_refcnt); + #ifdef HAVE_BLKDEV_RELEASE_RETURN_INT return 0; #endif @@ -653,7 +638,7 @@ static int lo_ioctl(struct block_device *bdev, fmode_t mode, switch (cmd) { case LL_IOC_LLOOP_DETACH: { err = loop_clr_fd(lo, bdev, 2); - if (err == 0) + if (!err) blkdev_put(bdev, 0); /* grabbed in LLOOP_ATTACH */ break; } @@ -777,7 +762,7 @@ static enum llioc_iter lloop_ioctl(struct inode *unused, struct file *file, bdev = lo->lo_device; err = loop_clr_fd(lo, bdev, 1); - if (err == 0) + if (!err) blkdev_put(bdev, 0); /* grabbed in LLOOP_ATTACH */ break; @@ -845,8 +830,8 @@ static int __init lloop_init(void) if (!lo->lo_queue) goto out_mem4; - mutex_init(&lo->lo_ctl_mutex); - sema_init(&lo->lo_sem, 0); + atomic_set(&lo->lo_refcnt, 0); + init_completion(&lo->lo_completion); init_waitqueue_head(&lo->lo_bh_wait); lo->lo_number = i; spin_lock_init(&lo->lo_lock); @@ -876,7 +861,6 @@ out_mem2: out_mem1: unregister_blkdev(lloop_major, "lloop"); ll_iocontrol_unregister(ll_iocontrol_magic); - CERROR("lloop: ran out of memory\n"); return -ENOMEM; } diff --git a/lustre/osc/osc_io.c b/lustre/osc/osc_io.c index c6285fd..ee35c9d 100644 --- a/lustre/osc/osc_io.c +++ b/lustre/osc/osc_io.c @@ -1057,6 +1057,7 @@ static void osc_req_attr_set(const struct lu_env *env, opg = osc_cl_page_osc(apage, NULL); lock = osc_dlmlock_at_pgoff(env, cl2osc(obj), osc_index(opg), OSC_DAP_FL_TEST_LOCK | OSC_DAP_FL_CANCELING); +#if 0 if (lock == NULL && !opg->ops_srvlock) { struct ldlm_resource *res; struct ldlm_res_id *resname; @@ -1073,6 +1074,7 @@ static void osc_req_attr_set(const struct lu_env *env, libcfs_debug_dumpstack(NULL); LBUG(); } +#endif /* check for lockless io. */ if (lock != NULL) { diff --git a/lustre/utils/obd.c b/lustre/utils/obd.c index 54f55d2..ea7c510 100644 --- a/lustre/utils/obd.c +++ b/lustre/utils/obd.c @@ -2906,7 +2906,7 @@ int jt_blockdev_attach(int argc, char **argv) filename = argv[1]; devname = argv[2]; - fd = open(filename, O_RDWR); + fd = open(filename, O_RDWR | O_DIRECT); if (fd < 0) { fprintf(stderr, "file %s can't be opened(%s)\n\n", filename, strerror(errno)); -- 1.9.1