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                         rc = mds_pack_md(mds, req, offset + 1, body, inode);
 222
 223                 /* This isn't an error for RECREATE. */
 224                 if (rec->ur_opcode & REINT_REPLAYING) {
 225                         CDEBUG(D_INODE, "EEXIST suppressed for REPLAYING\n");
 226                         rc = 0;
 227                 } else {
 228                         rc = -EEXIST;
 229                 }
 230                 GOTO(out_create_dchild, rc);
 231         }
 232
 233         OBD_FAIL_WRITE(OBD_FAIL_MDS_REINT_CREATE_WRITE,
 234                        to_kdev_t(dir->i_sb->s_dev));
 235
 236         if (dir->i_mode & S_ISGID) {
 237                 rec->ur_gid = dir->i_gid;
 238                 if (S_ISDIR(rec->ur_mode))
 239                         rec->ur_mode |= S_ISGID;
 240         }
 241
 242         switch (type) {
 243         case S_IFREG:{
 244                 handle = mds_fs_start(mds, dir, MDS_FSOP_CREATE);
 245                 if (!handle)
 246                         GOTO(out_create_dchild, PTR_ERR(handle));
 247                 rc = vfs_create(dir, dchild, rec->ur_mode);
 248                 EXIT;
 249                 break;
 250         }
 251         case S_IFDIR:{
 252                 handle = mds_fs_start(mds, dir, MDS_FSOP_MKDIR);
 253                 if (!handle)
 254                         GOTO(out_create_dchild, PTR_ERR(handle));
 255                 rc = vfs_mkdir(dir, dchild, rec->ur_mode);
 256                 EXIT;
 257                 break;
 258         }
 259         case S_IFLNK:{
 260                 handle = mds_fs_start(mds, dir, MDS_FSOP_SYMLINK);
 261                 if (!handle)
 262                         GOTO(out_create_dchild, PTR_ERR(handle));
 263                 rc = vfs_symlink(dir, dchild, rec->ur_name);
 264                 EXIT;
 265                 break;
 266         }
 267         case S_IFCHR:
 268         case S_IFBLK:
 269         case S_IFIFO:
 270         case S_IFSOCK:{
 271                 int rdev = rec->ur_rdev;
 272                 handle = mds_fs_start(mds, dir, MDS_FSOP_MKNOD);
 273                 if (!handle)
 274                         GOTO(out_create_dchild, PTR_ERR(handle));
 275                 rc = vfs_mknod(dir, dchild, rec->ur_mode, rdev);
 276                 EXIT;
 277                 break;
 278         }
 279         default:
 280                 CERROR("bad file type %o creating %s\n", type, rec->ur_name);
 281                 GOTO(out_create_dchild, rc = -EINVAL);
 282         }
 283
 284         if (rc) {
 285                 CDEBUG(D_INODE, "error during create: %d\n", rc);
 286                 GOTO(out_create_commit, rc);
 287         } else {
 288                 struct iattr iattr;
 289                 struct inode *inode = dchild->d_inode;
 290                 struct mds_body *body;
 291
 292                 iattr.ia_atime = rec->ur_time;
 293                 iattr.ia_ctime = rec->ur_time;
 294                 iattr.ia_mtime = rec->ur_time;
 295                 iattr.ia_uid = rec->ur_uid;
 296                 iattr.ia_gid = rec->ur_gid;
 297                 iattr.ia_valid = ATTR_UID | ATTR_GID | ATTR_ATIME |
 298                         ATTR_MTIME | ATTR_CTIME;
 299
 300                 if (rec->ur_fid2->id) {
 301                         LASSERT(rec->ur_opcode & REINT_REPLAYING);
 302                         inode->i_generation = rec->ur_fid2->generation;
 303                         /* Dirtied and committed by this setattr: */
 304                         CDEBUG(D_INODE, "recreated ino %ld with gen %ld\n",
 305                                inode->i_ino, inode->i_generation);
 306                 } else {
 307                         CDEBUG(D_INODE, "created ino %ld\n", inode->i_ino);
 308                 }
 309
 310                 rc = mds_fs_setattr(mds, dchild, handle, &iattr);
 311                 if (rc) {
 312                         CERROR("error on setattr: rc = %d\n", rc);
 313                         /* XXX should we abort here in case of error? */
 314                 }
 315
 316                 rc = mds_update_last_rcvd(mds, handle, req);
 317                 if (rc) {
 318                         CERROR("error on mds_update_last_rcvd: rc = %d\n", rc);
 319                         GOTO(out_create_unlink, rc);
 320                 }
 321
 322                 body = lustre_msg_buf(req->rq_repmsg, offset);
 323                 mds_pack_inode2fid(&body->fid1, inode);
 324                 mds_pack_inode2body(body, inode);
 325         }
 326         EXIT;
 327 out_create_commit:
 328         err = mds_fs_commit(mds, dir, handle);
 329         if (err) {
 330                 CERROR("error on commit: err = %d\n", err);
 331                 if (!rc)
 332                         rc = err;
 333         }
 334 out_create_dchild:
 335         l_dput(dchild);
 336         ldlm_lock_decref(&lockh, lock_mode);
 337 out_create_de:
 338         up(&dir->i_sem);
 339         l_dput(de);
 340 out_create:
 341         req->rq_status = rc;
 342         return 0;
 343
 344 out_create_unlink:
 345         /* Destroy the file we just created.  This should not need extra
 346          * journal credits, as we have already modified all of the blocks
 347          * needed in order to create the file in the first place.
 348          */
 349         switch (type) {
 350         case S_IFDIR:
 351                 err = vfs_rmdir(dir, dchild);
 352                 if (err)
 353                         CERROR("failed rmdir in error path: rc = %d\n", err);
 354                 break;
 355         default:
 356                 err = vfs_unlink(dir, dchild);
 357                 if (err)
 358                         CERROR("failed unlink in error path: rc = %d\n", err);
 359                 break;
 360         }
 361
 362         goto out_create_commit;
 363 }
 364
 365 static int mds_reint_unlink(struct mds_update_record *rec, int offset,
 366                             struct ptlrpc_request *req)
 367 {
 368         struct dentry *de = NULL;
 369         struct dentry *dchild = NULL;
 370         struct mds_obd *mds = mds_req2mds(req);
 371         struct obd_device *obd = req->rq_export->exp_obd;
 372         struct mds_body *body = NULL;
 373         char *name;
 374         struct inode *dir, *inode;
 375         struct lustre_handle lockh, child_lockh;
 376         void *handle;
 377         int namelen, lock_mode, err, rc = 0;
 378         ENTRY;
 379
 380         /* a name was supplied by the client; fid1 is the directory */
 381         lock_mode = (req->rq_reqmsg->opc == MDS_REINT) ? LCK_PW : LCK_PW;
 382         de = mds_fid2locked_dentry(obd, rec->ur_fid1, NULL, lock_mode, &lockh);
 383         if (IS_ERR(de)) {
 384                 LBUG();
 385                 RETURN(PTR_ERR(de));
 386         }
 387
 388         if (OBD_FAIL_CHECK(OBD_FAIL_MDS_REINT_UNLINK))
 389                 GOTO(out_unlink, rc = -ENOENT);
 390
 391         name = lustre_msg_buf(req->rq_reqmsg, offset + 1);
 392         namelen = req->rq_reqmsg->buflens[offset + 1] - 1;
 393 #warning "FIXME: if mds_name2locked_dentry decrefs this lock, we must not"
 394         memcpy(&child_lockh, &lockh, sizeof(child_lockh));
 395         dchild = mds_name2locked_dentry(obd, de, NULL, name, namelen,
 396                                         LCK_EX, &child_lockh, lock_mode);
 397
 398         if (IS_ERR(dchild)) {
 399                 LBUG();
 400                 GOTO(out_unlink, rc = PTR_ERR(dchild));
 401         }
 402
 403         dir = de->d_inode;
 404         inode = dchild->d_inode;
 405         CDEBUG(D_INODE, "parent ino %ld\n", dir->i_ino);
 406
 407         if (!inode) {
 408                 if (rec->ur_opcode & REINT_REPLAYING) {
 409                         CDEBUG(D_INODE,
 410                                "child missing (%ld/%s); OK for REPLAYING\n",
 411                                dir->i_ino, rec->ur_name);
 412                         rc = 0;
 413                 } else {
 414                         CDEBUG(D_INODE,
 415                                "child doesn't exist (dir %ld, name %s)\n",
 416                                dir->i_ino, rec->ur_name);
 417                         rc = -ENOENT;
 418                 }
 419                 /* going to out_unlink_cancel causes an LBUG, don't know why */
 420                 GOTO(out_unlink_dchild, rc);
 421         }
 422
 423         if (offset) {
 424                 /* XXX offset? */
 425                 offset = 1;
 426
 427                 body = lustre_msg_buf(req->rq_repmsg, offset);
 428                 mds_pack_inode2fid(&body->fid1, inode);
 429                 mds_pack_inode2body(body, inode);
 430         }
 431
 432         OBD_FAIL_WRITE(OBD_FAIL_MDS_REINT_UNLINK_WRITE,
 433                        to_kdev_t(dir->i_sb->s_dev));
 434
 435         switch (rec->ur_mode /* & S_IFMT ? */) {
 436         case S_IFDIR:
 437                 handle = mds_fs_start(mds, dir, MDS_FSOP_RMDIR);
 438                 if (!handle)
 439                         GOTO(out_unlink_cancel, rc = PTR_ERR(handle));
 440                 rc = vfs_rmdir(dir, dchild);
 441                 break;
 442         case S_IFREG:
 443                 /* get OBD EA data first so client can also destroy object */
 444                 if ((inode->i_mode & S_IFMT) == S_IFREG && offset)
 445                         rc = mds_pack_md(mds, req, offset + 1, body, inode);
 446                 /* no break */
 447         case S_IFLNK:
 448         case S_IFCHR:
 449         case S_IFBLK:
 450         case S_IFIFO:
 451         case S_IFSOCK:
 452                 handle = mds_fs_start(mds, dir, MDS_FSOP_UNLINK);
 453                 if (!handle)
 454                         GOTO(out_unlink_cancel, rc = PTR_ERR(handle));
 455                 rc = vfs_unlink(dir, dchild);
 456                 break;
 457         default:
 458                 CERROR("bad file type %o unlinking %s\n", rec->ur_mode, name);
 459                 handle = NULL;
 460                 LBUG();
 461                 GOTO(out_unlink_cancel, rc = -EINVAL);
 462         }
 463
 464         if (!rc)
 465                 rc = mds_update_last_rcvd(mds, handle, req);
 466         err = mds_fs_commit(mds, dir, handle);
 467         if (err) {
 468                 CERROR("error on commit: err = %d\n", err);
 469                 if (!rc)
 470                         rc = err;
 471         }
 472
 473         EXIT;
 474
 475 out_unlink_cancel:
 476         ldlm_lock_decref(&child_lockh, LCK_EX);
 477         err = ldlm_cli_cancel(&child_lockh);
 478         if (err < 0) {
 479                 CERROR("failed to cancel child inode lock: err = %d\n", err);
 480                 if (!rc)
 481                         rc = -ENOLCK;   /*XXX translate LDLM lock error */
 482         }
 483 out_unlink_dchild:
 484         l_dput(dchild);
 485         up(&dir->i_sem);
 486 out_unlink:
 487         ldlm_lock_decref(&lockh, lock_mode);
 488         l_dput(de);
 489         req->rq_status = rc;
 490         return 0;
 491 }
 492
 493 static int mds_reint_link(struct mds_update_record *rec, int offset,
 494                           struct ptlrpc_request *req)
 495 {
 496         struct obd_device *obd = req->rq_export->exp_obd;
 497         struct dentry *de_src = NULL;
 498         struct dentry *de_tgt_dir = NULL;
 499         struct dentry *dchild = NULL;
 500         struct mds_obd *mds = mds_req2mds(req);
 501         struct lustre_handle *handle, tgtlockh, srclockh;
 502         int lock_mode;
 503         __u64 res_id[3] = { 0 };
 504         int flags = 0;
 505         int rc = 0, err;
 506
 507         ENTRY;
 508         de_src = mds_fid2dentry(mds, rec->ur_fid1, NULL);
 509         if (IS_ERR(de_src) || OBD_FAIL_CHECK(OBD_FAIL_MDS_REINT_LINK)) {
 510                 GOTO(out_link, rc = -ESTALE);
 511         }
 512
 513         /* plan to change the link count on this inode: write lock */
 514         lock_mode = (req->rq_reqmsg->opc == MDS_REINT) ? LCK_PW : LCK_PW;
 515         res_id[0] = de_src->d_inode->i_ino;
 516         res_id[1] = de_src->d_inode->i_generation;
 517
 518         rc = ldlm_lock_match(obd->obd_namespace, res_id, LDLM_PLAIN,
 519                              NULL, 0, lock_mode, &srclockh);
 520         if (rc == 0) {
 521                 LDLM_DEBUG_NOLOCK("enqueue res "LPU64, res_id[0]);
 522                 rc = ldlm_cli_enqueue(NULL, NULL, obd->obd_namespace, NULL,
 523                                       res_id, LDLM_PLAIN, NULL, 0, lock_mode,
 524                                       &flags, ldlm_completion_ast,
 525                                       mds_blocking_ast, NULL, 0, &srclockh);
 526                 if (rc != ELDLM_OK) {
 527                         CERROR("lock enqueue: err: %d\n", rc);
 528                         GOTO(out_link_src_put, rc = -EIO);
 529                 }
 530         } else
 531                 ldlm_lock_dump((void *)(unsigned long)srclockh.addr);
 532
 533         de_tgt_dir = mds_fid2dentry(mds, rec->ur_fid2, NULL);
 534         if (IS_ERR(de_tgt_dir)) {
 535                 GOTO(out_link_src, rc = -ESTALE);
 536         }
 537
 538         lock_mode = (req->rq_reqmsg->opc == MDS_REINT) ? LCK_PW : LCK_PW;
 539         res_id[0] = de_tgt_dir->d_inode->i_ino;
 540         res_id[1] = de_tgt_dir->d_inode->i_generation;
 541
 542         rc = ldlm_lock_match(obd->obd_namespace, res_id, LDLM_PLAIN,
 543                              NULL, 0, lock_mode, &tgtlockh);
 544         if (rc == 0) {
 545                 LDLM_DEBUG_NOLOCK("enqueue res "LPU64, res_id[0]);
 546                 rc = ldlm_cli_enqueue(NULL, NULL, obd->obd_namespace, NULL,
 547                                       res_id, LDLM_PLAIN, NULL, 0, lock_mode,
 548                                       &flags, ldlm_completion_ast,
 549                                       mds_blocking_ast, NULL, 0, &tgtlockh);
 550                 if (rc != ELDLM_OK) {
 551                         CERROR("lock enqueue: err: %d\n", rc);
 552                         GOTO(out_link_tgt_dir_put, rc = -EIO);
 553                 }
 554         } else
 555                 ldlm_lock_dump((void *)(unsigned long)tgtlockh.addr);
 556
 557         down(&de_tgt_dir->d_inode->i_sem);
 558         dchild = lookup_one_len(rec->ur_name, de_tgt_dir, rec->ur_namelen - 1);
 559         if (IS_ERR(dchild)) {
 560                 CERROR("child lookup error %ld\n", PTR_ERR(dchild));
 561                 GOTO(out_link_tgt_dir, rc = -ESTALE);
 562         }
 563
 564         if (dchild->d_inode) {
 565                 struct inode *inode = dchild->d_inode;
 566                 /* in intent case ship back attributes to client */
 567                 if (offset) {
 568                         struct mds_body *body =
 569                                 lustre_msg_buf(req->rq_repmsg, 1);
 570
 571                         mds_pack_inode2fid(&body->fid1, inode);
 572                         mds_pack_inode2body(body, inode);
 573                         if (S_ISREG(inode->i_mode))
 574                                 rc = mds_pack_md(mds, req, 2, body, inode);
 575                 }
 576                 if (rec->ur_opcode & REINT_REPLAYING) {
 577                         /* XXX verify that the link is to the the right file? */
 578                         rc = 0;
 579                         CDEBUG(D_INODE,
 580                                "child exists (dir %ld, name %s) (REPLAYING)\n",
 581                                de_tgt_dir->d_inode->i_ino, rec->ur_name);
 582                 } else {
 583                         rc = -EEXIST;
 584                         CERROR("child exists (dir %ld, name %s)\n",
 585                                de_tgt_dir->d_inode->i_ino, rec->ur_name);
 586                 }
 587                 GOTO(out_link_dchild, rc);
 588         }
 589
 590         OBD_FAIL_WRITE(OBD_FAIL_MDS_REINT_LINK_WRITE,
 591                        to_kdev_t(de_src->d_inode->i_sb->s_dev));
 592
 593         handle = mds_fs_start(mds, de_tgt_dir->d_inode, MDS_FSOP_LINK);
 594         if (!handle)
 595                 GOTO(out_link_dchild, rc = PTR_ERR(handle));
 596
 597         rc = vfs_link(de_src, de_tgt_dir->d_inode, dchild);
 598         if (rc)
 599                 CERROR("link error %d\n", rc);
 600         if (!rc)
 601                 rc = mds_update_last_rcvd(mds, handle, req);
 602
 603         err = mds_fs_commit(mds, de_tgt_dir->d_inode, handle);
 604         if (err) {
 605                 CERROR("error on commit: err = %d\n", err);
 606                 if (!rc)
 607                         rc = err;
 608         }
 609         EXIT;
 610
 611 out_link_dchild:
 612         l_dput(dchild);
 613 out_link_tgt_dir:
 614         ldlm_lock_decref(&tgtlockh, lock_mode);
 615 out_link_tgt_dir_put:
 616         up(&de_tgt_dir->d_inode->i_sem);
 617         l_dput(de_tgt_dir);
 618 out_link_src:
 619         ldlm_lock_decref(&srclockh, lock_mode);
 620 out_link_src_put:
 621         l_dput(de_src);
 622 out_link:
 623         req->rq_status = rc;
 624         return 0;
 625 }
 626
 627 static int mds_reint_rename(struct mds_update_record *rec, int offset,
 628                             struct ptlrpc_request *req)
 629 {
 630         struct obd_device *obd = req->rq_export->exp_obd;
 631         struct dentry *de_srcdir = NULL;
 632         struct dentry *de_tgtdir = NULL;
 633         struct dentry *de_old = NULL;
 634         struct dentry *de_new = NULL;
 635         struct mds_obd *mds = mds_req2mds(req);
 636         struct lustre_handle tgtlockh, srclockh, oldhandle;
 637         int flags = 0, lock_mode, rc = 0, err;
 638         void *handle;
 639         __u64 res_id[3] = { 0 };
 640         ENTRY;
 641
 642         de_srcdir = mds_fid2dentry(mds, rec->ur_fid1, NULL);
 643         if (IS_ERR(de_srcdir))
 644                 GOTO(out_rename, rc = -ESTALE);
 645
 646         lock_mode = (req->rq_reqmsg->opc == MDS_REINT) ? LCK_PW : LCK_PW;
 647         res_id[0] = de_srcdir->d_inode->i_ino;
 648         res_id[1] = de_srcdir->d_inode->i_generation;
 649
 650         rc = ldlm_lock_match(obd->obd_namespace, res_id, LDLM_PLAIN,
 651                              NULL, 0, lock_mode, &srclockh);
 652         if (rc == 0) {
 653                 LDLM_DEBUG_NOLOCK("enqueue res "LPU64, res_id[0]);
 654                 rc = ldlm_cli_enqueue(NULL, NULL, obd->obd_namespace, NULL,
 655                                       res_id, LDLM_PLAIN, NULL, 0, lock_mode,
 656                                       &flags, ldlm_completion_ast,
 657                                       mds_blocking_ast, NULL, 0, &srclockh);
 658                 if (rc != ELDLM_OK) {
 659                         CERROR("lock enqueue: err: %d\n", rc);
 660                         GOTO(out_rename_srcput, rc = -EIO);
 661                 }
 662         } else
 663                 ldlm_lock_dump((void *)(unsigned long)srclockh.addr);
 664
 665         de_tgtdir = mds_fid2dentry(mds, rec->ur_fid2, NULL);
 666         if (IS_ERR(de_tgtdir))
 667                 GOTO(out_rename_srcdir, rc = -ESTALE);
 668
 669         lock_mode = (req->rq_reqmsg->opc == MDS_REINT) ? LCK_PW : LCK_PW;
 670         res_id[0] = de_tgtdir->d_inode->i_ino;
 671         res_id[1] = de_tgtdir->d_inode->i_generation;
 672
 673         rc = ldlm_lock_match(obd->obd_namespace, res_id, LDLM_PLAIN,
 674                              NULL, 0, lock_mode, &tgtlockh);
 675         if (rc == 0) {
 676                 flags = 0;
 677                 LDLM_DEBUG_NOLOCK("enqueue res "LPU64, res_id[0]);
 678                 rc = ldlm_cli_enqueue(NULL, NULL, obd->obd_namespace, NULL,
 679                                       res_id, LDLM_PLAIN, NULL, 0, lock_mode,
 680                                       &flags, ldlm_completion_ast,
 681                                       mds_blocking_ast, NULL, 0, &tgtlockh);
 682                 if (rc != ELDLM_OK) {
 683                         CERROR("lock enqueue: err: %d\n", rc);
 684                         GOTO(out_rename_tgtput, rc = -EIO);
 685                 }
 686         } else
 687                 ldlm_lock_dump((void *)(unsigned long)tgtlockh.addr);
 688
 689 #if (LINUX_VERSION_CODE < KERNEL_VERSION(2,5,0))
 690         double_lock(de_tgtdir, de_srcdir);
 691 #endif
 692         de_old = lookup_one_len(rec->ur_name, de_srcdir, rec->ur_namelen - 1);
 693         if (IS_ERR(de_old)) {
 694                 CERROR("old child lookup error (%*s): %ld\n",
 695                        rec->ur_namelen - 1, rec->ur_name, PTR_ERR(de_old));
 696                 GOTO(out_rename_tgtdir, rc = -ENOENT);
 697         }
 698
 699         de_new = lookup_one_len(rec->ur_tgt, de_tgtdir, rec->ur_tgtlen - 1);
 700         if (IS_ERR(de_new)) {
 701                 CERROR("new child lookup error (%*s): %ld\n",
 702                        rec->ur_tgtlen - 1, rec->ur_tgt, PTR_ERR(de_new));
 703                 GOTO(out_rename_deold, rc = -ENOENT);
 704         }
 705
 706         /* in intent case ship back attributes to client */
 707         if (offset) {
 708                 struct mds_body *body = lustre_msg_buf(req->rq_repmsg, 1);
 709                 struct inode *inode = de_new->d_inode;
 710
 711                 if (!inode) {
 712                         body->valid = 0;
 713                 } else {
 714                         mds_pack_inode2fid(&body->fid1, inode);
 715                         mds_pack_inode2body(body, inode);
 716                         if (S_ISREG(inode->i_mode))
 717                                 rc = mds_pack_md(mds, req, 2, body, inode);
 718                 }
 719         }
 720
 721         OBD_FAIL_WRITE(OBD_FAIL_MDS_REINT_RENAME_WRITE,
 722                        to_kdev_t(de_srcdir->d_inode->i_sb->s_dev));
 723
 724         handle = mds_fs_start(mds, de_tgtdir->d_inode, MDS_FSOP_RENAME);
 725         if (!handle)
 726                 GOTO(out_rename_denew, rc = PTR_ERR(handle));
 727         lock_kernel();
 728         rc = vfs_rename(de_srcdir->d_inode, de_old, de_tgtdir->d_inode, de_new,
 729                         NULL);
 730         unlock_kernel();
 731
 732         if (!rc)
 733                 rc = mds_update_last_rcvd(mds, handle, req);
 734
 735         err = mds_fs_commit(mds, de_tgtdir->d_inode, handle);
 736         if (err) {
 737                 CERROR("error on commit: err = %d\n", err);
 738                 if (!rc)
 739                         rc = err;
 740         }
 741         EXIT;
 742
 743 out_rename_denew:
 744         l_dput(de_new);
 745 out_rename_deold:
 746         if (!rc) {
 747                 res_id[0] = de_old->d_inode->i_ino;
 748                 res_id[1] = de_old->d_inode->i_generation;
 749                 flags = 0;
 750                 /* Take an exclusive lock on the resource that we're
 751                  * about to free, to force everyone to drop their
 752                  * locks. */
 753                 LDLM_DEBUG_NOLOCK("getting EX lock res "LPU64, res_id[0]);
 754                 rc = ldlm_cli_enqueue(NULL, NULL, obd->obd_namespace, NULL,
 755                                       res_id, LDLM_PLAIN, NULL, 0, LCK_EX,
 756                                       &flags, ldlm_completion_ast,
 757                                       mds_blocking_ast, NULL, 0, &oldhandle);
 758                 if (rc)
 759                         CERROR("failed to get child inode lock (child ino "
 760                                LPD64" dir ino %ld)\n",
 761                                res_id[0], de_old->d_inode->i_ino);
 762         }
 763
 764         l_dput(de_old);
 765
 766         if (!rc) {
 767                 ldlm_lock_decref(&oldhandle, LCK_EX);
 768                 rc = ldlm_cli_cancel(&oldhandle);
 769                 if (rc < 0)
 770                         CERROR("failed to cancel child inode lock ino "
 771                                LPD64": %d\n", res_id[0], rc);
 772         }
 773 out_rename_tgtdir:
 774 #if (LINUX_VERSION_CODE < KERNEL_VERSION(2,5,0))
 775         double_up(&de_srcdir->d_inode->i_sem, &de_tgtdir->d_inode->i_sem);
 776 #endif
 777         ldlm_lock_decref(&tgtlockh, lock_mode);
 778 out_rename_tgtput:
 779         l_dput(de_tgtdir);
 780 out_rename_srcdir:
 781         ldlm_lock_decref(&srclockh, lock_mode);
 782 out_rename_srcput:
 783         l_dput(de_srcdir);
 784 out_rename:
 785         req->rq_status = rc;
 786         return 0;
 787 }
 788
 789 typedef int (*mds_reinter) (struct mds_update_record *, int offset,
 790                             struct ptlrpc_request *);
 791
 792 static mds_reinter reinters[REINT_MAX + 1] = {
 793         [REINT_SETATTR] mds_reint_setattr,
 794         [REINT_CREATE] mds_reint_create,
 795         [REINT_UNLINK] mds_reint_unlink,
 796         [REINT_LINK] mds_reint_link,
 797         [REINT_RENAME] mds_reint_rename,
 798 };
 799
 800 int mds_reint_rec(struct mds_update_record *rec, int offset,
 801                   struct ptlrpc_request *req)
 802 {
 803         struct mds_obd *mds = mds_req2mds(req);
 804         struct obd_run_ctxt saved;
 805         struct obd_ucred uc;
 806         int realop = rec->ur_opcode & REINT_OPCODE_MASK;
 807         int rc;
 808
 809         if (realop < 1 || realop > REINT_MAX) {
 810                 CERROR("opcode %d not valid (%sREPLAYING)\n", realop,
 811                        rec->ur_opcode & REINT_REPLAYING ? "" : "not ");
 812                 rc = req->rq_status = -EINVAL;
 813                 RETURN(rc);
 814         }
 815
 816         uc.ouc_fsuid = rec->ur_fsuid;
 817         uc.ouc_fsgid = rec->ur_fsgid;
 818         uc.ouc_cap = rec->ur_cap;
 819
 820         push_ctxt(&saved, &mds->mds_ctxt, &uc);
 821         rc = reinters[realop] (rec, offset, req);
 822         pop_ctxt(&saved);
 823
 824         return rc;
 825 }