* Copyright (c) 2001, 2010, Oracle and/or its affiliates. All rights reserved.
* Use is subject to license terms.
*
- * Copyright (c) 2011, 2013, Intel Corporation.
+ * Copyright (c) 2011, 2014, Intel Corporation.
*/
/*
* This file is part of Lustre, http://www.lustre.org/
#define DEBUG_SUBSYSTEM S_MDC
-#ifdef __KERNEL__
-# include <linux/module.h>
-# include <linux/pagemap.h>
-# include <linux/miscdevice.h>
-# include <linux/init.h>
-# include <linux/utsname.h>
-#else
-# include <liblustre.h>
+#include <linux/module.h>
+#include <linux/pagemap.h>
+#include <linux/miscdevice.h>
+#include <linux/init.h>
+#include <linux/utsname.h>
+#include <linux/kthread.h>
+#include <linux/user_namespace.h>
+#ifdef HAVE_UIDGID_HEADER
+# include <linux/uidgid.h>
#endif
#include <lustre_acl.h>
+#include <lustre_ioctl.h>
#include <obd_class.h>
#include <lustre_lmv.h>
#include <lustre_fid.h>
#include <lprocfs_status.h>
#include <lustre_param.h>
#include <lustre_log.h>
+#include <lustre_kernelcomm.h>
#include <cl_object.h>
-#include <lclient.h>
#include "mdc_internal.h"
static int mdc_cleanup(struct obd_device *obd);
-int mdc_unpack_capa(struct obd_export *exp, struct ptlrpc_request *req,
- const struct req_msg_field *field, struct obd_capa **oc)
+static int mdc_unpack_capa(struct obd_export *exp, struct ptlrpc_request *req,
+ const struct req_msg_field *field,
+ struct obd_capa **oc)
{
struct lustre_capa *capa;
struct obd_capa *c;
struct client_obd *cli = &req->rq_import->imp_obd->u.cli;
int rc;
- /* mdc_enter_request() ensures that this client has no more
+ /* obd_get_request_slot() ensures that this client has no more
* than cl_max_rpcs_in_flight RPCs simultaneously inf light
* against an MDT. */
- rc = mdc_enter_request(cli);
+ rc = obd_get_request_slot(cli);
if (rc != 0)
return rc;
rc = ptlrpc_queue_wait(req);
- mdc_exit_request(cli);
+ obd_put_request_slot(cli);
return rc;
}
if (body == NULL)
GOTO(out, rc = -EPROTO);
- if (body->valid & OBD_MD_FLMDSCAPA) {
- rc = mdc_unpack_capa(NULL, req, &RMF_CAPA1, pc);
- if (rc)
- GOTO(out, rc);
- }
+ if (body->mbo_valid & OBD_MD_FLMDSCAPA) {
+ rc = mdc_unpack_capa(NULL, req, &RMF_CAPA1, pc);
+ if (rc)
+ GOTO(out, rc);
+ }
- *rootfid = body->fid1;
- CDEBUG(D_NET,
- "root fid="DFID", last_committed="LPU64"\n",
- PFID(rootfid),
- lustre_msg_get_last_committed(req->rq_repmsg));
+ *rootfid = body->mbo_fid1;
+ CDEBUG(D_NET, "root fid="DFID", last_committed="LPU64"\n",
+ PFID(rootfid), lustre_msg_get_last_committed(req->rq_repmsg));
EXIT;
out:
ptlrpc_req_finished(req);
}
/* This should be mdc_get_info("rootfid") */
-int mdc_getstatus(struct obd_export *exp, struct lu_fid *rootfid,
- struct obd_capa **pc)
+static int mdc_getstatus(struct obd_export *exp, struct lu_fid *rootfid,
+ struct obd_capa **pc)
{
return send_getstatus(class_exp2cliimp(exp), rootfid, pc,
LUSTRE_IMP_FULL, 0);
if (body == NULL)
RETURN(-EPROTO);
- CDEBUG(D_NET, "mode: %o\n", body->mode);
+ CDEBUG(D_NET, "mode: %o\n", body->mbo_mode);
mdc_update_max_ea_from_body(exp, body);
- if (body->eadatasize != 0) {
+ if (body->mbo_eadatasize != 0) {
eadata = req_capsule_server_sized_get(pill, &RMF_MDT_MD,
- body->eadatasize);
+ body->mbo_eadatasize);
if (eadata == NULL)
RETURN(-EPROTO);
}
- if (body->valid & OBD_MD_FLRMTPERM) {
+ if (body->mbo_valid & OBD_MD_FLRMTPERM) {
struct mdt_remote_perm *perm;
LASSERT(client_is_remote(exp));
RETURN(-EPROTO);
}
- if (body->valid & OBD_MD_FLMDSCAPA) {
+ if (body->mbo_valid & OBD_MD_FLMDSCAPA) {
struct lustre_capa *capa;
capa = req_capsule_server_get(pill, &RMF_CAPA1);
if (capa == NULL)
RETURN(0);
}
-int mdc_getattr(struct obd_export *exp, struct md_op_data *op_data,
- struct ptlrpc_request **request)
+static int mdc_getattr(struct obd_export *exp, struct md_op_data *op_data,
+ struct ptlrpc_request **request)
{
struct ptlrpc_request *req;
int rc;
RETURN(rc);
}
-int mdc_getattr_name(struct obd_export *exp, struct md_op_data *op_data,
- struct ptlrpc_request **request)
+static int mdc_getattr_name(struct obd_export *exp, struct md_op_data *op_data,
+ struct ptlrpc_request **request)
{
struct ptlrpc_request *req;
int rc;
RETURN(rc);
}
-static int mdc_is_subdir(struct obd_export *exp,
- const struct lu_fid *pfid,
- const struct lu_fid *cfid,
- struct ptlrpc_request **request)
-{
- struct ptlrpc_request *req;
- int rc;
-
- ENTRY;
-
- *request = NULL;
- req = ptlrpc_request_alloc_pack(class_exp2cliimp(exp),
- &RQF_MDS_IS_SUBDIR, LUSTRE_MDS_VERSION,
- MDS_IS_SUBDIR);
- if (req == NULL)
- RETURN(-ENOMEM);
-
- mdc_is_subdir_pack(req, pfid, cfid, 0);
- ptlrpc_request_set_replen(req);
-
- rc = ptlrpc_queue_wait(req);
- if (rc && rc != -EREMOTE)
- ptlrpc_req_finished(req);
- else
- *request = req;
- RETURN(rc);
-}
-
static int mdc_xattr_common(struct obd_export *exp,const struct req_format *fmt,
- const struct lu_fid *fid,
- struct obd_capa *oc, int opcode, obd_valid valid,
- const char *xattr_name, const char *input,
- int input_size, int output_size, int flags,
- __u32 suppgid, struct ptlrpc_request **request)
+ const struct lu_fid *fid,
+ struct obd_capa *oc, int opcode, u64 valid,
+ const char *xattr_name, const char *input,
+ int input_size, int output_size, int flags,
+ __u32 suppgid, struct ptlrpc_request **request)
{
struct ptlrpc_request *req;
int xattr_namelen = 0;
/* Flush local XATTR locks to get rid of a possible cancel RPC */
if (opcode == MDS_REINT && fid_is_sane(fid) &&
exp->exp_connect_data.ocd_ibits_known & MDS_INODELOCK_XATTR) {
- CFS_LIST_HEAD(cancels);
+ struct list_head cancels = LIST_HEAD_INIT(cancels);
int count;
/* Without that packing would fail */
sizeof(struct mdt_rec_reint));
rec = req_capsule_client_get(&req->rq_pill, &RMF_REC_REINT);
rec->sx_opcode = REINT_SETXATTR;
- rec->sx_fsuid = current_fsuid();
- rec->sx_fsgid = current_fsgid();
+ rec->sx_fsuid = from_kuid(&init_user_ns, current_fsuid());
+ rec->sx_fsgid = from_kgid(&init_user_ns, current_fsgid());
rec->sx_cap = cfs_curproc_cap_pack();
rec->sx_suppgid1 = suppgid;
rec->sx_suppgid2 = -1;
RETURN(rc);
}
-int mdc_setxattr(struct obd_export *exp, const struct lu_fid *fid,
- struct obd_capa *oc, obd_valid valid, const char *xattr_name,
- const char *input, int input_size, int output_size,
- int flags, __u32 suppgid, struct ptlrpc_request **request)
+static int mdc_setxattr(struct obd_export *exp, const struct lu_fid *fid,
+ struct obd_capa *oc, u64 valid,
+ const char *xattr_name,
+ const char *input, int input_size, int output_size,
+ int flags, __u32 suppgid,
+ struct ptlrpc_request **request)
{
return mdc_xattr_common(exp, &RQF_MDS_REINT_SETXATTR,
fid, oc, MDS_REINT, valid, xattr_name,
suppgid, request);
}
-int mdc_getxattr(struct obd_export *exp, const struct lu_fid *fid,
- struct obd_capa *oc, obd_valid valid, const char *xattr_name,
- const char *input, int input_size, int output_size,
- int flags, struct ptlrpc_request **request)
+static int mdc_getxattr(struct obd_export *exp, const struct lu_fid *fid,
+ struct obd_capa *oc, u64 valid,
+ const char *xattr_name,
+ const char *input, int input_size, int output_size,
+ int flags, struct ptlrpc_request **request)
{
return mdc_xattr_common(exp, &RQF_MDS_GETXATTR,
fid, oc, MDS_GETXATTR, valid, xattr_name,
int rc;
ENTRY;
- if (!body->aclsize)
- RETURN(0);
+ if (!body->mbo_aclsize)
+ RETURN(0);
- buf = req_capsule_server_sized_get(pill, &RMF_ACL, body->aclsize);
+ buf = req_capsule_server_sized_get(pill, &RMF_ACL, body->mbo_aclsize);
- if (!buf)
- RETURN(-EPROTO);
+ if (!buf)
+ RETURN(-EPROTO);
- acl = posix_acl_from_xattr(&init_user_ns, buf, body->aclsize);
+ acl = posix_acl_from_xattr(&init_user_ns, buf, body->mbo_aclsize);
+ if (acl == NULL)
+ RETURN(0);
if (IS_ERR(acl)) {
rc = PTR_ERR(acl);
CERROR("convert xattr to acl: %d\n", rc);
md->body = req_capsule_server_get(pill, &RMF_MDT_BODY);
LASSERT(md->body != NULL);
- if (md->body->valid & OBD_MD_FLEASIZE) {
- int lmmsize;
- struct lov_mds_md *lmm;
+ if (md->body->mbo_valid & OBD_MD_FLEASIZE) {
+ int lmmsize;
+ struct lov_mds_md *lmm;
- if (!S_ISREG(md->body->mode)) {
- CDEBUG(D_INFO, "OBD_MD_FLEASIZE set, should be a "
- "regular file, but is not\n");
- GOTO(out, rc = -EPROTO);
- }
+ if (!S_ISREG(md->body->mbo_mode)) {
+ CDEBUG(D_INFO, "OBD_MD_FLEASIZE set, should be a "
+ "regular file, but is not\n");
+ GOTO(out, rc = -EPROTO);
+ }
- if (md->body->eadatasize == 0) {
- CDEBUG(D_INFO, "OBD_MD_FLEASIZE set, "
- "but eadatasize 0\n");
- GOTO(out, rc = -EPROTO);
- }
- lmmsize = md->body->eadatasize;
+ if (md->body->mbo_eadatasize == 0) {
+ CDEBUG(D_INFO, "OBD_MD_FLEASIZE set, "
+ "but eadatasize 0\n");
+ GOTO(out, rc = -EPROTO);
+ }
+
+ lmmsize = md->body->mbo_eadatasize;
lmm = req_capsule_server_sized_get(pill, &RMF_MDT_MD, lmmsize);
if (!lmm)
GOTO(out, rc = -EPROTO);
if (rc < 0)
GOTO(out, rc);
- if (rc < sizeof(*md->lsm)) {
+ if (rc < (typeof(rc))sizeof(*md->lsm)) {
CDEBUG(D_INFO, "lsm size too small: "
"rc < sizeof (*md->lsm) (%d < %d)\n",
rc, (int)sizeof(*md->lsm));
GOTO(out, rc = -EPROTO);
}
- } else if (md->body->valid & OBD_MD_FLDIREA) {
- int lmvsize;
- struct lov_mds_md *lmv;
+ } else if (md->body->mbo_valid & OBD_MD_FLDIREA) {
+ int lmvsize;
+ struct lov_mds_md *lmv;
- if(!S_ISDIR(md->body->mode)) {
- CDEBUG(D_INFO, "OBD_MD_FLDIREA set, should be a "
- "directory, but is not\n");
- GOTO(out, rc = -EPROTO);
- }
+ if (!S_ISDIR(md->body->mbo_mode)) {
+ CDEBUG(D_INFO, "OBD_MD_FLDIREA set, should be a "
+ "directory, but is not\n");
+ GOTO(out, rc = -EPROTO);
+ }
- if (md->body->eadatasize == 0) {
- CDEBUG(D_INFO, "OBD_MD_FLDIREA is set, "
- "but eadatasize 0\n");
- RETURN(-EPROTO);
- }
- if (md->body->valid & OBD_MD_MEA) {
- lmvsize = md->body->eadatasize;
+ if (md->body->mbo_eadatasize == 0) {
+ CDEBUG(D_INFO, "OBD_MD_FLDIREA is set, "
+ "but eadatasize 0\n");
+ RETURN(-EPROTO);
+ }
+
+ if (md->body->mbo_valid & OBD_MD_MEA) {
+ lmvsize = md->body->mbo_eadatasize;
lmv = req_capsule_server_sized_get(pill, &RMF_MDT_MD,
lmvsize);
if (!lmv)
if (rc < 0)
GOTO(out, rc);
- if (rc < sizeof(*md->lmv)) {
+ if (rc < (typeof(rc))sizeof(*md->lmv)) {
CDEBUG(D_INFO, "size too small: "
"rc < sizeof(*md->lmv) (%d < %d)\n",
rc, (int)sizeof(*md->lmv));
}
rc = 0;
- if (md->body->valid & OBD_MD_FLRMTPERM) {
- /* remote permission */
- LASSERT(client_is_remote(exp));
- md->remote_perm = req_capsule_server_swab_get(pill, &RMF_ACL,
- lustre_swab_mdt_remote_perm);
- if (!md->remote_perm)
- GOTO(out, rc = -EPROTO);
- }
- else if (md->body->valid & OBD_MD_FLACL) {
- /* for ACL, it's possible that FLACL is set but aclsize is zero.
- * only when aclsize != 0 there's an actual segment for ACL
- * in reply buffer.
- */
- if (md->body->aclsize) {
+ if (md->body->mbo_valid & OBD_MD_FLRMTPERM) {
+ /* remote permission */
+ LASSERT(client_is_remote(exp));
+ md->remote_perm = req_capsule_server_swab_get(pill, &RMF_ACL,
+ lustre_swab_mdt_remote_perm);
+ if (!md->remote_perm)
+ GOTO(out, rc = -EPROTO);
+ } else if (md->body->mbo_valid & OBD_MD_FLACL) {
+ /* for ACL, it's possible that FLACL is set but aclsize is zero.
+ * only when aclsize != 0 there's an actual segment for ACL
+ * in reply buffer.
+ */
+ if (md->body->mbo_aclsize) {
rc = mdc_unpack_acl(req, md);
if (rc)
GOTO(out, rc);
#endif
}
}
- if (md->body->valid & OBD_MD_FLMDSCAPA) {
+ if (md->body->mbo_valid & OBD_MD_FLMDSCAPA) {
struct obd_capa *oc = NULL;
rc = mdc_unpack_capa(NULL, req, &RMF_CAPA1, &oc);
md->mds_capa = oc;
}
- if (md->body->valid & OBD_MD_FLOSSCAPA) {
+ if (md->body->mbo_valid & OBD_MD_FLOSSCAPA) {
struct obd_capa *oc = NULL;
rc = mdc_unpack_capa(NULL, req, &RMF_CAPA2, &oc);
RETURN(0);
}
-/**
- * Handles both OPEN and SETATTR RPCs for OPEN-CLOSE and SETATTR-DONE_WRITING
- * RPC chains.
- */
void mdc_replay_open(struct ptlrpc_request *req)
{
struct md_open_data *mod = req->rq_cb_data;
LASSERT(och->och_magic == OBD_CLIENT_HANDLE_MAGIC);
file_fh = &och->och_fh;
- CDEBUG(D_HA, "updating handle from "LPX64" to "LPX64"\n",
- file_fh->cookie, body->handle.cookie);
- old = *file_fh;
- *file_fh = body->handle;
+ CDEBUG(D_HA, "updating handle from "LPX64" to "LPX64"\n",
+ file_fh->cookie, body->mbo_handle.cookie);
+ old = *file_fh;
+ *file_fh = body->mbo_handle;
}
close_req = mod->mod_close_req;
if (close_req != NULL) {
__u32 opc = lustre_msg_get_opc(close_req->rq_reqmsg);
struct mdt_ioepoch *epoch;
- LASSERT(opc == MDS_CLOSE || opc == MDS_DONE_WRITING);
- epoch = req_capsule_client_get(&close_req->rq_pill,
- &RMF_MDT_EPOCH);
- LASSERT(epoch);
+ LASSERT(opc == MDS_CLOSE);
+ epoch = req_capsule_client_get(&close_req->rq_pill,
+ &RMF_MDT_EPOCH);
+ LASSERT(epoch);
- if (och != NULL)
- LASSERT(!memcmp(&old, &epoch->handle, sizeof(old)));
- DEBUG_REQ(D_HA, close_req, "updating close body with new fh");
- epoch->handle = body->handle;
- }
- EXIT;
+ if (och != NULL)
+ LASSERT(!memcmp(&old, &epoch->mio_handle, sizeof(old)));
+
+ DEBUG_REQ(D_HA, close_req, "updating close body with new fh");
+ epoch->mio_handle = body->mbo_handle;
+ }
+ EXIT;
}
void mdc_commit_open(struct ptlrpc_request *req)
spin_unlock(&open_req->rq_lock);
}
- rec->cr_fid2 = body->fid1;
- rec->cr_ioepoch = body->ioepoch;
- rec->cr_old_handle.cookie = body->handle.cookie;
- open_req->rq_replay_cb = mdc_replay_open;
- if (!fid_is_sane(&body->fid1)) {
+ rec->cr_fid2 = body->mbo_fid1;
+ rec->cr_ioepoch = body->mbo_ioepoch;
+ rec->cr_old_handle.cookie = body->mbo_handle.cookie;
+ open_req->rq_replay_cb = mdc_replay_open;
+ if (!fid_is_sane(&body->mbo_fid1)) {
DEBUG_REQ(D_ERROR, open_req, "Saving replay request with "
"insane fid");
LBUG();
RETURN(0);
}
-/* Prepares the request for the replay by the given reply */
-static void mdc_close_handle_reply(struct ptlrpc_request *req,
- struct md_op_data *op_data, int rc) {
- struct mdt_body *repbody;
- struct mdt_ioepoch *epoch;
-
- if (req && rc == -EAGAIN) {
- repbody = req_capsule_server_get(&req->rq_pill, &RMF_MDT_BODY);
- epoch = req_capsule_client_get(&req->rq_pill, &RMF_MDT_EPOCH);
-
- epoch->flags |= MF_SOM_AU;
- if (repbody->valid & OBD_MD_FLGETATTRLOCK)
- op_data->op_flags |= MF_GETATTR_LOCK;
- }
-}
-
-int mdc_close(struct obd_export *exp, struct md_op_data *op_data,
- struct md_open_data *mod, struct ptlrpc_request **request)
+static int mdc_close(struct obd_export *exp, struct md_op_data *op_data,
+ struct md_open_data *mod, struct ptlrpc_request **request)
{
struct obd_device *obd = class_exp2obd(exp);
struct ptlrpc_request *req;
req_fmt = &RQF_MDS_RELEASE_CLOSE;
/* allocate a FID for volatile file */
- rc = mdc_fid_alloc(exp, &op_data->op_fid2, op_data);
+ rc = mdc_fid_alloc(NULL, exp, &op_data->op_fid2, op_data);
if (rc < 0) {
CERROR("%s: "DFID" failed to allocate FID: %d\n",
obd->obd_name, PFID(&op_data->op_fid1), rc);
mdc_close_pack(req, op_data);
- req_capsule_set_size(&req->rq_pill, &RMF_MDT_MD, RCL_SERVER,
- obd->u.cli.cl_max_mds_easize);
- req_capsule_set_size(&req->rq_pill, &RMF_LOGCOOKIES, RCL_SERVER,
- obd->u.cli.cl_max_mds_cookiesize);
+ req_capsule_set_size(&req->rq_pill, &RMF_MDT_MD, RCL_SERVER,
+ obd->u.cli.cl_default_mds_easize);
+ req_capsule_set_size(&req->rq_pill, &RMF_LOGCOOKIES, RCL_SERVER,
+ obd->u.cli.cl_default_mds_cookiesize);
ptlrpc_request_set_replen(req);
obd_mod_put(mod);
}
*request = req;
- mdc_close_handle_reply(req, op_data, rc);
- RETURN(rc < 0 ? rc : saved_rc);
-}
-
-int mdc_done_writing(struct obd_export *exp, struct md_op_data *op_data,
- struct md_open_data *mod)
-{
- struct obd_device *obd = class_exp2obd(exp);
- struct ptlrpc_request *req;
- int rc;
- ENTRY;
-
- req = ptlrpc_request_alloc(class_exp2cliimp(exp),
- &RQF_MDS_DONE_WRITING);
- if (req == NULL)
- RETURN(-ENOMEM);
-
- mdc_set_capa_size(req, &RMF_CAPA1, op_data->op_capa1);
- rc = ptlrpc_request_pack(req, LUSTRE_MDS_VERSION, MDS_DONE_WRITING);
- if (rc) {
- ptlrpc_request_free(req);
- RETURN(rc);
- }
-
- if (mod != NULL) {
- LASSERTF(mod->mod_open_req != NULL &&
- mod->mod_open_req->rq_type != LI_POISON,
- "POISONED setattr %p!\n", mod->mod_open_req);
-
- mod->mod_close_req = req;
- DEBUG_REQ(D_HA, mod->mod_open_req, "matched setattr");
- /* We no longer want to preserve this setattr for replay even
- * though the open was committed. b=3632, b=3633 */
- spin_lock(&mod->mod_open_req->rq_lock);
- mod->mod_open_req->rq_replay = 0;
- spin_unlock(&mod->mod_open_req->rq_lock);
- }
-
- mdc_close_pack(req, op_data);
- ptlrpc_request_set_replen(req);
-
- mdc_get_rpc_lock(obd->u.cli.cl_close_lock, NULL);
- rc = ptlrpc_queue_wait(req);
- mdc_put_rpc_lock(obd->u.cli.cl_close_lock, NULL);
-
- if (rc == -ESTALE) {
- /**
- * it can be allowed error after 3633 if open or setattr were
- * committed and server failed before close was sent.
- * Let's check if mod exists and return no error in that case
- */
- if (mod) {
- LASSERT(mod->mod_open_req != NULL);
- if (mod->mod_open_req->rq_committed)
- rc = 0;
- }
- }
-
- if (mod) {
- if (rc != 0)
- mod->mod_close_req = NULL;
- LASSERT(mod->mod_open_req != NULL);
- mdc_free_open(mod);
- /* Since now, mod is accessed through setattr req only,
- * thus DW req does not keep a reference on mod anymore. */
- obd_mod_put(mod);
- }
-
- mdc_close_handle_reply(req, op_data, rc);
- ptlrpc_req_finished(req);
- RETURN(rc);
+ RETURN(rc < 0 ? rc : saved_rc);
}
-#ifdef HAVE_SPLIT_SUPPORT
-int mdc_sendpage(struct obd_export *exp, const struct lu_fid *fid,
- const struct page *page, int offset)
-{
- struct ptlrpc_request *req;
- struct ptlrpc_bulk_desc *desc;
- int rc;
- ENTRY;
-
- req = ptlrpc_request_alloc(class_exp2cliimp(exp), &RQF_MDS_WRITEPAGE);
- if (req == NULL)
- RETURN(-ENOMEM);
-
- /* FIXME: capa doesn't support split yet */
- mdc_set_capa_size(req, &RMF_CAPA1, NULL);
-
- rc = ptlrpc_request_pack(req, LUSTRE_MDS_VERSION, MDS_WRITEPAGE);
- if (rc) {
- ptlrpc_request_free(req);
- RETURN(rc);
- }
-
- req->rq_request_portal = MDS_READPAGE_PORTAL;
- ptlrpc_at_set_req_timeout(req);
-
- desc = ptlrpc_prep_bulk_imp(req, 1, 1,BULK_GET_SOURCE, MDS_BULK_PORTAL);
- if (desc == NULL)
- GOTO(out, rc = -ENOMEM);
-
- /* NB req now owns desc and will free it when it gets freed. */
- ptlrpc_prep_bulk_page(desc, (struct page *)page, 0, offset);
- mdc_readdir_pack(req, 0, offset, fid, NULL);
-
- ptlrpc_request_set_replen(req);
- rc = ptlrpc_queue_wait(req);
- if (rc)
- GOTO(out, rc);
-
- rc = sptlrpc_cli_unwrap_bulk_write(req, req->rq_bulk);
-out:
- ptlrpc_req_finished(req);
- return rc;
-}
-EXPORT_SYMBOL(mdc_sendpage);
-#endif
-
static int mdc_getpage(struct obd_export *exp, const struct lu_fid *fid,
__u64 offset, struct obd_capa *oc,
struct page **pages, int npages,
RETURN(0);
}
-#ifdef __KERNEL__
static void mdc_release_page(struct page *page, int remove)
{
- kunmap(page);
if (remove) {
lock_page(page);
if (likely(page->mapping != NULL))
spin_lock_irq(&mapping->tree_lock);
found = radix_tree_gang_lookup(&mapping->page_tree,
(void **)&page, offset, 1);
- if (found > 0) {
+ if (found > 0 && !radix_tree_exceptional_entry(page)) {
struct lu_dirpage *dp;
page_cache_get(page);
*start = le64_to_cpu(dp->ldp_hash_start);
*end = le64_to_cpu(dp->ldp_hash_end);
}
- LASSERTF(*start <= *hash, "start = "LPX64",end = "
- LPX64",hash = "LPX64"\n", *start, *end, *hash);
- CDEBUG(D_VFSTRACE, "page%lu [%llu %llu], hash"LPU64"\n",
- offset, *start, *end, *hash);
+ if (unlikely(*start == 1 && *hash == 0))
+ *hash = *start;
+ else
+ LASSERTF(*start <= *hash, "start = "LPX64
+ ",end = "LPX64",hash = "LPX64"\n",
+ *start, *end, *hash);
+ CDEBUG(D_VFSTRACE, "offset %lx ["LPX64" "LPX64"],"
+ " hash "LPX64"\n", offset, *start, *end, *hash);
if (*hash > *end) {
+ kunmap(page);
mdc_release_page(page, 0);
page = NULL;
} else if (*end != *start && *hash == *end) {
/*
* upon hash collision, remove this page,
* otherwise put page reference, and
- * ll_get_dir_page() will issue RPC to fetch
- * the page we want.
+ * mdc_read_page_remote() will issue RPC to
+ * fetch the page we want.
*/
+ kunmap(page);
mdc_release_page(page,
le32_to_cpu(dp->ldp_flags) & LDF_COLLIDE);
page = NULL;
* | |
* .|--------v------- -----.
* |s|e|f|p|ent|ent| ... |ent|
- * '--|-------------- -----' Each CFS_PAGE contains a single
+ * '--|-------------- -----' Each PAGE contains a single
* '------. lu_dirpage.
* .---------v------- -----.
* |s|e|f|p|ent| 0 | ... | 0 |
* larger than LU_PAGE_SIZE, a single host page may contain multiple
* lu_dirpages. After reading the lu_dirpages from the MDS, the
* ldp_hash_end of the first lu_dirpage refers to the one immediately
- * after it in the same CFS_PAGE (arrows simplified for brevity, but
+ * after it in the same PAGE (arrows simplified for brevity, but
* in general e0==s1, e1==s2, etc.):
*
* .-------------------- -----.
* |s0|e0|f0|p|ent|ent| ... |ent|
* |---v---------------- -----|
* |s1|e1|f1|p|ent|ent| ... |ent|
- * |---v---------------- -----| Here, each CFS_PAGE contains
+ * |---v---------------- -----| Here, each PAGE contains
* ... multiple lu_dirpages.
* |---v---------------- -----|
* |s'|e'|f'|p|ent|ent| ... |ent|
* '---|---------------- -----'
* v
* .----------------------------.
- * | next CFS_PAGE |
+ * | next PAGE |
*
* This structure is transformed into a single logical lu_dirpage as follows:
*
* - Replace e0 with e' so the request for the next lu_dirpage gets the page
- * labeled 'next CFS_PAGE'.
+ * labeled 'next PAGE'.
*
* - Copy the LDF_COLLIDE flag from f' to f0 to correctly reflect whether
* a hash collision with the next page exists.
struct lu_dirpage *first = dp;
struct lu_dirent *end_dirent = NULL;
struct lu_dirent *ent;
- __u64 hash_end = dp->ldp_hash_end;
- __u32 flags = dp->ldp_flags;
+ __u64 hash_end = le64_to_cpu(dp->ldp_hash_end);
+ __u32 flags = le32_to_cpu(dp->ldp_flags);
while (--lu_pgs > 0) {
ent = lu_dirent_start(dp);
/* Advance dp to next lu_dirpage. */
dp = (struct lu_dirpage *)((char *)dp + LU_PAGE_SIZE);
- /* Check if we've reached the end of the CFS_PAGE. */
- if (!((unsigned long)dp & ~CFS_PAGE_MASK))
+ /* Check if we've reached the end of the PAGE. */
+ if (!((unsigned long)dp & ~PAGE_MASK))
break;
/* Save the hash and flags of this lu_dirpage. */
- hash_end = dp->ldp_hash_end;
- flags = dp->ldp_flags;
+ hash_end = le64_to_cpu(dp->ldp_hash_end);
+ flags = le32_to_cpu(dp->ldp_flags);
/* Check if lu_dirpage contains no entries. */
if (end_dirent == NULL)
kunmap(pages[i]);
}
- LASSERTF(lu_pgs == 0, "left = %d", lu_pgs);
+ LASSERTF(lu_pgs == 0, "left = %d\n", lu_pgs);
}
#else
#define mdc_adjust_dirpages(pages, cfs_pgs, lu_pgs) do {} while (0)
struct md_callback *rp_cb;
};
+#ifndef HAVE_DELETE_FROM_PAGE_CACHE
+static inline void delete_from_page_cache(struct page *page)
+{
+ remove_from_page_cache(page);
+ page_cache_release(page);
+}
+#endif
+
/**
* Read pages from server.
*
* a header lu_dirpage which describes the start/end hash, and whether this
* page is empty (contains no dir entry) or hash collide with next page.
* After client receives reply, several pages will be integrated into dir page
- * in CFS_PAGE_SIZE (if CFS_PAGE_SIZE greater than LU_PAGE_SIZE), and the
+ * in PAGE_SIZE (if PAGE_SIZE greater than LU_PAGE_SIZE), and the
* lu_dirpage for this integrated page will be adjusted.
**/
static int mdc_read_page_remote(void *data, struct page *page0)
ENTRY;
LASSERT(max_pages > 0 && max_pages <= PTLRPC_MAX_BRW_PAGES);
- if (op_data->op_mea1 != NULL) {
- __u32 index = op_data->op_stripe_offset;
-
- inode = op_data->op_mea1->lsm_md_oinfo[index].lmo_root;
- fid = &op_data->op_mea1->lsm_md_oinfo[index].lmo_fid;
- } else {
- inode = op_data->op_data;
- fid = &op_data->op_fid1;
- }
+ inode = op_data->op_data;
+ fid = &op_data->op_fid1;
LASSERT(inode != NULL);
OBD_ALLOC(page_pool, sizeof(page_pool[0]) * max_pages);
rc = mdc_getpage(rp->rp_exp, fid, rp->rp_off, op_data->op_capa1,
page_pool, npages, &req);
- if (rc == 0) {
+ if (rc < 0) {
+ /* page0 is special, which was added into page cache early */
+ delete_from_page_cache(page0);
+ } else {
int lu_pgs;
rd_pgs = (req->rq_bulk->bd_nob_transferred +
LU_PAGE_SHIFT;
LASSERT(!(req->rq_bulk->bd_nob_transferred & ~LU_PAGE_MASK));
- CDEBUG(D_INODE, "read %d(%d)/%d pages\n", rd_pgs, lu_pgs,
- op_data->op_npages);
+ CDEBUG(D_INODE, "read %d(%d) pages\n", rd_pgs, lu_pgs);
mdc_adjust_dirpages(page_pool, rd_pgs, lu_pgs);
SetPageUptodate(page0);
}
-
unlock_page(page0);
+
ptlrpc_req_finished(req);
CDEBUG(D_CACHE, "read %d/%d pages\n", rd_pgs, npages);
for (i = 1; i < npages; i++) {
/**
* Read dir page from cache first, if it can not find it, read it from
* server and add into the cache.
+ *
+ * \param[in] exp MDC export
+ * \param[in] op_data client MD stack parameters, transfering parameters
+ * between different layers on client MD stack.
+ * \param[in] cb_op callback required for ldlm lock enqueue during
+ * read page
+ * \param[in] hash_offset the hash offset of the page to be read
+ * \param[in] ppage the page to be read
+ *
+ * retval = 0 get the page successfully
+ * errno(<0) get the page failed
*/
static int mdc_read_page(struct obd_export *exp, struct md_op_data *op_data,
- struct md_callback *cb_op, struct page **ppage)
+ struct md_callback *cb_op, __u64 hash_offset,
+ struct page **ppage)
{
struct lookup_intent it = { .it_op = IT_READDIR };
struct page *page;
- struct inode *dir = NULL;
+ struct inode *dir = op_data->op_data;
struct address_space *mapping;
struct lu_dirpage *dp;
__u64 start = 0;
*ppage = NULL;
- if (op_data->op_mea1 != NULL) {
- __u32 index = op_data->op_stripe_offset;
-
- dir = op_data->op_mea1->lsm_md_oinfo[index].lmo_root;
- } else {
- dir = op_data->op_data;
- }
LASSERT(dir != NULL);
-
mapping = dir->i_mapping;
- rc = mdc_intent_lock(exp, op_data, NULL, 0, &it, 0, &enq_req,
+ rc = mdc_intent_lock(exp, op_data, &it, &enq_req,
cb_op->md_blocking_ast, 0);
if (enq_req != NULL)
ptlrpc_req_finished(enq_req);
rc = 0;
mdc_set_lock_data(exp, &it.d.lustre.it_lock_handle, dir, NULL);
- rp_param.rp_off = op_data->op_hash_offset;
+ rp_param.rp_off = hash_offset;
rp_param.rp_hash64 = op_data->op_cli_flags & CLI_HASH64;
page = mdc_page_locate(mapping, &rp_param.rp_off, &start, &end,
rp_param.rp_hash64);
rp_param.rp_hash64),
mdc_read_page_remote, &rp_param);
if (IS_ERR(page)) {
- CERROR("%s: read cache page: "DFID" at "LPU64": rc %ld\n",
+ CDEBUG(D_INFO, "%s: read cache page: "DFID" at "LPU64": %ld\n",
exp->exp_obd->obd_name, PFID(&op_data->op_fid1),
rp_param.rp_off, PTR_ERR(page));
GOTO(out_unlock, rc = PTR_ERR(page));
if (BITS_PER_LONG == 32 && rp_param.rp_hash64) {
start = le64_to_cpu(dp->ldp_hash_start) >> 32;
end = le64_to_cpu(dp->ldp_hash_end) >> 32;
- rp_param.rp_off = op_data->op_hash_offset >> 32;
+ rp_param.rp_off = hash_offset >> 32;
} else {
start = le64_to_cpu(dp->ldp_hash_start);
end = le64_to_cpu(dp->ldp_hash_end);
- rp_param.rp_off = op_data->op_hash_offset;
+ rp_param.rp_off = hash_offset;
}
if (end == start) {
LASSERT(start == rp_param.rp_off);
#if BITS_PER_LONG == 32
CWARN("Real page-wide hash collision at ["LPU64" "LPU64"] with "
"hash "LPU64"\n", le64_to_cpu(dp->ldp_hash_start),
- le64_to_cpu(dp->ldp_hash_end), op_data->op_hash_offset);
+ le64_to_cpu(dp->ldp_hash_end), hash_offset);
#endif
/*
goto out_unlock;
}
-/**
- * Read one directory entry from the cache.
- */
-int mdc_read_entry(struct obd_export *exp, struct md_op_data *op_data,
- struct md_callback *cb_op, struct lu_dirent **entp)
-{
- struct page *page = NULL;
- struct lu_dirpage *dp;
- struct lu_dirent *ent;
- int rc = 0;
- int index = 0;
- ENTRY;
-
- if (op_data->op_hash_offset == MDS_DIR_END_OFF) {
- *entp = NULL;
- RETURN(0);
- }
-
- rc = mdc_read_page(exp, op_data, cb_op, &page);
- if (rc != 0)
- RETURN(rc);
-
- if (op_data->op_cli_flags & CLI_READENT_END) {
- mdc_release_page(page, 0);
- RETURN(0);
- }
-
- dp = kmap(page);
- for (ent = lu_dirent_start(dp); ent != NULL;
- ent = lu_dirent_next(ent)) {
- index++;
- if (ent->lde_hash > op_data->op_hash_offset)
- break;
- }
- kunmap(page);
-
- /* If it can not find entry in current page, try next page. */
- if (ent == NULL) {
- __u64 orig_offset = op_data->op_hash_offset;
-
- if (dp->ldp_hash_end == MDS_DIR_END_OFF) {
- mdc_release_page(page, 0);
- RETURN(0);
- }
-
- op_data->op_hash_offset = dp->ldp_hash_end;
- mdc_release_page(page,
- le32_to_cpu(dp->ldp_flags) & LDF_COLLIDE);
- rc = mdc_read_page(exp, op_data, cb_op, &page);
- if (rc != 0)
- RETURN(rc);
-
- if (page != NULL) {
- dp = kmap(page);
- ent = lu_dirent_start(dp);
- kunmap(page);
- }
-
- op_data->op_hash_offset = orig_offset;
- }
-
- *entp = ent;
-
- RETURN(rc);
-}
-
-#else /* __KERNEL__ */
-
-static struct page
-*mdc_read_page_remote(struct obd_export *exp, const struct lmv_oinfo *lmo,
- const __u64 hash, struct obd_capa *oc)
-{
- struct ptlrpc_request *req = NULL;
- struct page *page;
- int rc;
-
- OBD_PAGE_ALLOC(page, 0);
- if (page == NULL)
- return ERR_PTR(-ENOMEM);
-
- rc = mdc_getpage(exp, &lmo->lmo_fid, hash, oc, &page, 1, &req);
- if (req != NULL)
- ptlrpc_req_finished(req);
-
- if (unlikely(rc)) {
- OBD_PAGE_FREE(page);
- return ERR_PTR(rc);
- }
- return page;
-}
-
-
-static int mdc_read_page(struct obd_export *exp, struct md_op_data *op_data,
- struct md_callback *cb_op,
- struct page **ppage)
-{
- struct page *page;
- struct lmv_oinfo *lmo;
- int rc = 0;
-
- /* No local cache for liblustre, always read entry remotely */
- lmo = &op_data->op_mea1->lsm_md_oinfo[op_data->op_stripe_offset];
- page = mdc_read_page_remote(exp, lmo, op_data->op_hash_offset,
- op_data->op_capa1);
- if (IS_ERR(page))
- return PTR_ERR(page);
-
- *ppage = page;
-
- return rc;
-}
-
-int mdc_read_entry(struct obd_export *exp, struct md_op_data *op_data,
- struct md_callback *cb_op, struct lu_dirent **entp)
-{
- struct page *page = NULL;
- struct lu_dirpage *dp;
- struct lu_dirent *ent;
- int rc;
- ENTRY;
-
- rc = mdc_read_page(exp, op_data, cb_op, &page);
- if (rc != 0)
- RETURN(rc);
-
- dp = page_address(page);
- if (dp->ldp_hash_end < op_data->op_hash_offset)
- GOTO(out, *entp = NULL);
-
- for (ent = lu_dirent_start(dp); ent != NULL;
- ent = lu_dirent_next(ent))
- if (ent->lde_hash >= op_data->op_hash_offset)
- break;
- *entp = ent;
-out:
-
- OBD_PAGE_FREE(page);
- RETURN(rc);
-}
-
-#endif
static int mdc_statfs(const struct lu_env *env,
struct obd_export *exp, struct obd_statfs *osfs,
/* Val is struct getinfo_fid2path result plus path */
vallen = sizeof(*gf) + gf->gf_pathlen;
- rc = obd_get_info(NULL, exp, keylen, key, &vallen, gf, NULL);
+ rc = obd_get_info(NULL, exp, keylen, key, &vallen, gf);
if (rc != 0 && rc != -EREMOTE)
GOTO(out, rc);
ptlrpc_request_set_replen(req);
- rc = mdc_queue_wait(req);
+ mdc_get_rpc_lock(exp->exp_obd->u.cli.cl_rpc_lock, NULL);
+ rc = ptlrpc_queue_wait(req);
+ mdc_put_rpc_lock(exp->exp_obd->u.cli.cl_rpc_lock, NULL);
+
GOTO(out, rc);
out:
ptlrpc_req_finished(req);
ptlrpc_request_set_replen(req);
- rc = mdc_queue_wait(req);
- GOTO(out, rc);
+ mdc_get_rpc_lock(exp->exp_obd->u.cli.cl_rpc_lock, NULL);
+ rc = ptlrpc_queue_wait(req);
+ mdc_put_rpc_lock(exp->exp_obd->u.cli.cl_rpc_lock, NULL);
- EXIT;
+ GOTO(out, rc);
out:
ptlrpc_req_finished(req);
return rc;
ptlrpc_request_set_replen(req);
- rc = mdc_queue_wait(req);
+ mdc_get_rpc_lock(exp->exp_obd->u.cli.cl_rpc_lock, NULL);
+ rc = ptlrpc_queue_wait(req);
+ mdc_put_rpc_lock(exp->exp_obd->u.cli.cl_rpc_lock, NULL);
+
GOTO(out, rc);
out:
return rc;
}
-static struct kuc_hdr *changelog_kuc_hdr(char *buf, int len, int flags)
+static struct kuc_hdr *changelog_kuc_hdr(char *buf, size_t len, __u32 flags)
{
struct kuc_hdr *lh = (struct kuc_hdr *)buf;
return lh;
}
-#define D_CHANGELOG 0
-
struct changelog_show {
- __u64 cs_startrec;
- __u32 cs_flags;
- struct file *cs_fp;
- char *cs_buf;
- struct obd_device *cs_obd;
+ __u64 cs_startrec;
+ enum changelog_send_flag cs_flags;
+ struct file *cs_fp;
+ char *cs_buf;
+ struct obd_device *cs_obd;
};
+static inline char *cs_obd_name(struct changelog_show *cs)
+{
+ return cs->cs_obd->obd_name;
+}
+
static int changelog_kkuc_cb(const struct lu_env *env, struct llog_handle *llh,
struct llog_rec_hdr *hdr, void *data)
{
- struct changelog_show *cs = data;
- struct llog_changelog_rec *rec = (struct llog_changelog_rec *)hdr;
- struct kuc_hdr *lh;
- int len, rc;
+ struct changelog_show *cs = data;
+ struct llog_changelog_rec *rec = (struct llog_changelog_rec *)hdr;
+ struct kuc_hdr *lh;
+ size_t len;
+ int rc;
ENTRY;
if (rec->cr_hdr.lrh_type != CHANGELOG_REC) {
rc = -EINVAL;
CERROR("%s: not a changelog rec %x/%d: rc = %d\n",
- cs->cs_obd->obd_name, rec->cr_hdr.lrh_type,
+ cs_obd_name(cs), rec->cr_hdr.lrh_type,
rec->cr.cr_type, rc);
RETURN(rc);
}
if (rec->cr.cr_index < cs->cs_startrec) {
/* Skip entries earlier than what we are interested in */
- CDEBUG(D_CHANGELOG, "rec="LPU64" start="LPU64"\n",
+ CDEBUG(D_HSM, "rec="LPU64" start="LPU64"\n",
rec->cr.cr_index, cs->cs_startrec);
RETURN(0);
}
- CDEBUG(D_CHANGELOG, LPU64" %02d%-5s "LPU64" 0x%x t="DFID" p="DFID
- " %.*s\n", rec->cr.cr_index, rec->cr.cr_type,
- changelog_type2str(rec->cr.cr_type), rec->cr.cr_time,
- rec->cr.cr_flags & CLF_FLAGMASK,
- PFID(&rec->cr.cr_tfid), PFID(&rec->cr.cr_pfid),
- rec->cr.cr_namelen, changelog_rec_name(&rec->cr));
+ CDEBUG(D_HSM, LPU64" %02d%-5s "LPU64" 0x%x t="DFID" p="DFID" %.*s\n",
+ rec->cr.cr_index, rec->cr.cr_type,
+ changelog_type2str(rec->cr.cr_type), rec->cr.cr_time,
+ rec->cr.cr_flags & CLF_FLAGMASK,
+ PFID(&rec->cr.cr_tfid), PFID(&rec->cr.cr_pfid),
+ rec->cr.cr_namelen, changelog_rec_name(&rec->cr));
len = sizeof(*lh) + changelog_rec_size(&rec->cr) + rec->cr.cr_namelen;
memcpy(lh + 1, &rec->cr, len - sizeof(*lh));
rc = libcfs_kkuc_msg_put(cs->cs_fp, lh);
- CDEBUG(D_CHANGELOG, "kucmsg fp %p len %d rc %d\n", cs->cs_fp, len,rc);
+ CDEBUG(D_HSM, "kucmsg fp %p len %zu rc %d\n", cs->cs_fp, len, rc);
RETURN(rc);
}
static int mdc_changelog_send_thread(void *csdata)
{
- struct changelog_show *cs = csdata;
- struct llog_ctxt *ctxt = NULL;
- struct llog_handle *llh = NULL;
- struct kuc_hdr *kuch;
- int rc;
+ struct changelog_show *cs = csdata;
+ struct llog_ctxt *ctxt = NULL;
+ struct llog_handle *llh = NULL;
+ struct kuc_hdr *kuch;
+ enum llog_flag flags = LLOG_F_IS_CAT;
+ int rc;
- CDEBUG(D_CHANGELOG, "changelog to fp=%p start "LPU64"\n",
+ CDEBUG(D_HSM, "changelog to fp=%p start "LPU64"\n",
cs->cs_fp, cs->cs_startrec);
OBD_ALLOC(cs->cs_buf, KUC_CHANGELOG_MSG_MAXSIZE);
if (cs->cs_buf == NULL)
GOTO(out, rc = -ENOMEM);
- /* Set up the remote catalog handle */
- ctxt = llog_get_context(cs->cs_obd, LLOG_CHANGELOG_REPL_CTXT);
- if (ctxt == NULL)
- GOTO(out, rc = -ENOENT);
+ /* Set up the remote catalog handle */
+ ctxt = llog_get_context(cs->cs_obd, LLOG_CHANGELOG_REPL_CTXT);
+ if (ctxt == NULL)
+ GOTO(out, rc = -ENOENT);
rc = llog_open(NULL, ctxt, &llh, NULL, CHANGELOG_CATALOG,
LLOG_OPEN_EXISTS);
if (rc) {
CERROR("%s: fail to open changelog catalog: rc = %d\n",
- cs->cs_obd->obd_name, rc);
+ cs_obd_name(cs), rc);
GOTO(out, rc);
}
- rc = llog_init_handle(NULL, llh, LLOG_F_IS_CAT, NULL);
+
+ if (cs->cs_flags & CHANGELOG_FLAG_JOBID)
+ flags |= LLOG_F_EXT_JOBID;
+
+ rc = llog_init_handle(NULL, llh, flags, NULL);
if (rc) {
CERROR("llog_init_handle failed %d\n", rc);
GOTO(out, rc);
rc = llog_cat_process(NULL, llh, changelog_kkuc_cb, cs, 0, 0);
- /* Send EOF no matter what our result */
- if ((kuch = changelog_kuc_hdr(cs->cs_buf, sizeof(*kuch),
- cs->cs_flags))) {
- kuch->kuc_msgtype = CL_EOF;
- libcfs_kkuc_msg_put(cs->cs_fp, kuch);
- }
+ /* Send EOF no matter what our result */
+ kuch = changelog_kuc_hdr(cs->cs_buf, sizeof(*kuch), cs->cs_flags);
+ kuch->kuc_msgtype = CL_EOF;
+ libcfs_kkuc_msg_put(cs->cs_fp, kuch);
out:
fput(cs->cs_fp);
if (llh)
llog_cat_close(NULL, llh);
- if (ctxt)
- llog_ctxt_put(ctxt);
+ if (ctxt)
+ llog_ctxt_put(ctxt);
if (cs->cs_buf)
OBD_FREE(cs->cs_buf, KUC_CHANGELOG_MSG_MAXSIZE);
OBD_FREE_PTR(cs);
static int mdc_ioc_changelog_send(struct obd_device *obd,
struct ioc_changelog *icc)
{
- struct changelog_show *cs;
- int rc;
+ struct changelog_show *cs;
+ struct task_struct *task;
+ int rc;
/* Freed in mdc_changelog_send_thread */
OBD_ALLOC_PTR(cs);
* New thread because we should return to user app before
* writing into our pipe
*/
- rc = PTR_ERR(kthread_run(mdc_changelog_send_thread, cs,
- "mdc_clg_send_thread"));
- if (!IS_ERR_VALUE(rc)) {
- CDEBUG(D_CHANGELOG, "start changelog thread\n");
- return 0;
+ task = kthread_run(mdc_changelog_send_thread, cs,
+ "mdc_clg_send_thread");
+ if (IS_ERR(task)) {
+ rc = PTR_ERR(task);
+ CERROR("%s: cannot start changelog thread: rc = %d\n",
+ cs_obd_name(cs), rc);
+ OBD_FREE_PTR(cs);
+ } else {
+ rc = 0;
+ CDEBUG(D_HSM, "%s: started changelog thread\n",
+ cs_obd_name(cs));
}
- CERROR("Failed to start changelog thread: %d\n", rc);
- OBD_FREE_PTR(cs);
- return rc;
+ return rc;
}
static int mdc_ioc_hsm_ct_start(struct obd_export *exp,
static int mdc_ioc_swap_layouts(struct obd_export *exp,
struct md_op_data *op_data)
{
- CFS_LIST_HEAD(cancels);
+ struct list_head cancels = LIST_HEAD_INIT(cancels);
struct ptlrpc_request *req;
int rc, count;
struct mdc_swap_layouts *msl, *payload;
* with the request RPC to avoid extra RPC round trips
*/
count = mdc_resource_get_unused(exp, &op_data->op_fid1, &cancels,
- LCK_CR, MDS_INODELOCK_LAYOUT);
+ LCK_EX, MDS_INODELOCK_LAYOUT |
+ MDS_INODELOCK_XATTR);
count += mdc_resource_get_unused(exp, &op_data->op_fid2, &cancels,
- LCK_CR, MDS_INODELOCK_LAYOUT);
+ LCK_EX, MDS_INODELOCK_LAYOUT |
+ MDS_INODELOCK_XATTR);
req = ptlrpc_request_alloc(class_exp2cliimp(exp),
&RQF_MDS_SWAP_LAYOUTS);
}
static int mdc_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;
ENTRY;
if (!try_module_get(THIS_MODULE)) {
- CERROR("Can't get module. Is it alive?");
+ CERROR("%s: cannot get module '%s'\n", obd->obd_name,
+ module_name(THIS_MODULE));
return -EINVAL;
}
switch (cmd) {
return rc;
}
-int mdc_get_info_rpc(struct obd_export *exp,
- obd_count keylen, void *key,
- int vallen, void *val)
+static int mdc_get_info_rpc(struct obd_export *exp,
+ u32 keylen, void *key,
+ u32 vallen, void *val)
{
struct obd_import *imp = class_exp2cliimp(exp);
struct ptlrpc_request *req;
req_capsule_set_size(&req->rq_pill, &RMF_GETINFO_KEY,
RCL_CLIENT, keylen);
req_capsule_set_size(&req->rq_pill, &RMF_GETINFO_VALLEN,
- RCL_CLIENT, sizeof(__u32));
+ RCL_CLIENT, sizeof(vallen));
rc = ptlrpc_request_pack(req, LUSTRE_MDS_VERSION, MDS_GET_INFO);
if (rc) {
tmp = req_capsule_client_get(&req->rq_pill, &RMF_GETINFO_KEY);
memcpy(tmp, key, keylen);
tmp = req_capsule_client_get(&req->rq_pill, &RMF_GETINFO_VALLEN);
- memcpy(tmp, &vallen, sizeof(__u32));
+ memcpy(tmp, &vallen, sizeof(vallen));
req_capsule_set_size(&req->rq_pill, &RMF_GETINFO_VAL,
RCL_SERVER, vallen);
static void lustre_swab_hal(struct hsm_action_list *h)
{
struct hsm_action_item *hai;
- int i;
+ __u32 i;
__swab32s(&h->hal_version);
__swab32s(&h->hal_count);
* @param val KUC message (kuc_hdr + hsm_action_list)
* @param len total length of message
*/
-static int mdc_hsm_copytool_send(int len, void *val)
+static int mdc_hsm_copytool_send(size_t len, void *val)
{
struct kuc_hdr *lh = (struct kuc_hdr *)val;
struct hsm_action_list *hal = (struct hsm_action_list *)(lh + 1);
ENTRY;
if (len < sizeof(*lh) + sizeof(*hal)) {
- CERROR("Short HSM message %d < %d\n", len,
- (int) (sizeof(*lh) + sizeof(*hal)));
+ CERROR("Short HSM message %zu < %zu\n", len,
+ sizeof(*lh) + sizeof(*hal));
RETURN(-EPROTO);
}
if (lh->kuc_magic == __swab16(KUC_MAGIC)) {
(void *)imp);
}
-int mdc_set_info_async(const struct lu_env *env,
- struct obd_export *exp,
- obd_count keylen, void *key,
- obd_count vallen, void *val,
- struct ptlrpc_request_set *set)
+static int mdc_set_info_async(const struct lu_env *env,
+ struct obd_export *exp,
+ u32 keylen, void *key,
+ u32 vallen, void *val,
+ struct ptlrpc_request_set *set)
{
struct obd_import *imp = class_exp2cliimp(exp);
int rc;
RETURN(rc);
}
+ if (KEY_IS(KEY_DEFAULT_EASIZE)) {
+ __u32 *default_easize = val;
+
+ exp->exp_obd->u.cli.cl_default_mds_easize = *default_easize;
+ RETURN(0);
+ }
+
CERROR("Unknown key %s\n", (char *)key);
RETURN(-EINVAL);
}
-int mdc_get_info(const struct lu_env *env, struct obd_export *exp,
- __u32 keylen, void *key, __u32 *vallen, void *val,
- struct lov_stripe_md *lsm)
+static int mdc_get_info(const struct lu_env *env, struct obd_export *exp,
+ __u32 keylen, void *key, __u32 *vallen, void *val)
{
- int rc = -EINVAL;
+ int rc = -EINVAL;
- if (KEY_IS(KEY_MAX_EASIZE)) {
- int mdsize, *max_easize;
+ if (KEY_IS(KEY_MAX_EASIZE)) {
+ __u32 mdsize, *max_easize;
- if (*vallen != sizeof(int))
- RETURN(-EINVAL);
- mdsize = *(int*)val;
- if (mdsize > exp->exp_obd->u.cli.cl_max_mds_easize)
- exp->exp_obd->u.cli.cl_max_mds_easize = mdsize;
- max_easize = val;
- *max_easize = exp->exp_obd->u.cli.cl_max_mds_easize;
- RETURN(0);
+ if (*vallen != sizeof(int))
+ RETURN(-EINVAL);
+ mdsize = *(__u32 *)val;
+ if (mdsize > exp->exp_obd->u.cli.cl_max_mds_easize)
+ exp->exp_obd->u.cli.cl_max_mds_easize = mdsize;
+ max_easize = val;
+ *max_easize = exp->exp_obd->u.cli.cl_max_mds_easize;
+ RETURN(0);
+ } else if (KEY_IS(KEY_DEFAULT_EASIZE)) {
+ __u32 *default_easize;
+
+ if (*vallen != sizeof(int))
+ RETURN(-EINVAL);
+ default_easize = val;
+ *default_easize = exp->exp_obd->u.cli.cl_default_mds_easize;
+ RETURN(0);
+ } else if (KEY_IS(KEY_MAX_COOKIESIZE)) {
+ __u32 *max_cookiesize;
+
+ if (*vallen != sizeof(int))
+ RETURN(-EINVAL);
+ max_cookiesize = val;
+ *max_cookiesize = exp->exp_obd->u.cli.cl_max_mds_cookiesize;
+ RETURN(0);
+ } else if (KEY_IS(KEY_DEFAULT_COOKIESIZE)) {
+ __u32 *default_cookiesize;
+
+ if (*vallen != sizeof(int))
+ RETURN(-EINVAL);
+ default_cookiesize = val;
+ *default_cookiesize =
+ exp->exp_obd->u.cli.cl_default_mds_cookiesize;
+ RETURN(0);
} else if (KEY_IS(KEY_CONN_DATA)) {
struct obd_import *imp = class_exp2cliimp(exp);
struct obd_connect_data *data = val;
*data = imp->imp_connect_data;
RETURN(0);
} else if (KEY_IS(KEY_TGT_COUNT)) {
- *((int *)val) = 1;
+ *((__u32 *)val) = 1;
RETURN(0);
}
RETURN(rc);
}
-static int mdc_pin(struct obd_export *exp, const struct lu_fid *fid,
- struct obd_capa *oc, struct obd_client_handle *handle,
- int flags)
-{
- struct ptlrpc_request *req;
- struct mdt_body *body;
- int rc;
- ENTRY;
-
- req = ptlrpc_request_alloc(class_exp2cliimp(exp), &RQF_MDS_PIN);
- if (req == NULL)
- RETURN(-ENOMEM);
-
- mdc_set_capa_size(req, &RMF_CAPA1, oc);
-
- rc = ptlrpc_request_pack(req, LUSTRE_MDS_VERSION, MDS_PIN);
- if (rc) {
- ptlrpc_request_free(req);
- RETURN(rc);
- }
-
- mdc_pack_body(req, fid, oc, 0, 0, -1, flags);
-
- ptlrpc_request_set_replen(req);
-
- mdc_get_rpc_lock(exp->exp_obd->u.cli.cl_rpc_lock, NULL);
- rc = ptlrpc_queue_wait(req);
- mdc_put_rpc_lock(exp->exp_obd->u.cli.cl_rpc_lock, NULL);
- if (rc) {
- CERROR("Pin failed: %d\n", rc);
- GOTO(err_out, rc);
- }
-
- body = req_capsule_server_get(&req->rq_pill, &RMF_MDT_BODY);
- if (body == NULL)
- GOTO(err_out, rc = -EPROTO);
-
- handle->och_fh = body->handle;
- handle->och_magic = OBD_CLIENT_HANDLE_MAGIC;
-
- handle->och_mod = obd_mod_alloc();
- if (handle->och_mod == NULL) {
- DEBUG_REQ(D_ERROR, req, "can't allocate md_open_data");
- GOTO(err_out, rc = -ENOMEM);
- }
- handle->och_mod->mod_open_req = req; /* will be dropped by unpin */
-
- RETURN(0);
-
-err_out:
- ptlrpc_req_finished(req);
- RETURN(rc);
-}
-
-static int mdc_unpin(struct obd_export *exp, struct obd_client_handle *handle,
- int flag)
-{
- struct ptlrpc_request *req;
- struct mdt_body *body;
- int rc;
- ENTRY;
-
- req = ptlrpc_request_alloc_pack(class_exp2cliimp(exp), &RQF_MDS_UNPIN,
- LUSTRE_MDS_VERSION, MDS_UNPIN);
- if (req == NULL)
- RETURN(-ENOMEM);
-
- body = req_capsule_client_get(&req->rq_pill, &RMF_MDT_BODY);
- body->handle = handle->och_fh;
- body->flags = flag;
-
- ptlrpc_request_set_replen(req);
-
- mdc_get_rpc_lock(exp->exp_obd->u.cli.cl_rpc_lock, NULL);
- rc = ptlrpc_queue_wait(req);
- mdc_put_rpc_lock(exp->exp_obd->u.cli.cl_rpc_lock, NULL);
-
- if (rc != 0)
- CERROR("Unpin failed: %d\n", rc);
-
- ptlrpc_req_finished(req);
- ptlrpc_req_finished(handle->och_mod->mod_open_req);
-
- obd_mod_put(handle->och_mod);
- RETURN(rc);
-}
-
-int mdc_fsync(struct obd_export *exp, const struct lu_fid *fid,
- struct obd_capa *oc, struct ptlrpc_request **request)
+static int mdc_fsync(struct obd_export *exp, const struct lu_fid *fid,
+ struct obd_capa *oc, struct ptlrpc_request **request)
{
struct ptlrpc_request *req;
int rc;
RETURN(rc);
}
-int mdc_fid_alloc(struct obd_export *exp, struct lu_fid *fid,
- struct md_op_data *op_data)
+int mdc_fid_alloc(const struct lu_env *env, struct obd_export *exp,
+ struct lu_fid *fid, struct md_op_data *op_data)
{
- struct client_obd *cli = &exp->exp_obd->u.cli;
- struct lu_client_seq *seq = cli->cl_seq;
- ENTRY;
- RETURN(seq_client_alloc_fid(NULL, seq, fid));
+ struct client_obd *cli = &exp->exp_obd->u.cli;
+ struct lu_client_seq *seq = cli->cl_seq;
+ ENTRY;
+ RETURN(seq_client_alloc_fid(env, seq, fid));
}
-struct obd_uuid *mdc_get_uuid(struct obd_export *exp) {
+static struct obd_uuid *mdc_get_uuid(struct obd_export *exp)
+{
struct client_obd *cli = &exp->exp_obd->u.cli;
return &cli->cl_target_uuid;
}
* recovery, non zero value will be return if the lock can be canceled,
* or zero returned for not
*/
-static int mdc_cancel_for_recovery(struct ldlm_lock *lock)
+static int mdc_cancel_weight(struct ldlm_lock *lock)
{
- if (lock->l_resource->lr_type != LDLM_IBITS)
- RETURN(0);
+ if (lock->l_resource->lr_type != LDLM_IBITS)
+ RETURN(0);
- /* FIXME: if we ever get into a situation where there are too many
- * opened files with open locks on a single node, then we really
- * should replay these open locks to reget it */
- if (lock->l_policy_data.l_inodebits.bits & MDS_INODELOCK_OPEN)
- RETURN(0);
+ /* FIXME: if we ever get into a situation where there are too many
+ * opened files with open locks on a single node, then we really
+ * should replay these open locks to reget it */
+ if (lock->l_policy_data.l_inodebits.bits & MDS_INODELOCK_OPEN)
+ RETURN(0);
- RETURN(1);
+ RETURN(1);
}
static int mdc_resource_inode_free(struct ldlm_resource *res)
return 0;
}
-struct ldlm_valblock_ops inode_lvbo = {
- lvbo_free: mdc_resource_inode_free
+static struct ldlm_valblock_ops inode_lvbo = {
+ .lvbo_free = mdc_resource_inode_free
};
+static int mdc_llog_init(struct obd_device *obd)
+{
+ struct obd_llog_group *olg = &obd->obd_olg;
+ struct llog_ctxt *ctxt;
+ int rc;
+
+ ENTRY;
+
+ rc = llog_setup(NULL, obd, olg, LLOG_CHANGELOG_REPL_CTXT, obd,
+ &llog_client_ops);
+ if (rc < 0)
+ RETURN(rc);
+
+ ctxt = llog_group_get_ctxt(olg, LLOG_CHANGELOG_REPL_CTXT);
+ llog_initiator_connect(ctxt);
+ llog_ctxt_put(ctxt);
+
+ RETURN(0);
+}
+
+static void mdc_llog_finish(struct obd_device *obd)
+{
+ struct llog_ctxt *ctxt;
+
+ ENTRY;
+
+ ctxt = llog_get_context(obd, LLOG_CHANGELOG_REPL_CTXT);
+ if (ctxt != NULL)
+ llog_cleanup(NULL, ctxt);
+
+ EXIT;
+}
+
static int mdc_setup(struct obd_device *obd, struct lustre_cfg *cfg)
{
struct client_obd *cli = &obd->u.cli;
rc = client_obd_setup(obd, cfg);
if (rc)
GOTO(err_close_lock, rc);
-#ifdef LPROCFS
+#ifdef CONFIG_PROC_FS
obd->obd_vars = lprocfs_mdc_obd_vars;
- lprocfs_seq_obd_setup(obd);
+ lprocfs_obd_setup(obd);
lprocfs_alloc_md_stats(obd, 0);
#endif
sptlrpc_lprocfs_cliobd_attach(obd);
ptlrpc_lprocfs_register_obd(obd);
- ns_register_cancel(obd->obd_namespace, mdc_cancel_for_recovery);
+ ns_register_cancel(obd->obd_namespace, mdc_cancel_weight);
obd->obd_namespace->ns_lvbo = &inode_lvbo;
- rc = obd_llog_init(obd, &obd->obd_olg, obd, NULL);
+ rc = mdc_llog_init(obd);
if (rc) {
mdc_cleanup(obd);
CERROR("failed to setup llogging subsystems\n");
}
/* Initialize the default and maximum LOV EA and cookie sizes. This allows
- * us to make MDS RPCs with large enough reply buffers to hold the
- * maximum-sized (= maximum striped) EA and cookie without having to
- * calculate this (via a call into the LOV + OSCs) each time we make an RPC. */
-static int mdc_init_ea_size(struct obd_export *exp, int easize,
- int def_easize, int cookiesize)
+ * us to make MDS RPCs with large enough reply buffers to hold a default
+ * sized EA and cookie without having to calculate this (via a call into the
+ * LOV + OSCs) each time we make an RPC. The maximum size is also tracked
+ * but not used to avoid wastefully vmalloc()'ing large reply buffers when
+ * a large number of stripes is possible. If a larger reply buffer is
+ * required it will be reallocated in the ptlrpc layer due to overflow.
+ */
+static int mdc_init_ea_size(struct obd_export *exp, __u32 easize,
+ __u32 def_easize, __u32 cookiesize,
+ __u32 def_cookiesize)
{
- struct obd_device *obd = exp->exp_obd;
- struct client_obd *cli = &obd->u.cli;
- ENTRY;
+ struct obd_device *obd = exp->exp_obd;
+ struct client_obd *cli = &obd->u.cli;
+ ENTRY;
- if (cli->cl_max_mds_easize < easize)
- cli->cl_max_mds_easize = easize;
+ if (cli->cl_max_mds_easize < easize)
+ cli->cl_max_mds_easize = easize;
- if (cli->cl_default_mds_easize < def_easize)
- cli->cl_default_mds_easize = def_easize;
+ if (cli->cl_default_mds_easize < def_easize)
+ cli->cl_default_mds_easize = def_easize;
- if (cli->cl_max_mds_cookiesize < cookiesize)
- cli->cl_max_mds_cookiesize = cookiesize;
+ if (cli->cl_max_mds_cookiesize < cookiesize)
+ cli->cl_max_mds_cookiesize = cookiesize;
- RETURN(0);
+ if (cli->cl_default_mds_cookiesize < def_cookiesize)
+ cli->cl_default_mds_cookiesize = def_cookiesize;
+
+ RETURN(0);
}
static int mdc_precleanup(struct obd_device *obd, enum obd_cleanup_stage stage)
ptlrpc_lprocfs_unregister_obd(obd);
lprocfs_obd_cleanup(obd);
lprocfs_free_md_stats(obd);
-
- rc = obd_llog_finish(obd, 0);
- if (rc != 0)
- CERROR("failed to cleanup llogging subsystems\n");
+ mdc_llog_finish(obd);
break;
}
RETURN(rc);
return client_obd_cleanup(obd);
}
-
-static int mdc_llog_init(struct obd_device *obd, struct obd_llog_group *olg,
- struct obd_device *tgt, int *index)
-{
- struct llog_ctxt *ctxt;
- int rc;
-
- ENTRY;
-
- LASSERT(olg == &obd->obd_olg);
-
- rc = llog_setup(NULL, obd, olg, LLOG_CHANGELOG_REPL_CTXT, tgt,
- &llog_client_ops);
- if (rc)
- RETURN(rc);
-
- ctxt = llog_group_get_ctxt(olg, LLOG_CHANGELOG_REPL_CTXT);
- llog_initiator_connect(ctxt);
- llog_ctxt_put(ctxt);
-
- RETURN(0);
-}
-
-static int mdc_llog_finish(struct obd_device *obd, int count)
-{
- struct llog_ctxt *ctxt;
-
- ENTRY;
-
- ctxt = llog_get_context(obd, LLOG_CHANGELOG_REPL_CTXT);
- if (ctxt)
- llog_cleanup(NULL, ctxt);
-
- RETURN(0);
-}
-
-static int mdc_process_config(struct obd_device *obd, obd_count len, void *buf)
+static int mdc_process_config(struct obd_device *obd, size_t len, void *buf)
{
struct lustre_cfg *lcfg = buf;
- int rc = class_process_proc_seq_param(PARAM_MDC, obd->obd_vars,
- lcfg, obd);
+ int rc = class_process_proc_param(PARAM_MDC, obd->obd_vars, lcfg, obd);
return (rc > 0 ? 0: rc);
}
/* get remote permission for current user on fid */
-int mdc_get_remote_perm(struct obd_export *exp, const struct lu_fid *fid,
- struct obd_capa *oc, __u32 suppgid,
- struct ptlrpc_request **request)
+static int mdc_get_remote_perm(struct obd_export *exp, const struct lu_fid *fid,
+ struct obd_capa *oc, __u32 suppgid,
+ struct ptlrpc_request **request)
{
struct ptlrpc_request *req;
int rc;
if (body == NULL)
GOTO(out, capa = ERR_PTR(-EFAULT));
- if ((body->valid & OBD_MD_FLOSSCAPA) == 0)
+ if ((body->mbo_valid & OBD_MD_FLOSSCAPA) == 0)
GOTO(out, capa = ERR_PTR(-ENOENT));
capa = req_capsule_server_get(&req->rq_pill, &RMF_CAPA2);
RETURN(0);
}
-struct obd_ops mdc_obd_ops = {
+static struct obd_ops mdc_obd_ops = {
.o_owner = THIS_MODULE,
.o_setup = mdc_setup,
.o_precleanup = mdc_precleanup,
.o_iocontrol = mdc_iocontrol,
.o_set_info_async = mdc_set_info_async,
.o_statfs = mdc_statfs,
- .o_pin = mdc_pin,
- .o_unpin = mdc_unpin,
.o_fid_init = client_fid_init,
.o_fid_fini = client_fid_fini,
.o_fid_alloc = mdc_fid_alloc,
.o_import_event = mdc_import_event,
- .o_llog_init = mdc_llog_init,
- .o_llog_finish = mdc_llog_finish,
.o_get_info = mdc_get_info,
.o_process_config = mdc_process_config,
.o_get_uuid = mdc_get_uuid,
.o_quotacheck = mdc_quotacheck
};
-struct md_ops mdc_md_ops = {
+static struct md_ops mdc_md_ops = {
.m_getstatus = mdc_getstatus,
.m_null_inode = mdc_null_inode,
.m_find_cbdata = mdc_find_cbdata,
.m_close = mdc_close,
.m_create = mdc_create,
- .m_done_writing = mdc_done_writing,
.m_enqueue = mdc_enqueue,
.m_getattr = mdc_getattr,
.m_getattr_name = mdc_getattr_name,
.m_intent_lock = mdc_intent_lock,
.m_link = mdc_link,
- .m_is_subdir = mdc_is_subdir,
.m_rename = mdc_rename,
.m_setattr = mdc_setattr,
.m_setxattr = mdc_setxattr,
.m_getxattr = mdc_getxattr,
.m_fsync = mdc_fsync,
- .m_read_entry = mdc_read_entry,
+ .m_read_page = mdc_read_page,
.m_unlink = mdc_unlink,
.m_cancel_unused = mdc_cancel_unused,
.m_init_ea_size = mdc_init_ea_size,
.m_revalidate_lock = mdc_revalidate_lock
};
-int __init mdc_init(void)
+static int __init mdc_init(void)
{
- return class_register_type(&mdc_obd_ops, &mdc_md_ops, NULL,
-#ifndef HAVE_ONLY_PROCFS_SEQ
- NULL,
-#endif
- LUSTRE_MDC_NAME, NULL);
+ return class_register_type(&mdc_obd_ops, &mdc_md_ops, true, NULL,
+ LUSTRE_MDC_NAME, NULL);
}
-#ifdef __KERNEL__
static void /*__exit*/ mdc_exit(void)
{
class_unregister_type(LUSTRE_MDC_NAME);
module_init(mdc_init);
module_exit(mdc_exit);
-#endif
git://git.whamcloud.com / fs / lustre-release.git / blobdiff