X-Git-Url: https://git.whamcloud.com/?p=fs%2Flustre-release.git;a=blobdiff_plain;f=lustre%2Fmdc%2Fmdc_request.c;h=9f3dea4ce3343007e16f6df5d18264f2e7d7242d;hp=478929568ef2d1f634488b055e3a7cb4e9eca00f;hb=16e0fe6eab3e54aa7beddb75bf7dae05645408e9;hpb=fbf5870b9848929d352460f1f005b79c0b5ccc5a diff --git a/lustre/mdc/mdc_request.c b/lustre/mdc/mdc_request.c index 4789295..9f3dea4 100644 --- a/lustre/mdc/mdc_request.c +++ b/lustre/mdc/mdc_request.c @@ -1,6 +1,4 @@ -/* -*- mode: c; c-basic-offset: 8; indent-tabs-mode: nil; -*- - * vim:expandtab:shiftwidth=8:tabstop=8: - * +/* * GPL HEADER START * * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER. @@ -26,17 +24,16 @@ * GPL HEADER END */ /* - * Copyright 2008 Sun Microsystems, Inc. All rights reserved + * Copyright (c) 2001, 2010, Oracle and/or its affiliates. All rights reserved. * Use is subject to license terms. + * + * Copyright (c) 2011, 2012, Intel Corporation. */ /* * This file is part of Lustre, http://www.lustre.org/ * Lustre is a trademark of Sun Microsystems, Inc. */ -#ifndef EXPORT_SYMTAB -# define EXPORT_SYMTAB -#endif #define DEBUG_SUBSYSTEM S_MDC #ifdef __KERNEL__ @@ -50,40 +47,61 @@ #include #include -#include #include -#include #include #include -#include "mdc_internal.h" +#include -quota_interface_t *quota_interface; +#include "mdc_internal.h" #define REQUEST_MINOR 244 -extern quota_interface_t mdc_quota_interface; +struct mdc_renew_capa_args { + struct obd_capa *ra_oc; + renew_capa_cb_t ra_cb; +}; static int mdc_cleanup(struct obd_device *obd); -static struct obd_capa *mdc_unpack_capa(struct ptlrpc_request *req, - const struct req_msg_field *field) +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 *oc; + struct obd_capa *c; + ENTRY; /* swabbed already in mdc_enqueue */ capa = req_capsule_server_get(&req->rq_pill, field); if (capa == NULL) - return ERR_PTR(-EPROTO); + RETURN(-EPROTO); - oc = alloc_capa(CAPA_SITE_CLIENT); - if (!oc) { + c = alloc_capa(CAPA_SITE_CLIENT); + if (IS_ERR(c)) { CDEBUG(D_INFO, "alloc capa failed!\n"); - return ERR_PTR(-ENOMEM); + RETURN(PTR_ERR(c)); + } else { + c->c_capa = *capa; + *oc = c; + RETURN(0); } - oc->c_capa = *capa; +} + +static inline int mdc_queue_wait(struct ptlrpc_request *req) +{ + struct client_obd *cli = &req->rq_import->imp_obd->u.cli; + int rc; + + /* mdc_enter_request() 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); + if (rc != 0) + return rc; + + rc = ptlrpc_queue_wait(req); + mdc_exit_request(cli); - return oc; + return rc; } /* Helper that implements most of mdc_getstatus and signal_completed_replay. */ @@ -116,20 +134,16 @@ static int send_getstatus(struct obd_import *imp, struct lu_fid *rootfid, GOTO(out, rc = -EPROTO); if (body->valid & OBD_MD_FLMDSCAPA) { - struct obd_capa *oc; - - oc = mdc_unpack_capa(req, &RMF_CAPA1); - if (IS_ERR(oc)) - GOTO(out, rc = PTR_ERR(oc)); - *pc = oc; + 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", last_xid="LPU64"\n", + "root fid="DFID", last_committed="LPU64"\n", PFID(rootfid), - lustre_msg_get_last_committed(req->rq_repmsg), - lustre_msg_get_last_xid(req->rq_repmsg)); + lustre_msg_get_last_committed(req->rq_repmsg)); EXIT; out: ptlrpc_req_finished(req); @@ -176,21 +190,14 @@ static int mdc_getattr_common(struct obd_export *exp, CDEBUG(D_NET, "mode: %o\n", body->mode); if (body->eadatasize != 0) { + mdc_update_max_ea_from_body(exp, body); + eadata = req_capsule_server_sized_get(pill, &RMF_MDT_MD, body->eadatasize); if (eadata == NULL) RETURN(-EPROTO); } - if (body->valid & OBD_MD_FLMODEASIZE) { - struct client_obd *cli = &exp->exp_obd->u.cli; - - if (cli->cl_max_mds_easize < body->max_mdsize) - cli->cl_max_mds_easize = body->max_mdsize; - if (cli->cl_max_mds_cookiesize < body->max_cookiesize) - cli->cl_max_mds_cookiesize = body->max_cookiesize; - } - if (body->valid & OBD_MD_FLRMTPERM) { struct mdt_remote_perm *perm; @@ -211,8 +218,7 @@ static int mdc_getattr_common(struct obd_export *exp, RETURN(0); } -int mdc_getattr(struct obd_export *exp, const struct lu_fid *fid, - struct obd_capa *oc, obd_valid valid, int ea_size, +int mdc_getattr(struct obd_export *exp, struct md_op_data *op_data, struct ptlrpc_request **request) { struct ptlrpc_request *req; @@ -224,7 +230,7 @@ int mdc_getattr(struct obd_export *exp, const struct lu_fid *fid, if (req == NULL) RETURN(-ENOMEM); - mdc_set_capa_size(req, &RMF_CAPA1, oc); + mdc_set_capa_size(req, &RMF_CAPA1, op_data->op_capa1); rc = ptlrpc_request_pack(req, LUSTRE_MDS_VERSION, MDS_GETATTR); if (rc) { @@ -232,11 +238,12 @@ int mdc_getattr(struct obd_export *exp, const struct lu_fid *fid, RETURN(rc); } - /* MDS_BFLAG_EXT_FLAGS: request "new" flags(bug 9486) */ - mdc_pack_body(req, fid, oc, valid, ea_size, -1, MDS_BFLAG_EXT_FLAGS); + mdc_pack_body(req, &op_data->op_fid1, op_data->op_capa1, + op_data->op_valid, op_data->op_mode, -1, 0); - req_capsule_set_size(&req->rq_pill, &RMF_MDT_MD, RCL_SERVER, ea_size); - if (valid & OBD_MD_FLRMTPERM) { + req_capsule_set_size(&req->rq_pill, &RMF_MDT_MD, RCL_SERVER, + op_data->op_mode); + if (op_data->op_valid & OBD_MD_FLRMTPERM) { LASSERT(client_is_remote(exp)); req_capsule_set_size(&req->rq_pill, &RMF_ACL, RCL_SERVER, sizeof(struct mdt_remote_perm)); @@ -251,9 +258,7 @@ int mdc_getattr(struct obd_export *exp, const struct lu_fid *fid, RETURN(rc); } -int mdc_getattr_name(struct obd_export *exp, const struct lu_fid *fid, - struct obd_capa *oc, const char *filename, int namelen, - obd_valid valid, int ea_size, __u32 suppgid, +int mdc_getattr_name(struct obd_export *exp, struct md_op_data *op_data, struct ptlrpc_request **request) { struct ptlrpc_request *req; @@ -266,8 +271,9 @@ int mdc_getattr_name(struct obd_export *exp, const struct lu_fid *fid, if (req == NULL) RETURN(-ENOMEM); - mdc_set_capa_size(req, &RMF_CAPA1, oc); - req_capsule_set_size(&req->rq_pill, &RMF_NAME, RCL_CLIENT, namelen); + mdc_set_capa_size(req, &RMF_CAPA1, op_data->op_capa1); + req_capsule_set_size(&req->rq_pill, &RMF_NAME, RCL_CLIENT, + op_data->op_namelen + 1); rc = ptlrpc_request_pack(req, LUSTRE_MDS_VERSION, MDS_GETATTR_NAME); if (rc) { @@ -275,17 +281,19 @@ int mdc_getattr_name(struct obd_export *exp, const struct lu_fid *fid, RETURN(rc); } - /* MDS_BFLAG_EXT_FLAGS: request "new" flags(bug 9486) */ - mdc_pack_body(req, fid, oc, valid, ea_size, suppgid, - MDS_BFLAG_EXT_FLAGS); + mdc_pack_body(req, &op_data->op_fid1, op_data->op_capa1, + op_data->op_valid, op_data->op_mode, + op_data->op_suppgids[0], 0); - if (filename) { + if (op_data->op_name) { char *name = req_capsule_client_get(&req->rq_pill, &RMF_NAME); - LASSERT(strnlen(filename, namelen) == namelen - 1); - memcpy(name, filename, namelen); + LASSERT(strnlen(op_data->op_name, op_data->op_namelen) == + op_data->op_namelen); + memcpy(name, op_data->op_name, op_data->op_namelen); } - req_capsule_set_size(&req->rq_pill, &RMF_MDT_MD, RCL_SERVER, ea_size); + req_capsule_set_size(&req->rq_pill, &RMF_MDT_MD, RCL_SERVER, + op_data->op_mode); ptlrpc_request_set_replen(req); rc = mdc_getattr_common(exp, req); @@ -371,8 +379,8 @@ static int mdc_xattr_common(struct obd_export *exp,const struct req_format *fmt, * cfs_curproc_fs{u,g}id() should replace * current->fs{u,g}id for portability. */ - rec->sx_fsuid = current->fsuid; - rec->sx_fsgid = current->fsgid; + rec->sx_fsuid = cfs_curproc_fsuid(); + rec->sx_fsgid = cfs_curproc_fsgid(); rec->sx_cap = cfs_curproc_cap_pack(); rec->sx_suppgid1 = suppgid; rec->sx_suppgid2 = -1; @@ -584,28 +592,34 @@ int mdc_get_lustre_md(struct obd_export *exp, struct ptlrpc_request *req, } } if (md->body->valid & OBD_MD_FLMDSCAPA) { - struct obd_capa *oc = mdc_unpack_capa(req, &RMF_CAPA1); + struct obd_capa *oc = NULL; - if (IS_ERR(oc)) - GOTO(out, rc = PTR_ERR(oc)); + rc = mdc_unpack_capa(NULL, req, &RMF_CAPA1, &oc); + if (rc) + GOTO(out, rc); md->mds_capa = oc; } if (md->body->valid & OBD_MD_FLOSSCAPA) { - struct obd_capa *oc = mdc_unpack_capa(req, &RMF_CAPA2); + struct obd_capa *oc = NULL; - if (IS_ERR(oc)) - GOTO(out, rc = PTR_ERR(oc)); + rc = mdc_unpack_capa(NULL, req, &RMF_CAPA2, &oc); + if (rc) + GOTO(out, rc); md->oss_capa = oc; } EXIT; out: if (rc) { - if (md->oss_capa) - free_capa(md->oss_capa); - if (md->mds_capa) - free_capa(md->mds_capa); + if (md->oss_capa) { + capa_put(md->oss_capa); + md->oss_capa = NULL; + } + if (md->mds_capa) { + capa_put(md->mds_capa); + md->mds_capa = NULL; + } #ifdef CONFIG_FS_POSIX_ACL posix_acl_release(md->posix_acl); #endif @@ -621,10 +635,14 @@ int mdc_free_lustre_md(struct obd_export *exp, struct lustre_md *md) RETURN(0); } -static void mdc_replay_open(struct ptlrpc_request *req) +/** + * 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; - struct ptlrpc_request *cur, *tmp; + struct ptlrpc_request *close_req; struct obd_client_handle *och; struct lustre_handle old; struct mdt_body *body; @@ -652,76 +670,50 @@ static void mdc_replay_open(struct ptlrpc_request *req) old = *file_fh; *file_fh = body->handle; } - list_for_each_entry_safe(cur, tmp, &mod->mod_replay_list, rq_mod_list) { - int opc = lustre_msg_get_opc(cur->rq_reqmsg); - struct mdt_epoch *epoch = NULL; + close_req = mod->mod_close_req; + if (close_req != NULL) { + __u32 opc = lustre_msg_get_opc(close_req->rq_reqmsg); + struct mdt_ioepoch *epoch; - if (opc == MDS_CLOSE || opc == MDS_DONE_WRITING) { - epoch = req_capsule_client_get(&cur->rq_pill, - &RMF_MDT_EPOCH); - LASSERT(epoch); - DEBUG_REQ(D_HA, cur, "updating %s body with new fh", - opc == MDS_CLOSE ? "CLOSE" : "DONE_WRITING"); - } else if (opc == MDS_REINT) { - struct mdt_rec_setattr *rec; - - /* Check this is REINT_SETATTR. */ - rec = req_capsule_client_get(&cur->rq_pill, - &RMF_REC_REINT); - LASSERT(rec && rec->sa_opcode == REINT_SETATTR); - - epoch = req_capsule_client_get(&cur->rq_pill, + LASSERT(opc == MDS_CLOSE || opc == MDS_DONE_WRITING); + epoch = req_capsule_client_get(&close_req->rq_pill, &RMF_MDT_EPOCH); - LASSERT(epoch); - DEBUG_REQ(D_HA, cur, "updating REINT_SETATTR body " - "with new fh"); - } - if (epoch) { - if (och != NULL) - LASSERT(!memcmp(&old, &epoch->handle, - sizeof(old))); - epoch->handle = body->handle; - } + 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; } -void mdc_commit_delayed(struct ptlrpc_request *req) +void mdc_commit_open(struct ptlrpc_request *req) { struct md_open_data *mod = req->rq_cb_data; - struct ptlrpc_request *cur, *tmp; - - DEBUG_REQ(D_HA, req, "req committed"); - if (mod == NULL) return; - req->rq_cb_data = NULL; - req->rq_commit_cb = NULL; - list_del_init(&req->rq_mod_list); - if (req->rq_sequence) { - list_for_each_entry_safe(cur, tmp, &mod->mod_replay_list, - rq_mod_list) { - LASSERT(cur != LP_POISON); - LASSERT(cur->rq_type != LI_POISON); - DEBUG_REQ(D_HA, cur, "req balanced"); - LASSERT(cur->rq_transno != 0); - LASSERT(cur->rq_import == req->rq_import); - - /* We no longer want to preserve this for transno- - * unconditional replay. */ - spin_lock(&cur->rq_lock); - cur->rq_replay = 0; - spin_unlock(&cur->rq_lock); - } - } - - if (list_empty(&mod->mod_replay_list)) { - if (mod->mod_och != NULL) - mod->mod_och->och_mod = NULL; + /** + * No need to touch md_open_data::mod_och, it holds a reference on + * \var mod and will zero references to each other, \var mod will be + * freed after that when md_open_data::mod_och will put the reference. + */ - OBD_FREE_PTR(mod); - } + /** + * Do not let open request to disappear as it still may be needed + * for close rpc to happen (it may happen on evict only, otherwise + * ptlrpc_request::rq_replay does not let mdc_commit_open() to be + * called), just mark this rpc as committed to distinguish these 2 + * cases, see mdc_close() for details. The open request reference will + * be put along with freeing \var mod. + */ + ptlrpc_request_addref(req); + spin_lock(&req->rq_lock); + req->rq_committed = 1; + spin_unlock(&req->rq_lock); + req->rq_cb_data = NULL; + obd_mod_put(mod); } int mdc_set_open_replay_data(struct obd_export *exp, @@ -746,21 +738,29 @@ int mdc_set_open_replay_data(struct obd_export *exp, /* Only if the import is replayable, we set replay_open data */ if (och && imp->imp_replayable) { - OBD_ALLOC_PTR(mod); + mod = obd_mod_alloc(); if (mod == NULL) { DEBUG_REQ(D_ERROR, open_req, "Can't allocate md_open_data"); RETURN(0); } - CFS_INIT_LIST_HEAD(&mod->mod_replay_list); - spin_lock(&open_req->rq_lock); - och->och_mod = mod; - mod->mod_och = och; - open_req->rq_cb_data = mod; - list_add_tail(&open_req->rq_mod_list, &mod->mod_replay_list); - open_req->rq_commit_cb = mdc_commit_delayed; - spin_unlock(&open_req->rq_lock); + /** + * Take a reference on \var mod, to be freed on mdc_close(). + * It protects \var mod from being freed on eviction (commit + * callback is called despite rq_replay flag). + * Another reference for \var och. + */ + obd_mod_get(mod); + obd_mod_get(mod); + + spin_lock(&open_req->rq_lock); + och->och_mod = mod; + mod->mod_och = och; + mod->mod_open_req = open_req; + open_req->rq_cb_data = mod; + open_req->rq_commit_cb = mdc_commit_open; + spin_unlock(&open_req->rq_lock); } rec->cr_fid2 = body->fid1; @@ -783,20 +783,38 @@ int mdc_clear_open_replay_data(struct obd_export *exp, struct md_open_data *mod = och->och_mod; ENTRY; - /* - * Don't free the structure now (it happens in mdc_commit_delayed(), - * after the last request is removed from its replay list), - * but make sure that replay doesn't poke at the och, which is about to - * be freed. - */ + /** + * It is possible to not have \var mod in a case of eviction between + * lookup and ll_file_open(). + **/ + if (mod == NULL) + RETURN(0); + LASSERT(mod != LP_POISON); - if (mod != NULL) - mod->mod_och = NULL; + mod->mod_och = NULL; och->och_mod = NULL; + obd_mod_put(mod); + 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) { @@ -825,10 +843,22 @@ int mdc_close(struct obd_export *exp, struct md_op_data *op_data, ptlrpc_at_set_req_timeout(req); /* Ensure that this close's handle is fixed up during replay. */ - if (likely(mod != NULL)) - list_add_tail(&req->rq_mod_list, &mod->mod_replay_list); - else - CDEBUG(D_HA, "couldn't find open req; expecting close error\n"); + if (likely(mod != NULL)) { + LASSERTF(mod->mod_open_req != NULL && + mod->mod_open_req->rq_type != LI_POISON, + "POISONED open %p!\n", mod->mod_open_req); + + mod->mod_close_req = req; + + DEBUG_REQ(D_HA, mod->mod_open_req, "matched open"); + /* We no longer want to preserve this open 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); + } else { + CDEBUG(D_HA, "couldn't find open req; expecting close error\n"); + } mdc_close_pack(req, op_data); @@ -839,11 +869,6 @@ int mdc_close(struct obd_export *exp, struct md_op_data *op_data, ptlrpc_request_set_replen(req); - req->rq_commit_cb = mdc_commit_delayed; - req->rq_replay = 1; - LASSERT(req->rq_cb_data == NULL); - req->rq_cb_data = mod; - 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); @@ -862,25 +887,34 @@ int mdc_close(struct obd_export *exp, struct md_op_data *op_data, "= %d", rc); if (rc > 0) rc = -rc; - } else if (mod == NULL) { - if (req->rq_import->imp_replayable) - CERROR("Unexpected: can't find md_open_data," - "but close succeeded with replayable imp" - "Please tell " - "http://bugzilla.lustre.org/\n"); } - body = req_capsule_server_get(&req->rq_pill, &RMF_MDT_BODY); if (body == NULL) rc = -EPROTO; + } else if (rc == -ESTALE) { + /** + * it can be allowed error after 3633 if open was committed and + * server failed before close was sent. Let's check if mod + * exists and return no error in that case + */ + if (mod) { + DEBUG_REQ(D_HA, req, "Reset ESTALE = %d", rc); + LASSERT(mod->mod_open_req != NULL); + if (mod->mod_open_req->rq_committed) + rc = 0; + } } - EXIT; - if (rc != 0 && rc != -EAGAIN && req && req->rq_commit_cb) - req->rq_commit_cb(req); - + if (mod) { + if (rc != 0) + mod->mod_close_req = NULL; + /* Since now, mod is accessed through open_req only, + * thus close req does not keep a reference on mod anymore. */ + obd_mod_put(mod); + } *request = req; - return rc; + mdc_close_handle_reply(req, op_data, rc); + RETURN(rc); } int mdc_done_writing(struct obd_export *exp, struct md_op_data *op_data, @@ -903,30 +937,49 @@ int mdc_done_writing(struct obd_export *exp, struct md_op_data *op_data, RETURN(rc); } - /* XXX: add DONE_WRITING request to och -- when Size-on-MDS - * recovery will be ready. */ + 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); - req->rq_replay = 1; - req->rq_cb_data = mod; - req->rq_commit_cb = mdc_commit_delayed; - if (likely(mod != NULL)) - list_add_tail(&req->rq_mod_list, &mod->mod_replay_list); - else - CDEBUG(D_HA, "couldn't find open req; expecting close error\n"); 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); - /* Close the open replay sequence if an error occured or no SOM - * attribute update is needed. */ - if (rc != -EAGAIN) - ptlrpc_close_replay_seq(req); + 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 (rc && rc != -EAGAIN && req->rq_commit_cb) - req->rq_commit_cb(req); + if (mod) { + if (rc != 0) + mod->mod_close_req = NULL; + /* 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); } @@ -956,9 +1009,9 @@ int mdc_sendpage(struct obd_export *exp, const struct lu_fid *fid, req->rq_request_portal = MDS_READPAGE_PORTAL; ptlrpc_at_set_req_timeout(req); - desc = ptlrpc_prep_bulk_imp(req, 1, BULK_GET_SOURCE, MDS_BULK_PORTAL); - if (desc == NULL) - GOTO(out, rc = -ENOMEM); + 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); @@ -966,7 +1019,10 @@ int mdc_sendpage(struct obd_export *exp, const struct lu_fid *fid, ptlrpc_request_set_replen(req); rc = ptlrpc_queue_wait(req); - GOTO(out, rc); + if (rc) + GOTO(out, rc); + + rc = sptlrpc_cli_unwrap_bulk_write(req, req->rq_bulk); out: ptlrpc_req_finished(req); return rc; @@ -974,21 +1030,27 @@ out: EXPORT_SYMBOL(mdc_sendpage); #endif -int mdc_readpage(struct obd_export *exp, const struct lu_fid *fid, - struct obd_capa *oc, __u64 offset, struct page *page, - struct ptlrpc_request **request) +int mdc_readpage(struct obd_export *exp, struct md_op_data *op_data, + struct page **pages, struct ptlrpc_request **request) { struct ptlrpc_request *req; struct ptlrpc_bulk_desc *desc; + int i; + cfs_waitq_t waitq; + int resends = 0; + struct l_wait_info lwi; int rc; ENTRY; *request = NULL; + cfs_waitq_init(&waitq); + +restart_bulk: req = ptlrpc_request_alloc(class_exp2cliimp(exp), &RQF_MDS_READPAGE); if (req == NULL) RETURN(-ENOMEM); - mdc_set_capa_size(req, &RMF_CAPA1, oc); + mdc_set_capa_size(req, &RMF_CAPA1, op_data->op_capa1); rc = ptlrpc_request_pack(req, LUSTRE_MDS_VERSION, MDS_READPAGE); if (rc) { @@ -999,244 +1061,77 @@ int mdc_readpage(struct obd_export *exp, const struct lu_fid *fid, req->rq_request_portal = MDS_READPAGE_PORTAL; ptlrpc_at_set_req_timeout(req); - desc = ptlrpc_prep_bulk_imp(req, 1, BULK_PUT_SINK, MDS_BULK_PORTAL); + desc = ptlrpc_prep_bulk_imp(req, op_data->op_npages, 1, BULK_PUT_SINK, + MDS_BULK_PORTAL); if (desc == NULL) { ptlrpc_request_free(req); RETURN(-ENOMEM); } /* NB req now owns desc and will free it when it gets freed */ - ptlrpc_prep_bulk_page(desc, page, 0, CFS_PAGE_SIZE); - mdc_readdir_pack(req, offset, CFS_PAGE_SIZE, fid, oc); + for (i = 0; i < op_data->op_npages; i++) + ptlrpc_prep_bulk_page_pin(desc, pages[i], 0, CFS_PAGE_SIZE); + + mdc_readdir_pack(req, op_data->op_offset, + CFS_PAGE_SIZE * op_data->op_npages, + &op_data->op_fid1, op_data->op_capa1); ptlrpc_request_set_replen(req); rc = ptlrpc_queue_wait(req); if (rc) { ptlrpc_req_finished(req); - RETURN(rc); - } - - if (req->rq_bulk->bd_nob_transferred != CFS_PAGE_SIZE) { - CERROR("Unexpected # bytes transferred: %d (%ld expected)\n", - req->rq_bulk->bd_nob_transferred, CFS_PAGE_SIZE); - ptlrpc_req_finished(req); - RETURN(-EPROTO); - } - - *request = req; - RETURN(0); -} - -static int mdc_iocontrol(unsigned int cmd, struct obd_export *exp, int len, - void *karg, void *uarg) -{ - struct obd_device *obd = exp->exp_obd; - struct obd_ioctl_data *data = karg; - struct obd_import *imp = obd->u.cli.cl_import; - struct llog_ctxt *ctxt; - int rc; - ENTRY; - - if (!try_module_get(THIS_MODULE)) { - CERROR("Can't get module. Is it alive?"); - return -EINVAL; - } - switch (cmd) { - case OBD_IOC_CLIENT_RECOVER: - rc = ptlrpc_recover_import(imp, data->ioc_inlbuf1); - if (rc < 0) - GOTO(out, rc); - GOTO(out, rc = 0); - case IOC_OSC_SET_ACTIVE: - rc = ptlrpc_set_import_active(imp, data->ioc_offset); - GOTO(out, rc); - case OBD_IOC_PARSE: { - ctxt = llog_get_context(exp->exp_obd, LLOG_CONFIG_REPL_CTXT); - rc = class_config_parse_llog(ctxt, data->ioc_inlbuf1, NULL); - llog_ctxt_put(ctxt); - GOTO(out, rc); - } -#ifdef __KERNEL__ - case OBD_IOC_LLOG_INFO: - case OBD_IOC_LLOG_PRINT: { - ctxt = llog_get_context(obd, LLOG_CONFIG_REPL_CTXT); - rc = llog_ioctl(ctxt, cmd, data); - llog_ctxt_put(ctxt); - GOTO(out, rc); - } -#endif - case OBD_IOC_POLL_QUOTACHECK: - rc = lquota_poll_check(quota_interface, exp, - (struct if_quotacheck *)karg); - GOTO(out, rc); - default: - CERROR("mdc_ioctl(): unrecognised ioctl %#x\n", cmd); - GOTO(out, rc = -ENOTTY); - } -out: - module_put(THIS_MODULE); - - return rc; -} + if (rc != -ETIMEDOUT) + RETURN(rc); -static int do_set_info_async(struct obd_export *exp, - obd_count keylen, void *key, - obd_count vallen, void *val, - struct ptlrpc_request_set *set) -{ - struct obd_import *imp = class_exp2cliimp(exp); - struct ptlrpc_request *req; - char *tmp; - int rc; - ENTRY; - - if (vallen != sizeof(int)) - RETURN(-EINVAL); - - spin_lock(&imp->imp_lock); - if (*((int *)val)) { - imp->imp_connect_flags_orig |= OBD_CONNECT_RDONLY; - imp->imp_connect_data.ocd_connect_flags |= OBD_CONNECT_RDONLY; - } else { - imp->imp_connect_flags_orig &= ~OBD_CONNECT_RDONLY; - imp->imp_connect_data.ocd_connect_flags &= ~OBD_CONNECT_RDONLY; - } - spin_unlock(&imp->imp_lock); - - req = ptlrpc_request_alloc(imp, &RQF_MDS_SET_INFO); - if (req == NULL) - RETURN(-ENOMEM); + resends++; + if (!client_should_resend(resends, &exp->exp_obd->u.cli)) { + CERROR("too many resend retries, returning error\n"); + RETURN(-EIO); + } + lwi = LWI_TIMEOUT_INTR(cfs_time_seconds(resends), NULL, NULL, NULL); + l_wait_event(waitq, 0, &lwi); - req_capsule_set_size(&req->rq_pill, &RMF_SETINFO_KEY, - RCL_CLIENT, keylen); - req_capsule_set_size(&req->rq_pill, &RMF_SETINFO_VAL, - RCL_CLIENT, vallen); - rc = ptlrpc_request_pack(req, LUSTRE_MDS_VERSION, MDS_SET_INFO); - if (rc) { - ptlrpc_request_free(req); - RETURN(rc); + goto restart_bulk; } - tmp = req_capsule_client_get(&req->rq_pill, &RMF_SETINFO_KEY); - memcpy(tmp, key, keylen); - tmp = req_capsule_client_get(&req->rq_pill, &RMF_SETINFO_VAL); - memcpy(tmp, val, vallen); - - ptlrpc_request_set_replen(req); - - if (set) { - ptlrpc_set_add_req(set, req); - ptlrpc_check_set(NULL, set); - } else { - rc = ptlrpc_queue_wait(req); + rc = sptlrpc_cli_unwrap_bulk_read(req, req->rq_bulk, + req->rq_bulk->bd_nob_transferred); + if (rc < 0) { ptlrpc_req_finished(req); - } - - RETURN(rc); -} - -int mdc_set_info_async(struct obd_export *exp, - obd_count keylen, void *key, - obd_count vallen, void *val, - struct ptlrpc_request_set *set) -{ - struct obd_import *imp = class_exp2cliimp(exp); - int rc = -EINVAL; - ENTRY; - - if (KEY_IS(KEY_INIT_RECOV)) { - if (vallen != sizeof(int)) - RETURN(-EINVAL); - spin_lock(&imp->imp_lock); - imp->imp_initial_recov = *(int *)val; - spin_unlock(&imp->imp_lock); - CDEBUG(D_HA, "%s: set imp_initial_recov = %d\n", - exp->exp_obd->obd_name, imp->imp_initial_recov); - RETURN(0); - } - /* Turn off initial_recov after we try all backup servers once */ - if (KEY_IS(KEY_INIT_RECOV_BACKUP)) { - if (vallen != sizeof(int)) - RETURN(-EINVAL); - spin_lock(&imp->imp_lock); - imp->imp_initial_recov_bk = *(int *)val; - if (imp->imp_initial_recov_bk) - imp->imp_initial_recov = 1; - spin_unlock(&imp->imp_lock); - CDEBUG(D_HA, "%s: set imp_initial_recov_bk = %d\n", - exp->exp_obd->obd_name, imp->imp_initial_recov_bk); - RETURN(0); - } - if (KEY_IS(KEY_READ_ONLY)) { - rc = do_set_info_async(exp, keylen, key, vallen, val, set); RETURN(rc); } - if (KEY_IS(KEY_FLUSH_CTX)) { - sptlrpc_import_flush_my_ctx(imp); - RETURN(0); - } - if (KEY_IS(KEY_MDS_CONN)) { - struct obd_import *imp = class_exp2cliimp(exp); - - /* mds-mds import */ - spin_lock(&imp->imp_lock); - imp->imp_server_timeout = 1; - spin_unlock(&imp->imp_lock); - imp->imp_client->cli_request_portal = MDS_MDS_PORTAL; - CDEBUG(D_OTHER|D_WARNING, "%s: timeout / 2\n", exp->exp_obd->obd_name); - RETURN(0); - } - - RETURN(rc); -} - -int mdc_get_info(struct obd_export *exp, __u32 keylen, void *key, - __u32 *vallen, void *val, struct lov_stripe_md *lsm) -{ - int rc = -EINVAL; - - if (KEY_IS(KEY_MAX_EASIZE)) { - int 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 (KEY_IS(KEY_CONN_DATA)) { - struct obd_import *imp = class_exp2cliimp(exp); - struct obd_connect_data *data = val; - if (*vallen != sizeof(*data)) - RETURN(-EINVAL); - - *data = imp->imp_connect_data; - RETURN(0); + if (req->rq_bulk->bd_nob_transferred & ~LU_PAGE_MASK) { + CERROR("Unexpected # bytes transferred: %d (%ld expected)\n", + req->rq_bulk->bd_nob_transferred, + CFS_PAGE_SIZE * op_data->op_npages); + ptlrpc_req_finished(req); + RETURN(-EPROTO); } - RETURN(rc); + *request = req; + RETURN(0); } -static int mdc_statfs(struct obd_device *obd, struct obd_statfs *osfs, +static int mdc_statfs(const struct lu_env *env, + struct obd_export *exp, struct obd_statfs *osfs, __u64 max_age, __u32 flags) { + struct obd_device *obd = class_exp2obd(exp); struct ptlrpc_request *req; struct obd_statfs *msfs; struct obd_import *imp = NULL; int rc; ENTRY; - - /*Since the request might also come from lprocfs, so we need - *sync this with client_disconnect_export Bug15684*/ - down_read(&obd->u.cli.cl_sem); + /* + * Since the request might also come from lprocfs, so we need + * sync this with client_disconnect_export Bug15684 + */ + down_read(&obd->u.cli.cl_sem); if (obd->u.cli.cl_import) imp = class_import_get(obd->u.cli.cl_import); - up_read(&obd->u.cli.cl_sem); + up_read(&obd->u.cli.cl_sem); if (!imp) RETURN(-ENODEV); @@ -1274,54 +1169,1106 @@ output: 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) +static int mdc_ioc_fid2path(struct obd_export *exp, struct getinfo_fid2path *gf) { - struct ptlrpc_request *req; - struct mdt_body *body; - int rc; - ENTRY; + __u32 keylen, vallen; + void *key; + int rc; - req = ptlrpc_request_alloc(class_exp2cliimp(exp), &RQF_MDS_PIN); - if (req == NULL) + if (gf->gf_pathlen > PATH_MAX) + RETURN(-ENAMETOOLONG); + if (gf->gf_pathlen < 2) + RETURN(-EOVERFLOW); + + /* Key is KEY_FID2PATH + getinfo_fid2path description */ + keylen = cfs_size_round(sizeof(KEY_FID2PATH)) + sizeof(*gf); + OBD_ALLOC(key, keylen); + if (key == NULL) RETURN(-ENOMEM); + memcpy(key, KEY_FID2PATH, sizeof(KEY_FID2PATH)); + memcpy(key + cfs_size_round(sizeof(KEY_FID2PATH)), gf, sizeof(*gf)); - mdc_set_capa_size(req, &RMF_CAPA1, oc); + CDEBUG(D_IOCTL, "path get "DFID" from "LPU64" #%d\n", + PFID(&gf->gf_fid), gf->gf_recno, gf->gf_linkno); - rc = ptlrpc_request_pack(req, LUSTRE_MDS_VERSION, MDS_PIN); - if (rc) { - ptlrpc_request_free(req); - RETURN(rc); - } + if (!fid_is_sane(&gf->gf_fid)) + GOTO(out, rc = -EINVAL); - mdc_pack_body(req, fid, oc, 0, 0, -1, flags); + /* Val is struct getinfo_fid2path result plus path */ + vallen = sizeof(*gf) + gf->gf_pathlen; - ptlrpc_request_set_replen(req); + rc = obd_get_info(NULL, exp, keylen, key, &vallen, gf, NULL); + if (rc != 0 && rc != -EREMOTE) + 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); - if (rc) { - CERROR("Pin failed: %d\n", rc); - GOTO(err_out, rc); - } + if (vallen <= sizeof(*gf)) + GOTO(out, rc = -EPROTO); + else if (vallen > sizeof(*gf) + gf->gf_pathlen) + GOTO(out, rc = -EOVERFLOW); - body = req_capsule_server_get(&req->rq_pill, &RMF_MDT_BODY); - if (body == NULL) + CDEBUG(D_IOCTL, "path get "DFID" from "LPU64" #%d\n%s\n", + PFID(&gf->gf_fid), gf->gf_recno, gf->gf_linkno, gf->gf_path); + +out: + OBD_FREE(key, keylen); + return rc; +} + +static int mdc_ioc_hsm_progress(struct obd_export *exp, + struct hsm_progress_kernel *hpk) +{ + struct obd_import *imp = class_exp2cliimp(exp); + struct hsm_progress_kernel *req_hpk; + struct ptlrpc_request *req; + int rc; + ENTRY; + + req = ptlrpc_request_alloc_pack(imp, &RQF_MDS_HSM_PROGRESS, + LUSTRE_MDS_VERSION, MDS_HSM_PROGRESS); + if (req == NULL) + GOTO(out, rc = -ENOMEM); + + mdc_pack_body(req, NULL, NULL, OBD_MD_FLRMTPERM, 0, 0, 0); + + /* Copy hsm_progress struct */ + req_hpk = req_capsule_client_get(&req->rq_pill, &RMF_MDS_HSM_PROGRESS); + if (req_hpk == NULL) + GOTO(out, rc = -EPROTO); + + *req_hpk = *hpk; + + ptlrpc_request_set_replen(req); + + rc = mdc_queue_wait(req); + GOTO(out, rc); +out: + ptlrpc_req_finished(req); + return rc; +} + +static int mdc_ioc_hsm_ct_register(struct obd_import *imp, __u32 archives) +{ + __u32 *archive_mask; + struct ptlrpc_request *req; + int rc; + ENTRY; + + req = ptlrpc_request_alloc_pack(imp, &RQF_MDS_HSM_CT_REGISTER, + LUSTRE_MDS_VERSION, + MDS_HSM_CT_REGISTER); + if (req == NULL) + GOTO(out, rc = -ENOMEM); + + mdc_pack_body(req, NULL, NULL, OBD_MD_FLRMTPERM, 0, 0, 0); + + /* Copy hsm_progress struct */ + archive_mask = req_capsule_client_get(&req->rq_pill, + &RMF_MDS_HSM_ARCHIVE); + if (archive_mask == NULL) + GOTO(out, rc = -EPROTO); + + *archive_mask = archives; + + ptlrpc_request_set_replen(req); + + rc = mdc_queue_wait(req); + GOTO(out, rc); +out: + ptlrpc_req_finished(req); + return rc; +} + +static int mdc_ioc_hsm_current_action(struct obd_export *exp, + struct md_op_data *op_data) +{ + struct hsm_current_action *hca = op_data->op_data; + struct hsm_current_action *req_hca; + struct ptlrpc_request *req; + int rc; + ENTRY; + + req = ptlrpc_request_alloc(class_exp2cliimp(exp), + &RQF_MDS_HSM_ACTION); + 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_HSM_ACTION); + if (rc) { + ptlrpc_request_free(req); + RETURN(rc); + } + + mdc_pack_body(req, &op_data->op_fid1, op_data->op_capa1, + OBD_MD_FLRMTPERM, 0, op_data->op_suppgids[0], 0); + + ptlrpc_request_set_replen(req); + + rc = mdc_queue_wait(req); + if (rc) + GOTO(out, rc); + + req_hca = req_capsule_server_get(&req->rq_pill, + &RMF_MDS_HSM_CURRENT_ACTION); + if (req_hca == NULL) + GOTO(out, rc = -EPROTO); + + *hca = *req_hca; + + EXIT; +out: + ptlrpc_req_finished(req); + return rc; +} + +static int mdc_ioc_hsm_ct_unregister(struct obd_import *imp) +{ + struct ptlrpc_request *req; + int rc; + ENTRY; + + req = ptlrpc_request_alloc_pack(imp, &RQF_MDS_HSM_CT_UNREGISTER, + LUSTRE_MDS_VERSION, + MDS_HSM_CT_UNREGISTER); + if (req == NULL) + GOTO(out, rc = -ENOMEM); + + mdc_pack_body(req, NULL, NULL, OBD_MD_FLRMTPERM, 0, 0, 0); + + ptlrpc_request_set_replen(req); + + rc = mdc_queue_wait(req); + GOTO(out, rc); +out: + ptlrpc_req_finished(req); + return rc; +} + +static int mdc_ioc_hsm_state_get(struct obd_export *exp, + struct md_op_data *op_data) +{ + struct hsm_user_state *hus = op_data->op_data; + struct hsm_user_state *req_hus; + struct ptlrpc_request *req; + int rc; + ENTRY; + + req = ptlrpc_request_alloc(class_exp2cliimp(exp), + &RQF_MDS_HSM_STATE_GET); + 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_HSM_STATE_GET); + if (rc != 0) { + ptlrpc_request_free(req); + RETURN(rc); + } + + mdc_pack_body(req, &op_data->op_fid1, op_data->op_capa1, + OBD_MD_FLRMTPERM, 0, op_data->op_suppgids[0], 0); + + ptlrpc_request_set_replen(req); + + rc = mdc_queue_wait(req); + if (rc) + GOTO(out, rc); + + req_hus = req_capsule_server_get(&req->rq_pill, &RMF_HSM_USER_STATE); + if (req_hus == NULL) + GOTO(out, rc = -EPROTO); + + *hus = *req_hus; + + EXIT; +out: + ptlrpc_req_finished(req); + return rc; +} + +static int mdc_ioc_hsm_state_set(struct obd_export *exp, + struct md_op_data *op_data) +{ + struct hsm_state_set *hss = op_data->op_data; + struct hsm_state_set *req_hss; + struct ptlrpc_request *req; + int rc; + ENTRY; + + req = ptlrpc_request_alloc(class_exp2cliimp(exp), + &RQF_MDS_HSM_STATE_SET); + 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_HSM_STATE_SET); + if (rc) { + ptlrpc_request_free(req); + RETURN(rc); + } + + mdc_pack_body(req, &op_data->op_fid1, op_data->op_capa1, + OBD_MD_FLRMTPERM, 0, op_data->op_suppgids[0], 0); + + /* Copy states */ + req_hss = req_capsule_client_get(&req->rq_pill, &RMF_HSM_STATE_SET); + if (req_hss == NULL) + GOTO(out, rc = -EPROTO); + *req_hss = *hss; + + ptlrpc_request_set_replen(req); + + rc = mdc_queue_wait(req); + GOTO(out, rc); + + EXIT; +out: + ptlrpc_req_finished(req); + return rc; +} + +static int mdc_ioc_hsm_request(struct obd_export *exp, + struct hsm_user_request *hur) +{ + struct obd_import *imp = class_exp2cliimp(exp); + struct ptlrpc_request *req; + struct hsm_request *req_hr; + struct hsm_user_item *req_hui; + char *req_opaque; + int rc; + ENTRY; + + req = ptlrpc_request_alloc(imp, &RQF_MDS_HSM_REQUEST); + if (req == NULL) + GOTO(out, rc = -ENOMEM); + + req_capsule_set_size(&req->rq_pill, &RMF_MDS_HSM_USER_ITEM, RCL_CLIENT, + hur->hur_request.hr_itemcount + * sizeof(struct hsm_user_item)); + req_capsule_set_size(&req->rq_pill, &RMF_GENERIC_DATA, RCL_CLIENT, + hur->hur_request.hr_data_len); + + rc = ptlrpc_request_pack(req, LUSTRE_MDS_VERSION, MDS_HSM_REQUEST); + if (rc) { + ptlrpc_request_free(req); + RETURN(rc); + } + + mdc_pack_body(req, NULL, NULL, OBD_MD_FLRMTPERM, 0, 0, 0); + + /* Copy hsm_request struct */ + req_hr = req_capsule_client_get(&req->rq_pill, &RMF_MDS_HSM_REQUEST); + if (req_hr == NULL) + GOTO(out, rc = -EPROTO); + *req_hr = hur->hur_request; + + /* Copy hsm_user_item structs */ + req_hui = req_capsule_client_get(&req->rq_pill, &RMF_MDS_HSM_USER_ITEM); + if (req_hui == NULL) + GOTO(out, rc = -EPROTO); + memcpy(req_hui, hur->hur_user_item, + hur->hur_request.hr_itemcount * sizeof(struct hsm_user_item)); + + /* Copy opaque field */ + req_opaque = req_capsule_client_get(&req->rq_pill, &RMF_GENERIC_DATA); + if (req_opaque == NULL) + GOTO(out, rc = -EPROTO); + memcpy(req_opaque, hur_data(hur), hur->hur_request.hr_data_len); + + ptlrpc_request_set_replen(req); + + rc = mdc_queue_wait(req); + GOTO(out, rc); + +out: + ptlrpc_req_finished(req); + return rc; +} + +static struct kuc_hdr *changelog_kuc_hdr(char *buf, int len, int flags) +{ + struct kuc_hdr *lh = (struct kuc_hdr *)buf; + + LASSERT(len <= CR_MAXSIZE); + + lh->kuc_magic = KUC_MAGIC; + lh->kuc_transport = KUC_TRANSPORT_CHANGELOG; + lh->kuc_flags = flags; + lh->kuc_msgtype = CL_RECORD; + lh->kuc_msglen = len; + 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; +}; + +static int changelog_show_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; + ENTRY; + + if ((rec->cr_hdr.lrh_type != CHANGELOG_REC) || + (rec->cr.cr_type >= CL_LAST)) { + CERROR("Not a changelog rec %d/%d\n", rec->cr_hdr.lrh_type, + rec->cr.cr_type); + RETURN(-EINVAL); + } + + 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", + 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)); + + len = sizeof(*lh) + changelog_rec_size(&rec->cr) + rec->cr.cr_namelen; + + /* Set up the message */ + lh = changelog_kuc_hdr(cs->cs_buf, len, cs->cs_flags); + 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); + + 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; + + CDEBUG(D_CHANGELOG, "changelog to fp=%p start "LPU64"\n", + cs->cs_fp, cs->cs_startrec); + + /* + * It's important to daemonize here to close unused FDs. + * The write fd from pipe is already opened by the caller, + * so it's fine to clear all files here + */ + cfs_daemonize("mdc_clg_send_thread"); + + OBD_ALLOC(cs->cs_buf, CR_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); + 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); + GOTO(out, rc); + } + rc = llog_init_handle(NULL, llh, LLOG_F_IS_CAT, NULL); + if (rc) { + CERROR("llog_init_handle failed %d\n", rc); + GOTO(out, rc); + } + + rc = llog_cat_process(NULL, llh, changelog_show_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); + } + +out: + fput(cs->cs_fp); + if (llh) + llog_cat_close(NULL, llh); + if (ctxt) + llog_ctxt_put(ctxt); + if (cs->cs_buf) + OBD_FREE(cs->cs_buf, CR_MAXSIZE); + OBD_FREE_PTR(cs); + /* detach from parent process so we get cleaned up */ + cfs_daemonize("cl_send"); + return rc; +} + +static int mdc_ioc_changelog_send(struct obd_device *obd, + struct ioc_changelog *icc) +{ + struct changelog_show *cs; + int rc; + + /* Freed in mdc_changelog_send_thread */ + OBD_ALLOC_PTR(cs); + if (!cs) + return -ENOMEM; + + cs->cs_obd = obd; + cs->cs_startrec = icc->icc_recno; + /* matching fput in mdc_changelog_send_thread */ + cs->cs_fp = fget(icc->icc_id); + cs->cs_flags = icc->icc_flags; + + /* New thread because we should return to user app before + writing into our pipe */ + rc = cfs_create_thread(mdc_changelog_send_thread, cs, CFS_DAEMON_FLAGS); + if (rc >= 0) { + CDEBUG(D_CHANGELOG, "start changelog thread: %d\n", rc); + return 0; + } + + CERROR("Failed to start changelog thread: %d\n", rc); + OBD_FREE_PTR(cs); + return rc; +} + +static int mdc_ioc_hsm_ct_start(struct obd_export *exp, + struct lustre_kernelcomm *lk); + +static int mdc_quotacheck(struct obd_device *unused, struct obd_export *exp, + struct obd_quotactl *oqctl) +{ + struct client_obd *cli = &exp->exp_obd->u.cli; + struct ptlrpc_request *req; + struct obd_quotactl *body; + int rc; + ENTRY; + + req = ptlrpc_request_alloc_pack(class_exp2cliimp(exp), + &RQF_MDS_QUOTACHECK, LUSTRE_MDS_VERSION, + MDS_QUOTACHECK); + if (req == NULL) + RETURN(-ENOMEM); + + body = req_capsule_client_get(&req->rq_pill, &RMF_OBD_QUOTACTL); + *body = *oqctl; + + ptlrpc_request_set_replen(req); + + /* the next poll will find -ENODATA, that means quotacheck is + * going on */ + cli->cl_qchk_stat = -ENODATA; + rc = ptlrpc_queue_wait(req); + if (rc) + cli->cl_qchk_stat = rc; + ptlrpc_req_finished(req); + RETURN(rc); +} + +static int mdc_quota_poll_check(struct obd_export *exp, + struct if_quotacheck *qchk) +{ + struct client_obd *cli = &exp->exp_obd->u.cli; + int rc; + ENTRY; + + qchk->obd_uuid = cli->cl_target_uuid; + memcpy(qchk->obd_type, LUSTRE_MDS_NAME, strlen(LUSTRE_MDS_NAME)); + + rc = cli->cl_qchk_stat; + /* the client is not the previous one */ + if (rc == CL_NOT_QUOTACHECKED) + rc = -EINTR; + RETURN(rc); +} + +static int mdc_quotactl(struct obd_device *unused, struct obd_export *exp, + struct obd_quotactl *oqctl) +{ + struct ptlrpc_request *req; + struct obd_quotactl *oqc; + int rc; + ENTRY; + + req = ptlrpc_request_alloc_pack(class_exp2cliimp(exp), + &RQF_MDS_QUOTACTL, LUSTRE_MDS_VERSION, + MDS_QUOTACTL); + if (req == NULL) + RETURN(-ENOMEM); + + oqc = req_capsule_client_get(&req->rq_pill, &RMF_OBD_QUOTACTL); + *oqc = *oqctl; + + ptlrpc_request_set_replen(req); + ptlrpc_at_set_req_timeout(req); + req->rq_no_resend = 1; + + rc = ptlrpc_queue_wait(req); + if (rc) + CERROR("ptlrpc_queue_wait failed, rc: %d\n", rc); + + if (req->rq_repmsg && + (oqc = req_capsule_server_get(&req->rq_pill, &RMF_OBD_QUOTACTL))) { + *oqctl = *oqc; + } else if (!rc) { + CERROR ("Can't unpack obd_quotactl\n"); + rc = -EPROTO; + } + ptlrpc_req_finished(req); + + RETURN(rc); +} + +static int mdc_ioc_swap_layouts(struct obd_export *exp, + struct md_op_data *op_data) +{ + CFS_LIST_HEAD(cancels); + struct ptlrpc_request *req; + int rc, count; + struct mdc_swap_layouts *msl, *payload; + ENTRY; + + msl = op_data->op_data; + + /* When the MDT will get the MDS_SWAP_LAYOUTS RPC the + * first thing it will do is to cancel the 2 layout + * locks hold by this client. + * So the client must cancel its layout locks on the 2 fids + * 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); + count += mdc_resource_get_unused(exp, &op_data->op_fid2, &cancels, + LCK_CR, MDS_INODELOCK_LAYOUT); + + req = ptlrpc_request_alloc(class_exp2cliimp(exp), + &RQF_MDS_SWAP_LAYOUTS); + if (req == NULL) { + ldlm_lock_list_put(&cancels, l_bl_ast, count); + RETURN(-ENOMEM); + } + + mdc_set_capa_size(req, &RMF_CAPA1, op_data->op_capa1); + mdc_set_capa_size(req, &RMF_CAPA2, op_data->op_capa2); + + rc = mdc_prep_elc_req(exp, req, MDS_SWAP_LAYOUTS, &cancels, count); + if (rc) { + ptlrpc_request_free(req); + RETURN(rc); + } + + mdc_swap_layouts_pack(req, op_data); + + payload = req_capsule_client_get(&req->rq_pill, &RMF_SWAP_LAYOUTS); + LASSERT(payload); + + *payload = *msl; + + ptlrpc_request_set_replen(req); + + rc = ptlrpc_queue_wait(req); + if (rc) + GOTO(out, rc); + EXIT; + +out: + ptlrpc_req_finished(req); + return rc; +} + +static int mdc_iocontrol(unsigned int cmd, struct obd_export *exp, int len, + void *karg, void *uarg) +{ + struct obd_device *obd = exp->exp_obd; + struct obd_ioctl_data *data = karg; + struct obd_import *imp = obd->u.cli.cl_import; + struct llog_ctxt *ctxt; + int rc; + ENTRY; + + if (!cfs_try_module_get(THIS_MODULE)) { + CERROR("Can't get module. Is it alive?"); + return -EINVAL; + } + switch (cmd) { + case OBD_IOC_CHANGELOG_SEND: + rc = mdc_ioc_changelog_send(obd, karg); + GOTO(out, rc); + case OBD_IOC_CHANGELOG_CLEAR: { + struct ioc_changelog *icc = karg; + struct changelog_setinfo cs = + {.cs_recno = icc->icc_recno, .cs_id = icc->icc_id}; + rc = obd_set_info_async(NULL, exp, strlen(KEY_CHANGELOG_CLEAR), + KEY_CHANGELOG_CLEAR, sizeof(cs), &cs, + NULL); + GOTO(out, rc); + } + case OBD_IOC_FID2PATH: + rc = mdc_ioc_fid2path(exp, karg); + GOTO(out, rc); + case LL_IOC_HSM_CT_START: + rc = mdc_ioc_hsm_ct_start(exp, karg); + GOTO(out, rc); + case LL_IOC_HSM_PROGRESS: + rc = mdc_ioc_hsm_progress(exp, karg); + GOTO(out, rc); + case LL_IOC_HSM_STATE_GET: + rc = mdc_ioc_hsm_state_get(exp, karg); + GOTO(out, rc); + case LL_IOC_HSM_STATE_SET: + rc = mdc_ioc_hsm_state_set(exp, karg); + case LL_IOC_HSM_ACTION: + rc = mdc_ioc_hsm_current_action(exp, karg); + GOTO(out, rc); + case LL_IOC_HSM_REQUEST: + rc = mdc_ioc_hsm_request(exp, karg); + GOTO(out, rc); + case OBD_IOC_CLIENT_RECOVER: + rc = ptlrpc_recover_import(imp, data->ioc_inlbuf1, 0); + if (rc < 0) + GOTO(out, rc); + GOTO(out, rc = 0); + case IOC_OSC_SET_ACTIVE: + rc = ptlrpc_set_import_active(imp, data->ioc_offset); + GOTO(out, rc); + case OBD_IOC_PARSE: { + ctxt = llog_get_context(exp->exp_obd, LLOG_CONFIG_REPL_CTXT); + rc = class_config_parse_llog(NULL, ctxt, data->ioc_inlbuf1, + NULL); + llog_ctxt_put(ctxt); + GOTO(out, rc); + } +#ifdef __KERNEL__ + case OBD_IOC_LLOG_INFO: + case OBD_IOC_LLOG_PRINT: { + ctxt = llog_get_context(obd, LLOG_CONFIG_REPL_CTXT); + rc = llog_ioctl(NULL, ctxt, cmd, data); + llog_ctxt_put(ctxt); + GOTO(out, rc); + } +#endif + case OBD_IOC_POLL_QUOTACHECK: + rc = mdc_quota_poll_check(exp, (struct if_quotacheck *)karg); + GOTO(out, rc); + case OBD_IOC_PING_TARGET: + rc = ptlrpc_obd_ping(obd); + GOTO(out, rc); + /* + * Normally IOC_OBD_STATFS, OBD_IOC_QUOTACTL iocontrol are handled by + * LMV instead of MDC. But when the cluster is upgraded from 1.8, + * there'd be no LMV layer thus we might be called here. Eventually + * this code should be removed. + * bz20731, LU-592. + */ + case IOC_OBD_STATFS: { + struct obd_statfs stat_buf = {0}; + + if (*((__u32 *) data->ioc_inlbuf2) != 0) + GOTO(out, rc = -ENODEV); + + /* copy UUID */ + if (cfs_copy_to_user(data->ioc_pbuf2, obd2cli_tgt(obd), + min((int) data->ioc_plen2, + (int) sizeof(struct obd_uuid)))) + GOTO(out, rc = -EFAULT); + + rc = mdc_statfs(NULL, obd->obd_self_export, &stat_buf, + cfs_time_shift_64(-OBD_STATFS_CACHE_SECONDS), + 0); + if (rc != 0) + GOTO(out, rc); + + if (cfs_copy_to_user(data->ioc_pbuf1, &stat_buf, + min((int) data->ioc_plen1, + (int) sizeof(stat_buf)))) + GOTO(out, rc = -EFAULT); + + GOTO(out, rc = 0); + } + case OBD_IOC_QUOTACTL: { + struct if_quotactl *qctl = karg; + struct obd_quotactl *oqctl; + + OBD_ALLOC_PTR(oqctl); + if (!oqctl) + RETURN(-ENOMEM); + + QCTL_COPY(oqctl, qctl); + rc = obd_quotactl(exp, oqctl); + if (rc == 0) { + QCTL_COPY(qctl, oqctl); + qctl->qc_valid = QC_MDTIDX; + qctl->obd_uuid = obd->u.cli.cl_target_uuid; + } + OBD_FREE_PTR(oqctl); + break; + } + case LL_IOC_GET_CONNECT_FLAGS: { + if (cfs_copy_to_user(uarg, + exp_connect_flags_ptr(exp), + sizeof(__u64))) + GOTO(out, rc = -EFAULT); + else + GOTO(out, rc = 0); + } + case LL_IOC_LOV_SWAP_LAYOUTS: { + rc = mdc_ioc_swap_layouts(exp, karg); + break; + } + default: + CERROR("mdc_ioctl(): unrecognised ioctl %#x\n", cmd); + GOTO(out, rc = -ENOTTY); + } +out: + cfs_module_put(THIS_MODULE); + + return rc; +} + +int mdc_get_info_rpc(struct obd_export *exp, + obd_count keylen, void *key, + int vallen, void *val) +{ + struct obd_import *imp = class_exp2cliimp(exp); + struct ptlrpc_request *req; + char *tmp; + int rc = -EINVAL; + ENTRY; + + req = ptlrpc_request_alloc(imp, &RQF_MDS_GET_INFO); + if (req == NULL) + RETURN(-ENOMEM); + + 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)); + + rc = ptlrpc_request_pack(req, LUSTRE_MDS_VERSION, MDS_GET_INFO); + if (rc) { + ptlrpc_request_free(req); + RETURN(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)); + + req_capsule_set_size(&req->rq_pill, &RMF_GETINFO_VAL, + RCL_SERVER, vallen); + ptlrpc_request_set_replen(req); + + rc = ptlrpc_queue_wait(req); + /* -EREMOTE means the get_info result is partial, and it needs to + * continue on another MDT, see fid2path part in lmv_iocontrol */ + if (rc == 0 || rc == -EREMOTE) { + tmp = req_capsule_server_get(&req->rq_pill, &RMF_GETINFO_VAL); + memcpy(val, tmp, vallen); + if (ptlrpc_rep_need_swab(req)) { + if (KEY_IS(KEY_FID2PATH)) + lustre_swab_fid2path(val); + } + } + ptlrpc_req_finished(req); + + RETURN(rc); +} + +static void lustre_swab_hai(struct hsm_action_item *h) +{ + __swab32s(&h->hai_len); + __swab32s(&h->hai_action); + lustre_swab_lu_fid(&h->hai_fid); + lustre_swab_lu_fid(&h->hai_dfid); + __swab64s(&h->hai_cookie); + __swab64s(&h->hai_extent.offset); + __swab64s(&h->hai_extent.length); + __swab64s(&h->hai_gid); +} + +static void lustre_swab_hal(struct hsm_action_list *h) +{ + struct hsm_action_item *hai; + int i; + + __swab32s(&h->hal_version); + __swab32s(&h->hal_count); + __swab32s(&h->hal_archive_id); + __swab64s(&h->hal_flags); + hai = hai_zero(h); + for (i = 0; i < h->hal_count; i++) { + lustre_swab_hai(hai); + hai = hai_next(hai); + } +} + +static void lustre_swab_kuch(struct kuc_hdr *l) +{ + __swab16s(&l->kuc_magic); + /* __u8 l->kuc_transport */ + __swab16s(&l->kuc_msgtype); + __swab16s(&l->kuc_msglen); +} + +static int mdc_ioc_hsm_ct_start(struct obd_export *exp, + struct lustre_kernelcomm *lk) +{ + struct obd_import *imp = class_exp2cliimp(exp); + __u32 archive = lk->lk_data; + int rc = 0; + + if (lk->lk_group != KUC_GRP_HSM) { + CERROR("Bad copytool group %d\n", lk->lk_group); + return -EINVAL; + } + + CDEBUG(D_HSM, "CT start r%d w%d u%d g%d f%#x\n", lk->lk_rfd, lk->lk_wfd, + lk->lk_uid, lk->lk_group, lk->lk_flags); + + if (lk->lk_flags & LK_FLG_STOP) { + rc = libcfs_kkuc_group_rem(lk->lk_uid, lk->lk_group); + /* Unregister with the coordinator */ + if (rc == 0) + rc = mdc_ioc_hsm_ct_unregister(imp); + } else { + struct file *fp = fget(lk->lk_wfd); + + rc = libcfs_kkuc_group_add(fp, lk->lk_uid, lk->lk_group, + lk->lk_data); + if (rc && fp) + fput(fp); + if (rc == 0) + rc = mdc_ioc_hsm_ct_register(imp, archive); + } + + return rc; +} + +/** + * Send a message to any listening copytools + * @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) +{ + struct kuc_hdr *lh = (struct kuc_hdr *)val; + struct hsm_action_list *hal = (struct hsm_action_list *)(lh + 1); + int rc; + ENTRY; + + if (len < sizeof(*lh) + sizeof(*hal)) { + CERROR("Short HSM message %d < %d\n", len, + (int) (sizeof(*lh) + sizeof(*hal))); + RETURN(-EPROTO); + } + if (lh->kuc_magic == __swab16(KUC_MAGIC)) { + lustre_swab_kuch(lh); + lustre_swab_hal(hal); + } else if (lh->kuc_magic != KUC_MAGIC) { + CERROR("Bad magic %x!=%x\n", lh->kuc_magic, KUC_MAGIC); + RETURN(-EPROTO); + } + + CDEBUG(D_HSM, " Received message mg=%x t=%d m=%d l=%d actions=%d " + "on %s\n", + lh->kuc_magic, lh->kuc_transport, lh->kuc_msgtype, + lh->kuc_msglen, hal->hal_count, hal->hal_fsname); + + /* Broadcast to HSM listeners */ + rc = libcfs_kkuc_group_put(KUC_GRP_HSM, lh); + + RETURN(rc); +} + +/** + * callback function passed to kuc for re-registering each HSM copytool + * running on MDC, after MDT shutdown/recovery. + * @param data archive id served by the copytool + * @param cb_arg callback argument (obd_import) + */ +static int mdc_hsm_ct_reregister(__u32 data, void *cb_arg) +{ + struct obd_import *imp = (struct obd_import *)cb_arg; + __u32 archive = data; + int rc; + + CDEBUG(D_HA, "recover copytool registration to MDT (archive=%#x)\n", + archive); + rc = mdc_ioc_hsm_ct_register(imp, archive); + + /* ignore error if the copytool is already registered */ + return ((rc != 0) && (rc != -EEXIST)) ? rc : 0; +} + +/** + * Re-establish all kuc contexts with MDT + * after MDT shutdown/recovery. + */ +static int mdc_kuc_reregister(struct obd_import *imp) +{ + /* re-register HSM agents */ + return libcfs_kkuc_group_foreach(KUC_GRP_HSM, mdc_hsm_ct_reregister, + (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) +{ + struct obd_import *imp = class_exp2cliimp(exp); + int rc; + ENTRY; + + if (KEY_IS(KEY_READ_ONLY)) { + if (vallen != sizeof(int)) + RETURN(-EINVAL); + + spin_lock(&imp->imp_lock); + if (*((int *)val)) { + imp->imp_connect_flags_orig |= OBD_CONNECT_RDONLY; + imp->imp_connect_data.ocd_connect_flags |= + OBD_CONNECT_RDONLY; + } else { + imp->imp_connect_flags_orig &= ~OBD_CONNECT_RDONLY; + imp->imp_connect_data.ocd_connect_flags &= + ~OBD_CONNECT_RDONLY; + } + spin_unlock(&imp->imp_lock); + + rc = do_set_info_async(imp, MDS_SET_INFO, LUSTRE_MDS_VERSION, + keylen, key, vallen, val, set); + RETURN(rc); + } + if (KEY_IS(KEY_SPTLRPC_CONF)) { + sptlrpc_conf_client_adapt(exp->exp_obd); + RETURN(0); + } + if (KEY_IS(KEY_FLUSH_CTX)) { + sptlrpc_import_flush_my_ctx(imp); + RETURN(0); + } + if (KEY_IS(KEY_MDS_CONN)) { + /* mds-mds import */ + spin_lock(&imp->imp_lock); + imp->imp_server_timeout = 1; + spin_unlock(&imp->imp_lock); + imp->imp_client->cli_request_portal = MDS_MDS_PORTAL; + CDEBUG(D_OTHER, "%s: timeout / 2\n", exp->exp_obd->obd_name); + RETURN(0); + } + if (KEY_IS(KEY_CHANGELOG_CLEAR)) { + rc = do_set_info_async(imp, MDS_SET_INFO, LUSTRE_MDS_VERSION, + keylen, key, vallen, val, set); + RETURN(rc); + } + if (KEY_IS(KEY_HSM_COPYTOOL_SEND)) { + rc = mdc_hsm_copytool_send(vallen, val); + RETURN(rc); + } + + 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) +{ + int rc = -EINVAL; + + if (KEY_IS(KEY_MAX_EASIZE)) { + int 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); + } else if (KEY_IS(KEY_CONN_DATA)) { + struct obd_import *imp = class_exp2cliimp(exp); + struct obd_connect_data *data = val; + + if (*vallen != sizeof(*data)) + RETURN(-EINVAL); + + *data = imp->imp_connect_data; + RETURN(0); + } else if (KEY_IS(KEY_TGT_COUNT)) { + *((int *)val) = 1; + RETURN(0); + } + + rc = mdc_get_info_rpc(exp, keylen, key, *vallen, val); + + 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; - OBD_ALLOC_PTR(handle->och_mod); + handle->och_mod = obd_mod_alloc(); if (handle->och_mod == NULL) { - DEBUG_REQ(D_ERROR, req, "can't allocate mdc_open_data"); + DEBUG_REQ(D_ERROR, req, "can't allocate md_open_data"); GOTO(err_out, rc = -ENOMEM); } - /* will be dropped by unpin */ - CFS_INIT_LIST_HEAD(&handle->och_mod->mod_replay_list); - list_add_tail(&req->rq_mod_list, &handle->och_mod->mod_replay_list); + handle->och_mod->mod_open_req = req; /* will be dropped by unpin */ RETURN(0); @@ -1357,15 +2304,9 @@ static int mdc_unpin(struct obd_export *exp, struct obd_client_handle *handle, CERROR("Unpin failed: %d\n", rc); ptlrpc_req_finished(req); + ptlrpc_req_finished(handle->och_mod->mod_open_req); - LASSERT(!list_empty(&handle->och_mod->mod_replay_list)); - req = list_entry(handle->och_mod->mod_replay_list.next, - typeof(*req), rq_mod_list); - list_del_init(&req->rq_mod_list); - ptlrpc_req_finished(req); - LASSERT(list_empty(&handle->och_mod->mod_replay_list)); - - OBD_FREE(handle->och_mod, sizeof(*handle->och_mod)); + obd_mod_put(handle->och_mod); RETURN(rc); } @@ -1421,12 +2362,9 @@ static int mdc_import_event(struct obd_device *obd, struct obd_import *imp, /* * Flush current sequence to make client obtain new one * from server in case of disconnect/reconnect. - * If range is already empty then no need to flush it. */ - if (cli->cl_seq != NULL && - !range_is_exhausted(&cli->cl_seq->lcs_space)) { + if (cli->cl_seq != NULL) seq_client_flush(cli->cl_seq); - } rc = obd_notify_observer(obd, obd, OBD_NOTIFY_INACTIVE, NULL); break; @@ -1438,14 +2376,18 @@ static int mdc_import_event(struct obd_device *obd, struct obd_import *imp, break; } - case IMP_EVENT_ACTIVE: { - rc = obd_notify_observer(obd, obd, OBD_NOTIFY_ACTIVE, NULL); - break; - } + case IMP_EVENT_ACTIVE: + rc = obd_notify_observer(obd, obd, OBD_NOTIFY_ACTIVE, NULL); + /* restore re-establish kuc registration after reconnecting */ + if (rc == 0) + rc = mdc_kuc_reregister(imp); + break; case IMP_EVENT_OCD: rc = obd_notify_observer(obd, obd, OBD_NOTIFY_OCD, NULL); break; - + case IMP_EVENT_DEACTIVATE: + case IMP_EVENT_ACTIVATE: + break; default: CERROR("Unknown import event %x\n", event); LBUG(); @@ -1453,76 +2395,50 @@ static int mdc_import_event(struct obd_device *obd, struct obd_import *imp, RETURN(rc); } -static int mdc_fid_init(struct obd_export *exp) +int mdc_fid_alloc(struct obd_export *exp, struct lu_fid *fid, + struct md_op_data *op_data) { struct client_obd *cli = &exp->exp_obd->u.cli; - char *prefix; - int rc; + struct lu_client_seq *seq = cli->cl_seq; ENTRY; - - OBD_ALLOC_PTR(cli->cl_seq); - if (cli->cl_seq == NULL) - RETURN(-ENOMEM); - - OBD_ALLOC(prefix, MAX_OBD_NAME + 5); - if (prefix == NULL) - GOTO(out_free_seq, rc = -ENOMEM); - - snprintf(prefix, MAX_OBD_NAME + 5, "srv-%s", - exp->exp_obd->obd_name); - - /* Init client side sequence-manager */ - rc = seq_client_init(cli->cl_seq, exp, - LUSTRE_SEQ_METADATA, - prefix, NULL); - OBD_FREE(prefix, MAX_OBD_NAME + 5); - if (rc) - GOTO(out_free_seq, rc); - - RETURN(rc); -out_free_seq: - OBD_FREE_PTR(cli->cl_seq); - cli->cl_seq = NULL; - return rc; + RETURN(seq_client_alloc_fid(NULL, seq, fid)); } -static int mdc_fid_fini(struct obd_export *exp) -{ +struct obd_uuid *mdc_get_uuid(struct obd_export *exp) { struct client_obd *cli = &exp->exp_obd->u.cli; - ENTRY; - - if (cli->cl_seq != NULL) { - seq_client_fini(cli->cl_seq); - OBD_FREE_PTR(cli->cl_seq); - cli->cl_seq = NULL; - } - - RETURN(0); + return &cli->cl_target_uuid; } -int mdc_fid_alloc(struct obd_export *exp, struct lu_fid *fid, - struct md_op_data *op_data) +/** + * Determine whether the lock can be canceled before replaying it during + * 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) { - struct client_obd *cli = &exp->exp_obd->u.cli; - struct lu_client_seq *seq = cli->cl_seq; - ENTRY; - RETURN(seq_client_alloc_fid(seq, fid)); + 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); + + RETURN(1); } -/* XXX This method is used only to clear current fid seq - * once fld/mds insert failed */ -static int mdc_fid_delete(struct obd_export *exp, const struct lu_fid *fid) +static int mdc_resource_inode_free(struct ldlm_resource *res) { - struct client_obd *cli = &exp->exp_obd->u.cli; + if (res->lr_lvb_inode) + res->lr_lvb_inode = NULL; - seq_client_flush(cli->cl_seq); - return 0; + return 0; } -struct obd_uuid *mdc_get_uuid(struct obd_export *exp) { - struct client_obd *cli = &exp->exp_obd->u.cli; - return &cli->cl_target_uuid; -} +struct ldlm_valblock_ops inode_lvbo = { + lvbo_free: mdc_resource_inode_free +}; static int mdc_setup(struct obd_device *obd, struct lustre_cfg *cfg) { @@ -1538,14 +2454,9 @@ static int mdc_setup(struct obd_device *obd, struct lustre_cfg *cfg) ptlrpcd_addref(); - OBD_ALLOC(cli->cl_setattr_lock, sizeof (*cli->cl_setattr_lock)); - if (!cli->cl_setattr_lock) - GOTO(err_rpc_lock, rc = -ENOMEM); - mdc_init_rpc_lock(cli->cl_setattr_lock); - OBD_ALLOC(cli->cl_close_lock, sizeof (*cli->cl_close_lock)); if (!cli->cl_close_lock) - GOTO(err_setattr_lock, rc = -ENOMEM); + GOTO(err_rpc_lock, rc = -ENOMEM); mdc_init_rpc_lock(cli->cl_close_lock); rc = client_obd_setup(obd, cfg); @@ -1556,7 +2467,11 @@ static int mdc_setup(struct obd_device *obd, struct lustre_cfg *cfg) sptlrpc_lprocfs_cliobd_attach(obd); ptlrpc_lprocfs_register_obd(obd); - rc = obd_llog_init(obd, &obd->obd_olg, obd, 0, NULL, NULL); + ns_register_cancel(obd->obd_namespace, mdc_cancel_for_recovery); + + obd->obd_namespace->ns_lvbo = &inode_lvbo; + + rc = obd_llog_init(obd, &obd->obd_olg, obd, NULL); if (rc) { mdc_cleanup(obd); CERROR("failed to setup llogging subsystems\n"); @@ -1566,8 +2481,6 @@ static int mdc_setup(struct obd_device *obd, struct lustre_cfg *cfg) err_close_lock: OBD_FREE(cli->cl_close_lock, sizeof (*cli->cl_close_lock)); -err_setattr_lock: - OBD_FREE(cli->cl_setattr_lock, sizeof (*cli->cl_setattr_lock)); err_rpc_lock: OBD_FREE(cli->cl_rpc_lock, sizeof (*cli->cl_rpc_lock)); ptlrpcd_decref(); @@ -1604,19 +2517,16 @@ static int mdc_precleanup(struct obd_device *obd, enum obd_cleanup_stage stage) switch (stage) { case OBD_CLEANUP_EARLY: + break; case OBD_CLEANUP_EXPORTS: - /* If we set up but never connected, the - client import will not have been cleaned. */ - if (obd->u.cli.cl_import) { - struct obd_import *imp; - down_write(&obd->u.cli.cl_sem); - imp = obd->u.cli.cl_import; - CERROR("client import never connected\n"); - ptlrpc_invalidate_import(imp); - class_destroy_import(imp); - up_write(&obd->u.cli.cl_sem); - obd->u.cli.cl_import = NULL; - } + /* Failsafe, ok if racy */ + if (obd->obd_type->typ_refcnt <= 1) + libcfs_kkuc_group_rem(0, KUC_GRP_HSM); + + obd_cleanup_client_import(obd); + ptlrpc_lprocfs_unregister_obd(obd); + lprocfs_obd_cleanup(obd); + rc = obd_llog_finish(obd, 0); if (rc != 0) CERROR("failed to cleanup llogging subsystems\n"); @@ -1630,11 +2540,8 @@ static int mdc_cleanup(struct obd_device *obd) struct client_obd *cli = &obd->u.cli; OBD_FREE(cli->cl_rpc_lock, sizeof (*cli->cl_rpc_lock)); - OBD_FREE(cli->cl_setattr_lock, sizeof (*cli->cl_setattr_lock)); OBD_FREE(cli->cl_close_lock, sizeof (*cli->cl_close_lock)); - ptlrpc_lprocfs_unregister_obd(obd); - lprocfs_obd_cleanup(obd); ptlrpcd_decref(); return client_obd_cleanup(obd); @@ -1642,37 +2549,38 @@ static int mdc_cleanup(struct obd_device *obd) static int mdc_llog_init(struct obd_device *obd, struct obd_llog_group *olg, - struct obd_device *tgt, int count, - struct llog_catid *logid, struct obd_uuid *uuid) + struct obd_device *tgt, int *index) { - struct llog_ctxt *ctxt; - int rc; - ENTRY; + struct llog_ctxt *ctxt; + int rc; - LASSERT(olg == &obd->obd_olg); + ENTRY; - rc = llog_setup(obd, olg, LLOG_LOVEA_REPL_CTXT, tgt, 0, - NULL, &llog_client_ops); - if (rc == 0) { - ctxt = llog_get_context(obd, LLOG_LOVEA_REPL_CTXT); - llog_initiator_connect(ctxt); - llog_ctxt_put(ctxt); - } + LASSERT(olg == &obd->obd_olg); - RETURN(rc); + 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; - int rc = 0; - ENTRY; + struct llog_ctxt *ctxt; - ctxt = llog_get_context(obd, LLOG_LOVEA_REPL_CTXT); - if (ctxt) - rc = llog_cleanup(ctxt); + ENTRY; - RETURN(rc); + 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) @@ -1683,12 +2591,11 @@ static int mdc_process_config(struct obd_device *obd, obd_count len, void *buf) lprocfs_mdc_init_vars(&lvars); switch (lcfg->lcfg_command) { - case LCFG_SPTLRPC_CONF: - rc = sptlrpc_cliobd_process_config(obd, lcfg); - break; default: rc = class_process_proc_param(PARAM_MDC, lvars.obd_vars, lcfg, obd); + if (rc > 0) + rc = 0; break; } return(rc); @@ -1735,11 +2642,10 @@ int mdc_get_remote_perm(struct obd_export *exp, const struct lu_fid *fid, } static int mdc_interpret_renew_capa(const struct lu_env *env, - struct ptlrpc_request *req, void *unused, + struct ptlrpc_request *req, void *args, int status) { - struct obd_capa *oc = req->rq_async_args.pointer_arg[0]; - renew_capa_cb_t cb = req->rq_async_args.pointer_arg[1]; + struct mdc_renew_capa_args *ra = args; struct mdt_body *body = NULL; struct lustre_capa *capa; ENTRY; @@ -1759,7 +2665,7 @@ static int mdc_interpret_renew_capa(const struct lu_env *env, GOTO(out, capa = ERR_PTR(-EFAULT)); EXIT; out: - cb(oc, capa); + ra->ra_cb(ra->ra_oc, capa); return 0; } @@ -1767,6 +2673,7 @@ static int mdc_renew_capa(struct obd_export *exp, struct obd_capa *oc, renew_capa_cb_t cb) { struct ptlrpc_request *req; + struct mdc_renew_capa_args *ra; ENTRY; req = ptlrpc_request_alloc_pack(class_exp2cliimp(exp), &RQF_MDS_GETATTR, @@ -1780,32 +2687,34 @@ static int mdc_renew_capa(struct obd_export *exp, struct obd_capa *oc, mdc_pack_body(req, &oc->c_capa.lc_fid, oc, OBD_MD_FLOSSCAPA, 0, -1, 0); ptlrpc_request_set_replen(req); - req->rq_async_args.pointer_arg[0] = oc; - req->rq_async_args.pointer_arg[1] = cb; + CLASSERT(sizeof(*ra) <= sizeof(req->rq_async_args)); + ra = ptlrpc_req_async_args(req); + ra->ra_oc = oc; + ra->ra_cb = cb; req->rq_interpret_reply = mdc_interpret_renew_capa; - ptlrpcd_add_req(req, PSCOPE_OTHER); + ptlrpcd_add_req(req, PDL_POLICY_LOCAL, -1); RETURN(0); } static int mdc_connect(const struct lu_env *env, - struct lustre_handle *dlm_handle, + struct obd_export **exp, struct obd_device *obd, struct obd_uuid *cluuid, struct obd_connect_data *data, void *localdata) { - struct obd_import *imp = obd->u.cli.cl_import; + struct obd_import *imp = obd->u.cli.cl_import; - /* mds-mds import features */ - if (data && (data->ocd_connect_flags & OBD_CONNECT_MDS_MDS)) { - spin_lock(&imp->imp_lock); - imp->imp_server_timeout = 1; - spin_unlock(&imp->imp_lock); + /* mds-mds import features */ + if (data && (data->ocd_connect_flags & OBD_CONNECT_MDS_MDS)) { + spin_lock(&imp->imp_lock); + imp->imp_server_timeout = 1; + spin_unlock(&imp->imp_lock); imp->imp_client->cli_request_portal = MDS_MDS_PORTAL; CDEBUG(D_OTHER, "%s: Set 'mds' portal and timeout\n", obd->obd_name); } - return client_connect_import(env, dlm_handle, obd, cluuid, data, NULL); + return client_connect_import(env, exp, obd, cluuid, data, NULL); } struct obd_ops mdc_obd_ops = { @@ -1822,21 +2731,23 @@ struct obd_ops mdc_obd_ops = { .o_statfs = mdc_statfs, .o_pin = mdc_pin, .o_unpin = mdc_unpin, - .o_fid_init = mdc_fid_init, - .o_fid_fini = mdc_fid_fini, + .o_fid_init = client_fid_init, + .o_fid_fini = client_fid_fini, .o_fid_alloc = mdc_fid_alloc, - .o_fid_delete = mdc_fid_delete, .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_quotactl = mdc_quotactl, + .o_quotacheck = mdc_quotacheck }; struct md_ops mdc_md_ops = { .m_getstatus = mdc_getstatus, - .m_change_cbdata = mdc_change_cbdata, + .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, @@ -1862,37 +2773,26 @@ struct md_ops mdc_md_ops = { .m_set_open_replay_data = mdc_set_open_replay_data, .m_clear_open_replay_data = mdc_clear_open_replay_data, .m_renew_capa = mdc_renew_capa, + .m_unpack_capa = mdc_unpack_capa, .m_get_remote_perm = mdc_get_remote_perm, .m_intent_getattr_async = mdc_intent_getattr_async, .m_revalidate_lock = mdc_revalidate_lock }; -extern quota_interface_t mdc_quota_interface; - int __init mdc_init(void) { int rc; struct lprocfs_static_vars lvars = { 0 }; lprocfs_mdc_init_vars(&lvars); - request_module("lquota"); - quota_interface = PORTAL_SYMBOL_GET(mdc_quota_interface); - init_obd_quota_ops(quota_interface, &mdc_obd_ops); - rc = class_register_type(&mdc_obd_ops, &mdc_md_ops, lvars.module_vars, LUSTRE_MDC_NAME, NULL); - if (rc && quota_interface) - PORTAL_SYMBOL_PUT(mdc_quota_interface); - RETURN(rc); } #ifdef __KERNEL__ static void /*__exit*/ mdc_exit(void) { - if (quota_interface) - PORTAL_SYMBOL_PUT(mdc_quota_interface); - class_unregister_type(LUSTRE_MDC_NAME); }