mds_open.c

lustre 1.6.5 source code

                /* We didn't find the correct inode on disk either, so we
                 * need to re-create it via a regular replay. */
                if (!(rec->ur_flags & MDS_OPEN_CREAT)) {
                        DEBUG_REQ(D_ERROR, req,"OPEN_CREAT not in open replay");
                        RETURN(-EFAULT);
                }
        } else if (rec->ur_fid2->id) {
                DEBUG_REQ(D_ERROR, req, "fid2 "LPU64"/%u on open non-replay",
                          rec->ur_fid2->id, rec->ur_fid2->generation);
                RETURN(-EFAULT);
        }

        /* If we got here, we must be called via intent */
        LASSERT(offset == DLM_INTENT_REC_OFF);

        med = &req->rq_export->exp_mds_data;
        if (OBD_FAIL_CHECK(OBD_FAIL_MDS_OPEN_PACK)) {
                CERROR("test case OBD_FAIL_MDS_OPEN_PACK\n");
                RETURN(-ENOMEM);
        }

        /* Step 1: Find and lock the parent */
        if (rec->ur_flags & (MDS_OPEN_CREAT | MDS_OPEN_JOIN_FILE))
                parent_mode = LCK_EX;
        dparent = mds_fid2locked_dentry(obd, rec->ur_fid1, NULL, parent_mode,
                                        &parent_lockh, rec->ur_name,
                                        rec->ur_namelen - 1,
                                        MDS_INODELOCK_UPDATE);
        if (IS_ERR(dparent)) {
                rc = PTR_ERR(dparent);
                if (rc != -ENOENT) {
                        CERROR("parent "LPU64"/%u lookup error %d\n",
                               rec->ur_fid1->id, rec->ur_fid1->generation, rc);
                } else {
                        /* Just cannot find parent - make it look like
                         * usual negative lookup to avoid extra MDS RPC */
                        intent_set_disposition(rep, DISP_LOOKUP_EXECD);
                        intent_set_disposition(rep, DISP_LOOKUP_NEG);
                }
                GOTO(cleanup, rc);
        }
        LASSERT(dparent->d_inode != NULL);

        cleanup_phase = 1; /* parent dentry and lock */

        if (rec->ur_flags & MDS_OPEN_JOIN_FILE) {
                dchild = dget(dparent);
                cleanup_phase = 2; /* child dentry */
                acc_mode = accmode(dchild->d_inode, rec->ur_flags);
                GOTO(found_child, rc);
        }

        /* Step 2: Lookup the child */
      
        if (!(lustre_msg_get_flags(req->rq_reqmsg) & MSG_REPLAY) &&
            (rec->ur_flags & MDS_OPEN_LOCK) && (rec->ur_namelen == 1)) {
                /* hack for nfsd with no_subtree_check, it will use anon
                 * dentry w/o filename to open the file. the anon dentry's
                 * parent was set to itself, so rec->ur_fid1 is the file.
                 * And in MDC it cannot derive the dentry's parent dentry,
                 * hence the file's name, so we hack here in MDS, 
                 * refer to bug 13030. */
                dchild = mds_fid2dentry(mds, rec->ur_fid1, NULL);
        } else {
                dchild = ll_lookup_one_len(rec->ur_name, dparent,
                                           rec->ur_namelen - 1);
        }
        if (IS_ERR(dchild)) {
                rc = PTR_ERR(dchild);
                dchild = NULL; /* don't confuse mds_finish_transno() below */
                GOTO(cleanup, rc);
        }

        cleanup_phase = 2; /* child dentry */

        intent_set_disposition(rep, DISP_LOOKUP_EXECD);
        if (dchild->d_inode)
                intent_set_disposition(rep, DISP_LOOKUP_POS);
        else
                intent_set_disposition(rep, DISP_LOOKUP_NEG);

        /*Step 3: If the child was negative, and we're supposed to, create it.*/
        if (dchild->d_inode == NULL) {
                unsigned long ino = rec->ur_fid2->id;
                struct iattr iattr;
                struct inode *inode;

                if (!(rec->ur_flags & MDS_OPEN_CREAT)) {
                        /* It's negative and we weren't supposed to create it */
                        GOTO(cleanup, rc = -ENOENT);
                }

                if (req->rq_export->exp_connect_flags & OBD_CONNECT_RDONLY)
                        GOTO(cleanup, rc = -EROFS);

                if (dparent->d_inode->i_mode & S_ISGID)
                        gid = dparent->d_inode->i_gid;
                else
                        gid = current->fsgid;

                /* we try to get enough quota to write here, and let ldiskfs
                 * decide if it is out of quota or not b=14783 */
                lquota_chkquota(mds_quota_interface_ref, obd,
                                current->fsuid, gid, 1, &rec_pending);

                intent_set_disposition(rep, DISP_OPEN_CREATE);
                handle = fsfilt_start(obd, dparent->d_inode, FSFILT_OP_CREATE,
                                      NULL);
                if (IS_ERR(handle)) {
                        rc = PTR_ERR(handle);
                        handle = NULL;
                        GOTO(cleanup, rc);
                }
                dchild->d_fsdata = (void *) &dp;
                dp.ldp_ptr = req;
                dp.ldp_inum = ino;

                rc = ll_vfs_create(dparent->d_inode, dchild, rec->ur_mode,NULL);
                if (dchild->d_fsdata == (void *)(unsigned long)ino)
                        dchild->d_fsdata = NULL;

                if (rc) {
                        CDEBUG(D_INODE, "error during create: %d\n", rc);
                        GOTO(cleanup, rc);
                }
                inode = dchild->d_inode;
                if (ino) {
                        LASSERT(ino == inode->i_ino);
                        /* Written as part of setattr */
                        inode->i_generation = rec->ur_fid2->generation;
                        CDEBUG(D_HA, "recreated ino %lu with gen %u\n",
                               inode->i_ino, inode->i_generation);
                }

                created = 1;
                LTIME_S(iattr.ia_atime) = rec->ur_time;
                LTIME_S(iattr.ia_ctime) = rec->ur_time;
                LTIME_S(iattr.ia_mtime) = rec->ur_time;

                iattr.ia_uid = current->fsuid;  /* set by push_ctxt already */
                iattr.ia_gid = gid;

                iattr.ia_valid = ATTR_UID | ATTR_GID | ATTR_ATIME |
                        ATTR_MTIME | ATTR_CTIME;

                rc = fsfilt_setattr(obd, dchild, handle, &iattr, 0);
                if (rc)
                        CERROR("error on child setattr: rc = %d\n", rc);

                iattr.ia_valid = ATTR_MTIME | ATTR_CTIME;

                rc = fsfilt_setattr(obd, dparent, handle, &iattr, 0);
                if (rc)
                        CERROR("error on parent setattr: rc = %d\n", rc);

                rc = fsfilt_commit(obd, dchild->d_inode, handle, 0);
                handle = NULL;
                acc_mode = 0;           /* Don't check for permissions */
        } else {
                acc_mode = accmode(dchild->d_inode, rec->ur_flags);
        }

        LASSERTF(!mds_inode_is_orphan(dchild->d_inode),
                 "dchild %.*s (%p) inode %p/%lu/%u\n", dchild->d_name.len,
                 dchild->d_name.name, dchild, dchild->d_inode,
                 dchild->d_inode->i_ino, dchild->d_inode->i_generation);

found_child:
        mds_pack_inode2fid(&body->fid1, dchild->d_inode);
        mds_pack_inode2body(body, dchild->d_inode);

        if (S_ISREG(dchild->d_inode->i_mode)) {
                /* Check permissions etc */
                rc = ll_permission(dchild->d_inode, acc_mode, NULL);
                if (rc != 0)
                        GOTO(cleanup, rc);

                if ((req->rq_export->exp_connect_flags & OBD_CONNECT_RDONLY) &&
                    (acc_mode & MAY_WRITE))
                        GOTO(cleanup, rc = -EROFS);

                /* An append-only file must be opened in append mode for
                 * writing */
                if (IS_APPEND(dchild->d_inode) && (acc_mode & MAY_WRITE) != 0 &&
                    ((rec->ur_flags & MDS_OPEN_APPEND) == 0 ||
                     (rec->ur_flags & MDS_OPEN_TRUNC) != 0))
                        GOTO(cleanup, rc = -EPERM);
        }

        if (!created && (rec->ur_flags & MDS_OPEN_CREAT) &&
            (rec->ur_flags & MDS_OPEN_EXCL)) {
                /* File already exists, we didn't just create it, and we
                 * were passed O_EXCL; err-or. */
                GOTO(cleanup, rc = -EEXIST); // returns a lock to the client
        }

        /* if we are following a symlink, don't open */
        if (S_ISLNK(dchild->d_inode->i_mode))
                GOTO(cleanup_no_trans, rc = 0);

        if (S_ISDIR(dchild->d_inode->i_mode)) {
                if (rec->ur_flags & MDS_OPEN_CREAT ||
                    rec->ur_flags & FMODE_WRITE) {
                        /* we are trying to create or write a exist dir */
                        GOTO(cleanup, rc = -EISDIR);
                }
                if (rec->ur_flags & MDS_FMODE_EXEC) {
                        /* we are trying to exec a directory */
                        GOTO(cleanup, rc = -EACCES);
                }
                if (ll_permission(dchild->d_inode, acc_mode, NULL)) {
                        intent_set_disposition(rep, DISP_OPEN_OPEN);
                        GOTO(cleanup, rc = -EACCES);
                }
        } else if (rec->ur_flags & MDS_OPEN_DIRECTORY) {
                GOTO(cleanup, rc = -ENOTDIR);
        }

        if (OBD_FAIL_CHECK(OBD_FAIL_MDS_OPEN_CREATE)) {
                obd_fail_loc = OBD_FAIL_LDLM_REPLY | OBD_FAIL_ONCE;
                GOTO(cleanup, rc = -EAGAIN);
        }

        /* Obtain OPEN lock as well */
        policy.l_inodebits.bits |= MDS_INODELOCK_OPEN;

        /* We cannot use acc_mode here, because it is zeroed in case of
           creating a file, so we get wrong lockmode */
        if (rec->ur_flags & FMODE_WRITE)
                child_mode = LCK_CW;
        else if (rec->ur_flags & MDS_FMODE_EXEC)
                child_mode = LCK_PR;
        else
                child_mode = LCK_CR;

        if (!(lustre_msg_get_flags(req->rq_reqmsg) & MSG_REPLAY) && 
             (rec->ur_flags & MDS_OPEN_LOCK)) {
                /* In case of replay we do not get a lock assuming that the
                   caller has it already */
                child_res_id.name[0] = dchild->d_inode->i_ino;
                child_res_id.name[1] = dchild->d_inode->i_generation;

                rc = ldlm_cli_enqueue_local(obd->obd_namespace, &child_res_id,
                                            LDLM_IBITS, &policy, child_mode, 
                                            &lock_flags, ldlm_blocking_ast, 
                                            ldlm_completion_ast, NULL, NULL,
                                            0, NULL, child_lockh);
                if (rc != ELDLM_OK)
                        GOTO(cleanup, rc);

                /* Let mds_intent_policy know that we have a lock to return */
                intent_set_disposition(rep, DISP_OPEN_LOCK);
                cleanup_phase = 3;
        }

        if (!S_ISREG(dchild->d_inode->i_mode) &&
            !S_ISDIR(dchild->d_inode->i_mode) &&
            (req->rq_export->exp_connect_flags & OBD_CONNECT_NODEVOH)) {
                /* If client supports this, do not return open handle for
                 * special device nodes */
                GOTO(cleanup_no_trans, rc = 0);
        }

        /* Step 5: mds_open it */
        rc = mds_finish_open(req, dchild, body, rec->ur_flags, &handle, rec,
                             rep, &parent_lockh);
        GOTO(cleanup, rc);

 cleanup:
        rc = mds_finish_transno(mds, dchild ? dchild->d_inode : NULL, handle,
                                req, rc, rep ? rep->lock_policy_res1 : 0, 0);

 cleanup_no_trans:
        if (rec_pending)
                lquota_pending_commit(mds_quota_interface_ref, obd,
                                      current->fsuid, gid, 1);
        switch (cleanup_phase) {
        case 3:
                if (rc)
                        /* It is safe to leave IT_OPEN_LOCK set, if rc is not 0,
                         * mds_intent_policy won't try to return any locks */
                        ldlm_lock_decref(child_lockh, child_mode);
        case 2:
                if (rc && created) {
                        int err = vfs_unlink(dparent->d_inode, dchild);
                        if (err) {
                                CERROR("unlink(%.*s) in error path: %d\n",
                                       dchild->d_name.len, dchild->d_name.name,
                                       err);
                        }
                } else if (created) {
                        mds_lock_new_child(obd, dchild->d_inode, NULL);
                        /* save uid/gid for quota acquire/release */
                        qpids[USRQUOTA] = dparent->d_inode->i_uid;
                        qpids[GRPQUOTA] = dparent->d_inode->i_gid;
                }
                l_dput(dchild);
        case 1:
                if (dparent == NULL)
                        break;

                l_dput(dparent);
                if (rc)
                        ldlm_lock_decref(&parent_lockh, parent_mode);
                else
                        ptlrpc_save_lock(req, &parent_lockh, parent_mode);
        }
        /* trigger dqacq on the owner of child and parent */
        lquota_adjust(mds_quota_interface_ref, obd, qcids, qpids, rc,