lustre/mds/mds_reint.c

   1 /* -*- mode: c; c-basic-offset: 8; indent-tabs-mode: nil; -*-
   2  * vim:expandtab:shiftwidth=8:tabstop=8:
   3  *
   4  *  linux/mds/mds_reint.c
   5  *  Lustre Metadata Server (mds) reintegration routines
   6  *
   7  *  Copyright (C) 2002 Cluster File Systems, Inc.
   8  *
   9  *   This file is part of Lustre, http://www.lustre.org.
  10  *
  11  *   Lustre is free software; you can redistribute it and/or
  12  *   modify it under the terms of version 2 of the GNU General Public
  13  *   License as published by the Free Software Foundation.
  14  *
  15  *   Lustre is distributed in the hope that it will be useful,
  16  *   but WITHOUT ANY WARRANTY; without even the implied warranty of
  17  *   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  18  *   GNU General Public License for more details.
  19  *
  20  *   You should have received a copy of the GNU General Public License
  21  *   along with Lustre; if not, write to the Free Software
  22  *   Foundation, Inc., 675 Mass Ave, Cambridge, MA 02139, USA.
  23  *
  24  *  Author: Peter Braam <braam@clusterfs.com>
  25  *  Author: Andreas Dilger <adilger@clusterfs.com>
  26  */
  27
  28 #define EXPORT_SYMTAB
  29 #define DEBUG_SUBSYSTEM S_MDS
  30
  31 #include <linux/obd_support.h>
  32 #include <linux/obd_class.h>
  33 #include <linux/obd.h>
  34 #include <linux/lustre_lib.h>
  35 #include <linux/lustre_idl.h>
  36 #include <linux/lustre_mds.h>
  37 #include <linux/lustre_dlm.h>
  38 #include <linux/obd_class.h>
  39
  40 extern inline struct mds_obd *mds_req2mds(struct ptlrpc_request *req);
  41
  42 /* Assumes caller has already pushed us into the kernel context. */
  43 int mds_update_last_rcvd(struct mds_obd *mds, void *handle,
  44                          struct ptlrpc_request *req)
  45 {
  46         struct mds_export_data *med = &req->rq_export->exp_mds_data;
  47         struct mds_client_data *mcd = med->med_mcd;
  48         __u64 last_rcvd;
  49         loff_t off;
  50         int rc;
  51
  52         /* we don't allocate new transnos for replayed requests */
  53         if (req->rq_level == LUSTRE_CONN_RECOVD)
  54                 RETURN(0);
  55
  56         off = MDS_LR_CLIENT + med->med_off * MDS_LR_SIZE;
  57
  58         spin_lock(&mds->mds_last_lock);
  59         last_rcvd = ++mds->mds_last_rcvd;
  60         spin_unlock(&mds->mds_last_lock);
  61         req->rq_repmsg->transno = HTON__u64(last_rcvd);
  62         mcd->mcd_last_rcvd = cpu_to_le64(last_rcvd);
  63         mcd->mcd_mount_count = cpu_to_le64(mds->mds_mount_count);
  64         mcd->mcd_last_xid = cpu_to_le64(req->rq_xid);
  65
  66         mds_fs_set_last_rcvd(mds, handle);
  67         rc = lustre_fwrite(mds->mds_rcvd_filp, (char *)mcd, sizeof(*mcd), &off);
  68         CDEBUG(D_INODE, "wrote trans #"LPD64" for client '%s' at #%d: rc = "
  69                "%d\n", last_rcvd, mcd->mcd_uuid, med->med_off, rc);
  70
  71         if (rc == sizeof(*mcd))
  72                 rc = 0;
  73         else {
  74                 CERROR("error writing to last_rcvd file: rc = %d\n", rc);
  75                 if (rc >= 0)
  76                         rc = -EIO;
  77         }
  78
  79         return rc;
  80 }
  81
  82 /* In the write-back case, the client holds a lock on a subtree.
  83  * In the intent case, the client holds a lock on the child inode.
  84  * In the pathname case, the client (may) hold a lock on the child inode. */
  85 static int mds_reint_setattr(struct mds_update_record *rec, int offset,
  86                              struct ptlrpc_request *req)
  87 {
  88         struct mds_obd *mds = mds_req2mds(req);
  89         struct obd_device *obd = req->rq_export->exp_obd;
  90         struct mds_body *body;
  91         struct dentry *de;
  92         struct inode *inode;
  93         void *handle;
  94         struct lustre_handle child_lockh;
  95         int rc = 0, err;
  96
  97         if (req->rq_reqmsg->bufcount > offset + 1) {
  98                 struct dentry *dir;
  99                 struct lustre_handle dir_lockh;
 100                 char *name;
 101                 int namelen;
 102
 103                 /* a name was supplied by the client; fid1 is the directory */
 104                 dir = mds_fid2locked_dentry(obd, rec->ur_fid1, NULL, LCK_PR,
 105                                             &dir_lockh);
 106                 if (IS_ERR(dir)) {
 107                         LBUG();
 108                         GOTO(out_setattr, rc = PTR_ERR(dir));
 109                 }
 110
 111                 name = lustre_msg_buf(req->rq_reqmsg, offset + 1);
 112                 namelen = req->rq_reqmsg->buflens[offset + 1] - 1;
 113                 de = mds_name2locked_dentry(obd, dir, NULL, name, namelen,
 114                                             0, &child_lockh, LCK_PR);
 115                 l_dput(dir);
 116                 if (IS_ERR(de)) {
 117                         LBUG();
 118                         GOTO(out_setattr_de, rc = PTR_ERR(de));
 119                 }
 120         } else {
 121                 de = mds_fid2dentry(mds, rec->ur_fid1, NULL);
 122                 if (!de || IS_ERR(de)) {
 123                         LBUG();
 124                         GOTO(out_setattr_de, rc = -ESTALE);
 125                 }
 126         }
 127         inode = de->d_inode;
 128         CDEBUG(D_INODE, "ino %ld\n", inode->i_ino);
 129
 130         OBD_FAIL_WRITE(OBD_FAIL_MDS_REINT_SETATTR_WRITE,
 131                        to_kdev_t(inode->i_sb->s_dev));
 132
 133         handle = mds_fs_start(mds, inode, MDS_FSOP_SETATTR);
 134         if (!handle)
 135                 GOTO(out_setattr_de, rc = PTR_ERR(handle));
 136
 137         rc = mds_fs_setattr(mds, de, handle, &rec->ur_iattr);
 138
 139         if (offset) {
 140                 body = lustre_msg_buf(req->rq_repmsg, 1);
 141                 mds_pack_inode2fid(&body->fid1, inode);
 142                 mds_pack_inode2body(body, inode);
 143         }
 144
 145         if (!rc)
 146                 rc = mds_update_last_rcvd(mds, handle, req);
 147
 148         err = mds_fs_commit(mds, de->d_inode, handle);
 149         if (err) {
 150                 CERROR("error on commit: err = %d\n", err);
 151                 if (!rc)
 152                         rc = err;
 153         }
 154
 155         EXIT;
 156 out_setattr_de:
 157         l_dput(de);
 158 out_setattr:
 159         req->rq_status = rc;
 160         return 0;
 161 }
 162
 163 static int mds_reint_create(struct mds_update_record *rec, int offset,
 164                             struct ptlrpc_request *req)
 165 {
 166         struct dentry *de = NULL;
 167         struct mds_obd *mds = mds_req2mds(req);
 168         struct obd_device *obd = req->rq_export->exp_obd;
 169         struct dentry *dchild = NULL;
 170         struct inode *dir;
 171         void *handle;
 172         struct lustre_handle lockh;
 173         int rc = 0, err, lock_mode, type = rec->ur_mode & S_IFMT;
 174         ENTRY;
 175
 176         /* requests were at offset 2, replies go back at 1 */
 177         if (offset)
 178                 offset = 1;
 179
 180         LASSERT(!strcmp(req->rq_export->exp_obd->obd_type->typ_name, "mds"));
 181
 182         lock_mode = (req->rq_reqmsg->opc == MDS_REINT) ? LCK_CW : LCK_PW;
 183
 184         if (OBD_FAIL_CHECK(OBD_FAIL_MDS_REINT_CREATE))
 185                 GOTO(out_create, rc = -ESTALE);
 186
 187         de = mds_fid2locked_dentry(obd, rec->ur_fid1, NULL, lock_mode, &lockh);
 188         if (IS_ERR(de)) {
 189                 rc = PTR_ERR(de);
 190                 CERROR("parent lookup error %d\n", rc);
 191                 LBUG();
 192                 GOTO(out_create, rc);
 193         }
 194         dir = de->d_inode;
 195         CDEBUG(D_INODE, "parent ino %ld name %s mode %o\n",
 196                dir->i_ino, rec->ur_name, rec->ur_mode);
 197
 198         ldlm_lock_dump((void *)(unsigned long)lockh.addr);
 199
 200         down(&dir->i_sem);
 201         dchild = lookup_one_len(rec->ur_name, de, rec->ur_namelen - 1);
 202         if (IS_ERR(dchild)) {
 203                 CERROR("child lookup error %ld\n", PTR_ERR(dchild));
 204                 LBUG();
 205                 GOTO(out_create_de, rc = -ESTALE);
 206         }
 207
 208         if (dchild->d_inode) {
 209                 struct mds_body *body;
 210                 struct inode *inode = dchild->d_inode;
 211
 212                 CDEBUG(D_INODE, "child exists (dir %ld, name %s, ino %ld)\n",
 213                        dir->i_ino, rec->ur_name, dchild->d_inode->i_ino);
 214
 215                 /* XXX check that mode is correct? */
 216
 217                 body = lustre_msg_buf(req->rq_repmsg, offset);
 218                 mds_pack_inode2fid(&body->fid1, inode);
 219                 mds_pack_inode2body(body, inode);
 220                 if (S_ISREG(inode->i_mode)) {
 221                         struct lov_mds_md *lmm;
 222
 223                         lmm = lustre_msg_buf(req->rq_repmsg, offset + 1);
 224                         lmm->lmm_easize = mds->mds_max_mdsize;
 225
 226                         if (mds_fs_get_md(mds, inode, lmm) < 0) {
 227                                 CDEBUG(D_INFO,"No md for %ld: rc %d\n",
 228                                        inode->i_ino, rc);
 229                                 memset(lmm, 0, lmm->lmm_easize);
 230                         } else
 231                                 body->valid |= OBD_MD_FLEASIZE;
 232                 }
 233
 234                 /* This isn't an error for RECREATE. */
 235                 if (rec->ur_opcode & REINT_REPLAYING) {
 236                         CDEBUG(D_INODE, "EEXIST suppressed for REPLAYING\n");
 237                         rc = 0;
 238                 } else {
 239                         rc = -EEXIST;
 240                 }
 241                 GOTO(out_create_dchild, rc);
 242         }
 243
 244         OBD_FAIL_WRITE(OBD_FAIL_MDS_REINT_CREATE_WRITE,
 245                        to_kdev_t(dir->i_sb->s_dev));
 246
 247         if (dir->i_mode & S_ISGID) {
 248                 rec->ur_gid = dir->i_gid;
 249                 if (S_ISDIR(rec->ur_mode))
 250                         rec->ur_mode |= S_ISGID;
 251         }
 252
 253         switch (type) {
 254         case S_IFREG:{
 255                 handle = mds_fs_start(mds, dir, MDS_FSOP_CREATE);
 256                 if (!handle)
 257                         GOTO(out_create_dchild, PTR_ERR(handle));
 258                 rc = vfs_create(dir, dchild, rec->ur_mode);
 259                 EXIT;
 260                 break;
 261         }
 262         case S_IFDIR:{
 263                 handle = mds_fs_start(mds, dir, MDS_FSOP_MKDIR);
 264                 if (!handle)
 265                         GOTO(out_create_dchild, PTR_ERR(handle));
 266                 rc = vfs_mkdir(dir, dchild, rec->ur_mode);
 267                 EXIT;
 268                 break;
 269         }
 270         case S_IFLNK:{
 271                 handle = mds_fs_start(mds, dir, MDS_FSOP_SYMLINK);
 272                 if (!handle)
 273                         GOTO(out_create_dchild, PTR_ERR(handle));
 274                 rc = vfs_symlink(dir, dchild, rec->ur_tgt);
 275                 EXIT;
 276                 break;
 277         }
 278         case S_IFCHR:
 279         case S_IFBLK:
 280         case S_IFIFO:
 281         case S_IFSOCK:{
 282                 int rdev = rec->ur_rdev;
 283                 handle = mds_fs_start(mds, dir, MDS_FSOP_MKNOD);
 284                 if (!handle)
 285                         GOTO(out_create_dchild, PTR_ERR(handle));
 286                 rc = vfs_mknod(dir, dchild, rec->ur_mode, rdev);
 287                 EXIT;
 288                 break;
 289         }
 290         default:
 291                 CERROR("bad file type %o creating %s\n", type, rec->ur_name);
 292                 GOTO(out_create_dchild, rc = -EINVAL);
 293         }
 294
 295         if (rc) {
 296                 CDEBUG(D_INODE, "error during create: %d\n", rc);
 297                 GOTO(out_create_commit, rc);
 298         } else {
 299                 struct iattr iattr;
 300                 struct inode *inode = dchild->d_inode;
 301                 struct mds_body *body;
 302
 303                 CDEBUG(D_INODE, "created ino %ld\n", dchild->d_inode->i_ino);
 304
 305                 iattr.ia_atime = rec->ur_time;
 306                 iattr.ia_ctime = rec->ur_time;
 307                 iattr.ia_mtime = rec->ur_time;
 308                 iattr.ia_uid = rec->ur_uid;
 309                 iattr.ia_gid = rec->ur_gid;
 310                 iattr.ia_valid = ATTR_UID | ATTR_GID | ATTR_ATIME |
 311                         ATTR_MTIME | ATTR_CTIME;
 312
 313                 rc = mds_fs_setattr(mds, dchild, handle, &iattr);
 314                 if (rc) {
 315                         CERROR("error on setattr: rc = %d\n", rc);
 316                         /* XXX should we abort here in case of error? */
 317                 }
 318
 319                 rc = mds_update_last_rcvd(mds, handle, req);
 320                 if (rc) {
 321                         CERROR("error on mds_update_last_rcvd: rc = %d\n", rc);
 322                         GOTO(out_create_unlink, rc);
 323                 }
 324
 325                 body = lustre_msg_buf(req->rq_repmsg, offset);
 326                 mds_pack_inode2fid(&body->fid1, inode);
 327                 mds_pack_inode2body(body, inode);
 328         }
 329         EXIT;
 330 out_create_commit:
 331         err = mds_fs_commit(mds, dir, handle);
 332         if (err) {
 333                 CERROR("error on commit: err = %d\n", err);
 334                 if (!rc)
 335                         rc = err;
 336         }
 337 out_create_dchild:
 338         l_dput(dchild);
 339         ldlm_lock_decref(&lockh, lock_mode);
 340 out_create_de:
 341         up(&dir->i_sem);
 342         l_dput(de);
 343 out_create:
 344         req->rq_status = rc;
 345         return 0;
 346
 347 out_create_unlink:
 348         /* Destroy the file we just created.  This should not need extra
 349          * journal credits, as we have already modified all of the blocks
 350          * needed in order to create the file in the first place.
 351          */
 352         switch (type) {
 353         case S_IFDIR:
 354                 err = vfs_rmdir(dir, dchild);
 355                 if (err)
 356                         CERROR("failed rmdir in error path: rc = %d\n", err);
 357                 break;
 358         default:
 359                 err = vfs_unlink(dir, dchild);
 360                 if (err)
 361                         CERROR("failed unlink in error path: rc = %d\n", err);
 362                 break;
 363         }
 364
 365         goto out_create_commit;
 366 }
 367
 368 static int mds_reint_unlink(struct mds_update_record *rec, int offset,
 369                             struct ptlrpc_request *req)
 370 {
 371         struct dentry *de = NULL;
 372         struct dentry *dchild = NULL;
 373         struct mds_obd *mds = mds_req2mds(req);
 374         struct obd_device *obd = req->rq_export->exp_obd;
 375         struct mds_body *body = NULL;
 376         char *name;
 377         struct inode *dir, *inode;
 378         struct lustre_handle lockh, child_lockh;
 379         void *handle;
 380         int namelen, lock_mode, err, rc = 0;
 381         ENTRY;
 382
 383         /* a name was supplied by the client; fid1 is the directory */
 384         lock_mode = (req->rq_reqmsg->opc == MDS_REINT) ? LCK_PW : LCK_PW;
 385         de = mds_fid2locked_dentry(obd, rec->ur_fid1, NULL, lock_mode, &lockh);
 386         if (IS_ERR(de)) {
 387                 LBUG();
 388                 RETURN(PTR_ERR(de));
 389         }
 390
 391         if (OBD_FAIL_CHECK(OBD_FAIL_MDS_REINT_UNLINK))
 392                 GOTO(out_unlink, rc = -ENOENT);
 393
 394         name = lustre_msg_buf(req->rq_reqmsg, offset + 1);
 395         namelen = req->rq_reqmsg->buflens[offset + 1] - 1;
 396 #warning "FIXME: if mds_name2locked_dentry decrefs this lock, we must not"
 397         memcpy(&child_lockh, &lockh, sizeof(child_lockh));
 398         dchild = mds_name2locked_dentry(obd, de, NULL, name, namelen,
 399                                         LCK_EX, &child_lockh, lock_mode);
 400
 401         if (IS_ERR(dchild)) {
 402                 LBUG();
 403                 GOTO(out_unlink, rc = PTR_ERR(dchild));
 404         }
 405
 406         dir = de->d_inode;
 407         inode = dchild->d_inode;
 408         CDEBUG(D_INODE, "parent ino %ld\n", dir->i_ino);
 409
 410         if (!inode) {
 411                 CDEBUG(D_INODE, "child doesn't exist (dir %ld, name %s\n",
 412                        dir->i_ino, rec->ur_name);
 413                 /* going to out_unlink_cancel causes an LBUG, don't know why */
 414                 GOTO(out_unlink_dchild, rc = -ENOENT);
 415         }
 416
 417         if (offset) {
 418                 /* XXX offset? */
 419                 offset = 1;
 420
 421                 body = lustre_msg_buf(req->rq_repmsg, offset);
 422                 mds_pack_inode2fid(&body->fid1, inode);
 423                 mds_pack_inode2body(body, inode);
 424         }
 425
 426         OBD_FAIL_WRITE(OBD_FAIL_MDS_REINT_UNLINK_WRITE,
 427                        to_kdev_t(dir->i_sb->s_dev));
 428
 429         switch (rec->ur_mode /* & S_IFMT ? */) {
 430         case S_IFDIR:
 431                 handle = mds_fs_start(mds, dir, MDS_FSOP_RMDIR);
 432                 if (!handle)
 433                         GOTO(out_unlink_cancel, rc = PTR_ERR(handle));
 434                 rc = vfs_rmdir(dir, dchild);
 435                 break;
 436         case S_IFREG:
 437                 /* get OBD EA data first so client can also destroy object */
 438                 if ((inode->i_mode & S_IFMT) == S_IFREG && offset) {
 439                         struct lov_mds_md *lmm;
 440
 441                         lmm = lustre_msg_buf(req->rq_repmsg, offset + 1);
 442                         lmm->lmm_easize = mds->mds_max_mdsize;
 443                         if ((rc = mds_fs_get_md(mds, inode, lmm)) < 0) {
 444                                 CDEBUG(D_INFO, "No md for ino %ld: rc = %d\n",
 445                                        inode->i_ino, rc);
 446                                 memset(lmm, 0, lmm->lmm_easize);
 447                         } else
 448                                 body->valid |= OBD_MD_FLEASIZE;
 449                 }
 450                 /* no break */
 451         case S_IFLNK:
 452         case S_IFCHR:
 453         case S_IFBLK:
 454         case S_IFIFO:
 455         case S_IFSOCK:
 456                 handle = mds_fs_start(mds, dir, MDS_FSOP_UNLINK);
 457                 if (!handle)
 458                         GOTO(out_unlink_cancel, rc = PTR_ERR(handle));
 459                 rc = vfs_unlink(dir, dchild);
 460                 break;
 461         default:
 462                 CERROR("bad file type %o unlinking %s\n", rec->ur_mode, name);
 463                 handle = NULL;
 464                 LBUG();
 465                 GOTO(out_unlink_cancel, rc = -EINVAL);
 466         }
 467
 468         if (!rc)
 469                 rc = mds_update_last_rcvd(mds, handle, req);
 470         err = mds_fs_commit(mds, dir, handle);
 471         if (err) {
 472                 CERROR("error on commit: err = %d\n", err);
 473                 if (!rc)
 474                         rc = err;
 475         }
 476
 477         EXIT;
 478
 479 out_unlink_cancel:
 480         ldlm_lock_decref(&child_lockh, LCK_EX);
 481         err = ldlm_cli_cancel(&child_lockh);
 482         if (err < 0) {
 483                 CERROR("failed to cancel child inode lock: err = %d\n", err);
 484                 if (!rc)
 485                         rc = -ENOLCK;   /*XXX translate LDLM lock error */
 486         }
 487 out_unlink_dchild:
 488         l_dput(dchild);
 489         up(&dir->i_sem);
 490 out_unlink:
 491         ldlm_lock_decref(&lockh, lock_mode);
 492         l_dput(de);
 493         req->rq_status = rc;
 494         return 0;
 495 }
 496
 497 static int mds_reint_link(struct mds_update_record *rec, int offset,
 498                           struct ptlrpc_request *req)
 499 {
 500         struct obd_device *obd = req->rq_export->exp_obd;
 501         struct dentry *de_src = NULL;
 502         struct dentry *de_tgt_dir = NULL;
 503         struct dentry *dchild = NULL;
 504         struct mds_obd *mds = mds_req2mds(req);
 505         struct lustre_handle *handle, tgtlockh, srclockh;
 506         int lock_mode;
 507         __u64 res_id[3] = { 0 };
 508         int flags = 0;
 509         int rc = 0, err;
 510
 511         ENTRY;
 512         de_src = mds_fid2dentry(mds, rec->ur_fid1, NULL);
 513         if (IS_ERR(de_src) || OBD_FAIL_CHECK(OBD_FAIL_MDS_REINT_LINK)) {
 514                 GOTO(out_link, rc = -ESTALE);
 515         }
 516
 517         /* plan to change the link count on this inode: write lock */
 518         lock_mode = (req->rq_reqmsg->opc == MDS_REINT) ? LCK_PW : LCK_PW;
 519         res_id[0] = de_src->d_inode->i_ino;
 520
 521         rc = ldlm_lock_match(obd->obd_namespace, res_id, LDLM_PLAIN,
 522                              NULL, 0, lock_mode, &srclockh);
 523         if (rc == 0) {
 524                 LDLM_DEBUG_NOLOCK("enqueue res "LPU64, res_id[0]);
 525                 rc = ldlm_cli_enqueue(NULL, NULL, obd->obd_namespace, NULL,
 526                                       res_id, LDLM_PLAIN, NULL, 0, lock_mode,
 527                                       &flags, ldlm_completion_ast,
 528                                       mds_blocking_ast, NULL, 0, &srclockh);
 529                 if (rc != ELDLM_OK) {
 530                         CERROR("lock enqueue: err: %d\n", rc);
 531                         GOTO(out_link_src_put, rc = -EIO);
 532                 }
 533         } else
 534                 ldlm_lock_dump((void *)(unsigned long)srclockh.addr);
 535
 536         de_tgt_dir = mds_fid2dentry(mds, rec->ur_fid2, NULL);
 537         if (IS_ERR(de_tgt_dir)) {
 538                 GOTO(out_link_src, rc = -ESTALE);
 539         }
 540
 541         lock_mode = (req->rq_reqmsg->opc == MDS_REINT) ? LCK_PW : LCK_PW;
 542         res_id[0] = de_tgt_dir->d_inode->i_ino;
 543
 544         rc = ldlm_lock_match(obd->obd_namespace, res_id, LDLM_PLAIN,
 545                              NULL, 0, lock_mode, &tgtlockh);
 546         if (rc == 0) {
 547                 LDLM_DEBUG_NOLOCK("enqueue res "LPU64, res_id[0]);
 548                 rc = ldlm_cli_enqueue(NULL, NULL, obd->obd_namespace, NULL,
 549                                       res_id, LDLM_PLAIN, NULL, 0, lock_mode,
 550                                       &flags, ldlm_completion_ast,
 551                                       mds_blocking_ast, NULL, 0, &tgtlockh);
 552                 if (rc != ELDLM_OK) {
 553                         CERROR("lock enqueue: err: %d\n", rc);
 554                         GOTO(out_link_tgt_dir_put, rc = -EIO);
 555                 }
 556         } else
 557                 ldlm_lock_dump((void *)(unsigned long)tgtlockh.addr);
 558
 559         down(&de_tgt_dir->d_inode->i_sem);
 560         dchild = lookup_one_len(rec->ur_name, de_tgt_dir, rec->ur_namelen - 1);
 561         if (IS_ERR(dchild)) {
 562                 CERROR("child lookup error %ld\n", PTR_ERR(dchild));
 563                 GOTO(out_link_tgt_dir, rc = -ESTALE);
 564         }
 565
 566         if (dchild->d_inode) {
 567                 struct inode *inode = dchild->d_inode;
 568                 /* in intent case ship back attributes to client */
 569                 if (offset) {
 570                         struct mds_body *body =
 571                                 lustre_msg_buf(req->rq_repmsg, 1);
 572
 573                         mds_pack_inode2fid(&body->fid1, inode);
 574                         mds_pack_inode2body(body, inode);
 575                         if (S_ISREG(inode->i_mode)) {
 576                                 struct lov_mds_md *lmm;
 577
 578                                 lmm = lustre_msg_buf(req->rq_repmsg, 2);
 579                                 lmm->lmm_easize = mds->mds_max_mdsize;
 580                                 if ((rc = mds_fs_get_md(mds, inode, lmm)) < 0) {
 581                                         CDEBUG(D_INFO,"No md for %ld: rc %d\n",
 582                                                inode->i_ino, rc);
 583                                         memset(lmm, 0, lmm->lmm_easize);
 584                                 } else
 585                                         body->valid |= OBD_MD_FLEASIZE;
 586                         }
 587                 }
 588                 if (rec->ur_opcode & REINT_REPLAYING) {
 589                         rc = 0;
 590                         CDEBUG(D_INODE,
 591                                "child exists (dir %ld, name %s) (REPLAYING)\n",
 592                                de_tgt_dir->d_inode->i_ino, rec->ur_name);
 593                 } else {
 594                         rc = -EEXIST;
 595                         CERROR("child exists (dir %ld, name %s)\n",
 596                                de_tgt_dir->d_inode->i_ino, rec->ur_name);
 597                 }
 598                 GOTO(out_link_dchild, rc);
 599         }
 600
 601         OBD_FAIL_WRITE(OBD_FAIL_MDS_REINT_LINK_WRITE,
 602                        to_kdev_t(de_src->d_inode->i_sb->s_dev));
 603
 604         handle = mds_fs_start(mds, de_tgt_dir->d_inode, MDS_FSOP_LINK);
 605         if (!handle)
 606                 GOTO(out_link_dchild, rc = PTR_ERR(handle));
 607
 608         rc = vfs_link(de_src, de_tgt_dir->d_inode, dchild);
 609         if (rc)
 610                 CERROR("link error %d\n", rc);
 611         if (!rc)
 612                 rc = mds_update_last_rcvd(mds, handle, req);
 613
 614         err = mds_fs_commit(mds, de_tgt_dir->d_inode, handle);
 615         if (err) {
 616                 CERROR("error on commit: err = %d\n", err);
 617                 if (!rc)
 618                         rc = err;
 619         }
 620         EXIT;
 621
 622
 623
 624 out_link_dchild:
 625         l_dput(dchild);
 626 out_link_tgt_dir:
 627         ldlm_lock_decref(&tgtlockh, lock_mode);
 628 out_link_tgt_dir_put:
 629         up(&de_tgt_dir->d_inode->i_sem);
 630         l_dput(de_tgt_dir);
 631 out_link_src:
 632         ldlm_lock_decref(&srclockh, lock_mode);
 633 out_link_src_put:
 634         l_dput(de_src);
 635 out_link:
 636         req->rq_status = rc;
 637         return 0;
 638 }
 639
 640 static int mds_reint_rename(struct mds_update_record *rec, int offset,
 641                             struct ptlrpc_request *req)
 642 {
 643         struct obd_device *obd = req->rq_export->exp_obd;
 644         struct dentry *de_srcdir = NULL;
 645         struct dentry *de_tgtdir = NULL;
 646         struct dentry *de_old = NULL;
 647         struct dentry *de_new = NULL;
 648         struct mds_obd *mds = mds_req2mds(req);
 649         struct lustre_handle tgtlockh, srclockh, oldhandle;
 650         int flags, lock_mode, rc = 0, err;
 651         void *handle;
 652         __u64 res_id[3] = { 0 };
 653         ENTRY;
 654
 655         de_srcdir = mds_fid2dentry(mds, rec->ur_fid1, NULL);
 656         if (IS_ERR(de_srcdir))
 657                 GOTO(out_rename, rc = -ESTALE);
 658
 659         lock_mode = (req->rq_reqmsg->opc == MDS_REINT) ? LCK_PW : LCK_PW;
 660         res_id[0] = de_srcdir->d_inode->i_ino;
 661
 662         rc = ldlm_lock_match(obd->obd_namespace, res_id, LDLM_PLAIN,
 663                              NULL, 0, lock_mode, &srclockh);
 664         if (rc == 0) {
 665                 LDLM_DEBUG_NOLOCK("enqueue res "LPU64, res_id[0]);
 666                 rc = ldlm_cli_enqueue(NULL, NULL, obd->obd_namespace, NULL,
 667                                       res_id, LDLM_PLAIN, NULL, 0, lock_mode,
 668                                       &flags, ldlm_completion_ast,
 669                                       mds_blocking_ast, NULL, 0, &srclockh);
 670                 if (rc != ELDLM_OK) {
 671                         CERROR("lock enqueue: err: %d\n", rc);
 672                         GOTO(out_rename_srcput, rc = -EIO);
 673                 }
 674         } else
 675                 ldlm_lock_dump((void *)(unsigned long)srclockh.addr);
 676
 677         de_tgtdir = mds_fid2dentry(mds, rec->ur_fid2, NULL);
 678         if (IS_ERR(de_tgtdir))
 679                 GOTO(out_rename_srcdir, rc = -ESTALE);
 680
 681         lock_mode = (req->rq_reqmsg->opc == MDS_REINT) ? LCK_PW : LCK_PW;
 682         res_id[0] = de_tgtdir->d_inode->i_ino;
 683
 684         rc = ldlm_lock_match(obd->obd_namespace, res_id, LDLM_PLAIN,
 685                              NULL, 0, lock_mode, &tgtlockh);
 686         if (rc == 0) {
 687                 LDLM_DEBUG_NOLOCK("enqueue res "LPU64, res_id[0]);
 688                 rc = ldlm_cli_enqueue(NULL, NULL, obd->obd_namespace, NULL,
 689                                       res_id, LDLM_PLAIN, NULL, 0, lock_mode,
 690                                       &flags, ldlm_completion_ast,
 691                                       mds_blocking_ast, NULL, 0, &tgtlockh);
 692                 if (rc != ELDLM_OK) {
 693                         CERROR("lock enqueue: err: %d\n", rc);
 694                         GOTO(out_rename_tgtput, rc = -EIO);
 695                 }
 696         } else
 697                 ldlm_lock_dump((void *)(unsigned long)tgtlockh.addr);
 698
 699 #if (LINUX_VERSION_CODE < KERNEL_VERSION(2,5,0))
 700         double_lock(de_tgtdir, de_srcdir);
 701 #endif
 702         de_old = lookup_one_len(rec->ur_name, de_srcdir, rec->ur_namelen - 1);
 703         if (IS_ERR(de_old)) {
 704                 CERROR("old child lookup error (%*s): %ld\n",
 705                        rec->ur_namelen - 1, rec->ur_name, PTR_ERR(de_old));
 706                 GOTO(out_rename_tgtdir, rc = -ENOENT);
 707         }
 708
 709         de_new = lookup_one_len(rec->ur_tgt, de_tgtdir, rec->ur_tgtlen - 1);
 710         if (IS_ERR(de_new)) {
 711                 CERROR("new child lookup error (%*s): %ld\n",
 712                        rec->ur_tgtlen - 1, rec->ur_tgt, PTR_ERR(de_new));
 713                 GOTO(out_rename_deold, rc = -ENOENT);
 714         }
 715
 716         /* in intent case ship back attributes to client */
 717         if (offset) {
 718                 struct mds_body *body = lustre_msg_buf(req->rq_repmsg, 1);
 719                 struct inode *inode = de_new->d_inode;
 720
 721                 if (!inode) {
 722                         body->valid = 0;
 723                 } else {
 724                         mds_pack_inode2fid(&body->fid1, inode);
 725                         mds_pack_inode2body(body, inode);
 726                         if (S_ISREG(inode->i_mode)) {
 727                                 struct lov_mds_md *lmm;
 728
 729                                 lmm = lustre_msg_buf(req->rq_repmsg, 2);
 730                                 lmm->lmm_easize = mds->mds_max_mdsize;
 731                                 if ((rc = mds_fs_get_md(mds, inode, lmm)) < 0) {
 732                                         CDEBUG(D_INFO,"No md for %ld: rc %d\n",
 733                                                inode->i_ino, rc);
 734                                         memset(lmm, 0, lmm->lmm_easize);
 735                                 } else
 736                                         body->valid |= OBD_MD_FLEASIZE;
 737                         }
 738                 }
 739         }
 740
 741         OBD_FAIL_WRITE(OBD_FAIL_MDS_REINT_RENAME_WRITE,
 742                        to_kdev_t(de_srcdir->d_inode->i_sb->s_dev));
 743
 744         handle = mds_fs_start(mds, de_tgtdir->d_inode, MDS_FSOP_RENAME);
 745         if (!handle)
 746                 GOTO(out_rename_denew, rc = PTR_ERR(handle));
 747         lock_kernel();
 748         rc = vfs_rename(de_srcdir->d_inode, de_old, de_tgtdir->d_inode, de_new,
 749                         NULL);
 750         unlock_kernel();
 751
 752         if (!rc)
 753                 rc = mds_update_last_rcvd(mds, handle, req);
 754
 755         err = mds_fs_commit(mds, de_tgtdir->d_inode, handle);
 756         if (err) {
 757                 CERROR("error on commit: err = %d\n", err);
 758                 if (!rc)
 759                         rc = err;
 760         }
 761         EXIT;
 762
 763 out_rename_denew:
 764         l_dput(de_new);
 765 out_rename_deold:
 766         if (!rc) {
 767                 res_id[0] = de_old->d_inode->i_ino;
 768                 /* Take an exclusive lock on the resource that we're
 769                  * about to free, to force everyone to drop their
 770                  * locks. */
 771                 LDLM_DEBUG_NOLOCK("getting EX lock res "LPU64, res_id[0]);
 772                 rc = ldlm_cli_enqueue(NULL, NULL, obd->obd_namespace, NULL,
 773                                       res_id, LDLM_PLAIN, NULL, 0, LCK_EX,
 774                                       &flags, ldlm_completion_ast,
 775                                       mds_blocking_ast, NULL, 0, &oldhandle);
 776                 if (rc)
 777                         CERROR("failed to get child inode lock (child ino "
 778                                LPD64" dir ino %ld)\n",
 779                                res_id[0], de_old->d_inode->i_ino);
 780         }
 781
 782         l_dput(de_old);
 783
 784         if (!rc) {
 785                 ldlm_lock_decref(&oldhandle, LCK_EX);
 786                 rc = ldlm_cli_cancel(&oldhandle);
 787                 if (rc < 0)
 788                         CERROR("failed to cancel child inode lock ino "
 789                                LPD64": %d\n", res_id[0], rc);
 790         }
 791 out_rename_tgtdir:
 792 #if (LINUX_VERSION_CODE < KERNEL_VERSION(2,5,0))
 793         double_up(&de_srcdir->d_inode->i_sem, &de_tgtdir->d_inode->i_sem);
 794 #endif
 795         ldlm_lock_decref(&tgtlockh, lock_mode);
 796 out_rename_tgtput:
 797         l_dput(de_tgtdir);
 798 out_rename_srcdir:
 799         ldlm_lock_decref(&srclockh, lock_mode);
 800 out_rename_srcput:
 801         l_dput(de_srcdir);
 802 out_rename:
 803         req->rq_status = rc;
 804         return 0;
 805 }
 806
 807 typedef int (*mds_reinter) (struct mds_update_record *, int offset,
 808                             struct ptlrpc_request *);
 809
 810 static mds_reinter reinters[REINT_MAX + 1] = {
 811         [REINT_SETATTR] mds_reint_setattr,
 812         [REINT_CREATE] mds_reint_create,
 813         [REINT_UNLINK] mds_reint_unlink,
 814         [REINT_LINK] mds_reint_link,
 815         [REINT_RENAME] mds_reint_rename,
 816 };
 817
 818 int mds_reint_rec(struct mds_update_record *rec, int offset,
 819                   struct ptlrpc_request *req)
 820 {
 821         struct mds_obd *mds = mds_req2mds(req);
 822         struct obd_run_ctxt saved;
 823         struct obd_ucred uc;
 824         int realop = rec->ur_opcode & REINT_OPCODE_MASK;
 825         int rc;
 826
 827         if (realop < 1 || realop > REINT_MAX) {
 828                 CERROR("opcode %d not valid (%sREPLAYING)\n", realop,
 829                        rec->ur_opcode & REINT_REPLAYING ? "" : "not ");
 830                 rc = req->rq_status = -EINVAL;
 831                 RETURN(rc);
 832         }
 833
 834         uc.ouc_fsuid = rec->ur_fsuid;
 835         uc.ouc_fsgid = rec->ur_fsgid;
 836
 837         push_ctxt(&saved, &mds->mds_ctxt, &uc);
 838         rc = reinters[realop] (rec, offset, req);
 839         pop_ctxt(&saved);
 840
 841         return rc;
 842 }