mds_open.c

lustre 1.6.5 source code

                      FSFILT_OP_CREATE);
        RETURN(rc);
}

/* Close a "file descriptor" and possibly unlink an orphan from the
 * PENDING directory.  Caller must hold child->i_mutex, this drops it.
 *
 * If we are being called from mds_disconnect() because the client has
 * disappeared, then req == NULL and we do not update last_rcvd because
 * there is nothing that could be recovered by the client at this stage
 * (it will not even _have_ an entry in last_rcvd anymore).
 *
 * Returns EAGAIN if the client needs to get more data and re-close. */
int mds_mfd_close(struct ptlrpc_request *req, int offset,
                  struct obd_device *obd, struct mds_file_data *mfd,
                  int unlink_orphan, struct lov_mds_md *lmm, int lmm_size,
                  struct llog_cookie *logcookies, int cookies_size,
                  __u64 *valid)
{
        struct inode *inode = mfd->mfd_dentry->d_inode;
        char fidname[LL_FID_NAMELEN];
        int last_orphan, fidlen, rc = 0, cleanup_phase = 0;
        struct dentry *pending_child = NULL;
        struct mds_obd *mds = &obd->u.mds;
        struct inode *pending_dir = mds->mds_pending_dir->d_inode;
        void *handle = NULL;
        struct mds_body *request_body = NULL, *reply_body = NULL;
        struct lvfs_dentry_params dp = LVFS_DENTRY_PARAMS_INIT;
        struct iattr iattr = { 0 };
        ENTRY;

        if (req && req->rq_reqmsg != NULL)
                request_body = lustre_msg_buf(req->rq_reqmsg, offset,
                                              sizeof(*request_body));
        if (req && req->rq_repmsg != NULL)
                reply_body = lustre_msg_buf(req->rq_repmsg, REPLY_REC_OFF,
                                            sizeof(*reply_body));

        fidlen = ll_fid2str(fidname, inode->i_ino, inode->i_generation);

        CDEBUG(D_INODE, "inode %p ino %s nlink %d orphan %d\n", inode, fidname,
               inode->i_nlink, mds_orphan_open_count(inode));

        last_orphan = mds_orphan_open_dec_test(inode) &&
                mds_inode_is_orphan(inode);
        MDS_UP_WRITE_ORPHAN_SEM(inode);

        /* this is half of the actual "close" */
        if (mfd->mfd_mode & FMODE_WRITE) {
                rc = mds_put_write_access(mds, inode, request_body,
                                          last_orphan && unlink_orphan);
        } else if (mfd->mfd_mode & MDS_FMODE_EXEC) {
                mds_allow_write_access(inode);
        }

        if (last_orphan && unlink_orphan) {
                int stripe_count = 0;
                LASSERT(rc == 0); /* mds_put_write_access must have succeeded */

                CDEBUG(D_INODE, "destroying orphan object %s\n", fidname);

                if ((S_ISREG(inode->i_mode) && inode->i_nlink != 1) ||
                    (S_ISDIR(inode->i_mode) && inode->i_nlink != 2))
                        CERROR("found \"orphan\" %s %s with link count %d\n",
                               S_ISREG(inode->i_mode) ? "file" : "dir",
                               fidname, inode->i_nlink);

                /* Sadly, there is no easy way to save pending_child from
                 * mds_reint_unlink() into mfd, so we need to re-lookup,
                 * but normally it will still be in the dcache. */
                LOCK_INODE_MUTEX(pending_dir);
                cleanup_phase = 1; /* UNLOCK_INODE_MUTEX(pending_dir) when finished */
                pending_child = lookup_one_len(fidname, mds->mds_pending_dir,
                                               fidlen);
                if (IS_ERR(pending_child))
                        GOTO(cleanup, rc = PTR_ERR(pending_child));
                LASSERT(pending_child->d_inode != NULL);

                cleanup_phase = 2; /* dput(pending_child) when finished */
                if (S_ISDIR(pending_child->d_inode->i_mode)) {
                        rc = vfs_rmdir(pending_dir, pending_child);
                        if (rc)
                                CERROR("error unlinking orphan dir %s: rc %d\n",
                                       fidname,rc);
                        goto out;
                }

                if (lmm != NULL) {
                        stripe_count = le32_to_cpu(lmm->lmm_stripe_count);
                }

                handle = fsfilt_start_log(obd, pending_dir, FSFILT_OP_UNLINK,
                                          NULL, stripe_count);
                if (IS_ERR(handle)) {
                        rc = PTR_ERR(handle);
                        handle = NULL;
                        GOTO(cleanup, rc);
                }
                if (lmm != NULL && (*valid & OBD_MD_FLEASIZE) &&
                    mds_log_op_unlink(obd, lmm, lmm_size,
                                      logcookies, cookies_size) > 0) {
                        *valid |= OBD_MD_FLCOOKIE;
                }

                dp.ldp_inum = 0;
                dp.ldp_ptr = req;
                pending_child->d_fsdata = (void *) &dp;
                rc = vfs_unlink(pending_dir, pending_child);
                if (rc)
                        CERROR("error unlinking orphan %s: rc %d\n",fidname,rc);

                goto out; /* Don't bother updating attrs on unlinked inode */
        }

#if 0
        if (request_body != NULL && mfd->mfd_mode & FMODE_WRITE && rc == 0) {
                /* Update the on-disk attributes if this was the last write
                 * close, and all information was provided (i.e., rc == 0)
                 *
                 * XXX this should probably be abstracted with mds_reint_setattr
                 */

                if (request_body->valid & OBD_MD_FLMTIME &&
                    LTIME_S(iattr.ia_mtime) > LTIME_S(inode->i_mtime)) {
                        LTIME_S(iattr.ia_mtime) = request_body->mtime;
                        iattr.ia_valid |= ATTR_MTIME;
                }
                if (request_body->valid & OBD_MD_FLCTIME &&
                    LTIME_S(iattr.ia_ctime) > LTIME_S(inode->i_ctime)) {
                        LTIME_S(iattr.ia_ctime) = request_body->ctime;
                        iattr.ia_valid |= ATTR_CTIME;
                }

                /* XXX can't set block count with fsfilt_setattr (!) */
                if (request_body->valid & OBD_MD_FLSIZE) {
                        iattr.ia_valid |= ATTR_SIZE;
                        iattr.ia_size = request_body->size;
                }
                /* iattr.ia_blocks = request_body->blocks */

        }
#endif
        if (request_body != NULL && request_body->valid & OBD_MD_FLATIME) {
                /* Only start a transaction to write out only the atime if
                 * it is more out-of-date than the specified limit.  If we
                 * are already going to write out the atime then do it anyway.
                 * */
                LTIME_S(iattr.ia_atime) = request_body->atime;
                if ((LTIME_S(iattr.ia_atime) >
                     LTIME_S(inode->i_atime) + mds->mds_atime_diff) ||
                    (iattr.ia_valid != 0 &&
                     LTIME_S(iattr.ia_atime) > LTIME_S(inode->i_atime)))
                        iattr.ia_valid |= ATTR_ATIME;
        }

        if (iattr.ia_valid != 0) {
                handle = fsfilt_start(obd, inode, FSFILT_OP_SETATTR, NULL);
                if (IS_ERR(handle)) {
                        rc = PTR_ERR(handle);
                        handle = NULL;
                        GOTO(cleanup, rc);
                }
                rc = fsfilt_setattr(obd, mfd->mfd_dentry, handle, &iattr, 0);
                if (rc)
                        CERROR("error in setattr(%s): rc %d\n", fidname, rc);
        }
out:
        /* If other clients have this file open for write, rc will be > 0 */
        if (rc > 0)
                rc = 0;
        l_dput(mfd->mfd_dentry);
        mds_mfd_put(mfd);

 cleanup:
        if (req != NULL && reply_body != NULL) {
                rc = mds_finish_transno(mds, pending_dir, handle, req, rc, 0, 0);
        } else if (handle) {
                int err = fsfilt_commit(obd, pending_dir, handle, 0);
                if (err) {
                        CERROR("error committing close: %d\n", err);
                        if (!rc)
                                rc = err;
                }
        }

        switch (cleanup_phase) {
        case 2:
                dput(pending_child);
        case 1:
                UNLOCK_INODE_MUTEX(pending_dir);
        }
        RETURN(rc);
}

int mds_close(struct ptlrpc_request *req, int offset)
{
        struct mds_export_data *med = &req->rq_export->exp_mds_data;
        struct obd_device *obd = req->rq_export->exp_obd;
        struct mds_body *body;
        struct mds_file_data *mfd;
        struct lvfs_run_ctxt saved;
        struct inode *inode;
        int rc, repsize[4] = { sizeof(struct ptlrpc_body),
                               sizeof(struct mds_body),
                               obd->u.mds.mds_max_mdsize,
                               obd->u.mds.mds_max_cookiesize };
        struct mds_body *reply_body;
        struct lov_mds_md *lmm;
        int lmm_size;
        struct llog_cookie *logcookies;
        int cookies_size;
        ENTRY;

        rc = lustre_pack_reply(req, 4, repsize, NULL);
        if (rc)
                req->rq_status = rc;
                /* continue on to drop local open even if we can't send reply */
        else
                MDS_CHECK_RESENT(req, mds_reconstruct_generic(req));

        CDEBUG(D_INODE, "close req->rep_len %d mdsize %d cookiesize %d\n",
               req->rq_replen,
               obd->u.mds.mds_max_mdsize, obd->u.mds.mds_max_cookiesize);
        mds_counter_incr(req->rq_export, LPROC_MDS_CLOSE);

        body = lustre_swab_reqbuf(req, offset, sizeof(*body),
                                  lustre_swab_mds_body);
        if (body == NULL) {
                CERROR("Can't unpack body\n");
                req->rq_status = -EFAULT;
                RETURN(-EFAULT);
        }

        if (body->flags & MDS_BFLAG_UNCOMMITTED_WRITES)
                /* do some stuff */ ;

        spin_lock(&med->med_open_lock);
        mfd = mds_handle2mfd(&body->handle);
        if (mfd == NULL) {
                spin_unlock(&med->med_open_lock);
                DEBUG_REQ(D_ERROR, req, "no handle for file close ino "LPD64
                          ": cookie "LPX64, body->fid1.id, body->handle.cookie);
                req->rq_status = -ESTALE;
                RETURN(-ESTALE);
        }
        /* Remove mfd handle so it can't be found again.  We consume mfd_list
         * reference here, but still have mds_handle2mfd ref until mfd_close. */
        mds_mfd_unlink(mfd, 1);
        spin_unlock(&med->med_open_lock);

        inode = mfd->mfd_dentry->d_inode;
        /* child orphan sem protects orphan_dec_test && is_orphan race */
        MDS_DOWN_WRITE_ORPHAN_SEM(inode); /* mds_mfd_close drops this */
        if (mds_inode_is_orphan(inode) && mds_orphan_open_count(inode) == 1) {
                body = lustre_msg_buf(req->rq_repmsg, REPLY_REC_OFF,
                                      sizeof(*body));
                LASSERT(body != NULL);

                mds_pack_inode2fid(&body->fid1, inode);
                mds_pack_inode2body(body, inode);
                mds_pack_md(obd, req->rq_repmsg, REPLY_REC_OFF + 1, body, inode,
                            MDS_PACK_MD_LOCK, 0);
        }

        push_ctxt(&saved, &obd->obd_lvfs_ctxt, NULL);
        reply_body = lustre_msg_buf(req->rq_repmsg, REPLY_REC_OFF, sizeof(*reply_body));
        lmm = lustre_msg_buf(req->rq_repmsg, REPLY_REC_OFF + 1, 0);
        lmm_size = lustre_msg_buflen(req->rq_repmsg, REPLY_REC_OFF + 1),
        logcookies = lustre_msg_buf(req->rq_repmsg, REPLY_REC_OFF + 2, 0);
        cookies_size = lustre_msg_buflen(req->rq_repmsg, REPLY_REC_OFF + 2);
        req->rq_status = mds_mfd_close(req, offset, obd, mfd, 1,
                                       lmm, lmm_size, logcookies, cookies_size,
                                       &reply_body->valid);
        pop_ctxt(&saved, &obd->obd_lvfs_ctxt, NULL);

        mds_shrink_reply(obd, req, body, REPLY_REC_OFF + 1);
        if (OBD_FAIL_CHECK(OBD_FAIL_MDS_CLOSE_PACK)) {
                CERROR("test case OBD_FAIL_MDS_CLOSE_PACK\n");
                req->rq_status = -ENOMEM;
                RETURN(-ENOMEM);
        }

        RETURN(rc);
}

int mds_done_writing(struct ptlrpc_request *req, int offset)
{
        struct mds_body *body;
        int rc, size[2] = { sizeof(struct ptlrpc_body),
                            sizeof(struct mds_body) };
        ENTRY;

        MDS_CHECK_RESENT(req, mds_reconstruct_generic(req));

        body = lustre_swab_reqbuf(req, offset, sizeof(*body),
                                  lustre_swab_mds_body);
        if (body == NULL) {
                CERROR("Can't unpack body\n");
                req->rq_status = -EFAULT;
                RETURN(-EFAULT);
        }

        rc = lustre_pack_reply(req, 2, size, NULL);
        if (rc)
                req->rq_status = rc;

        RETURN(0);
}