llite_lib.c

/*
 * GPL HEADER START
 *
 * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.
 *
 * This program is free software; you can redistribute it and/or modify
 * it under the terms of the GNU General Public License version 2 only,
 * as published by the Free Software Foundation.
 *
 * This program is distributed in the hope that it will be useful, but
 * WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
 * General Public License version 2 for more details (a copy is included
 * in the LICENSE file that accompanied this code).
 *
 * You should have received a copy of the GNU General Public License
 * version 2 along with this program; If not, see
 * http://www.gnu.org/licenses/gpl-2.0.html
 *
 * GPL HEADER END
 */
/*
 * Copyright (c) 2003, 2010, Oracle and/or its affiliates. All rights reserved.
 * Use is subject to license terms.
 *
 * Copyright (c) 2011, 2017, Intel Corporation.
 */
/*
 * This file is part of Lustre, http://www.lustre.org/
 * Lustre is a trademark of Sun Microsystems, Inc.
 *
 * lustre/llite/llite_lib.c
 *
 * Lustre Light Super operations
 */

#define DEBUG_SUBSYSTEM S_LLITE

#include <linux/cpu.h>
#include <linux/module.h>
#include <linux/random.h>
#include <linux/statfs.h>
#include <linux/time.h>
#include <linux/types.h>
#include <libcfs/linux/linux-uuid.h>
#include <linux/version.h>
#include <linux/mm.h>
#include <linux/user_namespace.h>
#include <linux/delay.h>
#include <linux/uidgid.h>
#include <linux/security.h>
#include <linux/fs_struct.h>

#ifndef HAVE_CPUS_READ_LOCK
#include <libcfs/linux/linux-cpu.h>
#endif
#include <uapi/linux/lustre/lustre_ioctl.h>
#ifdef HAVE_UAPI_LINUX_MOUNT_H
#include <uapi/linux/mount.h>
#endif

#include <lustre_ha.h>
#include <lustre_dlm.h>
#include <lprocfs_status.h>
#include <lustre_disk.h>
#include <uapi/linux/lustre/lustre_param.h>
#include <lustre_log.h>
#include <cl_object.h>
#include <obd_cksum.h>
#include "llite_internal.h"

struct kmem_cache *ll_file_data_slab;

#ifndef log2
#define log2(n) ffz(~(n))
#endif

/**
 * If there is only one number of core visible to Lustre,
 * async readahead will be disabled, to avoid massive over
 * subscription, we use 1/2 of active cores as default max
 * async readahead requests.
 */
static inline unsigned int ll_get_ra_async_max_active(void)
{
	return cfs_cpt_weight(cfs_cpt_tab, CFS_CPT_ANY) >> 1;
}

static struct ll_sb_info *ll_init_sbi(void)
{
	struct ll_sb_info *sbi = NULL;
	unsigned long pages;
	unsigned long lru_page_max;
	struct sysinfo si;
	int rc;
	int i;

	ENTRY;

	OBD_ALLOC_PTR(sbi);
	if (sbi == NULL)
		RETURN(ERR_PTR(-ENOMEM));

	rc = pcc_super_init(&sbi->ll_pcc_super);
	if (rc < 0)
		GOTO(out_sbi, rc);

	spin_lock_init(&sbi->ll_lock);
	mutex_init(&sbi->ll_lco.lco_lock);
	spin_lock_init(&sbi->ll_pp_extent_lock);
	spin_lock_init(&sbi->ll_process_lock);
        sbi->ll_rw_stats_on = 0;
	sbi->ll_statfs_max_age = OBD_STATFS_CACHE_SECONDS;

        si_meminfo(&si);
        pages = si.totalram - si.totalhigh;
	lru_page_max = pages / 2;

	sbi->ll_ra_info.ra_async_max_active = ll_get_ra_async_max_active();
	sbi->ll_ra_info.ll_readahead_wq =
		cfs_cpt_bind_workqueue("ll-readahead-wq", cfs_cpt_tab,
				       0, CFS_CPT_ANY,
				       sbi->ll_ra_info.ra_async_max_active);
	if (IS_ERR(sbi->ll_ra_info.ll_readahead_wq))
		GOTO(out_pcc, rc = PTR_ERR(sbi->ll_ra_info.ll_readahead_wq));

	/* initialize ll_cache data */
	sbi->ll_cache = cl_cache_init(lru_page_max);
	if (sbi->ll_cache == NULL)
		GOTO(out_destroy_ra, rc = -ENOMEM);

	sbi->ll_ra_info.ra_max_pages_per_file = min(pages / 32,
						    SBI_DEFAULT_READ_AHEAD_MAX);
	sbi->ll_ra_info.ra_async_pages_per_file_threshold =
				sbi->ll_ra_info.ra_max_pages_per_file;
	sbi->ll_ra_info.ra_max_pages = sbi->ll_ra_info.ra_max_pages_per_file;
	sbi->ll_ra_info.ra_max_read_ahead_whole_pages = -1;
	atomic_set(&sbi->ll_ra_info.ra_async_inflight, 0);

        sbi->ll_flags |= LL_SBI_VERBOSE;
#ifdef ENABLE_CHECKSUM
        sbi->ll_flags |= LL_SBI_CHECKSUM;
#endif
#ifdef ENABLE_FLOCK
	sbi->ll_flags |= LL_SBI_FLOCK;
#endif

#ifdef HAVE_LRU_RESIZE_SUPPORT
        sbi->ll_flags |= LL_SBI_LRU_RESIZE;
#endif
	sbi->ll_flags |= LL_SBI_LAZYSTATFS;

        for (i = 0; i <= LL_PROCESS_HIST_MAX; i++) {
		spin_lock_init(&sbi->ll_rw_extents_info.pp_extents[i].
			       pp_r_hist.oh_lock);
		spin_lock_init(&sbi->ll_rw_extents_info.pp_extents[i].
			       pp_w_hist.oh_lock);
        }

	/* metadata statahead is enabled by default */
	sbi->ll_sa_running_max = LL_SA_RUNNING_DEF;
	sbi->ll_sa_max = LL_SA_RPC_DEF;
	atomic_set(&sbi->ll_sa_total, 0);
	atomic_set(&sbi->ll_sa_wrong, 0);
	atomic_set(&sbi->ll_sa_running, 0);
	atomic_set(&sbi->ll_agl_total, 0);
	sbi->ll_flags |= LL_SBI_AGL_ENABLED;
	sbi->ll_flags |= LL_SBI_FAST_READ;
	sbi->ll_flags |= LL_SBI_TINY_WRITE;
	ll_sbi_set_encrypt(sbi, true);

	/* root squash */
	sbi->ll_squash.rsi_uid = 0;
	sbi->ll_squash.rsi_gid = 0;
	INIT_LIST_HEAD(&sbi->ll_squash.rsi_nosquash_nids);
	spin_lock_init(&sbi->ll_squash.rsi_lock);

	/* Per-filesystem file heat */
	sbi->ll_heat_decay_weight = SBI_DEFAULT_HEAT_DECAY_WEIGHT;
	sbi->ll_heat_period_second = SBI_DEFAULT_HEAT_PERIOD_SECOND;
	RETURN(sbi);
out_destroy_ra:
	destroy_workqueue(sbi->ll_ra_info.ll_readahead_wq);
out_pcc:
	pcc_super_fini(&sbi->ll_pcc_super);
out_sbi:
	OBD_FREE_PTR(sbi);
	RETURN(ERR_PTR(rc));
}

static void ll_free_sbi(struct super_block *sb)
{
	struct ll_sb_info *sbi = ll_s2sbi(sb);
	ENTRY;

	if (sbi != NULL) {
		if (!list_empty(&sbi->ll_squash.rsi_nosquash_nids))
			cfs_free_nidlist(&sbi->ll_squash.rsi_nosquash_nids);
		if (sbi->ll_ra_info.ll_readahead_wq)
			destroy_workqueue(sbi->ll_ra_info.ll_readahead_wq);
		if (sbi->ll_cache != NULL) {
			cl_cache_decref(sbi->ll_cache);
			sbi->ll_cache = NULL;
		}
		pcc_super_fini(&sbi->ll_pcc_super);
		OBD_FREE(sbi, sizeof(*sbi));
	}
	EXIT;
}

static int client_common_fill_super(struct super_block *sb, char *md, char *dt)
{
	struct inode *root = NULL;
	struct ll_sb_info *sbi = ll_s2sbi(sb);
	struct obd_statfs *osfs = NULL;
	struct ptlrpc_request *request = NULL;
	struct obd_connect_data *data = NULL;
	struct obd_uuid *uuid;
	struct md_op_data *op_data;
	struct lustre_md lmd;
	u64 valid;
	int size, err, checksum;

	ENTRY;
	sbi->ll_md_obd = class_name2obd(md);
	if (!sbi->ll_md_obd) {
                CERROR("MD %s: not setup or attached\n", md);
                RETURN(-EINVAL);
        }

        OBD_ALLOC_PTR(data);
        if (data == NULL)
                RETURN(-ENOMEM);

        OBD_ALLOC_PTR(osfs);
        if (osfs == NULL) {
                OBD_FREE_PTR(data);
                RETURN(-ENOMEM);
        }

	/* pass client page size via ocd_grant_blkbits, the server should report
	 * back its backend blocksize for grant calculation purpose */
	data->ocd_grant_blkbits = PAGE_SHIFT;

	/* indicate MDT features supported by this client */
	data->ocd_connect_flags = OBD_CONNECT_IBITS    | OBD_CONNECT_NODEVOH  |
				  OBD_CONNECT_ATTRFID  | OBD_CONNECT_GRANT |
				  OBD_CONNECT_VERSION  | OBD_CONNECT_BRW_SIZE |
				  OBD_CONNECT_SRVLOCK  | OBD_CONNECT_TRUNCLOCK|
				  OBD_CONNECT_MDS_CAPA | OBD_CONNECT_OSS_CAPA |
				  OBD_CONNECT_CANCELSET | OBD_CONNECT_FID     |
				  OBD_CONNECT_AT       | OBD_CONNECT_LOV_V3   |
				  OBD_CONNECT_VBR | OBD_CONNECT_FULL20 |
				  OBD_CONNECT_64BITHASH |
				  OBD_CONNECT_EINPROGRESS |
				  OBD_CONNECT_JOBSTATS | OBD_CONNECT_LVB_TYPE |
				  OBD_CONNECT_LAYOUTLOCK | OBD_CONNECT_PINGLESS|
				  OBD_CONNECT_MAX_EASIZE |
				  OBD_CONNECT_FLOCK_DEAD |
				  OBD_CONNECT_DISP_STRIPE | OBD_CONNECT_LFSCK |
				  OBD_CONNECT_OPEN_BY_FID |
				  OBD_CONNECT_DIR_STRIPE |
				  OBD_CONNECT_BULK_MBITS | OBD_CONNECT_CKSUM |
				  OBD_CONNECT_SUBTREE |
				  OBD_CONNECT_MULTIMODRPCS |
				  OBD_CONNECT_GRANT_PARAM |
				  OBD_CONNECT_SHORTIO | OBD_CONNECT_FLAGS2;

	data->ocd_connect_flags2 = OBD_CONNECT2_DIR_MIGRATE |
				   OBD_CONNECT2_SUM_STATFS |
				   OBD_CONNECT2_OVERSTRIPING |
				   OBD_CONNECT2_FLR |
				   OBD_CONNECT2_LOCK_CONVERT |
				   OBD_CONNECT2_ARCHIVE_ID_ARRAY |
				   OBD_CONNECT2_INC_XID |
				   OBD_CONNECT2_LSOM |
				   OBD_CONNECT2_ASYNC_DISCARD |
				   OBD_CONNECT2_PCC |
				   OBD_CONNECT2_CRUSH |
				   OBD_CONNECT2_GETATTR_PFID;

#ifdef HAVE_LRU_RESIZE_SUPPORT
        if (sbi->ll_flags & LL_SBI_LRU_RESIZE)
                data->ocd_connect_flags |= OBD_CONNECT_LRU_RESIZE;
#endif
	data->ocd_connect_flags |= OBD_CONNECT_ACL_FLAGS;

	data->ocd_cksum_types = obd_cksum_types_supported_client();

	if (OBD_FAIL_CHECK(OBD_FAIL_MDC_LIGHTWEIGHT))
		/* flag mdc connection as lightweight, only used for test
		 * purpose, use with care */
		data->ocd_connect_flags |= OBD_CONNECT_LIGHTWEIGHT;

	data->ocd_ibits_known = MDS_INODELOCK_FULL;
	data->ocd_version = LUSTRE_VERSION_CODE;

	if (sb->s_flags & SB_RDONLY)
		data->ocd_connect_flags |= OBD_CONNECT_RDONLY;
	if (sbi->ll_flags & LL_SBI_USER_XATTR)
		data->ocd_connect_flags |= OBD_CONNECT_XATTR;

#ifdef SB_NOSEC
	/* Setting this indicates we correctly support S_NOSEC (See kernel
	 * commit 9e1f1de02c2275d7172e18dc4e7c2065777611bf)
	 */
	sb->s_flags |= SB_NOSEC;
#endif

	if (sbi->ll_flags & LL_SBI_FLOCK)
		sbi->ll_fop = &ll_file_operations_flock;
	else if (sbi->ll_flags & LL_SBI_LOCALFLOCK)
		sbi->ll_fop = &ll_file_operations;
	else
		sbi->ll_fop = &ll_file_operations_noflock;

	/* always ping even if server suppress_pings */
	if (sbi->ll_flags & LL_SBI_ALWAYS_PING)
		data->ocd_connect_flags &= ~OBD_CONNECT_PINGLESS;

	obd_connect_set_secctx(data);
	if (ll_sbi_has_encrypt(sbi))
		obd_connect_set_enc(data);

#if defined(CONFIG_SECURITY)
	data->ocd_connect_flags2 |= OBD_CONNECT2_SELINUX_POLICY;
#endif

	data->ocd_brw_size = MD_MAX_BRW_SIZE;

	err = obd_connect(NULL, &sbi->ll_md_exp, sbi->ll_md_obd,
			  &sbi->ll_sb_uuid, data, sbi->ll_cache);
	if (err == -EBUSY) {
		LCONSOLE_ERROR_MSG(0x14f, "An MDT (md %s) is performing "
				   "recovery, of which this client is not a "
				   "part. Please wait for recovery to complete,"
				   " abort, or time out.\n", md);
		GOTO(out, err);
	} else if (err) {
		CERROR("cannot connect to %s: rc = %d\n", md, err);
		GOTO(out, err);
	}

	sbi->ll_md_exp->exp_connect_data = *data;

	err = obd_fid_init(sbi->ll_md_exp->exp_obd, sbi->ll_md_exp,
			   LUSTRE_SEQ_METADATA);
	if (err) {
		CERROR("%s: Can't init metadata layer FID infrastructure, "
		       "rc = %d\n", sbi->ll_md_exp->exp_obd->obd_name, err);
		GOTO(out_md, err);
	}

	/* For mount, we only need fs info from MDT0, and also in DNE, it
	 * can make sure the client can be mounted as long as MDT0 is
	 * avaible */
	err = obd_statfs(NULL, sbi->ll_md_exp, osfs,
			ktime_get_seconds() - sbi->ll_statfs_max_age,
			OBD_STATFS_FOR_MDT0);
	if (err)
		GOTO(out_md_fid, err);

	/* This needs to be after statfs to ensure connect has finished.
	 * Note that "data" does NOT contain the valid connect reply.
	 * If connecting to a 1.8 server there will be no LMV device, so
	 * we can access the MDC export directly and exp_connect_flags will
	 * be non-zero, but if accessing an upgraded 2.1 server it will
	 * have the correct flags filled in.
	 * XXX: fill in the LMV exp_connect_flags from MDC(s). */
	valid = exp_connect_flags(sbi->ll_md_exp) & CLIENT_CONNECT_MDT_REQD;
	if (exp_connect_flags(sbi->ll_md_exp) != 0 &&
	    valid != CLIENT_CONNECT_MDT_REQD) {
		char *buf;

		OBD_ALLOC_WAIT(buf, PAGE_SIZE);
		obd_connect_flags2str(buf, PAGE_SIZE,
				      valid ^ CLIENT_CONNECT_MDT_REQD, 0, ",");
		LCONSOLE_ERROR_MSG(0x170, "Server %s does not support "
				   "feature(s) needed for correct operation "
				   "of this client (%s). Please upgrade "
				   "server or downgrade client.\n",
				   sbi->ll_md_exp->exp_obd->obd_name, buf);
		OBD_FREE(buf, PAGE_SIZE);
		GOTO(out_md_fid, err = -EPROTO);
	}

	size = sizeof(*data);
	err = obd_get_info(NULL, sbi->ll_md_exp, sizeof(KEY_CONN_DATA),
			   KEY_CONN_DATA,  &size, data);
	if (err) {
		CERROR("%s: Get connect data failed: rc = %d\n",
		       sbi->ll_md_exp->exp_obd->obd_name, err);
		GOTO(out_md_fid, err);
	}

	LASSERT(osfs->os_bsize);
	sb->s_blocksize = osfs->os_bsize;
	sb->s_blocksize_bits = log2(osfs->os_bsize);
	sb->s_magic = LL_SUPER_MAGIC;
	sb->s_maxbytes = MAX_LFS_FILESIZE;
	sbi->ll_namelen = osfs->os_namelen;
	sbi->ll_mnt.mnt = current->fs->root.mnt;

	if ((sbi->ll_flags & LL_SBI_USER_XATTR) &&
	    !(data->ocd_connect_flags & OBD_CONNECT_XATTR)) {
		LCONSOLE_INFO("Disabling user_xattr feature because "
			      "it is not supported on the server\n");
		sbi->ll_flags &= ~LL_SBI_USER_XATTR;
	}

	if (data->ocd_connect_flags & OBD_CONNECT_ACL) {
#ifdef SB_POSIXACL
		sb->s_flags |= SB_POSIXACL;
#endif
		sbi->ll_flags |= LL_SBI_ACL;
	} else {
		LCONSOLE_INFO("client wants to enable acl, but mdt not!\n");
#ifdef SB_POSIXACL
		sb->s_flags &= ~SB_POSIXACL;
#endif
		sbi->ll_flags &= ~LL_SBI_ACL;
	}

	if (data->ocd_connect_flags & OBD_CONNECT_64BITHASH)
		sbi->ll_flags |= LL_SBI_64BIT_HASH;

	if (data->ocd_connect_flags & OBD_CONNECT_LAYOUTLOCK)
		sbi->ll_flags |= LL_SBI_LAYOUT_LOCK;

	if (obd_connect_has_secctx(data))
		sbi->ll_flags |= LL_SBI_FILE_SECCTX;

	if (ll_sbi_has_encrypt(sbi) && !obd_connect_has_enc(data)) {
		if (ll_sbi_has_test_dummy_encryption(sbi))
			LCONSOLE_WARN("%s: server %s does not support encryption feature, encryption deactivated.\n",
				      sbi->ll_fsname,
				      sbi->ll_md_exp->exp_obd->obd_name);
		ll_sbi_set_encrypt(sbi, false);
	}

	if (data->ocd_ibits_known & MDS_INODELOCK_XATTR) {
		if (!(data->ocd_connect_flags & OBD_CONNECT_MAX_EASIZE)) {
			LCONSOLE_INFO("%s: disabling xattr cache due to "
				      "unknown maximum xattr size.\n", dt);
		} else if (!sbi->ll_xattr_cache_set) {
			/* If xattr_cache is already set (no matter 0 or 1)
			 * during processing llog, it won't be enabled here. */
			sbi->ll_flags |= LL_SBI_XATTR_CACHE;
			sbi->ll_xattr_cache_enabled = 1;
		}
	}

	sbi->ll_dt_obd = class_name2obd(dt);
	if (!sbi->ll_dt_obd) {
		CERROR("DT %s: not setup or attached\n", dt);
		GOTO(out_md_fid, err = -ENODEV);
	}

	/* pass client page size via ocd_grant_blkbits, the server should report
	 * back its backend blocksize for grant calculation purpose */
	data->ocd_grant_blkbits = PAGE_SHIFT;

	/* indicate OST features supported by this client */
	data->ocd_connect_flags = OBD_CONNECT_GRANT | OBD_CONNECT_VERSION |
				  OBD_CONNECT_REQPORTAL | OBD_CONNECT_BRW_SIZE |
				  OBD_CONNECT_CANCELSET | OBD_CONNECT_FID |
				  OBD_CONNECT_SRVLOCK | OBD_CONNECT_TRUNCLOCK|
				  OBD_CONNECT_AT | OBD_CONNECT_OSS_CAPA |
				  OBD_CONNECT_VBR | OBD_CONNECT_FULL20 |
				  OBD_CONNECT_64BITHASH | OBD_CONNECT_MAXBYTES |
				  OBD_CONNECT_EINPROGRESS |
				  OBD_CONNECT_JOBSTATS | OBD_CONNECT_LVB_TYPE |
				  OBD_CONNECT_LAYOUTLOCK |
				  OBD_CONNECT_PINGLESS | OBD_CONNECT_LFSCK |
				  OBD_CONNECT_BULK_MBITS | OBD_CONNECT_SHORTIO |
				  OBD_CONNECT_FLAGS2 | OBD_CONNECT_GRANT_SHRINK;
	data->ocd_connect_flags2 = OBD_CONNECT2_LOCKAHEAD |
				   OBD_CONNECT2_INC_XID;

	if (!OBD_FAIL_CHECK(OBD_FAIL_OSC_CONNECT_GRANT_PARAM))
		data->ocd_connect_flags |= OBD_CONNECT_GRANT_PARAM;

	/* OBD_CONNECT_CKSUM should always be set, even if checksums are
	 * disabled by default, because it can still be enabled on the
	 * fly via /sys. As a consequence, we still need to come to an
	 * agreement on the supported algorithms at connect time
	 */
	data->ocd_connect_flags |= OBD_CONNECT_CKSUM;

	if (OBD_FAIL_CHECK(OBD_FAIL_OSC_CKSUM_ADLER_ONLY))
		data->ocd_cksum_types = OBD_CKSUM_ADLER;
	else
		data->ocd_cksum_types = obd_cksum_types_supported_client();

#ifdef HAVE_LRU_RESIZE_SUPPORT
	data->ocd_connect_flags |= OBD_CONNECT_LRU_RESIZE;
#endif
	/* always ping even if server suppress_pings */
	if (sbi->ll_flags & LL_SBI_ALWAYS_PING)
		data->ocd_connect_flags &= ~OBD_CONNECT_PINGLESS;

	if (ll_sbi_has_encrypt(sbi))
		obd_connect_set_enc(data);

	CDEBUG(D_RPCTRACE, "ocd_connect_flags: %#llx ocd_version: %d "
	       "ocd_grant: %d\n", data->ocd_connect_flags,
	       data->ocd_version, data->ocd_grant);

	sbi->ll_dt_obd->obd_upcall.onu_owner = &sbi->ll_lco;
	sbi->ll_dt_obd->obd_upcall.onu_upcall = cl_ocd_update;

	data->ocd_brw_size = DT_MAX_BRW_SIZE;

	err = obd_connect(NULL, &sbi->ll_dt_exp, sbi->ll_dt_obd,
			  &sbi->ll_sb_uuid, data, sbi->ll_cache);
	if (err == -EBUSY) {
		LCONSOLE_ERROR_MSG(0x150, "An OST (dt %s) is performing "
				   "recovery, of which this client is not a "
				   "part.  Please wait for recovery to "
				   "complete, abort, or time out.\n", dt);
		GOTO(out_md, err);
	} else if (err) {
		CERROR("%s: Cannot connect to %s: rc = %d\n",
		       sbi->ll_dt_exp->exp_obd->obd_name, dt, err);
		GOTO(out_md, err);
	}

	if (ll_sbi_has_encrypt(sbi) &&
	    !obd_connect_has_enc(&sbi->ll_dt_obd->u.lov.lov_ocd)) {
		if (ll_sbi_has_test_dummy_encryption(sbi))
			LCONSOLE_WARN("%s: server %s does not support encryption feature, encryption deactivated.\n",
				      sbi->ll_fsname, dt);
		ll_sbi_set_encrypt(sbi, false);
	} else if (ll_sbi_has_test_dummy_encryption(sbi)) {
		LCONSOLE_WARN("Test dummy encryption mode enabled\n");
	}

	sbi->ll_dt_exp->exp_connect_data = *data;

	/* Don't change value if it was specified in the config log */
	if (sbi->ll_ra_info.ra_max_read_ahead_whole_pages == -1) {
		sbi->ll_ra_info.ra_max_read_ahead_whole_pages =
			max_t(unsigned long, SBI_DEFAULT_READ_AHEAD_WHOLE_MAX,
			      (data->ocd_brw_size >> PAGE_SHIFT));
		if (sbi->ll_ra_info.ra_max_read_ahead_whole_pages >
		    sbi->ll_ra_info.ra_max_pages_per_file)
			sbi->ll_ra_info.ra_max_read_ahead_whole_pages =
				sbi->ll_ra_info.ra_max_pages_per_file;
	}

	err = obd_fid_init(sbi->ll_dt_exp->exp_obd, sbi->ll_dt_exp,
			   LUSTRE_SEQ_METADATA);
	if (err) {
		CERROR("%s: Can't init data layer FID infrastructure, "
		       "rc = %d\n", sbi->ll_dt_exp->exp_obd->obd_name, err);
		GOTO(out_dt, err);
	}

	mutex_lock(&sbi->ll_lco.lco_lock);
	sbi->ll_lco.lco_flags = data->ocd_connect_flags;
	sbi->ll_lco.lco_md_exp = sbi->ll_md_exp;
	sbi->ll_lco.lco_dt_exp = sbi->ll_dt_exp;
	mutex_unlock(&sbi->ll_lco.lco_lock);

	fid_zero(&sbi->ll_root_fid);
	err = md_get_root(sbi->ll_md_exp, get_mount_fileset(sb),
			   &sbi->ll_root_fid);
	if (err) {
		CERROR("cannot mds_connect: rc = %d\n", err);
		GOTO(out_lock_cn_cb, err);
	}
	if (!fid_is_sane(&sbi->ll_root_fid)) {
		CERROR("%s: Invalid root fid "DFID" during mount\n",
		       sbi->ll_md_exp->exp_obd->obd_name,
		       PFID(&sbi->ll_root_fid));
		GOTO(out_lock_cn_cb, err = -EINVAL);
	}
	CDEBUG(D_SUPER, "rootfid "DFID"\n", PFID(&sbi->ll_root_fid));

	sb->s_op = &lustre_super_operations;
	sb->s_xattr = ll_xattr_handlers;
#if THREAD_SIZE >= 8192 /*b=17630*/
	sb->s_export_op = &lustre_export_operations;
#endif
#ifdef HAVE_LUSTRE_CRYPTO
	llcrypt_set_ops(sb, &lustre_cryptops);
#endif

	/* make root inode
	 * XXX: move this to after cbd setup? */
	valid = OBD_MD_FLGETATTR | OBD_MD_FLBLOCKS | OBD_MD_FLMODEASIZE;
	if (sbi->ll_flags & LL_SBI_ACL)
		valid |= OBD_MD_FLACL;

	OBD_ALLOC_PTR(op_data);
	if (op_data == NULL)
		GOTO(out_lock_cn_cb, err = -ENOMEM);

	op_data->op_fid1 = sbi->ll_root_fid;
	op_data->op_mode = 0;
	op_data->op_valid = valid;

	err = md_getattr(sbi->ll_md_exp, op_data, &request);

	OBD_FREE_PTR(op_data);
	if (err) {
		CERROR("%s: md_getattr failed for root: rc = %d\n",
		       sbi->ll_md_exp->exp_obd->obd_name, err);
		GOTO(out_lock_cn_cb, err);
	}

	err = md_get_lustre_md(sbi->ll_md_exp, request, sbi->ll_dt_exp,
			       sbi->ll_md_exp, &lmd);
	if (err) {
		CERROR("failed to understand root inode md: rc = %d\n", err);
		ptlrpc_req_finished(request);
		GOTO(out_lock_cn_cb, err);
	}

	LASSERT(fid_is_sane(&sbi->ll_root_fid));
	root = ll_iget(sb, cl_fid_build_ino(&sbi->ll_root_fid,
					    sbi->ll_flags & LL_SBI_32BIT_API),
		       &lmd);
	md_free_lustre_md(sbi->ll_md_exp, &lmd);
	ptlrpc_req_finished(request);

	if (IS_ERR(root)) {
		lmd_clear_acl(&lmd);
		err = IS_ERR(root) ? PTR_ERR(root) : -EBADF;
		root = NULL;
		CERROR("%s: bad ll_iget() for root: rc = %d\n",
		       sbi->ll_fsname, err);
		GOTO(out_root, err);
	}

	checksum = sbi->ll_flags & LL_SBI_CHECKSUM;
	if (sbi->ll_checksum_set) {
		err = obd_set_info_async(NULL, sbi->ll_dt_exp,
					 sizeof(KEY_CHECKSUM), KEY_CHECKSUM,
					 sizeof(checksum), &checksum, NULL);
		if (err) {
			CERROR("%s: Set checksum failed: rc = %d\n",
			       sbi->ll_dt_exp->exp_obd->obd_name, err);
			GOTO(out_root, err);
		}
	}
	cl_sb_init(sb);

	sb->s_root = d_make_root(root);
	if (sb->s_root == NULL) {
		err = -ENOMEM;
		CERROR("%s: can't make root dentry: rc = %d\n",
		       sbi->ll_fsname, err);
		GOTO(out_root, err);
	}

	sbi->ll_sdev_orig = sb->s_dev;

	/* We set sb->s_dev equal on all lustre clients in order to support
	 * NFS export clustering.  NFSD requires that the FSID be the same
	 * on all clients. */
	/* s_dev is also used in lt_compare() to compare two fs, but that is
	 * only a node-local comparison. */
	uuid = obd_get_uuid(sbi->ll_md_exp);
	if (uuid != NULL)
		sb->s_dev = get_uuid2int(uuid->uuid, strlen(uuid->uuid));

	if (data != NULL)
		OBD_FREE_PTR(data);
	if (osfs != NULL)
		OBD_FREE_PTR(osfs);

	if (sbi->ll_dt_obd) {
		err = sysfs_create_link(&sbi->ll_kset.kobj,
					&sbi->ll_dt_obd->obd_kset.kobj,
					sbi->ll_dt_obd->obd_type->typ_name);
		if (err < 0) {
			CERROR("%s: could not register %s in llite: rc = %d\n",
			       dt, sbi->ll_fsname, err);
			err = 0;
		}
	}

	if (sbi->ll_md_obd) {
		err = sysfs_create_link(&sbi->ll_kset.kobj,
					&sbi->ll_md_obd->obd_kset.kobj,
					sbi->ll_md_obd->obd_type->typ_name);
		if (err < 0) {
			CERROR("%s: could not register %s in llite: rc = %d\n",
			       md, sbi->ll_fsname, err);
			err = 0;
		}
	}

	RETURN(err);
out_root:
	if (root)
		iput(root);
out_lock_cn_cb:
	obd_fid_fini(sbi->ll_dt_exp->exp_obd);
out_dt:
	obd_disconnect(sbi->ll_dt_exp);
	sbi->ll_dt_exp = NULL;
	sbi->ll_dt_obd = NULL;
out_md_fid:
	obd_fid_fini(sbi->ll_md_exp->exp_obd);
out_md:
	obd_disconnect(sbi->ll_md_exp);
	sbi->ll_md_exp = NULL;
	sbi->ll_md_obd = NULL;
out:
	if (data != NULL)
		OBD_FREE_PTR(data);
	if (osfs != NULL)
		OBD_FREE_PTR(osfs);
	return err;
}

int ll_get_max_mdsize(struct ll_sb_info *sbi, int *lmmsize)
{
	int size, rc;

	size = sizeof(*lmmsize);
	rc = obd_get_info(NULL, sbi->ll_dt_exp, sizeof(KEY_MAX_EASIZE),
			  KEY_MAX_EASIZE, &size, lmmsize);
	if (rc != 0) {
		CERROR("%s: cannot get max LOV EA size: rc = %d\n",
		       sbi->ll_dt_exp->exp_obd->obd_name, rc);
		RETURN(rc);
	}

	CDEBUG(D_INFO, "max LOV ea size: %d\n", *lmmsize);

	size = sizeof(int);
	rc = obd_get_info(NULL, sbi->ll_md_exp, sizeof(KEY_MAX_EASIZE),
			  KEY_MAX_EASIZE, &size, lmmsize);
	if (rc)
		CERROR("Get max mdsize error rc %d\n", rc);

	CDEBUG(D_INFO, "max LMV ea size: %d\n", *lmmsize);

	RETURN(rc);
}

/**
 * Get the value of the default_easize parameter.
 *
 * \see client_obd::cl_default_mds_easize
 *
 * \param[in] sbi	superblock info for this filesystem
 * \param[out] lmmsize	pointer to storage location for value
 *
 * \retval 0		on success
 * \retval negative	negated errno on failure
 */
int ll_get_default_mdsize(struct ll_sb_info *sbi, int *lmmsize)
{
	int size, rc;

	size = sizeof(int);
	rc = obd_get_info(NULL, sbi->ll_md_exp, sizeof(KEY_DEFAULT_EASIZE),
			 KEY_DEFAULT_EASIZE, &size, lmmsize);
	if (rc)
		CERROR("Get default mdsize error rc %d\n", rc);

	RETURN(rc);
}

/**
 * Set the default_easize parameter to the given value.
 *
 * \see client_obd::cl_default_mds_easize
 *
 * \param[in] sbi	superblock info for this filesystem
 * \param[in] lmmsize	the size to set
 *
 * \retval 0		on success
 * \retval negative	negated errno on failure
 */
int ll_set_default_mdsize(struct ll_sb_info *sbi, int lmmsize)
{
	int rc;

	if (lmmsize < sizeof(struct lov_mds_md) ||
	    lmmsize > OBD_MAX_DEFAULT_EA_SIZE)
		return -EINVAL;

	rc = obd_set_info_async(NULL, sbi->ll_md_exp,
				sizeof(KEY_DEFAULT_EASIZE), KEY_DEFAULT_EASIZE,
				sizeof(int), &lmmsize, NULL);

	RETURN(rc);
}

static void client_common_put_super(struct super_block *sb)
{
	struct ll_sb_info *sbi = ll_s2sbi(sb);
	ENTRY;

	cl_sb_fini(sb);

	obd_fid_fini(sbi->ll_dt_exp->exp_obd);
	obd_disconnect(sbi->ll_dt_exp);
	sbi->ll_dt_exp = NULL;

	ll_debugfs_unregister_super(sb);

	obd_fid_fini(sbi->ll_md_exp->exp_obd);
	obd_disconnect(sbi->ll_md_exp);
	sbi->ll_md_exp = NULL;

	EXIT;
}

void ll_kill_super(struct super_block *sb)
{
	struct ll_sb_info *sbi;
	ENTRY;

	/* not init sb ?*/
	if (!(sb->s_flags & SB_ACTIVE))
		return;

	sbi = ll_s2sbi(sb);
	/* we need restore s_dev from changed for clustred NFS before put_super
	 * because new kernels have cached s_dev and change sb->s_dev in
	 * put_super not affected real removing devices */
	if (sbi) {
		sb->s_dev = sbi->ll_sdev_orig;

		/* wait running statahead threads to quit */
		while (atomic_read(&sbi->ll_sa_running) > 0)
			schedule_timeout_uninterruptible(
				cfs_time_seconds(1) >> 3);
	}

	EXIT;
}

static inline int ll_set_opt(const char *opt, char *data, int fl)
{
	if (strncmp(opt, data, strlen(opt)) != 0)
		return 0;
	else
		return fl;
}

/* non-client-specific mount options are parsed in lmd_parse */
static int ll_options(char *options, struct ll_sb_info *sbi)
{
	int tmp;
	char *s1 = options, *s2;
	int *flags = &sbi->ll_flags;
	ENTRY;

	if (!options)
		RETURN(0);

	CDEBUG(D_CONFIG, "Parsing opts %s\n", options);

	while (*s1) {
		CDEBUG(D_SUPER, "next opt=%s\n", s1);
		tmp = ll_set_opt("nolock", s1, LL_SBI_NOLCK);
		if (tmp) {
			*flags |= tmp;
			goto next;
		}
		tmp = ll_set_opt("flock", s1, LL_SBI_FLOCK);
		if (tmp) {
			*flags = (*flags & ~LL_SBI_LOCALFLOCK) | tmp;
			goto next;
		}
		tmp = ll_set_opt("localflock", s1, LL_SBI_LOCALFLOCK);
		if (tmp) {
			*flags = (*flags & ~LL_SBI_FLOCK) | tmp;
			goto next;
		}
		tmp = ll_set_opt("noflock", s1, LL_SBI_FLOCK|LL_SBI_LOCALFLOCK);
		if (tmp) {
			*flags &= ~tmp;
			goto next;
		}
		tmp = ll_set_opt("user_xattr", s1, LL_SBI_USER_XATTR);
		if (tmp) {
			*flags |= tmp;
			goto next;
		}
		tmp = ll_set_opt("nouser_xattr", s1, LL_SBI_USER_XATTR);
		if (tmp) {
			*flags &= ~tmp;
			goto next;
		}
		tmp = ll_set_opt("context", s1, 1);
		if (tmp)
			goto next;
		tmp = ll_set_opt("fscontext", s1, 1);
		if (tmp)
			goto next;
		tmp = ll_set_opt("defcontext", s1, 1);
		if (tmp)
			goto next;
		tmp = ll_set_opt("rootcontext", s1, 1);
		if (tmp)
			goto next;
		tmp = ll_set_opt("user_fid2path", s1, LL_SBI_USER_FID2PATH);
		if (tmp) {
			*flags |= tmp;
			goto next;
		}
		tmp = ll_set_opt("nouser_fid2path", s1, LL_SBI_USER_FID2PATH);
		if (tmp) {
			*flags &= ~tmp;
			goto next;
		}

		tmp = ll_set_opt("checksum", s1, LL_SBI_CHECKSUM);
		if (tmp) {
			*flags |= tmp;
			sbi->ll_checksum_set = 1;
			goto next;
		}
		tmp = ll_set_opt("nochecksum",