X-Git-Url: https://git.whamcloud.com/?a=blobdiff_plain;f=lustre%2Fobdclass%2Fcl_io.c;h=8878ed89516cf7156c6af0f23b938d925878ff79;hb=refs%2Fchanges%2F25%2F10525%2F5;hp=9ebc40de285e4fcf2998158c48909165e709b1e9;hpb=e8ffe16619baf1ef7c5c6b117d338956372aa752;p=fs%2Flustre-release.git diff --git a/lustre/obdclass/cl_io.c b/lustre/obdclass/cl_io.c index 9ebc40d..8878ed8 100644 --- a/lustre/obdclass/cl_io.c +++ b/lustre/obdclass/cl_io.c @@ -1,6 +1,4 @@ -/* -*- mode: c; c-basic-offset: 8; indent-tabs-mode: nil; -*- - * vim:expandtab:shiftwidth=8:tabstop=8: - * +/* * GPL HEADER START * * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER. @@ -28,6 +26,8 @@ /* * Copyright (c) 2008, 2010, Oracle and/or its affiliates. All rights reserved. * Use is subject to license terms. + * + * Copyright (c) 2011, 2013, Intel Corporation. */ /* * This file is part of Lustre, http://www.lustre.org/ @@ -36,20 +36,15 @@ * Client IO. * * Author: Nikita Danilov + * Author: Jinshan Xiong */ #define DEBUG_SUBSYSTEM S_CLASS -#ifndef EXPORT_SYMTAB -# define EXPORT_SYMTAB -#endif #include #include #include #include -/* lu_time_global_{init,fini}() */ -#include - #include #include "cl_internal.h" @@ -60,9 +55,9 @@ */ #define cl_io_for_each(slice, io) \ - cfs_list_for_each_entry((slice), &io->ci_layers, cis_linkage) + list_for_each_entry((slice), &io->ci_layers, cis_linkage) #define cl_io_for_each_reverse(slice, io) \ - cfs_list_for_each_entry_reverse((slice), &io->ci_layers, cis_linkage) + list_for_each_entry_reverse((slice), &io->ci_layers, cis_linkage) static inline int cl_io_type_is_valid(enum cl_io_type type) { @@ -107,17 +102,17 @@ static int cl_io_invariant(const struct cl_io *io) */ void cl_io_fini(const struct lu_env *env, struct cl_io *io) { - struct cl_io_slice *slice; - struct cl_thread_info *info; + struct cl_io_slice *slice; + struct cl_thread_info *info; LINVRNT(cl_io_type_is_valid(io->ci_type)); LINVRNT(cl_io_invariant(io)); ENTRY; - while (!cfs_list_empty(&io->ci_layers)) { - slice = container_of(io->ci_layers.next, struct cl_io_slice, + while (!list_empty(&io->ci_layers)) { + slice = container_of(io->ci_layers.prev, struct cl_io_slice, cis_linkage); - cfs_list_del_init(&slice->cis_linkage); + list_del_init(&slice->cis_linkage); if (slice->cis_iop->op[io->ci_type].cio_fini != NULL) slice->cis_iop->op[io->ci_type].cio_fini(env, slice); /* @@ -131,7 +126,26 @@ void cl_io_fini(const struct lu_env *env, struct cl_io *io) info = cl_env_info(env); if (info->clt_current_io == io) info->clt_current_io = NULL; - EXIT; + + /* sanity check for layout change */ + switch(io->ci_type) { + case CIT_READ: + case CIT_WRITE: + break; + case CIT_FAULT: + case CIT_FSYNC: + LASSERT(!io->ci_need_restart); + break; + case CIT_SETATTR: + case CIT_MISC: + /* Check ignore layout change conf */ + LASSERT(ergo(io->ci_ignore_layout || !io->ci_verify_layout, + !io->ci_need_restart)); + break; + default: + LBUG(); + } + EXIT; } EXPORT_SYMBOL(cl_io_fini); @@ -147,10 +161,10 @@ static int cl_io_init0(const struct lu_env *env, struct cl_io *io, ENTRY; io->ci_type = iot; - CFS_INIT_LIST_HEAD(&io->ci_lockset.cls_todo); - CFS_INIT_LIST_HEAD(&io->ci_lockset.cls_curr); - CFS_INIT_LIST_HEAD(&io->ci_lockset.cls_done); - CFS_INIT_LIST_HEAD(&io->ci_layers); + INIT_LIST_HEAD(&io->ci_lockset.cls_todo); + INIT_LIST_HEAD(&io->ci_lockset.cls_curr); + INIT_LIST_HEAD(&io->ci_lockset.cls_done); + INIT_LIST_HEAD(&io->ci_layers); result = 0; cl_object_for_each(scan, obj) { @@ -285,34 +299,33 @@ static void cl_io_locks_sort(struct cl_io *io) done = 1; prev = NULL; - cfs_list_for_each_entry_safe(curr, temp, - &io->ci_lockset.cls_todo, - cill_linkage) { - if (prev != NULL) { - switch (cl_lock_descr_sort(&prev->cill_descr, - &curr->cill_descr)) { - case 0: - /* - * IMPOSSIBLE: Identical locks are - * already removed at - * this point. - */ - default: - LBUG(); - case +1: - cfs_list_move_tail(&curr->cill_linkage, - &prev->cill_linkage); - done = 0; - continue; /* don't change prev: it's - * still "previous" */ - case -1: /* already in order */ - break; - } - } - prev = curr; - } - } while (!done); - EXIT; + list_for_each_entry_safe(curr, temp, &io->ci_lockset.cls_todo, + cill_linkage) { + if (prev != NULL) { + switch (cl_lock_descr_sort(&prev->cill_descr, + &curr->cill_descr)) { + case 0: + /* + * IMPOSSIBLE: Identical locks are + * already removed at + * this point. + */ + default: + LBUG(); + case +1: + list_move_tail(&curr->cill_linkage, + &prev->cill_linkage); + done = 0; + continue; /* don't change prev: it's + * still "previous" */ + case -1: /* already in order */ + break; + } + } + prev = curr; + } + } while (!done); + EXIT; } /** @@ -321,37 +334,36 @@ static void cl_io_locks_sort(struct cl_io *io) * \retval +ve there is a matching lock in the \a queue * \retval 0 there are no matching locks in the \a queue */ -int cl_queue_match(const cfs_list_t *queue, +int cl_queue_match(const struct list_head *queue, const struct cl_lock_descr *need) { - struct cl_io_lock_link *scan; + struct cl_io_lock_link *scan; + ENTRY; - ENTRY; - cfs_list_for_each_entry(scan, queue, cill_linkage) { - if (cl_lock_descr_match(&scan->cill_descr, need)) - RETURN(+1); - } - RETURN(0); + list_for_each_entry(scan, queue, cill_linkage) { + if (cl_lock_descr_match(&scan->cill_descr, need)) + RETURN(+1); + } + RETURN(0); } EXPORT_SYMBOL(cl_queue_match); -static int cl_queue_merge(const cfs_list_t *queue, +static int cl_queue_merge(const struct list_head *queue, const struct cl_lock_descr *need) { - struct cl_io_lock_link *scan; - - ENTRY; - cfs_list_for_each_entry(scan, queue, cill_linkage) { - if (cl_lock_descr_cmp(&scan->cill_descr, need)) - continue; - cl_lock_descr_merge(&scan->cill_descr, need); - CDEBUG(D_VFSTRACE, "lock: %d: [%lu, %lu]\n", - scan->cill_descr.cld_mode, scan->cill_descr.cld_start, - scan->cill_descr.cld_end); - RETURN(+1); - } - RETURN(0); + struct cl_io_lock_link *scan; + ENTRY; + list_for_each_entry(scan, queue, cill_linkage) { + if (cl_lock_descr_cmp(&scan->cill_descr, need)) + continue; + cl_lock_descr_merge(&scan->cill_descr, need); + CDEBUG(D_VFSTRACE, "lock: %d: [%lu, %lu]\n", + scan->cill_descr.cld_mode, scan->cill_descr.cld_start, + scan->cill_descr.cld_end); + RETURN(+1); + } + RETURN(0); } static int cl_lockset_match(const struct cl_lockset *set, @@ -377,21 +389,15 @@ static int cl_lockset_lock_one(const struct lu_env *env, ENTRY; - if (io->ci_lockreq == CILR_PEEK) { - lock = cl_lock_peek(env, io, &link->cill_descr, "io", io); - if (lock == NULL) - lock = ERR_PTR(-ENODATA); - } else - lock = cl_lock_request(env, io, &link->cill_descr, "io", io); + lock = cl_lock_request(env, io, &link->cill_descr, "io", io); if (!IS_ERR(lock)) { link->cill_lock = lock; - cfs_list_move(&link->cill_linkage, &set->cls_curr); + list_move(&link->cill_linkage, &set->cls_curr); if (!(link->cill_descr.cld_enq_flags & CEF_ASYNC)) { result = cl_wait(env, lock); if (result == 0) - cfs_list_move(&link->cill_linkage, - &set->cls_done); + list_move(&link->cill_linkage, &set->cls_done); } else result = 0; } else @@ -405,7 +411,7 @@ static void cl_lock_link_fini(const struct lu_env *env, struct cl_io *io, struct cl_lock *lock = link->cill_lock; ENTRY; - cfs_list_del_init(&link->cill_linkage); + list_del_init(&link->cill_linkage); if (lock != NULL) { cl_lock_release(env, lock, "io", io); link->cill_lock = NULL; @@ -418,36 +424,35 @@ static void cl_lock_link_fini(const struct lu_env *env, struct cl_io *io, static int cl_lockset_lock(const struct lu_env *env, struct cl_io *io, struct cl_lockset *set) { - struct cl_io_lock_link *link; - struct cl_io_lock_link *temp; - struct cl_lock *lock; - int result; - - ENTRY; - result = 0; - cfs_list_for_each_entry_safe(link, temp, &set->cls_todo, cill_linkage) { - if (!cl_lockset_match(set, &link->cill_descr)) { - /* XXX some locking to guarantee that locks aren't - * expanded in between. */ - result = cl_lockset_lock_one(env, io, set, link); - if (result != 0) - break; - } else - cl_lock_link_fini(env, io, link); - } - if (result == 0) { - cfs_list_for_each_entry_safe(link, temp, - &set->cls_curr, cill_linkage) { - lock = link->cill_lock; - result = cl_wait(env, lock); - if (result == 0) - cfs_list_move(&link->cill_linkage, - &set->cls_done); - else - break; - } - } - RETURN(result); + struct cl_io_lock_link *link; + struct cl_io_lock_link *temp; + struct cl_lock *lock; + int result; + + ENTRY; + result = 0; + list_for_each_entry_safe(link, temp, &set->cls_todo, cill_linkage) { + if (!cl_lockset_match(set, &link->cill_descr)) { + /* XXX some locking to guarantee that locks aren't + * expanded in between. */ + result = cl_lockset_lock_one(env, io, set, link); + if (result != 0) + break; + } else + cl_lock_link_fini(env, io, link); + } + if (result == 0) { + list_for_each_entry_safe(link, temp, &set->cls_curr, + cill_linkage) { + lock = link->cill_lock; + result = cl_wait(env, lock); + if (result == 0) + list_move(&link->cill_linkage, &set->cls_done); + else + break; + } + } + RETURN(result); } /** @@ -503,23 +508,23 @@ void cl_io_unlock(const struct lu_env *env, struct cl_io *io) ENTRY; set = &io->ci_lockset; - cfs_list_for_each_entry_safe(link, temp, &set->cls_todo, cill_linkage) - cl_lock_link_fini(env, io, link); + list_for_each_entry_safe(link, temp, &set->cls_todo, cill_linkage) + cl_lock_link_fini(env, io, link); - cfs_list_for_each_entry_safe(link, temp, &set->cls_curr, cill_linkage) - cl_lock_link_fini(env, io, link); + list_for_each_entry_safe(link, temp, &set->cls_curr, cill_linkage) + cl_lock_link_fini(env, io, link); - cfs_list_for_each_entry_safe(link, temp, &set->cls_done, cill_linkage) { - cl_unuse(env, link->cill_lock); - cl_lock_link_fini(env, io, link); - } - cl_io_for_each_reverse(scan, io) { - if (scan->cis_iop->op[io->ci_type].cio_unlock != NULL) - scan->cis_iop->op[io->ci_type].cio_unlock(env, scan); - } - io->ci_state = CIS_UNLOCKED; - LASSERT(!cl_env_info(env)->clt_counters[CNL_TOP].ctc_nr_locks_acquired); - EXIT; + list_for_each_entry_safe(link, temp, &set->cls_done, cill_linkage) { + cl_unuse(env, link->cill_lock); + cl_lock_link_fini(env, io, link); + } + cl_io_for_each_reverse(scan, io) { + if (scan->cis_iop->op[io->ci_type].cio_unlock != NULL) + scan->cis_iop->op[io->ci_type].cio_unlock(env, scan); + } + io->ci_state = CIS_UNLOCKED; + LASSERT(!cl_env_info(env)->clt_counters[CNL_TOP].ctc_nr_locks_acquired); + EXIT; } EXPORT_SYMBOL(cl_io_unlock); @@ -617,7 +622,7 @@ int cl_io_lock_add(const struct lu_env *env, struct cl_io *io, if (cl_lockset_merge(&io->ci_lockset, &link->cill_descr)) result = +1; else { - cfs_list_add(&link->cill_linkage, &io->ci_lockset.cls_todo); + list_add(&link->cill_linkage, &io->ci_lockset.cls_todo); result = 0; } RETURN(result); @@ -715,41 +720,6 @@ cl_io_slice_page(const struct cl_io_slice *ios, struct cl_page *page) } /** - * True iff \a page is within \a io range. - */ -static int cl_page_in_io(const struct cl_page *page, const struct cl_io *io) -{ - int result = 1; - loff_t start; - loff_t end; - pgoff_t idx; - - idx = page->cp_index; - switch (io->ci_type) { - case CIT_READ: - case CIT_WRITE: - /* - * check that [start, end) and [pos, pos + count) extents - * overlap. - */ - if (!cl_io_is_append(io)) { - const struct cl_io_rw_common *crw = &(io->u.ci_rw); - start = cl_offset(page->cp_obj, idx); - end = cl_offset(page->cp_obj, idx + 1); - result = crw->crw_pos < end && - start < crw->crw_pos + crw->crw_count; - } - break; - case CIT_FAULT: - result = io->u.ci_fault.ft_index == idx; - break; - default: - LBUG(); - } - return result; -} - -/** * Called by read io, when page has to be read from the server. * * \see cl_io_operations::cio_read_page() @@ -764,7 +734,6 @@ int cl_io_read_page(const struct lu_env *env, struct cl_io *io, LINVRNT(io->ci_type == CIT_READ || io->ci_type == CIT_FAULT); LINVRNT(cl_page_is_owned(page, io)); LINVRNT(io->ci_state == CIS_IO_GOING || io->ci_state == CIS_LOCKED); - LINVRNT(cl_page_in_io(page, io)); LINVRNT(cl_io_invariant(io)); ENTRY; @@ -792,8 +761,8 @@ int cl_io_read_page(const struct lu_env *env, struct cl_io *io, break; } } - if (result == 0) - result = cl_io_submit_rw(env, io, CRT_READ, queue, CRP_NORMAL); + if (result == 0 && queue->c2_qin.pl_nr > 0) + result = cl_io_submit_rw(env, io, CRT_READ, queue); /* * Unlock unsent pages in case of error. */ @@ -804,79 +773,30 @@ int cl_io_read_page(const struct lu_env *env, struct cl_io *io, EXPORT_SYMBOL(cl_io_read_page); /** - * Called by write io to prepare page to receive data from user buffer. - * - * \see cl_io_operations::cio_prepare_write() - */ -int cl_io_prepare_write(const struct lu_env *env, struct cl_io *io, - struct cl_page *page, unsigned from, unsigned to) -{ - const struct cl_io_slice *scan; - int result = 0; - - LINVRNT(io->ci_type == CIT_WRITE); - LINVRNT(cl_page_is_owned(page, io)); - LINVRNT(io->ci_state == CIS_IO_GOING || io->ci_state == CIS_LOCKED); - LINVRNT(cl_io_invariant(io)); - LASSERT(cl_page_in_io(page, io)); - ENTRY; - - cl_io_for_each_reverse(scan, io) { - if (scan->cis_iop->cio_prepare_write != NULL) { - const struct cl_page_slice *slice; - - slice = cl_io_slice_page(scan, page); - result = scan->cis_iop->cio_prepare_write(env, scan, - slice, - from, to); - if (result != 0) - break; - } - } - RETURN(result); -} -EXPORT_SYMBOL(cl_io_prepare_write); - -/** - * Called by write io after user data were copied into a page. + * Commit a list of contiguous pages into writeback cache. * - * \see cl_io_operations::cio_commit_write() + * \returns 0 if all pages committed, or errcode if error occurred. + * \see cl_io_operations::cio_commit_async() */ -int cl_io_commit_write(const struct lu_env *env, struct cl_io *io, - struct cl_page *page, unsigned from, unsigned to) -{ - const struct cl_io_slice *scan; - int result = 0; - - LINVRNT(io->ci_type == CIT_WRITE); - LINVRNT(io->ci_state == CIS_IO_GOING || io->ci_state == CIS_LOCKED); - LINVRNT(cl_io_invariant(io)); - /* - * XXX Uh... not nice. Top level cl_io_commit_write() call (vvp->lov) - * already called cl_page_cache_add(), moving page into CPS_CACHED - * state. Better (and more general) way of dealing with such situation - * is needed. - */ - LASSERT(cl_page_is_owned(page, io) || page->cp_parent != NULL); - LASSERT(cl_page_in_io(page, io)); - ENTRY; - - cl_io_for_each(scan, io) { - if (scan->cis_iop->cio_commit_write != NULL) { - const struct cl_page_slice *slice; - - slice = cl_io_slice_page(scan, page); - result = scan->cis_iop->cio_commit_write(env, scan, - slice, - from, to); - if (result != 0) - break; - } - } - LINVRNT(result <= 0); - RETURN(result); -} -EXPORT_SYMBOL(cl_io_commit_write); +int cl_io_commit_async(const struct lu_env *env, struct cl_io *io, + struct cl_page_list *queue, int from, int to, + cl_commit_cbt cb) +{ + const struct cl_io_slice *scan; + int result = 0; + ENTRY; + + cl_io_for_each(scan, io) { + if (scan->cis_iop->cio_commit_async == NULL) + continue; + result = scan->cis_iop->cio_commit_async(env, scan, queue, + from, to, cb); + if (result != 0) + break; + } + RETURN(result); +} +EXPORT_SYMBOL(cl_io_commit_async); /** * Submits a list of pages for immediate io. @@ -889,28 +809,24 @@ EXPORT_SYMBOL(cl_io_commit_write); * \see cl_io_operations::cio_submit() */ int cl_io_submit_rw(const struct lu_env *env, struct cl_io *io, - enum cl_req_type crt, struct cl_2queue *queue, - enum cl_req_priority priority) -{ - const struct cl_io_slice *scan; - int result = 0; - - LINVRNT(crt < ARRAY_SIZE(scan->cis_iop->req_op)); - ENTRY; - - cl_io_for_each(scan, io) { - if (scan->cis_iop->req_op[crt].cio_submit == NULL) - continue; - result = scan->cis_iop->req_op[crt].cio_submit(env, scan, crt, - queue, priority); - if (result != 0) - break; - } - /* - * If ->cio_submit() failed, no pages were sent. - */ - LASSERT(ergo(result != 0, cfs_list_empty(&queue->c2_qout.pl_pages))); - RETURN(result); + enum cl_req_type crt, struct cl_2queue *queue) +{ + const struct cl_io_slice *scan; + int result = 0; + ENTRY; + + cl_io_for_each(scan, io) { + if (scan->cis_iop->cio_submit == NULL) + continue; + result = scan->cis_iop->cio_submit(env, scan, crt, queue); + if (result != 0) + break; + } + /* + * If ->cio_submit() failed, no pages were sent. + */ + LASSERT(ergo(result != 0, list_empty(&queue->c2_qout.pl_pages))); + RETURN(result); } EXPORT_SYMBOL(cl_io_submit_rw); @@ -920,21 +836,19 @@ EXPORT_SYMBOL(cl_io_submit_rw); */ int cl_io_submit_sync(const struct lu_env *env, struct cl_io *io, enum cl_req_type iot, struct cl_2queue *queue, - enum cl_req_priority prio, long timeout) + long timeout) { struct cl_sync_io *anchor = &cl_env_info(env)->clt_anchor; struct cl_page *pg; int rc; - LASSERT(prio == CRP_NORMAL || prio == CRP_CANCEL); - cl_page_list_for_each(pg, &queue->c2_qin) { LASSERT(pg->cp_sync_io == NULL); pg->cp_sync_io = anchor; } cl_sync_io_init(anchor, queue->c2_qin.pl_nr); - rc = cl_io_submit_rw(env, io, iot, queue, prio); + rc = cl_io_submit_rw(env, io, iot, queue); if (rc == 0) { /* * If some pages weren't sent for any reason (e.g., @@ -951,7 +865,7 @@ int cl_io_submit_sync(const struct lu_env *env, struct cl_io *io, rc = cl_sync_io_wait(env, io, &queue->c2_qout, anchor, timeout); } else { - LASSERT(cfs_list_empty(&queue->c2_qout.pl_pages)); + LASSERT(list_empty(&queue->c2_qout.pl_pages)); cl_page_list_for_each(pg, &queue->c2_qin) pg->cp_sync_io = NULL; } @@ -972,7 +886,6 @@ int cl_io_cancel(const struct lu_env *env, struct cl_io *io, cl_page_list_for_each(page, queue) { int rc; - LINVRNT(cl_page_in_io(page, io)); rc = cl_page_cancel(env, page); result = result ?: rc; } @@ -1036,7 +949,9 @@ int cl_io_loop(const struct lu_env *env, struct cl_io *io) } cl_io_iter_fini(env, io); } while (result == 0 && io->ci_continue); - RETURN(result < 0 ? result : 0); + if (result == 0) + result = io->ci_result; + RETURN(result < 0 ? result : 0); } EXPORT_SYMBOL(cl_io_loop); @@ -1053,13 +968,13 @@ void cl_io_slice_add(struct cl_io *io, struct cl_io_slice *slice, struct cl_object *obj, const struct cl_io_operations *ops) { - cfs_list_t *linkage = &slice->cis_linkage; + struct list_head *linkage = &slice->cis_linkage; LASSERT((linkage->prev == NULL && linkage->next == NULL) || - cfs_list_empty(linkage)); + list_empty(linkage)); ENTRY; - cfs_list_add_tail(linkage, &io->ci_layers); + list_add_tail(linkage, &io->ci_layers); slice->cis_io = io; slice->cis_obj = obj; slice->cis_iop = ops; @@ -1073,11 +988,11 @@ EXPORT_SYMBOL(cl_io_slice_add); */ void cl_page_list_init(struct cl_page_list *plist) { - ENTRY; - plist->pl_nr = 0; - CFS_INIT_LIST_HEAD(&plist->pl_pages); - plist->pl_owner = cfs_current(); - EXIT; + ENTRY; + plist->pl_nr = 0; + INIT_LIST_HEAD(&plist->pl_pages); + plist->pl_owner = current; + EXIT; } EXPORT_SYMBOL(cl_page_list_init); @@ -1086,21 +1001,18 @@ EXPORT_SYMBOL(cl_page_list_init); */ void cl_page_list_add(struct cl_page_list *plist, struct cl_page *page) { - ENTRY; - /* it would be better to check that page is owned by "current" io, but - * it is not passed here. */ - LASSERT(page->cp_owner != NULL); - LINVRNT(plist->pl_owner == cfs_current()); - - cfs_lockdep_off(); - cfs_mutex_lock(&page->cp_mutex); - cfs_lockdep_on(); - LASSERT(cfs_list_empty(&page->cp_batch)); - cfs_list_add_tail(&page->cp_batch, &plist->pl_pages); - ++plist->pl_nr; - page->cp_queue_ref = lu_ref_add(&page->cp_reference, "queue", plist); - cl_page_get(page); - EXIT; + ENTRY; + /* it would be better to check that page is owned by "current" io, but + * it is not passed here. */ + LASSERT(page->cp_owner != NULL); + LINVRNT(plist->pl_owner == current); + + LASSERT(list_empty(&page->cp_batch)); + list_add_tail(&page->cp_batch, &plist->pl_pages); + ++plist->pl_nr; + lu_ref_add_at(&page->cp_reference, &page->cp_queue_ref, "queue", plist); + cl_page_get(page); + EXIT; } EXPORT_SYMBOL(cl_page_list_add); @@ -1108,20 +1020,18 @@ EXPORT_SYMBOL(cl_page_list_add); * Removes a page from a page list. */ void cl_page_list_del(const struct lu_env *env, - struct cl_page_list *plist, struct cl_page *page) + struct cl_page_list *plist, struct cl_page *page) { - LASSERT(plist->pl_nr > 0); - LINVRNT(plist->pl_owner == cfs_current()); + LASSERT(plist->pl_nr > 0); + LASSERT(cl_page_is_vmlocked(env, page)); + LINVRNT(plist->pl_owner == current); - ENTRY; - cfs_list_del_init(&page->cp_batch); - cfs_lockdep_off(); - cfs_mutex_unlock(&page->cp_mutex); - cfs_lockdep_on(); - --plist->pl_nr; - lu_ref_del_at(&page->cp_reference, page->cp_queue_ref, "queue", plist); - cl_page_put(env, page); - EXIT; + ENTRY; + list_del_init(&page->cp_batch); + --plist->pl_nr; + lu_ref_del_at(&page->cp_reference, &page->cp_queue_ref, "queue", plist); + cl_page_put(env, page); + EXIT; } EXPORT_SYMBOL(cl_page_list_del); @@ -1129,37 +1039,57 @@ EXPORT_SYMBOL(cl_page_list_del); * Moves a page from one page list to another. */ void cl_page_list_move(struct cl_page_list *dst, struct cl_page_list *src, - struct cl_page *page) + struct cl_page *page) { - LASSERT(src->pl_nr > 0); - LINVRNT(dst->pl_owner == cfs_current()); - LINVRNT(src->pl_owner == cfs_current()); + LASSERT(src->pl_nr > 0); + LINVRNT(dst->pl_owner == current); + LINVRNT(src->pl_owner == current); - ENTRY; - cfs_list_move_tail(&page->cp_batch, &dst->pl_pages); - --src->pl_nr; - ++dst->pl_nr; - lu_ref_set_at(&page->cp_reference, - page->cp_queue_ref, "queue", src, dst); - EXIT; + ENTRY; + list_move_tail(&page->cp_batch, &dst->pl_pages); + --src->pl_nr; + ++dst->pl_nr; + lu_ref_set_at(&page->cp_reference, &page->cp_queue_ref, "queue", + src, dst); + EXIT; } EXPORT_SYMBOL(cl_page_list_move); /** + * Moves a page from one page list to the head of another list. + */ +void cl_page_list_move_head(struct cl_page_list *dst, struct cl_page_list *src, + struct cl_page *page) +{ + LASSERT(src->pl_nr > 0); + LINVRNT(dst->pl_owner == current); + LINVRNT(src->pl_owner == current); + + ENTRY; + list_move(&page->cp_batch, &dst->pl_pages); + --src->pl_nr; + ++dst->pl_nr; + lu_ref_set_at(&page->cp_reference, &page->cp_queue_ref, "queue", + src, dst); + EXIT; +} +EXPORT_SYMBOL(cl_page_list_move_head); + +/** * splice the cl_page_list, just as list head does */ void cl_page_list_splice(struct cl_page_list *list, struct cl_page_list *head) { - struct cl_page *page; - struct cl_page *tmp; + struct cl_page *page; + struct cl_page *tmp; - LINVRNT(list->pl_owner == cfs_current()); - LINVRNT(head->pl_owner == cfs_current()); + LINVRNT(list->pl_owner == current); + LINVRNT(head->pl_owner == current); - ENTRY; - cl_page_list_for_each_safe(page, tmp, list) - cl_page_list_move(head, list, page); - EXIT; + ENTRY; + cl_page_list_for_each_safe(page, tmp, list) + cl_page_list_move(head, list, page); + EXIT; } EXPORT_SYMBOL(cl_page_list_splice); @@ -1170,35 +1100,33 @@ void cl_page_disown0(const struct lu_env *env, * Disowns pages in a queue. */ void cl_page_list_disown(const struct lu_env *env, - struct cl_io *io, struct cl_page_list *plist) -{ - struct cl_page *page; - struct cl_page *temp; - - LINVRNT(plist->pl_owner == cfs_current()); - - ENTRY; - cl_page_list_for_each_safe(page, temp, plist) { - LASSERT(plist->pl_nr > 0); - - cfs_list_del_init(&page->cp_batch); - cfs_lockdep_off(); - cfs_mutex_unlock(&page->cp_mutex); - cfs_lockdep_on(); - --plist->pl_nr; - /* - * cl_page_disown0 rather than usual cl_page_disown() is used, - * because pages are possibly in CPS_FREEING state already due - * to the call to cl_page_list_discard(). - */ - /* - * XXX cl_page_disown0() will fail if page is not locked. - */ - cl_page_disown0(env, io, page); - lu_ref_del(&page->cp_reference, "queue", plist); - cl_page_put(env, page); - } - EXIT; + struct cl_io *io, struct cl_page_list *plist) +{ + struct cl_page *page; + struct cl_page *temp; + + LINVRNT(plist->pl_owner == current); + + ENTRY; + cl_page_list_for_each_safe(page, temp, plist) { + LASSERT(plist->pl_nr > 0); + + list_del_init(&page->cp_batch); + --plist->pl_nr; + /* + * cl_page_disown0 rather than usual cl_page_disown() is used, + * because pages are possibly in CPS_FREEING state already due + * to the call to cl_page_list_discard(). + */ + /* + * XXX cl_page_disown0() will fail if page is not locked. + */ + cl_page_disown0(env, io, page); + lu_ref_del_at(&page->cp_reference, &page->cp_queue_ref, "queue", + plist); + cl_page_put(env, page); + } + EXIT; } EXPORT_SYMBOL(cl_page_list_disown); @@ -1207,16 +1135,16 @@ EXPORT_SYMBOL(cl_page_list_disown); */ void cl_page_list_fini(const struct lu_env *env, struct cl_page_list *plist) { - struct cl_page *page; - struct cl_page *temp; + struct cl_page *page; + struct cl_page *temp; - LINVRNT(plist->pl_owner == cfs_current()); + LINVRNT(plist->pl_owner == current); - ENTRY; - cl_page_list_for_each_safe(page, temp, plist) - cl_page_list_del(env, plist, page); - LASSERT(plist->pl_nr == 0); - EXIT; + ENTRY; + cl_page_list_for_each_safe(page, temp, plist) + cl_page_list_del(env, plist, page); + LASSERT(plist->pl_nr == 0); + EXIT; } EXPORT_SYMBOL(cl_page_list_fini); @@ -1224,26 +1152,23 @@ EXPORT_SYMBOL(cl_page_list_fini); * Owns all pages in a queue. */ int cl_page_list_own(const struct lu_env *env, - struct cl_io *io, struct cl_page_list *plist) + struct cl_io *io, struct cl_page_list *plist) { - struct cl_page *page; - struct cl_page *temp; - pgoff_t index = 0; - int result; + struct cl_page *page; + struct cl_page *temp; + int result; - LINVRNT(plist->pl_owner == cfs_current()); + LINVRNT(plist->pl_owner == current); - ENTRY; - result = 0; - cl_page_list_for_each_safe(page, temp, plist) { - LASSERT(index <= page->cp_index); - index = page->cp_index; - if (cl_page_own(env, io, page) == 0) - result = result ?: page->cp_error; - else - cl_page_list_del(env, plist, page); - } - RETURN(result); + ENTRY; + result = 0; + cl_page_list_for_each_safe(page, temp, plist) { + if (cl_page_own(env, io, page) == 0) + result = result ?: page->cp_error; + else + cl_page_list_del(env, plist, page); + } + RETURN(result); } EXPORT_SYMBOL(cl_page_list_own); @@ -1251,14 +1176,14 @@ EXPORT_SYMBOL(cl_page_list_own); * Assumes all pages in a queue. */ void cl_page_list_assume(const struct lu_env *env, - struct cl_io *io, struct cl_page_list *plist) + struct cl_io *io, struct cl_page_list *plist) { - struct cl_page *page; + struct cl_page *page; - LINVRNT(plist->pl_owner == cfs_current()); + LINVRNT(plist->pl_owner == current); - cl_page_list_for_each(page, plist) - cl_page_assume(env, io, page); + cl_page_list_for_each(page, plist) + cl_page_assume(env, io, page); } EXPORT_SYMBOL(cl_page_list_assume); @@ -1266,40 +1191,19 @@ EXPORT_SYMBOL(cl_page_list_assume); * Discards all pages in a queue. */ void cl_page_list_discard(const struct lu_env *env, struct cl_io *io, - struct cl_page_list *plist) + struct cl_page_list *plist) { - struct cl_page *page; + struct cl_page *page; - LINVRNT(plist->pl_owner == cfs_current()); - ENTRY; - cl_page_list_for_each(page, plist) - cl_page_discard(env, io, page); - EXIT; + LINVRNT(plist->pl_owner == current); + ENTRY; + cl_page_list_for_each(page, plist) + cl_page_discard(env, io, page); + EXIT; } EXPORT_SYMBOL(cl_page_list_discard); /** - * Unmaps all pages in a queue from user virtual memory. - */ -int cl_page_list_unmap(const struct lu_env *env, struct cl_io *io, - struct cl_page_list *plist) -{ - struct cl_page *page; - int result; - - LINVRNT(plist->pl_owner == cfs_current()); - ENTRY; - result = 0; - cl_page_list_for_each(page, plist) { - result = cl_page_unmap(env, io, page); - if (result != 0) - break; - } - RETURN(result); -} -EXPORT_SYMBOL(cl_page_list_unmap); - -/** * Initialize dual page queue. */ void cl_2queue_init(struct cl_2queue *queue) @@ -1386,7 +1290,7 @@ EXPORT_SYMBOL(cl_2queue_init_page); /** * Returns top-level io. * - * \see cl_object_top(), cl_page_top(). + * \see cl_object_top() */ struct cl_io *cl_io_top(struct cl_io *io) { @@ -1419,7 +1323,7 @@ void cl_req_slice_add(struct cl_req *req, struct cl_req_slice *slice, const struct cl_req_operations *ops) { ENTRY; - cfs_list_add_tail(&slice->crs_linkage, &req->crq_layers); + list_add_tail(&slice->crs_linkage, &req->crq_layers); slice->crs_dev = dev; slice->crs_ops = ops; slice->crs_req = req; @@ -1431,21 +1335,21 @@ static void cl_req_free(const struct lu_env *env, struct cl_req *req) { unsigned i; - LASSERT(cfs_list_empty(&req->crq_pages)); + LASSERT(list_empty(&req->crq_pages)); LASSERT(req->crq_nrpages == 0); - LINVRNT(cfs_list_empty(&req->crq_layers)); + LINVRNT(list_empty(&req->crq_layers)); LINVRNT(equi(req->crq_nrobjs > 0, req->crq_o != NULL)); ENTRY; if (req->crq_o != NULL) { for (i = 0; i < req->crq_nrobjs; ++i) { struct cl_object *obj = req->crq_o[i].ro_obj; - if (obj != NULL) { - lu_object_ref_del_at(&obj->co_lu, - req->crq_o[i].ro_obj_ref, - "cl_req", req); - cl_object_put(env, obj); - } + if (obj != NULL) { + lu_object_ref_del_at(&obj->co_lu, + &req->crq_o[i].ro_obj_ref, + "cl_req", req); + cl_object_put(env, obj); + } } OBD_FREE(req->crq_o, req->crq_nrobjs * sizeof req->crq_o[0]); } @@ -1456,26 +1360,22 @@ static void cl_req_free(const struct lu_env *env, struct cl_req *req) static int cl_req_init(const struct lu_env *env, struct cl_req *req, struct cl_page *page) { - struct cl_device *dev; - struct cl_page_slice *slice; - int result; + struct cl_device *dev; + struct cl_page_slice *slice; + int result; - ENTRY; - result = 0; - page = cl_page_top(page); - do { - cfs_list_for_each_entry(slice, &page->cp_layers, cpl_linkage) { - dev = lu2cl_dev(slice->cpl_obj->co_lu.lo_dev); - if (dev->cd_ops->cdo_req_init != NULL) { - result = dev->cd_ops->cdo_req_init(env, - dev, req); - if (result != 0) - break; - } - } - page = page->cp_child; - } while (page != NULL && result == 0); - RETURN(result); + ENTRY; + result = 0; + list_for_each_entry(slice, &page->cp_layers, cpl_linkage) { + dev = lu2cl_dev(slice->cpl_obj->co_lu.lo_dev); + if (dev->cd_ops->cdo_req_init != NULL) { + result = dev->cd_ops->cdo_req_init(env, + dev, req); + if (result != 0) + break; + } + } + RETURN(result); } /** @@ -1490,10 +1390,10 @@ void cl_req_completion(const struct lu_env *env, struct cl_req *req, int rc) /* * for the lack of list_for_each_entry_reverse_safe()... */ - while (!cfs_list_empty(&req->crq_layers)) { - slice = cfs_list_entry(req->crq_layers.prev, - struct cl_req_slice, crs_linkage); - cfs_list_del_init(&slice->crs_linkage); + while (!list_empty(&req->crq_layers)) { + slice = list_entry(req->crq_layers.prev, + struct cl_req_slice, crs_linkage); + list_del_init(&slice->crs_linkage); if (slice->crs_ops->cro_completion != NULL) slice->crs_ops->cro_completion(env, slice, rc); } @@ -1517,12 +1417,13 @@ struct cl_req *cl_req_alloc(const struct lu_env *env, struct cl_page *page, if (req != NULL) { int result; + req->crq_type = crt; + INIT_LIST_HEAD(&req->crq_pages); + INIT_LIST_HEAD(&req->crq_layers); + OBD_ALLOC(req->crq_o, nr_objects * sizeof req->crq_o[0]); if (req->crq_o != NULL) { req->crq_nrobjs = nr_objects; - req->crq_type = crt; - CFS_INIT_LIST_HEAD(&req->crq_pages); - CFS_INIT_LIST_HEAD(&req->crq_layers); result = cl_req_init(env, req, page); } else result = -ENOMEM; @@ -1542,20 +1443,19 @@ EXPORT_SYMBOL(cl_req_alloc); void cl_req_page_add(const struct lu_env *env, struct cl_req *req, struct cl_page *page) { - struct cl_object *obj; - struct cl_req_obj *rqo; - int i; + struct cl_object *obj; + struct cl_req_obj *rqo; + int i; - ENTRY; - page = cl_page_top(page); + ENTRY; - LASSERT(cfs_list_empty(&page->cp_flight)); + LASSERT(list_empty(&page->cp_flight)); LASSERT(page->cp_req == NULL); CL_PAGE_DEBUG(D_PAGE, env, page, "req %p, %d, %u\n", req, req->crq_type, req->crq_nrpages); - cfs_list_add_tail(&page->cp_flight, &req->crq_pages); + list_add_tail(&page->cp_flight, &req->crq_pages); ++req->crq_nrpages; page->cp_req = req; obj = cl_object_top(page->cp_obj); @@ -1563,13 +1463,13 @@ void cl_req_page_add(const struct lu_env *env, if (rqo->ro_obj == NULL) { rqo->ro_obj = obj; cl_object_get(obj); - rqo->ro_obj_ref = lu_object_ref_add(&obj->co_lu, - "cl_req", req); - break; - } - } - LASSERT(i < req->crq_nrobjs); - EXIT; + lu_object_ref_add_at(&obj->co_lu, &rqo->ro_obj_ref, + "cl_req", req); + break; + } + } + LASSERT(i < req->crq_nrobjs); + EXIT; } EXPORT_SYMBOL(cl_req_page_add); @@ -1578,18 +1478,17 @@ EXPORT_SYMBOL(cl_req_page_add); */ void cl_req_page_done(const struct lu_env *env, struct cl_page *page) { - struct cl_req *req = page->cp_req; + struct cl_req *req = page->cp_req; - ENTRY; - page = cl_page_top(page); + ENTRY; - LASSERT(!cfs_list_empty(&page->cp_flight)); - LASSERT(req->crq_nrpages > 0); + LASSERT(!list_empty(&page->cp_flight)); + LASSERT(req->crq_nrpages > 0); - cfs_list_del_init(&page->cp_flight); - --req->crq_nrpages; - page->cp_req = NULL; - EXIT; + list_del_init(&page->cp_flight); + --req->crq_nrpages; + page->cp_req = NULL; + EXIT; } EXPORT_SYMBOL(cl_req_page_done); @@ -1612,7 +1511,7 @@ int cl_req_prep(const struct lu_env *env, struct cl_req *req) LASSERT(req->crq_o[i].ro_obj != NULL); result = 0; - cfs_list_for_each_entry(slice, &req->crq_layers, crs_linkage) { + list_for_each_entry(slice, &req->crq_layers, crs_linkage) { if (slice->crs_ops->cro_prep != NULL) { result = slice->crs_ops->cro_prep(env, slice); if (result != 0) @@ -1635,14 +1534,14 @@ void cl_req_attr_set(const struct lu_env *env, struct cl_req *req, struct cl_page *page; int i; - LASSERT(!cfs_list_empty(&req->crq_pages)); + LASSERT(!list_empty(&req->crq_pages)); ENTRY; /* Take any page to use as a model. */ - page = cfs_list_entry(req->crq_pages.next, struct cl_page, cp_flight); + page = list_entry(req->crq_pages.next, struct cl_page, cp_flight); for (i = 0; i < req->crq_nrobjs; ++i) { - cfs_list_for_each_entry(slice, &req->crq_layers, crs_linkage) { + list_for_each_entry(slice, &req->crq_layers, crs_linkage) { const struct cl_page_slice *scan; const struct cl_object *obj; @@ -1672,11 +1571,12 @@ EXPORT_SYMBOL(cl_req_attr_set); */ void cl_sync_io_init(struct cl_sync_io *anchor, int nrpages) { - ENTRY; - cfs_waitq_init(&anchor->csi_waitq); - cfs_atomic_set(&anchor->csi_sync_nr, nrpages); - anchor->csi_sync_rc = 0; - EXIT; + ENTRY; + init_waitqueue_head(&anchor->csi_waitq); + atomic_set(&anchor->csi_sync_nr, nrpages); + atomic_set(&anchor->csi_barrier, nrpages > 0); + anchor->csi_sync_rc = 0; + EXIT; } EXPORT_SYMBOL(cl_sync_io_init); @@ -1696,26 +1596,34 @@ int cl_sync_io_wait(const struct lu_env *env, struct cl_io *io, LASSERT(timeout >= 0); rc = l_wait_event(anchor->csi_waitq, - cfs_atomic_read(&anchor->csi_sync_nr) == 0, + atomic_read(&anchor->csi_sync_nr) == 0, &lwi); if (rc < 0) { CERROR("SYNC IO failed with error: %d, try to cancel " "%d remaining pages\n", - rc, cfs_atomic_read(&anchor->csi_sync_nr)); + rc, atomic_read(&anchor->csi_sync_nr)); (void)cl_io_cancel(env, io, queue); lwi = (struct l_wait_info) { 0 }; (void)l_wait_event(anchor->csi_waitq, - cfs_atomic_read(&anchor->csi_sync_nr) == 0, + atomic_read(&anchor->csi_sync_nr) == 0, &lwi); } else { rc = anchor->csi_sync_rc; } - LASSERT(cfs_atomic_read(&anchor->csi_sync_nr) == 0); + LASSERT(atomic_read(&anchor->csi_sync_nr) == 0); cl_page_list_assume(env, io, queue); - POISON(anchor, 0x5a, sizeof *anchor); - RETURN(rc); + + /* wait until cl_sync_io_note() has done wakeup */ + while (unlikely(atomic_read(&anchor->csi_barrier) != 0)) { +#ifdef __KERNEL__ + cpu_relax(); +#endif + } + + POISON(anchor, 0x5a, sizeof *anchor); + RETURN(rc); } EXPORT_SYMBOL(cl_sync_io_wait); @@ -1732,9 +1640,12 @@ void cl_sync_io_note(struct cl_sync_io *anchor, int ioret) * ->{prepare,commit}_write(). Completion is used to signal the end of * IO. */ - LASSERT(cfs_atomic_read(&anchor->csi_sync_nr) > 0); - if (cfs_atomic_dec_and_test(&anchor->csi_sync_nr)) - cfs_waitq_broadcast(&anchor->csi_waitq); - EXIT; + LASSERT(atomic_read(&anchor->csi_sync_nr) > 0); + if (atomic_dec_and_test(&anchor->csi_sync_nr)) { + wake_up_all(&anchor->csi_waitq); + /* it's safe to nuke or reuse anchor now */ + atomic_set(&anchor->csi_barrier, 0); + } + EXIT; } EXPORT_SYMBOL(cl_sync_io_note);