Whamcloud - gitweb
LU-2139 osc: Use SOFT_SYNC to urge server commit
[fs/lustre-release.git] / lustre / osc / osc_page.c
index e72e6d2..affd0d2 100644 (file)
@@ -1,6 +1,4 @@
-/* -*- mode: c; c-basic-offset: 8; indent-tabs-mode: nil; -*-
- * vim:expandtab:shiftwidth=8:tabstop=8:
- *
+/*
  * GPL HEADER START
  *
  * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.
  * GPL HEADER END
  */
 /*
- * Copyright 2008 Sun Microsystems, Inc.  All rights reserved.
+ * Copyright (c) 2008, 2010, Oracle and/or its affiliates. All rights reserved.
  * Use is subject to license terms.
+ *
+ * Copyright (c) 2011, 2013, Intel Corporation.
  */
 /*
  * This file is part of Lustre, http://www.lustre.org/
  *   Author: Nikita Danilov <nikita.danilov@sun.com>
  */
 
-/** \addtogroup osc osc @{ */
-
 #define DEBUG_SUBSYSTEM S_OSC
 
 #include "osc_cl_internal.h"
 
+static void osc_lru_del(struct client_obd *cli, struct osc_page *opg, bool del);
+static void osc_lru_add(struct client_obd *cli, struct osc_page *opg);
+static int osc_lru_reserve(const struct lu_env *env, struct osc_object *obj,
+                          struct osc_page *opg);
+
+/** \addtogroup osc
+ *  @{
+ */
+
+/*
+ * Comment out osc_page_protected because it may sleep inside the
+ * the client_obd_list_lock.
+ * client_obd_list_lock -> osc_ap_completion -> osc_completion ->
+ *   -> osc_page_protected -> osc_page_is_dlocked -> osc_match_base
+ *   -> ldlm_lock_match -> sptlrpc_import_check_ctx -> sleep.
+ */
+#if 0
 static int osc_page_is_dlocked(const struct lu_env *env,
                                const struct osc_page *opg,
                                enum cl_lock_mode mode, int pending, int unref)
 {
-        struct cl_page         *page;
-        struct osc_object      *obj;
-        struct osc_thread_info *info;
-        struct ldlm_res_id     *resname;
-        struct lustre_handle   *lockh;
-        ldlm_policy_data_t     *policy;
-        ldlm_mode_t             dlmmode;
-        int                     flags;
-
-        info = osc_env_info(env);
-        resname = &info->oti_resname;
-        policy = &info->oti_policy;
-        lockh = &info->oti_handle;
-        page = opg->ops_cl.cpl_page;
-        obj = cl2osc(opg->ops_cl.cpl_obj);
+       struct cl_page         *page;
+       struct osc_object      *obj;
+       struct osc_thread_info *info;
+       struct ldlm_res_id     *resname;
+       struct lustre_handle   *lockh;
+       ldlm_policy_data_t     *policy;
+       ldlm_mode_t             dlmmode;
+       __u64                   flags;
+
+       might_sleep();
+
+       info = osc_env_info(env);
+       resname = &info->oti_resname;
+       policy = &info->oti_policy;
+       lockh = &info->oti_handle;
+       page = opg->ops_cl.cpl_page;
+       obj = cl2osc(opg->ops_cl.cpl_obj);
 
         flags = LDLM_FL_TEST_LOCK | LDLM_FL_BLOCK_GRANTED;
         if (pending)
@@ -75,6 +92,10 @@ static int osc_page_is_dlocked(const struct lu_env *env,
                               dlmmode, &flags, NULL, lockh, unref);
 }
 
+/**
+ * Checks an invariant that a page in the cache is covered by a lock, as
+ * needed.
+ */
 static int osc_page_protected(const struct lu_env *env,
                               const struct osc_page *opg,
                               enum cl_lock_mode mode, int unref)
@@ -87,17 +108,26 @@ static int osc_page_protected(const struct lu_env *env,
 
         LINVRNT(!opg->ops_temp);
 
+        page = opg->ops_cl.cpl_page;
+        if (page->cp_owner != NULL &&
+            cl_io_top(page->cp_owner)->ci_lockreq == CILR_NEVER)
+                /*
+                 * If IO is done without locks (liblustre, or lloop), lock is
+                 * not required.
+                 */
+                result = 1;
+        else
+                /* otherwise check for a DLM lock */
         result = osc_page_is_dlocked(env, opg, mode, 1, unref);
         if (result == 0) {
                 /* maybe this page is a part of a lockless io? */
                 hdr = cl_object_header(opg->ops_cl.cpl_obj);
-                page = opg->ops_cl.cpl_page;
                 descr = &osc_env_info(env)->oti_descr;
                 descr->cld_mode = mode;
                 descr->cld_start = page->cp_index;
                 descr->cld_end   = page->cp_index;
-                spin_lock(&hdr->coh_lock_guard);
-                list_for_each_entry(scan, &hdr->coh_locks, cll_linkage) {
+               spin_lock(&hdr->coh_lock_guard);
+                cfs_list_for_each_entry(scan, &hdr->coh_locks, cll_linkage) {
                         /*
                          * Lock-less sub-lock has to be either in HELD state
                          * (when io is actively going on), or in CACHED state,
@@ -114,10 +144,18 @@ static int osc_page_protected(const struct lu_env *env,
                                 break;
                         }
                 }
-                spin_unlock(&hdr->coh_lock_guard);
+               spin_unlock(&hdr->coh_lock_guard);
         }
         return result;
 }
+#else
+static int osc_page_protected(const struct lu_env *env,
+                              const struct osc_page *opg,
+                              enum cl_lock_mode mode, int unref)
+{
+        return 1;
+}
+#endif
 
 /*****************************************************************************
  *
@@ -129,7 +167,7 @@ static void osc_page_fini(const struct lu_env *env,
 {
         struct osc_page *opg = cl2osc_page(slice);
         CDEBUG(D_TRACE, "%p\n", opg);
-        OBD_SLAB_FREE_PTR(opg, osc_page_kmem);
+        LASSERT(opg->ops_lock == NULL);
 }
 
 static void osc_page_transfer_get(struct osc_page *opg, const char *label)
@@ -162,48 +200,47 @@ static void osc_page_transfer_put(const struct lu_env *env,
 static void osc_page_transfer_add(const struct lu_env *env,
                                   struct osc_page *opg, enum cl_req_type crt)
 {
-        struct osc_object *obj;
+       struct osc_object *obj = cl2osc(opg->ops_cl.cpl_obj);
 
-        LINVRNT(cl_page_is_vmlocked(env, opg->ops_cl.cpl_page));
+       /* ops_lru and ops_inflight share the same field, so take it from LRU
+        * first and then use it as inflight. */
+       osc_lru_del(osc_cli(obj), opg, false);
 
-        obj = cl2osc(opg->ops_cl.cpl_obj);
-        spin_lock(&obj->oo_seatbelt);
-        list_add(&opg->ops_inflight, &obj->oo_inflight[crt]);
-        opg->ops_submitter = cfs_current();
-        spin_unlock(&obj->oo_seatbelt);
+       spin_lock(&obj->oo_seatbelt);
+       cfs_list_add(&opg->ops_inflight, &obj->oo_inflight[crt]);
+       opg->ops_submitter = current;
+       spin_unlock(&obj->oo_seatbelt);
 }
 
 static int osc_page_cache_add(const struct lu_env *env,
-                              const struct cl_page_slice *slice,
-                              struct cl_io *_)
+                             const struct cl_page_slice *slice,
+                             struct cl_io *io)
 {
-        struct osc_page   *opg = cl2osc_page(slice);
-        struct osc_object *obj = cl2osc(opg->ops_cl.cpl_obj);
-        struct osc_io     *oio = osc_env_io(env);
-        int result;
-        int brw_flags;
-        int noquota = 0;
-
-        LINVRNT(osc_page_protected(env, opg, CLM_WRITE, 0));
-        ENTRY;
-
-        /* Set the OBD_BRW_SRVLOCK before the page is queued. */
-        brw_flags = oio->oi_lockless ? OBD_BRW_SRVLOCK : 0;
-        if (!client_is_remote(osc_export(obj)) &&
-            cfs_capable(CFS_CAP_SYS_RESOURCE)) {
-                brw_flags |= OBD_BRW_NOQUOTA;
-                noquota = OBD_BRW_NOQUOTA;
-        }
-
-        osc_page_transfer_get(opg, "transfer\0cache");
-        result = osc_queue_async_io(env, osc_export(obj), NULL, obj->oo_oinfo,
-                                    &opg->ops_oap, OBD_BRW_WRITE | noquota,
-                                    0, 0, brw_flags, 0);
-        if (result != 0)
-                osc_page_transfer_put(env, opg);
-        else
-                osc_page_transfer_add(env, opg, CRT_WRITE);
-        RETURN(result);
+       struct osc_io   *oio = osc_env_io(env);
+       struct osc_page *opg = cl2osc_page(slice);
+       int result;
+       ENTRY;
+
+       LINVRNT(osc_page_protected(env, opg, CLM_WRITE, 0));
+
+       osc_page_transfer_get(opg, "transfer\0cache");
+       result = osc_queue_async_io(env, io, opg);
+       if (result != 0)
+               osc_page_transfer_put(env, opg);
+       else
+               osc_page_transfer_add(env, opg, CRT_WRITE);
+
+       /* for sync write, kernel will wait for this page to be flushed before
+        * osc_io_end() is called, so release it earlier.
+        * for mkwrite(), it's known there is no further pages. */
+       if (cl_io_is_sync_write(io) || cl_io_is_mkwrite(io)) {
+               if (oio->oi_active != NULL) {
+                       osc_extent_release(env, oio->oi_active);
+                       oio->oi_active = NULL;
+               }
+       }
+
+       RETURN(result);
 }
 
 void osc_index2policy(ldlm_policy_data_t *policy, const struct cl_object *obj,
@@ -214,26 +251,95 @@ void osc_index2policy(ldlm_policy_data_t *policy, const struct cl_object *obj,
         policy->l_extent.end   = cl_offset(obj, end + 1) - 1;
 }
 
+static int osc_page_addref_lock(const struct lu_env *env,
+                                struct osc_page *opg,
+                                struct cl_lock *lock)
+{
+        struct osc_lock *olock;
+        int              rc;
+
+        LASSERT(opg->ops_lock == NULL);
+
+        olock = osc_lock_at(lock);
+        if (cfs_atomic_inc_return(&olock->ols_pageref) <= 0) {
+                cfs_atomic_dec(&olock->ols_pageref);
+                rc = -ENODATA;
+        } else {
+               cl_lock_get(lock);
+                opg->ops_lock = lock;
+                rc = 0;
+        }
+        return rc;
+}
+
+static void osc_page_putref_lock(const struct lu_env *env,
+                                 struct osc_page *opg)
+{
+        struct cl_lock  *lock = opg->ops_lock;
+        struct osc_lock *olock;
+
+        LASSERT(lock != NULL);
+        olock = osc_lock_at(lock);
+
+        cfs_atomic_dec(&olock->ols_pageref);
+        opg->ops_lock = NULL;
+
+        cl_lock_put(env, lock);
+}
+
 static int osc_page_is_under_lock(const struct lu_env *env,
                                   const struct cl_page_slice *slice,
-                                  struct cl_io *_)
+                                  struct cl_io *unused)
 {
         struct cl_lock *lock;
-        int             result;
+        int             result = -ENODATA;
 
         ENTRY;
         lock = cl_lock_at_page(env, slice->cpl_obj, slice->cpl_page,
                                NULL, 1, 0);
         if (lock != NULL) {
-                cl_lock_put(env, lock);
-                result = -EBUSY;
-        } else
-                result = -ENODATA;
+               if (osc_page_addref_lock(env, cl2osc_page(slice), lock) == 0)
+                       result = -EBUSY;
+               cl_lock_put(env, lock);
+       }
         RETURN(result);
 }
 
+static void osc_page_disown(const struct lu_env *env,
+                            const struct cl_page_slice *slice,
+                            struct cl_io *io)
+{
+        struct osc_page *opg = cl2osc_page(slice);
+
+        if (unlikely(opg->ops_lock))
+                osc_page_putref_lock(env, opg);
+}
+
+static void osc_page_completion_read(const struct lu_env *env,
+                                     const struct cl_page_slice *slice,
+                                     int ioret)
+{
+       struct osc_page   *opg = cl2osc_page(slice);
+       struct osc_object *obj = cl2osc(opg->ops_cl.cpl_obj);
+
+       if (likely(opg->ops_lock))
+               osc_page_putref_lock(env, opg);
+       osc_lru_add(osc_cli(obj), opg);
+}
+
+static void osc_page_completion_write(const struct lu_env *env,
+                                     const struct cl_page_slice *slice,
+                                     int ioret)
+{
+       struct osc_page   *opg = cl2osc_page(slice);
+       struct osc_object *obj = cl2osc(slice->cpl_obj);
+
+       osc_lru_add(osc_cli(obj), opg);
+}
+
 static int osc_page_fail(const struct lu_env *env,
-                         const struct cl_page_slice *slice, struct cl_io *_)
+                         const struct cl_page_slice *slice,
+                         struct cl_io *unused)
 {
         /*
          * Cached read?
@@ -243,9 +349,17 @@ static int osc_page_fail(const struct lu_env *env,
 }
 
 
-static const char *osc_list(struct list_head *head)
+static const char *osc_list(cfs_list_t *head)
 {
-        return list_empty(head) ? "-" : "+";
+        return cfs_list_empty(head) ? "-" : "+";
+}
+
+static inline cfs_time_t osc_submit_duration(struct osc_page *opg)
+{
+        if (opg->ops_submit_time == 0)
+                return 0;
+
+        return (cfs_time_current() - opg->ops_submit_time);
 }
 
 static int osc_page_print(const struct lu_env *env,
@@ -254,26 +368,55 @@ static int osc_page_print(const struct lu_env *env,
 {
         struct osc_page       *opg = cl2osc_page(slice);
         struct osc_async_page *oap = &opg->ops_oap;
+        struct osc_object     *obj = cl2osc(slice->cpl_obj);
+        struct client_obd     *cli = &osc_export(obj)->exp_obd->u.cli;
 
         return (*printer)(env, cookie, LUSTRE_OSC_NAME"-page@%p: "
-                          "%#x %d %u %s %s %s %llu %u %#x %p %p %p %p %p\n",
-                          opg, oap->oap_magic, oap->oap_cmd,
+                         "1< %#x %d %u %s %s > "
+                         "2< "LPU64" %u %u %#x %#x | %p %p %p > "
+                         "3< %s %p %d %lu %d > "
+                         "4< %d %d %d %lu %s | %s %s %s %s > "
+                         "5< %s %s %s %s | %d %s | %d %s %s>\n",
+                          opg,
+                          /* 1 */
+                          oap->oap_magic, oap->oap_cmd,
                           oap->oap_interrupted,
                           osc_list(&oap->oap_pending_item),
-                          osc_list(&oap->oap_urgent_item),
                           osc_list(&oap->oap_rpc_item),
-                          oap->oap_obj_off, oap->oap_page_off,
-                          oap->oap_async_flags, oap->oap_request,
-                          oap->oap_cli, oap->oap_loi, oap->oap_caller_ops,
-                          oap->oap_caller_data);
+                          /* 2 */
+                          oap->oap_obj_off, oap->oap_page_off, oap->oap_count,
+                          oap->oap_async_flags, oap->oap_brw_flags,
+                         oap->oap_request, oap->oap_cli, obj,
+                          /* 3 */
+                          osc_list(&opg->ops_inflight),
+                          opg->ops_submitter, opg->ops_transfer_pinned,
+                          osc_submit_duration(opg), opg->ops_srvlock,
+                          /* 4 */
+                          cli->cl_r_in_flight, cli->cl_w_in_flight,
+                          cli->cl_max_rpcs_in_flight,
+                          cli->cl_avail_grant,
+                          osc_list(&cli->cl_cache_waiters),
+                          osc_list(&cli->cl_loi_ready_list),
+                          osc_list(&cli->cl_loi_hp_ready_list),
+                          osc_list(&cli->cl_loi_write_list),
+                          osc_list(&cli->cl_loi_read_list),
+                          /* 5 */
+                         osc_list(&obj->oo_ready_item),
+                         osc_list(&obj->oo_hp_ready_item),
+                         osc_list(&obj->oo_write_item),
+                         osc_list(&obj->oo_read_item),
+                         cfs_atomic_read(&obj->oo_nr_reads),
+                         osc_list(&obj->oo_reading_exts),
+                         cfs_atomic_read(&obj->oo_nr_writes),
+                         osc_list(&obj->oo_hp_exts),
+                         osc_list(&obj->oo_urgent_exts));
 }
 
 static void osc_page_delete(const struct lu_env *env,
                             const struct cl_page_slice *slice)
 {
-        struct osc_page       *opg = cl2osc_page(slice);
-        struct osc_object     *obj = cl2osc(opg->ops_cl.cpl_obj);
-        struct osc_async_page *oap = &opg->ops_oap;
+       struct osc_page   *opg = cl2osc_page(slice);
+       struct osc_object *obj = cl2osc(opg->ops_cl.cpl_obj);
         int rc;
 
         LINVRNT(opg->ops_temp || osc_page_protected(env, opg, CLM_READ, 1));
@@ -281,12 +424,23 @@ static void osc_page_delete(const struct lu_env *env,
         ENTRY;
         CDEBUG(D_TRACE, "%p\n", opg);
         osc_page_transfer_put(env, opg);
-        rc = osc_teardown_async_page(osc_export(obj), NULL, obj->oo_oinfo, oap);
-        LASSERTF(rc == 0, "%i\n", rc);
-        spin_lock(&obj->oo_seatbelt);
-        list_del_init(&opg->ops_inflight);
-        spin_unlock(&obj->oo_seatbelt);
-        EXIT;
+       rc = osc_teardown_async_page(env, obj, opg);
+        if (rc) {
+                CL_PAGE_DEBUG(D_ERROR, env, cl_page_top(slice->cpl_page),
+                              "Trying to teardown failed: %d\n", rc);
+                LASSERT(0);
+        }
+
+       spin_lock(&obj->oo_seatbelt);
+       if (opg->ops_submitter != NULL) {
+               LASSERT(!cfs_list_empty(&opg->ops_inflight));
+               cfs_list_del_init(&opg->ops_inflight);
+               opg->ops_submitter = NULL;
+       }
+       spin_unlock(&obj->oo_seatbelt);
+
+       osc_lru_del(osc_cli(obj), opg, true);
+       EXIT;
 }
 
 void osc_page_clip(const struct lu_env *env, const struct cl_page_slice *slice,
@@ -299,241 +453,505 @@ void osc_page_clip(const struct lu_env *env, const struct cl_page_slice *slice,
 
         opg->ops_from = from;
         opg->ops_to   = to;
-        oap->oap_async_flags |= ASYNC_COUNT_STABLE;
+       spin_lock(&oap->oap_lock);
+       oap->oap_async_flags |= ASYNC_COUNT_STABLE;
+       spin_unlock(&oap->oap_lock);
 }
 
 static int osc_page_cancel(const struct lu_env *env,
                            const struct cl_page_slice *slice)
 {
-        struct osc_page *opg       = cl2osc_page(slice);
-        struct osc_async_page *oap = &opg->ops_oap;
+       struct osc_page *opg = cl2osc_page(slice);
         int rc = 0;
 
         LINVRNT(osc_page_protected(env, opg, CLM_READ, 0));
 
-        client_obd_list_lock(&oap->oap_cli->cl_loi_list_lock);
         /* Check if the transferring against this page
          * is completed, or not even queued. */
         if (opg->ops_transfer_pinned)
                 /* FIXME: may not be interrupted.. */
-                rc = osc_oap_interrupted(env, oap);
+               rc = osc_cancel_async_page(env, opg);
         LASSERT(ergo(rc == 0, opg->ops_transfer_pinned == 0));
-        client_obd_list_unlock(&oap->oap_cli->cl_loi_list_lock);
         return rc;
 }
 
+static int osc_page_flush(const struct lu_env *env,
+                         const struct cl_page_slice *slice,
+                         struct cl_io *io)
+{
+       struct osc_page *opg = cl2osc_page(slice);
+       int rc = 0;
+       ENTRY;
+       rc = osc_flush_async_page(env, io, opg);
+       RETURN(rc);
+}
+
 static const struct cl_page_operations osc_page_ops = {
         .cpo_fini          = osc_page_fini,
         .cpo_print         = osc_page_print,
         .cpo_delete        = osc_page_delete,
         .cpo_is_under_lock = osc_page_is_under_lock,
+        .cpo_disown        = osc_page_disown,
         .io = {
                 [CRT_READ] = {
-                        .cpo_cache_add = osc_page_fail
+                        .cpo_cache_add  = osc_page_fail,
+                        .cpo_completion = osc_page_completion_read
                 },
                 [CRT_WRITE] = {
-                        .cpo_cache_add = osc_page_cache_add
-                }
-        },
-        .cpo_clip           = osc_page_clip,
-        .cpo_cancel         = osc_page_cancel
+                       .cpo_cache_add  = osc_page_cache_add,
+                       .cpo_completion = osc_page_completion_write
+               }
+       },
+       .cpo_clip           = osc_page_clip,
+       .cpo_cancel         = osc_page_cancel,
+       .cpo_flush          = osc_page_flush
 };
 
-static int osc_make_ready(const struct lu_env *env, void *data, int cmd)
+int osc_page_init(const struct lu_env *env, struct cl_object *obj,
+               struct cl_page *page, struct page *vmpage)
 {
-        struct osc_page *opg  = data;
-        struct cl_page  *page = cl_page_top(opg->ops_cl.cpl_page);
-        int result;
+       struct osc_object *osc = cl2osc(obj);
+       struct osc_page   *opg = cl_object_page_slice(obj, page);
+       int result;
+
+       opg->ops_from = 0;
+       opg->ops_to   = PAGE_CACHE_SIZE;
+
+       result = osc_prep_async_page(osc, opg, vmpage,
+                                       cl_offset(obj, page->cp_index));
+       if (result == 0) {
+               struct osc_io *oio = osc_env_io(env);
+               opg->ops_srvlock = osc_io_srvlock(oio);
+               cl_page_slice_add(page, &opg->ops_cl, obj,
+                               &osc_page_ops);
+       }
+       /*
+        * Cannot assert osc_page_protected() here as read-ahead
+        * creates temporary pages outside of a lock.
+        */
+#ifdef CONFIG_LUSTRE_DEBUG_EXPENSIVE_CHECK
+       opg->ops_temp = !osc_page_protected(env, opg, CLM_READ, 1);
+#endif
+       /* ops_inflight and ops_lru are the same field, but it doesn't
+        * hurt to initialize it twice :-) */
+       CFS_INIT_LIST_HEAD(&opg->ops_inflight);
+       CFS_INIT_LIST_HEAD(&opg->ops_lru);
 
-        LASSERT(cmd == OBD_BRW_WRITE); /* no cached reads */
-        LINVRNT(osc_page_protected(env, opg, CLM_WRITE, 1));
+       /* reserve an LRU space for this page */
+       if (page->cp_type == CPT_CACHEABLE && result == 0)
+               result = osc_lru_reserve(env, osc, opg);
 
-        ENTRY;
-        result = cl_page_make_ready(env, page, CRT_WRITE);
-        RETURN(result);
+       return result;
 }
 
-static int osc_refresh_count(const struct lu_env *env, void *data, int cmd)
+int osc_over_unstable_soft_limit(struct client_obd *cli)
 {
-        struct cl_page   *page;
-        struct osc_page  *osc = data;
-        struct cl_object *obj;
-        struct cl_attr   *attr = &osc_env_info(env)->oti_attr;
+       long obd_upages, obd_dpages, osc_upages;
+
+       /* Can't check cli->cl_unstable_count, therefore, no soft limit */
+       if (cli == NULL)
+               return 0;
+
+       obd_upages = cfs_atomic_read(&obd_unstable_pages);
+       obd_dpages = cfs_atomic_read(&obd_dirty_pages);
+
+       osc_upages = cfs_atomic_read(&cli->cl_unstable_count);
+
+       /* obd_max_dirty_pages is the max number of (dirty + unstable)
+        * pages allowed at any given time. To simulate an unstable page
+        * only limit, we subtract the current number of dirty pages
+        * from this max. This difference is roughly the amount of pages
+        * currently available for unstable pages. Thus, the soft limit
+        * is half of that difference. Check osc_upages to ensure we don't
+        * set SOFT_SYNC for OSCs without any outstanding unstable pages. */
+       return osc_upages != 0 &&
+              obd_upages >= (obd_max_dirty_pages - obd_dpages) / 2;
+}
 
-        int result;
-        loff_t kms;
-
-        LINVRNT(osc_page_protected(env, osc, CLM_READ, 1));
-
-        /* readpage queues with _COUNT_STABLE, shouldn't get here. */
-        LASSERT(!(cmd & OBD_BRW_READ));
-        LASSERT(osc != NULL);
-        page = osc->ops_cl.cpl_page;
-        obj = osc->ops_cl.cpl_obj;
-
-        cl_object_attr_lock(obj);
-        result = cl_object_attr_get(env, obj, attr);
-        cl_object_attr_unlock(obj);
-        if (result < 0)
-                return result;
-        kms = attr->cat_kms;
-        if (cl_offset(obj, page->cp_index) >= kms)
-                /* catch race with truncate */
-                return 0;
-        else if (cl_offset(obj, page->cp_index + 1) > kms)
-                /* catch sub-page write at end of file */
-                return kms % CFS_PAGE_SIZE;
-        else
-                return CFS_PAGE_SIZE;
+/**
+ * Helper function called by osc_io_submit() for every page in an immediate
+ * transfer (i.e., transferred synchronously).
+ */
+void osc_page_submit(const struct lu_env *env, struct osc_page *opg,
+                    enum cl_req_type crt, int brw_flags)
+{
+       struct osc_async_page *oap = &opg->ops_oap;
+       struct osc_object     *obj = oap->oap_obj;
+
+       LINVRNT(osc_page_protected(env, opg,
+                                  crt == CRT_WRITE ? CLM_WRITE : CLM_READ, 1));
+
+       LASSERTF(oap->oap_magic == OAP_MAGIC, "Bad oap magic: oap %p, "
+                "magic 0x%x\n", oap, oap->oap_magic);
+       LASSERT(oap->oap_async_flags & ASYNC_READY);
+       LASSERT(oap->oap_async_flags & ASYNC_COUNT_STABLE);
+
+       oap->oap_cmd       = crt == CRT_WRITE ? OBD_BRW_WRITE : OBD_BRW_READ;
+       oap->oap_page_off  = opg->ops_from;
+       oap->oap_count     = opg->ops_to - opg->ops_from;
+       oap->oap_brw_flags = OBD_BRW_SYNC | brw_flags;
+
+       if (osc_over_unstable_soft_limit(oap->oap_cli))
+               oap->oap_brw_flags |= OBD_BRW_SOFT_SYNC;
+
+       if (!client_is_remote(osc_export(obj)) &&
+                       cfs_capable(CFS_CAP_SYS_RESOURCE)) {
+               oap->oap_brw_flags |= OBD_BRW_NOQUOTA;
+               oap->oap_cmd |= OBD_BRW_NOQUOTA;
+       }
+
+       opg->ops_submit_time = cfs_time_current();
+       osc_page_transfer_get(opg, "transfer\0imm");
+       osc_page_transfer_add(env, opg, crt);
 }
 
-static int osc_completion(const struct lu_env *env,
-                          void *data, int cmd, struct obdo *oa, int rc)
+/* --------------- LRU page management ------------------ */
+
+/* OSC is a natural place to manage LRU pages as applications are specialized
+ * to write OSC by OSC. Ideally, if one OSC is used more frequently it should
+ * occupy more LRU slots. On the other hand, we should avoid using up all LRU
+ * slots (client_obd::cl_lru_left) otherwise process has to be put into sleep
+ * for free LRU slots - this will be very bad so the algorithm requires each
+ * OSC to free slots voluntarily to maintain a reasonable number of free slots
+ * at any time.
+ */
+
+static CFS_DECL_WAITQ(osc_lru_waitq);
+static cfs_atomic_t osc_lru_waiters = CFS_ATOMIC_INIT(0);
+/* LRU pages are freed in batch mode. OSC should at least free this
+ * number of pages to avoid running out of LRU budget, and.. */
+static const int lru_shrink_min = 2 << (20 - PAGE_CACHE_SHIFT);  /* 2M */
+/* free this number at most otherwise it will take too long time to finsih. */
+static const int lru_shrink_max = 32 << (20 - PAGE_CACHE_SHIFT); /* 32M */
+
+/* Check if we can free LRU slots from this OSC. If there exists LRU waiters,
+ * we should free slots aggressively. In this way, slots are freed in a steady
+ * step to maintain fairness among OSCs.
+ *
+ * Return how many LRU pages should be freed. */
+static int osc_cache_too_much(struct client_obd *cli)
 {
-        struct osc_page       *opg  = data;
-        struct osc_async_page *oap  = &opg->ops_oap;
-        struct cl_page        *page = cl_page_top(opg->ops_cl.cpl_page);
-        struct osc_object     *obj  = cl2osc(opg->ops_cl.cpl_obj);
-        enum cl_req_type crt;
+       struct cl_client_cache *cache = cli->cl_cache;
+       int pages = cfs_atomic_read(&cli->cl_lru_in_list) >> 1;
 
-        LINVRNT(osc_page_protected(env, opg, CLM_READ, 1));
-        LINVRNT(cl_page_is_vmlocked(env, page));
+       if (cfs_atomic_read(&osc_lru_waiters) > 0 &&
+           cfs_atomic_read(cli->cl_lru_left) < lru_shrink_max)
+               /* drop lru pages aggressively */
+               return min(pages, lru_shrink_max);
 
-        ENTRY;
+       /* if it's going to run out LRU slots, we should free some, but not
+        * too much to maintain faireness among OSCs. */
+       if (cfs_atomic_read(cli->cl_lru_left) < cache->ccc_lru_max >> 4) {
+               unsigned long tmp;
 
-        cmd &= ~OBD_BRW_NOQUOTA;
-        LASSERT(equi(page->cp_state == CPS_PAGEIN,  cmd == OBD_BRW_READ));
-        LASSERT(equi(page->cp_state == CPS_PAGEOUT, cmd == OBD_BRW_WRITE));
-        LASSERT(opg->ops_transfer_pinned);
+               tmp = cache->ccc_lru_max / cfs_atomic_read(&cache->ccc_users);
+               if (pages > tmp)
+                       return min(pages, lru_shrink_max);
 
-        /*
-         * page->cp_req can be NULL if io submission failed before
-         * cl_req was allocated.
-         */
-        if (page->cp_req != NULL)
-                cl_req_page_done(env, page);
-        LASSERT(page->cp_req == NULL);
-
-        /* As the transfer for this page is being done, clear the flags */
-        oap->oap_async_flags = 0;
-
-        crt = cmd == OBD_BRW_READ ? CRT_READ : CRT_WRITE;
-        /* Clear opg->ops_transfer_pinned before VM lock is released. */
-        opg->ops_transfer_pinned = 0;
-
-        spin_lock(&obj->oo_seatbelt);
-        LASSERT(opg->ops_submitter != NULL);
-        LASSERT(!list_empty(&opg->ops_inflight));
-        list_del_init(&opg->ops_inflight);
-        spin_unlock(&obj->oo_seatbelt);
-
-        cl_page_completion(env, page, crt, rc);
-
-        /* statistic */
-        if (rc == 0 && oap->oap_brw_flags & OBD_BRW_SRVLOCK) {
-                struct lu_device *ld    = opg->ops_cl.cpl_obj->co_lu.lo_dev;
-                struct osc_stats *stats = &lu2osc_dev(ld)->od_stats;
-                int bytes = opg->ops_to - opg->ops_from;
-
-                if (crt == CRT_READ)
-                        stats->os_lockless_reads += bytes;
-                else
-                        stats->os_lockless_writes += bytes;
-        }
+               return pages > lru_shrink_min ? lru_shrink_min : 0;
+       }
 
-        /*
-         * This has to be the last operation with the page, as locks are
-         * released in cl_page_completion() and nothing except for the
-         * reference counter protects page from concurrent reclaim.
-         */
-        lu_ref_del(&page->cp_reference, "transfer", page);
-        /*
-         * As page->cp_obj is pinned by a reference from page->cp_req, it is
-         * safe to call cl_page_put() without risking object destruction in a
-         * non-blocking context.
-         */
-        cl_page_put(env, page);
-        RETURN(0);
+       return 0;
 }
 
-const static struct obd_async_page_ops osc_async_page_ops = {
-        .ap_make_ready    = osc_make_ready,
-        .ap_refresh_count = osc_refresh_count,
-        .ap_completion    = osc_completion
-};
-
-struct cl_page *osc_page_init(const struct lu_env *env,
-                              struct cl_object *obj,
-                              struct cl_page *page, cfs_page_t *vmpage)
+/* Return how many pages are not discarded in @pvec. */
+static int discard_pagevec(const struct lu_env *env, struct cl_io *io,
+                          struct cl_page **pvec, int max_index)
 {
-        struct osc_object *osc = cl2osc(obj);
-        struct osc_page   *opg;
-        int result;
-
-        OBD_SLAB_ALLOC_PTR(opg, osc_page_kmem);
-        if (opg != NULL) {
-                void *oap = &opg->ops_oap;
-
-                opg->ops_from = 0;
-                opg->ops_to   = CFS_PAGE_SIZE;
-
-                result = osc_prep_async_page(osc_export(osc),
-                                             NULL, osc->oo_oinfo, vmpage,
-                                             cl_offset(obj, page->cp_index),
-                                             &osc_async_page_ops,
-                                             opg, (void **)&oap, 1, NULL);
-                if (result == 0)
-                        cl_page_slice_add(page, &opg->ops_cl, obj,
-                                          &osc_page_ops);
-                /*
-                 * Cannot assert osc_page_protected() here as read-ahead
-                 * creates temporary pages outside of a lock.
-                 */
-#ifdef INVARIANT_CHECK
-                opg->ops_temp = !osc_page_protected(env, opg, CLM_READ, 1);
-#endif
-                CFS_INIT_LIST_HEAD(&opg->ops_inflight);
-        } else
-                result = -ENOMEM;
-        return ERR_PTR(result);
+       int count;
+       int i;
+
+       for (count = 0, i = 0; i < max_index; i++) {
+               struct cl_page *page = pvec[i];
+               if (cl_page_own_try(env, io, page) == 0) {
+                       /* free LRU page only if nobody is using it.
+                        * This check is necessary to avoid freeing the pages
+                        * having already been removed from LRU and pinned
+                        * for IO. */
+                       if (!cl_page_in_use(page)) {
+                               cl_page_unmap(env, io, page);
+                               cl_page_discard(env, io, page);
+                               ++count;
+                       }
+                       cl_page_disown(env, io, page);
+               }
+               cl_page_put(env, page);
+               pvec[i] = NULL;
+       }
+       return max_index - count;
 }
 
 /**
- * Helper function called by osc_io_submit() for every page in an immediate
- * transfer (i.e., transferred synchronously).
+ * Drop @target of pages from LRU at most.
  */
-void osc_io_submit_page(const struct lu_env *env,
-                        struct osc_io *oio, struct osc_page *opg,
-                        enum cl_req_type crt)
+int osc_lru_shrink(struct client_obd *cli, int target)
 {
-        struct osc_async_page *oap = &opg->ops_oap;
-        struct client_obd     *cli = oap->oap_cli;
-
-        LINVRNT(osc_page_protected(env, opg,
-                                   crt == CRT_WRITE ? CLM_WRITE : CLM_READ, 1));
-
-        oap->oap_page_off   = opg->ops_from;
-        oap->oap_count      = opg->ops_to - opg->ops_from;
-        oap->oap_brw_flags |= OBD_BRW_SYNC;
-        if (oio->oi_lockless)
-                oap->oap_brw_flags |= OBD_BRW_SRVLOCK;
-
-        oap->oap_cmd = crt == CRT_WRITE ? OBD_BRW_WRITE : OBD_BRW_READ;
-        if (!client_is_remote(osc_export(cl2osc(opg->ops_cl.cpl_obj))) &&
-            cfs_capable(CFS_CAP_SYS_RESOURCE)) {
-                oap->oap_brw_flags |= OBD_BRW_NOQUOTA;
-                oap->oap_cmd |= OBD_BRW_NOQUOTA;
-        }
+       struct cl_env_nest nest;
+       struct lu_env *env;
+       struct cl_io *io;
+       struct cl_object *clobj = NULL;
+       struct cl_page **pvec;
+       struct osc_page *opg;
+       int maxscan = 0;
+       int count = 0;
+       int index = 0;
+       int rc = 0;
+       ENTRY;
+
+       LASSERT(cfs_atomic_read(&cli->cl_lru_in_list) >= 0);
+       if (cfs_atomic_read(&cli->cl_lru_in_list) == 0 || target <= 0)
+               RETURN(0);
+
+       env = cl_env_nested_get(&nest);
+       if (IS_ERR(env))
+               RETURN(PTR_ERR(env));
+
+       pvec = osc_env_info(env)->oti_pvec;
+       io = &osc_env_info(env)->oti_io;
+
+       client_obd_list_lock(&cli->cl_lru_list_lock);
+       cfs_atomic_inc(&cli->cl_lru_shrinkers);
+       maxscan = min(target << 1, cfs_atomic_read(&cli->cl_lru_in_list));
+       while (!cfs_list_empty(&cli->cl_lru_list)) {
+               struct cl_page *page;
+
+               if (--maxscan < 0)
+                       break;
+
+               opg = cfs_list_entry(cli->cl_lru_list.next, struct osc_page,
+                                    ops_lru);
+               page = cl_page_top(opg->ops_cl.cpl_page);
+               if (cl_page_in_use_noref(page)) {
+                       cfs_list_move_tail(&opg->ops_lru, &cli->cl_lru_list);
+                       continue;
+               }
+
+               LASSERT(page->cp_obj != NULL);
+               if (clobj != page->cp_obj) {
+                       struct cl_object *tmp = page->cp_obj;
+
+                       cl_object_get(tmp);
+                       client_obd_list_unlock(&cli->cl_lru_list_lock);
+
+                       if (clobj != NULL) {
+                               count -= discard_pagevec(env, io, pvec, index);
+                               index = 0;
+
+                               cl_io_fini(env, io);
+                               cl_object_put(env, clobj);
+                               clobj = NULL;
+                       }
+
+                       clobj = tmp;
+                       io->ci_obj = clobj;
+                       io->ci_ignore_layout = 1;
+                       rc = cl_io_init(env, io, CIT_MISC, clobj);
+
+                       client_obd_list_lock(&cli->cl_lru_list_lock);
+
+                       if (rc != 0)
+                               break;
+
+                       ++maxscan;
+                       continue;
+               }
+
+               /* move this page to the end of list as it will be discarded
+                * soon. The page will be finally removed from LRU list in
+                * osc_page_delete().  */
+               cfs_list_move_tail(&opg->ops_lru, &cli->cl_lru_list);
+
+               /* it's okay to grab a refcount here w/o holding lock because
+                * it has to grab cl_lru_list_lock to delete the page. */
+               cl_page_get(page);
+               pvec[index++] = page;
+               if (++count >= target)
+                       break;
+
+               if (unlikely(index == OTI_PVEC_SIZE)) {
+                       client_obd_list_unlock(&cli->cl_lru_list_lock);
+                       count -= discard_pagevec(env, io, pvec, index);
+                       index = 0;
+
+                       client_obd_list_lock(&cli->cl_lru_list_lock);
+               }
+       }
+       client_obd_list_unlock(&cli->cl_lru_list_lock);
+
+       if (clobj != NULL) {
+               count -= discard_pagevec(env, io, pvec, index);
+
+               cl_io_fini(env, io);
+               cl_object_put(env, clobj);
+       }
+       cl_env_nested_put(&nest, env);
+
+       cfs_atomic_dec(&cli->cl_lru_shrinkers);
+       RETURN(count > 0 ? count : rc);
+}
+
+static void osc_lru_add(struct client_obd *cli, struct osc_page *opg)
+{
+       bool wakeup = false;
+
+       if (!opg->ops_in_lru)
+               return;
+
+       cfs_atomic_dec(&cli->cl_lru_busy);
+       client_obd_list_lock(&cli->cl_lru_list_lock);
+       if (cfs_list_empty(&opg->ops_lru)) {
+               cfs_list_move_tail(&opg->ops_lru, &cli->cl_lru_list);
+               cfs_atomic_inc_return(&cli->cl_lru_in_list);
+               wakeup = cfs_atomic_read(&osc_lru_waiters) > 0;
+       }
+       client_obd_list_unlock(&cli->cl_lru_list_lock);
+
+       if (wakeup) {
+               osc_lru_shrink(cli, osc_cache_too_much(cli));
+               wake_up_all(&osc_lru_waitq);
+       }
+}
+
+/* delete page from LRUlist. The page can be deleted from LRUlist for two
+ * reasons: redirtied or deleted from page cache. */
+static void osc_lru_del(struct client_obd *cli, struct osc_page *opg, bool del)
+{
+       if (opg->ops_in_lru) {
+               client_obd_list_lock(&cli->cl_lru_list_lock);
+               if (!cfs_list_empty(&opg->ops_lru)) {
+                       LASSERT(cfs_atomic_read(&cli->cl_lru_in_list) > 0);
+                       cfs_list_del_init(&opg->ops_lru);
+                       cfs_atomic_dec(&cli->cl_lru_in_list);
+                       if (!del)
+                               cfs_atomic_inc(&cli->cl_lru_busy);
+               } else if (del) {
+                       LASSERT(cfs_atomic_read(&cli->cl_lru_busy) > 0);
+                       cfs_atomic_dec(&cli->cl_lru_busy);
+               }
+               client_obd_list_unlock(&cli->cl_lru_list_lock);
+               if (del) {
+                       cfs_atomic_inc(cli->cl_lru_left);
+                       /* this is a great place to release more LRU pages if
+                        * this osc occupies too many LRU pages and kernel is
+                        * stealing one of them.
+                        * cl_lru_shrinkers is to avoid recursive call in case
+                        * we're already in the context of osc_lru_shrink(). */
+                       if (cfs_atomic_read(&cli->cl_lru_shrinkers) == 0 &&
+                           !memory_pressure_get())
+                               osc_lru_shrink(cli, osc_cache_too_much(cli));
+                       wake_up(&osc_lru_waitq);
+               }
+       } else {
+               LASSERT(cfs_list_empty(&opg->ops_lru));
+       }
+}
 
-        oap->oap_async_flags |= OSC_FLAGS;
-        if (oap->oap_cmd & OBD_BRW_READ)
-                oap->oap_async_flags |= ASYNC_COUNT_STABLE;
-        else if (!(oap->oap_brw_page.flag & OBD_BRW_FROM_GRANT))
-                osc_enter_cache_try(env, cli, oap->oap_loi, oap, 1);
+static inline int max_to_shrink(struct client_obd *cli)
+{
+       return min(cfs_atomic_read(&cli->cl_lru_in_list) >> 1, lru_shrink_max);
+}
 
-        osc_oap_to_pending(oap);
-        osc_page_transfer_get(opg, "transfer\0imm");
-        osc_page_transfer_add(env, opg, crt);
+static int osc_lru_reclaim(struct client_obd *cli)
+{
+       struct cl_client_cache *cache = cli->cl_cache;
+       int max_scans;
+       int rc;
+
+       LASSERT(cache != NULL);
+       LASSERT(!cfs_list_empty(&cache->ccc_lru));
+
+       rc = osc_lru_shrink(cli, lru_shrink_min);
+       if (rc != 0) {
+               CDEBUG(D_CACHE, "%s: Free %d pages from own LRU: %p.\n",
+                       cli->cl_import->imp_obd->obd_name, rc, cli);
+               return rc;
+       }
+
+       CDEBUG(D_CACHE, "%s: cli %p no free slots, pages: %d, busy: %d.\n",
+               cli->cl_import->imp_obd->obd_name, cli,
+               cfs_atomic_read(&cli->cl_lru_in_list),
+               cfs_atomic_read(&cli->cl_lru_busy));
+
+       /* Reclaim LRU slots from other client_obd as it can't free enough
+        * from its own. This should rarely happen. */
+       spin_lock(&cache->ccc_lru_lock);
+       cache->ccc_lru_shrinkers++;
+       cfs_list_move_tail(&cli->cl_lru_osc, &cache->ccc_lru);
+
+       max_scans = cfs_atomic_read(&cache->ccc_users);
+       while (--max_scans > 0 && !cfs_list_empty(&cache->ccc_lru)) {
+               cli = cfs_list_entry(cache->ccc_lru.next, struct client_obd,
+                                       cl_lru_osc);
+
+               CDEBUG(D_CACHE, "%s: cli %p LRU pages: %d, busy: %d.\n",
+                       cli->cl_import->imp_obd->obd_name, cli,
+                       cfs_atomic_read(&cli->cl_lru_in_list),
+                       cfs_atomic_read(&cli->cl_lru_busy));
+
+               cfs_list_move_tail(&cli->cl_lru_osc, &cache->ccc_lru);
+               if (cfs_atomic_read(&cli->cl_lru_in_list) > 0) {
+                       spin_unlock(&cache->ccc_lru_lock);
+
+                       rc = osc_lru_shrink(cli, max_to_shrink(cli));
+                       spin_lock(&cache->ccc_lru_lock);
+                       if (rc != 0)
+                               break;
+               }
+       }
+       spin_unlock(&cache->ccc_lru_lock);
+
+       CDEBUG(D_CACHE, "%s: cli %p freed %d pages.\n",
+               cli->cl_import->imp_obd->obd_name, cli, rc);
+       return rc;
+}
+
+static int osc_lru_reserve(const struct lu_env *env, struct osc_object *obj,
+                          struct osc_page *opg)
+{
+       struct l_wait_info lwi = LWI_INTR(LWI_ON_SIGNAL_NOOP, NULL);
+       struct client_obd *cli = osc_cli(obj);
+       int rc = 0;
+       ENTRY;
+
+       if (cli->cl_cache == NULL) /* shall not be in LRU */
+               RETURN(0);
+
+       LASSERT(cfs_atomic_read(cli->cl_lru_left) >= 0);
+       while (!cfs_atomic_add_unless(cli->cl_lru_left, -1, 0)) {
+               int gen;
+
+               /* run out of LRU spaces, try to drop some by itself */
+               rc = osc_lru_reclaim(cli);
+               if (rc < 0)
+                       break;
+               if (rc > 0)
+                       continue;
+
+               cond_resched();
+
+               /* slowest case, all of caching pages are busy, notifying
+                * other OSCs that we're lack of LRU slots. */
+               cfs_atomic_inc(&osc_lru_waiters);
+
+               gen = cfs_atomic_read(&cli->cl_lru_in_list);
+               rc = l_wait_event(osc_lru_waitq,
+                               cfs_atomic_read(cli->cl_lru_left) > 0 ||
+                               (cfs_atomic_read(&cli->cl_lru_in_list) > 0 &&
+                                gen != cfs_atomic_read(&cli->cl_lru_in_list)),
+                               &lwi);
+
+               cfs_atomic_dec(&osc_lru_waiters);
+               if (rc < 0)
+                       break;
+       }
+
+       if (rc >= 0) {
+               cfs_atomic_inc(&cli->cl_lru_busy);
+               opg->ops_in_lru = 1;
+               rc = 0;
+       }
+
+       RETURN(rc);
 }
 
 /** @} osc */