X-Git-Url: https://git.whamcloud.com/?a=blobdiff_plain;f=lustre%2Fmdc%2Fmdc_lib.c;h=9d09ab6e3ceb22e0d1737cae23789c5d55efa324;hb=c49ff6ff89725052a29aa91d780a881e7829dfa2;hp=1396f8da6f38993a9d66b6aa20581dd1f6fcf99d;hpb=96ec6856f91f7f9031cfce4273c714d72cfe59ae;p=fs%2Flustre-release.git diff --git a/lustre/mdc/mdc_lib.c b/lustre/mdc/mdc_lib.c index 1396f8d..9d09ab6 100644 --- a/lustre/mdc/mdc_lib.c +++ b/lustre/mdc/mdc_lib.c @@ -1,282 +1,554 @@ -/* -*- mode: c; c-basic-offset: 8; indent-tabs-mode: nil; -*- - * vim:expandtab:shiftwidth=8:tabstop=8: +/* + * GPL HEADER START * - * Copyright (c) 2003 Cluster File Systems, Inc. + * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER. * - * This file is part of Lustre, http://www.lustre.org. + * This program is free software; you can redistribute it and/or modify + * it under the terms of the GNU General Public License version 2 only, + * as published by the Free Software Foundation. * - * Lustre is free software; you can redistribute it and/or - * modify it under the terms of version 2 of the GNU General Public - * License as published by the Free Software Foundation. + * This program is distributed in the hope that it will be useful, but + * WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU + * General Public License version 2 for more details (a copy is included + * in the LICENSE file that accompanied this code). * - * Lustre is distributed in the hope that it will be useful, - * but WITHOUT ANY WARRANTY; without even the implied warranty of - * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the - * GNU General Public License for more details. + * You should have received a copy of the GNU General Public License + * version 2 along with this program; If not, see + * http://www.sun.com/software/products/lustre/docs/GPLv2.pdf * - * You should have received a copy of the GNU General Public License - * along with Lustre; if not, write to the Free Software - * Foundation, Inc., 675 Mass Ave, Cambridge, MA 02139, USA. + * Please contact Sun Microsystems, Inc., 4150 Network Circle, Santa Clara, + * CA 95054 USA or visit www.sun.com if you need additional information or + * have any questions. + * + * GPL HEADER END + */ +/* + * Copyright (c) 2003, 2010, Oracle and/or its affiliates. All rights reserved. + * Use is subject to license terms. + * + * Copyright (c) 2011, Whamcloud, Inc. + */ +/* + * This file is part of Lustre, http://www.lustre.org/ + * Lustre is a trademark of Sun Microsystems, Inc. */ -#define DEBUG_SUBSYSTEM S_MDS +#define DEBUG_SUBSYSTEM S_MDC #ifndef __KERNEL__ +# include # include #endif -#include -#include -#include -#include +#include +#include +#include "mdc_internal.h" + +#ifndef __KERNEL__ +/* some liblustre hackings here */ +#ifndef O_DIRECTORY +#define O_DIRECTORY 0 +#endif +#endif -void mds_readdir_pack(struct ptlrpc_request *req, __u64 offset, __u32 size, - obd_id ino, int type, __u64 xid) +static void __mdc_pack_body(struct mdt_body *b, __u32 suppgid) { - struct mds_body *b; - - b = lustre_msg_buf(req->rq_reqmsg, 0, sizeof (*b)); - b->fsuid = current->fsuid; - b->fsgid = current->fsgid; - b->capability = current->cap_effective; - b->fid1.id = ino; - b->fid1.f_type = type; - b->size = offset; /* !! */ - b->suppgid = -1; - b->blocks = xid; /* !! */ - b->nlink = size; /* !! */ + LASSERT (b != NULL); + + b->suppgid = suppgid; + b->uid = cfs_curproc_uid(); + b->gid = cfs_curproc_gid(); + b->fsuid = cfs_curproc_fsuid(); + b->fsgid = cfs_curproc_fsgid(); + b->capability = cfs_curproc_cap_pack(); } -static void mds_pack_body(struct mds_body *b) +void mdc_pack_capa(struct ptlrpc_request *req, const struct req_msg_field *field, + struct obd_capa *oc) { - LASSERT (b != NULL); + struct req_capsule *pill = &req->rq_pill; + struct lustre_capa *c; - b->fsuid = current->fsuid; - b->fsgid = current->fsgid; - b->capability = current->cap_effective; + if (oc == NULL) { + LASSERT(req_capsule_get_size(pill, field, RCL_CLIENT) == 0); + return; + } + + c = req_capsule_client_get(pill, field); + LASSERT(c != NULL); + capa_cpy(c, oc); + DEBUG_CAPA(D_SEC, c, "pack"); +} + +void mdc_is_subdir_pack(struct ptlrpc_request *req, const struct lu_fid *pfid, + const struct lu_fid *cfid, int flags) +{ + struct mdt_body *b = req_capsule_client_get(&req->rq_pill, + &RMF_MDT_BODY); + + if (pfid) { + b->fid1 = *pfid; + b->valid = OBD_MD_FLID; + } + if (cfid) + b->fid2 = *cfid; + b->flags = flags; } -void mds_pack_req_body(struct ptlrpc_request *req) +void mdc_pack_body(struct ptlrpc_request *req, + const struct lu_fid *fid, struct obd_capa *oc, + __u64 valid, int ea_size, __u32 suppgid, int flags) { - struct mds_body *b = lustre_msg_buf(req->rq_reqmsg, 0, sizeof (*b)); - mds_pack_body(b); + struct mdt_body *b = req_capsule_client_get(&req->rq_pill, + &RMF_MDT_BODY); + LASSERT(b != NULL); + b->valid = valid; + b->eadatasize = ea_size; + b->flags = flags; + __mdc_pack_body(b, suppgid); + if (fid) { + b->fid1 = *fid; + b->valid |= OBD_MD_FLID; + mdc_pack_capa(req, &RMF_CAPA1, oc); + } +} + +void mdc_readdir_pack(struct ptlrpc_request *req, __u64 pgoff, + __u32 size, const struct lu_fid *fid, struct obd_capa *oc) +{ + struct mdt_body *b = req_capsule_client_get(&req->rq_pill, + &RMF_MDT_BODY); + b->fid1 = *fid; + b->valid |= OBD_MD_FLID; + b->size = pgoff; /* !! */ + b->nlink = size; /* !! */ + __mdc_pack_body(b, -1); + b->mode = LUDA_FID | LUDA_TYPE; + + mdc_pack_capa(req, &RMF_CAPA1, oc); } /* packing of MDS records */ -void mds_create_pack(struct ptlrpc_request *req, int offset, - struct mdc_op_data *op_data, - __u32 mode, __u64 rdev, __u32 uid, __u32 gid, __u64 time, - const void *data, int datalen) +void mdc_create_pack(struct ptlrpc_request *req, struct md_op_data *op_data, + const void *data, int datalen, __u32 mode, + __u32 uid, __u32 gid, cfs_cap_t cap_effective, __u64 rdev) { - struct mds_rec_create *rec; - char *tmp; - rec = lustre_msg_buf(req->rq_reqmsg, offset, sizeof (*rec)); - - rec->cr_opcode = REINT_CREATE; - rec->cr_fsuid = current->fsuid; - rec->cr_fsgid = current->fsgid; - rec->cr_cap = current->cap_effective; - ll_ino2fid(&rec->cr_fid, op_data->ino1, op_data->gen1, op_data->typ1); - memset(&rec->cr_replayfid, 0, sizeof(rec->cr_replayfid)); - rec->cr_mode = mode; - rec->cr_rdev = rdev; - rec->cr_uid = uid; - rec->cr_gid = gid; - rec->cr_time = time; - if (in_group_p(op_data->gid1)) - rec->cr_suppgid = op_data->gid1; - else - rec->cr_suppgid = -1; + struct mdt_rec_create *rec; + char *tmp; - tmp = lustre_msg_buf(req->rq_reqmsg, offset + 1, op_data->namelen + 1); - LOGL0(op_data->name, op_data->namelen, tmp); + CLASSERT(sizeof(struct mdt_rec_reint) == sizeof(struct mdt_rec_create)); + rec = req_capsule_client_get(&req->rq_pill, &RMF_REC_REINT); + + + rec->cr_opcode = REINT_CREATE; + rec->cr_fsuid = uid; + rec->cr_fsgid = gid; + rec->cr_cap = cap_effective; + rec->cr_fid1 = op_data->op_fid1; + rec->cr_fid2 = op_data->op_fid2; + rec->cr_mode = mode; + rec->cr_rdev = rdev; + rec->cr_time = op_data->op_mod_time; + rec->cr_suppgid1 = op_data->op_suppgids[0]; + rec->cr_suppgid2 = op_data->op_suppgids[1]; + set_mrc_cr_flags(rec, op_data->op_flags & MF_SOM_LOCAL_FLAGS); + rec->cr_bias = op_data->op_bias; + rec->cr_umask = cfs_curproc_umask(); + + mdc_pack_capa(req, &RMF_CAPA1, op_data->op_capa1); + + tmp = req_capsule_client_get(&req->rq_pill, &RMF_NAME); + LOGL0(op_data->op_name, op_data->op_namelen, tmp); if (data) { - tmp = lustre_msg_buf(req->rq_reqmsg, offset + 2, datalen); - memcpy (tmp, data, datalen); + tmp = req_capsule_client_get(&req->rq_pill, &RMF_EADATA); + memcpy(tmp, data, datalen); } } + +static __u64 mds_pack_open_flags(__u32 flags, __u32 mode) +{ + __u64 cr_flags = (flags & (FMODE_READ | FMODE_WRITE | + MDS_OPEN_HAS_EA | MDS_OPEN_HAS_OBJS | + MDS_OPEN_OWNEROVERRIDE | MDS_OPEN_LOCK | + MDS_OPEN_BY_FID)); + if (flags & O_CREAT) + cr_flags |= MDS_OPEN_CREAT; + if (flags & O_EXCL) + cr_flags |= MDS_OPEN_EXCL; + if (flags & O_TRUNC) + cr_flags |= MDS_OPEN_TRUNC; + if (flags & O_APPEND) + cr_flags |= MDS_OPEN_APPEND; + if (flags & O_SYNC) + cr_flags |= MDS_OPEN_SYNC; + if (flags & O_DIRECTORY) + cr_flags |= MDS_OPEN_DIRECTORY; +#ifdef FMODE_EXEC + if (flags & FMODE_EXEC) + cr_flags |= MDS_FMODE_EXEC; +#endif + if (flags & O_LOV_DELAY_CREATE) + cr_flags |= MDS_OPEN_DELAY_CREATE; + + if ((flags & O_NOACCESS) || (flags & O_NONBLOCK)) + cr_flags |= MDS_OPEN_NORESTORE; + + return cr_flags; +} + /* packing of MDS records */ -void mds_open_pack(struct ptlrpc_request *req, int offset, - struct mdc_op_data *op_data, - __u32 mode, __u64 rdev, __u32 uid, __u32 gid, __u64 time, - __u32 flags, - const void *data, int datalen) +void mdc_open_pack(struct ptlrpc_request *req, struct md_op_data *op_data, + __u32 mode, __u64 rdev, __u32 flags, const void *lmm, + int lmmlen) { - struct mds_rec_create *rec; + struct mdt_rec_create *rec; char *tmp; - rec = lustre_msg_buf(req->rq_reqmsg, offset, sizeof (*rec)); + __u64 cr_flags; - /* XXX do something about time, uid, gid */ - rec->cr_opcode = REINT_OPEN; - rec->cr_fsuid = current->fsuid; - rec->cr_fsgid = current->fsgid; - rec->cr_cap = current->cap_effective; - ll_ino2fid(&rec->cr_fid, op_data->ino1, - op_data->gen1, op_data->typ1); - memset(&rec->cr_replayfid, 0, sizeof(rec->cr_replayfid)); - rec->cr_mode = mode; - rec->cr_flags = flags; - rec->cr_rdev = rdev; - rec->cr_uid = uid; - rec->cr_gid = gid; - rec->cr_time = time; - if (in_group_p(op_data->gid1)) - rec->cr_suppgid = op_data->gid1; - else - rec->cr_suppgid = -1; + CLASSERT(sizeof(struct mdt_rec_reint) == sizeof(struct mdt_rec_create)); + rec = req_capsule_client_get(&req->rq_pill, &RMF_REC_REINT); - tmp = lustre_msg_buf(req->rq_reqmsg, offset + 1, op_data->namelen + 1); - LOGL0(op_data->name, op_data->namelen, tmp); + /* XXX do something about time, uid, gid */ + rec->cr_opcode = REINT_OPEN; + rec->cr_fsuid = cfs_curproc_fsuid(); + rec->cr_fsgid = cfs_curproc_fsgid(); + rec->cr_cap = cfs_curproc_cap_pack(); + if (op_data != NULL) { + rec->cr_fid1 = op_data->op_fid1; + rec->cr_fid2 = op_data->op_fid2; + } + rec->cr_mode = mode; + cr_flags = mds_pack_open_flags(flags, mode); + rec->cr_rdev = rdev; + rec->cr_time = op_data->op_mod_time; + rec->cr_suppgid1 = op_data->op_suppgids[0]; + rec->cr_suppgid2 = op_data->op_suppgids[1]; + rec->cr_bias = op_data->op_bias; + rec->cr_umask = cfs_curproc_umask(); + + mdc_pack_capa(req, &RMF_CAPA1, op_data->op_capa1); + /* the next buffer is child capa, which is used for replay, + * will be packed from the data in reply message. */ + + if (op_data->op_name) { + tmp = req_capsule_client_get(&req->rq_pill, &RMF_NAME); + LOGL0(op_data->op_name, op_data->op_namelen, tmp); + } - if (data) { - tmp = lustre_msg_buf(req->rq_reqmsg, offset + 2, datalen); - memcpy (tmp, data, datalen); + if (lmm) { + cr_flags |= MDS_OPEN_HAS_EA; +#ifndef __KERNEL__ + /*XXX a hack for liblustre to set EA (LL_IOC_LOV_SETSTRIPE) */ + rec->cr_fid2 = op_data->op_fid2; +#endif + tmp = req_capsule_client_get(&req->rq_pill, &RMF_EADATA); + memcpy (tmp, lmm, lmmlen); } + set_mrc_cr_flags(rec, cr_flags); +} + +static inline __u64 attr_pack(unsigned int ia_valid) { + __u64 sa_valid = 0; + + if (ia_valid & ATTR_MODE) + sa_valid |= MDS_ATTR_MODE; + if (ia_valid & ATTR_UID) + sa_valid |= MDS_ATTR_UID; + if (ia_valid & ATTR_GID) + sa_valid |= MDS_ATTR_GID; + if (ia_valid & ATTR_SIZE) + sa_valid |= MDS_ATTR_SIZE; + if (ia_valid & ATTR_ATIME) + sa_valid |= MDS_ATTR_ATIME; + if (ia_valid & ATTR_MTIME) + sa_valid |= MDS_ATTR_MTIME; + if (ia_valid & ATTR_CTIME) + sa_valid |= MDS_ATTR_CTIME; + if (ia_valid & ATTR_ATIME_SET) + sa_valid |= MDS_ATTR_ATIME_SET; + if (ia_valid & ATTR_MTIME_SET) + sa_valid |= MDS_ATTR_MTIME_SET; + if (ia_valid & ATTR_FORCE) + sa_valid |= MDS_ATTR_FORCE; + if (ia_valid & ATTR_ATTR_FLAG) + sa_valid |= MDS_ATTR_ATTR_FLAG; + if (ia_valid & ATTR_KILL_SUID) + sa_valid |= MDS_ATTR_KILL_SUID; + if (ia_valid & ATTR_KILL_SGID) + sa_valid |= MDS_ATTR_KILL_SGID; + if (ia_valid & ATTR_CTIME_SET) + sa_valid |= MDS_ATTR_CTIME_SET; + if (ia_valid & ATTR_FROM_OPEN) + sa_valid |= MDS_ATTR_FROM_OPEN; + if (ia_valid & ATTR_BLOCKS) + sa_valid |= MDS_ATTR_BLOCKS; + if (ia_valid & MDS_OPEN_OWNEROVERRIDE) + /* NFSD hack (see bug 5781) */ + sa_valid |= MDS_OPEN_OWNEROVERRIDE; + return sa_valid; +} + +static void mdc_setattr_pack_rec(struct mdt_rec_setattr *rec, + struct md_op_data *op_data) +{ + rec->sa_opcode = REINT_SETATTR; + rec->sa_fsuid = cfs_curproc_fsuid(); + rec->sa_fsgid = cfs_curproc_fsgid(); + rec->sa_cap = cfs_curproc_cap_pack(); + rec->sa_suppgid = -1; + + rec->sa_fid = op_data->op_fid1; + rec->sa_valid = attr_pack(op_data->op_attr.ia_valid); + rec->sa_mode = op_data->op_attr.ia_mode; + rec->sa_uid = op_data->op_attr.ia_uid; + rec->sa_gid = op_data->op_attr.ia_gid; + rec->sa_size = op_data->op_attr.ia_size; + rec->sa_blocks = op_data->op_attr_blocks; + rec->sa_atime = LTIME_S(op_data->op_attr.ia_atime); + rec->sa_mtime = LTIME_S(op_data->op_attr.ia_mtime); + rec->sa_ctime = LTIME_S(op_data->op_attr.ia_ctime); + rec->sa_attr_flags = ((struct ll_iattr *)&op_data->op_attr)->ia_attr_flags; + if ((op_data->op_attr.ia_valid & ATTR_GID) && + cfs_curproc_is_in_groups(op_data->op_attr.ia_gid)) + rec->sa_suppgid = op_data->op_attr.ia_gid; + else + rec->sa_suppgid = op_data->op_suppgids[0]; +} + +static void mdc_ioepoch_pack(struct mdt_ioepoch *epoch, + struct md_op_data *op_data) +{ + memcpy(&epoch->handle, &op_data->op_handle, sizeof(epoch->handle)); + epoch->ioepoch = op_data->op_ioepoch; + epoch->flags = op_data->op_flags & MF_SOM_LOCAL_FLAGS; } -void mds_setattr_pack(struct ptlrpc_request *req, - struct mdc_op_data *data, - struct iattr *iattr, void *ea, int ealen) + +void mdc_setattr_pack(struct ptlrpc_request *req, struct md_op_data *op_data, + void *ea, int ealen, void *ea2, int ea2len) { - struct mds_rec_setattr *rec = lustre_msg_buf(req->rq_reqmsg, 0, - sizeof (*rec)); - rec->sa_opcode = REINT_SETATTR; - rec->sa_fsuid = current->fsuid; - rec->sa_fsgid = current->fsgid; - rec->sa_cap = current->cap_effective; - ll_ino2fid(&rec->sa_fid, data->ino1, data->gen1, data->typ1); - - if (iattr) { - rec->sa_valid = iattr->ia_valid; - rec->sa_mode = iattr->ia_mode; - rec->sa_uid = iattr->ia_uid; - rec->sa_gid = iattr->ia_gid; - rec->sa_size = iattr->ia_size; - rec->sa_atime = LTIME_S(iattr->ia_atime); - rec->sa_mtime = LTIME_S(iattr->ia_mtime); - rec->sa_ctime = LTIME_S(iattr->ia_ctime); - rec->sa_attr_flags = iattr->ia_attr_flags; - - if ((iattr->ia_valid & ATTR_GID) && in_group_p(iattr->ia_gid)) - rec->sa_suppgid = iattr->ia_gid; - else if ((iattr->ia_valid & ATTR_MODE) && - in_group_p(data->gid1)) - rec->sa_suppgid = data->gid1; - else - rec->sa_suppgid = -1; + struct mdt_rec_setattr *rec; + struct mdt_ioepoch *epoch; + struct lov_user_md *lum = NULL; + + CLASSERT(sizeof(struct mdt_rec_reint) ==sizeof(struct mdt_rec_setattr)); + rec = req_capsule_client_get(&req->rq_pill, &RMF_REC_REINT); + mdc_setattr_pack_rec(rec, op_data); + + mdc_pack_capa(req, &RMF_CAPA1, op_data->op_capa1); + + if (op_data->op_flags & (MF_SOM_CHANGE | MF_EPOCH_OPEN)) { + epoch = req_capsule_client_get(&req->rq_pill, &RMF_MDT_EPOCH); + mdc_ioepoch_pack(epoch, op_data); } - if (ealen != 0) - memcpy(lustre_msg_buf(req->rq_reqmsg, 1, ealen), ea, ealen); + if (ealen == 0) + return; + + lum = req_capsule_client_get(&req->rq_pill, &RMF_EADATA); + if (ea == NULL) { /* Remove LOV EA */ + lum->lmm_magic = LOV_USER_MAGIC_V1; + lum->lmm_stripe_size = 0; + lum->lmm_stripe_count = 0; + lum->lmm_stripe_offset = (typeof(lum->lmm_stripe_offset))(-1); + } else { + memcpy(lum, ea, ealen); + } + + if (ea2len == 0) + return; + + memcpy(req_capsule_client_get(&req->rq_pill, &RMF_LOGCOOKIES), ea2, + ea2len); } -void mds_unlink_pack(struct ptlrpc_request *req, int offset, - struct mdc_op_data *data) +void mdc_unlink_pack(struct ptlrpc_request *req, struct md_op_data *op_data) { - struct mds_rec_unlink *rec; + struct mdt_rec_unlink *rec; char *tmp; - - rec = lustre_msg_buf(req->rq_reqmsg, offset, sizeof (*rec)); + + CLASSERT(sizeof(struct mdt_rec_reint) == sizeof(struct mdt_rec_unlink)); + rec = req_capsule_client_get(&req->rq_pill, &RMF_REC_REINT); LASSERT (rec != NULL); - rec->ul_opcode = REINT_UNLINK; - rec->ul_fsuid = current->fsuid; - rec->ul_fsgid = current->fsgid; - rec->ul_cap = current->cap_effective; - rec->ul_mode = data->mode; - if (in_group_p(data->gid1)) - rec->ul_suppgid = data->gid1; - else - rec->ul_suppgid = -1; - ll_ino2fid(&rec->ul_fid1, data->ino1, data->gen1, data->typ1); - if (data->ino2) - ll_ino2fid(&rec->ul_fid2, data->ino2, data->gen2, data->typ2); - - tmp = lustre_msg_buf(req->rq_reqmsg, offset + 1, data->namelen + 1); - LASSERT (tmp != NULL); - LOGL0(data->name, data->namelen, tmp); + rec->ul_opcode = REINT_UNLINK; + rec->ul_fsuid = op_data->op_fsuid; + rec->ul_fsgid = op_data->op_fsgid; + rec->ul_cap = op_data->op_cap; + rec->ul_mode = op_data->op_mode; + rec->ul_suppgid1= op_data->op_suppgids[0]; + rec->ul_suppgid2= -1; + rec->ul_fid1 = op_data->op_fid1; + rec->ul_fid2 = op_data->op_fid2; + rec->ul_time = op_data->op_mod_time; + rec->ul_bias = op_data->op_bias; + + mdc_pack_capa(req, &RMF_CAPA1, op_data->op_capa1); + + tmp = req_capsule_client_get(&req->rq_pill, &RMF_NAME); + LASSERT(tmp != NULL); + LOGL0(op_data->op_name, op_data->op_namelen, tmp); } -void mds_link_pack(struct ptlrpc_request *req, int offset, - struct mdc_op_data *data) +void mdc_link_pack(struct ptlrpc_request *req, struct md_op_data *op_data) { - struct mds_rec_link *rec; + struct mdt_rec_link *rec; char *tmp; - rec = lustre_msg_buf(req->rq_reqmsg, offset, sizeof (*rec)); - - rec->lk_opcode = REINT_LINK; - rec->lk_fsuid = current->fsuid; - rec->lk_fsgid = current->fsgid; - rec->lk_cap = current->cap_effective; - if (in_group_p(data->gid1)) - rec->lk_suppgid1 = data->gid1; - else - rec->lk_suppgid1 = -1; - if (in_group_p(data->gid2)) - rec->lk_suppgid2 = data->gid2; - else - rec->lk_suppgid2 = -1; - ll_ino2fid(&rec->lk_fid1, data->ino1, data->gen1, data->typ1); - ll_ino2fid(&rec->lk_fid2, data->ino2, data->gen2, data->typ2); + CLASSERT(sizeof(struct mdt_rec_reint) == sizeof(struct mdt_rec_link)); + rec = req_capsule_client_get(&req->rq_pill, &RMF_REC_REINT); + LASSERT (rec != NULL); - tmp = lustre_msg_buf(req->rq_reqmsg, offset + 1, data->namelen + 1); - LOGL0(data->name, data->namelen, tmp); + rec->lk_opcode = REINT_LINK; + rec->lk_fsuid = op_data->op_fsuid;//current->fsuid; + rec->lk_fsgid = op_data->op_fsgid;//current->fsgid; + rec->lk_cap = op_data->op_cap;//current->cap_effective; + rec->lk_suppgid1 = op_data->op_suppgids[0]; + rec->lk_suppgid2 = op_data->op_suppgids[1]; + rec->lk_fid1 = op_data->op_fid1; + rec->lk_fid2 = op_data->op_fid2; + rec->lk_time = op_data->op_mod_time; + rec->lk_bias = op_data->op_bias; + + mdc_pack_capa(req, &RMF_CAPA1, op_data->op_capa1); + mdc_pack_capa(req, &RMF_CAPA2, op_data->op_capa2); + + tmp = req_capsule_client_get(&req->rq_pill, &RMF_NAME); + LOGL0(op_data->op_name, op_data->op_namelen, tmp); } -void mds_rename_pack(struct ptlrpc_request *req, int offset, - struct mdc_op_data *data, +void mdc_rename_pack(struct ptlrpc_request *req, struct md_op_data *op_data, const char *old, int oldlen, const char *new, int newlen) { - struct mds_rec_rename *rec; + struct mdt_rec_rename *rec; char *tmp; - rec = lustre_msg_buf(req->rq_reqmsg, offset, sizeof (*rec)); + CLASSERT(sizeof(struct mdt_rec_reint) == sizeof(struct mdt_rec_rename)); + rec = req_capsule_client_get(&req->rq_pill, &RMF_REC_REINT); /* XXX do something about time, uid, gid */ - rec->rn_opcode = REINT_RENAME; - rec->rn_fsuid = current->fsuid; - rec->rn_fsgid = current->fsgid; - rec->rn_cap = current->cap_effective; - if (in_group_p(data->gid1)) - rec->rn_suppgid1 = data->gid1; - else - rec->rn_suppgid1 = -1; - if (in_group_p(data->gid2)) - rec->rn_suppgid2 = data->gid2; - else - rec->rn_suppgid2 = -1; - ll_ino2fid(&rec->rn_fid1, data->ino1, data->gen1, data->typ1); - ll_ino2fid(&rec->rn_fid2, data->ino2, data->gen2, data->typ2); - - tmp = lustre_msg_buf(req->rq_reqmsg, offset + 1, oldlen + 1); + rec->rn_opcode = REINT_RENAME; + rec->rn_fsuid = op_data->op_fsuid; + rec->rn_fsgid = op_data->op_fsgid; + rec->rn_cap = op_data->op_cap; + rec->rn_suppgid1 = op_data->op_suppgids[0]; + rec->rn_suppgid2 = op_data->op_suppgids[1]; + rec->rn_fid1 = op_data->op_fid1; + rec->rn_fid2 = op_data->op_fid2; + rec->rn_time = op_data->op_mod_time; + rec->rn_mode = op_data->op_mode; + rec->rn_bias = op_data->op_bias; + + mdc_pack_capa(req, &RMF_CAPA1, op_data->op_capa1); + mdc_pack_capa(req, &RMF_CAPA2, op_data->op_capa2); + + tmp = req_capsule_client_get(&req->rq_pill, &RMF_NAME); LOGL0(old, oldlen, tmp); if (new) { - tmp = lustre_msg_buf(req->rq_reqmsg, offset + 2, newlen + 1); + tmp = req_capsule_client_get(&req->rq_pill, &RMF_SYMTGT); LOGL0(new, newlen, tmp); } } -void mds_getattr_pack(struct ptlrpc_request *req, int valid, int offset, - int flags, struct mdc_op_data *data) +void mdc_getattr_pack(struct ptlrpc_request *req, __u64 valid, int flags, + struct md_op_data *op_data, int ea_size) { - struct mds_body *b; - b = lustre_msg_buf(req->rq_reqmsg, offset, sizeof (*b)); + struct mdt_body *b = req_capsule_client_get(&req->rq_pill, + &RMF_MDT_BODY); - b->fsuid = current->fsuid; - b->fsgid = current->fsgid; - b->capability = current->cap_effective; b->valid = valid; + if (op_data->op_bias & MDS_CHECK_SPLIT) + b->valid |= OBD_MD_FLCKSPLIT; + if (op_data->op_bias & MDS_CROSS_REF) + b->valid |= OBD_MD_FLCROSSREF; + b->eadatasize = ea_size; b->flags = flags; - if (in_group_p(data->gid1)) - b->suppgid = data->gid1; - else - b->suppgid = -1; - - ll_ino2fid(&b->fid1, data->ino1, data->gen1, data->typ1); - if (data->name) { - char *tmp; - tmp = lustre_msg_buf(req->rq_reqmsg, offset + 1, - data->namelen + 1); - LOGL0(data->name, data->namelen, tmp); + __mdc_pack_body(b, op_data->op_suppgids[0]); + + b->fid1 = op_data->op_fid1; + b->fid2 = op_data->op_fid2; + b->valid |= OBD_MD_FLID; + + mdc_pack_capa(req, &RMF_CAPA1, op_data->op_capa1); + + if (op_data->op_name) { + char *tmp = req_capsule_client_get(&req->rq_pill, &RMF_NAME); + LOGL0(op_data->op_name, op_data->op_namelen, tmp); + } } + +void mdc_close_pack(struct ptlrpc_request *req, struct md_op_data *op_data) +{ + struct mdt_ioepoch *epoch; + struct mdt_rec_setattr *rec; + + epoch = req_capsule_client_get(&req->rq_pill, &RMF_MDT_EPOCH); + rec = req_capsule_client_get(&req->rq_pill, &RMF_REC_REINT); + + mdc_setattr_pack_rec(rec, op_data); + mdc_pack_capa(req, &RMF_CAPA1, op_data->op_capa1); + mdc_ioepoch_pack(epoch, op_data); +} + +static int mdc_req_avail(struct client_obd *cli, struct mdc_cache_waiter *mcw) +{ + int rc; + ENTRY; + client_obd_list_lock(&cli->cl_loi_list_lock); + rc = cfs_list_empty(&mcw->mcw_entry); + client_obd_list_unlock(&cli->cl_loi_list_lock); + RETURN(rc); +}; + +/* We record requests in flight in cli->cl_r_in_flight here. + * There is only one write rpc possible in mdc anyway. If this to change + * in the future - the code may need to be revisited. */ +int mdc_enter_request(struct client_obd *cli) +{ + int rc = 0; + struct mdc_cache_waiter mcw; + struct l_wait_info lwi = LWI_INTR(LWI_ON_SIGNAL_NOOP, NULL); + + client_obd_list_lock(&cli->cl_loi_list_lock); + if (cli->cl_r_in_flight >= cli->cl_max_rpcs_in_flight) { + cfs_list_add_tail(&mcw.mcw_entry, &cli->cl_cache_waiters); + cfs_waitq_init(&mcw.mcw_waitq); + client_obd_list_unlock(&cli->cl_loi_list_lock); + rc = l_wait_event(mcw.mcw_waitq, mdc_req_avail(cli, &mcw), &lwi); + if (rc) { + client_obd_list_lock(&cli->cl_loi_list_lock); + if (cfs_list_empty(&mcw.mcw_entry)) + cli->cl_r_in_flight--; + cfs_list_del_init(&mcw.mcw_entry); + client_obd_list_unlock(&cli->cl_loi_list_lock); + } + } else { + cli->cl_r_in_flight++; + client_obd_list_unlock(&cli->cl_loi_list_lock); + } + return rc; +} + +void mdc_exit_request(struct client_obd *cli) +{ + cfs_list_t *l, *tmp; + struct mdc_cache_waiter *mcw; + + client_obd_list_lock(&cli->cl_loi_list_lock); + cli->cl_r_in_flight--; + cfs_list_for_each_safe(l, tmp, &cli->cl_cache_waiters) { + if (cli->cl_r_in_flight >= cli->cl_max_rpcs_in_flight) { + /* No free request slots anymore */ + break; + } + + mcw = cfs_list_entry(l, struct mdc_cache_waiter, mcw_entry); + cfs_list_del_init(&mcw->mcw_entry); + cli->cl_r_in_flight++; + cfs_waitq_signal(&mcw->mcw_waitq); + } + /* Empty waiting list? Decrease reqs in-flight number */ + + client_obd_list_unlock(&cli->cl_loi_list_lock); +}