Branch HEAD

[fs/lustre-release.git] / lustre / obdfilter / filter_io.c

diff --git a/lustre/obdfilter/filter_io.c b/lustre/obdfilter/filter_io.c
index 7e75c80..ea2721e 100644 (file)
--- a/lustre/obdfilter/filter_io.c
+++ b/lustre/obdfilter/filter_io.c
@@ -264,20 +264,26 @@ long filter_grant(struct obd_export *exp, obd_size current_grant,
 /*
  * 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_NOFS for requests from a local client not allowing to enter FS
+ * as we might end up 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
+ * use more agressive GFP_HIGHUSER flags from non-local clients to be able to
+ * generate more memory pressure, but at the same time use __GFP_NOMEMALLOC
+ * in order not to exhaust emergency reserves.
+ *
+ * See Bug 19529 and Bug 19917 for details. 
  */
 static struct page *filter_get_page(struct obd_device *obd,
                                     struct inode *inode,
-                                    obd_off offset)
+                                    obd_off offset,
+                                    int localreq)
 {
         struct page *page;
 
         page = find_or_create_page(inode->i_mapping, offset >> CFS_PAGE_SHIFT,
-                                   GFP_NOFS | __GFP_HIGHMEM);
+                                   (localreq ? (GFP_NOFS | __GFP_HIGHMEM) 
+                                             : (GFP_HIGHUSER | __GFP_NOMEMALLOC)));
         if (unlikely(page == NULL))
                 lprocfs_counter_add(obd->obd_stats, LPROC_FILTER_NO_PAGE, 1);
 
@@ -359,6 +365,34 @@ void filter_invalidate_cache(struct obd_device *obd, struct obd_ioobj *obj,
         }
 }
 
+/*
+ * the invalidate above doesn't work during read because lnet pins pages.
+ * The truncate is used here instead to drop pages from cache
+ */
+void filter_truncate_cache(struct obd_device *obd, struct obd_ioobj *obj,
+                           struct niobuf_remote *nb, int pages,
+                           struct niobuf_local *res, struct inode *inode)
+{
+        struct niobuf_remote *rnb;
+        int i;
+
+        LASSERT(inode != NULL);
+#ifdef HAVE_TRUNCATE_RANGE
+        for (i = 0, rnb = nb; i < obj->ioo_bufcnt; i++, rnb++) {
+                /* remove pages in which range is fit */
+                truncate_inode_pages_range(inode->i_mapping,
+                                           rnb->offset & CFS_PAGE_MASK,
+                                           (rnb->offset + rnb->len - 1) |
+                                           ~CFS_PAGE_MASK);
+        }
+#elif (defined HAVE_TRUNCATE_COMPLETE)
+        for (i = 0, lnb = res; i < pages; i++, lnb++)
+                truncate_complete_page(inode->i_mapping, lnb->page);
+#else
+#error "Nor truncate_inode_pages_range or truncate_complete_page are supported"
+#endif
+}
+
 static int filter_preprw_read(int cmd, struct obd_export *exp, struct obdo *oa,
                               int objcount, struct obd_ioobj *obj,
                               struct niobuf_remote *nb,
@@ -431,7 +465,7 @@ static int filter_preprw_read(int cmd, struct obd_export *exp, struct obdo *oa,
                          * so it's easy to detect later. */
                         break;
 
-                lnb->page = filter_get_page(obd, inode, lnb->offset);
+                lnb->page = filter_get_page(obd, inode, lnb->offset, 0);
                 if (lnb->page == NULL)
                         GOTO(cleanup, rc = -ENOMEM);
 
@@ -645,7 +679,7 @@ static int filter_preprw_write(int cmd, struct obd_export *exp, struct obdo *oa,
         void *iobuf;
         obd_size left;
         unsigned long now = jiffies, timediff;
-        int rc = 0, i, tot_bytes = 0, cleanup_phase = 0;
+        int rc = 0, i, tot_bytes = 0, cleanup_phase = 0, localreq = 0;
         ENTRY;
         LASSERT(objcount == 1);
         LASSERT(obj->ioo_bufcnt > 0);
@@ -655,6 +689,9 @@ static int filter_preprw_write(int cmd, struct obd_export *exp, struct obdo *oa,
         if (rc)
                 RETURN(rc);
 
+        if (exp->exp_connection->c_peer.nid == exp->exp_connection->c_self) 
+                localreq = 1;
+
         push_ctxt(&saved, &obd->obd_lvfs_ctxt, NULL);
         iobuf = filter_iobuf_get(&obd->u.filter, oti);
         if (IS_ERR(iobuf))
@@ -740,7 +777,8 @@ static int filter_preprw_write(int cmd, struct obd_export *exp, struct obdo *oa,
                  * needs to keep the pages all aligned properly. */
                 lnb->dentry = dentry;
 
-                lnb->page = filter_get_page(obd, dentry->d_inode, lnb->offset);
+                lnb->page = filter_get_page(obd, dentry->d_inode, lnb->offset,
+                                            localreq);
                 if (lnb->page == NULL)
                         GOTO(cleanup, rc = -ENOMEM);
 
@@ -891,16 +929,17 @@ static int filter_commitrw_read(struct obd_export *exp, struct obdo *oa,
         if (res->dentry != NULL)
                 inode = res->dentry->d_inode;
 
-        for (i = 0, lnb = res; i < npages; i++, lnb++) {
-                if (lnb->page != NULL) {
+        for (i = 0, lnb = res; i < npages; i++, lnb++)
+                if (lnb->page != NULL)
                         page_cache_release(lnb->page);
-                        lnb->page = NULL;
-                }
-        }
 
         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);
+                      i_size_read(inode) > fo->fo_readcache_max_filesize))
+                filter_truncate_cache(exp->exp_obd, obj, rnb, npages, res,
+                                      inode);
+
+        for (i = 0, lnb = res; i < npages; i++, lnb++)
+                lnb->page = NULL;
 
         if (res->dentry != NULL)
                 f_dput(res->dentry);