lustre/cmm/mdc_object.c

   1 /*
   2  * GPL HEADER START
   3  *
   4  * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.
   5  *
   6  * This program is free software; you can redistribute it and/or modify
   7  * it under the terms of the GNU General Public License version 2 only,
   8  * as published by the Free Software Foundation.
   9  *
  10  * This program is distributed in the hope that it will be useful, but
  11  * WITHOUT ANY WARRANTY; without even the implied warranty of
  12  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
  13  * General Public License version 2 for more details (a copy is included
  14  * in the LICENSE file that accompanied this code).
  15  *
  16  * You should have received a copy of the GNU General Public License
  17  * version 2 along with this program; If not, see
  18  * http://www.sun.com/software/products/lustre/docs/GPLv2.pdf
  19  *
  20  * Please contact Sun Microsystems, Inc., 4150 Network Circle, Santa Clara,
  21  * CA 95054 USA or visit www.sun.com if you need additional information or
  22  * have any questions.
  23  *
  24  * GPL HEADER END
  25  */
  26 /*
  27  * Copyright (c) 2007, 2010, Oracle and/or its affiliates. All rights reserved.
  28  * Use is subject to license terms.
  29  */
  30 /*
  31  * This file is part of Lustre, http://www.lustre.org/
  32  * Lustre is a trademark of Sun Microsystems, Inc.
  33  *
  34  * lustre/cmm/mdc_object.c
  35  *
  36  * Lustre Cluster Metadata Manager (cmm)
  37  *
  38  * Author: Mike Pershin <tappro@clusterfs.com>
  39  */
  40
  41 #define DEBUG_SUBSYSTEM S_MDS
  42 #include <obd_support.h>
  43 #include <lustre_lib.h>
  44 #include <obd_class.h>
  45 #include <lustre_mdc.h>
  46 #include "cmm_internal.h"
  47 #include "mdc_internal.h"
  48
  49 static const struct md_object_operations mdc_mo_ops;
  50 static const struct md_dir_operations mdc_dir_ops;
  51 static const struct lu_object_operations mdc_obj_ops;
  52
  53 extern struct lu_context_key mdc_thread_key;
  54 /**
  55  * \addtogroup cmm_mdc
  56  * @{
  57  */
  58 /**
  59  * Allocate new mdc object.
  60  */
  61 struct lu_object *mdc_object_alloc(const struct lu_env *env,
  62                                    const struct lu_object_header *hdr,
  63                                    struct lu_device *ld)
  64 {
  65         struct mdc_object *mco;
  66         ENTRY;
  67
  68         OBD_ALLOC_PTR(mco);
  69         if (mco != NULL) {
  70                 struct lu_object *lo;
  71
  72                 lo = &mco->mco_obj.mo_lu;
  73                 lu_object_init(lo, NULL, ld);
  74                 mco->mco_obj.mo_ops = &mdc_mo_ops;
  75                 mco->mco_obj.mo_dir_ops = &mdc_dir_ops;
  76                 lo->lo_ops = &mdc_obj_ops;
  77                 RETURN(lo);
  78         } else
  79                 RETURN(NULL);
  80 }
  81
  82 /** Free current mdc object */
  83 static void mdc_object_free(const struct lu_env *env, struct lu_object *lo)
  84 {
  85         struct mdc_object *mco = lu2mdc_obj(lo);
  86         lu_object_fini(lo);
  87         OBD_FREE_PTR(mco);
  88 }
  89
  90 /**
  91  * Initialize mdc object. All of them have loh_attr::LOHA_REMOTE set.
  92  */
  93 static int mdc_object_init(const struct lu_env *env, struct lu_object *lo,
  94                            const struct lu_object_conf *unused)
  95 {
  96         ENTRY;
  97         lo->lo_header->loh_attr |= LOHA_REMOTE;
  98         RETURN(0);
  99 }
 100
 101 /**
 102  * Instance of lu_object_operations for mdc.
 103  */
 104 static const struct lu_object_operations mdc_obj_ops = {
 105         .loo_object_init    = mdc_object_init,
 106         .loo_object_free    = mdc_object_free,
 107 };
 108
 109 /**
 110  * \name The set of md_object_operations.
 111  * @{
 112  */
 113 /**
 114  * Get mdc_thread_info from lu_context
 115  */
 116 static
 117 struct mdc_thread_info *mdc_info_get(const struct lu_env *env)
 118 {
 119         struct mdc_thread_info *mci;
 120
 121         mci = lu_context_key_get(&env->le_ctx, &mdc_thread_key);
 122         LASSERT(mci);
 123         return mci;
 124 }
 125
 126 /**
 127  * Initialize mdc_thread_info.
 128  */
 129 static
 130 struct mdc_thread_info *mdc_info_init(const struct lu_env *env)
 131 {
 132         struct mdc_thread_info *mci = mdc_info_get(env);
 133         memset(mci, 0, sizeof(*mci));
 134         return mci;
 135 }
 136
 137 /**
 138  * Convert attributes from mdt_body to the md_attr.
 139  */
 140 static void mdc_body2attr(struct mdt_body *body, struct md_attr *ma)
 141 {
 142         struct lu_attr *la = &ma->ma_attr;
 143         /* update time */
 144         if (body->valid & OBD_MD_FLCTIME && body->ctime >= la->la_ctime) {
 145                 la->la_ctime = body->ctime;
 146                 if (body->valid & OBD_MD_FLMTIME)
 147                         la->la_mtime = body->mtime;
 148         }
 149
 150         if (body->valid & OBD_MD_FLMODE)
 151                 la->la_mode = body->mode;
 152         if (body->valid & OBD_MD_FLSIZE)
 153                 la->la_size = body->size;
 154         if (body->valid & OBD_MD_FLBLOCKS)
 155                 la->la_blocks = body->blocks;
 156         if (body->valid & OBD_MD_FLUID)
 157                 la->la_uid = body->uid;
 158         if (body->valid & OBD_MD_FLGID)
 159                 la->la_gid = body->gid;
 160         if (body->valid & OBD_MD_FLFLAGS)
 161                 la->la_flags = body->flags;
 162         if (body->valid & OBD_MD_FLNLINK)
 163                 la->la_nlink = body->nlink;
 164         if (body->valid & OBD_MD_FLRDEV)
 165                 la->la_rdev = body->rdev;
 166
 167         la->la_valid = body->valid;
 168         ma->ma_valid = MA_INODE;
 169 }
 170
 171 /**
 172  * Fill the md_attr \a ma with attributes from request.
 173  */
 174 static int mdc_req2attr_update(const struct lu_env *env,
 175                                struct md_attr *ma)
 176 {
 177         struct mdc_thread_info *mci;
 178         struct ptlrpc_request *req;
 179         struct mdt_body *body;
 180         struct lov_mds_md *md;
 181         struct llog_cookie *cookie;
 182         void *acl;
 183
 184         ENTRY;
 185         mci = mdc_info_get(env);
 186         req = mci->mci_req;
 187         LASSERT(req);
 188         body = req_capsule_server_get(&req->rq_pill, &RMF_MDT_BODY);
 189         LASSERT(body);
 190         mdc_body2attr(body, ma);
 191
 192         if (body->valid & OBD_MD_FLMDSCAPA) {
 193                 struct lustre_capa *capa;
 194
 195                 /* create for cross-ref will fetch mds capa from remote obj */
 196                 capa = req_capsule_server_get(&req->rq_pill, &RMF_CAPA1);
 197                 LASSERT(capa != NULL);
 198                 LASSERT(ma->ma_capa != NULL);
 199                 *ma->ma_capa = *capa;
 200         }
 201
 202         if ((body->valid & OBD_MD_FLEASIZE) || (body->valid & OBD_MD_FLDIREA)) {
 203                 if (body->eadatasize == 0) {
 204                         CERROR("No size defined for easize field\n");
 205                         RETURN(-EPROTO);
 206                 }
 207
 208                 md = req_capsule_server_sized_get(&req->rq_pill, &RMF_MDT_MD,
 209                                                   body->eadatasize);
 210                 if (md == NULL)
 211                         RETURN(-EPROTO);
 212
 213                 LASSERT(ma->ma_lmm != NULL);
 214                 LASSERT(ma->ma_lmm_size >= body->eadatasize);
 215                 ma->ma_lmm_size = body->eadatasize;
 216                 memcpy(ma->ma_lmm, md, ma->ma_lmm_size);
 217                 ma->ma_valid |= MA_LOV;
 218         }
 219
 220         if (body->valid & OBD_MD_FLCOOKIE) {
 221                 /*
 222                  * ACL and cookie share the same body->aclsize, we need
 223                  * to make sure that they both never come here.
 224                  */
 225                 LASSERT(!(body->valid & OBD_MD_FLACL));
 226
 227                 if (body->aclsize == 0) {
 228                         CERROR("No size defined for cookie field\n");
 229                         RETURN(-EPROTO);
 230                 }
 231
 232                 cookie = req_capsule_server_sized_get(&req->rq_pill,
 233                                                       &RMF_LOGCOOKIES,
 234                                                       body->aclsize);
 235                 if (cookie == NULL)
 236                         RETURN(-EPROTO);
 237
 238                 LASSERT(ma->ma_cookie != NULL);
 239                 LASSERT(ma->ma_cookie_size == body->aclsize);
 240                 memcpy(ma->ma_cookie, cookie, ma->ma_cookie_size);
 241                 ma->ma_valid |= MA_COOKIE;
 242         }
 243
 244 #ifdef CONFIG_FS_POSIX_ACL
 245         if (body->valid & OBD_MD_FLACL) {
 246                 if (body->aclsize == 0) {
 247                         CERROR("No size defined for acl field\n");
 248                         RETURN(-EPROTO);
 249                 }
 250
 251                 acl = req_capsule_server_sized_get(&req->rq_pill,
 252                                                    &RMF_ACL,
 253                                                    body->aclsize);
 254                 if (acl == NULL)
 255                         RETURN(-EPROTO);
 256
 257                 LASSERT(ma->ma_acl != NULL);
 258                 LASSERT(ma->ma_acl_size == body->aclsize);
 259                 memcpy(ma->ma_acl, acl, ma->ma_acl_size);
 260                 ma->ma_valid |= MA_ACL_DEF;
 261         }
 262 #endif
 263
 264         RETURN(0);
 265 }
 266
 267 /**
 268  * The md_object_operations::moo_attr_get() in mdc.
 269  */
 270 static int mdc_attr_get(const struct lu_env *env, struct md_object *mo,
 271                         struct md_attr *ma)
 272 {
 273         struct mdc_device *mc = md2mdc_dev(md_obj2dev(mo));
 274         struct mdc_thread_info *mci;
 275         int rc;
 276         ENTRY;
 277
 278         mci = lu_context_key_get(&env->le_ctx, &mdc_thread_key);
 279         LASSERT(mci);
 280
 281         memset(&mci->mci_opdata, 0, sizeof(mci->mci_opdata));
 282
 283         memcpy(&mci->mci_opdata.op_fid1, lu_object_fid(&mo->mo_lu),
 284                sizeof (struct lu_fid));
 285         mci->mci_opdata.op_valid = OBD_MD_FLMODE | OBD_MD_FLUID |
 286                                    OBD_MD_FLGID | OBD_MD_FLFLAGS |
 287                                    OBD_MD_FLCROSSREF;
 288
 289         rc = md_getattr(mc->mc_desc.cl_exp, &mci->mci_opdata, &mci->mci_req);
 290         if (rc == 0) {
 291                 /* get attr from request */
 292                 rc = mdc_req2attr_update(env, ma);
 293         }
 294
 295         ptlrpc_req_finished(mci->mci_req);
 296
 297         RETURN(rc);
 298 }
 299
 300 /**
 301  * Helper to init timspec \a t.
 302  */
 303 static inline struct timespec *mdc_attr_time(struct timespec *t, obd_time seconds)
 304 {
 305         t->tv_sec = seconds;
 306         t->tv_nsec = 0;
 307         return t;
 308 }
 309
 310 /**
 311  * The md_object_operations::moo_attr_set() in mdc.
 312  *
 313  * \note It is only used for set ctime when rename's source on remote MDS.
 314  */
 315 static int mdc_attr_set(const struct lu_env *env, struct md_object *mo,
 316                         const struct md_attr *ma)
 317 {
 318         struct mdc_device *mc = md2mdc_dev(md_obj2dev(mo));
 319         const struct lu_attr *la = &ma->ma_attr;
 320         struct mdc_thread_info *mci;
 321         struct md_ucred *uc = md_ucred(env);
 322         int rc;
 323         ENTRY;
 324
 325         LASSERT(ma->ma_attr.la_valid & LA_CTIME);
 326
 327         mci = lu_context_key_get(&env->le_ctx, &mdc_thread_key);
 328         LASSERT(mci);
 329
 330         memset(&mci->mci_opdata, 0, sizeof(mci->mci_opdata));
 331
 332         mci->mci_opdata.op_fid1 = *lu_object_fid(&mo->mo_lu);
 333         mdc_attr_time(&mci->mci_opdata.op_attr.ia_ctime, la->la_ctime);
 334         mci->mci_opdata.op_attr.ia_mode = la->la_mode;
 335         mci->mci_opdata.op_attr.ia_valid = ATTR_CTIME_SET;
 336         if (uc &&
 337             ((uc->mu_valid == UCRED_OLD) || (uc->mu_valid == UCRED_NEW))) {
 338                 mci->mci_opdata.op_fsuid = uc->mu_fsuid;
 339                 mci->mci_opdata.op_fsgid = uc->mu_fsgid;
 340                 mci->mci_opdata.op_cap = uc->mu_cap;
 341                 if (uc->mu_ginfo || (uc->mu_valid == UCRED_OLD)) {
 342                         mci->mci_opdata.op_suppgids[0] = uc->mu_suppgids[0];
 343                         mci->mci_opdata.op_suppgids[1] = uc->mu_suppgids[1];
 344                 } else {
 345                         mci->mci_opdata.op_suppgids[0] =
 346                                 mci->mci_opdata.op_suppgids[1] = -1;
 347                 }
 348         } else {
 349                 mci->mci_opdata.op_fsuid = la->la_uid;
 350                 mci->mci_opdata.op_fsgid = la->la_gid;
 351                 mci->mci_opdata.op_cap = cfs_curproc_cap_pack();
 352                 mci->mci_opdata.op_suppgids[0] =
 353                                 mci->mci_opdata.op_suppgids[1] = -1;
 354         }
 355
 356         rc = md_setattr(mc->mc_desc.cl_exp, &mci->mci_opdata,
 357                         NULL, 0, NULL, 0, &mci->mci_req, NULL);
 358
 359         ptlrpc_req_finished(mci->mci_req);
 360
 361         RETURN(rc);
 362 }
 363
 364 /**
 365  * The md_object_operations::moo_object_create() in mdc.
 366  */
 367 static int mdc_object_create(const struct lu_env *env,
 368                              struct md_object *mo,
 369                              const struct md_op_spec *spec,
 370                              struct md_attr *ma)
 371 {
 372         struct mdc_device *mc = md2mdc_dev(md_obj2dev(mo));
 373         struct lu_attr *la = &ma->ma_attr;
 374         struct mdc_thread_info *mci;
 375         const void *symname;
 376         struct md_ucred *uc = md_ucred(env);
 377         int rc, symlen;
 378         uid_t uid;
 379         gid_t gid;
 380         cfs_cap_t cap;
 381         ENTRY;
 382
 383         LASSERT(S_ISDIR(la->la_mode));
 384         LASSERT(spec->u.sp_pfid != NULL);
 385
 386         mci = mdc_info_init(env);
 387         mci->mci_opdata.op_bias = MDS_CROSS_REF;
 388         mci->mci_opdata.op_fid2 = *lu_object_fid(&mo->mo_lu);
 389
 390         /* Parent fid is needed to create dotdot on the remote node. */
 391         mci->mci_opdata.op_fid1 = *(spec->u.sp_pfid);
 392         mci->mci_opdata.op_mod_time = la->la_ctime;
 393         if (uc &&
 394             ((uc->mu_valid == UCRED_OLD) || (uc->mu_valid == UCRED_NEW))) {
 395                 uid = uc->mu_fsuid;
 396                 if (la->la_mode & S_ISGID)
 397                         gid = la->la_gid;
 398                 else
 399                         gid = uc->mu_fsgid;
 400                 cap = uc->mu_cap;
 401                 if (uc->mu_ginfo || (uc->mu_valid == UCRED_OLD))
 402                         mci->mci_opdata.op_suppgids[0] = uc->mu_suppgids[0];
 403                 else
 404                         mci->mci_opdata.op_suppgids[0] = -1;
 405         } else {
 406                 uid = la->la_uid;
 407                 gid = la->la_gid;
 408                 cap = 0;
 409                 mci->mci_opdata.op_suppgids[0] = -1;
 410         }
 411
 412         /* get data from spec */
 413         if (spec->sp_cr_flags & MDS_CREATE_SLAVE_OBJ) {
 414                 symname = spec->u.sp_ea.eadata;
 415                 symlen = spec->u.sp_ea.eadatalen;
 416                 mci->mci_opdata.op_fid1 = *(spec->u.sp_ea.fid);
 417                 mci->mci_opdata.op_flags |= MDS_CREATE_SLAVE_OBJ;
 418 #ifdef CONFIG_FS_POSIX_ACL
 419         } else if (spec->sp_cr_flags & MDS_CREATE_RMT_ACL) {
 420                 symname = spec->u.sp_ea.eadata;
 421                 symlen = spec->u.sp_ea.eadatalen;
 422                 mci->mci_opdata.op_fid1 = *(spec->u.sp_ea.fid);
 423                 mci->mci_opdata.op_flags |= MDS_CREATE_RMT_ACL;
 424 #endif
 425         } else {
 426                 symname = spec->u.sp_symname;
 427                 symlen = symname ? strlen(symname) + 1 : 0;
 428         }
 429
 430         rc = md_create(mc->mc_desc.cl_exp, &mci->mci_opdata,
 431                        symname, symlen, la->la_mode, uid, gid,
 432                        cap, la->la_rdev, &mci->mci_req);
 433
 434         if (rc == 0) {
 435                 /* get attr from request */
 436                 rc = mdc_req2attr_update(env, ma);
 437         }
 438
 439         ptlrpc_req_finished(mci->mci_req);
 440
 441         RETURN(rc);
 442 }
 443
 444 /**
 445  * The md_object_operations::moo_ref_add() in mdc.
 446  */
 447 static int mdc_ref_add(const struct lu_env *env, struct md_object *mo,
 448                        const struct md_attr *ma)
 449 {
 450         struct mdc_device *mc = md2mdc_dev(md_obj2dev(mo));
 451         const struct lu_attr *la = &ma->ma_attr;
 452         struct mdc_thread_info *mci;
 453         struct md_ucred *uc = md_ucred(env);
 454         int rc;
 455         ENTRY;
 456
 457         mci = lu_context_key_get(&env->le_ctx, &mdc_thread_key);
 458         LASSERT(mci);
 459
 460         memset(&mci->mci_opdata, 0, sizeof(mci->mci_opdata));
 461         mci->mci_opdata.op_bias = MDS_CROSS_REF;
 462         mci->mci_opdata.op_fid1 = *lu_object_fid(&mo->mo_lu);
 463         mci->mci_opdata.op_mod_time = la->la_ctime;
 464         if (uc &&
 465             ((uc->mu_valid == UCRED_OLD) || (uc->mu_valid == UCRED_NEW))) {
 466                 mci->mci_opdata.op_fsuid = uc->mu_fsuid;
 467                 mci->mci_opdata.op_fsgid = uc->mu_fsgid;
 468                 mci->mci_opdata.op_cap = uc->mu_cap;
 469                 if (uc->mu_ginfo || (uc->mu_valid == UCRED_OLD)) {
 470                         mci->mci_opdata.op_suppgids[0] = uc->mu_suppgids[0];
 471                         mci->mci_opdata.op_suppgids[1] = uc->mu_suppgids[1];
 472                 } else {
 473                         mci->mci_opdata.op_suppgids[0] =
 474                                 mci->mci_opdata.op_suppgids[1] = -1;
 475                 }
 476         } else {
 477                 mci->mci_opdata.op_fsuid = la->la_uid;
 478                 mci->mci_opdata.op_fsgid = la->la_gid;
 479                 mci->mci_opdata.op_cap = cfs_curproc_cap_pack();
 480                 mci->mci_opdata.op_suppgids[0] =
 481                                 mci->mci_opdata.op_suppgids[1] = -1;
 482         }
 483
 484
 485         rc = md_link(mc->mc_desc.cl_exp, &mci->mci_opdata, &mci->mci_req);
 486
 487         ptlrpc_req_finished(mci->mci_req);
 488
 489         RETURN(rc);
 490 }
 491
 492 /**
 493  * The md_object_operations::moo_ref_del() in mdc.
 494  */
 495 static int mdc_ref_del(const struct lu_env *env, struct md_object *mo,
 496                        struct md_attr *ma)
 497 {
 498         struct mdc_device *mc = md2mdc_dev(md_obj2dev(mo));
 499         struct lu_attr *la = &ma->ma_attr;
 500         struct mdc_thread_info *mci;
 501         struct md_ucred *uc = md_ucred(env);
 502         int rc;
 503         ENTRY;
 504
 505         mci = mdc_info_init(env);
 506         mci->mci_opdata.op_bias = MDS_CROSS_REF;
 507         if (ma->ma_attr_flags & MDS_VTX_BYPASS)
 508                 mci->mci_opdata.op_bias |= MDS_VTX_BYPASS;
 509         else
 510                 mci->mci_opdata.op_bias &= ~MDS_VTX_BYPASS;
 511         mci->mci_opdata.op_fid1 = *lu_object_fid(&mo->mo_lu);
 512         mci->mci_opdata.op_mode = la->la_mode;
 513         mci->mci_opdata.op_mod_time = la->la_ctime;
 514         if (uc &&
 515             ((uc->mu_valid == UCRED_OLD) || (uc->mu_valid == UCRED_NEW))) {
 516                 mci->mci_opdata.op_fsuid = uc->mu_fsuid;
 517                 mci->mci_opdata.op_fsgid = uc->mu_fsgid;
 518                 mci->mci_opdata.op_cap = uc->mu_cap;
 519                 if (uc->mu_ginfo || (uc->mu_valid == UCRED_OLD))
 520                         mci->mci_opdata.op_suppgids[0] = uc->mu_suppgids[0];
 521                 else
 522                         mci->mci_opdata.op_suppgids[0] = -1;
 523         } else {
 524                 mci->mci_opdata.op_fsuid = la->la_uid;
 525                 mci->mci_opdata.op_fsgid = la->la_gid;
 526                 mci->mci_opdata.op_cap = cfs_curproc_cap_pack();
 527                 mci->mci_opdata.op_suppgids[0] = -1;
 528         }
 529
 530         rc = md_unlink(mc->mc_desc.cl_exp, &mci->mci_opdata, &mci->mci_req);
 531         if (rc == 0) {
 532                 /* get attr from request */
 533                 rc = mdc_req2attr_update(env, ma);
 534         }
 535
 536         ptlrpc_req_finished(mci->mci_req);
 537
 538         RETURN(rc);
 539 }
 540
 541 #ifdef HAVE_SPLIT_SUPPORT
 542 /** Send page with directory entries to another MDS. */
 543 int mdc_send_page(struct cmm_device *cm, const struct lu_env *env,
 544                   struct md_object *mo, struct page *page, __u32 offset)
 545 {
 546         struct mdc_device *mc = md2mdc_dev(md_obj2dev(mo));
 547         int rc;
 548         ENTRY;
 549
 550         rc = mdc_sendpage(mc->mc_desc.cl_exp, lu_object_fid(&mo->mo_lu),
 551                           page, offset);
 552         CDEBUG(rc ? D_ERROR : D_INFO, "send page %p  offset %d fid "DFID
 553                " rc %d \n", page, offset, PFID(lu_object_fid(&mo->mo_lu)), rc);
 554         RETURN(rc);
 555 }
 556 #endif
 557
 558 /**
 559  * Instance of md_object_operations for mdc.
 560  */
 561 static const struct md_object_operations mdc_mo_ops = {
 562         .moo_attr_get       = mdc_attr_get,
 563         .moo_attr_set       = mdc_attr_set,
 564         .moo_object_create  = mdc_object_create,
 565         .moo_ref_add        = mdc_ref_add,
 566         .moo_ref_del        = mdc_ref_del,
 567 };
 568 /** @} */
 569
 570 /**
 571  * \name The set of md_dir_operations.
 572  * @{
 573  */
 574 /**
 575  * The md_dir_operations::mdo_rename_tgt in mdc.
 576  */
 577 static int mdc_rename_tgt(const struct lu_env *env, struct md_object *mo_p,
 578                           struct md_object *mo_t, const struct lu_fid *lf,
 579                           const struct lu_name *lname, struct md_attr *ma)
 580 {
 581         struct mdc_device *mc = md2mdc_dev(md_obj2dev(mo_p));
 582         struct lu_attr *la = &ma->ma_attr;
 583         struct mdc_thread_info *mci;
 584         struct md_ucred *uc = md_ucred(env);
 585         int rc;
 586         ENTRY;
 587
 588         mci = mdc_info_init(env);
 589         mci->mci_opdata.op_bias = MDS_CROSS_REF;
 590         if (ma->ma_attr_flags & MDS_VTX_BYPASS)
 591                 mci->mci_opdata.op_bias |= MDS_VTX_BYPASS;
 592         else
 593                 mci->mci_opdata.op_bias &= ~MDS_VTX_BYPASS;
 594         mci->mci_opdata.op_fid1 = *lu_object_fid(&mo_p->mo_lu);
 595         mci->mci_opdata.op_fid2 = *lf;
 596         mci->mci_opdata.op_mode = la->la_mode;
 597         mci->mci_opdata.op_mod_time = la->la_ctime;
 598         if (uc &&
 599             ((uc->mu_valid == UCRED_OLD) || (uc->mu_valid == UCRED_NEW))) {
 600                 mci->mci_opdata.op_fsuid = uc->mu_fsuid;
 601                 mci->mci_opdata.op_fsgid = uc->mu_fsgid;
 602                 mci->mci_opdata.op_cap = uc->mu_cap;
 603                 if (uc->mu_ginfo || (uc->mu_valid == UCRED_OLD)) {
 604                         mci->mci_opdata.op_suppgids[0] = uc->mu_suppgids[0];
 605                         mci->mci_opdata.op_suppgids[1] = uc->mu_suppgids[1];
 606                 } else {
 607                         mci->mci_opdata.op_suppgids[0] =
 608                                 mci->mci_opdata.op_suppgids[1] = -1;
 609                 }
 610         } else {
 611                 mci->mci_opdata.op_fsuid = la->la_uid;
 612                 mci->mci_opdata.op_fsgid = la->la_gid;
 613                 mci->mci_opdata.op_cap = cfs_curproc_cap_pack();
 614                 mci->mci_opdata.op_suppgids[0] =
 615                                 mci->mci_opdata.op_suppgids[1] = -1;
 616         }
 617
 618         rc = md_rename(mc->mc_desc.cl_exp, &mci->mci_opdata, NULL, 0,
 619                        lname->ln_name, lname->ln_namelen, &mci->mci_req);
 620         if (rc == 0) {
 621                 /* get attr from request */
 622                 mdc_req2attr_update(env, ma);
 623         }
 624
 625         ptlrpc_req_finished(mci->mci_req);
 626
 627         RETURN(rc);
 628 }
 629 /**
 630  * Check the fids are not relatives.
 631  * The md_dir_operations::mdo_is_subdir() in mdc.
 632  *
 633  * Return resulting fid in sfid.
 634  * \retval \a sfid = 0 fids are not relatives
 635  * \retval \a sfid = FID at which search stopped
 636  */
 637 static int mdc_is_subdir(const struct lu_env *env, struct md_object *mo,
 638                          const struct lu_fid *fid, struct lu_fid *sfid)
 639 {
 640         struct mdc_device *mc = md2mdc_dev(md_obj2dev(mo));
 641         struct mdc_thread_info *mci;
 642         struct mdt_body *body;
 643         int rc;
 644         ENTRY;
 645
 646         mci = mdc_info_init(env);
 647
 648         rc = md_is_subdir(mc->mc_desc.cl_exp, lu_object_fid(&mo->mo_lu),
 649                           fid, &mci->mci_req);
 650         if (rc == 0 || rc == -EREMOTE) {
 651                 body = req_capsule_server_get(&mci->mci_req->rq_pill,
 652                                               &RMF_MDT_BODY);
 653                 LASSERT(body->valid & OBD_MD_FLID);
 654
 655                 CDEBUG(D_INFO, "Remote mdo_is_subdir(), new src "DFID"\n",
 656                        PFID(&body->fid1));
 657                 *sfid = body->fid1;
 658         }
 659         ptlrpc_req_finished(mci->mci_req);
 660         RETURN(rc);
 661 }
 662
 663 /** Instance of md_dir_operations for mdc. */
 664 static const struct md_dir_operations mdc_dir_ops = {
 665         .mdo_is_subdir   = mdc_is_subdir,
 666         .mdo_rename_tgt  = mdc_rename_tgt
 667 };
 668 /** @} */