/* -*- mode: c; c-basic-offset: 8; indent-tabs-mode: nil; -*-
* vim:expandtab:shiftwidth=8:tabstop=8:
*
- * Lustre Lite I/O Page Cache
+ * Lustre Lite I/O page cache routines shared by different kernel revs
*
* Copyright (c) 2001-2003 Cluster File Systems, Inc.
*
* Foundation, Inc., 675 Mass Ave, Cambridge, MA 02139, USA.
*/
+#ifndef AUTOCONF_INCLUDED
#include <linux/config.h>
+#endif
#include <linux/kernel.h>
#include <linux/mm.h>
#include <linux/string.h>
#include <asm/uaccess.h>
#include <linux/fs.h>
-#if (LINUX_VERSION_CODE > KERNEL_VERSION(2,5,0))
-#include <linux/buffer_head.h>
-#include <linux/mpage.h>
-#include <linux/writeback.h>
-#else
-#include <linux/iobuf.h>
-#endif
#include <linux/stat.h>
#include <asm/uaccess.h>
#include <asm/segment.h>
#define DEBUG_SUBSYSTEM S_LLITE
-#include <linux/lustre_mds.h>
-#include <linux/lustre_lite.h>
+//#include <lustre_mdc.h>
+#include <lustre_lite.h>
#include "llite_internal.h"
#include <linux/lustre_compat25.h>
-/*
- * Remove page from dirty list
- */
-static void __set_page_clean(struct page *page)
-{
- struct address_space *mapping = page->mapping;
- struct inode *inode;
-
- if (!mapping)
- return;
-
- PGCACHE_WRLOCK(mapping);
-
- list_del(&page->list);
- list_add(&page->list, &mapping->clean_pages);
-
- /* XXX doesn't inode_lock protect i_state ? */
- inode = mapping->host;
- if (list_empty(&mapping->dirty_pages)) {
- CDEBUG(D_INODE, "inode clean\n");
- inode->i_state &= ~I_DIRTY_PAGES;
- }
-
- PGCACHE_WRUNLOCK(mapping);
- EXIT;
-}
+#ifndef list_for_each_prev_safe
+#define list_for_each_prev_safe(pos, n, head) \
+ for (pos = (head)->prev, n = pos->prev; pos != (head); \
+ pos = n, n = pos->prev )
+#endif
-void set_page_clean(struct page *page)
-{
- if (PageDirty(page)) {
- ClearPageDirty(page);
- __set_page_clean(page);
- }
-}
+cfs_mem_cache_t *ll_async_page_slab = NULL;
+size_t ll_async_page_slab_size = 0;
/* SYNCHRONOUS I/O to object storage for an inode */
static int ll_brw(int cmd, struct inode *inode, struct obdo *oa,
{
struct ll_inode_info *lli = ll_i2info(inode);
struct lov_stripe_md *lsm = lli->lli_smd;
+ struct obd_info oinfo = { { { 0 } } };
struct brw_page pg;
- int rc;
+ int opc, rc;
ENTRY;
pg.pg = page;
- pg.off = ((obd_off)page->index) << PAGE_SHIFT;
+ pg.off = ((obd_off)page->index) << CFS_PAGE_SHIFT;
- if (cmd == OBD_BRW_WRITE && (pg.off + PAGE_SIZE > inode->i_size))
- pg.count = inode->i_size % PAGE_SIZE;
+ if ((cmd & OBD_BRW_WRITE) && (pg.off+CFS_PAGE_SIZE>i_size_read(inode)))
+ pg.count = i_size_read(inode) % CFS_PAGE_SIZE;
else
- pg.count = PAGE_SIZE;
+ pg.count = CFS_PAGE_SIZE;
- CDEBUG(D_PAGE, "%s %d bytes ino %lu at "LPU64"/"LPX64"\n",
- cmd & OBD_BRW_WRITE ? "write" : "read", pg.count, inode->i_ino,
- pg.off, pg.off);
+ LL_CDEBUG_PAGE(D_PAGE, page, "%s %d bytes ino %lu at "LPU64"/"LPX64"\n",
+ cmd & OBD_BRW_WRITE ? "write" : "read", pg.count,
+ inode->i_ino, pg.off, pg.off);
if (pg.count == 0) {
CERROR("ZERO COUNT: ino %lu: size %p:%Lu(%p:%Lu) idx %lu off "
- LPU64"\n",
- inode->i_ino, inode, inode->i_size, page->mapping->host,
- page->mapping->host->i_size, page->index, pg.off);
+ LPU64"\n", inode->i_ino, inode, i_size_read(inode),
+ page->mapping->host, i_size_read(page->mapping->host),
+ page->index, pg.off);
}
pg.flag = flags;
- if (cmd == OBD_BRW_WRITE)
- lprocfs_counter_add(ll_i2sbi(inode)->ll_stats,
- LPROC_LL_BRW_WRITE, pg.count);
+ if (cmd & OBD_BRW_WRITE)
+ ll_stats_ops_tally(ll_i2sbi(inode), LPROC_LL_BRW_WRITE,
+ pg.count);
else
- lprocfs_counter_add(ll_i2sbi(inode)->ll_stats,
- LPROC_LL_BRW_READ, pg.count);
- rc = obd_brw(cmd, ll_i2obdconn(inode), oa, lsm, 1, &pg, NULL);
- if (rc != 0 && rc != -EIO)
+ ll_stats_ops_tally(ll_i2sbi(inode), LPROC_LL_BRW_READ,
+ pg.count);
+ oinfo.oi_oa = oa;
+ oinfo.oi_md = lsm;
+ /* NB partial write, so we might not have CAPA_OPC_OSS_READ capa */
+ opc = cmd & OBD_BRW_WRITE ? CAPA_OPC_OSS_WRITE : CAPA_OPC_OSS_RW;
+ oinfo.oi_capa = ll_osscapa_get(inode, opc);
+ rc = obd_brw(cmd, ll_i2dtexp(inode), &oinfo, 1, &pg, NULL);
+ capa_put(oinfo.oi_capa);
+ if (rc == 0)
+ obdo_to_inode(inode, oa, OBD_MD_FLBLOCKS);
+ else if (rc != -EIO)
CERROR("error from obd_brw: rc = %d\n", rc);
-
RETURN(rc);
}
-/*
- * we were asked to read a single page but we're going to try and read a batch
- * of pages all at once. this vaguely simulates 2.5's readpages.
- */
-static int ll_readpage(struct file *file, struct page *first_page)
+/* this isn't where truncate starts. roughly:
+ * sys_truncate->ll_setattr_raw->vmtruncate->ll_truncate. setattr_raw grabs
+ * DLM lock on [size, EOF], i_mutex, ->lli_size_sem, and WRITE_I_ALLOC_SEM to
+ * avoid races.
+ *
+ * must be called under ->lli_size_sem */
+void ll_truncate(struct inode *inode)
{
- struct inode *inode = first_page->mapping->host;
struct ll_inode_info *lli = ll_i2info(inode);
- struct page *page = first_page;
- struct list_head *pos;
- struct brw_page *pgs;
- struct obdo *oa;
- unsigned long end_index, extent_end = 0;
- struct ptlrpc_request_set *set;
- int npgs = 0, rc = 0, max_pages;
+ struct obd_info oinfo = { { { 0 } } };
+ struct ost_lvb lvb;
+ struct obdo oa;
+ int rc;
ENTRY;
+ CDEBUG(D_VFSTRACE, "VFS Op:inode=%lu/%u(%p) to %Lu=%#Lx\n",inode->i_ino,
+ inode->i_generation, inode, i_size_read(inode),
+ i_size_read(inode));
- LASSERT(PageLocked(page));
- LASSERT(!PageUptodate(page));
- CDEBUG(D_VFSTRACE, "VFS Op:inode=%lu/%u(%p),offset="LPX64"\n",
- inode->i_ino, inode->i_generation, inode,
- (((obd_off)page->index) << PAGE_SHIFT));
- LASSERT(atomic_read(&file->f_dentry->d_inode->i_count) > 0);
-
- if (inode->i_size <= ((obd_off)page->index) << PAGE_SHIFT) {
- CERROR("reading beyond EOF\n");
- memset(kmap(page), 0, PAGE_SIZE);
- kunmap(page);
- SetPageUptodate(page);
- unlock_page(page);
- RETURN(rc);
+ ll_stats_ops_tally(ll_i2sbi(inode), LPROC_LL_TRUNC, 1);
+ if (lli->lli_size_sem_owner != current) {
+ EXIT;
+ return;
}
- /* try to read the file's preferred block size in a one-er */
- end_index = first_page->index +
- (inode->i_blksize >> PAGE_CACHE_SHIFT);
- if (end_index > (inode->i_size >> PAGE_CACHE_SHIFT))
- end_index = inode->i_size >> PAGE_CACHE_SHIFT;
-
- max_pages = ((end_index - first_page->index) << PAGE_CACHE_SHIFT) >>
- PAGE_SHIFT;
- pgs = kmalloc(max_pages * sizeof(*pgs), GFP_USER);
- if (pgs == NULL)
- RETURN(-ENOMEM);
-
- /*
- * find how far we're allowed to read under the extent ll_file_read
- * is passing us..
- */
- spin_lock(&lli->lli_read_extent_lock);
- list_for_each(pos, &lli->lli_read_extents) {
- struct ll_read_extent *rextent;
- rextent = list_entry(pos, struct ll_read_extent, re_lli_item);
- if (rextent->re_task != current)
- continue;
-
- if (rextent->re_extent.end + PAGE_SIZE < rextent->re_extent.end)
- /* extent wrapping */
- extent_end = ~0;
- else {
- extent_end = (rextent->re_extent.end + PAGE_SIZE)
- << PAGE_CACHE_SHIFT;
- /* 32bit indexes, 64bit extents.. */
- if (((u64)extent_end >> PAGE_CACHE_SHIFT) <
- rextent->re_extent.end)
- extent_end = ~0;
- }
- break;
+ if (!lli->lli_smd) {
+ CDEBUG(D_INODE, "truncate on inode %lu with no objects\n",
+ inode->i_ino);
+ GOTO(out_unlock, 0);
}
- spin_unlock(&lli->lli_read_extent_lock);
-
- if (extent_end == 0) {
- static long next_print;
- if (time_after(jiffies, next_print)) {
- next_print = jiffies + 30 * HZ;
- CDEBUG(D_INODE, "mmap readpage - check locks\n");
- }
- end_index = page->index + 1;
- } else if (extent_end < end_index)
- end_index = extent_end;
-
- /* to balance the find_get_page ref the other pages get that is
- * decrefed on teardown.. */
- page_cache_get(page);
- do {
- unsigned long index ;
-
- pgs[npgs].pg = page;
- pgs[npgs].off = ((obd_off)page->index) << PAGE_CACHE_SHIFT;
- pgs[npgs].flag = 0;
- pgs[npgs].count = PAGE_SIZE;
- /* XXX Workaround for BA OSTs returning short reads at EOF.
- * The linux OST will return the full page, zero-filled at the
- * end, which will just overwrite the data we set here. Bug
- * 593 relates to fixing this properly.
- */
- if (inode->i_size < pgs[npgs].off + PAGE_SIZE) {
- int count = inode->i_size - pgs[npgs].off;
- void *addr = kmap(page);
- pgs[npgs].count = count;
- //POISON(addr, 0x7c, count);
- memset(addr + count, 0, PAGE_SIZE - count);
- kunmap(page);
- }
- npgs++;
- if (npgs == max_pages)
- break;
+ LASSERT(atomic_read(&lli->lli_size_sem.count) <= 0);
+
+ /* XXX I'm pretty sure this is a hack to paper over a more fundamental
+ * race condition. */
+ lov_stripe_lock(lli->lli_smd);
+ inode_init_lvb(inode, &lvb);
+ obd_merge_lvb(ll_i2dtexp(inode), lli->lli_smd, &lvb, 0);
+ if (lvb.lvb_size == i_size_read(inode)) {
+ CDEBUG(D_VFSTRACE, "skipping punch for obj "LPX64", %Lu=%#Lx\n",
+ lli->lli_smd->lsm_object_id, i_size_read(inode),
+ i_size_read(inode));
+ lov_stripe_unlock(lli->lli_smd);
+ GOTO(out_unlock, 0);
+ }
- /*
- * find pages ahead of us that we can read in.
- * grab_cache_page waits on pages that are locked so
- * we first try find_get_page, which doesn't. this stops
- * the worst case behaviour of racing threads waiting on
- * each other, but doesn't remove it entirely.
- */
- for (index = page->index + 1, page = NULL;
- page == NULL && index < end_index; index++) {
-
- /* see if the page already exists and needs updating */
- page = find_get_page(inode->i_mapping, index);
- if (page) {
- if (Page_Uptodate(page) || TryLockPage(page))
- goto out_release;
- if (!page->mapping || Page_Uptodate(page))
- goto out_unlock;
- } else {
- /* ok, we have to create it.. */
- page = grab_cache_page(inode->i_mapping, index);
- if (page == NULL)
- continue;
- if (Page_Uptodate(page))
- goto out_unlock;
+ obd_adjust_kms(ll_i2dtexp(inode), lli->lli_smd, i_size_read(inode), 1);
+ lov_stripe_unlock(lli->lli_smd);
+
+ if (unlikely((ll_i2sbi(inode)->ll_flags & LL_SBI_CHECKSUM) &&
+ (i_size_read(inode) & ~CFS_PAGE_MASK))) {
+ /* If the truncate leaves behind a partial page, update its
+ * checksum. */
+ struct page *page = find_get_page(inode->i_mapping,
+ i_size_read(inode) >>
+ CFS_PAGE_SHIFT);
+ if (page != NULL) {
+ struct ll_async_page *llap = llap_cast_private(page);
+ if (llap != NULL) {
+ llap->llap_checksum =
+ crc32_le(0, kmap(page), CFS_PAGE_SIZE);
+ kunmap(page);
}
-
- break;
-
- out_unlock:
- unlock_page(page);
- out_release:
page_cache_release(page);
- page = NULL;
}
-
- } while (page);
-
- if ((oa = obdo_alloc()) == NULL) {
- CERROR("ENOMEM allocing obdo\n");
- rc = -ENOMEM;
- } else if ((set = ptlrpc_prep_set()) == NULL) {
- CERROR("ENOMEM allocing request set\n");
- obdo_free(oa);
- rc = -ENOMEM;
- } else {
- struct ll_file_data *fd = file->private_data;
-
- oa->o_id = lli->lli_smd->lsm_object_id;
- memcpy(obdo_handle(oa), &fd->fd_ost_och.och_fh,
- sizeof(fd->fd_ost_och.och_fh));
- oa->o_valid = OBD_MD_FLID | OBD_MD_FLHANDLE;
- obdo_from_inode(oa, inode, OBD_MD_FLTYPE | OBD_MD_FLATIME);
-
- rc = obd_brw_async(OBD_BRW_READ, ll_i2obdconn(inode), oa,
- ll_i2info(inode)->lli_smd, npgs, pgs,
- set, NULL);
- if (rc == 0)
- rc = ptlrpc_set_wait(set);
- ptlrpc_set_destroy(set);
- if (rc == 0)
- obdo_refresh_inode(inode, oa, oa->o_valid);
- if (rc && rc != -EIO)
- CERROR("error from obd_brw_async: rc = %d\n", rc);
- obdo_free(oa);
- }
-
- while (npgs-- > 0) {
- page = pgs[npgs].pg;
-
- if (rc == 0)
- SetPageUptodate(page);
- unlock_page(page);
- page_cache_release(page);
}
- kfree(pgs);
- RETURN(rc);
-} /* ll_readpage */
+ CDEBUG(D_INFO, "calling punch for "LPX64" (new size %Lu=%#Lx)\n",
+ lli->lli_smd->lsm_object_id, i_size_read(inode), i_size_read(inode));
-/* this isn't where truncate starts. roughly:
- * sys_truncate->ll_setattr_raw->vmtruncate->ll_truncate
- * we grab the lock back in setattr_raw to avoid races. */
-void ll_truncate(struct inode *inode)
-{
- struct lov_stripe_md *lsm = ll_i2info(inode)->lli_smd;
- struct obdo oa;
- int err;
- ENTRY;
- CDEBUG(D_VFSTRACE, "VFS Op:inode=%lu/%u(%p)\n", inode->i_ino,
- inode->i_generation, inode);
+ oinfo.oi_md = lli->lli_smd;
+ oinfo.oi_policy.l_extent.start = i_size_read(inode);
+ oinfo.oi_policy.l_extent.end = OBD_OBJECT_EOF;
+ oinfo.oi_oa = &oa;
+ oa.o_id = lli->lli_smd->lsm_object_id;
+ oa.o_gr = lli->lli_smd->lsm_object_gr;
+ oa.o_valid = OBD_MD_FLID | OBD_MD_FLGROUP;
- /* object not yet allocated */
- if (!lsm) {
- CERROR("truncate on inode %lu with no objects\n", inode->i_ino);
- EXIT;
- return;
- }
+ obdo_from_inode(&oa, inode, OBD_MD_FLTYPE | OBD_MD_FLMODE |
+ OBD_MD_FLATIME | OBD_MD_FLMTIME | OBD_MD_FLCTIME |
+ OBD_MD_FLFID | OBD_MD_FLGENER);
- /* vmtruncate will just throw away our dirty pages, make sure
- * we don't think they're still dirty, being careful to round
- * i_size to the first whole page that was tossed */
- err = ll_clear_dirty_pages(ll_i2obdconn(inode), lsm,
- (inode->i_size + PAGE_CACHE_SIZE-1) >> PAGE_CACHE_SHIFT,
- ~0);
+ ll_inode_size_unlock(inode, 0);
- oa.o_id = lsm->lsm_object_id;
- oa.o_valid = OBD_MD_FLID;
- obdo_from_inode(&oa, inode, OBD_MD_FLTYPE|OBD_MD_FLMODE|OBD_MD_FLATIME|
- OBD_MD_FLMTIME | OBD_MD_FLCTIME);
-
- CDEBUG(D_INFO, "calling punch for "LPX64" (all bytes after %Lu)\n",
- oa.o_id, inode->i_size);
-
- /* truncate == punch from new size to absolute end of file */
- err = obd_punch(ll_i2obdconn(inode), &oa, lsm, inode->i_size,
- OBD_OBJECT_EOF, NULL);
- if (err)
- CERROR("obd_truncate fails (%d) ino %lu\n", err, inode->i_ino);
+ oinfo.oi_capa = ll_osscapa_get(inode, CAPA_OPC_OSS_TRUNC);
+ rc = obd_punch_rqset(ll_i2dtexp(inode), &oinfo, NULL);
+ ll_truncate_free_capa(oinfo.oi_capa);
+ if (rc)
+ CERROR("obd_truncate fails (%d) ino %lu\n", rc, inode->i_ino);
else
obdo_to_inode(inode, &oa, OBD_MD_FLSIZE | OBD_MD_FLBLOCKS |
- OBD_MD_FLATIME | OBD_MD_FLMTIME |
- OBD_MD_FLCTIME);
-
+ OBD_MD_FLATIME | OBD_MD_FLMTIME | OBD_MD_FLCTIME);
EXIT;
return;
-} /* ll_truncate */
-//#if (LINUX_VERSION_CODE < KERNEL_VERSION(2,5,0))
+ out_unlock:
+ ll_inode_size_unlock(inode, 0);
+} /* ll_truncate */
-static int ll_prepare_write(struct file *file, struct page *page, unsigned from,
- unsigned to)
+int ll_prepare_write(struct file *file, struct page *page, unsigned from,
+ unsigned to)
{
struct inode *inode = page->mapping->host;
struct ll_inode_info *lli = ll_i2info(inode);
- struct ll_file_data *fd = file->private_data;
struct lov_stripe_md *lsm = lli->lli_smd;
- obd_off offset = ((obd_off)page->index) << PAGE_SHIFT;
- struct brw_page pg;
+ obd_off offset = ((obd_off)page->index) << CFS_PAGE_SHIFT;
+ struct obd_info oinfo = { { { 0 } } };
+ struct brw_page pga;
struct obdo oa;
+ struct ost_lvb lvb;
int rc = 0;
ENTRY;
- if (!PageLocked(page))
- LBUG();
-
- if (PageUptodate(page))
- RETURN(0);
-
- //POISON(addr + from, 0xca, to - from);
+ LASSERT(PageLocked(page));
+ (void)llap_cast_private(page); /* assertion */
/* Check to see if we should return -EIO right away */
- pg.pg = page;
- pg.off = offset;
- pg.count = PAGE_SIZE;
- pg.flag = 0;
- rc = obd_brw(OBD_BRW_CHECK, ll_i2obdconn(inode), NULL, lsm, 1,&pg,NULL);
+ pga.pg = page;
+ pga.off = offset;
+ pga.count = CFS_PAGE_SIZE;
+ pga.flag = 0;
+
+ oa.o_mode = inode->i_mode;
+ oa.o_id = lsm->lsm_object_id;
+ oa.o_gr = lsm->lsm_object_gr;
+ oa.o_valid = OBD_MD_FLID | OBD_MD_FLMODE |
+ OBD_MD_FLTYPE | OBD_MD_FLGROUP;
+ obdo_from_inode(&oa, inode, OBD_MD_FLFID | OBD_MD_FLGENER);
+
+ oinfo.oi_oa = &oa;
+ oinfo.oi_md = lsm;
+ rc = obd_brw(OBD_BRW_CHECK, ll_i2dtexp(inode), &oinfo, 1, &pga, NULL);
if (rc)
RETURN(rc);
+ if (PageUptodate(page)) {
+ LL_CDEBUG_PAGE(D_PAGE, page, "uptodate\n");
+ RETURN(0);
+ }
+
/* We're completely overwriting an existing page, so _don't_ set it up
* to date until commit_write */
- if (from == 0 && to == PAGE_SIZE)
+ if (from == 0 && to == CFS_PAGE_SIZE) {
+ LL_CDEBUG_PAGE(D_PAGE, page, "full page write\n");
+ POISON_PAGE(page, 0x11);
RETURN(0);
+ }
- /* If are writing to a new page, no need to read old data.
- * the extent locking and getattr procedures in ll_file_write have
- * guaranteed that i_size is stable enough for our zeroing needs */
- if (inode->i_size <= offset) {
- memset(kmap(page), 0, PAGE_SIZE);
+ /* If are writing to a new page, no need to read old data. The extent
+ * locking will have updated the KMS, and for our purposes here we can
+ * treat it like i_size. */
+ lov_stripe_lock(lsm);
+ inode_init_lvb(inode, &lvb);
+ obd_merge_lvb(ll_i2dtexp(inode), lsm, &lvb, 1);
+ lov_stripe_unlock(lsm);
+ if (lvb.lvb_size <= offset) {
+ LL_CDEBUG_PAGE(D_PAGE, page, "kms "LPU64" <= offset "LPU64"\n",
+ lvb.lvb_size, offset);
+ memset(kmap(page), 0, CFS_PAGE_SIZE);
kunmap(page);
GOTO(prepare_done, rc = 0);
}
- oa.o_id = lsm->lsm_object_id;
- oa.o_mode = inode->i_mode;
- memcpy(obdo_handle(&oa), &fd->fd_ost_och.och_fh,
- sizeof(fd->fd_ost_och.och_fh));
- oa.o_valid = OBD_MD_FLID |OBD_MD_FLMODE |OBD_MD_FLTYPE |OBD_MD_FLHANDLE;
-
+ /* XXX could be an async ocp read.. read-ahead? */
rc = ll_brw(OBD_BRW_READ, inode, &oa, page, 0);
- if (rc == 0)
+ if (rc == 0) {
+ /* bug 1598: don't clobber blksize */
+ oa.o_valid &= ~(OBD_MD_FLSIZE | OBD_MD_FLBLKSZ);
obdo_refresh_inode(inode, &oa, oa.o_valid);
+ }
EXIT;
prepare_done:
return rc;
}
-/*
- * background file writeback. This is called regularly from kupdated to write
- * dirty data, from kswapd when memory is low, and from filemap_fdatasync when
- * super blocks or inodes are synced..
+static int ll_ap_make_ready(void *data, int cmd)
+{
+ struct ll_async_page *llap;
+ struct page *page;
+ ENTRY;
+
+ llap = LLAP_FROM_COOKIE(data);
+ page = llap->llap_page;
+
+ LASSERTF(!(cmd & OBD_BRW_READ), "cmd %x page %p ino %lu index %lu\n", cmd, page,
+ page->mapping->host->i_ino, page->index);
+
+ /* we're trying to write, but the page is locked.. come back later */
+ if (TryLockPage(page))
+ RETURN(-EAGAIN);
+
+ LASSERT(!PageWriteback(page));
+
+ /* if we left PageDirty we might get another writepage call
+ * in the future. list walkers are bright enough
+ * to check page dirty so we can leave it on whatever list
+ * its on. XXX also, we're called with the cli list so if
+ * we got the page cache list we'd create a lock inversion
+ * with the removepage path which gets the page lock then the
+ * cli lock */
+ LASSERTF(!PageWriteback(page),"cmd %x page %p ino %lu index %lu\n", cmd, page,
+ page->mapping->host->i_ino, page->index);
+ clear_page_dirty_for_io(page);
+
+ /* This actually clears the dirty bit in the radix tree.*/
+ set_page_writeback(page);
+
+ LL_CDEBUG_PAGE(D_PAGE, page, "made ready\n");
+ page_cache_get(page);
+
+ RETURN(0);
+}
+
+/* We have two reasons for giving llite the opportunity to change the
+ * write length of a given queued page as it builds the RPC containing
+ * the page:
*
- * obd_brw errors down in _batch_writepage are ignored, so pages are always
- * unlocked. Also, there is nobody to return an error code to from here - the
- * application may not even be running anymore.
+ * 1) Further extending writes may have landed in the page cache
+ * since a partial write first queued this page requiring us
+ * to write more from the page cache. (No further races are possible, since
+ * by the time this is called, the page is locked.)
+ * 2) We might have raced with truncate and want to avoid performing
+ * write RPCs that are just going to be thrown away by the
+ * truncate's punch on the storage targets.
*
- * this should be async so that things like kswapd can have a chance to
- * free some more pages that our allocating writeback may need, but it isn't
- * yet.
+ * The kms serves these purposes as it is set at both truncate and extending
+ * writes.
*/
-#if (LINUX_VERSION_CODE < KERNEL_VERSION(2,5,0))
-static unsigned long ll_local_cache_dirty_pages;
-static unsigned long ll_max_dirty_pages = 20 * 1024 * 1024 / PAGE_SIZE;
+static int ll_ap_refresh_count(void *data, int cmd)
+{
+ struct ll_inode_info *lli;
+ struct ll_async_page *llap;
+ struct lov_stripe_md *lsm;
+ struct page *page;
+ struct inode *inode;
+ struct ost_lvb lvb;
+ __u64 kms;
+ ENTRY;
-static spinlock_t ll_local_cache_page_count_lock = SPIN_LOCK_UNLOCKED;
+ /* readpage queues with _COUNT_STABLE, shouldn't get here. */
+ LASSERT(cmd != OBD_BRW_READ);
-int ll_rd_dirty_pages(char *page, char **start, off_t off, int count, int *eof,
- void *data)
-{
- unsigned long dirty_count;
- spin_lock(&ll_local_cache_page_count_lock);
- dirty_count = ll_local_cache_dirty_pages;
- spin_unlock(&ll_local_cache_page_count_lock);
- return snprintf(page, count, "%lu\n", dirty_count);
+ llap = LLAP_FROM_COOKIE(data);
+ page = llap->llap_page;
+ inode = page->mapping->host;
+ lli = ll_i2info(inode);
+ lsm = lli->lli_smd;
+
+ lov_stripe_lock(lsm);
+ inode_init_lvb(inode, &lvb);
+ obd_merge_lvb(ll_i2dtexp(inode), lsm, &lvb, 1);
+ kms = lvb.lvb_size;
+ lov_stripe_unlock(lsm);
+
+ /* catch race with truncate */
+ if (((__u64)page->index << CFS_PAGE_SHIFT) >= kms)
+ return 0;
+
+ /* catch sub-page write at end of file */
+ if (((__u64)page->index << CFS_PAGE_SHIFT) + CFS_PAGE_SIZE > kms)
+ return kms % CFS_PAGE_SIZE;
+
+ return CFS_PAGE_SIZE;
}
-int ll_rd_max_dirty_pages(char *page, char **start, off_t off, int count,
- int *eof, void *data)
+void ll_inode_fill_obdo(struct inode *inode, int cmd, struct obdo *oa)
{
- unsigned long max_dirty;
- spin_lock(&ll_local_cache_page_count_lock);
- max_dirty = ll_max_dirty_pages;
- spin_unlock(&ll_local_cache_page_count_lock);
- return snprintf(page, count, "%lu\n", max_dirty);
+ struct lov_stripe_md *lsm;
+ obd_flag valid_flags;
+
+ lsm = ll_i2info(inode)->lli_smd;
+
+ oa->o_id = lsm->lsm_object_id;
+ oa->o_gr = lsm->lsm_object_gr;
+ oa->o_valid = OBD_MD_FLID | OBD_MD_FLGROUP;
+ valid_flags = OBD_MD_FLTYPE | OBD_MD_FLATIME;
+ if (cmd & OBD_BRW_WRITE) {
+ oa->o_valid |= OBD_MD_FLEPOCH;
+ oa->o_easize = ll_i2info(inode)->lli_ioepoch;
+
+ valid_flags |= OBD_MD_FLMTIME | OBD_MD_FLCTIME |
+ OBD_MD_FLUID | OBD_MD_FLGID |
+ OBD_MD_FLFID | OBD_MD_FLGENER;
+ }
+
+ obdo_from_inode(oa, inode, valid_flags);
}
-int ll_wr_max_dirty_pages(struct file *file, const char *buffer,
- unsigned long count, void *data)
+static void ll_ap_fill_obdo(void *data, int cmd, struct obdo *oa)
{
- unsigned long max_dirty;
- signed long max_dirty_signed;
- char kernbuf[20], *end;
-
- if (count > (sizeof(kernbuf) - 1))
- return -EINVAL;
+ struct ll_async_page *llap;
+ ENTRY;
- if (copy_from_user(kernbuf, buffer, count))
- return -EFAULT;
+ llap = LLAP_FROM_COOKIE(data);
+ ll_inode_fill_obdo(llap->llap_page->mapping->host, cmd, oa);
- kernbuf[count] = '\0';
+ EXIT;
+}
- max_dirty_signed = simple_strtol(kernbuf, &end, 0);
- if (kernbuf == end)
- return -EINVAL;
- max_dirty = (unsigned long)max_dirty_signed;
+static void ll_ap_update_obdo(void *data, int cmd, struct obdo *oa,
+ obd_valid valid)
+{
+ struct ll_async_page *llap;
+ ENTRY;
-#if 0
- if (max_dirty < ll_local_cache_dirty_pages)
- flush_to_new_max_dirty();
-#endif
+ llap = LLAP_FROM_COOKIE(data);
+ obdo_from_inode(oa, llap->llap_page->mapping->host, valid);
- spin_lock(&ll_local_cache_page_count_lock);
- CDEBUG(D_CACHE, "changing max_dirty from %lu to %lu\n",
- ll_max_dirty_pages, max_dirty);
- ll_max_dirty_pages = max_dirty;
- spin_unlock(&ll_local_cache_page_count_lock);
- return count;
+ EXIT;
}
-static int ll_local_cache_full(void)
+static struct obd_capa *ll_ap_lookup_capa(void *data, int cmd)
{
- int full = 0;
- spin_lock(&ll_local_cache_page_count_lock);
- if (ll_max_dirty_pages &&
- ll_local_cache_dirty_pages >= ll_max_dirty_pages) {
- full = 1;
- }
- spin_unlock(&ll_local_cache_page_count_lock);
- /* XXX instrument? */
- /* XXX trigger async writeback when full, or 75% of full? */
- return full;
-}
+ struct ll_async_page *llap = LLAP_FROM_COOKIE(data);
+ int opc = cmd & OBD_BRW_WRITE ? CAPA_OPC_OSS_WRITE : CAPA_OPC_OSS_RW;
-static void ll_local_cache_flushed_pages(unsigned long pgcount)
-{
- unsigned long dirty_count;
- spin_lock(&ll_local_cache_page_count_lock);
- dirty_count = ll_local_cache_dirty_pages;
- ll_local_cache_dirty_pages -= pgcount;
- CDEBUG(D_CACHE, "dirty pages: %lu->%lu)\n",
- dirty_count, ll_local_cache_dirty_pages);
- spin_unlock(&ll_local_cache_page_count_lock);
- LASSERT(dirty_count >= pgcount);
+ return ll_osscapa_get(llap->llap_page->mapping->host, opc);
}
-static void ll_local_cache_dirtied_pages(unsigned long pgcount)
+static struct obd_async_page_ops ll_async_page_ops = {
+ .ap_make_ready = ll_ap_make_ready,
+ .ap_refresh_count = ll_ap_refresh_count,
+ .ap_fill_obdo = ll_ap_fill_obdo,
+ .ap_update_obdo = ll_ap_update_obdo,
+ .ap_completion = ll_ap_completion,
+ .ap_lookup_capa = ll_ap_lookup_capa,
+};
+
+struct ll_async_page *llap_cast_private(struct page *page)
{
- unsigned long dirty_count;
- spin_lock(&ll_local_cache_page_count_lock);
- dirty_count = ll_local_cache_dirty_pages;
- ll_local_cache_dirty_pages += pgcount;
- CDEBUG(D_CACHE, "dirty pages: %lu->%lu\n",
- dirty_count, ll_local_cache_dirty_pages);
- spin_unlock(&ll_local_cache_page_count_lock);
- /* XXX track maximum cached, report to lprocfs */
+ struct ll_async_page *llap = (struct ll_async_page *)page_private(page);
+
+ LASSERTF(llap == NULL || llap->llap_magic == LLAP_MAGIC,
+ "page %p private %lu gave magic %d which != %d\n",
+ page, page_private(page), llap->llap_magic, LLAP_MAGIC);
+
+ return llap;
}
-int ll_clear_dirty_pages(struct lustre_handle *conn, struct lov_stripe_md *lsm,
- unsigned long start, unsigned long end)
+/* Try to shrink the page cache for the @sbi filesystem by 1/@shrink_fraction.
+ *
+ * There is an llap attached onto every page in lustre, linked off @sbi.
+ * We add an llap to the list so we don't lose our place during list walking.
+ * If llaps in the list are being moved they will only move to the end
+ * of the LRU, and we aren't terribly interested in those pages here (we
+ * start at the beginning of the list where the least-used llaps are.
+ */
+int llap_shrink_cache(struct ll_sb_info *sbi, int shrink_fraction)
{
- unsigned long cleared;
- int rc;
+ struct ll_async_page *llap, dummy_llap = { .llap_magic = 0xd11ad11a };
+ unsigned long total, want, count = 0;
- ENTRY;
- rc = obd_clear_dirty_pages(conn, lsm, start, end, &cleared);
- if (!rc)
- ll_local_cache_flushed_pages(cleared);
- RETURN(rc);
+ total = sbi->ll_async_page_count;
+
+ /* There can be a large number of llaps (600k or more in a large
+ * memory machine) so the VM 1/6 shrink ratio is likely too much.
+ * Since we are freeing pages also, we don't necessarily want to
+ * shrink so much. Limit to 40MB of pages + llaps per call. */
+ if (shrink_fraction == 0)
+ want = sbi->ll_async_page_count - sbi->ll_async_page_max + 32;
+ else
+ want = (total + shrink_fraction - 1) / shrink_fraction;
+
+ if (want > 40 << (20 - CFS_PAGE_SHIFT))
+ want = 40 << (20 - CFS_PAGE_SHIFT);
+
+ CDEBUG(D_CACHE, "shrinking %lu of %lu pages (1/%d)\n",
+ want, total, shrink_fraction);
+
+ spin_lock(&sbi->ll_lock);
+ list_add(&dummy_llap.llap_pglist_item, &sbi->ll_pglist);
+
+ while (--total >= 0 && count < want) {
+ struct page *page;
+ int keep;
+
+ if (unlikely(need_resched())) {
+ spin_unlock(&sbi->ll_lock);
+ cond_resched();
+ spin_lock(&sbi->ll_lock);
+ }
+
+ llap = llite_pglist_next_llap(sbi,&dummy_llap.llap_pglist_item);
+ list_del_init(&dummy_llap.llap_pglist_item);
+ if (llap == NULL)
+ break;
+
+ page = llap->llap_page;
+ LASSERT(page != NULL);
+
+ list_add(&dummy_llap.llap_pglist_item, &llap->llap_pglist_item);
+
+ /* Page needs/undergoing IO */
+ if (TryLockPage(page)) {
+ LL_CDEBUG_PAGE(D_PAGE, page, "can't lock\n");
+ continue;
+ }
+
+ keep = (llap->llap_write_queued || PageDirty(page) ||
+ PageWriteback(page) || (!PageUptodate(page) &&
+ llap->llap_origin != LLAP_ORIGIN_READAHEAD));
+
+ LL_CDEBUG_PAGE(D_PAGE, page,"%s LRU page: %s%s%s%s%s origin %s\n",
+ keep ? "keep" : "drop",
+ llap->llap_write_queued ? "wq " : "",
+ PageDirty(page) ? "pd " : "",
+ PageUptodate(page) ? "" : "!pu ",
+ PageWriteback(page) ? "wb" : "",
+ llap->llap_defer_uptodate ? "" : "!du",
+ llap_origins[llap->llap_origin]);
+
+ /* If page is dirty or undergoing IO don't discard it */
+ if (keep) {
+ unlock_page(page);
+ continue;
+ }
+
+ page_cache_get(page);
+ spin_unlock(&sbi->ll_lock);
+
+ if (page->mapping != NULL) {
+ ll_teardown_mmaps(page->mapping,
+ (__u64)page->index << CFS_PAGE_SHIFT,
+ ((__u64)page->index << CFS_PAGE_SHIFT)|
+ ~CFS_PAGE_MASK);
+ if (!PageDirty(page) && !page_mapped(page)) {
+ ll_ra_accounting(llap, page->mapping);
+ ll_truncate_complete_page(page);
+ ++count;
+ } else {
+ LL_CDEBUG_PAGE(D_PAGE, page, "Not dropping page"
+ " because it is "
+ "%s\n",
+ PageDirty(page)?
+ "dirty":"mapped");
+ }
+ }
+ unlock_page(page);
+ page_cache_release(page);
+
+ spin_lock(&sbi->ll_lock);
+ }
+ list_del(&dummy_llap.llap_pglist_item);
+ spin_unlock(&sbi->ll_lock);
+
+ CDEBUG(D_CACHE, "shrank %lu/%lu and left %lu unscanned\n",
+ count, want, total);
+
+ return count;
}
-int ll_mark_dirty_page(struct lustre_handle *conn, struct lov_stripe_md *lsm,
- unsigned long index)
+struct ll_async_page *llap_from_page(struct page *page, unsigned origin)
{
+ struct ll_async_page *llap;
+ struct obd_export *exp;
+ struct inode *inode = page->mapping->host;
+ struct ll_sb_info *sbi;
int rc;
-
ENTRY;
- if (ll_local_cache_full())
- RETURN(-EDQUOT);
- rc = obd_mark_page_dirty(conn, lsm, index);
- if (!rc)
- ll_local_cache_dirtied_pages(1);
- RETURN(rc);
+ if (!inode) {
+ static int triggered;
+
+ if (!triggered) {
+ LL_CDEBUG_PAGE(D_ERROR, page, "Bug 10047. Wrong anon "
+ "page received\n");
+ libcfs_debug_dumpstack(NULL);
+ triggered = 1;
+ }
+ RETURN(ERR_PTR(-EINVAL));
+ }
+ sbi = ll_i2sbi(inode);
+ LASSERT(ll_async_page_slab);
+ LASSERTF(origin < LLAP__ORIGIN_MAX, "%u\n", origin);
+
+ llap = llap_cast_private(page);
+ if (llap != NULL) {
+ /* move to end of LRU list, except when page is just about to
+ * die */
+ if (origin != LLAP_ORIGIN_REMOVEPAGE) {
+ spin_lock(&sbi->ll_lock);
+ sbi->ll_pglist_gen++;
+ list_del_init(&llap->llap_pglist_item);
+ list_add_tail(&llap->llap_pglist_item, &sbi->ll_pglist);
+ spin_unlock(&sbi->ll_lock);
+ }
+ GOTO(out, llap);
+ }
+
+ exp = ll_i2dtexp(page->mapping->host);
+ if (exp == NULL)
+ RETURN(ERR_PTR(-EINVAL));
+
+ /* limit the number of lustre-cached pages */
+ if (sbi->ll_async_page_count >= sbi->ll_async_page_max)
+ llap_shrink_cache(sbi, 0);
+
+ OBD_SLAB_ALLOC(llap, ll_async_page_slab, CFS_ALLOC_STD,
+ ll_async_page_slab_size);
+ if (llap == NULL)
+ RETURN(ERR_PTR(-ENOMEM));
+ llap->llap_magic = LLAP_MAGIC;
+ llap->llap_cookie = (void *)llap + size_round(sizeof(*llap));
+
+ rc = obd_prep_async_page(exp, ll_i2info(inode)->lli_smd, NULL, page,
+ (obd_off)page->index << CFS_PAGE_SHIFT,
+ &ll_async_page_ops, llap, &llap->llap_cookie);
+ if (rc) {
+ OBD_SLAB_FREE(llap, ll_async_page_slab,
+ ll_async_page_slab_size);
+ RETURN(ERR_PTR(rc));
+ }
+
+ CDEBUG(D_CACHE, "llap %p page %p cookie %p obj off "LPU64"\n", llap,
+ page, llap->llap_cookie, (obd_off)page->index << CFS_PAGE_SHIFT);
+ /* also zeroing the PRIVBITS low order bitflags */
+ __set_page_ll_data(page, llap);
+ llap->llap_page = page;
+ spin_lock(&sbi->ll_lock);
+ sbi->ll_pglist_gen++;
+ sbi->ll_async_page_count++;
+ list_add_tail(&llap->llap_pglist_item, &sbi->ll_pglist);
+ INIT_LIST_HEAD(&llap->llap_pending_write);
+ spin_unlock(&sbi->ll_lock);
+
+ out:
+ if (unlikely(sbi->ll_flags & LL_SBI_CHECKSUM)) {
+ __u32 csum = 0;
+ csum = crc32_le(csum, kmap(page), CFS_PAGE_SIZE);
+ kunmap(page);
+ if (origin == LLAP_ORIGIN_READAHEAD ||
+ origin == LLAP_ORIGIN_READPAGE) {
+ llap->llap_checksum = 0;
+ } else if (origin == LLAP_ORIGIN_COMMIT_WRITE ||
+ llap->llap_checksum == 0) {
+ llap->llap_checksum = csum;
+ CDEBUG(D_PAGE, "page %p cksum %x\n", page, csum);
+ } else if (llap->llap_checksum == csum) {
+ /* origin == LLAP_ORIGIN_WRITEPAGE */
+ CDEBUG(D_PAGE, "page %p cksum %x confirmed\n",
+ page, csum);
+ } else {
+ /* origin == LLAP_ORIGIN_WRITEPAGE */
+ LL_CDEBUG_PAGE(D_ERROR, page, "old cksum %x != new "
+ "%x!\n", llap->llap_checksum, csum);
+ }
+ }
+
+ llap->llap_origin = origin;
+ RETURN(llap);
}
-static int ll_writepage(struct page *page)
+static int queue_or_sync_write(struct obd_export *exp, struct inode *inode,
+ struct ll_async_page *llap,
+ unsigned to, obd_flag async_flags)
{
- struct inode *inode = page->mapping->host;
- struct obdo oa;
+ unsigned long size_index = i_size_read(inode) >> CFS_PAGE_SHIFT;
+ struct obd_io_group *oig;
+ struct ll_sb_info *sbi = ll_i2sbi(inode);
+ int rc, noquot = llap->llap_ignore_quota ? OBD_BRW_NOQUOTA : 0;
ENTRY;
- CDEBUG(D_CACHE, "page %p [lau %d] inode %p\n", page,
- PageLaunder(page), inode);
- LASSERT(PageLocked(page));
+ /* _make_ready only sees llap once we've unlocked the page */
+ llap->llap_write_queued = 1;
+ rc = obd_queue_async_io(exp, ll_i2info(inode)->lli_smd, NULL,
+ llap->llap_cookie, OBD_BRW_WRITE | noquot,
+ 0, 0, 0, async_flags);
+ if (rc == 0) {
+ LL_CDEBUG_PAGE(D_PAGE, llap->llap_page, "write queued\n");
+ GOTO(out, 0);
+ }
+
+ llap->llap_write_queued = 0;
+ /* Do not pass llap here as it is sync write. */
+ llap_write_pending(inode, NULL);
+
+ rc = oig_init(&oig);
+ if (rc)
+ GOTO(out, rc);
+
+ /* make full-page requests if we are not at EOF (bug 4410) */
+ if (to != CFS_PAGE_SIZE && llap->llap_page->index < size_index) {
+ LL_CDEBUG_PAGE(D_PAGE, llap->llap_page,
+ "sync write before EOF: size_index %lu, to %d\n",
+ size_index, to);
+ to = CFS_PAGE_SIZE;
+ } else if (to != CFS_PAGE_SIZE && llap->llap_page->index == size_index) {
+ int size_to = i_size_read(inode) & ~CFS_PAGE_MASK;
+ LL_CDEBUG_PAGE(D_PAGE, llap->llap_page,
+ "sync write at EOF: size_index %lu, to %d/%d\n",
+ size_index, to, size_to);
+ if (to < size_to)
+ to = size_to;
+ }
+
+ /* compare the checksum once before the page leaves llite */
+ if (unlikely((sbi->ll_flags & LL_SBI_CHECKSUM) &&
+ llap->llap_checksum != 0)) {
+ __u32 csum = 0;
+ struct page *page = llap->llap_page;
+ csum = crc32_le(csum, kmap(page), CFS_PAGE_SIZE);
+ kunmap(page);
+ if (llap->llap_checksum == csum) {
+ CDEBUG(D_PAGE, "page %p cksum %x confirmed\n",
+ page, csum);
+ } else {
+ CERROR("page %p old cksum %x != new cksum %x!\n",
+ page, llap->llap_checksum, csum);
+ }
+ }
+
+ rc = obd_queue_group_io(exp, ll_i2info(inode)->lli_smd, NULL, oig,
+ llap->llap_cookie, OBD_BRW_WRITE | noquot,
+ 0, to, 0, ASYNC_READY | ASYNC_URGENT |
+ ASYNC_COUNT_STABLE | ASYNC_GROUP_SYNC);
+ if (rc)
+ GOTO(free_oig, rc);
+
+ rc = obd_trigger_group_io(exp, ll_i2info(inode)->lli_smd, NULL, oig);
+ if (rc)
+ GOTO(free_oig, rc);
+
+ rc = oig_wait(oig);
+
+ if (!rc && async_flags & ASYNC_READY) {
+ unlock_page(llap->llap_page);
+ if (PageWriteback(llap->llap_page)) {
+ end_page_writeback(llap->llap_page);
+ }
+ }
+
+ if (rc == 0 && llap_write_complete(inode, llap))
+ ll_queue_done_writing(inode, 0);
- oa.o_id = ll_i2info(inode)->lli_smd->lsm_object_id;
- oa.o_valid = OBD_MD_FLID;
- obdo_from_inode(&oa, inode, OBD_MD_FLTYPE | OBD_MD_FLATIME |
- OBD_MD_FLMTIME | OBD_MD_FLCTIME);
+ LL_CDEBUG_PAGE(D_PAGE, llap->llap_page, "sync write returned %d\n", rc);
- RETURN(ll_batch_writepage(inode, &oa, page));
+free_oig:
+ oig_release(oig);
+out:
+ RETURN(rc);
}
-/*
- * we really don't want to start writeback here, we want to give callers some
- * time to further dirty the pages before we write them out.
- */
-static int ll_commit_write(struct file *file, struct page *page,
- unsigned from, unsigned to)
+/* update our write count to account for i_size increases that may have
+ * happened since we've queued the page for io. */
+
+/* be careful not to return success without setting the page Uptodate or
+ * the next pass through prepare_write will read in stale data from disk. */
+int ll_commit_write(struct file *file, struct page *page, unsigned from,
+ unsigned to)
{
struct inode *inode = page->mapping->host;
+ struct ll_inode_info *lli = ll_i2info(inode);
+ struct lov_stripe_md *lsm = lli->lli_smd;
+ struct obd_export *exp;
+ struct ll_async_page *llap;
loff_t size;
int rc = 0;
ENTRY;
CDEBUG(D_INODE, "inode %p is writing page %p from %d to %d at %lu\n",
inode, page, from, to, page->index);
+
+ llap = llap_from_page(page, LLAP_ORIGIN_COMMIT_WRITE);
+ if (IS_ERR(llap))
+ RETURN(PTR_ERR(llap));
+
+ exp = ll_i2dtexp(inode);
+ if (exp == NULL)
+ RETURN(-EINVAL);
+
+ llap->llap_ignore_quota = capable(CAP_SYS_RESOURCE);
+
+ /*
+ * queue a write for some time in the future the first time we
+ * dirty the page.
+ *
+ * This is different from what other file systems do: they usually
+ * just mark page (and some of its buffers) dirty and rely on
+ * balance_dirty_pages() to start a write-back. Lustre wants write-back
+ * to be started earlier for the following reasons:
+ *
+ * (1) with a large number of clients we need to limit the amount
+ * of cached data on the clients a lot;
+ *
+ * (2) large compute jobs generally want compute-only then io-only
+ * and the IO should complete as quickly as possible;
+ *
+ * (3) IO is batched up to the RPC size and is async until the
+ * client max cache is hit
+ * (/proc/fs/lustre/osc/OSC.../max_dirty_mb)
+ *
+ */
if (!PageDirty(page)) {
- lprocfs_counter_incr(ll_i2sbi(inode)->ll_stats,
- LPROC_LL_DIRTY_MISSES);
- rc = ll_mark_dirty_page(ll_i2obdconn(inode),
- ll_i2info(inode)->lli_smd,
- page->index);
- if (rc < 0 && rc != -EDQUOT)
- RETURN(rc); /* XXX lproc counter here? */
+ ll_stats_ops_tally(ll_i2sbi(inode), LPROC_LL_DIRTY_MISSES, 1);
+
+ rc = queue_or_sync_write(exp, inode, llap, to, 0);
+ if (rc)
+ GOTO(out, rc);
} else {
- lprocfs_counter_incr(ll_i2sbi(inode)->ll_stats,
- LPROC_LL_DIRTY_HITS);
+ ll_stats_ops_tally(ll_i2sbi(inode), LPROC_LL_DIRTY_HITS, 1);
}
- size = (((obd_off)page->index) << PAGE_SHIFT) + to;
- if (size > inode->i_size)
- inode->i_size = size;
+ /* put the page in the page cache, from now on ll_removepage is
+ * responsible for cleaning up the llap.
+ * only set page dirty when it's queued to be write out */
+ if (llap->llap_write_queued)
+ set_page_dirty(page);
+
+out:
+ size = (((obd_off)page->index) << CFS_PAGE_SHIFT) + to;
+ ll_inode_size_lock(inode, 0);
+ if (rc == 0) {
+ lov_stripe_lock(lsm);
+ obd_adjust_kms(exp, lsm, size, 0);
+ lov_stripe_unlock(lsm);
+ if (size > i_size_read(inode))
+ i_size_write(inode, size);
+ SetPageUptodate(page);
+ } else if (size > i_size_read(inode)) {
+ /* this page beyond the pales of i_size, so it can't be
+ * truncated in ll_p_r_e during lock revoking. we must
+ * teardown our book-keeping here. */
+ ll_removepage(page);
+ }
+ ll_inode_size_unlock(inode, 0);
+ RETURN(rc);
+}
- SetPageUptodate(page);
- set_page_dirty(page);
+static unsigned long ll_ra_count_get(struct ll_sb_info *sbi, unsigned long len)
+{
+ struct ll_ra_info *ra = &sbi->ll_ra_info;
+ unsigned long ret;
+ ENTRY;
- /* This means that we've hit either the local cache limit or the limit
- * of the OST's grant. */
- if (rc == -EDQUOT) {
- struct ll_file_data *fd = file->private_data;
- struct obdo oa;
- int rc;
+ spin_lock(&sbi->ll_lock);
+ ret = min(ra->ra_max_pages - ra->ra_cur_pages, len);
+ ra->ra_cur_pages += ret;
+ spin_unlock(&sbi->ll_lock);
- oa.o_id = ll_i2info(inode)->lli_smd->lsm_object_id;
- memcpy(obdo_handle(&oa), &fd->fd_ost_och.och_fh,
- sizeof(fd->fd_ost_och.och_fh));
- oa.o_valid = OBD_MD_FLID | OBD_MD_FLHANDLE;
- obdo_from_inode(&oa, inode, OBD_MD_FLTYPE | OBD_MD_FLATIME |
- OBD_MD_FLMTIME | OBD_MD_FLCTIME);
+ RETURN(ret);
+}
- rc = ll_batch_writepage(inode, &oa, page);
- lock_page(page); /* caller expects to unlock */
- RETURN(rc);
+static void ll_ra_count_put(struct ll_sb_info *sbi, unsigned long len)
+{
+ struct ll_ra_info *ra = &sbi->ll_ra_info;
+ spin_lock(&sbi->ll_lock);
+ LASSERTF(ra->ra_cur_pages >= len, "r_c_p %lu len %lu\n",
+ ra->ra_cur_pages, len);
+ ra->ra_cur_pages -= len;
+ spin_unlock(&sbi->ll_lock);
+}
+
+/* called for each page in a completed rpc.*/
+int ll_ap_completion(void *data, int cmd, struct obdo *oa, int rc)
+{
+ struct ll_async_page *llap;
+ struct page *page;
+ int ret = 0;
+ ENTRY;
+
+ llap = LLAP_FROM_COOKIE(data);
+ page = llap->llap_page;
+ LASSERT(PageLocked(page));
+ LASSERT(CheckWriteback(page,cmd));
+
+ LL_CDEBUG_PAGE(D_PAGE, page, "completing cmd %d with %d\n", cmd, rc);
+
+ if (cmd & OBD_BRW_READ && llap->llap_defer_uptodate)
+ ll_ra_count_put(ll_i2sbi(page->mapping->host), 1);
+
+ if (rc == 0) {
+ if (cmd & OBD_BRW_READ) {
+ if (!llap->llap_defer_uptodate)
+ SetPageUptodate(page);
+ } else {
+ llap->llap_write_queued = 0;
+ }
+ ClearPageError(page);
+ } else {
+ if (cmd & OBD_BRW_READ) {
+ llap->llap_defer_uptodate = 0;
+ }
+ SetPageError(page);
+ if (rc == -ENOSPC)
+ set_bit(AS_ENOSPC, &page->mapping->flags);
+ else
+ set_bit(AS_EIO, &page->mapping->flags);
}
- RETURN(0);
-} /* ll_commit_write */
-#else
-static int ll_writepage(struct page *page,
- struct writeback_control *wbc)
+ unlock_page(page);
+
+ if (cmd & OBD_BRW_WRITE) {
+ /* Only rc == 0, write succeed, then this page could be deleted
+ * from the pending_writing list
+ */
+ if (rc == 0 && llap_write_complete(page->mapping->host, llap))
+ ll_queue_done_writing(page->mapping->host, 0);
+ }
+
+ if (PageWriteback(page)) {
+ end_page_writeback(page);
+ }
+ page_cache_release(page);
+
+ RETURN(ret);
+}
+
+/* the kernel calls us here when a page is unhashed from the page cache.
+ * the page will be locked and the kernel is holding a spinlock, so
+ * we need to be careful. we're just tearing down our book-keeping
+ * here. */
+void ll_removepage(struct page *page)
{
+ struct inode *inode = page->mapping->host;
+ struct obd_export *exp;
+ struct ll_async_page *llap;
+ struct ll_sb_info *sbi = ll_i2sbi(inode);
+ int rc;
+ ENTRY;
+
+ LASSERT(!in_interrupt());
+
+ /* sync pages or failed read pages can leave pages in the page
+ * cache that don't have our data associated with them anymore */
+ if (page_private(page) == 0) {
+ EXIT;
+ return;
+ }
+
+ LL_CDEBUG_PAGE(D_PAGE, page, "being evicted\n");
+
+ exp = ll_i2dtexp(inode);
+ if (exp == NULL) {
+ CERROR("page %p ind %lu gave null export\n", page, page->index);
+ EXIT;
+ return;
+ }
+
+ llap = llap_from_page(page, LLAP_ORIGIN_REMOVEPAGE);
+ if (IS_ERR(llap)) {
+ CERROR("page %p ind %lu couldn't find llap: %ld\n", page,
+ page->index, PTR_ERR(llap));
+ EXIT;
+ return;
+ }
- return 0;
+ if (llap_write_complete(inode, llap))
+ ll_queue_done_writing(inode, 0);
+
+ rc = obd_teardown_async_page(exp, ll_i2info(inode)->lli_smd, NULL,
+ llap->llap_cookie);
+ if (rc != 0)
+ CERROR("page %p ind %lu failed: %d\n", page, page->index, rc);
+
+ /* this unconditional free is only safe because the page lock
+ * is providing exclusivity to memory pressure/truncate/writeback..*/
+ __clear_page_ll_data(page);
+
+ spin_lock(&sbi->ll_lock);
+ if (!list_empty(&llap->llap_pglist_item))
+ list_del_init(&llap->llap_pglist_item);
+ sbi->ll_pglist_gen++;
+ sbi->ll_async_page_count--;
+ spin_unlock(&sbi->ll_lock);
+ OBD_SLAB_FREE(llap, ll_async_page_slab, ll_async_page_slab_size);
+ EXIT;
}
-static int ll_commit_write(struct file *file, struct page *page,
- unsigned from, unsigned to)
+
+static int ll_page_matches(struct page *page, int fd_flags)
{
- return 0;
+ struct lustre_handle match_lockh = {0};
+ struct inode *inode = page->mapping->host;
+ ldlm_policy_data_t page_extent;
+ int flags, matches;
+ ENTRY;
+
+ if (unlikely(fd_flags & LL_FILE_GROUP_LOCKED))
+ RETURN(1);
+
+ page_extent.l_extent.start = (__u64)page->index << CFS_PAGE_SHIFT;
+ page_extent.l_extent.end =
+ page_extent.l_extent.start + CFS_PAGE_SIZE - 1;
+ flags = LDLM_FL_TEST_LOCK | LDLM_FL_BLOCK_GRANTED;
+ if (!(fd_flags & LL_FILE_READAHEAD))
+ flags |= LDLM_FL_CBPENDING;
+ matches = obd_match(ll_i2sbi(inode)->ll_dt_exp,
+ ll_i2info(inode)->lli_smd, LDLM_EXTENT,
+ &page_extent, LCK_PR | LCK_PW, &flags, inode,
+ &match_lockh);
+ RETURN(matches);
}
-#endif
-#if (LINUX_VERSION_CODE < KERNEL_VERSION(2,5,0))
-static int ll_direct_IO(int rw, struct inode *inode, struct kiobuf *iobuf,
- unsigned long blocknr, int blocksize)
+static int ll_issue_page_read(struct obd_export *exp,
+ struct ll_async_page *llap,
+ struct obd_io_group *oig, int defer)
{
- struct ll_inode_info *lli = ll_i2info(inode);
- struct lov_stripe_md *lsm = lli->lli_smd;
- struct brw_page *pga;
- struct ptlrpc_request_set *set;
- struct obdo oa;
- int length, i, flags, rc = 0;
- loff_t offset;
+ struct page *page = llap->llap_page;
+ int rc;
+
+ page_cache_get(page);
+ llap->llap_defer_uptodate = defer;
+ llap->llap_ra_used = 0;
+ rc = obd_queue_group_io(exp, ll_i2info(page->mapping->host)->lli_smd,
+ NULL, oig, llap->llap_cookie, OBD_BRW_READ, 0,
+ CFS_PAGE_SIZE, 0, ASYNC_COUNT_STABLE |
+ ASYNC_READY | ASYNC_URGENT);
+ if (rc) {
+ LL_CDEBUG_PAGE(D_ERROR, page, "read queue failed: rc %d\n", rc);
+ page_cache_release(page);
+ }
+ RETURN(rc);
+}
+
+static void ll_ra_stats_inc_unlocked(struct ll_ra_info *ra, enum ra_stat which)
+{
+ LASSERTF(which >= 0 && which < _NR_RA_STAT, "which: %u\n", which);
+ ra->ra_stats[which]++;
+}
+
+static void ll_ra_stats_inc(struct address_space *mapping, enum ra_stat which)
+{
+ struct ll_sb_info *sbi = ll_i2sbi(mapping->host);
+ struct ll_ra_info *ra = &ll_i2sbi(mapping->host)->ll_ra_info;
+
+ spin_lock(&sbi->ll_lock);
+ ll_ra_stats_inc_unlocked(ra, which);
+ spin_unlock(&sbi->ll_lock);
+}
+
+void ll_ra_accounting(struct ll_async_page *llap, struct address_space *mapping)
+{
+ if (!llap->llap_defer_uptodate || llap->llap_ra_used)
+ return;
+
+ ll_ra_stats_inc(mapping, RA_STAT_DISCARDED);
+}
+
+#define RAS_CDEBUG(ras) \
+ CDEBUG(D_READA, \
+ "lrp %lu cr %lu cp %lu ws %lu wl %lu nra %lu r %lu ri %lu\n", \
+ ras->ras_last_readpage, ras->ras_consecutive_requests, \
+ ras->ras_consecutive_pages, ras->ras_window_start, \
+ ras->ras_window_len, ras->ras_next_readahead, \
+ ras->ras_requests, ras->ras_request_index);
+
+static int index_in_window(unsigned long index, unsigned long point,
+ unsigned long before, unsigned long after)
+{
+ unsigned long start = point - before, end = point + after;
+
+ if (start > point)
+ start = 0;
+ if (end < point)
+ end = ~0;
+
+ return start <= index && index <= end;
+}
+
+static struct ll_readahead_state *ll_ras_get(struct file *f)
+{
+ struct ll_file_data *fd;
+
+ fd = LUSTRE_FPRIVATE(f);
+ return &fd->fd_ras;
+}
+
+void ll_ra_read_in(struct file *f, struct ll_ra_read *rar)
+{
+ struct ll_readahead_state *ras;
+
+ ras = ll_ras_get(f);
+
+ spin_lock(&ras->ras_lock);
+ ras->ras_requests++;
+ ras->ras_request_index = 0;
+ ras->ras_consecutive_requests++;
+ rar->lrr_reader = current;
+
+ list_add(&rar->lrr_linkage, &ras->ras_read_beads);
+ spin_unlock(&ras->ras_lock);
+}
+
+void ll_ra_read_ex(struct file *f, struct ll_ra_read *rar)
+{
+ struct ll_readahead_state *ras;
+
+ ras = ll_ras_get(f);
+
+ spin_lock(&ras->ras_lock);
+ list_del_init(&rar->lrr_linkage);
+ spin_unlock(&ras->ras_lock);
+}
+
+static struct ll_ra_read *ll_ra_read_get_locked(struct ll_readahead_state *ras)
+{
+ struct ll_ra_read *scan;
+
+ list_for_each_entry(scan, &ras->ras_read_beads, lrr_linkage) {
+ if (scan->lrr_reader == current)
+ return scan;
+ }
+ return NULL;
+}
+
+struct ll_ra_read *ll_ra_read_get(struct file *f)
+{
+ struct ll_readahead_state *ras;
+ struct ll_ra_read *bead;
+
+ ras = ll_ras_get(f);
+
+ spin_lock(&ras->ras_lock);
+ bead = ll_ra_read_get_locked(ras);
+ spin_unlock(&ras->ras_lock);
+ return bead;
+}
+
+static int ll_readahead(struct ll_readahead_state *ras,
+ struct obd_export *exp, struct address_space *mapping,
+ struct obd_io_group *oig, int flags)
+{
+ unsigned long i, start = 0, end = 0, reserved;
+ struct ll_async_page *llap;
+ struct page *page;
+ int rc, ret = 0, match_failed = 0;
+ __u64 kms;
+ unsigned int gfp_mask;
+ struct inode *inode;
+ struct lov_stripe_md *lsm;
+ struct ll_ra_read *bead;
+ struct ost_lvb lvb;
ENTRY;
- if (!lsm || !lsm->lsm_object_id)
- RETURN(-EBADF);
+ inode = mapping->host;
+ lsm = ll_i2info(inode)->lli_smd;
+
+ lov_stripe_lock(lsm);
+ inode_init_lvb(inode, &lvb);
+ obd_merge_lvb(ll_i2dtexp(inode), lsm, &lvb, 1);
+ kms = lvb.lvb_size;
+ lov_stripe_unlock(lsm);
+ if (kms == 0) {
+ ll_ra_stats_inc(mapping, RA_STAT_ZERO_LEN);
+ RETURN(0);
+ }
- /* FIXME: io smaller than PAGE_SIZE is broken on ia64 */
- if ((iobuf->offset & (PAGE_SIZE - 1)) ||
- (iobuf->length & (PAGE_SIZE - 1)))
- RETURN(-EINVAL);
+ spin_lock(&ras->ras_lock);
+ bead = ll_ra_read_get_locked(ras);
+ /* Enlarge the RA window to encompass the full read */
+ if (bead != NULL && ras->ras_window_start + ras->ras_window_len <
+ bead->lrr_start + bead->lrr_count) {
+ ras->ras_window_len = bead->lrr_start + bead->lrr_count -
+ ras->ras_window_start;
+ }
+ /* Reserve a part of the read-ahead window that we'll be issuing */
+ if (ras->ras_window_len) {
+ start = ras->ras_next_readahead;
+ end = ras->ras_window_start + ras->ras_window_len - 1;
+ }
+ if (end != 0) {
+ /* Truncate RA window to end of file */
+ end = min(end, (unsigned long)((kms - 1) >> CFS_PAGE_SHIFT));
+ ras->ras_next_readahead = max(end, end + 1);
+ RAS_CDEBUG(ras);
+ }
+ spin_unlock(&ras->ras_lock);
- set = ptlrpc_prep_set();
- if (set == NULL)
- RETURN(-ENOMEM);
-
- OBD_ALLOC(pga, sizeof(*pga) * iobuf->nr_pages);
- if (!pga) {
- ptlrpc_set_destroy(set);
- RETURN(-ENOMEM);
- }
-
- flags = (rw == WRITE ? OBD_BRW_CREATE : 0) /* | OBD_BRW_DIRECTIO */;
- offset = ((obd_off)blocknr << inode->i_blkbits);
- length = iobuf->length;
-
- for (i = 0, length = iobuf->length; length > 0;
- length -= pga[i].count, offset += pga[i].count, i++) { /*i last!*/
- pga[i].pg = iobuf->maplist[i];
- pga[i].off = offset;
- /* To the end of the page, or the length, whatever is less */
- pga[i].count = min_t(int, PAGE_SIZE - (offset & ~PAGE_MASK),
- length);
- pga[i].flag = flags;
- if (rw == READ) {
- //POISON(kmap(iobuf->maplist[i]), 0xc5, PAGE_SIZE);
- //kunmap(iobuf->maplist[i]);
+ if (end == 0) {
+ ll_ra_stats_inc(mapping, RA_STAT_ZERO_WINDOW);
+ RETURN(0);
+ }
+
+ reserved = ll_ra_count_get(ll_i2sbi(inode), end - start + 1);
+ if (reserved < end - start + 1)
+ ll_ra_stats_inc(mapping, RA_STAT_MAX_IN_FLIGHT);
+
+ gfp_mask = GFP_HIGHUSER & ~__GFP_WAIT;
+#ifdef __GFP_NOWARN
+ gfp_mask |= __GFP_NOWARN;
+#endif
+
+ for (i = start; reserved > 0 && !match_failed && i <= end; i++) {
+ /* skip locked pages from previous readpage calls */
+ page = grab_cache_page_nowait_gfp(mapping, i, gfp_mask);
+ if (page == NULL) {
+ ll_ra_stats_inc(mapping, RA_STAT_FAILED_GRAB_PAGE);
+ CDEBUG(D_READA, "g_c_p_n failed\n");
+ continue;
}
+
+ /* Check if page was truncated or reclaimed */
+ if (page->mapping != mapping) {
+ ll_ra_stats_inc(mapping, RA_STAT_WRONG_GRAB_PAGE);
+ CDEBUG(D_READA, "g_c_p_n returned invalid page\n");
+ goto next_page;
+ }
+
+ /* we do this first so that we can see the page in the /proc
+ * accounting */
+ llap = llap_from_page(page, LLAP_ORIGIN_READAHEAD);
+ if (IS_ERR(llap) || llap->llap_defer_uptodate)
+ goto next_page;
+
+ /* skip completed pages */
+ if (Page_Uptodate(page))
+ goto next_page;
+
+ /* bail when we hit the end of the lock. */
+ if ((rc = ll_page_matches(page, flags|LL_FILE_READAHEAD)) <= 0){
+ LL_CDEBUG_PAGE(D_READA | D_PAGE, page,
+ "lock match failed: rc %d\n", rc);
+ ll_ra_stats_inc(mapping, RA_STAT_FAILED_MATCH);
+ match_failed = 1;
+ goto next_page;
+ }
+
+ rc = ll_issue_page_read(exp, llap, oig, 1);
+ if (rc == 0) {
+ reserved--;
+ ret++;
+ LL_CDEBUG_PAGE(D_READA| D_PAGE, page,
+ "started read-ahead\n");
+ } else {
+ next_page:
+ LL_CDEBUG_PAGE(D_READA | D_PAGE, page,
+ "skipping read-ahead\n");
+
+ unlock_page(page);
+ }
+ page_cache_release(page);
}
- oa.o_id = lsm->lsm_object_id;
- oa.o_valid = OBD_MD_FLID;
- obdo_from_inode(&oa, inode, OBD_MD_FLTYPE | OBD_MD_FLATIME |
- OBD_MD_FLMTIME | OBD_MD_FLCTIME);
+ LASSERTF(reserved >= 0, "reserved %lu\n", reserved);
+ if (reserved != 0)
+ ll_ra_count_put(ll_i2sbi(inode), reserved);
+ if (i == end + 1 && end == (kms >> CFS_PAGE_SHIFT))
+ ll_ra_stats_inc(mapping, RA_STAT_EOF);
+
+ /* if we didn't get to the end of the region we reserved from
+ * the ras we need to go back and update the ras so that the
+ * next read-ahead tries from where we left off. we only do so
+ * if the region we failed to issue read-ahead on is still ahead
+ * of the app and behind the next index to start read-ahead from */
+ if (i != end + 1) {
+ spin_lock(&ras->ras_lock);
+ if (i < ras->ras_next_readahead &&
+ index_in_window(i, ras->ras_window_start, 0,
+ ras->ras_window_len)) {
+ ras->ras_next_readahead = i;
+ RAS_CDEBUG(ras);
+ }
+ spin_unlock(&ras->ras_lock);
+ }
- if (rw == WRITE)
- lprocfs_counter_add(ll_i2sbi(inode)->ll_stats,
- LPROC_LL_DIRECT_WRITE, iobuf->length);
- else
- lprocfs_counter_add(ll_i2sbi(inode)->ll_stats,
- LPROC_LL_DIRECT_READ, iobuf->length);
- rc = obd_brw_async(rw == WRITE ? OBD_BRW_WRITE : OBD_BRW_READ,
- ll_i2obdconn(inode), &oa, lsm, iobuf->nr_pages, pga,
- set, NULL);
- if (rc) {
- CDEBUG(rc == -ENOSPC ? D_INODE : D_ERROR,
- "error from obd_brw_async: rc = %d\n", rc);
- } else {
- rc = ptlrpc_set_wait(set);
- if (rc)
- CERROR("error from callback: rc = %d\n", rc);
+ RETURN(ret);
+}
+
+static void ras_set_start(struct ll_readahead_state *ras, unsigned long index)
+{
+ ras->ras_window_start = index & (~((1024 * 1024 >> CFS_PAGE_SHIFT) - 1));
+}
+
+/* called with the ras_lock held or from places where it doesn't matter */
+static void ras_reset(struct ll_readahead_state *ras, unsigned long index)
+{
+ ras->ras_last_readpage = index;
+ ras->ras_consecutive_requests = 0;
+ ras->ras_consecutive_pages = 0;
+ ras->ras_window_len = 0;
+ ras_set_start(ras, index);
+ ras->ras_next_readahead = max(ras->ras_window_start, index);
+
+ RAS_CDEBUG(ras);
+}
+
+void ll_readahead_init(struct inode *inode, struct ll_readahead_state *ras)
+{
+ spin_lock_init(&ras->ras_lock);
+ ras_reset(ras, 0);
+ ras->ras_requests = 0;
+ INIT_LIST_HEAD(&ras->ras_read_beads);
+}
+
+static void ras_update(struct ll_sb_info *sbi, struct inode *inode,
+ struct ll_readahead_state *ras, unsigned long index,
+ unsigned hit)
+{
+ struct ll_ra_info *ra = &sbi->ll_ra_info;
+ int zero = 0;
+ ENTRY;
+
+ spin_lock(&sbi->ll_lock);
+ spin_lock(&ras->ras_lock);
+
+ ll_ra_stats_inc_unlocked(ra, hit ? RA_STAT_HIT : RA_STAT_MISS);
+
+ /* reset the read-ahead window in two cases. First when the app seeks
+ * or reads to some other part of the file. Secondly if we get a
+ * read-ahead miss that we think we've previously issued. This can
+ * be a symptom of there being so many read-ahead pages that the VM is
+ * reclaiming it before we get to it. */
+ if (!index_in_window(index, ras->ras_last_readpage, 8, 8)) {
+ zero = 1;
+ ll_ra_stats_inc_unlocked(ra, RA_STAT_DISTANT_READPAGE);
+ } else if (!hit && ras->ras_window_len &&
+ index < ras->ras_next_readahead &&
+ index_in_window(index, ras->ras_window_start, 0,
+ ras->ras_window_len)) {
+ zero = 1;
+ ll_ra_stats_inc_unlocked(ra, RA_STAT_MISS_IN_WINDOW);
+ }
+
+ /* On the second access to a file smaller than the tunable
+ * ra_max_read_ahead_whole_pages trigger RA on all pages in the
+ * file up to ra_max_pages. This is simply a best effort and
+ * only occurs once per open file. Normal RA behavior is reverted
+ * to for subsequent IO. The mmap case does not increment
+ * ras_requests and thus can never trigger this behavior. */
+ if (ras->ras_requests == 2 && !ras->ras_request_index) {
+ __u64 kms_pages;
+
+ kms_pages = (i_size_read(inode) + CFS_PAGE_SIZE - 1) >>
+ CFS_PAGE_SHIFT;
+
+ CDEBUG(D_READA, "kmsp "LPU64" mwp %lu mp %lu\n", kms_pages,
+ ra->ra_max_read_ahead_whole_pages, ra->ra_max_pages);
+
+ if (kms_pages &&
+ kms_pages <= ra->ra_max_read_ahead_whole_pages) {
+ ras->ras_window_start = 0;
+ ras->ras_last_readpage = 0;
+ ras->ras_next_readahead = 0;
+ ras->ras_window_len = min(ra->ra_max_pages,
+ ra->ra_max_read_ahead_whole_pages);
+ GOTO(out_unlock, 0);
+ }
+ }
+
+ if (zero) {
+ ras_reset(ras, index);
+ GOTO(out_unlock, 0);
}
- ptlrpc_set_destroy(set);
- if (rc == 0)
- rc = iobuf->length;
- OBD_FREE(pga, sizeof(*pga) * iobuf->nr_pages);
+ ras->ras_last_readpage = index;
+ ras->ras_consecutive_pages++;
+ ras_set_start(ras, index);
+ ras->ras_next_readahead = max(ras->ras_window_start,
+ ras->ras_next_readahead);
+
+ /* Trigger RA in the mmap case where ras_consecutive_requests
+ * is not incremented and thus can't be used to trigger RA */
+ if (!ras->ras_window_len && ras->ras_consecutive_pages == 3) {
+ ras->ras_window_len = 1024 * 1024 >> CFS_PAGE_SHIFT;
+ GOTO(out_unlock, 0);
+ }
+
+ /* The initial ras_window_len is set to the request size. To avoid
+ * uselessly reading and discarding pages for random IO the window is
+ * only increased once per consecutive request received. */
+ if (ras->ras_consecutive_requests > 1 && !ras->ras_request_index) {
+ ras->ras_window_len = min(ras->ras_window_len +
+ (1024 * 1024 >> CFS_PAGE_SHIFT),
+ ra->ra_max_pages);
+ }
+
+ EXIT;
+out_unlock:
+ RAS_CDEBUG(ras);
+ ras->ras_request_index++;
+ spin_unlock(&ras->ras_lock);
+ spin_unlock(&sbi->ll_lock);
+ return;
+}
+
+int ll_writepage(struct page *page)
+{
+ struct inode *inode = page->mapping->host;
+ struct ll_inode_info *lli = ll_i2info(inode);
+ struct obd_export *exp;
+ struct ll_async_page *llap;
+ int rc = 0;
+ ENTRY;
+
+ LASSERT(!PageDirty(page));
+ LASSERT(PageLocked(page));
+
+ exp = ll_i2dtexp(inode);
+ if (exp == NULL)
+ GOTO(out, rc = -EINVAL);
+
+ llap = llap_from_page(page, LLAP_ORIGIN_WRITEPAGE);
+ if (IS_ERR(llap))
+ GOTO(out, rc = PTR_ERR(llap));
+
+ LASSERT(!PageWriteback(page));
+ set_page_writeback(page);
+
+ page_cache_get(page);
+ if (llap->llap_write_queued) {
+ LL_CDEBUG_PAGE(D_PAGE, page, "marking urgent\n");
+ rc = obd_set_async_flags(exp, lli->lli_smd, NULL,
+ llap->llap_cookie,
+ ASYNC_READY | ASYNC_URGENT);
+ } else {
+ rc = queue_or_sync_write(exp, inode, llap, CFS_PAGE_SIZE,
+ ASYNC_READY | ASYNC_URGENT);
+ }
+ if (rc)
+ page_cache_release(page);
+out:
+ if (rc) {
+ if (!lli->lli_async_rc)
+ lli->lli_async_rc = rc;
+ /* re-dirty page on error so it retries write */
+ if (PageWriteback(page)) {
+ end_page_writeback(page);
+ }
+ /* resend page only for not started IO*/
+ if (!PageError(page))
+ ll_redirty_page(page);
+ unlock_page(page);
+ }
RETURN(rc);
}
-#endif
-//#endif
+/*
+ * for now we do our readpage the same on both 2.4 and 2.5. The kernel's
+ * read-ahead assumes it is valid to issue readpage all the way up to
+ * i_size, but our dlm locks make that not the case. We disable the
+ * kernel's read-ahead and do our own by walking ahead in the page cache
+ * checking for dlm lock coverage. the main difference between 2.4 and
+ * 2.6 is how read-ahead gets batched and issued, but we're using our own,
+ * so they look the same.
+ */
+int ll_readpage(struct file *filp, struct page *page)
+{
+ struct ll_file_data *fd = LUSTRE_FPRIVATE(filp);
+ struct inode *inode = page->mapping->host;
+ struct obd_export *exp;
+ struct ll_async_page *llap;
+ struct obd_io_group *oig = NULL;
+ int rc;
+ ENTRY;
+
+ LASSERT(PageLocked(page));
+ LASSERT(!PageUptodate(page));
+ CDEBUG(D_VFSTRACE, "VFS Op:inode=%lu/%u(%p),offset=%Lu=%#Lx\n",
+ inode->i_ino, inode->i_generation, inode,
+ (((loff_t)page->index) << CFS_PAGE_SHIFT),
+ (((loff_t)page->index) << CFS_PAGE_SHIFT));
+ LASSERT(atomic_read(&filp->f_dentry->d_inode->i_count) > 0);
+
+ if (!ll_i2info(inode)->lli_smd) {
+ /* File with no objects - one big hole */
+ /* We use this just for remove_from_page_cache that is not
+ * exported, we'd make page back up to date. */
+ ll_truncate_complete_page(page);
+ clear_page(kmap(page));
+ kunmap(page);
+ SetPageUptodate(page);
+ unlock_page(page);
+ RETURN(0);
+ }
-struct address_space_operations ll_aops = {
- readpage: ll_readpage,
-#if (LINUX_VERSION_CODE <= KERNEL_VERSION(2,5,0))
- direct_IO: ll_direct_IO,
-#endif
- writepage: ll_writepage,
- sync_page: block_sync_page,
- prepare_write: ll_prepare_write,
- commit_write: ll_commit_write,
- bmap: NULL
-//#endif
-};
+ rc = oig_init(&oig);
+ if (rc < 0)
+ GOTO(out, rc);
+
+ exp = ll_i2dtexp(inode);
+ if (exp == NULL)
+ GOTO(out, rc = -EINVAL);
+
+ llap = llap_from_page(page, LLAP_ORIGIN_READPAGE);
+ if (IS_ERR(llap))
+ GOTO(out, rc = PTR_ERR(llap));
+
+ if (ll_i2sbi(inode)->ll_ra_info.ra_max_pages)
+ ras_update(ll_i2sbi(inode), inode, &fd->fd_ras, page->index,
+ llap->llap_defer_uptodate);
+
+ if (llap->llap_defer_uptodate) {
+ llap->llap_ra_used = 1;
+ rc = ll_readahead(&fd->fd_ras, exp, page->mapping, oig,
+ fd->fd_flags);
+ if (rc > 0)
+ obd_trigger_group_io(exp, ll_i2info(inode)->lli_smd,
+ NULL, oig);
+ LL_CDEBUG_PAGE(D_PAGE, page, "marking uptodate from defer\n");
+ SetPageUptodate(page);
+ unlock_page(page);
+ GOTO(out_oig, rc = 0);
+ }
+
+ if (likely((fd->fd_flags & LL_FILE_IGNORE_LOCK) == 0)) {
+ rc = ll_page_matches(page, fd->fd_flags);
+ if (rc < 0) {
+ LL_CDEBUG_PAGE(D_ERROR, page, "lock match failed: rc %d\n", rc);
+ GOTO(out, rc);
+ }
+
+ if (rc == 0) {
+ CWARN("ino %lu page %lu (%llu) not covered by "
+ "a lock (mmap?). check debug logs.\n",
+ inode->i_ino, page->index,
+ (long long)page->index << CFS_PAGE_SHIFT);
+ }
+ }
+
+ rc = ll_issue_page_read(exp, llap, oig, 0);
+ if (rc)
+ GOTO(out, rc);
+
+ LL_CDEBUG_PAGE(D_PAGE, page, "queued readpage\n");
+ if (ll_i2sbi(inode)->ll_ra_info.ra_max_pages)
+ ll_readahead(&fd->fd_ras, exp, page->mapping, oig,
+ fd->fd_flags);
+
+ rc = obd_trigger_group_io(exp, ll_i2info(inode)->lli_smd, NULL, oig);
+
+out:
+ if (rc)
+ unlock_page(page);
+out_oig:
+ if (oig != NULL)
+ oig_release(oig);
+ RETURN(rc);
+}
git://git.whamcloud.com / fs / lustre-release.git / blobdiff