X-Git-Url: https://git.whamcloud.com/?a=blobdiff_plain;f=lustre%2Fobdfilter%2Ffilter_io.c;h=fa9a96f13a84c5156aad54c487a1659fe6f1318e;hb=b23da8fcae05b8c8a0ba679fed1ebe718d93ebb7;hp=c13b9868c1687527d59affa08bbba13563cebe70;hpb=374e8d1e867ee9a9c21b9835cd135271a299089d;p=fs%2Flustre-release.git diff --git a/lustre/obdfilter/filter_io.c b/lustre/obdfilter/filter_io.c index c13b986..fa9a96f1 100644 --- a/lustre/obdfilter/filter_io.c +++ b/lustre/obdfilter/filter_io.c @@ -1,403 +1,621 @@ /* -*- mode: c; c-basic-offset: 8; indent-tabs-mode: nil; -*- * vim:expandtab:shiftwidth=8:tabstop=8: * - * linux/fs/obdfilter/filter_io.c + * GPL HEADER START * - * Copyright (c) 2001-2003 Cluster File Systems, Inc. - * Author: Peter Braam - * Author: Andreas Dilger - * Author: Phil Schwan + * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER. * - * This file is part of Lustre, http://www.lustre.org. + * This program is free software; you can redistribute it and/or modify + * it under the terms of the GNU General Public License version 2 only, + * as published by the Free Software Foundation. * - * Lustre is free software; you can redistribute it and/or - * modify it under the terms of version 2 of the GNU General Public - * License as published by the Free Software Foundation. + * This program is distributed in the hope that it will be useful, but + * WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU + * General Public License version 2 for more details (a copy is included + * in the LICENSE file that accompanied this code). * - * Lustre is distributed in the hope that it will be useful, - * but WITHOUT ANY WARRANTY; without even the implied warranty of - * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the - * GNU General Public License for more details. + * You should have received a copy of the GNU General Public License + * version 2 along with this program; If not, see + * http://www.sun.com/software/products/lustre/docs/GPLv2.pdf * - * You should have received a copy of the GNU General Public License - * along with Lustre; if not, write to the Free Software - * Foundation, Inc., 675 Mass Ave, Cambridge, MA 02139, USA. + * Please contact Sun Microsystems, Inc., 4150 Network Circle, Santa Clara, + * CA 95054 USA or visit www.sun.com if you need additional information or + * have any questions. + * + * GPL HEADER END + */ +/* + * Copyright 2008 Sun Microsystems, Inc. All rights reserved + * Use is subject to license terms. + */ +/* + * This file is part of Lustre, http://www.lustre.org/ + * Lustre is a trademark of Sun Microsystems, Inc. + * + * lustre/obdfilter/filter_io.c + * + * Author: Peter Braam + * Author: Andreas Dilger + * Author: Phil Schwan */ #define DEBUG_SUBSYSTEM S_FILTER +#ifndef AUTOCONF_INCLUDED #include +#endif #include #include // XXX kill me soon #include -#include -#include +#include +#include +#include #include "filter_internal.h" -static int filter_start_page_read(struct inode *inode, struct niobuf_local *lnb) -{ - struct address_space *mapping = inode->i_mapping; - struct page *page; - unsigned long index = lnb->offset >> PAGE_SHIFT; - int rc; +int *obdfilter_created_scratchpad; - page = grab_cache_page(mapping, index); /* locked page */ - if (IS_ERR(page)) - return lnb->rc = PTR_ERR(page); - - LASSERT(page->mapping == mapping); - - lnb->page = page; +/* Grab the dirty and seen grant announcements from the incoming obdo. + * We will later calculate the clients new grant and return it. + * Caller must hold osfs lock */ +void filter_grant_incoming(struct obd_export *exp, struct obdo *oa) +{ + struct filter_export_data *fed; + struct obd_device *obd = exp->exp_obd; + ENTRY; - if (inode->i_size < lnb->offset + lnb->len - 1) - lnb->rc = inode->i_size - lnb->offset; - else - lnb->rc = lnb->len; + LASSERT_SPIN_LOCKED(&obd->obd_osfs_lock); - if (PageUptodate(page)) { - unlock_page(page); - return 0; + if ((oa->o_valid & (OBD_MD_FLBLOCKS|OBD_MD_FLGRANT)) != + (OBD_MD_FLBLOCKS|OBD_MD_FLGRANT)) { + oa->o_valid &= ~OBD_MD_FLGRANT; + EXIT; + return; } - rc = mapping->a_ops->readpage(NULL, page); - if (rc < 0) { - CERROR("page index %lu, rc = %d\n", index, rc); - lnb->page = NULL; - page_cache_release(page); - return lnb->rc = rc; + fed = &exp->exp_filter_data; + + /* Add some margin, since there is a small race if other RPCs arrive + * out-or-order and have already consumed some grant. We want to + * leave this here in case there is a large error in accounting. */ + CDEBUG(D_CACHE, + "%s: cli %s/%p reports grant: "LPU64" dropped: %u, local: %lu\n", + obd->obd_name, exp->exp_client_uuid.uuid, exp, oa->o_grant, + oa->o_dropped, fed->fed_grant); + + /* Update our accounting now so that statfs takes it into account. + * Note that fed_dirty is only approximate and can become incorrect + * if RPCs arrive out-of-order. No important calculations depend + * on fed_dirty however, but we must check sanity to not assert. */ + if ((long long)oa->o_dirty < 0) + oa->o_dirty = 0; + else if (oa->o_dirty > fed->fed_grant + 4 * FILTER_GRANT_CHUNK) + oa->o_dirty = fed->fed_grant + 4 * FILTER_GRANT_CHUNK; + obd->u.filter.fo_tot_dirty += oa->o_dirty - fed->fed_dirty; + if (fed->fed_grant < oa->o_dropped) { + CDEBUG(D_CACHE,"%s: cli %s/%p reports %u dropped > grant %lu\n", + obd->obd_name, exp->exp_client_uuid.uuid, exp, + oa->o_dropped, fed->fed_grant); + oa->o_dropped = 0; + } + if (obd->u.filter.fo_tot_granted < oa->o_dropped) { + CERROR("%s: cli %s/%p reports %u dropped > tot_grant "LPU64"\n", + obd->obd_name, exp->exp_client_uuid.uuid, exp, + oa->o_dropped, obd->u.filter.fo_tot_granted); + oa->o_dropped = 0; + } + obd->u.filter.fo_tot_granted -= oa->o_dropped; + fed->fed_grant -= oa->o_dropped; + fed->fed_dirty = oa->o_dirty; + + if (oa->o_flags & OBD_FL_SHRINK_GRANT) { + obd_size left_space = filter_grant_space_left(exp); + struct filter_obd *filter = &exp->exp_obd->u.filter; + + /*Only if left_space < fo_tot_clients * 32M, + *then the grant space could be shrinked */ + if (left_space < filter->fo_tot_granted_clients * + FILTER_GRANT_SHRINK_LIMIT) { + fed->fed_grant -= oa->o_grant; + filter->fo_tot_granted -= oa->o_grant; + CDEBUG(D_CACHE, "%s: cli %s/%p shrink "LPU64 + "fed_grant %ld total "LPU64"\n", + obd->obd_name, exp->exp_client_uuid.uuid, + exp, oa->o_grant, fed->fed_grant, + filter->fo_tot_granted); + oa->o_grant = 0; + } } - return 0; + if (fed->fed_dirty < 0 || fed->fed_grant < 0 || fed->fed_pending < 0) { + CERROR("%s: cli %s/%p dirty %ld pend %ld grant %ld\n", + obd->obd_name, exp->exp_client_uuid.uuid, exp, + fed->fed_dirty, fed->fed_pending, fed->fed_grant); + spin_unlock(&obd->obd_osfs_lock); + LBUG(); + } + EXIT; } -static int filter_finish_page_read(struct niobuf_local *lnb) +/* Figure out how much space is available between what we've granted + * and what remains in the filesystem. Compensate for ext3 indirect + * block overhead when computing how much free space is left ungranted. + * + * Caller must hold obd_osfs_lock. */ +obd_size filter_grant_space_left(struct obd_export *exp) { - if (lnb->page == NULL) - return 0; + struct obd_device *obd = exp->exp_obd; + int blockbits = obd->u.obt.obt_sb->s_blocksize_bits; + obd_size tot_granted = obd->u.filter.fo_tot_granted, avail, left = 0; + int rc, statfs_done = 0; + + LASSERT_SPIN_LOCKED(&obd->obd_osfs_lock); + + if (cfs_time_before_64(obd->obd_osfs_age, cfs_time_current_64() - HZ)) { +restat: + rc = fsfilt_statfs(obd, obd->u.obt.obt_sb, + cfs_time_current_64() + HZ); + if (rc) /* N.B. statfs can't really fail */ + RETURN(0); + statfs_done = 1; + } - if (PageUptodate(lnb->page)) - return 0; + avail = obd->obd_osfs.os_bavail; + left = avail - (avail >> (blockbits - 3)); /* (d)indirect */ + if (left > GRANT_FOR_LLOG(obd)) { + left = (left - GRANT_FOR_LLOG(obd)) << blockbits; + } else { + left = 0 /* << blockbits */; + } - wait_on_page(lnb->page); - if (!PageUptodate(lnb->page)) { - CERROR("page index %lu/offset "LPX64" not uptodate\n", - lnb->page->index, lnb->offset); - GOTO(err_page, lnb->rc = -EIO); + if (!statfs_done && left < 32 * FILTER_GRANT_CHUNK + tot_granted) { + CDEBUG(D_CACHE, "fs has no space left and statfs too old\n"); + goto restat; } - if (PageError(lnb->page)) { - CERROR("page index %lu/offset "LPX64" has error\n", - lnb->page->index, lnb->offset); - GOTO(err_page, lnb->rc = -EIO); + + if (left >= tot_granted) { + left -= tot_granted; + } else { + if (left < tot_granted - obd->u.filter.fo_tot_pending) { + CERROR("%s: cli %s/%p grant "LPU64" > available " + LPU64" and pending "LPU64"\n", obd->obd_name, + exp->exp_client_uuid.uuid, exp, tot_granted, + left, obd->u.filter.fo_tot_pending); + } + left = 0; } - return 0; + CDEBUG(D_CACHE, "%s: cli %s/%p free: "LPU64" avail: "LPU64" grant "LPU64 + " left: "LPU64" pending: "LPU64"\n", obd->obd_name, + exp->exp_client_uuid.uuid, exp, + obd->obd_osfs.os_bfree << blockbits, avail << blockbits, + tot_granted, left, obd->u.filter.fo_tot_pending); -err_page: - page_cache_release(lnb->page); - lnb->page = NULL; - return lnb->rc; + return left; } -static struct page *lustre_get_page_write(struct inode *inode, - unsigned long index) +/* Calculate how much grant space to allocate to this client, based on how + * much space is currently free and how much of that is already granted. + * + * Caller must hold obd_osfs_lock. */ +long filter_grant(struct obd_export *exp, obd_size current_grant, + obd_size want, obd_size fs_space_left) { - struct address_space *mapping = inode->i_mapping; - struct page *page; - int rc; - - page = grab_cache_page(mapping, index); /* locked page */ - - if (!IS_ERR(page)) { - /* Note: Called with "O" and "PAGE_SIZE" this is essentially - * a no-op for most filesystems, because we write the whole - * page. For partial-page I/O this will read in the page. - */ - rc = mapping->a_ops->prepare_write(NULL, page, 0, PAGE_SIZE); - if (rc) { - CERROR("page index %lu, rc = %d\n", index, rc); - if (rc != -ENOSPC) + struct obd_device *obd = exp->exp_obd; + struct filter_export_data *fed = &exp->exp_filter_data; + int blockbits = obd->u.obt.obt_sb->s_blocksize_bits; + __u64 grant = 0; + + LASSERT_SPIN_LOCKED(&obd->obd_osfs_lock); + + /* Grant some fraction of the client's requested grant space so that + * they are not always waiting for write credits (not all of it to + * avoid overgranting in face of multiple RPCs in flight). This + * essentially will be able to control the OSC_MAX_RIF for a client. + * + * If we do have a large disparity between what the client thinks it + * has and what we think it has, don't grant very much and let the + * client consume its grant first. Either it just has lots of RPCs + * in flight, or it was evicted and its grants will soon be used up. */ + if (want > 0x7fffffff) { + CERROR("%s: client %s/%p requesting > 2GB grant "LPU64"\n", + obd->obd_name, exp->exp_client_uuid.uuid, exp, want); + } else if (current_grant < want && + current_grant < fed->fed_grant + FILTER_GRANT_CHUNK) { + grant = min((want >> blockbits), + (fs_space_left >> blockbits) / 8); + grant <<= blockbits; + + if (grant) { + /* Allow >FILTER_GRANT_CHUNK size when clients + * reconnect due to a server reboot. + */ + if ((grant > FILTER_GRANT_CHUNK) && + (!obd->obd_recovering)) + grant = FILTER_GRANT_CHUNK; + + obd->u.filter.fo_tot_granted += grant; + fed->fed_grant += grant; + if (fed->fed_grant < 0) { + CERROR("%s: cli %s/%p grant %ld want "LPU64 + "current"LPU64"\n", + obd->obd_name, exp->exp_client_uuid.uuid, + exp, fed->fed_grant, want,current_grant); + spin_unlock(&obd->obd_osfs_lock); LBUG(); - GOTO(err_unlock, rc); - } - /* XXX not sure if we need this if we are overwriting page */ - if (PageError(page)) { - CERROR("error on page index %lu, rc = %d\n", index, rc); - LBUG(); - GOTO(err_unlock, rc = -EIO); + } } } - return page; -err_unlock: - unlock_page(page); - page_cache_release(page); - return ERR_PTR(rc); + CDEBUG(D_CACHE, + "%s: cli %s/%p wants: "LPU64" current grant "LPU64 + " granting: "LPU64"\n", obd->obd_name, exp->exp_client_uuid.uuid, + exp, want, current_grant, grant); + CDEBUG(D_CACHE, + "%s: cli %s/%p tot cached:"LPU64" granted:"LPU64 + " num_exports: %d\n", obd->obd_name, exp->exp_client_uuid.uuid, + exp, obd->u.filter.fo_tot_dirty, + obd->u.filter.fo_tot_granted, obd->obd_num_exports); + + return grant; } -#if (LINUX_VERSION_CODE < KERNEL_VERSION(2,5,0)) -int wait_on_page_locked(struct page *page) +/* + * the routine is used to request pages from pagecache + * + * use GFP_NOFS not allowing to enter FS as the client can run on this node + * and we might end waiting on a page he sent in the request we're serving. + * + * use __GFP_HIGHMEM so that the pages can use all of the available memory + * on 32-bit machines + */ +static struct page *filter_get_page(struct obd_device *obd, + struct inode *inode, + obd_off offset) { - waitfor_one_page(page); - return 0; -} + struct page *page; -/* We should only change the file mtime (and not the ctime, like - * update_inode_times() in generic_file_write()) when we only change data. */ -static inline void inode_update_time(struct inode *inode, int ctime_too) -{ - time_t now = CURRENT_TIME; - if (inode->i_mtime == now && (!ctime_too || inode->i_ctime == now)) - return; - inode->i_mtime = now; - if (ctime_too) - inode->i_ctime = now; - mark_inode_dirty_sync(inode); -} -#endif + page = find_or_create_page(inode->i_mapping, offset >> CFS_PAGE_SHIFT, + GFP_NOFS | __GFP_HIGHMEM); + if (unlikely(page == NULL)) + lprocfs_counter_add(obd->obd_stats, LPROC_FILTER_NO_PAGE, 1); -static int lustre_commit_write(struct niobuf_local *lnb) -{ - struct page *page = lnb->page; - unsigned from = lnb->offset & ~PAGE_MASK; - unsigned to = from + lnb->len; - struct inode *inode = page->mapping->host; - int err; - - LASSERT(to <= PAGE_SIZE); - err = page->mapping->a_ops->commit_write(NULL, page, from, to); -#warning 2.4 folks: wait_on_page_locked does NOT return its error here. - if (!err && IS_SYNC(inode)) - wait_on_page_locked(page); - //SetPageUptodate(page); // the client commit_write will do this - - SetPageReferenced(page); - unlock_page(page); - page_cache_release(page); - return err; + return page; } -int filter_get_page_write(struct inode *inode, struct niobuf_local *lnb, - int *pglocked) +/* + * the routine initializes array of local_niobuf from remote_niobuf + */ +static int filter_map_remote_to_local(int objcount, struct obd_ioobj *obj, + struct niobuf_remote *nb, + int *nrpages, struct niobuf_local *res) { - unsigned long index = lnb->offset >> PAGE_SHIFT; - struct address_space *mapping = inode->i_mapping; - struct page *page; - int rc; + struct niobuf_remote *rnb; + struct niobuf_local *lnb; + int i, max; + ENTRY; - //ASSERT_PAGE_INDEX(index, GOTO(err, rc = -EINVAL)); - if (*pglocked) - page = grab_cache_page_nowait(mapping, index); /* locked page */ - else - page = grab_cache_page(mapping, index); /* locked page */ + /* we don't support multiobject RPC yet + * ost_brw_read() and ost_brw_write() check this */ + LASSERT(objcount == 1); + max = *nrpages; + *nrpages = 0; + for (i = 0, rnb = nb, lnb = res; i < obj->ioo_bufcnt; i++, rnb++) { + obd_off offset = rnb->offset; + unsigned int len = rnb->len; - /* This page is currently locked, so get a temporary page instead. */ - if (page == NULL) { - CDEBUG(D_INFO, "ino %lu page %ld locked\n", inode->i_ino,index); - page = alloc_pages(GFP_KERNEL, 0); /* locked page */ - if (page == NULL) { - CERROR("no memory for a temp page\n"); - GOTO(err, rc = -ENOMEM); - } - page->index = index; - lnb->page = page; - lnb->flags |= N_LOCAL_TEMP_PAGE; - } else if (!IS_ERR(page)) { - unsigned from = lnb->offset & ~PAGE_MASK, to = from + lnb->len; - (*pglocked)++; - - rc = mapping->a_ops->prepare_write(NULL, page, from, to); - if (rc) { - if (rc != -ENOSPC) - CERROR("page index %lu, rc = %d\n", index, rc); - GOTO(err_unlock, rc); - } - /* XXX not sure if we need this if we are overwriting page */ - if (PageError(page)) { - CERROR("error on page index %lu, rc = %d\n", index, rc); - LBUG(); - GOTO(err_unlock, rc = -EIO); + while (len > 0) { + int poff = offset & (CFS_PAGE_SIZE - 1); + int plen = CFS_PAGE_SIZE - poff; + + if (*nrpages >= max) { + CERROR("small array of local bufs: %d\n", max); + RETURN(-EINVAL); + } + + if (plen > len) + plen = len; + lnb->offset = offset; + lnb->len = plen; + lnb->flags = rnb->flags; + lnb->page = NULL; + lnb->rc = 0; + lnb->lnb_grant_used = 0; + + LASSERTF(plen <= len, "plen %u, len %u\n", plen, len); + offset += plen; + len -= plen; + lnb++; + (*nrpages)++; } - lnb->page = page; } + RETURN(0); +} + +/* + * the function is used to free all pages used for request + * just to mimic cacheless OSS which don't occupy much memory + */ +void filter_invalidate_cache(struct obd_device *obd, struct obd_ioobj *obj, + struct niobuf_remote *nb, struct inode *inode) +{ + struct niobuf_remote *rnb; + int i; + + LASSERT(inode != NULL); - return 0; + for (i = 0, rnb = nb; i < obj->ioo_bufcnt; i++, rnb++) { + obd_off start; + obd_off end; -err_unlock: - unlock_page(page); - page_cache_release(page); -err: - return lnb->rc = rc; + start = rnb->offset >> CFS_PAGE_SHIFT; + end = (rnb->offset + rnb->len) >> CFS_PAGE_SHIFT; + invalidate_mapping_pages(inode->i_mapping, start, end); + /* just to avoid warnings */ + start = 0; + end = 0; + } } static int filter_preprw_read(int cmd, struct obd_export *exp, struct obdo *oa, int objcount, struct obd_ioobj *obj, - int niocount, struct niobuf_remote *nb, - struct niobuf_local *res, - struct obd_trans_info *oti) + struct niobuf_remote *nb, + int *npages, struct niobuf_local *res, + struct obd_trans_info *oti, + struct lustre_capa *capa) { - struct obd_run_ctxt saved; - struct obd_ioobj *o; - struct niobuf_remote *rnb; + struct obd_device *obd = exp->exp_obd; + struct timeval start, end; + struct lvfs_run_ctxt saved; struct niobuf_local *lnb; - struct fsfilt_objinfo *fso; - struct dentry *dentry; - struct inode *inode; - int rc = 0, i, j, tot_bytes = 0, cleanup_phase = 0; + struct dentry *dentry = NULL; + struct inode *inode = NULL; + void *iobuf = NULL; + int rc = 0, i, tot_bytes = 0; unsigned long now = jiffies; + long timediff; ENTRY; /* We are currently not supporting multi-obj BRW_READ RPCS at all. - * When we do this function's dentry cleanup will need to be fixed */ - LASSERT(objcount == 1); + * When we do this function's dentry cleanup will need to be fixed. + * These values are verified in ost_brw_write() from the wire. */ + LASSERTF(objcount == 1, "%d\n", objcount); + LASSERTF(obj->ioo_bufcnt > 0, "%d\n", obj->ioo_bufcnt); + + rc = filter_auth_capa(exp, NULL, obdo_mdsno(oa), capa, + CAPA_OPC_OSS_READ); + if (rc) + RETURN(rc); - OBD_ALLOC(fso, objcount * sizeof(*fso)); - if (fso == NULL) - RETURN(-ENOMEM); + if (oa && oa->o_valid & OBD_MD_FLGRANT) { + spin_lock(&obd->obd_osfs_lock); + filter_grant_incoming(exp, oa); - memset(res, 0, niocount * sizeof(*res)); + if (!(oa->o_flags & OBD_FL_SHRINK_GRANT)) + oa->o_grant = 0; + spin_unlock(&obd->obd_osfs_lock); + } - push_ctxt(&saved, &exp->exp_obd->u.filter.fo_ctxt, NULL); - for (i = 0, o = obj; i < objcount; i++, o++) { - struct filter_dentry_data *fdd; - LASSERT(o->ioo_bufcnt); + iobuf = filter_iobuf_get(&obd->u.filter, oti); + if (IS_ERR(iobuf)) + RETURN(PTR_ERR(iobuf)); - dentry = filter_oa2dentry(exp->exp_obd, oa); - if (IS_ERR(dentry)) - GOTO(cleanup, rc = PTR_ERR(dentry)); + push_ctxt(&saved, &obd->obd_lvfs_ctxt, NULL); + dentry = filter_oa2dentry(obd, oa); + if (IS_ERR(dentry)) { + rc = PTR_ERR(dentry); + dentry = NULL; + GOTO(cleanup, rc); + } - if (dentry->d_inode == NULL) { - CERROR("trying to BRW to non-existent file "LPU64"\n", - o->ioo_id); - f_dput(dentry); - GOTO(cleanup, rc = -ENOENT); - } + inode = dentry->d_inode; - fso[i].fso_dentry = dentry; - fso[i].fso_bufcnt = o->ioo_bufcnt; + obdo_to_inode(inode, oa, OBD_MD_FLATIME); - fdd = dentry->d_fsdata; - if (fdd == NULL || !atomic_read(&fdd->fdd_open_count)) - CDEBUG(D_PAGE, "I/O to unopened object "LPU64"\n", - o->ioo_id); - } + rc = filter_map_remote_to_local(objcount, obj, nb, npages, res); + if (rc) + GOTO(cleanup, rc); - if (time_after(jiffies, now + 15 * HZ)) - CERROR("slow prep setup %lus\n", (jiffies - now) / HZ); - - for (i = 0, o = obj, rnb = nb, lnb = res; i < objcount; i++, o++) { - dentry = fso[i].fso_dentry; - inode = dentry->d_inode; - - for (j = 0; j < o->ioo_bufcnt; j++, rnb++, lnb++) { - lnb->dentry = dentry; - lnb->offset = rnb->offset; - lnb->len = rnb->len; - lnb->flags = rnb->flags; - lnb->start = jiffies; - - if (inode->i_size <= rnb->offset) { - /* If there's no more data, abort early. - * lnb->page == NULL and lnb->rc == 0, so it's - * easy to detect later. */ - break; - } else { - rc = filter_start_page_read(inode, lnb); - } + fsfilt_check_slow(obd, now, "preprw_read setup"); - if (rc) { - CDEBUG(rc == -ENOSPC ? D_INODE : D_ERROR, - "page err %u@"LPU64" %u/%u %p: rc %d\n", - lnb->len, lnb->offset, j, o->ioo_bufcnt, - dentry, rc); - cleanup_phase = 1; - GOTO(cleanup, rc); - } + /* find pages for all segments, fill array with them */ + do_gettimeofday(&start); + for (i = 0, lnb = res; i < *npages; i++, lnb++) { - tot_bytes += lnb->rc; - if (lnb->rc < lnb->len) { - /* short read, be sure to wait on it */ - lnb++; - break; - } - } - } + lnb->dentry = dentry; - if (time_after(jiffies, now + 15 * HZ)) - CERROR("slow prep get page %lus\n", (jiffies - now) / HZ); + if (i_size_read(inode) <= lnb->offset) + /* If there's no more data, abort early. lnb->rc == 0, + * so it's easy to detect later. */ + break; - lprocfs_counter_add(exp->exp_obd->obd_stats, LPROC_FILTER_READ_BYTES, - tot_bytes); - while (lnb-- > res) { - rc = filter_finish_page_read(lnb); - if (rc) { - CERROR("error page %u@"LPU64" %u %p: rc %d\n", lnb->len, - lnb->offset, (int)(lnb - res), lnb->dentry, rc); - cleanup_phase = 1; - GOTO(cleanup, rc); + lnb->page = filter_get_page(obd, inode, lnb->offset); + if (lnb->page == NULL) + GOTO(cleanup, rc = -ENOMEM); + + lprocfs_counter_add(obd->obd_stats, LPROC_FILTER_CACHE_ACCESS, 1); + + if (i_size_read(inode) < lnb->offset + lnb->len - 1) + lnb->rc = i_size_read(inode) - lnb->offset; + else + lnb->rc = lnb->len; + + tot_bytes += lnb->rc; + + if (PageUptodate(lnb->page)) { + lprocfs_counter_add(obd->obd_stats, + LPROC_FILTER_CACHE_HIT, 1); + continue; } + + lprocfs_counter_add(obd->obd_stats, LPROC_FILTER_CACHE_MISS, 1); + filter_iobuf_add_page(obd, iobuf, inode, lnb->page); } + do_gettimeofday(&end); + timediff = cfs_timeval_sub(&end, &start, NULL); + lprocfs_counter_add(obd->obd_stats, LPROC_FILTER_GET_PAGE, timediff); + + if (OBD_FAIL_CHECK(OBD_FAIL_OST_NOMEM)) + GOTO(cleanup, rc = -ENOMEM); - if (time_after(jiffies, now + 15 * HZ)) - CERROR("slow prep finish page %lus\n", (jiffies - now) / HZ); + fsfilt_check_slow(obd, now, "start_page_read"); + + rc = filter_direct_io(OBD_BRW_READ, dentry, iobuf, + exp, NULL, NULL, NULL); + if (rc) + GOTO(cleanup, rc); + + lprocfs_counter_add(obd->obd_stats, LPROC_FILTER_READ_BYTES, tot_bytes); + + if (exp->exp_nid_stats && exp->exp_nid_stats->nid_stats) + lprocfs_counter_add(exp->exp_nid_stats->nid_stats, + LPROC_FILTER_READ_BYTES, tot_bytes); EXIT; cleanup: - switch (cleanup_phase) { - case 1: - for (lnb = res; lnb < (res + niocount); lnb++) { - if (lnb->page) + /* unlock pages to allow access from concurrent OST_READ */ + for (i = 0, lnb = res; i < *npages; i++, lnb++) { + if (lnb->page) { + LASSERT(PageLocked(lnb->page)); + unlock_page(lnb->page); + + if (rc) { page_cache_release(lnb->page); + lnb->page = NULL; + } } - if (res->dentry != NULL) - f_dput(res->dentry); - else - CERROR("NULL dentry in cleanup -- tell CFS\n"); - res->dentry = NULL; - case 0: - OBD_FREE(fso, objcount * sizeof(*fso)); - pop_ctxt(&saved, &exp->exp_obd->u.filter.fo_ctxt, NULL); } + + if (rc != 0) { + if (dentry != NULL) + f_dput(dentry); + } + + filter_iobuf_put(&obd->u.filter, iobuf, oti); + + pop_ctxt(&saved, &obd->obd_lvfs_ctxt, NULL); + if (rc) + CERROR("io error %d\n", rc); + return rc; } -/* We need to balance prepare_write() calls with commit_write() calls. - * If the page has been prepared, but we have no data for it, we don't - * want to overwrite valid data on disk, but we still need to zero out - * data for space which was newly allocated. Like part of what happens - * in __block_prepare_write() for newly allocated blocks. +/* When clients have dirtied as much space as they've been granted they + * fall through to sync writes. These sync writes haven't been expressed + * in grants and need to error with ENOSPC when there isn't room in the + * filesystem for them after grants are taken into account. However, + * writeback of the dirty data that was already granted space can write + * right on through. * - * XXX currently __block_prepare_write() creates buffers for all the - * pages, and the filesystems mark these buffers as BH_New if they - * were newly allocated from disk. We use the BH_New flag similarly. */ -static int filter_commit_write(struct niobuf_local *lnb, int err) + * Caller must hold obd_osfs_lock. */ +static int filter_grant_check(struct obd_export *exp, struct obdo *oa, + int objcount, struct fsfilt_objinfo *fso, + int niocount, struct niobuf_local *lnb, + obd_size *left, struct inode *inode) { -#if (LINUX_VERSION_CODE < KERNEL_VERSION(2,5,0)) - if (err) { - unsigned block_start, block_end; - struct buffer_head *bh, *head = lnb->page->buffers; - unsigned blocksize = head->b_size; - - /* debugging: just seeing if this ever happens */ - CDEBUG(err == -ENOSPC ? D_INODE : D_ERROR, - "called for ino %lu:%lu on err %d\n", - lnb->page->mapping->host->i_ino, lnb->page->index, err); - - /* Currently one buffer per page, but in the future... */ - for (bh = head, block_start = 0; bh != head || !block_start; - block_start = block_end, bh = bh->b_this_page) { - block_end = block_start + blocksize; - if (buffer_new(bh)) { - memset(kmap(lnb->page) + block_start, 0, - blocksize); - kunmap(lnb->page); + struct filter_export_data *fed = &exp->exp_filter_data; + int blocksize = exp->exp_obd->u.obt.obt_sb->s_blocksize; + unsigned long used = 0, ungranted = 0, using; + int i, rc = -ENOSPC, obj, n = 0; + + LASSERT_SPIN_LOCKED(&exp->exp_obd->obd_osfs_lock); + + for (obj = 0; obj < objcount; obj++) { + for (i = 0; i < fso[obj].fso_bufcnt; i++, n++) { + int tmp, bytes; + + /* should match the code in osc_exit_cache */ + bytes = lnb[n].len; + bytes += lnb[n].offset & (blocksize - 1); + tmp = (lnb[n].offset + lnb[n].len) & (blocksize - 1); + if (tmp) + bytes += blocksize - tmp; + + if ((lnb[n].flags & OBD_BRW_FROM_GRANT) && + (oa->o_valid & OBD_MD_FLGRANT)) { + if (fed->fed_grant < used + bytes) { + CDEBUG(D_CACHE, + "%s: cli %s/%p claims %ld+%d " + "GRANT, real grant %lu idx %d\n", + exp->exp_obd->obd_name, + exp->exp_client_uuid.uuid, exp, + used, bytes, fed->fed_grant, n); + } else { + used += bytes; + lnb[n].flags |= OBD_BRW_GRANTED; + lnb[n].lnb_grant_used = bytes; + CDEBUG(0, "idx %d used=%lu\n", n, used); + rc = 0; + continue; + } } + if (*left > ungranted + bytes) { + /* if enough space, pretend it was granted */ + ungranted += bytes; + lnb[n].flags |= OBD_BRW_GRANTED; + lnb[n].lnb_grant_used = bytes; + CDEBUG(0, "idx %d ungranted=%lu\n",n,ungranted); + rc = 0; + continue; + } + + /* We can't check for already-mapped blocks here, as + * it requires dropping the osfs lock to do the bmap. + * Instead, we return ENOSPC and in that case we need + * to go through and verify if all of the blocks not + * marked BRW_GRANTED are already mapped and we can + * ignore this error. */ + lnb[n].rc = -ENOSPC; + lnb[n].flags &= ~OBD_BRW_GRANTED; + CDEBUG(D_CACHE,"%s: cli %s/%p idx %d no space for %d\n", + exp->exp_obd->obd_name, + exp->exp_client_uuid.uuid, exp, n, bytes); } } -#endif - return lustre_commit_write(lnb); + + /* Now substract what client have used already. We don't subtract + * this from the tot_granted yet, so that other client's can't grab + * that space before we have actually allocated our blocks. That + * happens in filter_grant_commit() after the writes are done. */ + *left -= ungranted; + fed->fed_grant -= used; + fed->fed_pending += used + ungranted; + exp->exp_obd->u.filter.fo_tot_granted += ungranted; + exp->exp_obd->u.filter.fo_tot_pending += used + ungranted; + + CDEBUG(D_CACHE, + "%s: cli %s/%p used: %lu ungranted: %lu grant: %lu dirty: %lu\n", + exp->exp_obd->obd_name, exp->exp_client_uuid.uuid, exp, used, + ungranted, fed->fed_grant, fed->fed_dirty); + + /* Rough calc in case we don't refresh cached statfs data */ + using = (used + ungranted + 1 ) >> + exp->exp_obd->u.obt.obt_sb->s_blocksize_bits; + if (exp->exp_obd->obd_osfs.os_bavail > using) + exp->exp_obd->obd_osfs.os_bavail -= using; + else + exp->exp_obd->obd_osfs.os_bavail = 0; + + if (fed->fed_dirty < used) { + CERROR("%s: cli %s/%p claims used %lu > fed_dirty %lu\n", + exp->exp_obd->obd_name, exp->exp_client_uuid.uuid, exp, + used, fed->fed_dirty); + used = fed->fed_dirty; + } + exp->exp_obd->u.filter.fo_tot_dirty -= used; + fed->fed_dirty -= used; + + if (fed->fed_dirty < 0 || fed->fed_grant < 0 || fed->fed_pending < 0) { + CERROR("%s: cli %s/%p dirty %ld pend %ld grant %ld\n", + exp->exp_obd->obd_name, exp->exp_client_uuid.uuid, exp, + fed->fed_dirty, fed->fed_pending, fed->fed_grant); + spin_unlock(&exp->exp_obd->obd_osfs_lock); + LBUG(); + } + return rc; } /* If we ever start to support multi-object BRW RPCs, we will need to get locks @@ -412,388 +630,332 @@ static int filter_commit_write(struct niobuf_local *lnb, int err) * bug) or ensure we get the page locks in an appropriate order. */ static int filter_preprw_write(int cmd, struct obd_export *exp, struct obdo *oa, int objcount, struct obd_ioobj *obj, - int niocount, struct niobuf_remote *nb, + struct niobuf_remote *nb, int *npages, struct niobuf_local *res, - struct obd_trans_info *oti) + struct obd_trans_info *oti, + struct lustre_capa *capa) { - struct obd_run_ctxt saved; - struct obd_ioobj *o; - struct niobuf_remote *rnb; - struct niobuf_local *lnb; - struct fsfilt_objinfo *fso; - struct dentry *dentry; - int pglocked = 0, rc = 0, i, j, tot_bytes = 0; - unsigned long now = jiffies; + struct obd_device *obd = exp->exp_obd; + struct timeval start, end; + struct lvfs_run_ctxt saved; + struct niobuf_local *lnb = res; + struct fsfilt_objinfo fso; + struct filter_mod_data *fmd; + struct dentry *dentry = NULL; + void *iobuf; + obd_size left; + unsigned long now = jiffies, timediff; + int rc = 0, i, tot_bytes = 0, cleanup_phase = 0; ENTRY; LASSERT(objcount == 1); + LASSERT(obj->ioo_bufcnt > 0); - OBD_ALLOC(fso, objcount * sizeof(*fso)); - if (fso == NULL) - RETURN(-ENOMEM); - - memset(res, 0, niocount * sizeof(*res)); - - push_ctxt(&saved, &exp->exp_obd->u.filter.fo_ctxt, NULL); - for (i = 0, o = obj; i < objcount; i++, o++) { - struct filter_dentry_data *fdd; - LASSERT(o->ioo_bufcnt); + rc = filter_auth_capa(exp, NULL, obdo_mdsno(oa), capa, + CAPA_OPC_OSS_WRITE); + if (rc) + RETURN(rc); - dentry = filter_oa2dentry(exp->exp_obd, oa); - if (IS_ERR(dentry)) - GOTO(out_objinfo, rc = PTR_ERR(dentry)); + push_ctxt(&saved, &obd->obd_lvfs_ctxt, NULL); + iobuf = filter_iobuf_get(&obd->u.filter, oti); + if (IS_ERR(iobuf)) + GOTO(cleanup, rc = PTR_ERR(iobuf)); + cleanup_phase = 1; + + dentry = filter_fid2dentry(obd, NULL, obj->ioo_gr, + obj->ioo_id); + if (IS_ERR(dentry)) + GOTO(cleanup, rc = PTR_ERR(dentry)); + cleanup_phase = 2; + + if (dentry->d_inode == NULL) { + CERROR("%s: trying to BRW to non-existent file "LPU64"\n", + obd->obd_name, obj->ioo_id); + GOTO(cleanup, rc = -ENOENT); + } - if (dentry->d_inode == NULL) { - CERROR("trying to BRW to non-existent file "LPU64"\n", - o->ioo_id); - f_dput(dentry); - GOTO(out_objinfo, rc = -ENOENT); - } + if (oa->o_valid & (OBD_MD_FLUID | OBD_MD_FLGID) && + dentry->d_inode->i_mode & (S_ISUID | S_ISGID)) { + rc = filter_capa_fixoa(exp, oa, obdo_mdsno(oa), capa); + if (rc) + GOTO(cleanup, rc); + } - fso[i].fso_dentry = dentry; - fso[i].fso_bufcnt = o->ioo_bufcnt; + rc = filter_map_remote_to_local(objcount, obj, nb, npages, res); + if (rc) + GOTO(cleanup, rc); + + fsfilt_check_slow(obd, now, "preprw_write setup"); + + /* Don't update inode timestamps if this write is older than a + * setattr which modifies the timestamps. b=10150 */ + /* XXX when we start having persistent reservations this needs to + * be changed to filter_fmd_get() to create the fmd if it doesn't + * already exist so we can store the reservation handle there. */ + fmd = filter_fmd_find(exp, obj->ioo_id, obj->ioo_gr); + + LASSERT(oa != NULL); + spin_lock(&obd->obd_osfs_lock); + filter_grant_incoming(exp, oa); + if (fmd && fmd->fmd_mactime_xid > oti->oti_xid) + oa->o_valid &= ~(OBD_MD_FLMTIME | OBD_MD_FLCTIME | + OBD_MD_FLATIME); + else + obdo_to_inode(dentry->d_inode, oa, OBD_MD_FLATIME | + OBD_MD_FLMTIME | OBD_MD_FLCTIME); + cleanup_phase = 3; - down(&dentry->d_inode->i_sem); - fdd = dentry->d_fsdata; - if (fdd == NULL || !atomic_read(&fdd->fdd_open_count)) - CDEBUG(D_PAGE, "I/O to unopened object "LPU64"\n", - o->ioo_id); - } + left = filter_grant_space_left(exp); - if (time_after(jiffies, now + 15 * HZ)) - CERROR("slow prep setup %lus\n", (jiffies - now) / HZ); - - LASSERT(oti != NULL); - oti->oti_handle = fsfilt_brw_start(exp->exp_obd, objcount, fso, - niocount, oti); - if (IS_ERR(oti->oti_handle)) { - rc = PTR_ERR(oti->oti_handle); - CDEBUG(rc == -ENOSPC ? D_INODE : D_ERROR, - "error starting transaction: rc = %d\n", rc); - oti->oti_handle = NULL; - GOTO(out_objinfo, rc); - } + fso.fso_dentry = dentry; + fso.fso_bufcnt = *npages; - for (i = 0, o = obj, rnb = nb, lnb = res; i < objcount; i++, o++) { - dentry = fso[i].fso_dentry; - for (j = 0; j < o->ioo_bufcnt; j++, rnb++, lnb++) { - if (j == 0) - lnb->dentry = dentry; - else - lnb->dentry = dget(dentry); + rc = filter_grant_check(exp, oa, objcount, &fso, *npages, res, + &left, dentry->d_inode); - lnb->offset = rnb->offset; - lnb->len = rnb->len; - lnb->flags = rnb->flags; - lnb->start = jiffies; + /* do not zero out oa->o_valid as it is used in filter_commitrw_write() + * for setting UID/GID and fid EA in first write time. */ + if (oa->o_valid & OBD_MD_FLGRANT) + oa->o_grant = filter_grant(exp,oa->o_grant,oa->o_undirty,left); - rc = filter_get_page_write(dentry->d_inode, lnb, - &pglocked); - if (rc) - up(&dentry->d_inode->i_sem); + spin_unlock(&obd->obd_osfs_lock); + filter_fmd_put(exp, fmd); - if (rc) { - CDEBUG(rc == -ENOSPC ? D_INODE : D_ERROR, - "page err %u@"LPU64" %u/%u %p: rc %d\n", - lnb->len, lnb->offset, j, o->ioo_bufcnt, - dentry, rc); - f_dput(dentry); - GOTO(out_pages, rc); + if (rc) + GOTO(cleanup, rc); + + do_gettimeofday(&start); + for (i = 0, lnb = res; i < *npages; i++, lnb++) { + + /* We still set up for ungranted pages so that granted pages + * can be written to disk as they were promised, and portals + * needs to keep the pages all aligned properly. */ + lnb->dentry = dentry; + + lnb->page = filter_get_page(obd, dentry->d_inode, lnb->offset); + if (lnb->page == NULL) + GOTO(cleanup, rc = -ENOMEM); + cleanup_phase = 4; + + /* DLM locking protects us from write and truncate competing + * for same region, but truncate can leave dirty page in the + * cache. it's possible the writeout on a such a page is in + * progress when we access it. it's also possible that during + * this writeout we put new (partial) data, but then won't + * be able to proceed in filter_commitrw_write(). thus let's + * just wait for writeout completion, should be rare enough. + * -bzzz */ + if (obd->u.filter.fo_writethrough_cache) + wait_on_page_writeback(lnb->page); + BUG_ON(PageWriteback(lnb->page)); + + /* If the filter writes a partial page, then has the file + * extended, the client will read in the whole page. the + * filter has to be careful to zero the rest of the partial + * page on disk. we do it by hand for partial extending + * writes, send_bio() is responsible for zeroing pages when + * asked to read unmapped blocks -- brw_kiovec() does this. */ + if (lnb->len != CFS_PAGE_SIZE) { + __s64 maxidx; + + maxidx = ((i_size_read(dentry->d_inode) + + CFS_PAGE_SIZE - 1) >> CFS_PAGE_SHIFT) - 1; + if (maxidx >= lnb->page->index) { + LL_CDEBUG_PAGE(D_PAGE, lnb->page, "write %u @ " + LPU64" flg %x before EOF %llu\n", + lnb->len, lnb->offset,lnb->flags, + i_size_read(dentry->d_inode)); + filter_iobuf_add_page(obd, iobuf, + dentry->d_inode, + lnb->page); + } else { + long off; + char *p = kmap(lnb->page); + + off = lnb->offset & ~CFS_PAGE_MASK; + if (off) + memset(p, 0, off); + off = (lnb->offset + lnb->len) & ~CFS_PAGE_MASK; + if (off) + memset(p + off, 0, CFS_PAGE_SIZE - off); + kunmap(lnb->page); } - tot_bytes += lnb->len; } + if (lnb->rc == 0) + tot_bytes += lnb->len; } + do_gettimeofday(&end); + timediff = cfs_timeval_sub(&end, &start, NULL); + lprocfs_counter_add(obd->obd_stats, LPROC_FILTER_GET_PAGE, timediff); - if (time_after(jiffies, now + 15 * HZ)) - CERROR("slow prep get page %lus\n", (jiffies - now) / HZ); + if (OBD_FAIL_CHECK(OBD_FAIL_OST_NOMEM)) + GOTO(cleanup, rc = -ENOMEM); - lprocfs_counter_add(exp->exp_obd->obd_stats, LPROC_FILTER_WRITE_BYTES, - tot_bytes); + /* don't unlock pages to prevent any access */ + rc = filter_direct_io(OBD_BRW_READ, dentry, iobuf, exp, + NULL, NULL, NULL); - EXIT; -out: - OBD_FREE(fso, objcount * sizeof(*fso)); - /* we saved the journal handle into oti->oti_handle instead */ - current->journal_info = NULL; - pop_ctxt(&saved, &exp->exp_obd->u.filter.fo_ctxt, NULL); - return rc; + fsfilt_check_slow(obd, now, "start_page_write"); -out_pages: - while (lnb-- > res) { - filter_commit_write(lnb, rc); - up(&lnb->dentry->d_inode->i_sem); - f_dput(lnb->dentry); - } - filter_finish_transno(exp, oti, rc); - fsfilt_commit(exp->exp_obd, - filter_parent(exp->exp_obd,S_IFREG,obj->ioo_id)->d_inode, - oti->oti_handle, 0); - goto out; /* dropped the dentry refs already (one per page) */ - -out_objinfo: - for (i = 0; i < objcount && fso[i].fso_dentry; i++) { - up(&fso[i].fso_dentry->d_inode->i_sem); - f_dput(fso[i].fso_dentry); + if (exp->exp_nid_stats && exp->exp_nid_stats->nid_stats) + lprocfs_counter_add(exp->exp_nid_stats->nid_stats, + LPROC_FILTER_WRITE_BYTES, tot_bytes); + EXIT; +cleanup: + switch(cleanup_phase) { + case 4: + if (rc) { + for (i = 0, lnb = res; i < *npages; i++, lnb++) { + if (lnb->page != NULL) { + unlock_page(lnb->page); + page_cache_release(lnb->page); + lnb->page = NULL; + } + } + } + case 3: + filter_iobuf_put(&obd->u.filter, iobuf, oti); + case 2: + pop_ctxt(&saved, &obd->obd_lvfs_ctxt, NULL); + if (rc) + f_dput(dentry); + break; + case 1: + filter_iobuf_put(&obd->u.filter, iobuf, oti); + case 0: + spin_lock(&obd->obd_osfs_lock); + if (oa) + filter_grant_incoming(exp, oa); + spin_unlock(&obd->obd_osfs_lock); + pop_ctxt(&saved, &obd->obd_lvfs_ctxt, NULL); + break; + default:; } - goto out; + return rc; } int filter_preprw(int cmd, struct obd_export *exp, struct obdo *oa, - int objcount, struct obd_ioobj *obj, int niocount, - struct niobuf_remote *nb, struct niobuf_local *res, - struct obd_trans_info *oti) + int objcount, struct obd_ioobj *obj, + struct niobuf_remote *nb, int *npages, + struct niobuf_local *res, struct obd_trans_info *oti, + struct lustre_capa *capa) { if (cmd == OBD_BRW_WRITE) return filter_preprw_write(cmd, exp, oa, objcount, obj, - niocount, nb, res, oti); - + nb, npages, res, oti, capa); if (cmd == OBD_BRW_READ) return filter_preprw_read(cmd, exp, oa, objcount, obj, - niocount, nb, res, oti); - + nb, npages, res, oti, capa); LBUG(); - return -EPROTO; } -/* It is highly unlikely that we would ever get an error here. The page we want - * to get was previously locked, so it had to have already allocated the space, - * and we were just writing over the same data, so there would be no hole in the - * file. - * - * XXX: possibility of a race with truncate could exist, need to check that. - * There are no guarantees w.r.t. write order even on a local filesystem, - * although the normal response would be to return the number of bytes - * successfully written and leave the rest to the app. */ -static int filter_write_locked_page(struct niobuf_local *lnb) -{ - struct page *lpage; - void *lpage_addr, *lnb_addr; - int rc; - ENTRY; - - lpage = lustre_get_page_write(lnb->dentry->d_inode, lnb->page->index); - if (IS_ERR(lpage)) { - rc = PTR_ERR(lpage); - CERROR("error getting locked page index %ld: rc = %d\n", - lnb->page->index, rc); - LBUG(); - lustre_commit_write(lnb); - RETURN(rc); - } - - /* 2 kmaps == vanishingly small deadlock opportunity */ - lpage_addr = kmap(lpage); - lnb_addr = kmap(lnb->page); - - memcpy(lpage_addr, lnb_addr, PAGE_SIZE); - - kunmap(lnb->page); - kunmap(lpage); - - page_cache_release(lnb->page); - - lnb->page = lpage; - rc = lustre_commit_write(lnb); - if (rc) - CERROR("error committing locked page %ld: rc = %d\n", - lnb->page->index, rc); - RETURN(rc); -} - -static int filter_commitrw_read(struct obd_export *exp, int objcount, - struct obd_ioobj *obj, int niocount, - struct niobuf_local *res, - struct obd_trans_info *oti) +static int filter_commitrw_read(struct obd_export *exp, struct obdo *oa, + int objcount, struct obd_ioobj *obj, + struct niobuf_remote *rnb, + int npages, struct niobuf_local *res, + struct obd_trans_info *oti, int rc) { - struct obd_ioobj *o; + struct filter_obd *fo = &exp->exp_obd->u.filter; + struct inode *inode = NULL; + struct ldlm_res_id res_id; + struct ldlm_resource *resource = NULL; + struct ldlm_namespace *ns = exp->exp_obd->obd_namespace; struct niobuf_local *lnb; - int i, j; + int i; ENTRY; - for (i = 0, o = obj, lnb = res; i < objcount; i++, o++) { - for (j = 0 ; j < o->ioo_bufcnt ; j++, lnb++) { - if (lnb->page != NULL) - page_cache_release(lnb->page); + osc_build_res_name(obj->ioo_id, obj->ioo_gr, &res_id); + /* If oa != NULL then filter_preprw_read updated the inode atime + * and we should update the lvb so that other glimpses will also + * get the updated value. bug 5972 */ + if (oa && ns && ns->ns_lvbo && ns->ns_lvbo->lvbo_update) { + resource = ldlm_resource_get(ns, NULL, &res_id, LDLM_EXTENT, 0); + + if (resource != NULL) { + LDLM_RESOURCE_ADDREF(resource); + ns->ns_lvbo->lvbo_update(resource, NULL, 0, 1); + LDLM_RESOURCE_DELREF(resource); + ldlm_resource_putref(resource); } } - if (res->dentry != NULL) - f_dput(res->dentry); - RETURN(0); -} - -static int -filter_commitrw_write(int cmd, struct obd_export *exp, struct obdo *oa, - int objcount, struct obd_ioobj *obj, int niocount, - struct niobuf_local *res, struct obd_trans_info *oti) -{ - struct obd_run_ctxt saved; - struct obd_ioobj *o; - struct niobuf_local *lnb; - struct obd_device *obd = exp->exp_obd; - int found_locked = 0, rc = 0, i; - int nested_trans = current->journal_info != NULL; - unsigned long now = jiffies; /* DEBUGGING OST TIMEOUTS */ - ENTRY; - - push_ctxt(&saved, &obd->u.filter.fo_ctxt, NULL); - - if (cmd & OBD_BRW_WRITE) { - LASSERT(oti); - LASSERT(current->journal_info == NULL || - current->journal_info == oti->oti_handle); - current->journal_info = oti->oti_handle; - } - - for (i = 0, o = obj, lnb = res; i < objcount; i++, o++) { - struct inode *inode; - int j; - - /* If all of the page reads were beyond EOF, let's pretend - * this read didn't really happen at all. */ - if (lnb->dentry == NULL) { - oa->o_valid = OBD_MD_FLID|(oa->o_valid&OBD_MD_FLCKSUM); - continue; - } - - inode = igrab(lnb->dentry->d_inode); - - if (cmd & OBD_BRW_WRITE) { - /* FIXME: MULTI OBJECT BRW */ - if (oa && oa->o_valid & (OBD_MD_FLMTIME|OBD_MD_FLCTIME)) - obdo_refresh_inode(inode, oa, OBD_MD_FLATIME | - OBD_MD_FLMTIME | - OBD_MD_FLCTIME); - else - inode_update_time(lnb->dentry->d_inode, 1); - } else if (oa && oa->o_valid & OBD_MD_FLATIME) { - /* Note that we don't necessarily write this to disk */ - obdo_refresh_inode(inode, oa, OBD_MD_FLATIME); - } - - for (j = 0 ; j < o->ioo_bufcnt ; j++, lnb++) { - if (lnb->page == NULL) { - continue; - } - - if (lnb->flags & N_LOCAL_TEMP_PAGE) { - found_locked++; - continue; - } - - if (time_after(jiffies, lnb->start + 15 * HZ)) - CERROR("slow commitrw %lusi (%lus)\n", - (jiffies - lnb->start) / HZ, - (jiffies - now) / HZ); - - if (cmd & OBD_BRW_WRITE) { - int err = filter_commit_write(lnb, 0); - if (!rc) - rc = err; - } else { - page_cache_release(lnb->page); - } - - f_dput(lnb->dentry); - if (time_after(jiffies, lnb->start + 15 * HZ)) - CERROR("slow commit_write %lus (%lus)\n", - (jiffies - lnb->start) / HZ, - (jiffies - now) / HZ); - } - - /* FIXME: MULTI OBJECT BRW */ - if (oa) { - oa->o_valid = OBD_MD_FLID|(oa->o_valid&OBD_MD_FLCKSUM); - obdo_from_inode(oa, inode, FILTER_VALID_FLAGS); - } - - if (cmd & OBD_BRW_WRITE) - up(&inode->i_sem); - - iput(inode); - } - - for (i = 0, o = obj, lnb = res; found_locked > 0 && i < objcount; - i++, o++) { - int j; - - for (j = 0 ; j < o->ioo_bufcnt ; j++, lnb++) { - int err; - if (!(lnb->flags & N_LOCAL_TEMP_PAGE)) - continue; + if (res->dentry != NULL) + inode = res->dentry->d_inode; - if (time_after(jiffies, lnb->start + 15 * HZ)) - CERROR("slow commitrw locked %lus (%lus)\n", - (jiffies - lnb->start) / HZ, - (jiffies - now) / HZ); - - err = filter_write_locked_page(lnb); - if (!rc) - rc = err; - f_dput(lnb->dentry); - found_locked--; - - if (time_after(jiffies, lnb->start + 15 * HZ)) - CERROR("slow commit_write locked %lus (%lus)\n", - (jiffies - lnb->start) / HZ, - (jiffies - now) / HZ); + for (i = 0, lnb = res; i < npages; i++, lnb++) { + if (lnb->page != NULL) { + page_cache_release(lnb->page); + lnb->page = NULL; } } - if (cmd & OBD_BRW_WRITE) { - /* We just want any dentry for the commit, for now */ - struct dentry *dparent = filter_parent(obd, S_IFREG, 0); - int err; - - rc = filter_finish_transno(exp, oti, rc); - err = fsfilt_commit(obd, dparent->d_inode, oti->oti_handle, - obd_sync_filter); - if (err) - rc = err; - if (obd_sync_filter) - LASSERT(oti->oti_transno <= obd->obd_last_committed); - if (time_after(jiffies, now + 15 * HZ)) - CERROR("slow commitrw commit %lus\n", (jiffies-now)/HZ); - } + if (inode && (fo->fo_read_cache == 0 || + i_size_read(inode) > fo->fo_readcache_max_filesize)) + filter_invalidate_cache(exp->exp_obd, obj, rnb, inode); - LASSERT(nested_trans || current->journal_info == NULL); - pop_ctxt(&saved, &obd->u.filter.fo_ctxt, NULL); + if (res->dentry != NULL) + f_dput(res->dentry); RETURN(rc); } -/* XXX needs to trickle its oa down */ +void filter_grant_commit(struct obd_export *exp, int niocount, + struct niobuf_local *res) +{ + struct filter_obd *filter = &exp->exp_obd->u.filter; + struct niobuf_local *lnb = res; + unsigned long pending = 0; + int i; + + spin_lock(&exp->exp_obd->obd_osfs_lock); + for (i = 0, lnb = res; i < niocount; i++, lnb++) + pending += lnb->lnb_grant_used; + + LASSERTF(exp->exp_filter_data.fed_pending >= pending, + "%s: cli %s/%p fed_pending: %lu grant_used: %lu\n", + exp->exp_obd->obd_name, exp->exp_client_uuid.uuid, exp, + exp->exp_filter_data.fed_pending, pending); + exp->exp_filter_data.fed_pending -= pending; + LASSERTF(filter->fo_tot_granted >= pending, + "%s: cli %s/%p tot_granted: "LPU64" grant_used: %lu\n", + exp->exp_obd->obd_name, exp->exp_client_uuid.uuid, exp, + exp->exp_obd->u.filter.fo_tot_granted, pending); + filter->fo_tot_granted -= pending; + LASSERTF(filter->fo_tot_pending >= pending, + "%s: cli %s/%p tot_pending: "LPU64" grant_used: %lu\n", + exp->exp_obd->obd_name, exp->exp_client_uuid.uuid, exp, + filter->fo_tot_pending, pending); + filter->fo_tot_pending -= pending; + + spin_unlock(&exp->exp_obd->obd_osfs_lock); +} + int filter_commitrw(int cmd, struct obd_export *exp, struct obdo *oa, - int objcount, struct obd_ioobj *obj, int niocount, - struct niobuf_local *res, struct obd_trans_info *oti) + int objcount, struct obd_ioobj *obj, + struct niobuf_remote *nb, int npages, + struct niobuf_local *res, struct obd_trans_info *oti, + int rc) { if (cmd == OBD_BRW_WRITE) - return filter_commitrw_write(cmd, exp, oa, objcount, obj, - niocount, res, oti); + return filter_commitrw_write(exp, oa, objcount, obj, + nb, npages, res, oti, rc); if (cmd == OBD_BRW_READ) - return filter_commitrw_read(exp, objcount, obj, niocount, - res, oti); + return filter_commitrw_read(exp, oa, objcount, obj, + nb, npages, res, oti, rc); LBUG(); return -EPROTO; } -int filter_brw(int cmd, struct lustre_handle *conn, struct obdo *oa, - struct lov_stripe_md *lsm, obd_count oa_bufs, - struct brw_page *pga, struct obd_trans_info *oti) +int filter_brw(int cmd, struct obd_export *exp, struct obd_info *oinfo, + obd_count oa_bufs, struct brw_page *pga, + struct obd_trans_info *oti) { - struct obd_export *exp; struct obd_ioobj ioo; struct niobuf_local *lnb; struct niobuf_remote *rnb; obd_count i; - int ret = 0; + int ret = 0, npages; ENTRY; - exp = class_conn2export(conn); - if (exp == NULL) { - CDEBUG(D_IOCTL, "invalid client cookie "LPX64"\n",conn->cookie); - RETURN(-EINVAL); - } - OBD_ALLOC(lnb, oa_bufs * sizeof(struct niobuf_local)); OBD_ALLOC(rnb, oa_bufs * sizeof(struct niobuf_remote)); @@ -801,42 +963,28 @@ int filter_brw(int cmd, struct lustre_handle *conn, struct obdo *oa, GOTO(out, ret = -ENOMEM); for (i = 0; i < oa_bufs; i++) { + lnb[i].page = pga[i].pg; rnb[i].offset = pga[i].off; rnb[i].len = pga[i].count; } - ioo.ioo_id = oa->o_id; - ioo.ioo_gr = 0; - ioo.ioo_type = oa->o_mode & S_IFMT; + obdo_to_ioobj(oinfo->oi_oa, &ioo); ioo.ioo_bufcnt = oa_bufs; - ret = filter_preprw(cmd, exp, oa, 1, &ioo, oa_bufs, rnb, lnb, oti); + npages = oa_bufs; + ret = filter_preprw(cmd, exp, oinfo->oi_oa, 1, &ioo, + rnb, &npages, lnb, oti, oinfo_capa(oinfo)); if (ret != 0) GOTO(out, ret); + LASSERTF(oa_bufs == npages, "%u != %u\n", oa_bufs, npages); - for (i = 0; i < oa_bufs; i++) { - void *virt = kmap(pga[i].pg); - obd_off off = pga[i].off & ~PAGE_MASK; - void *addr = kmap(lnb[i].page); - - /* 2 kmaps == vanishingly small deadlock opportunity */ - - if (cmd & OBD_BRW_WRITE) - memcpy(addr + off, virt + off, pga[i].count); - else - memcpy(virt + off, addr + off, pga[i].count); - - kunmap(addr); - kunmap(virt); - } - - ret = filter_commitrw(cmd, exp, oa, 1, &ioo, oa_bufs, lnb, oti); + ret = filter_commitrw(cmd, exp, oinfo->oi_oa, 1, &ioo, rnb, + npages, lnb, oti, ret); out: if (lnb) OBD_FREE(lnb, oa_bufs * sizeof(struct niobuf_local)); if (rnb) OBD_FREE(rnb, oa_bufs * sizeof(struct niobuf_remote)); - class_export_put(exp); RETURN(ret); }