lustre/mds/mds_open.c

   1 /* -*- mode: c; c-basic-offset: 8; indent-tabs-mode: nil; -*-
   2  * vim:expandtab:shiftwidth=8:tabstop=8:
   3  *
   4  *  Copyright (c) 2003 Cluster File Systems, Inc.
   5  *   Author: Peter Braam <braam@clusterfs.com>
   6  *   Author: Andreas Dilger <adilger@clusterfs.com>
   7  *   Author: Phil Schwan <phil@clusterfs.com>
   8  *   Author: Mike Shaver <shaver@clusterfs.com>
   9  *
  10  *   This file is part of Lustre, http://www.lustre.org.
  11  *
  12  *   Lustre is free software; you can redistribute it and/or
  13  *   modify it under the terms of version 2 of the GNU General Public
  14  *   License as published by the Free Software Foundation.
  15  *
  16  *   Lustre is distributed in the hope that it will be useful,
  17  *   but WITHOUT ANY WARRANTY; without even the implied warranty of
  18  *   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  19  *   GNU General Public License for more details.
  20  *
  21  *   You should have received a copy of the GNU General Public License
  22  *   along with Lustre; if not, write to the Free Software
  23  *   Foundation, Inc., 675 Mass Ave, Cambridge, MA 02139, USA.
  24  */
  25
  26 #ifndef EXPORT_SYMTAB
  27 # define EXPORT_SYMTAB
  28 #endif
  29 #define DEBUG_SUBSYSTEM S_MDS
  30
  31 #include <linux/version.h>
  32 #include <linux/module.h>
  33 #include <linux/init.h>
  34 #include <linux/version.h>
  35 #if (LINUX_VERSION_CODE >= KERNEL_VERSION(2,5,0))
  36 # include <linux/buffer_head.h>
  37 # include <linux/workqueue.h>
  38 #else
  39 # include <linux/locks.h>
  40 #endif
  41
  42 #include <linux/obd_class.h>
  43 #include <linux/obd_lov.h>
  44 #include <linux/lustre_fsfilt.h>
  45 #include <linux/lprocfs_status.h>
  46
  47 #include "mds_internal.h"
  48
  49 /* Exported function from this file are:
  50  *
  51  * mds_open - called by the intent handler
  52  * mds_close - an rpc handling function
  53  * mds_pin - an rpc handling function - which will go away
  54  * mds_mfd_close - for force closing files when a client dies
  55  */
  56
  57 /*
  58  * MDS file data handling: file data holds a handle for a file opened
  59  * by a client.
  60  */
  61
  62 static void mds_mfd_addref(void *mfdp)
  63 {
  64         struct mds_file_data *mfd = mfdp;
  65
  66         atomic_inc(&mfd->mfd_refcount);
  67         CDEBUG(D_INFO, "GETting mfd %p : new refcount %d\n", mfd,
  68                atomic_read(&mfd->mfd_refcount));
  69 }
  70
  71 struct mds_file_data *mds_mfd_new(void)
  72 {
  73         struct mds_file_data *mfd;
  74
  75         OBD_ALLOC(mfd, sizeof *mfd);
  76         if (mfd == NULL) {
  77                 CERROR("mds: out of memory\n");
  78                 return NULL;
  79         }
  80
  81         atomic_set(&mfd->mfd_refcount, 2);
  82
  83         INIT_LIST_HEAD(&mfd->mfd_handle.h_link);
  84         class_handle_hash(&mfd->mfd_handle, mds_mfd_addref);
  85
  86         return mfd;
  87 }
  88
  89 static struct mds_file_data *mds_handle2mfd(struct lustre_handle *handle)
  90 {
  91         ENTRY;
  92         LASSERT(handle != NULL);
  93         RETURN(class_handle2object(handle->cookie));
  94 }
  95
  96 static void mds_mfd_put(struct mds_file_data *mfd)
  97 {
  98         CDEBUG(D_INFO, "PUTting mfd %p : new refcount %d\n", mfd,
  99                atomic_read(&mfd->mfd_refcount) - 1);
 100         LASSERT(atomic_read(&mfd->mfd_refcount) > 0 &&
 101                 atomic_read(&mfd->mfd_refcount) < 0x5a5a);
 102         if (atomic_dec_and_test(&mfd->mfd_refcount)) {
 103                 LASSERT(list_empty(&mfd->mfd_handle.h_link));
 104                 OBD_FREE(mfd, sizeof *mfd);
 105         }
 106 }
 107
 108 static void mds_mfd_destroy(struct mds_file_data *mfd)
 109 {
 110         class_handle_unhash(&mfd->mfd_handle);
 111         mds_mfd_put(mfd);
 112 }
 113
 114
 115 /* Caller must hold mds->mds_epoch_sem */
 116 static int mds_alloc_filterdata(struct inode *inode)
 117 {
 118         LASSERT(inode->i_filterdata == NULL);
 119         OBD_ALLOC(inode->i_filterdata, sizeof(struct mds_filter_data));
 120         if (inode->i_filterdata == NULL)
 121                 return -ENOMEM;
 122         LASSERT(igrab(inode) == inode);
 123         return 0;
 124 }
 125
 126 /* Caller must hold mds->mds_epoch_sem */
 127 static void mds_free_filterdata(struct inode *inode)
 128 {
 129         LASSERT(inode->i_filterdata != NULL);
 130         OBD_FREE(inode->i_filterdata, sizeof(struct mds_filter_data));
 131         inode->i_filterdata = NULL;
 132         iput(inode);
 133 }
 134
 135 /* Write access to a file: executors cause a negative count,
 136  * writers a positive count.  The semaphore is needed to perform
 137  * a check for the sign and then increment or decrement atomically.
 138  *
 139  * This code is closely tied to the allocation of the d_fsdata and the
 140  * MDS epoch, so we use the same semaphore for the whole lot.
 141  *
 142  * We could use a different semaphore for each file, if it ever shows
 143  * up in a profile, which it won't.
 144  *
 145  * epoch argument is nonzero during recovery */
 146 static int mds_get_write_access(struct mds_obd *mds, struct inode *inode,
 147                                 __u64 epoch)
 148 {
 149         int rc = 0;
 150
 151         down(&mds->mds_epoch_sem);
 152
 153         if (atomic_read(&inode->i_writecount) < 0) {
 154                 up(&mds->mds_epoch_sem);
 155                 RETURN(-ETXTBSY);
 156         }
 157
 158
 159         if (MDS_FILTERDATA(inode) && MDS_FILTERDATA(inode)->io_epoch != 0) {
 160                 CDEBUG(D_INODE, "continuing MDS epoch "LPU64" for ino %lu/%u\n",
 161                        MDS_FILTERDATA(inode)->io_epoch, inode->i_ino,
 162                        inode->i_generation);
 163                 goto out;
 164         }
 165
 166         if (inode->i_filterdata == NULL)
 167                 mds_alloc_filterdata(inode);
 168         if (inode->i_filterdata == NULL) {
 169                 rc = -ENOMEM;
 170                 goto out;
 171         }
 172         if (epoch > mds->mds_io_epoch)
 173                 mds->mds_io_epoch = epoch;
 174         else
 175                 mds->mds_io_epoch++;
 176         MDS_FILTERDATA(inode)->io_epoch = mds->mds_io_epoch;
 177         CDEBUG(D_INODE, "starting MDS epoch "LPU64" for ino %lu/%u\n",
 178                mds->mds_io_epoch, inode->i_ino, inode->i_generation);
 179  out:
 180         if (rc == 0)
 181                 atomic_inc(&inode->i_writecount);
 182         up(&mds->mds_epoch_sem);
 183         return rc;
 184 }
 185
 186 /* Returns EAGAIN if the client needs to get size and/or cookies and close
 187  * again -- which is never true if the file is about to be unlinked.  Otherwise
 188  * returns the number of remaining writers. */
 189 static int mds_put_write_access(struct mds_obd *mds, struct inode *inode,
 190                                 struct mds_body *body, int unlinking)
 191 {
 192         int rc = 0;
 193         ENTRY;
 194
 195         down(&mds->mds_epoch_sem);
 196         atomic_dec(&inode->i_writecount);
 197         rc = atomic_read(&inode->i_writecount);
 198         if (rc > 0)
 199                 GOTO(out, rc);
 200 #if 0
 201         if (!unlinking && !(body->valid & OBD_MD_FLSIZE))
 202                 GOTO(out, rc = EAGAIN);
 203 #endif
 204         mds_free_filterdata(inode);
 205  out:
 206         up(&mds->mds_epoch_sem);
 207         return rc;
 208 }
 209
 210 static int mds_deny_write_access(struct mds_obd *mds, struct inode *inode)
 211 {
 212         ENTRY;
 213         down(&mds->mds_epoch_sem);
 214         if (atomic_read(&inode->i_writecount) > 0) {
 215                 up(&mds->mds_epoch_sem);
 216                 RETURN(-ETXTBSY);
 217         }
 218         atomic_dec(&inode->i_writecount);
 219         up(&mds->mds_epoch_sem);
 220         RETURN(0);
 221 }
 222
 223 static void mds_allow_write_access(struct inode *inode)
 224 {
 225         ENTRY;
 226         atomic_inc(&inode->i_writecount);
 227 }
 228
 229 int mds_query_write_access(struct inode *inode)
 230 {
 231         ENTRY;
 232         RETURN(atomic_read(&inode->i_writecount));
 233 }
 234
 235 /* This replaces the VFS dentry_open, it manages mfd and writecount */
 236 static struct mds_file_data *mds_dentry_open(struct dentry *dentry,
 237                                              struct vfsmount *mnt, int flags,
 238                                              struct ptlrpc_request *req)
 239 {
 240         struct mds_export_data *med = &req->rq_export->exp_mds_data;
 241         struct mds_obd *mds = mds_req2mds(req);
 242         struct mds_file_data *mfd;
 243         struct mds_body *body;
 244         int error;
 245         ENTRY;
 246
 247         mfd = mds_mfd_new();
 248         if (mfd == NULL) {
 249                 CERROR("mds: out of memory\n");
 250                 GOTO(cleanup_dentry, error = -ENOMEM);
 251         }
 252
 253         body = lustre_msg_buf(req->rq_repmsg, 1, sizeof (*body));
 254
 255         if (flags & FMODE_WRITE) {
 256                 /* FIXME: in recovery, need to pass old epoch here */
 257                 error = mds_get_write_access(mds, dentry->d_inode, 0);
 258                 if (error)
 259                         GOTO(cleanup_mfd, error);
 260                 body->io_epoch = MDS_FILTERDATA(dentry->d_inode)->io_epoch;
 261         } else if (flags & FMODE_EXEC) {
 262                 error = mds_deny_write_access(mds, dentry->d_inode);
 263                 if (error)
 264                         GOTO(cleanup_mfd, error);
 265         }
 266
 267         dget(dentry);
 268
 269         /* Mark the file as open to handle open-unlink. */
 270         MDS_DOWN_WRITE_ORPHAN_SEM(dentry->d_inode);
 271         mds_orphan_open_inc(dentry->d_inode);
 272         MDS_UP_WRITE_ORPHAN_SEM(dentry->d_inode);
 273
 274         mfd->mfd_mode = flags;
 275         mfd->mfd_dentry = dentry;
 276         mfd->mfd_xid = req->rq_xid;
 277
 278         spin_lock(&med->med_open_lock);
 279         list_add(&mfd->mfd_list, &med->med_open_head);
 280         spin_unlock(&med->med_open_lock);
 281         mds_mfd_put(mfd);
 282
 283         body->handle.cookie = mfd->mfd_handle.h_cookie;
 284
 285         RETURN(mfd);
 286
 287 cleanup_mfd:
 288         mds_mfd_put(mfd);
 289         mds_mfd_destroy(mfd);
 290 cleanup_dentry:
 291         return ERR_PTR(error);
 292 }
 293
 294 /* Must be called with i_sem held */
 295 static int mds_create_objects(struct ptlrpc_request *req, int offset,
 296                               struct mds_update_record *rec,
 297                               struct mds_obd *mds, struct obd_device *obd,
 298                               struct dentry *dchild, void **handle, obd_id **ids,
 299                               struct llog_cookie **ret_logcookies,
 300                               int *setattr_async_flag)
 301 {
 302         struct obdo *oa;
 303         struct obd_trans_info oti = { 0 };
 304         struct mds_body *body;
 305         struct lov_stripe_md *lsm = NULL;
 306         struct lov_mds_md *lmm = NULL;
 307         struct inode *inode = dchild->d_inode;
 308         void *lmm_buf;
 309         int rc, lmm_bufsize, lmm_size;
 310         ENTRY;
 311
 312         if (rec->ur_flags & MDS_OPEN_DELAY_CREATE ||
 313             !(rec->ur_flags & FMODE_WRITE))
 314                 RETURN(0);
 315
 316         body = lustre_msg_buf(req->rq_repmsg, 1, sizeof(*body));
 317
 318         if (!S_ISREG(inode->i_mode))
 319                 RETURN(0);
 320         if (body->valid & OBD_MD_FLEASIZE)
 321                 RETURN(0);
 322
 323         OBD_ALLOC(*ids, mds->mds_lov_desc.ld_tgt_count * sizeof(**ids));
 324         if (*ids == NULL)
 325                 RETURN(-ENOMEM);
 326         oti.oti_objid = *ids;
 327
 328         /* replay case */
 329         if (lustre_msg_get_flags(req->rq_reqmsg) & MSG_REPLAY) {
 330                 if (rec->ur_fid2->id == 0) {
 331                         DEBUG_REQ(D_ERROR, req, "fid2 not set on open replay");
 332                         RETURN(-EFAULT);
 333                 }
 334
 335                 body->valid |= OBD_MD_FLBLKSZ | OBD_MD_FLEASIZE;
 336                 lmm_size = rec->ur_eadatalen;
 337                 lmm = rec->ur_eadata;
 338                 LASSERT(lmm);
 339
 340                 if (*handle == NULL)
 341                         *handle = fsfilt_start(obd,inode,FSFILT_OP_CREATE,NULL);
 342                 if (IS_ERR(*handle)) {
 343                         rc = PTR_ERR(*handle);
 344                         *handle = NULL;
 345                         GOTO(out_ids, rc);
 346                 }
 347
 348                 mds_objids_from_lmm(*ids, lmm, &mds->mds_lov_desc);
 349
 350                 lmm_buf = lustre_msg_buf(req->rq_repmsg, offset, 0);
 351                 lmm_bufsize = req->rq_repmsg->buflens[offset];
 352                 LASSERT(lmm_buf);
 353                 LASSERT(lmm_bufsize >= lmm_size);
 354                 memcpy(lmm_buf, lmm, lmm_size);
 355                 rc = fsfilt_set_md(obd, inode, *handle, lmm, lmm_size);
 356                 if (rc)
 357                         CERROR("open replay failed to set md:%d\n", rc);
 358                 RETURN(0);
 359         }
 360
 361         if (OBD_FAIL_CHECK_ONCE(OBD_FAIL_MDS_ALLOC_OBDO))
 362                 GOTO(out_ids, rc = -ENOMEM);
 363
 364         oa = obdo_alloc();
 365         if (oa == NULL)
 366                 GOTO(out_ids, rc = -ENOMEM);
 367         oa->o_mode = S_IFREG | 0600;
 368         oa->o_id = inode->i_ino;
 369         oa->o_generation = inode->i_generation;
 370         oa->o_uid = 0; /* must have 0 uid / gid on OST */
 371         oa->o_gid = 0;
 372         oa->o_valid = OBD_MD_FLID | OBD_MD_FLGENER | OBD_MD_FLTYPE |
 373                 OBD_MD_FLMODE | OBD_MD_FLUID | OBD_MD_FLGID;
 374         oa->o_size = 0;
 375
 376         obdo_from_inode(oa, inode, OBD_MD_FLTYPE|OBD_MD_FLATIME|OBD_MD_FLMTIME|
 377                         OBD_MD_FLCTIME);
 378
 379         if (!(rec->ur_flags & MDS_OPEN_HAS_OBJS)) {
 380                 /* check if things like lfs setstripe are sending us the ea */
 381                 if (rec->ur_flags & MDS_OPEN_HAS_EA) {
 382                         rc = obd_iocontrol(OBD_IOC_LOV_SETSTRIPE,
 383                                            mds->mds_osc_exp,
 384                                            0, &lsm, rec->ur_eadata);
 385                         if (rc)
 386                                 GOTO(out_oa, rc);
 387                 } else {
 388                         OBD_ALLOC(lmm, mds->mds_max_mdsize);
 389                         if (lmm == NULL)
 390                                 GOTO(out_oa, rc = -ENOMEM);
 391
 392                         lmm_size = mds->mds_max_mdsize;
 393                         rc = mds_get_md(obd, dchild->d_parent->d_inode,
 394                                         lmm, &lmm_size, 1);
 395                         if (rc > 0)
 396                                 rc = obd_iocontrol(OBD_IOC_LOV_SETSTRIPE,
 397                                                    mds->mds_osc_exp,
 398                                                    0, &lsm, lmm);
 399                         OBD_FREE(lmm, mds->mds_max_mdsize);
 400                         if (rc)
 401                                 GOTO(out_oa, rc);
 402                 }
 403                 rc = obd_create(mds->mds_osc_exp, oa, &lsm, &oti);
 404                 if (rc) {
 405                         int level = D_ERROR;
 406                         if (rc == -ENOSPC)
 407                                 level = D_INODE;
 408                         CDEBUG(level, "error creating objects for "
 409                                       "inode %lu: rc = %d\n",
 410                                inode->i_ino, rc);
 411                         if (rc > 0) {
 412                                 CERROR("obd_create returned invalid "
 413                                        "rc %d\n", rc);
 414                                 rc = -EIO;
 415                         }
 416                         GOTO(out_oa, rc);
 417                 }
 418                 *setattr_async_flag = 1;
 419         } else {
 420                 rc = obd_iocontrol(OBD_IOC_LOV_SETEA, mds->mds_osc_exp,
 421                                    0, &lsm, rec->ur_eadata);
 422                 if (rc) {
 423                         GOTO(out_oa, rc);
 424                 }
 425                 lsm->lsm_object_id = oa->o_id;
 426         }
 427         if (inode->i_size) {
 428                 oa->o_size = inode->i_size;
 429                 obdo_from_inode(oa, inode, OBD_MD_FLTYPE|OBD_MD_FLATIME|
 430                                 OBD_MD_FLMTIME| OBD_MD_FLCTIME| OBD_MD_FLSIZE);
 431                 rc = obd_setattr(mds->mds_osc_exp, oa, lsm, &oti);
 432                 if (rc) {
 433                         CERROR("error setting attrs for inode %lu: rc %d\n",
 434                                inode->i_ino, rc);
 435                         if (rc > 0) {
 436                                 CERROR("obd_setattr returned bad rc %d\n", rc);
 437                                 rc = -EIO;
 438                         }
 439                         GOTO(out_oa, rc);
 440                 }
 441         }
 442
 443         body->valid |= OBD_MD_FLBLKSZ | OBD_MD_FLEASIZE;
 444         obdo_refresh_inode(inode, oa, OBD_MD_FLBLKSZ);
 445
 446         LASSERT(lsm && lsm->lsm_object_id);
 447         lmm = NULL;
 448         rc = obd_packmd(mds->mds_osc_exp, &lmm, lsm);
 449         LASSERT(rc >= 0);
 450         lmm_size = rc;
 451         body->eadatasize = rc;
 452
 453         if (*handle == NULL) {
 454                 if (*setattr_async_flag)
 455                         *handle = fsfilt_start_log(obd, inode,
 456                                                    FSFILT_OP_CREATE, NULL,
 457                                                    le32_to_cpu(lmm->lmm_stripe_count));
 458                 else
 459                         *handle = fsfilt_start(obd, inode, FSFILT_OP_CREATE, NULL);
 460         }
 461         if (IS_ERR(*handle)) {
 462                 rc = PTR_ERR(*handle);
 463                 *handle = NULL;
 464                 GOTO(out_oa, rc);
 465         }
 466
 467         /* write mds setattr log for created objects */
 468         if (*setattr_async_flag && lmm_size) {
 469                 struct llog_cookie *logcookies = NULL;
 470
 471                 OBD_ALLOC(logcookies, mds->mds_max_cookiesize);
 472                 if (logcookies == NULL)
 473                         GOTO(out_oa, rc = -ENOMEM);
 474                 *ret_logcookies = logcookies;
 475                 if (mds_log_op_setattr(obd, inode, lmm, lmm_size, logcookies,
 476                                        mds->mds_max_cookiesize) <= 0) {
 477                         OBD_FREE(logcookies, mds->mds_max_cookiesize);
 478                         *ret_logcookies = NULL;
 479                }
 480         }
 481
 482         rc = fsfilt_set_md(obd, inode, *handle, lmm, lmm_size);
 483         lmm_buf = lustre_msg_buf(req->rq_repmsg, offset, 0);
 484         lmm_bufsize = req->rq_repmsg->buflens[offset];
 485         LASSERT(lmm_buf);
 486         LASSERT(lmm_bufsize >= lmm_size);
 487
 488         memcpy(lmm_buf, lmm, lmm_size);
 489         obd_free_diskmd(mds->mds_osc_exp, &lmm);
 490  out_oa:
 491         oti_free_cookies(&oti);
 492         obdo_free(oa);
 493  out_ids:
 494         if (rc) {
 495                 OBD_FREE(*ids, mds->mds_lov_desc.ld_tgt_count * sizeof(**ids));
 496                 *ids = NULL;
 497         }
 498         if (lsm)
 499                 obd_free_memmd(mds->mds_osc_exp, &lsm);
 500         RETURN(rc);
 501 }
 502
 503 static void reconstruct_open(struct mds_update_record *rec, int offset,
 504                              struct ptlrpc_request *req,
 505                              struct lustre_handle *child_lockh)
 506 {
 507         struct mds_export_data *med = &req->rq_export->exp_mds_data;
 508         struct mds_client_data *mcd = med->med_mcd;
 509         struct mds_obd *mds = mds_req2mds(req);
 510         struct mds_file_data *mfd;
 511         struct obd_device *obd = req->rq_export->exp_obd;
 512         struct dentry *parent, *dchild;
 513         struct ldlm_reply *rep;
 514         struct mds_body *body;
 515         int rc;
 516         struct list_head *t;
 517         int put_child = 1;
 518         ENTRY;
 519
 520         LASSERT(offset == 2);                  /* only called via intent */
 521         rep = lustre_msg_buf(req->rq_repmsg, 0, sizeof (*rep));
 522         body = lustre_msg_buf(req->rq_repmsg, 1, sizeof (*body));
 523
 524         /* copy rc, transno and disp; steal locks */
 525         mds_req_from_mcd(req, mcd);
 526         intent_set_disposition(rep, mcd->mcd_last_data);
 527
 528         /* Only replay if create or open actually happened. */
 529         if (!intent_disposition(rep, DISP_OPEN_CREATE | DISP_OPEN_OPEN) ) {
 530                 EXIT;
 531                 return; /* error looking up parent or child */
 532         }
 533
 534         parent = mds_fid2dentry(mds, rec->ur_fid1, NULL);
 535         LASSERT(!IS_ERR(parent));
 536
 537         dchild = ll_lookup_one_len(rec->ur_name, parent, rec->ur_namelen - 1);
 538         LASSERT(!IS_ERR(dchild));
 539
 540         if (!dchild->d_inode)
 541                 GOTO(out_dput, 0); /* child not present to open */
 542
 543         /* At this point, we know we have a child. We'll send
 544          * it back _unless_ it not created and open failed.
 545          */
 546         if (intent_disposition(rep, DISP_OPEN_OPEN) &&
 547             !intent_disposition(rep, DISP_OPEN_CREATE) &&
 548             req->rq_status) {
 549                 GOTO(out_dput, 0);
 550         }
 551
 552         mds_pack_inode2fid(&body->fid1, dchild->d_inode);
 553         mds_pack_inode2body(body, dchild->d_inode);
 554         if (S_ISREG(dchild->d_inode->i_mode)) {
 555                 rc = mds_pack_md(obd, req->rq_repmsg, 2, body,
 556                                  dchild->d_inode, 1);
 557
 558                 if (rc)
 559                         LASSERT(rc == req->rq_status);
 560
 561                 /* If we have LOV EA data, the OST holds size, mtime */
 562                 if (!(body->valid & OBD_MD_FLEASIZE))
 563                         body->valid |= (OBD_MD_FLSIZE | OBD_MD_FLBLOCKS |
 564                                         OBD_MD_FLATIME | OBD_MD_FLMTIME);
 565         }
 566
 567         /* If we have -EEXIST as the status, and we were asked to create
 568          * exclusively, we can tell we failed because the file already existed.
 569          */
 570         if (req->rq_status == -EEXIST &&
 571             ((rec->ur_flags & (MDS_OPEN_CREAT | MDS_OPEN_EXCL)) ==
 572              (MDS_OPEN_CREAT | MDS_OPEN_EXCL))) {
 573                 GOTO(out_dput, 0);
 574         }
 575
 576         /* If we didn't get as far as trying to open, then some locking thing
 577          * probably went wrong, and we'll just bail here.
 578          */
 579         if (!intent_disposition(rep, DISP_OPEN_OPEN))
 580                 GOTO(out_dput, 0);
 581
 582         /* If we failed, then we must have failed opening, so don't look for
 583          * file descriptor or anything, just give the client the bad news.
 584          */
 585         if (req->rq_status)
 586                 GOTO(out_dput, 0);
 587
 588         mfd = NULL;
 589         list_for_each(t, &med->med_open_head) {
 590                 mfd = list_entry(t, struct mds_file_data, mfd_list);
 591                 if (mfd->mfd_xid == req->rq_xid)
 592                         break;
 593                 mfd = NULL;
 594         }
 595
 596         /* #warning "XXX fixme" bug 2991 */
 597         /* Here it used to LASSERT(mfd) if exp_outstanding_reply != NULL.
 598          * Now that exp_outstanding_reply is a list, it's just using mfd != NULL
 599          * to detect a re-open */
 600         if (mfd == NULL) {
 601                 mntget(mds->mds_vfsmnt);
 602                 CERROR("Re-opened file \n");
 603                 mfd = mds_dentry_open(dchild, mds->mds_vfsmnt,
 604                                       rec->ur_flags & ~MDS_OPEN_TRUNC, req);
 605                 if (!mfd) {
 606                         CERROR("mds: out of memory\n");
 607                         GOTO(out_dput, req->rq_status = -ENOMEM);
 608                 }
 609                 put_child = 0;
 610         } else {
 611                 body->handle.cookie = mfd->mfd_handle.h_cookie;
 612                 CDEBUG(D_INODE, "resend mfd %p, cookie "LPX64"\n", mfd,
 613                        mfd->mfd_handle.h_cookie);
 614         }
 615
 616  out_dput:
 617         if (put_child)
 618                 l_dput(dchild);
 619         l_dput(parent);
 620         EXIT;
 621 }
 622
 623 /* do NOT or the MAY_*'s, you'll get the weakest */
 624 static int accmode(struct inode *inode, int flags)
 625 {
 626         int res = 0;
 627
 628         /* Sadly, NFSD reopens a file repeatedly during operation, so the
 629          * "acc_mode = 0" allowance for newly-created files isn't honoured.
 630          * NFSD uses the MDS_OPEN_OWNEROVERRIDE flag to say that a file
 631          * owner can write to a file even if it is marked readonly to hide
 632          * its brokenness. (bug 5781) */
 633         if (flags & MDS_OPEN_OWNEROVERRIDE && inode->i_uid == current->fsuid)
 634                 return 0;
 635
 636         if (flags & FMODE_READ)
 637                 res = MAY_READ;
 638         if (flags & (FMODE_WRITE|MDS_OPEN_TRUNC))
 639                 res |= MAY_WRITE;
 640         if (flags & FMODE_EXEC)
 641                 res = MAY_EXEC;
 642         return res;
 643 }
 644
 645 /* Handles object creation, actual opening, and I/O epoch */
 646 static int mds_finish_open(struct ptlrpc_request *req, struct dentry *dchild,
 647                            struct mds_body *body, int flags, void **handle,
 648                            struct mds_update_record *rec,struct ldlm_reply *rep,
 649                            struct llog_cookie **logcookies,
 650                            int *setattr_async_flag)
 651 {
 652         struct mds_obd *mds = mds_req2mds(req);
 653         struct obd_device *obd = req->rq_export->exp_obd;
 654         struct mds_file_data *mfd = NULL;
 655         obd_id *ids = NULL; /* object IDs created */
 656         int rc = 0;
 657         ENTRY;
 658
 659         /* atomically create objects if necessary */
 660         down(&dchild->d_inode->i_sem);
 661         if (S_ISREG(dchild->d_inode->i_mode) &&
 662             !(body->valid & OBD_MD_FLEASIZE)) {
 663                 rc = mds_pack_md(obd, req->rq_repmsg, 2, body,
 664                                  dchild->d_inode, 0);
 665                 if (rc) {
 666                         up(&dchild->d_inode->i_sem);
 667                         RETURN(rc);
 668                 }
 669         }
 670         if (rec != NULL) {
 671                 if ((body->valid & OBD_MD_FLEASIZE) &&
 672                     (rec->ur_flags & MDS_OPEN_HAS_EA)) {
 673                         up(&dchild->d_inode->i_sem);
 674                         RETURN(-EEXIST);
 675                 }
 676
 677                 if (!(body->valid & OBD_MD_FLEASIZE)) {
 678                         /* no EA: create objects */
 679                         rc = mds_create_objects(req, 2, rec, mds, obd,
 680                                                 dchild, handle, &ids,
 681                                                 logcookies, setattr_async_flag);
 682                         if (rc) {
 683                                 CERROR("mds_create_objects: rc = %d\n", rc);
 684                                 up(&dchild->d_inode->i_sem);
 685                                 RETURN(rc);
 686                         }
 687                 }
 688         }
 689         /* If the inode has no EA data, then MDS holds size, mtime */
 690         if (S_ISREG(dchild->d_inode->i_mode) &&
 691             !(body->valid & OBD_MD_FLEASIZE)) {
 692                 body->valid |= (OBD_MD_FLSIZE | OBD_MD_FLBLOCKS |
 693                                 OBD_MD_FLATIME | OBD_MD_FLMTIME);
 694         }
 695         up(&dchild->d_inode->i_sem);
 696
 697         intent_set_disposition(rep, DISP_OPEN_OPEN);
 698         mfd = mds_dentry_open(dchild, mds->mds_vfsmnt, flags, req);
 699         if (IS_ERR(mfd))
 700                 RETURN(PTR_ERR(mfd));
 701
 702         CDEBUG(D_INODE, "mfd %p, cookie "LPX64"\n", mfd,
 703                mfd->mfd_handle.h_cookie);
 704         if (ids != NULL) {
 705                 mds_lov_update_objids(obd, ids);
 706                 OBD_FREE(ids, sizeof(*ids) * mds->mds_lov_desc.ld_tgt_count);
 707         }
 708         //if (rc)
 709         //        mds_mfd_destroy(mfd);
 710         RETURN(rc);
 711 }
 712
 713 static int mds_open_by_fid(struct ptlrpc_request *req, struct ll_fid *fid,
 714                            struct mds_body *body, int flags,
 715                            struct mds_update_record *rec,struct ldlm_reply *rep)
 716 {
 717         struct obd_device *obd = req->rq_export->exp_obd;
 718         struct mds_obd *mds = mds_req2mds(req);
 719         struct dentry *dchild;
 720         char fidname[LL_FID_NAMELEN];
 721         int fidlen = 0, rc;
 722         void *handle = NULL;
 723         struct llog_cookie *logcookies = NULL;
 724         struct lov_mds_md *lmm = NULL;
 725         int lmm_size = 0;
 726         int setattr_async_flag = 0;
 727         ENTRY;
 728
 729         fidlen = ll_fid2str(fidname, fid->id, fid->generation);
 730         dchild = ll_lookup_one_len(fidname, mds->mds_pending_dir, fidlen);
 731         if (IS_ERR(dchild)) {
 732                 rc = PTR_ERR(dchild);
 733                 CERROR("error looking up %s in PENDING: rc = %d\n",fidname, rc);
 734                 RETURN(rc);
 735         }
 736
 737         if (dchild->d_inode != NULL) {
 738                 mds_inode_set_orphan(dchild->d_inode);
 739                 mds_pack_inode2fid(&body->fid1, dchild->d_inode);
 740                 mds_pack_inode2body(body, dchild->d_inode);
 741                 intent_set_disposition(rep, DISP_LOOKUP_EXECD);
 742                 intent_set_disposition(rep, DISP_LOOKUP_POS);
 743                 CWARN("Orphan %s found and opened in PENDING directory\n",
 744                        fidname);
 745                 goto open;
 746         }
 747         l_dput(dchild);
 748
 749         /* We didn't find it in PENDING so it isn't an orphan.  See
 750          * if it was a regular inode that was previously created. */
 751         dchild = mds_fid2dentry(mds, fid, NULL);
 752         if (IS_ERR(dchild))
 753                 RETURN(PTR_ERR(dchild));
 754
 755         mds_pack_inode2fid(&body->fid1, dchild->d_inode);
 756         mds_pack_inode2body(body, dchild->d_inode);
 757         intent_set_disposition(rep, DISP_LOOKUP_EXECD);
 758         intent_set_disposition(rep, DISP_LOOKUP_POS);
 759
 760  open:
 761         rc = mds_finish_open(req, dchild, body, flags, &handle, rec, rep,
 762                              &logcookies, &setattr_async_flag);
 763         rc = mds_finish_transno(mds, dchild ? dchild->d_inode : NULL, handle,
 764                                 req, rc, rep ? rep->lock_policy_res1 : 0);
 765         /* do mds to ost setattr for new created objects */
 766         if (rc == 0 && setattr_async_flag) {
 767                 lmm = lustre_msg_buf(req->rq_repmsg, 2, 0);
 768                 lmm_size = req->rq_repmsg->buflens[2];
 769                 rc = mds_osc_setattr_async(obd, dchild->d_inode, lmm, lmm_size,
 770                                            logcookies);
 771         }
 772         /* XXX what do we do here if mds_finish_transno itself failed? */
 773
 774         l_dput(dchild);
 775         RETURN(rc);
 776 }
 777
 778 int mds_pin(struct ptlrpc_request *req)
 779 {
 780         struct obd_device *obd = req->rq_export->exp_obd;
 781         struct mds_body *request_body, *reply_body;
 782         struct obd_run_ctxt saved;
 783         int rc, size = sizeof(*reply_body);
 784         ENTRY;
 785
 786         request_body = lustre_msg_buf(req->rq_reqmsg, 0, sizeof(*request_body));
 787
 788         rc = lustre_pack_reply(req, 1, &size, NULL);
 789         if (rc)
 790                 RETURN(rc);
 791         reply_body = lustre_msg_buf(req->rq_repmsg, 0, sizeof(*reply_body));
 792
 793         push_ctxt(&saved, &obd->obd_ctxt, NULL);
 794         rc = mds_open_by_fid(req, &request_body->fid1, reply_body,
 795                              request_body->flags, NULL, NULL);
 796         pop_ctxt(&saved, &obd->obd_ctxt, NULL);
 797
 798         RETURN(rc);
 799 }
 800
 801 /*  Get a lock on the ino to sync with creation WRT inode reuse (bug 2029).
 802  *  If child_lockh is NULL we just get the lock as a barrier to wait for
 803  *  other holders of this lock, and drop it right away again. */
 804 int mds_lock_new_child(struct obd_device *obd, struct inode *inode,
 805                        struct lustre_handle *child_lockh)
 806 {
 807         struct ldlm_res_id child_res_id = { .name = { inode->i_ino, 0, 1, 0 } };
 808         struct lustre_handle lockh;
 809         int lock_flags = 0;
 810         int rc;
 811
 812         if (child_lockh == NULL)
 813                 child_lockh = &lockh;
 814
 815         rc = ldlm_cli_enqueue(NULL, NULL, obd->obd_namespace, child_res_id,
 816                               LDLM_PLAIN, NULL, LCK_EX, &lock_flags,
 817                               mds_blocking_ast, ldlm_completion_ast, NULL, NULL,
 818                               NULL, 0, NULL, child_lockh);
 819         if (rc != ELDLM_OK)
 820                 CERROR("ldlm_cli_enqueue: %d\n", rc);
 821         else if (child_lockh == &lockh)
 822                 ldlm_lock_decref(child_lockh, LCK_EX);
 823
 824         RETURN(rc);
 825 }
 826
 827 int mds_open(struct mds_update_record *rec, int offset,
 828              struct ptlrpc_request *req, struct lustre_handle *child_lockh)
 829 {
 830         /* XXX ALLOCATE _something_ - 464 bytes on stack here */
 831         struct obd_device *obd = req->rq_export->exp_obd;
 832         struct mds_obd *mds = mds_req2mds(req);
 833         struct ldlm_reply *rep = NULL;
 834         struct mds_body *body = NULL;
 835         struct dentry *dchild = NULL, *dparent = NULL;
 836         struct mds_export_data *med;
 837         struct lustre_handle parent_lockh;
 838         int rc = 0, cleanup_phase = 0, acc_mode, created = 0;
 839         int parent_mode = LCK_PR;
 840         void *handle = NULL;
 841         struct dentry_params dp;
 842         struct lov_mds_md *lmm = NULL;
 843         int lmm_size = 0;
 844         struct llog_cookie *logcookies = NULL;
 845         int setattr_async_flag = 0;
 846         uid_t parent_uid = 0;
 847         gid_t parent_gid = 0;
 848         ENTRY;
 849
 850         if (offset == 2) { /* intent */
 851                 rep = lustre_msg_buf(req->rq_repmsg, 0, sizeof (*rep));
 852                 body = lustre_msg_buf(req->rq_repmsg, 1, sizeof (*body));
 853         } else if (offset == 0) { /* non-intent reint */
 854                 body = lustre_msg_buf(req->rq_repmsg, 0, sizeof (*body));
 855         } else {
 856                 body = NULL;
 857                 LBUG();
 858         }
 859
 860         MDS_CHECK_RESENT(req, reconstruct_open(rec, offset, req, child_lockh));
 861
 862         /* Step 0: If we are passed a fid, then we assume the client already
 863          * opened this file and is only replaying the RPC, so we open the
 864          * inode by fid (at some large expense in security). */
 865         if (lustre_msg_get_flags(req->rq_reqmsg) & MSG_REPLAY) {
 866                 if (rec->ur_fid2->id == 0) {
 867                         struct ldlm_lock *lock = ldlm_handle2lock(child_lockh);
 868                         if (lock) {
 869                                 LDLM_ERROR(lock, "fid2 not set on open replay");
 870                                 LDLM_LOCK_PUT(lock);
 871                         }
 872                         DEBUG_REQ(D_ERROR, req, "fid2 not set on open replay");
 873                         RETURN(-EFAULT);
 874                 }
 875
 876                 rc = mds_open_by_fid(req, rec->ur_fid2, body, rec->ur_flags,
 877                                      rec, rep);
 878                 if (rc != -ENOENT)
 879                         RETURN(rc);
 880
 881                 /* We didn't find the correct inode on disk either, so we
 882                  * need to re-create it via a regular replay. */
 883                 if (!(rec->ur_flags & MDS_OPEN_CREAT)) {
 884                         DEBUG_REQ(D_ERROR, req,"OPEN_CREAT not in open replay");
 885                         RETURN(-EFAULT);
 886                 }
 887         } else if (rec->ur_fid2->id) {
 888                 DEBUG_REQ(D_ERROR, req, "fid2 "LPU64"/%u on open non-replay",
 889                           rec->ur_fid2->id, rec->ur_fid2->generation);
 890                 RETURN(-EFAULT);
 891         }
 892
 893         LASSERT(offset == 2); /* If we got here, we must be called via intent */
 894
 895         med = &req->rq_export->exp_mds_data;
 896         if (OBD_FAIL_CHECK(OBD_FAIL_MDS_OPEN_PACK)) {
 897                 CERROR("test case OBD_FAIL_MDS_OPEN_PACK\n");
 898                 RETURN(-ENOMEM);
 899         }
 900
 901         /* Step 1: Find and lock the parent */
 902         if (rec->ur_flags & MDS_OPEN_CREAT)
 903                 parent_mode = LCK_PW;
 904         dparent = mds_fid2locked_dentry(obd, rec->ur_fid1, NULL, parent_mode,
 905                                         &parent_lockh, rec->ur_name,
 906                                         rec->ur_namelen - 1);
 907         if (IS_ERR(dparent)) {
 908                 rc = PTR_ERR(dparent);
 909                 if (rc != -ENOENT) {
 910                         CERROR("parent "LPU64"/%u lookup error %d\n",
 911                                rec->ur_fid1->id, rec->ur_fid1->generation, rc);
 912                 } else {
 913                         /* Just cannot find parent - make it look like
 914                          * usual negative lookup to avoid extra MDS RPC */
 915                         intent_set_disposition(rep, DISP_LOOKUP_EXECD);
 916                         intent_set_disposition(rep, DISP_LOOKUP_NEG);
 917                 }
 918                 GOTO(cleanup, rc);
 919         }
 920         LASSERT(dparent->d_inode != NULL);
 921
 922         cleanup_phase = 1; /* parent dentry and lock */
 923
 924         /* Step 2: Lookup the child */
 925         dchild = ll_lookup_one_len(rec->ur_name, dparent, rec->ur_namelen - 1);
 926         if (IS_ERR(dchild)) {
 927                 rc = PTR_ERR(dchild);
 928                 dchild = NULL; /* don't confuse mds_finish_transno() below */
 929                 GOTO(cleanup, rc);
 930         }
 931
 932         cleanup_phase = 2; /* child dentry */
 933
 934         intent_set_disposition(rep, DISP_LOOKUP_EXECD);
 935         if (dchild->d_inode)
 936                 intent_set_disposition(rep, DISP_LOOKUP_POS);
 937         else
 938                 intent_set_disposition(rep, DISP_LOOKUP_NEG);
 939
 940         /*Step 3: If the child was negative, and we're supposed to, create it.*/
 941         if (dchild->d_inode == NULL) {
 942                 unsigned long ino = rec->ur_fid2->id;
 943                 struct iattr iattr;
 944                 struct inode *inode;
 945
 946                 if (!(rec->ur_flags & MDS_OPEN_CREAT)) {
 947                         /* It's negative and we weren't supposed to create it */
 948                         GOTO(cleanup, rc = -ENOENT);
 949                 }
 950
 951                 if (req->rq_export->exp_connect_flags & OBD_CONNECT_RDONLY)
 952                         GOTO(cleanup, rc = -EROFS);
 953
 954                 intent_set_disposition(rep, DISP_OPEN_CREATE);
 955                 handle = fsfilt_start(obd, dparent->d_inode, FSFILT_OP_CREATE,
 956                                       NULL);
 957                 if (IS_ERR(handle)) {
 958                         rc = PTR_ERR(handle);
 959                         handle = NULL;
 960                         GOTO(cleanup, rc);
 961                 }
 962                 dchild->d_fsdata = (void *) &dp;
 963                 dp.p_ptr = req;
 964                 dp.p_inum = ino;
 965
 966                 rc = ll_vfs_create(dparent->d_inode, dchild, rec->ur_mode,NULL);
 967                 if (dchild->d_fsdata == (void *)(unsigned long)ino)
 968                         dchild->d_fsdata = NULL;
 969
 970                 if (rc) {
 971                         CDEBUG(D_INODE, "error during create: %d\n", rc);
 972                         GOTO(cleanup, rc);
 973                 }
 974                 inode = dchild->d_inode;
 975                 if (ino) {
 976                         LASSERT(ino == inode->i_ino);
 977                         /* Written as part of setattr */
 978                         inode->i_generation = rec->ur_fid2->generation;
 979                         CDEBUG(D_HA, "recreated ino %lu with gen %u\n",
 980                                inode->i_ino, inode->i_generation);
 981                 }
 982
 983                 created = 1;
 984                 LTIME_S(iattr.ia_atime) = rec->ur_time;
 985                 LTIME_S(iattr.ia_ctime) = rec->ur_time;
 986                 LTIME_S(iattr.ia_mtime) = rec->ur_time;
 987
 988                 iattr.ia_uid = rec->ur_fsuid;
 989                 if (dparent->d_inode->i_mode & S_ISGID)
 990                         iattr.ia_gid = dparent->d_inode->i_gid;
 991                 else
 992                         iattr.ia_gid = rec->ur_fsgid;
 993
 994                 iattr.ia_valid = ATTR_UID | ATTR_GID | ATTR_ATIME |
 995                         ATTR_MTIME | ATTR_CTIME;
 996
 997                 rc = fsfilt_setattr(obd, dchild, handle, &iattr, 0);
 998                 if (rc)
 999                         CERROR("error on child setattr: rc = %d\n", rc);
1000
1001                 iattr.ia_valid = ATTR_MTIME | ATTR_CTIME;
1002
1003                 rc = fsfilt_setattr(obd, dparent, handle, &iattr, 0);
1004                 if (rc)
1005                         CERROR("error on parent setattr: rc = %d\n", rc);
1006
1007                 rc = fsfilt_commit(obd, dchild->d_inode, handle, 0);
1008                 handle = NULL;
1009                 acc_mode = 0;           /* Don't check for permissions */
1010         } else {
1011                 acc_mode = accmode(dchild->d_inode, rec->ur_flags);
1012         }
1013
1014
1015         LASSERTF(!mds_inode_is_orphan(dchild->d_inode),
1016                  "dchild %.*s (%p) inode %p\n", dchild->d_name.len,
1017                  dchild->d_name.name, dchild, dchild->d_inode);
1018
1019         mds_pack_inode2fid(&body->fid1, dchild->d_inode);
1020         mds_pack_inode2body(body, dchild->d_inode);
1021
1022         if (S_ISREG(dchild->d_inode->i_mode)) {
1023                 /* Check permissions etc */
1024                 rc = ll_permission(dchild->d_inode, acc_mode, NULL);
1025                 if (rc != 0)
1026                         GOTO(cleanup, rc);
1027
1028                 if ((req->rq_export->exp_connect_flags & OBD_CONNECT_RDONLY) &&
1029                     (acc_mode & MAY_WRITE))
1030                         GOTO(cleanup, rc = -EROFS);
1031
1032                 /* Can't write to a read-only file */
1033                 if (IS_RDONLY(dchild->d_inode) && (acc_mode & MAY_WRITE) != 0)
1034                         GOTO(cleanup, rc = -EPERM);
1035
1036                 /* An append-only file must be opened in append mode for
1037                  * writing */
1038                 if (IS_APPEND(dchild->d_inode) && (acc_mode & MAY_WRITE) != 0 &&
1039                     ((rec->ur_flags & MDS_OPEN_APPEND) == 0 ||
1040                      (rec->ur_flags & MDS_OPEN_TRUNC) != 0))
1041                         GOTO(cleanup, rc = -EPERM);
1042         }
1043
1044         if (!created && (rec->ur_flags & MDS_OPEN_CREAT) &&
1045             (rec->ur_flags & MDS_OPEN_EXCL)) {
1046                 /* File already exists, we didn't just create it, and we
1047                  * were passed O_EXCL; err-or. */
1048                 GOTO(cleanup, rc = -EEXIST); // returns a lock to the client
1049         }
1050
1051         /* if we are following a symlink, don't open */
1052         if (S_ISLNK(dchild->d_inode->i_mode))
1053                 GOTO(cleanup_no_trans, rc = 0);
1054
1055         if (S_ISDIR(dchild->d_inode->i_mode)) {
1056                 if (rec->ur_flags & MDS_OPEN_CREAT ||
1057                     rec->ur_flags & FMODE_WRITE) {
1058                         /* we are trying to create or write a exist dir */
1059                         GOTO(cleanup, rc = -EISDIR);
1060                 }
1061                 if (ll_permission(dchild->d_inode, acc_mode, NULL)) {
1062                         intent_set_disposition(rep, DISP_OPEN_OPEN);
1063                         GOTO(cleanup, rc = -EACCES);
1064                 }
1065         } else if (rec->ur_flags & MDS_OPEN_DIRECTORY) {
1066                 GOTO(cleanup, rc = -ENOTDIR);
1067         }
1068
1069         if (OBD_FAIL_CHECK(OBD_FAIL_MDS_OPEN_CREATE)) {
1070                 obd_fail_loc = OBD_FAIL_LDLM_REPLY | OBD_FAIL_ONCE;
1071                 GOTO(cleanup, rc = -EAGAIN);
1072         }
1073
1074         /* Step 5: mds_open it */
1075         rc = mds_finish_open(req, dchild, body, rec->ur_flags, &handle, rec,
1076                              rep, &logcookies, &setattr_async_flag);
1077         GOTO(cleanup, rc);
1078
1079  cleanup:
1080         rc = mds_finish_transno(mds, dchild ? dchild->d_inode : NULL, handle,
1081                                 req, rc, rep ? rep->lock_policy_res1 : 0);
1082         /* do mds to ost setattr for new created objects */
1083         if (rc == 0 && setattr_async_flag) {
1084                 lmm = lustre_msg_buf(req->rq_repmsg, 2, 0);
1085                 lmm_size = req->rq_repmsg->buflens[2];
1086                 mds_osc_setattr_async(obd, dchild->d_inode, lmm, lmm_size,
1087                                       logcookies);
1088         }
1089
1090  cleanup_no_trans:
1091         switch (cleanup_phase) {
1092         case 2:
1093                 if (rc && created) {
1094                         int err = vfs_unlink(dparent->d_inode, dchild);
1095                         if (err) {
1096                                 CERROR("unlink(%.*s) in error path: %d\n",
1097                                        dchild->d_name.len, dchild->d_name.name,
1098                                        err);
1099                         }
1100                 } else if (created) {
1101                         mds_lock_new_child(obd, dchild->d_inode, NULL);
1102                         /* save uid/gid for quota acquire/release */
1103                         parent_uid = dparent->d_inode->i_uid;
1104                         parent_gid = dparent->d_inode->i_gid;
1105
1106                 }
1107                 l_dput(dchild);
1108         case 1:
1109                 if (dparent == NULL)
1110                         break;
1111
1112                 l_dput(dparent);
1113                 if (rc)
1114                         ldlm_lock_decref(&parent_lockh, parent_mode);
1115                 else
1116                         ptlrpc_save_lock (req, &parent_lockh, parent_mode);
1117         }
1118
1119         /* trigger dqacq on the owner of child and parent */
1120         mds_adjust_qunit(obd, current->fsuid, current->fsgid,
1121                          parent_uid, parent_gid, rc);
1122         RETURN(rc);
1123 }
1124
1125 /* Close a "file descriptor" and possibly unlink an orphan from the
1126  * PENDING directory.  Caller must hold child->i_sem, this drops it.
1127  *
1128  * If we are being called from mds_disconnect() because the client has
1129  * disappeared, then req == NULL and we do not update last_rcvd because
1130  * there is nothing that could be recovered by the client at this stage
1131  * (it will not even _have_ an entry in last_rcvd anymore).
1132  *
1133  * Returns EAGAIN if the client needs to get more data and re-close. */
1134 int mds_mfd_close(struct ptlrpc_request *req, struct obd_device *obd,
1135                   struct mds_file_data *mfd, int unlink_orphan)
1136 {
1137         struct inode *inode = mfd->mfd_dentry->d_inode;
1138         char fidname[LL_FID_NAMELEN];
1139         int last_orphan, fidlen, rc = 0, cleanup_phase = 0;
1140         struct dentry *pending_child = NULL;
1141         struct mds_obd *mds = &obd->u.mds;
1142         struct inode *pending_dir = mds->mds_pending_dir->d_inode;
1143         void *handle = NULL;
1144         struct mds_body *request_body = NULL, *reply_body = NULL;
1145         struct dentry_params dp;
1146         struct iattr iattr = { 0 };
1147         ENTRY;
1148
1149         if (req && req->rq_reqmsg != NULL)
1150                 request_body = lustre_msg_buf(req->rq_reqmsg, 0,
1151                                               sizeof(*request_body));
1152         if (req && req->rq_repmsg != NULL)
1153                 reply_body = lustre_msg_buf(req->rq_repmsg, 0,
1154                                             sizeof(*reply_body));
1155
1156         fidlen = ll_fid2str(fidname, inode->i_ino, inode->i_generation);
1157
1158         CDEBUG(D_INODE, "inode %p ino %s nlink %d orphan %d\n", inode, fidname,
1159                inode->i_nlink, mds_orphan_open_count(inode));
1160
1161         last_orphan = mds_orphan_open_dec_test(inode) &&
1162                 mds_inode_is_orphan(inode);
1163         MDS_UP_WRITE_ORPHAN_SEM(inode);
1164
1165         /* this is half of the actual "close" */
1166         if (mfd->mfd_mode & FMODE_WRITE) {
1167                 rc = mds_put_write_access(mds, inode, request_body,
1168                                           last_orphan && unlink_orphan);
1169         } else if (mfd->mfd_mode & FMODE_EXEC) {
1170                 mds_allow_write_access(inode);
1171         }
1172
1173         if (last_orphan && unlink_orphan) {
1174                 struct lov_mds_md *lmm = NULL;
1175                 int stripe_count = 0;
1176                 LASSERT(rc == 0); /* mds_put_write_access must have succeeded */
1177
1178                 CDEBUG(D_HA, "destroying orphan object %s\n", fidname);
1179
1180                 if ((S_ISREG(inode->i_mode) && inode->i_nlink != 1) ||
1181                     (S_ISDIR(inode->i_mode) && inode->i_nlink != 2))
1182                         CERROR("found \"orphan\" %s %s with link count %d\n",
1183                                S_ISREG(inode->i_mode) ? "file" : "dir",
1184                                fidname, inode->i_nlink);
1185
1186                 /* Sadly, there is no easy way to save pending_child from
1187                  * mds_reint_unlink() into mfd, so we need to re-lookup,
1188                  * but normally it will still be in the dcache. */
1189                 down(&pending_dir->i_sem);
1190                 cleanup_phase = 1; /* up(&pending_dir->i_sem) when finished */
1191                 pending_child = lookup_one_len(fidname, mds->mds_pending_dir,
1192                                                fidlen);
1193                 if (IS_ERR(pending_child))
1194                         GOTO(cleanup, rc = PTR_ERR(pending_child));
1195                 LASSERT(pending_child->d_inode != NULL);
1196
1197                 cleanup_phase = 2; /* dput(pending_child) when finished */
1198                 if (S_ISDIR(pending_child->d_inode->i_mode)) {
1199                         rc = vfs_rmdir(pending_dir, pending_child);
1200                         if (rc)
1201                                 CERROR("error unlinking orphan dir %s: rc %d\n",
1202                                        fidname,rc);
1203                         goto out;
1204                 }
1205
1206                 if (req != NULL && req->rq_repmsg != NULL) {
1207                         lmm = lustre_msg_buf(req->rq_repmsg, 1, 0);
1208                         stripe_count = le32_to_cpu(lmm->lmm_stripe_count);
1209                 }
1210
1211                 handle = fsfilt_start_log(obd, pending_dir, FSFILT_OP_UNLINK,
1212                                           NULL, stripe_count);
1213                 if (IS_ERR(handle)) {
1214                         rc = PTR_ERR(handle);
1215                         handle = NULL;
1216                         GOTO(cleanup, rc);
1217                 }
1218
1219                 if (req != NULL && req->rq_repmsg != NULL &&
1220                     (reply_body->valid & OBD_MD_FLEASIZE) &&
1221                     mds_log_op_unlink(obd, pending_child->d_inode, lmm,
1222                                       req->rq_repmsg->buflens[1],
1223                                       lustre_msg_buf(req->rq_repmsg, 2, 0),
1224                                       req->rq_repmsg->buflens[2]) > 0) {
1225                         reply_body->valid |= OBD_MD_FLCOOKIE;
1226                 }
1227
1228                 pending_child->d_fsdata = (void *) &dp;
1229                 dp.p_inum = 0;
1230                 dp.p_ptr = req;
1231                 rc = vfs_unlink(pending_dir, pending_child);
1232                 if (rc)
1233                         CERROR("error unlinking orphan %s: rc %d\n",fidname,rc);
1234
1235                 goto out; /* Don't bother updating attrs on unlinked inode */
1236         }
1237
1238 #if 0
1239         if (request_body != NULL && mfd->mfd_mode & FMODE_WRITE && rc == 0) {
1240                 /* Update the on-disk attributes if this was the last write
1241                  * close, and all information was provided (i.e., rc == 0)
1242                  *
1243                  * XXX this should probably be abstracted with mds_reint_setattr
1244                  */
1245
1246                 if (request_body->valid & OBD_MD_FLMTIME &&
1247                     LTIME_S(iattr.ia_mtime) > LTIME_S(inode->i_mtime)) {
1248                         LTIME_S(iattr.ia_mtime) = request_body->mtime;
1249                         iattr.ia_valid |= ATTR_MTIME;
1250                 }
1251                 if (request_body->valid & OBD_MD_FLCTIME &&
1252                     LTIME_S(iattr.ia_ctime) > LTIME_S(inode->i_ctime)) {
1253                         LTIME_S(iattr.ia_ctime) = request_body->ctime;
1254                         iattr.ia_valid |= ATTR_CTIME;
1255                 }
1256
1257                 /* XXX can't set block count with fsfilt_setattr (!) */
1258                 if (request_body->valid & OBD_MD_FLSIZE) {
1259                         iattr.ia_valid |= ATTR_SIZE;
1260                         iattr.ia_size = request_body->size;
1261                 }
1262                 /* iattr.ia_blocks = request_body->blocks */
1263
1264         }
1265 #endif
1266         if (request_body != NULL && request_body->valid & OBD_MD_FLATIME) {
1267                 /* Only start a transaction to write out only the atime if
1268                  * it is more out-of-date than the specified limit.  If we
1269                  * are already going to write out the atime then do it anyway.
1270                  * */
1271                 LTIME_S(iattr.ia_atime) = request_body->atime;
1272                 if ((LTIME_S(iattr.ia_atime) >
1273                      LTIME_S(inode->i_atime) + MAX_ATIME_DIFF) ||
1274                     (iattr.ia_valid != 0 &&
1275                      LTIME_S(iattr.ia_atime) > LTIME_S(inode->i_atime)))
1276                         iattr.ia_valid |= ATTR_ATIME;
1277         }
1278
1279         if (iattr.ia_valid != 0) {
1280                 handle = fsfilt_start(obd, inode, FSFILT_OP_SETATTR, NULL);
1281                 if (IS_ERR(handle))
1282                         GOTO(cleanup, rc = PTR_ERR(handle));
1283                 rc = fsfilt_setattr(obd, mfd->mfd_dentry, handle, &iattr, 0);
1284                 if (rc)
1285                         CERROR("error in setattr(%s): rc %d\n", fidname, rc);
1286         }
1287 out:
1288         /* If other clients have this file open for write, rc will be > 0 */
1289         if (rc > 0)
1290                 rc = 0;
1291         l_dput(mfd->mfd_dentry);
1292         mds_mfd_destroy(mfd);
1293
1294  cleanup:
1295         if (req != NULL && reply_body != NULL) {
1296                 rc = mds_finish_transno(mds, pending_dir, handle, req, rc, 0);
1297         } else if (handle) {
1298                 int err = fsfilt_commit(obd, pending_dir, handle, 0);
1299                 if (err) {
1300                         CERROR("error committing close: %d\n", err);
1301                         if (!rc)
1302                                 rc = err;
1303                 }
1304         }
1305
1306         switch (cleanup_phase) {
1307         case 2:
1308                 dput(pending_child);
1309         case 1:
1310                 up(&pending_dir->i_sem);
1311         }
1312         RETURN(rc);
1313 }
1314
1315 int mds_close(struct ptlrpc_request *req)
1316 {
1317         struct mds_export_data *med = &req->rq_export->exp_mds_data;
1318         struct obd_device *obd = req->rq_export->exp_obd;
1319         struct mds_body *body;
1320         struct mds_file_data *mfd;
1321         struct obd_run_ctxt saved;
1322         struct inode *inode;
1323         int rc, repsize[3] = {sizeof(struct mds_body),
1324                               obd->u.mds.mds_max_mdsize,
1325                               obd->u.mds.mds_max_cookiesize};
1326         ENTRY;
1327
1328         rc = lustre_pack_reply(req, 3, repsize, NULL);
1329         if (rc) {
1330                 CERROR("lustre_pack_reply: rc = %d\n", rc);
1331                 req->rq_status = rc;
1332         } else {
1333                 MDS_CHECK_RESENT(req, mds_reconstruct_generic(req));
1334         }
1335
1336         body = lustre_swab_reqbuf(req, 0, sizeof(*body), lustre_swab_mds_body);
1337         if (body == NULL) {
1338                 CERROR("Can't unpack body\n");
1339                 req->rq_status = -EFAULT;
1340                 RETURN(-EFAULT);
1341         }
1342
1343         if (body->flags & MDS_BFLAG_UNCOMMITTED_WRITES)
1344                 /* do some stuff */ ;
1345
1346         mfd = mds_handle2mfd(&body->handle);
1347         if (mfd == NULL) {
1348                 DEBUG_REQ(D_ERROR, req, "no handle for file close ino "LPD64
1349                           ": cookie "LPX64, body->fid1.id, body->handle.cookie);
1350                 req->rq_status = -ESTALE;
1351                 RETURN(-ESTALE);
1352         }
1353
1354         inode = mfd->mfd_dentry->d_inode;
1355         /* child orphan sem protects orphan_dec_test && is_orphan race */
1356         MDS_DOWN_WRITE_ORPHAN_SEM(inode); /* mds_mfd_close drops this */
1357         if (mds_inode_is_orphan(inode) && mds_orphan_open_count(inode) == 1) {
1358                 body = lustre_msg_buf(req->rq_repmsg, 0, sizeof (*body));
1359                 LASSERT(body != NULL);
1360
1361                 mds_pack_inode2fid(&body->fid1, inode);
1362                 mds_pack_inode2body(body, inode);
1363                 mds_pack_md(obd, req->rq_repmsg, 1,body,inode,MDS_PACK_MD_LOCK);
1364         }
1365         spin_lock(&med->med_open_lock);
1366         list_del(&mfd->mfd_list);
1367         spin_unlock(&med->med_open_lock);
1368
1369         push_ctxt(&saved, &obd->obd_ctxt, NULL);
1370         req->rq_status = mds_mfd_close(req, obd, mfd, 1);
1371         pop_ctxt(&saved, &obd->obd_ctxt, NULL);
1372
1373         mds_mfd_put(mfd);
1374         if (OBD_FAIL_CHECK(OBD_FAIL_MDS_CLOSE_PACK)) {
1375                 CERROR("test case OBD_FAIL_MDS_CLOSE_PACK\n");
1376                 req->rq_status = -ENOMEM;
1377                 RETURN(-ENOMEM);
1378         }
1379
1380         RETURN(rc);
1381 }
1382
1383 int mds_done_writing(struct ptlrpc_request *req)
1384 {
1385         struct mds_body *body;
1386         int rc, size = sizeof(struct mds_body);
1387         ENTRY;
1388
1389         MDS_CHECK_RESENT(req, mds_reconstruct_generic(req));
1390
1391         body = lustre_swab_reqbuf(req, 0, sizeof(*body), lustre_swab_mds_body);
1392         if (body == NULL) {
1393                 CERROR("Can't unpack body\n");
1394                 req->rq_status = -EFAULT;
1395                 RETURN(-EFAULT);
1396         }
1397
1398         rc = lustre_pack_reply(req, 1, &size, NULL);
1399         if (rc) {
1400                 CERROR("lustre_pack_reply: rc = %d\n", rc);
1401                 req->rq_status = rc;
1402         }
1403
1404         RETURN(0);
1405 }