Whamcloud - gitweb
LU-6401 headers: Create a header for obdo related functions
[fs/lustre-release.git] / lustre / osc / osc_request.c
index cadef12..f636360 100644 (file)
@@ -27,7 +27,7 @@
  * Copyright (c) 2002, 2010, Oracle and/or its affiliates. All rights reserved.
  * Use is subject to license terms.
  *
- * Copyright (c) 2011, 2014, Intel Corporation.
+ * Copyright (c) 2011, 2015, Intel Corporation.
  */
 /*
  * This file is part of Lustre, http://www.lustre.org/
 
 #include <libcfs/libcfs.h>
 
-#include <lustre_dlm.h>
-#include <lustre_net.h>
 #include <lustre/lustre_user.h>
-#include <obd_cksum.h>
-#include <lustre_ha.h>
+
 #include <lprocfs_status.h>
-#include <lustre_ioctl.h>
 #include <lustre_debug.h>
-#include <lustre_param.h>
+#include <lustre_dlm.h>
 #include <lustre_fid.h>
-#include <obd_class.h>
-#include <obd.h>
+#include <lustre_ha.h>
+#include <lustre_ioctl.h>
 #include <lustre_net.h>
-#include "osc_internal.h"
+#include <lustre_obdo.h>
+#include <lustre_param.h>
+#include <obd.h>
+#include <obd_cksum.h>
+#include <obd_class.h>
+
 #include "osc_cl_internal.h"
+#include "osc_internal.h"
 
 atomic_t osc_pool_req_count;
 unsigned int osc_reqpool_maxreqcount;
@@ -72,7 +74,6 @@ struct osc_brw_async_args {
        struct client_obd        *aa_cli;
        struct list_head          aa_oaps;
        struct list_head          aa_exts;
-       struct cl_req            *aa_clerq;
 };
 
 #define osc_grant_args osc_brw_async_args
@@ -574,7 +575,10 @@ static void osc_announce_cached(struct client_obd *cli, struct obdo *oa,
 
        oa->o_valid |= bits;
        spin_lock(&cli->cl_loi_list_lock);
-       oa->o_dirty = cli->cl_dirty_pages << PAGE_CACHE_SHIFT;
+       if (OCD_HAS_FLAG(&cli->cl_import->imp_connect_data, GRANT_PARAM))
+               oa->o_dirty = cli->cl_dirty_grant;
+       else
+               oa->o_dirty = cli->cl_dirty_pages << PAGE_CACHE_SHIFT;
        if (unlikely(cli->cl_dirty_pages - cli->cl_dirty_transit >
                     cli->cl_dirty_max_pages)) {
                CERROR("dirty %lu - %lu > dirty_max %lu\n",
@@ -598,11 +602,22 @@ static void osc_announce_cached(struct client_obd *cli, struct obdo *oa,
                       cli->cl_dirty_pages, cli->cl_dirty_max_pages);
                oa->o_undirty = 0;
        } else {
-               unsigned long max_in_flight = (cli->cl_max_pages_per_rpc <<
-                                     PAGE_CACHE_SHIFT) *
-                                    (cli->cl_max_rpcs_in_flight + 1);
-               oa->o_undirty = max(cli->cl_dirty_max_pages << PAGE_CACHE_SHIFT,
-                                   max_in_flight);
+               unsigned long nrpages;
+
+               nrpages = cli->cl_max_pages_per_rpc;
+               nrpages *= cli->cl_max_rpcs_in_flight + 1;
+               nrpages = max(nrpages, cli->cl_dirty_max_pages);
+               oa->o_undirty = nrpages << PAGE_CACHE_SHIFT;
+               if (OCD_HAS_FLAG(&cli->cl_import->imp_connect_data,
+                                GRANT_PARAM)) {
+                       int nrextents;
+
+                       /* take extent tax into account when asking for more
+                        * grant space */
+                       nrextents = (nrpages + cli->cl_max_extent_pages - 1)  /
+                                    cli->cl_max_extent_pages;
+                       oa->o_undirty += nrextents * cli->cl_grant_extent_tax;
+               }
         }
        oa->o_grant = cli->cl_avail_grant + cli->cl_reserved_grant;
         oa->o_dropped = cli->cl_lost_grant;
@@ -610,7 +625,6 @@ static void osc_announce_cached(struct client_obd *cli, struct obdo *oa,
        spin_unlock(&cli->cl_loi_list_lock);
         CDEBUG(D_CACHE,"dirty: "LPU64" undirty: %u dropped %u grant: "LPU64"\n",
                oa->o_dirty, oa->o_undirty, oa->o_dropped, oa->o_grant);
-
 }
 
 void osc_update_next_shrink(struct client_obd *cli)
@@ -808,11 +822,15 @@ static void osc_init_grant(struct client_obd *cli, struct obd_connect_data *ocd)
         * left EVICTED state, then cl_dirty_pages must be 0 already.
         */
        spin_lock(&cli->cl_loi_list_lock);
-       if (cli->cl_import->imp_state == LUSTRE_IMP_EVICTED)
-               cli->cl_avail_grant = ocd->ocd_grant;
-       else
-               cli->cl_avail_grant = ocd->ocd_grant -
-                                     (cli->cl_dirty_pages << PAGE_CACHE_SHIFT);
+       cli->cl_avail_grant = ocd->ocd_grant;
+       if (cli->cl_import->imp_state != LUSTRE_IMP_EVICTED) {
+               cli->cl_avail_grant -= cli->cl_reserved_grant;
+               if (OCD_HAS_FLAG(ocd, GRANT_PARAM))
+                       cli->cl_avail_grant -= cli->cl_dirty_grant;
+               else
+                       cli->cl_avail_grant -=
+                                       cli->cl_dirty_pages << PAGE_CACHE_SHIFT;
+       }
 
         if (cli->cl_avail_grant < 0) {
                CWARN("%s: available grant < 0: avail/ocd/dirty %ld/%u/%ld\n",
@@ -823,13 +841,31 @@ static void osc_init_grant(struct client_obd *cli, struct obd_connect_data *ocd)
                cli->cl_avail_grant = ocd->ocd_grant;
         }
 
-       /* determine the appropriate chunk size used by osc_extent. */
-       cli->cl_chunkbits = max_t(int, PAGE_CACHE_SHIFT, ocd->ocd_blocksize);
+       if (OCD_HAS_FLAG(ocd, GRANT_PARAM)) {
+               u64 size;
+
+               /* overhead for each extent insertion */
+               cli->cl_grant_extent_tax = ocd->ocd_grant_tax_kb << 10;
+               /* determine the appropriate chunk size used by osc_extent. */
+               cli->cl_chunkbits = max_t(int, PAGE_CACHE_SHIFT,
+                                         ocd->ocd_grant_blkbits);
+               /* determine maximum extent size, in #pages */
+               size = (u64)ocd->ocd_grant_max_blks << ocd->ocd_grant_blkbits;
+               cli->cl_max_extent_pages = size >> PAGE_CACHE_SHIFT;
+               if (cli->cl_max_extent_pages == 0)
+                       cli->cl_max_extent_pages = 1;
+       } else {
+               cli->cl_grant_extent_tax = 0;
+               cli->cl_chunkbits = PAGE_CACHE_SHIFT;
+               cli->cl_max_extent_pages = DT_MAX_BRW_PAGES;
+       }
        spin_unlock(&cli->cl_loi_list_lock);
 
        CDEBUG(D_CACHE, "%s, setting cl_avail_grant: %ld cl_lost_grant: %ld."
-              "chunk bits: %d.\n", cli_name(cli), cli->cl_avail_grant,
-              cli->cl_lost_grant, cli->cl_chunkbits);
+               "chunk bits: %d cl_max_extent_pages: %d\n",
+               cli_name(cli),
+               cli->cl_avail_grant, cli->cl_lost_grant, cli->cl_chunkbits,
+               cli->cl_max_extent_pages);
 
        if (ocd->ocd_connect_flags & OBD_CONNECT_GRANT_SHRINK &&
            list_empty(&cli->cl_grant_shrink_list))
@@ -1590,8 +1626,6 @@ static int brw_interpret(const struct lu_env *env,
        LASSERT(list_empty(&aa->aa_exts));
        LASSERT(list_empty(&aa->aa_oaps));
 
-       cl_req_completion(env, aa->aa_clerq, rc < 0 ? rc :
-                         req->rq_bulk->bd_nob_transferred);
        osc_release_ppga(aa->aa_ppga, aa->aa_page_count);
        ptlrpc_lprocfs_brw(req, req->rq_bulk->bd_nob_transferred);
 
@@ -1642,10 +1676,7 @@ int osc_build_rpc(const struct lu_env *env, struct client_obd *cli,
        struct osc_brw_async_args       *aa = NULL;
        struct obdo                     *oa = NULL;
        struct osc_async_page           *oap;
-       struct osc_async_page           *tmp;
-       struct cl_req                   *clerq = NULL;
-       enum cl_req_type                crt = (cmd & OBD_BRW_WRITE) ? CRT_WRITE :
-                                                                     CRT_READ;
+       struct osc_object               *obj = NULL;
        struct cl_req_attr              *crattr = NULL;
        loff_t                          starting_offset = OBD_OBJECT_EOF;
        loff_t                          ending_offset = 0;
@@ -1653,7 +1684,9 @@ int osc_build_rpc(const struct lu_env *env, struct client_obd *cli,
        int                             mem_tight = 0;
        int                             page_count = 0;
        bool                            soft_sync = false;
+       bool                            interrupted = false;
        int                             i;
+       int                             grant = 0;
        int                             rc;
        struct list_head                rpc_list = LIST_HEAD_INIT(rpc_list);
        struct ost_body                 *body;
@@ -1664,31 +1697,16 @@ int osc_build_rpc(const struct lu_env *env, struct client_obd *cli,
        list_for_each_entry(ext, ext_list, oe_link) {
                LASSERT(ext->oe_state == OES_RPC);
                mem_tight |= ext->oe_memalloc;
-               list_for_each_entry(oap, &ext->oe_pages, oap_pending_item) {
-                       ++page_count;
-                       list_add_tail(&oap->oap_rpc_item, &rpc_list);
-                       if (starting_offset == OBD_OBJECT_EOF ||
-                           starting_offset > oap->oap_obj_off)
-                               starting_offset = oap->oap_obj_off;
-                       else
-                               LASSERT(oap->oap_page_off == 0);
-                       if (ending_offset < oap->oap_obj_off + oap->oap_count)
-                               ending_offset = oap->oap_obj_off +
-                                               oap->oap_count;
-                       else
-                               LASSERT(oap->oap_page_off + oap->oap_count ==
-                                       PAGE_CACHE_SIZE);
-               }
+               grant += ext->oe_grants;
+               page_count += ext->oe_nr_pages;
+               if (obj == NULL)
+                       obj = ext->oe_obj;
        }
 
        soft_sync = osc_over_unstable_soft_limit(cli);
        if (mem_tight)
                mpflag = cfs_memory_pressure_get_and_set();
 
-       OBD_ALLOC(crattr, sizeof(*crattr));
-       if (crattr == NULL)
-               GOTO(out, rc = -ENOMEM);
-
        OBD_ALLOC(pga, sizeof(*pga) * page_count);
        if (pga == NULL)
                GOTO(out, rc = -ENOMEM);
@@ -1698,37 +1716,46 @@ int osc_build_rpc(const struct lu_env *env, struct client_obd *cli,
                GOTO(out, rc = -ENOMEM);
 
        i = 0;
-       list_for_each_entry(oap, &rpc_list, oap_rpc_item) {
-               struct cl_page *page = oap2cl_page(oap);
-               if (clerq == NULL) {
-                       clerq = cl_req_alloc(env, page, crt,
-                                            1 /* only 1-object rpcs for now */);
-                       if (IS_ERR(clerq))
-                               GOTO(out, rc = PTR_ERR(clerq));
+       list_for_each_entry(ext, ext_list, oe_link) {
+               list_for_each_entry(oap, &ext->oe_pages, oap_pending_item) {
+                       if (mem_tight)
+                               oap->oap_brw_flags |= OBD_BRW_MEMALLOC;
+                       if (soft_sync)
+                               oap->oap_brw_flags |= OBD_BRW_SOFT_SYNC;
+                       pga[i] = &oap->oap_brw_page;
+                       pga[i]->off = oap->oap_obj_off + oap->oap_page_off;
+                       i++;
+
+                       list_add_tail(&oap->oap_rpc_item, &rpc_list);
+                       if (starting_offset == OBD_OBJECT_EOF ||
+                           starting_offset > oap->oap_obj_off)
+                               starting_offset = oap->oap_obj_off;
+                       else
+                               LASSERT(oap->oap_page_off == 0);
+                       if (ending_offset < oap->oap_obj_off + oap->oap_count)
+                               ending_offset = oap->oap_obj_off +
+                                               oap->oap_count;
+                       else
+                               LASSERT(oap->oap_page_off + oap->oap_count ==
+                                       PAGE_CACHE_SIZE);
+                       if (oap->oap_interrupted)
+                               interrupted = true;
                }
-               if (mem_tight)
-                       oap->oap_brw_flags |= OBD_BRW_MEMALLOC;
-               if (soft_sync)
-                       oap->oap_brw_flags |= OBD_BRW_SOFT_SYNC;
-               pga[i] = &oap->oap_brw_page;
-               pga[i]->off = oap->oap_obj_off + oap->oap_page_off;
-               CDEBUG(0, "put page %p index %lu oap %p flg %x to pga\n",
-                      pga[i]->pg, page_index(oap->oap_page), oap,
-                      pga[i]->flag);
-               i++;
-               cl_req_page_add(env, clerq, page);
        }
 
-       /* always get the data for the obdo for the rpc */
-       LASSERT(clerq != NULL);
+       /* first page in the list */
+       oap = list_entry(rpc_list.next, typeof(*oap), oap_rpc_item);
+
+       crattr = &osc_env_info(env)->oti_req_attr;
+       memset(crattr, 0, sizeof(*crattr));
+       crattr->cra_type = (cmd & OBD_BRW_WRITE) ? CRT_WRITE : CRT_READ;
+       crattr->cra_flags = ~0ULL;
+       crattr->cra_page = oap2cl_page(oap);
        crattr->cra_oa = oa;
-       cl_req_attr_set(env, clerq, crattr, ~0ULL);
+       cl_req_attr_set(env, osc2cl(obj), crattr);
 
-       rc = cl_req_prep(env, clerq);
-       if (rc != 0) {
-               CERROR("cl_req_prep failed: %d\n", rc);
-               GOTO(out, rc);
-       }
+       if (cmd == OBD_BRW_WRITE)
+               oa->o_grant_used = grant;
 
        sort_brw_pages(pga, page_count);
        rc = osc_brw_prep_request(cmd, cli, oa, page_count, pga, &req, 0);
@@ -1739,9 +1766,10 @@ int osc_build_rpc(const struct lu_env *env, struct client_obd *cli,
 
        req->rq_commit_cb = brw_commit;
        req->rq_interpret_reply = brw_interpret;
-
-       if (mem_tight != 0)
-               req->rq_memalloc = 1;
+       req->rq_memalloc = mem_tight != 0;
+       oap->oap_request = ptlrpc_request_addref(req);
+       if (interrupted && !req->rq_intr)
+               ptlrpc_mark_interrupted(req);
 
        /* Need to update the timestamps after the request is built in case
         * we race with setattr (locally or in queue at OST).  If OST gets
@@ -1750,9 +1778,8 @@ int osc_build_rpc(const struct lu_env *env, struct client_obd *cli,
         * way to do this in a single call.  bug 10150 */
        body = req_capsule_client_get(&req->rq_pill, &RMF_OST_BODY);
        crattr->cra_oa = &body->oa;
-       cl_req_attr_set(env, clerq, crattr,
-                       OBD_MD_FLMTIME|OBD_MD_FLCTIME|OBD_MD_FLATIME);
-
+       crattr->cra_flags = OBD_MD_FLMTIME|OBD_MD_FLCTIME|OBD_MD_FLATIME;
+       cl_req_attr_set(env, osc2cl(obj), crattr);
        lustre_msg_set_jobid(req->rq_reqmsg, crattr->cra_jobid);
 
        CLASSERT(sizeof(*aa) <= sizeof(req->rq_async_args));
@@ -1761,23 +1788,6 @@ int osc_build_rpc(const struct lu_env *env, struct client_obd *cli,
        list_splice_init(&rpc_list, &aa->aa_oaps);
        INIT_LIST_HEAD(&aa->aa_exts);
        list_splice_init(ext_list, &aa->aa_exts);
-       aa->aa_clerq = clerq;
-
-       /* queued sync pages can be torn down while the pages
-        * were between the pending list and the rpc */
-       tmp = NULL;
-       list_for_each_entry(oap, &aa->aa_oaps, oap_rpc_item) {
-               /* only one oap gets a request reference */
-               if (tmp == NULL)
-                       tmp = oap;
-               if (oap->oap_interrupted && !req->rq_intr) {
-                       CDEBUG(D_INODE, "oap %p in req %p interrupted\n",
-                                       oap, req);
-                       ptlrpc_mark_interrupted(req);
-               }
-       }
-       if (tmp != NULL)
-               tmp->oap_request = ptlrpc_request_addref(req);
 
        spin_lock(&cli->cl_loi_list_lock);
        starting_offset >>= PAGE_CACHE_SHIFT;
@@ -1808,9 +1818,6 @@ out:
        if (mem_tight != 0)
                cfs_memory_pressure_restore(mpflag);
 
-       if (crattr != NULL)
-               OBD_FREE(crattr, sizeof(*crattr));
-
        if (rc != 0) {
                LASSERT(req == NULL);
 
@@ -1826,8 +1833,6 @@ out:
                        list_del_init(&ext->oe_link);
                        osc_extent_finish(env, ext, 0, rc);
                }
-               if (clerq && !IS_ERR(clerq))
-                       cl_req_completion(env, clerq, rc);
        }
        RETURN(rc);
 }
@@ -2048,7 +2053,7 @@ int osc_enqueue_base(struct obd_export *exp, struct ldlm_res_id *res_id,
        }
 
 no_match:
-       if (*flags & LDLM_FL_TEST_LOCK)
+       if (*flags & (LDLM_FL_TEST_LOCK | LDLM_FL_MATCH_LOCK))
                RETURN(-ENOLCK);
 
        if (intent) {
@@ -2302,7 +2307,7 @@ static int osc_statfs(const struct lu_env *env, struct obd_export *exp,
 }
 
 static int osc_iocontrol(unsigned int cmd, struct obd_export *exp, int len,
-                         void *karg, void *uarg)
+                        void *karg, void __user *uarg)
 {
         struct obd_device *obd = exp->exp_obd;
         struct obd_ioctl_data *data = karg;
@@ -2468,11 +2473,15 @@ static int osc_reconnect(const struct lu_env *env,
 
         if (data != NULL && (data->ocd_connect_flags & OBD_CONNECT_GRANT)) {
                 long lost_grant;
+               long grant;
 
                spin_lock(&cli->cl_loi_list_lock);
-               data->ocd_grant = (cli->cl_avail_grant +
-                                 (cli->cl_dirty_pages << PAGE_CACHE_SHIFT)) ?:
-                                 2 * cli_brw_size(obd);
+               grant = cli->cl_avail_grant + cli->cl_reserved_grant;
+               if (data->ocd_connect_flags & OBD_CONNECT_GRANT_PARAM)
+                       grant += cli->cl_dirty_grant;
+               else
+                       grant += cli->cl_dirty_pages << PAGE_CACHE_SHIFT;
+               data->ocd_grant = grant ? : 2 * cli_brw_size(obd);
                lost_grant = cli->cl_lost_grant;
                cli->cl_lost_grant = 0;
                spin_unlock(&cli->cl_loi_list_lock);
@@ -2513,6 +2522,37 @@ static int osc_disconnect(struct obd_export *exp)
         return rc;
 }
 
+static int osc_ldlm_resource_invalidate(struct cfs_hash *hs,
+       struct cfs_hash_bd *bd, struct hlist_node *hnode, void *arg)
+{
+       struct lu_env *env = arg;
+       struct ldlm_resource *res = cfs_hash_object(hs, hnode);
+       struct ldlm_lock *lock;
+       struct osc_object *osc = NULL;
+       ENTRY;
+
+       lock_res(res);
+       list_for_each_entry(lock, &res->lr_granted, l_res_link) {
+               if (lock->l_ast_data != NULL && osc == NULL) {
+                       osc = lock->l_ast_data;
+                       cl_object_get(osc2cl(osc));
+               }
+
+               /* clear LDLM_FL_CLEANED flag to make sure it will be canceled
+                * by the 2nd round of ldlm_namespace_clean() call in
+                * osc_import_event(). */
+               ldlm_clear_cleaned(lock);
+       }
+       unlock_res(res);
+
+       if (osc != NULL) {
+               osc_object_invalidate(env, osc);
+               cl_object_put(env, osc2cl(osc));
+       }
+
+       RETURN(0);
+}
+
 static int osc_import_event(struct obd_device *obd,
                             struct obd_import *imp,
                             enum obd_import_event event)
@@ -2539,18 +2579,20 @@ static int osc_import_event(struct obd_device *obd,
         case IMP_EVENT_INVALIDATE: {
                 struct ldlm_namespace *ns = obd->obd_namespace;
                 struct lu_env         *env;
-                int                    refcheck;
+               __u16                  refcheck;
+
+               ldlm_namespace_cleanup(ns, LDLM_FL_LOCAL_ONLY);
 
                 env = cl_env_get(&refcheck);
                 if (!IS_ERR(env)) {
-                        /* Reset grants */
-                        cli = &obd->u.cli;
-                        /* all pages go to failing rpcs due to the invalid
-                         * import */
-                       osc_io_unplug(env, cli, NULL);
-
-                        ldlm_namespace_cleanup(ns, LDLM_FL_LOCAL_ONLY);
-                        cl_env_put(env, &refcheck);
+                       osc_io_unplug(env, &obd->u.cli, NULL);
+
+                       cfs_hash_for_each_nolock(ns->ns_rs_hash,
+                                                osc_ldlm_resource_invalidate,
+                                                env, 0);
+                       cl_env_put(env, &refcheck);
+
+                       ldlm_namespace_cleanup(ns, LDLM_FL_LOCAL_ONLY);
                 } else
                         rc = PTR_ERR(env);
                 break;
@@ -2699,6 +2741,11 @@ int osc_setup(struct obd_device *obd, struct lustre_cfg *lcfg)
 
        INIT_LIST_HEAD(&cli->cl_grant_shrink_list);
        ns_register_cancel(obd->obd_namespace, osc_cancel_weight);
+
+       spin_lock(&osc_shrink_lock);
+       list_add_tail(&cli->cl_shrink_list, &osc_shrink_list);
+       spin_unlock(&osc_shrink_lock);
+
        RETURN(0);
 
 out_ptlrpcd_work:
@@ -2717,50 +2764,35 @@ out_ptlrpcd:
        RETURN(rc);
 }
 
-static int osc_precleanup(struct obd_device *obd, enum obd_cleanup_stage stage)
+static int osc_precleanup(struct obd_device *obd)
 {
-        int rc = 0;
-        ENTRY;
+       struct client_obd *cli = &obd->u.cli;
+       ENTRY;
 
-        switch (stage) {
-        case OBD_CLEANUP_EARLY: {
-                struct obd_import *imp;
-                imp = obd->u.cli.cl_import;
-                CDEBUG(D_HA, "Deactivating import %s\n", obd->obd_name);
-                /* ptlrpc_abort_inflight to stop an mds_lov_synchronize */
-                ptlrpc_deactivate_import(imp);
-               spin_lock(&imp->imp_lock);
-               imp->imp_pingable = 0;
-               spin_unlock(&imp->imp_lock);
-                break;
-        }
-        case OBD_CLEANUP_EXPORTS: {
-                struct client_obd *cli = &obd->u.cli;
-                /* LU-464
-                 * for echo client, export may be on zombie list, wait for
-                 * zombie thread to cull it, because cli.cl_import will be
-                 * cleared in client_disconnect_export():
-                 *   class_export_destroy() -> obd_cleanup() ->
-                 *   echo_device_free() -> echo_client_cleanup() ->
-                 *   obd_disconnect() -> osc_disconnect() ->
-                 *   client_disconnect_export()
-                 */
-                obd_zombie_barrier();
-                if (cli->cl_writeback_work) {
-                        ptlrpcd_destroy_work(cli->cl_writeback_work);
-                        cli->cl_writeback_work = NULL;
-                }
-               if (cli->cl_lru_work) {
-                       ptlrpcd_destroy_work(cli->cl_lru_work);
-                       cli->cl_lru_work = NULL;
-               }
-                obd_cleanup_client_import(obd);
-                ptlrpc_lprocfs_unregister_obd(obd);
-                lprocfs_obd_cleanup(obd);
-                break;
-                }
-        }
-        RETURN(rc);
+       /* LU-464
+        * for echo client, export may be on zombie list, wait for
+        * zombie thread to cull it, because cli.cl_import will be
+        * cleared in client_disconnect_export():
+        *   class_export_destroy() -> obd_cleanup() ->
+        *   echo_device_free() -> echo_client_cleanup() ->
+        *   obd_disconnect() -> osc_disconnect() ->
+        *   client_disconnect_export()
+        */
+       obd_zombie_barrier();
+       if (cli->cl_writeback_work) {
+               ptlrpcd_destroy_work(cli->cl_writeback_work);
+               cli->cl_writeback_work = NULL;
+       }
+
+       if (cli->cl_lru_work) {
+               ptlrpcd_destroy_work(cli->cl_lru_work);
+               cli->cl_lru_work = NULL;
+       }
+
+       obd_cleanup_client_import(obd);
+       ptlrpc_lprocfs_unregister_obd(obd);
+       lprocfs_obd_cleanup(obd);
+       RETURN(0);
 }
 
 int osc_cleanup(struct obd_device *obd)
@@ -2770,6 +2802,10 @@ int osc_cleanup(struct obd_device *obd)
 
        ENTRY;
 
+       spin_lock(&osc_shrink_lock);
+       list_del(&cli->cl_shrink_list);
+       spin_unlock(&osc_shrink_lock);
+
        /* lru cleanup */
        if (cli->cl_cache != NULL) {
                LASSERT(atomic_read(&cli->cl_cache->ccc_users) > 0);
@@ -2781,7 +2817,7 @@ int osc_cleanup(struct obd_device *obd)
                cli->cl_cache = NULL;
        }
 
-        /* free memory of osc quota cache */
+       /* free memory of osc quota cache */
        osc_quota_cleanup(obd);
 
        rc = client_obd_cleanup(obd);
@@ -2824,6 +2860,27 @@ static struct obd_ops osc_obd_ops = {
         .o_quotactl             = osc_quotactl,
 };
 
+static struct shrinker *osc_cache_shrinker;
+struct list_head osc_shrink_list = LIST_HEAD_INIT(osc_shrink_list);
+DEFINE_SPINLOCK(osc_shrink_lock);
+
+#ifndef HAVE_SHRINKER_COUNT
+static int osc_cache_shrink(SHRINKER_ARGS(sc, nr_to_scan, gfp_mask))
+{
+       struct shrink_control scv = {
+               .nr_to_scan = shrink_param(sc, nr_to_scan),
+               .gfp_mask   = shrink_param(sc, gfp_mask)
+       };
+#if !defined(HAVE_SHRINKER_WANT_SHRINK_PTR) && !defined(HAVE_SHRINK_CONTROL)
+       struct shrinker *shrinker = NULL;
+#endif
+
+       (void)osc_cache_shrink_scan(shrinker, &scv);
+
+       return osc_cache_shrink_count(shrinker, &scv);
+}
+#endif
+
 static int __init osc_init(void)
 {
        bool enable_proc = true;
@@ -2831,15 +2888,16 @@ static int __init osc_init(void)
        unsigned int reqpool_size;
        unsigned int reqsize;
        int rc;
-
+       DEF_SHRINKER_VAR(osc_shvar, osc_cache_shrink,
+                        osc_cache_shrink_count, osc_cache_shrink_scan);
        ENTRY;
 
-        /* print an address of _any_ initialized kernel symbol from this
-         * module, to allow debugging with gdb that doesn't support data
-         * symbols from modules.*/
-        CDEBUG(D_INFO, "Lustre OSC module (%p).\n", &osc_caches);
+       /* print an address of _any_ initialized kernel symbol from this
+        * module, to allow debugging with gdb that doesn't support data
+        * symbols from modules.*/
+       CDEBUG(D_INFO, "Lustre OSC module (%p).\n", &osc_caches);
 
-        rc = lu_kmem_init(osc_caches);
+       rc = lu_kmem_init(osc_caches);
        if (rc)
                RETURN(rc);
 
@@ -2852,6 +2910,8 @@ static int __init osc_init(void)
        if (rc)
                GOTO(out_kmem, rc);
 
+       osc_cache_shrinker = set_shrinker(DEFAULT_SEEKS, &osc_shvar);
+
        /* This is obviously too much memory, only prevent overflow here */
        if (osc_reqpool_mem_max >= 1 << 12 || osc_reqpool_mem_max == 0)
                GOTO(out_type, rc = -EINVAL);
@@ -2885,14 +2945,15 @@ out:
        RETURN(rc);
 }
 
-static void /*__exit*/ osc_exit(void)
+static void __exit osc_exit(void)
 {
+       remove_shrinker(osc_cache_shrinker);
        class_unregister_type(LUSTRE_OSC_NAME);
        lu_kmem_fini(osc_caches);
        ptlrpc_free_rq_pool(osc_rq_pool);
 }
 
-MODULE_AUTHOR("Sun Microsystems, Inc. <http://www.lustre.org/>");
+MODULE_AUTHOR("OpenSFS, Inc. <http://www.lustre.org/>");
 MODULE_DESCRIPTION("Lustre Object Storage Client (OSC)");
 MODULE_VERSION(LUSTRE_VERSION_STRING);
 MODULE_LICENSE("GPL");