X-Git-Url: https://git.whamcloud.com/?a=blobdiff_plain;f=lustre%2Fobdclass%2Fcl_io.c;h=aed5f5ed2db04cdb610147de1f29596dd2206d6e;hb=40137f1d07295ee4a0bb75d2ae553c42e66295f4;hp=0e2f342d423fa75df30e624d980d1bca0255e39d;hpb=89f659febd4220cc517a181f8ac6cc52235b76ca;p=fs%2Flustre-release.git diff --git a/lustre/obdclass/cl_io.c b/lustre/obdclass/cl_io.c index 0e2f342..aed5f5e 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. @@ -26,8 +24,10 @@ * GPL HEADER END */ /* - * Copyright 2008 Sun Microsystems, Inc. All rights reserved. + * Copyright (c) 2008, 2010, Oracle and/or its affiliates. All rights reserved. * Use is subject to license terms. + * + * Copyright (c) 2011, 2012, Intel Corporation. */ /* * This file is part of Lustre, http://www.lustre.org/ @@ -39,17 +39,11 @@ */ #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 +54,9 @@ */ #define cl_io_for_each(slice, io) \ - list_for_each_entry((slice), &io->ci_layers, cis_linkage) + cfs_list_for_each_entry((slice), &io->ci_layers, cis_linkage) #define cl_io_for_each_reverse(slice, io) \ - list_for_each_entry_reverse((slice), &io->ci_layers, cis_linkage) + cfs_list_for_each_entry_reverse((slice), &io->ci_layers, cis_linkage) static inline int cl_io_type_is_valid(enum cl_io_type type) { @@ -75,15 +69,6 @@ static inline int cl_io_is_loopable(const struct cl_io *io) } /** - * True, iff \a io is a sendfile(). - */ -int cl_io_is_sendfile(const struct cl_io *io) -{ - return io->ci_type == CIT_READ && io->u.ci_rd.rd_is_sendfile; -} -EXPORT_SYMBOL(cl_io_is_sendfile); - -/** * Returns true iff there is an IO ongoing in the given environment. */ int cl_io_is_going(const struct lu_env *env) @@ -116,17 +101,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 (!list_empty(&io->ci_layers)) { - slice = container_of(io->ci_layers.next, struct cl_io_slice, + while (!cfs_list_empty(&io->ci_layers)) { + slice = container_of(io->ci_layers.prev, struct cl_io_slice, cis_linkage); - list_del_init(&slice->cis_linkage); + cfs_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); /* @@ -140,7 +125,25 @@ 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: + 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); @@ -227,7 +230,7 @@ int cl_io_rw_init(const struct lu_env *env, struct cl_io *io, ENTRY; LU_OBJECT_HEADER(D_VFSTRACE, env, &io->ci_obj->co_lu, - "io range: %i [%llu, %llu) %i %i\n", + "io range: %u ["LPU64", "LPU64") %u %u\n", iot, (__u64)pos, (__u64)pos + count, io->u.ci_rw.crw_nonblock, io->u.ci_wr.wr_append); io->u.ci_rw.crw_pos = pos; @@ -242,13 +245,41 @@ cl_lock_descr_fid(const struct cl_lock_descr *descr) return lu_object_fid(&descr->cld_obj->co_lu); } -static int cl_lock_descr_cmp(const struct cl_lock_descr *d0, - const struct cl_lock_descr *d1) +static int cl_lock_descr_sort(const struct cl_lock_descr *d0, + const struct cl_lock_descr *d1) { return lu_fid_cmp(cl_lock_descr_fid(d0), cl_lock_descr_fid(d1)) ?: __diff_normalize(d0->cld_start, d1->cld_start); } +static int cl_lock_descr_cmp(const struct cl_lock_descr *d0, + const struct cl_lock_descr *d1) +{ + int ret; + + ret = lu_fid_cmp(cl_lock_descr_fid(d0), cl_lock_descr_fid(d1)); + if (ret) + return ret; + if (d0->cld_end < d1->cld_start) + return -1; + if (d0->cld_start > d0->cld_end) + return 1; + return 0; +} + +static void cl_lock_descr_merge(struct cl_lock_descr *d0, + const struct cl_lock_descr *d1) +{ + d0->cld_start = min(d0->cld_start, d1->cld_start); + d0->cld_end = max(d0->cld_end, d1->cld_end); + + if (d1->cld_mode == CLM_WRITE && d0->cld_mode != CLM_WRITE) + d0->cld_mode = CLM_WRITE; + + if (d1->cld_mode == CLM_GROUP && d0->cld_mode != CLM_GROUP) + d0->cld_mode = CLM_GROUP; +} + /* * Sort locks in lexicographical order of their (fid, start-offset) pairs. */ @@ -266,10 +297,11 @@ static void cl_io_locks_sort(struct cl_io *io) done = 1; prev = NULL; - list_for_each_entry_safe(curr, temp, &io->ci_lockset.cls_todo, - cill_linkage) { + cfs_list_for_each_entry_safe(curr, temp, + &io->ci_lockset.cls_todo, + cill_linkage) { if (prev != NULL) { - switch (cl_lock_descr_cmp(&prev->cill_descr, + switch (cl_lock_descr_sort(&prev->cill_descr, &curr->cill_descr)) { case 0: /* @@ -280,8 +312,8 @@ static void cl_io_locks_sort(struct cl_io *io) default: LBUG(); case +1: - list_move_tail(&curr->cill_linkage, - &prev->cill_linkage); + cfs_list_move_tail(&curr->cill_linkage, + &prev->cill_linkage); done = 0; continue; /* don't change prev: it's * still "previous" */ @@ -301,26 +333,51 @@ 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 struct list_head *queue, +int cl_queue_match(const cfs_list_t *queue, const struct cl_lock_descr *need) { struct cl_io_lock_link *scan; ENTRY; - list_for_each_entry(scan, queue, cill_linkage) { + cfs_list_for_each_entry(scan, queue, cill_linkage) { if (cl_lock_descr_match(&scan->cill_descr, need)) RETURN(+1); } - return 0; + RETURN(0); } EXPORT_SYMBOL(cl_queue_match); +static int cl_queue_merge(const cfs_list_t *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); + +} + static int cl_lockset_match(const struct cl_lockset *set, - const struct cl_lock_descr *need, int all_queues) + const struct cl_lock_descr *need) { - return (all_queues ? cl_queue_match(&set->cls_todo, need) : 0) || - cl_queue_match(&set->cls_curr, need) || - cl_queue_match(&set->cls_done, need); + return cl_queue_match(&set->cls_curr, need) || + cl_queue_match(&set->cls_done, need); +} + +static int cl_lockset_merge(const struct cl_lockset *set, + const struct cl_lock_descr *need) +{ + return cl_queue_merge(&set->cls_todo, need) || + cl_lockset_match(set, need); } static int cl_lockset_lock_one(const struct lu_env *env, @@ -332,15 +389,21 @@ static int cl_lockset_lock_one(const struct lu_env *env, ENTRY; - lock = cl_lock_request(env, io, &link->cill_descr, link->cill_enq_flags, - "io", io); + 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); + if (!IS_ERR(lock)) { link->cill_lock = lock; - list_move(&link->cill_linkage, &set->cls_curr); - if (!(link->cill_enq_flags & CEF_ASYNC)) { + cfs_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) - list_move(&link->cill_linkage, &set->cls_done); + cfs_list_move(&link->cill_linkage, + &set->cls_done); } else result = 0; } else @@ -354,7 +417,7 @@ static void cl_lock_link_fini(const struct lu_env *env, struct cl_io *io, struct cl_lock *lock = link->cill_lock; ENTRY; - list_del_init(&link->cill_linkage); + cfs_list_del_init(&link->cill_linkage); if (lock != NULL) { cl_lock_release(env, lock, "io", io); link->cill_lock = NULL; @@ -374,8 +437,8 @@ static int cl_lockset_lock(const struct lu_env *env, struct cl_io *io, ENTRY; result = 0; - list_for_each_entry_safe(link, temp, &set->cls_todo, cill_linkage) { - if (!cl_lockset_match(set, &link->cill_descr, 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); @@ -385,12 +448,13 @@ static int cl_lockset_lock(const struct lu_env *env, struct cl_io *io, cl_lock_link_fini(env, io, link); } if (result == 0) { - list_for_each_entry_safe(link, temp, - &set->cls_curr, cill_linkage) { + 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) - list_move(&link->cill_linkage, &set->cls_done); + cfs_list_move(&link->cill_linkage, + &set->cls_done); else break; } @@ -451,13 +515,13 @@ void cl_io_unlock(const struct lu_env *env, struct cl_io *io) ENTRY; set = &io->ci_lockset; - list_for_each_entry_safe(link, temp, &set->cls_todo, cill_linkage) + 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_curr, cill_linkage) + 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_done, cill_linkage) { + 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); } @@ -466,7 +530,7 @@ void cl_io_unlock(const struct lu_env *env, struct cl_io *io) scan->cis_iop->op[io->ci_type].cio_unlock(env, scan); } io->ci_state = CIS_UNLOCKED; - LASSERT(cl_env_info(env)->clt_nr_locks_acquired == 0); + LASSERT(!cl_env_info(env)->clt_counters[CNL_TOP].ctc_nr_locks_acquired); EXIT; } EXPORT_SYMBOL(cl_io_unlock); @@ -562,10 +626,10 @@ int cl_io_lock_add(const struct lu_env *env, struct cl_io *io, int result; ENTRY; - if (cl_lockset_match(&io->ci_lockset, &link->cill_descr, 1)) + if (cl_lockset_merge(&io->ci_lockset, &link->cill_descr)) result = +1; else { - list_add(&link->cill_linkage, &io->ci_lockset.cls_todo); + cfs_list_add(&link->cill_linkage, &io->ci_lockset.cls_todo); result = 0; } RETURN(result); @@ -582,7 +646,7 @@ static void cl_free_io_lock_link(const struct lu_env *env, * Allocates new lock link, and uses it to add a lock to a lockset. */ int cl_io_lock_alloc_add(const struct lu_env *env, struct cl_io *io, - struct cl_lock_descr *descr, int enqflags) + struct cl_lock_descr *descr) { struct cl_io_lock_link *link; int result; @@ -591,7 +655,6 @@ int cl_io_lock_alloc_add(const struct lu_env *env, struct cl_io *io, OBD_ALLOC_PTR(link); if (link != NULL) { link->cill_descr = *descr; - link->cill_enq_flags = enqflags; link->cill_fini = cl_free_io_lock_link; result = cl_io_lock_add(env, io, link); if (result) /* lock match */ @@ -668,7 +731,7 @@ cl_io_slice_page(const struct cl_io_slice *ios, struct cl_page *page) */ static int cl_page_in_io(const struct cl_page *page, const struct cl_io *io) { - int result; + int result = 1; loff_t start; loff_t end; pgoff_t idx; @@ -681,10 +744,13 @@ static int cl_page_in_io(const struct cl_page *page, const struct cl_io *io) * check that [start, end) and [pos, pos + count) extents * overlap. */ - start = cl_offset(page->cp_obj, idx); - end = cl_offset(page->cp_obj, idx + 1); - result = io->u.ci_rw.crw_pos < end && - start < io->u.ci_rw.crw_pos + io->u.ci_rw.crw_count; + 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; @@ -739,7 +805,7 @@ int cl_io_read_page(const struct lu_env *env, struct cl_io *io, } } if (result == 0) - result = cl_io_submit_rw(env, io, CRT_READ, queue); + result = cl_io_submit_rw(env, io, CRT_READ, queue); /* * Unlock unsent pages in case of error. */ @@ -835,7 +901,7 @@ 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_type crt, struct cl_2queue *queue) { const struct cl_io_slice *scan; int result = 0; @@ -847,19 +913,62 @@ int cl_io_submit_rw(const struct lu_env *env, struct cl_io *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); + queue); if (result != 0) break; } /* * If ->cio_submit() failed, no pages were sent. */ - LASSERT(ergo(result != 0, list_empty(&queue->c2_qout.pl_pages))); + LASSERT(ergo(result != 0, cfs_list_empty(&queue->c2_qout.pl_pages))); RETURN(result); } EXPORT_SYMBOL(cl_io_submit_rw); /** + * Submit a sync_io and wait for the IO to be finished, or error happens. + * If \a timeout is zero, it means to wait for the IO unconditionally. + */ +int cl_io_submit_sync(const struct lu_env *env, struct cl_io *io, + enum cl_req_type iot, struct cl_2queue *queue, + long timeout) +{ + struct cl_sync_io *anchor = &cl_env_info(env)->clt_anchor; + struct cl_page *pg; + int rc; + + 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); + if (rc == 0) { + /* + * If some pages weren't sent for any reason (e.g., + * read found up-to-date pages in the cache, or write found + * clean pages), count them as completed to avoid infinite + * wait. + */ + cl_page_list_for_each(pg, &queue->c2_qin) { + pg->cp_sync_io = NULL; + cl_sync_io_note(anchor, +1); + } + + /* wait for the IO to be finished. */ + rc = cl_sync_io_wait(env, io, &queue->c2_qout, + anchor, timeout); + } else { + LASSERT(cfs_list_empty(&queue->c2_qout.pl_pages)); + cl_page_list_for_each(pg, &queue->c2_qin) + pg->cp_sync_io = NULL; + } + return rc; +} +EXPORT_SYMBOL(cl_io_submit_sync); + +/** * Cancel an IO which has been submitted by cl_io_submit_rw. */ int cl_io_cancel(const struct lu_env *env, struct cl_io *io, @@ -936,7 +1045,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); @@ -953,13 +1064,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) { - struct list_head *linkage = &slice->cis_linkage; + cfs_list_t *linkage = &slice->cis_linkage; LASSERT((linkage->prev == NULL && linkage->next == NULL) || - list_empty(linkage)); + cfs_list_empty(linkage)); ENTRY; - list_add_tail(linkage, &io->ci_layers); + cfs_list_add_tail(linkage, &io->ci_layers); slice->cis_io = io; slice->cis_obj = obj; slice->cis_iop = ops; @@ -992,11 +1103,11 @@ void cl_page_list_add(struct cl_page_list *plist, struct cl_page *page) LASSERT(page->cp_owner != NULL); LINVRNT(plist->pl_owner == cfs_current()); - lockdep_off(); - mutex_lock(&page->cp_mutex); - lockdep_on(); - LASSERT(list_empty(&page->cp_batch)); - list_add_tail(&page->cp_batch, &plist->pl_pages); + lockdep_off(); + mutex_lock(&page->cp_mutex); + 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); @@ -1014,10 +1125,10 @@ void cl_page_list_del(const struct lu_env *env, LINVRNT(plist->pl_owner == cfs_current()); ENTRY; - list_del_init(&page->cp_batch); - lockdep_off(); - mutex_unlock(&page->cp_mutex); - lockdep_on(); + cfs_list_del_init(&page->cp_batch); + lockdep_off(); + mutex_unlock(&page->cp_mutex); + lockdep_on(); --plist->pl_nr; lu_ref_del_at(&page->cp_reference, page->cp_queue_ref, "queue", plist); cl_page_put(env, page); @@ -1036,7 +1147,7 @@ void cl_page_list_move(struct cl_page_list *dst, struct cl_page_list *src, LINVRNT(src->pl_owner == cfs_current()); ENTRY; - list_move_tail(&page->cp_batch, &dst->pl_pages); + cfs_list_move_tail(&page->cp_batch, &dst->pl_pages); --src->pl_nr; ++dst->pl_nr; lu_ref_set_at(&page->cp_reference, @@ -1081,10 +1192,10 @@ void cl_page_list_disown(const struct lu_env *env, cl_page_list_for_each_safe(page, temp, plist) { LASSERT(plist->pl_nr > 0); - list_del_init(&page->cp_batch); - lockdep_off(); - mutex_unlock(&page->cp_mutex); - lockdep_on(); + cfs_list_del_init(&page->cp_batch); + lockdep_off(); + mutex_unlock(&page->cp_mutex); + lockdep_on(); --plist->pl_nr; /* * cl_page_disown0 rather than usual cl_page_disown() is used, @@ -1128,6 +1239,7 @@ int cl_page_list_own(const struct lu_env *env, { struct cl_page *page; struct cl_page *temp; + pgoff_t index = 0; int result; LINVRNT(plist->pl_owner == cfs_current()); @@ -1135,8 +1247,10 @@ int cl_page_list_own(const struct lu_env *env, 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; + result = result ?: page->cp_error; else cl_page_list_del(env, plist, page); } @@ -1316,7 +1430,7 @@ void cl_req_slice_add(struct cl_req *req, struct cl_req_slice *slice, const struct cl_req_operations *ops) { ENTRY; - list_add_tail(&slice->crs_linkage, &req->crq_layers); + cfs_list_add_tail(&slice->crs_linkage, &req->crq_layers); slice->crs_dev = dev; slice->crs_ops = ops; slice->crs_req = req; @@ -1328,9 +1442,9 @@ static void cl_req_free(const struct lu_env *env, struct cl_req *req) { unsigned i; - LASSERT(list_empty(&req->crq_pages)); + LASSERT(cfs_list_empty(&req->crq_pages)); LASSERT(req->crq_nrpages == 0); - LINVRNT(list_empty(&req->crq_layers)); + LINVRNT(cfs_list_empty(&req->crq_layers)); LINVRNT(equi(req->crq_nrobjs > 0, req->crq_o != NULL)); ENTRY; @@ -1361,7 +1475,7 @@ static int cl_req_init(const struct lu_env *env, struct cl_req *req, result = 0; page = cl_page_top(page); do { - list_for_each_entry(slice, &page->cp_layers, cpl_linkage) { + 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, @@ -1387,10 +1501,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 (!list_empty(&req->crq_layers)) { - slice = list_entry(req->crq_layers.prev, - struct cl_req_slice, crs_linkage); - list_del_init(&slice->crs_linkage); + 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); if (slice->crs_ops->cro_completion != NULL) slice->crs_ops->cro_completion(env, slice, rc); } @@ -1446,11 +1560,13 @@ void cl_req_page_add(const struct lu_env *env, ENTRY; page = cl_page_top(page); - LINVRNT(cl_page_is_vmlocked(env, page)); - LASSERT(list_empty(&page->cp_flight)); + LASSERT(cfs_list_empty(&page->cp_flight)); LASSERT(page->cp_req == NULL); - list_add_tail(&page->cp_flight, &req->crq_pages); + 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); ++req->crq_nrpages; page->cp_req = req; obj = cl_object_top(page->cp_obj); @@ -1478,11 +1594,10 @@ void cl_req_page_done(const struct lu_env *env, struct cl_page *page) ENTRY; page = cl_page_top(page); - LINVRNT(cl_page_is_vmlocked(env, page)); - LASSERT(!list_empty(&page->cp_flight)); + LASSERT(!cfs_list_empty(&page->cp_flight)); LASSERT(req->crq_nrpages > 0); - list_del_init(&page->cp_flight); + cfs_list_del_init(&page->cp_flight); --req->crq_nrpages; page->cp_req = NULL; EXIT; @@ -1508,7 +1623,7 @@ int cl_req_prep(const struct lu_env *env, struct cl_req *req) LASSERT(req->crq_o[i].ro_obj != NULL); result = 0; - list_for_each_entry(slice, &req->crq_layers, crs_linkage) { + cfs_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) @@ -1531,14 +1646,14 @@ void cl_req_attr_set(const struct lu_env *env, struct cl_req *req, struct cl_page *page; int i; - LASSERT(!list_empty(&req->crq_pages)); + LASSERT(!cfs_list_empty(&req->crq_pages)); ENTRY; /* Take any page to use as a model. */ - page = list_entry(req->crq_pages.next, struct cl_page, cp_flight); + page = cfs_list_entry(req->crq_pages.next, struct cl_page, cp_flight); for (i = 0; i < req->crq_nrobjs; ++i) { - list_for_each_entry(slice, &req->crq_layers, crs_linkage) { + cfs_list_for_each_entry(slice, &req->crq_layers, crs_linkage) { const struct cl_page_slice *scan; const struct cl_object *obj; @@ -1569,8 +1684,8 @@ EXPORT_SYMBOL(cl_req_attr_set); void cl_sync_io_init(struct cl_sync_io *anchor, int nrpages) { ENTRY; - init_completion(&anchor->csi_sync_completion); - atomic_set(&anchor->csi_sync_nr, nrpages); + cfs_waitq_init(&anchor->csi_waitq); + cfs_atomic_set(&anchor->csi_sync_nr, nrpages); anchor->csi_sync_rc = 0; EXIT; } @@ -1581,25 +1696,34 @@ EXPORT_SYMBOL(cl_sync_io_init); * cl_sync_io_note() for every page. */ int cl_sync_io_wait(const struct lu_env *env, struct cl_io *io, - struct cl_page_list *queue, struct cl_sync_io *anchor) + struct cl_page_list *queue, struct cl_sync_io *anchor, + long timeout) { + struct l_wait_info lwi = LWI_TIMEOUT_INTR(cfs_time_seconds(timeout), + NULL, NULL, NULL); int rc; ENTRY; - rc = wait_for_completion_interruptible(&anchor->csi_sync_completion); + LASSERT(timeout >= 0); + + rc = l_wait_event(anchor->csi_waitq, + cfs_atomic_read(&anchor->csi_sync_nr) == 0, + &lwi); if (rc < 0) { - int rc2; - rc2 = cl_io_cancel(env, io, queue); - if (rc2 < 0) { - /* Too bad, some pages are still in IO. */ - CDEBUG(D_VFSTRACE, "Failed to cancel transfer (%i). " - "Waiting for %i pages\n", - rc2, atomic_read(&anchor->csi_sync_nr)); - wait_for_completion(&anchor->csi_sync_completion); - } - } else + CERROR("SYNC IO failed with error: %d, try to cancel " + "%d remaining pages\n", + rc, cfs_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, + &lwi); + } else { rc = anchor->csi_sync_rc; - LASSERT(atomic_read(&anchor->csi_sync_nr) == 0); + } + LASSERT(cfs_atomic_read(&anchor->csi_sync_nr) == 0); cl_page_list_assume(env, io, queue); POISON(anchor, 0x5a, sizeof *anchor); RETURN(rc); @@ -1619,8 +1743,9 @@ void cl_sync_io_note(struct cl_sync_io *anchor, int ioret) * ->{prepare,commit}_write(). Completion is used to signal the end of * IO. */ - if (atomic_dec_and_test(&anchor->csi_sync_nr)) - complete(&anchor->csi_sync_completion); + 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; } EXPORT_SYMBOL(cl_sync_io_note);