lustre/cmm/mdc_object.c

   1 /* -*- mode: c; c-basic-offset: 8; indent-tabs-mode: nil; -*-
   2  * vim:expandtab:shiftwidth=8:tabstop=8:
   3  *
   4  *  lustre/cmm/mdc_object.c
   5  *  Lustre Cluster Metadata Manager (cmm)
   6  *
   7  *  Copyright (c) 2006 Cluster File Systems, Inc.
   8  *   Author: Mike Pershin <tappro@clusterfs.com>
   9  *
  10  *   This file is part of the Lustre file system, http://www.lustre.org
  11  *   Lustre is a trademark of Cluster File Systems, Inc.
  12  *
  13  *   You may have signed or agreed to another license before downloading
  14  *   this software.  If so, you are bound by the terms and conditions
  15  *   of that agreement, and the following does not apply to you.  See the
  16  *   LICENSE file included with this distribution for more information.
  17  *
  18  *   If you did not agree to a different license, then this copy of Lustre
  19  *   is open source software; you can redistribute it and/or modify it
  20  *   under the terms of version 2 of the GNU General Public License as
  21  *   published by the Free Software Foundation.
  22  *
  23  *   In either case, Lustre is distributed in the hope that it will be
  24  *   useful, but WITHOUT ANY WARRANTY; without even the implied warranty
  25  *   of MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  26  *   license text for more details.
  27  */
  28
  29 #ifndef EXPORT_SYMTAB
  30 # define EXPORT_SYMTAB
  31 #endif
  32
  33 #define DEBUG_SUBSYSTEM S_MDS
  34 #include <obd_support.h>
  35 #include <lustre_lib.h>
  36 #include <obd_class.h>
  37 #include <lustre_mdc.h>
  38 #include "cmm_internal.h"
  39 #include "mdc_internal.h"
  40
  41 static struct md_object_operations mdc_mo_ops;
  42 static struct md_dir_operations mdc_dir_ops;
  43 static struct lu_object_operations mdc_obj_ops;
  44
  45 extern struct lu_context_key mdc_thread_key;
  46
  47 struct lu_object *mdc_object_alloc(const struct lu_env *env,
  48                                    const struct lu_object_header *hdr,
  49                                    struct lu_device *ld)
  50 {
  51         struct mdc_object *mco;
  52         ENTRY;
  53
  54         OBD_ALLOC_PTR(mco);
  55         if (mco != NULL) {
  56                 struct lu_object *lo;
  57
  58                 lo = &mco->mco_obj.mo_lu;
  59                 lu_object_init(lo, NULL, ld);
  60                 mco->mco_obj.mo_ops = &mdc_mo_ops;
  61                 mco->mco_obj.mo_dir_ops = &mdc_dir_ops;
  62                 lo->lo_ops = &mdc_obj_ops;
  63                 RETURN(lo);
  64         } else
  65                 RETURN(NULL);
  66 }
  67
  68 static void mdc_object_free(const struct lu_env *env, struct lu_object *lo)
  69 {
  70         struct mdc_object *mco = lu2mdc_obj(lo);
  71         lu_object_fini(lo);
  72         OBD_FREE_PTR(mco);
  73 }
  74
  75 static int mdc_object_init(const struct lu_env *env, struct lu_object *lo)
  76 {
  77         ENTRY;
  78         lo->lo_header->loh_attr |= LOHA_REMOTE;
  79         RETURN(0);
  80 }
  81
  82 static int mdc_object_print(const struct lu_env *env, void *cookie,
  83                             lu_printer_t p, const struct lu_object *lo)
  84 {
  85         return (*p)(env, cookie, LUSTRE_CMM_MDC_NAME"-object@%p", lo);
  86 }
  87
  88 static struct lu_object_operations mdc_obj_ops = {
  89         .loo_object_init    = mdc_object_init,
  90         .loo_object_free    = mdc_object_free,
  91         .loo_object_print   = mdc_object_print,
  92 };
  93
  94 /* md_object_operations */
  95 static
  96 struct mdc_thread_info *mdc_info_get(const struct lu_env *env)
  97 {
  98         struct mdc_thread_info *mci;
  99
 100         mci = lu_context_key_get(&env->le_ctx, &mdc_thread_key);
 101         LASSERT(mci);
 102         return mci;
 103 }
 104
 105 static
 106 struct mdc_thread_info *mdc_info_init(const struct lu_env *env)
 107 {
 108         struct mdc_thread_info *mci = mdc_info_get(env);
 109         memset(mci, 0, sizeof(*mci));
 110         return mci;
 111 }
 112
 113 static void mdc_body2attr(struct mdt_body *body, struct md_attr *ma)
 114 {
 115         struct lu_attr *la = &ma->ma_attr;
 116         /* update time */
 117         if (body->valid & OBD_MD_FLCTIME && body->ctime >= la->la_ctime) {
 118                 la->la_ctime = body->ctime;
 119                 if (body->valid & OBD_MD_FLMTIME)
 120                         la->la_mtime = body->mtime;
 121         }
 122
 123         if (body->valid & OBD_MD_FLMODE)
 124                 la->la_mode = body->mode;
 125         if (body->valid & OBD_MD_FLSIZE)
 126                 la->la_size = body->size;
 127         if (body->valid & OBD_MD_FLBLOCKS)
 128                 la->la_blocks = body->blocks;
 129         if (body->valid & OBD_MD_FLUID)
 130                 la->la_uid = body->uid;
 131         if (body->valid & OBD_MD_FLGID)
 132                 la->la_gid = body->gid;
 133         if (body->valid & OBD_MD_FLFLAGS)
 134                 la->la_flags = body->flags;
 135         if (body->valid & OBD_MD_FLNLINK)
 136                 la->la_nlink = body->nlink;
 137         if (body->valid & OBD_MD_FLRDEV)
 138                 la->la_rdev = body->rdev;
 139
 140         la->la_valid = body->valid;
 141         ma->ma_valid = MA_INODE;
 142 }
 143
 144 static int mdc_req2attr_update(const struct lu_env *env,
 145                                struct md_attr *ma)
 146 {
 147         struct mdc_thread_info *mci;
 148         struct ptlrpc_request *req;
 149         struct mdt_body *body;
 150         struct lov_mds_md *lov;
 151         struct llog_cookie *cookie;
 152
 153         ENTRY;
 154         mci = mdc_info_get(env);
 155         req = mci->mci_req;
 156         LASSERT(req);
 157         body = lustre_msg_buf(req->rq_repmsg, REPLY_REC_OFF, sizeof(*body));
 158         LASSERT(body);
 159         mdc_body2attr(body, ma);
 160
 161         if (body->valid & OBD_MD_FLMDSCAPA) {
 162                 struct lustre_capa *capa;
 163
 164                 /* create for cross-ref will fetch mds capa from remote obj */
 165                 capa = lustre_msg_buf(req->rq_repmsg, REPLY_REC_OFF + 1,
 166                                       sizeof(*capa));
 167                 LASSERT(capa != NULL);
 168                 LASSERT(ma->ma_capa != NULL);
 169                 *ma->ma_capa = *capa;
 170         }
 171
 172         if (!(body->valid & OBD_MD_FLEASIZE))
 173                 RETURN(0);
 174
 175         if (body->eadatasize == 0) {
 176                 CERROR("OBD_MD_FLEASIZE is set but eadatasize is zero\n");
 177                 RETURN(-EPROTO);
 178         }
 179
 180         lov = lustre_swab_repbuf(req, REPLY_REC_OFF + 1,
 181                                  body->eadatasize, NULL);
 182         if (lov == NULL) {
 183                 CERROR("Can't unpack MDS EA data\n");
 184                 RETURN(-EPROTO);
 185         }
 186
 187         LASSERT(ma->ma_lmm != NULL);
 188         LASSERT(ma->ma_lmm_size >= body->eadatasize);
 189         ma->ma_lmm_size = body->eadatasize;
 190         memcpy(ma->ma_lmm, lov, ma->ma_lmm_size);
 191         ma->ma_valid |= MA_LOV;
 192
 193         if (!(body->valid & OBD_MD_FLCOOKIE))
 194                 RETURN(0);
 195
 196         if (body->aclsize == 0) {
 197                 CERROR("OBD_MD_FLCOOKIE is set but cookie size is zero\n");
 198                 RETURN(-EPROTO);
 199         }
 200
 201         cookie = lustre_msg_buf(req->rq_repmsg,
 202                                 REPLY_REC_OFF + 2, body->aclsize);
 203         if (cookie == NULL) {
 204                 CERROR("Can't unpack unlink cookie data\n");
 205                 RETURN(-EPROTO);
 206         }
 207
 208         LASSERT(ma->ma_cookie != NULL);
 209         LASSERT(ma->ma_cookie_size == body->aclsize);
 210         memcpy(ma->ma_cookie, cookie, ma->ma_cookie_size);
 211         ma->ma_valid |= MA_COOKIE;
 212         RETURN(0);
 213 }
 214
 215 static int mdc_attr_get(const struct lu_env *env, struct md_object *mo,
 216                         struct md_attr *ma)
 217 {
 218         struct mdc_device *mc = md2mdc_dev(md_obj2dev(mo));
 219         struct mdc_thread_info *mci;
 220         int rc;
 221         ENTRY;
 222
 223         mci = lu_context_key_get(&env->le_ctx, &mdc_thread_key);
 224         LASSERT(mci);
 225
 226         memset(&mci->mci_opdata, 0, sizeof(mci->mci_opdata));
 227
 228         rc = md_getattr(mc->mc_desc.cl_exp, lu_object_fid(&mo->mo_lu),
 229                         NULL, OBD_MD_FLMODE | OBD_MD_FLUID | OBD_MD_FLGID |
 230                         OBD_MD_FLFLAGS | OBD_MD_FLCROSSREF, 0, &mci->mci_req);
 231         if (rc == 0) {
 232                 /* get attr from request */
 233                 rc = mdc_req2attr_update(env, ma);
 234         }
 235
 236         ptlrpc_req_finished(mci->mci_req);
 237
 238         RETURN(rc);
 239 }
 240
 241 static inline struct timespec *mdc_attr_time(struct timespec *t, __u64 seconds)
 242 {
 243         t->tv_sec = seconds;
 244         t->tv_nsec = 0;
 245         return t;
 246 }
 247
 248 /*
 249  * XXX: It is only used for set ctime when rename's source on remote MDS.
 250  */
 251 static int mdc_attr_set(const struct lu_env *env, struct md_object *mo,
 252                         const struct md_attr *ma)
 253 {
 254         struct mdc_device *mc = md2mdc_dev(md_obj2dev(mo));
 255         const struct lu_attr *la = &ma->ma_attr;
 256         struct mdc_thread_info *mci;
 257         struct md_ucred *uc = md_ucred(env);
 258         int rc;
 259         ENTRY;
 260
 261         LASSERT(ma->ma_attr.la_valid & LA_CTIME);
 262
 263         mci = lu_context_key_get(&env->le_ctx, &mdc_thread_key);
 264         LASSERT(mci);
 265
 266         memset(&mci->mci_opdata, 0, sizeof(mci->mci_opdata));
 267
 268         mci->mci_opdata.op_fid1 = *lu_object_fid(&mo->mo_lu);
 269         mdc_attr_time(&mci->mci_opdata.op_attr.ia_ctime, la->la_ctime);
 270         mci->mci_opdata.op_attr.ia_mode = la->la_mode;
 271         mci->mci_opdata.op_attr.ia_valid = ATTR_CTIME_SET;
 272         if (uc &&
 273             ((uc->mu_valid == UCRED_OLD) || (uc->mu_valid == UCRED_NEW))) {
 274                 mci->mci_opdata.op_fsuid = uc->mu_fsuid;
 275                 mci->mci_opdata.op_fsgid = uc->mu_fsgid;
 276                 mci->mci_opdata.op_cap = uc->mu_cap;
 277                 if (uc->mu_ginfo || (uc->mu_valid == UCRED_OLD)) {
 278                         mci->mci_opdata.op_suppgids[0] = uc->mu_suppgids[0];
 279                         mci->mci_opdata.op_suppgids[1] = uc->mu_suppgids[1];
 280                 } else {
 281                         mci->mci_opdata.op_suppgids[0] =
 282                                 mci->mci_opdata.op_suppgids[1] = -1;
 283                 }
 284         } else {
 285                 mci->mci_opdata.op_fsuid = la->la_uid;
 286                 mci->mci_opdata.op_fsgid = la->la_gid;
 287                 mci->mci_opdata.op_cap = current->cap_effective;
 288                 mci->mci_opdata.op_suppgids[0] =
 289                                 mci->mci_opdata.op_suppgids[1] = -1;
 290         }
 291
 292         rc = md_setattr(mc->mc_desc.cl_exp, &mci->mci_opdata,
 293                         NULL, 0, NULL, 0, &mci->mci_req, NULL);
 294
 295         ptlrpc_req_finished(mci->mci_req);
 296
 297         RETURN(rc);
 298 }
 299
 300 static int mdc_object_create(const struct lu_env *env,
 301                              struct md_object *mo,
 302                              const struct md_op_spec *spec,
 303                              struct md_attr *ma)
 304 {
 305         struct mdc_device *mc = md2mdc_dev(md_obj2dev(mo));
 306         struct lu_attr *la = &ma->ma_attr;
 307         struct mdc_thread_info *mci;
 308         const void *symname;
 309         struct md_ucred *uc = md_ucred(env);
 310         int rc, symlen;
 311         uid_t uid;
 312         gid_t gid;
 313         __u32 cap;
 314         ENTRY;
 315
 316         LASSERT(S_ISDIR(la->la_mode));
 317         LASSERT(spec->u.sp_pfid != NULL);
 318
 319         mci = mdc_info_init(env);
 320         mci->mci_opdata.op_bias = MDS_CROSS_REF;
 321         mci->mci_opdata.op_fid2 = *lu_object_fid(&mo->mo_lu);
 322
 323         /* Parent fid is needed to create dotdot on the remote node. */
 324         mci->mci_opdata.op_fid1 = *(spec->u.sp_pfid);
 325         mci->mci_opdata.op_mod_time = la->la_ctime;
 326         if (uc &&
 327             ((uc->mu_valid == UCRED_OLD) || (uc->mu_valid == UCRED_NEW))) {
 328                 uid = uc->mu_fsuid;
 329                 if (la->la_mode & S_ISGID)
 330                         gid = la->la_gid;
 331                 else
 332                         gid = uc->mu_fsgid;
 333                 cap = uc->mu_cap;
 334                 if (uc->mu_ginfo || (uc->mu_valid == UCRED_OLD))
 335                         mci->mci_opdata.op_suppgids[0] = uc->mu_suppgids[0];
 336                 else
 337                         mci->mci_opdata.op_suppgids[0] = -1;
 338         } else {
 339                 uid = la->la_uid;
 340                 gid = la->la_gid;
 341                 cap = 0;
 342                 mci->mci_opdata.op_suppgids[0] = -1;
 343         }
 344
 345         /* get data from spec */
 346         if (spec->sp_cr_flags & MDS_CREATE_SLAVE_OBJ) {
 347                 symname = spec->u.sp_ea.eadata;
 348                 symlen = spec->u.sp_ea.eadatalen;
 349                 mci->mci_opdata.op_fid1 = *(spec->u.sp_ea.fid);
 350                 mci->mci_opdata.op_flags |= MDS_CREATE_SLAVE_OBJ;
 351 #ifdef CONFIG_FS_POSIX_ACL
 352         } else if (spec->sp_cr_flags & MDS_CREATE_RMT_ACL) {
 353                 symname = spec->u.sp_ea.eadata;
 354                 symlen = spec->u.sp_ea.eadatalen;
 355                 mci->mci_opdata.op_fid1 = *(spec->u.sp_ea.fid);
 356                 mci->mci_opdata.op_flags |= MDS_CREATE_RMT_ACL;
 357 #endif
 358         } else {
 359                 symname = spec->u.sp_symname;
 360                 symlen = symname ? strlen(symname) + 1 : 0;
 361         }
 362
 363         rc = md_create(mc->mc_desc.cl_exp, &mci->mci_opdata,
 364                        symname, symlen, la->la_mode, uid, gid,
 365                        cap, la->la_rdev, &mci->mci_req);
 366
 367         if (rc == 0) {
 368                 /* get attr from request */
 369                 rc = mdc_req2attr_update(env, ma);
 370         }
 371
 372         ptlrpc_req_finished(mci->mci_req);
 373
 374         RETURN(rc);
 375 }
 376
 377 static int mdc_ref_add(const struct lu_env *env, struct md_object *mo,
 378                        const struct md_attr *ma)
 379 {
 380         struct mdc_device *mc = md2mdc_dev(md_obj2dev(mo));
 381         const struct lu_attr *la = &ma->ma_attr;
 382         struct mdc_thread_info *mci;
 383         struct md_ucred *uc = md_ucred(env);
 384         int rc;
 385         ENTRY;
 386
 387         mci = lu_context_key_get(&env->le_ctx, &mdc_thread_key);
 388         LASSERT(mci);
 389
 390         memset(&mci->mci_opdata, 0, sizeof(mci->mci_opdata));
 391         mci->mci_opdata.op_bias = MDS_CROSS_REF;
 392         mci->mci_opdata.op_fid1 = *lu_object_fid(&mo->mo_lu);
 393         mci->mci_opdata.op_mod_time = la->la_ctime;
 394         if (uc &&
 395             ((uc->mu_valid == UCRED_OLD) || (uc->mu_valid == UCRED_NEW))) {
 396                 mci->mci_opdata.op_fsuid = uc->mu_fsuid;
 397                 mci->mci_opdata.op_fsgid = uc->mu_fsgid;
 398                 mci->mci_opdata.op_cap = uc->mu_cap;
 399                 if (uc->mu_ginfo || (uc->mu_valid == UCRED_OLD)) {
 400                         mci->mci_opdata.op_suppgids[0] = uc->mu_suppgids[0];
 401                         mci->mci_opdata.op_suppgids[1] = uc->mu_suppgids[1];
 402                 } else {
 403                         mci->mci_opdata.op_suppgids[0] =
 404                                 mci->mci_opdata.op_suppgids[1] = -1;
 405                 }
 406         } else {
 407                 mci->mci_opdata.op_fsuid = la->la_uid;
 408                 mci->mci_opdata.op_fsgid = la->la_gid;
 409                 mci->mci_opdata.op_cap = current->cap_effective;
 410                 mci->mci_opdata.op_suppgids[0] =
 411                                 mci->mci_opdata.op_suppgids[1] = -1;
 412         }
 413
 414
 415         rc = md_link(mc->mc_desc.cl_exp, &mci->mci_opdata, &mci->mci_req);
 416
 417         ptlrpc_req_finished(mci->mci_req);
 418
 419         RETURN(rc);
 420 }
 421
 422 static int mdc_ref_del(const struct lu_env *env, struct md_object *mo,
 423                        struct md_attr *ma)
 424 {
 425         struct mdc_device *mc = md2mdc_dev(md_obj2dev(mo));
 426         struct lu_attr *la = &ma->ma_attr;
 427         struct mdc_thread_info *mci;
 428         struct md_ucred *uc = md_ucred(env);
 429         int rc;
 430         ENTRY;
 431
 432         mci = mdc_info_init(env);
 433         mci->mci_opdata.op_bias = MDS_CROSS_REF;
 434         if (ma->ma_attr_flags & MDS_VTX_BYPASS)
 435                 mci->mci_opdata.op_bias |= MDS_VTX_BYPASS;
 436         else
 437                 mci->mci_opdata.op_bias &= ~MDS_VTX_BYPASS;
 438         mci->mci_opdata.op_fid1 = *lu_object_fid(&mo->mo_lu);
 439         mci->mci_opdata.op_mode = la->la_mode;
 440         mci->mci_opdata.op_mod_time = la->la_ctime;
 441         if (uc &&
 442             ((uc->mu_valid == UCRED_OLD) || (uc->mu_valid == UCRED_NEW))) {
 443                 mci->mci_opdata.op_fsuid = uc->mu_fsuid;
 444                 mci->mci_opdata.op_fsgid = uc->mu_fsgid;
 445                 mci->mci_opdata.op_cap = uc->mu_cap;
 446                 if (uc->mu_ginfo || (uc->mu_valid == UCRED_OLD))
 447                         mci->mci_opdata.op_suppgids[0] = uc->mu_suppgids[0];
 448                 else
 449                         mci->mci_opdata.op_suppgids[0] = -1;
 450         } else {
 451                 mci->mci_opdata.op_fsuid = la->la_uid;
 452                 mci->mci_opdata.op_fsgid = la->la_gid;
 453                 mci->mci_opdata.op_cap = current->cap_effective;
 454                 mci->mci_opdata.op_suppgids[0] = -1;
 455         }
 456
 457         rc = md_unlink(mc->mc_desc.cl_exp, &mci->mci_opdata, &mci->mci_req);
 458         if (rc == 0) {
 459                 /* get attr from request */
 460                 rc = mdc_req2attr_update(env, ma);
 461         }
 462
 463         ptlrpc_req_finished(mci->mci_req);
 464
 465         RETURN(rc);
 466 }
 467
 468 #ifdef HAVE_SPLIT_SUPPORT
 469 int mdc_send_page(struct cmm_device *cm, const struct lu_env *env,
 470                   struct md_object *mo, struct page *page, __u32 offset)
 471 {
 472         struct mdc_device *mc = md2mdc_dev(md_obj2dev(mo));
 473         int rc;
 474         ENTRY;
 475
 476         rc = mdc_sendpage(mc->mc_desc.cl_exp, lu_object_fid(&mo->mo_lu),
 477                           page, offset);
 478         CDEBUG(rc ? D_ERROR : D_INFO, "send page %p  offset %d fid "DFID
 479                " rc %d \n", page, offset, PFID(lu_object_fid(&mo->mo_lu)), rc);
 480         RETURN(rc);
 481 }
 482 #endif
 483
 484 static struct md_object_operations mdc_mo_ops = {
 485         .moo_attr_get       = mdc_attr_get,
 486         .moo_attr_set       = mdc_attr_set,
 487         .moo_object_create  = mdc_object_create,
 488         .moo_ref_add        = mdc_ref_add,
 489         .moo_ref_del        = mdc_ref_del,
 490 };
 491
 492 /* md_dir_operations */
 493 static int mdc_rename_tgt(const struct lu_env *env, struct md_object *mo_p,
 494                           struct md_object *mo_t, const struct lu_fid *lf,
 495                           const struct lu_name *lname, struct md_attr *ma)
 496 {
 497         struct mdc_device *mc = md2mdc_dev(md_obj2dev(mo_p));
 498         struct lu_attr *la = &ma->ma_attr;
 499         struct mdc_thread_info *mci;
 500         struct md_ucred *uc = md_ucred(env);
 501         int rc;
 502         ENTRY;
 503
 504         mci = mdc_info_init(env);
 505         mci->mci_opdata.op_bias = MDS_CROSS_REF;
 506         if (ma->ma_attr_flags & MDS_VTX_BYPASS)
 507                 mci->mci_opdata.op_bias |= MDS_VTX_BYPASS;
 508         else
 509                 mci->mci_opdata.op_bias &= ~MDS_VTX_BYPASS;
 510         mci->mci_opdata.op_fid1 = *lu_object_fid(&mo_p->mo_lu);
 511         mci->mci_opdata.op_fid2 = *lf;
 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                         mci->mci_opdata.op_suppgids[1] = uc->mu_suppgids[1];
 522                 } else {
 523                         mci->mci_opdata.op_suppgids[0] =
 524                                 mci->mci_opdata.op_suppgids[1] = -1;
 525                 }
 526         } else {
 527                 mci->mci_opdata.op_fsuid = la->la_uid;
 528                 mci->mci_opdata.op_fsgid = la->la_gid;
 529                 mci->mci_opdata.op_cap = current->cap_effective;
 530                 mci->mci_opdata.op_suppgids[0] =
 531                                 mci->mci_opdata.op_suppgids[1] = -1;
 532         }
 533
 534         rc = md_rename(mc->mc_desc.cl_exp, &mci->mci_opdata, NULL, 0,
 535                        lname->ln_name, lname->ln_namelen, &mci->mci_req);
 536         if (rc == 0) {
 537                 /* get attr from request */
 538                 mdc_req2attr_update(env, ma);
 539         }
 540
 541         ptlrpc_req_finished(mci->mci_req);
 542
 543         RETURN(rc);
 544 }
 545 /*
 546  * Return resulting fid in sfid
 547  * 0: fids are not relatives
 548  * fid: fid at which search stopped
 549  */
 550 static int mdc_is_subdir(const struct lu_env *env, struct md_object *mo,
 551                          const struct lu_fid *fid, struct lu_fid *sfid)
 552 {
 553         struct mdc_device *mc = md2mdc_dev(md_obj2dev(mo));
 554         struct mdc_thread_info *mci;
 555         struct mdt_body *body;
 556         int rc;
 557         ENTRY;
 558
 559         mci = mdc_info_init(env);
 560
 561         rc = md_is_subdir(mc->mc_desc.cl_exp, lu_object_fid(&mo->mo_lu),
 562                           fid, &mci->mci_req);
 563         if (rc == 0 || rc == -EREMOTE) {
 564                 body = lustre_msg_buf(mci->mci_req->rq_repmsg, REPLY_REC_OFF,
 565                                       sizeof(*body));
 566                 LASSERT(body->valid & OBD_MD_FLID);
 567
 568                 CDEBUG(D_INFO, "Remote mdo_is_subdir(), new src "DFID"\n",
 569                        PFID(&body->fid1));
 570                 *sfid = body->fid1;
 571         }
 572         ptlrpc_req_finished(mci->mci_req);
 573         RETURN(rc);
 574 }
 575
 576 static struct md_dir_operations mdc_dir_ops = {
 577         .mdo_is_subdir   = mdc_is_subdir,
 578         .mdo_rename_tgt  = mdc_rename_tgt
 579 };