mds_open.c

lustre 1.6.5 source code

                if (IS_ERR(mfd)) {
                        req->rq_status = PTR_ERR(mfd);
                        mfd = NULL;
                        CERROR("%s: opening inode "LPU64" failed: rc %d\n",
                               req->rq_export->exp_obd->obd_name,
                               (__u64)dchild->d_inode->i_ino, req->rq_status);
                        GOTO(out_dput, req->rq_status);
                }
        } else {
                body->handle.cookie = mfd->mfd_handle.h_cookie;
                CDEBUG(D_INODE, "resend mfd %p, cookie "LPX64"\n", mfd,
                       mfd->mfd_handle.h_cookie);
        }

        mds_mfd_put(mfd);

 out_dput:
        if (put_child)
                l_dput(dchild);
        l_dput(parent);
        EXIT;
}

/* do NOT or the MAY_*'s, you'll get the weakest */
static int accmode(struct inode *inode, int flags)
{
        int res = 0;

        /* Sadly, NFSD reopens a file repeatedly during operation, so the
         * "acc_mode = 0" allowance for newly-created files isn't honoured.
         * NFSD uses the MDS_OPEN_OWNEROVERRIDE flag to say that a file
         * owner can write to a file even if it is marked readonly to hide
         * its brokenness. (bug 5781) */
        if (flags & MDS_OPEN_OWNEROVERRIDE && inode->i_uid == current->fsuid)
                return 0;

        if (flags & FMODE_READ)
                res = MAY_READ;
        if (flags & (FMODE_WRITE|MDS_OPEN_TRUNC))
                res |= MAY_WRITE;
        if (flags & MDS_FMODE_EXEC)
                res = MAY_EXEC;
        return res;
}

/* Handles object creation, actual opening, and I/O epoch */
static int mds_finish_open(struct ptlrpc_request *req, struct dentry *dchild,
                           struct mds_body *body, int flags, void **handle,
                           struct mds_update_record *rec,struct ldlm_reply *rep,
                           struct lustre_handle *lockh)
{
        struct mds_obd *mds = mds_req2mds(req);
        struct obd_device *obd = req->rq_export->exp_obd;
        struct mds_file_data *mfd = NULL;
        struct lov_mds_md *lmm = NULL; /* object IDs created */
        int rc = 0;
        ENTRY;

        /* atomically create objects if necessary */
        LOCK_INODE_MUTEX(dchild->d_inode);

        if (S_ISREG(dchild->d_inode->i_mode) &&
            !(body->valid & OBD_MD_FLEASIZE)) {
                rc = mds_pack_md(obd, req->rq_repmsg, DLM_REPLY_REC_OFF + 1,
                                 body, dchild->d_inode, 0, 0);
                if (rc) {
                        UNLOCK_INODE_MUTEX(dchild->d_inode);
                        RETURN(rc);
                }
        }
        if (rec != NULL) {
                if ((body->valid & OBD_MD_FLEASIZE) &&
                    (rec->ur_flags & MDS_OPEN_HAS_EA)) {
                        UNLOCK_INODE_MUTEX(dchild->d_inode);
                        RETURN(-EEXIST);
                }
                if (rec->ur_flags & MDS_OPEN_JOIN_FILE) {
                        UNLOCK_INODE_MUTEX(dchild->d_inode);
                        rc = mds_join_file(rec, req, dchild, lockh);
                        if (rc)
                                RETURN(rc);
                        LOCK_INODE_MUTEX(dchild->d_inode);
                }
                if (!(body->valid & OBD_MD_FLEASIZE) &&
                    !(body->valid & OBD_MD_FLMODEASIZE)) {
                        /* no EA: create objects */
                        rc = mds_create_objects(req, DLM_REPLY_REC_OFF + 1, rec,
                                                mds, obd, dchild, handle, &lmm);
                        if (rc) {
                                CERROR("mds_create_objects: rc = %d\n", rc);
                                UNLOCK_INODE_MUTEX(dchild->d_inode);
                                RETURN(rc);
                        }
                }
        }
        /* If the inode has no EA data, then MDS holds size, mtime */
        if (S_ISREG(dchild->d_inode->i_mode) &&
            !(body->valid & OBD_MD_FLEASIZE)) {
                body->valid |= (OBD_MD_FLSIZE | OBD_MD_FLBLOCKS |
                                OBD_MD_FLATIME | OBD_MD_FLMTIME);
        }
        UNLOCK_INODE_MUTEX(dchild->d_inode);

        if (rec && !(rec->ur_flags & MDS_OPEN_JOIN_FILE))
                lustre_shrink_reply(req, DLM_REPLY_REC_OFF + 1,
                                    body->eadatasize, 0);

        if (req->rq_export->exp_connect_flags & OBD_CONNECT_ACL &&
            rec && !(rec->ur_flags & MDS_OPEN_JOIN_FILE)) {
                int acl_off = DLM_REPLY_REC_OFF + (body->eadatasize ? 2 : 1);

                rc = mds_pack_acl(&req->rq_export->exp_mds_data,
                                  dchild->d_inode, req->rq_repmsg,
                                  body, acl_off);
                lustre_shrink_reply(req, acl_off, body->aclsize, 0);
                if (rc)
                        RETURN(rc);
        }

        intent_set_disposition(rep, DISP_OPEN_OPEN);
        mfd = mds_dentry_open(dchild, mds->mds_vfsmnt, flags, req);
        if (IS_ERR(mfd))
                RETURN(PTR_ERR(mfd));

        CDEBUG(D_INODE, "mfd %p, cookie "LPX64"\n", mfd,
               mfd->mfd_handle.h_cookie);

        mds_lov_update_objids(obd, lmm);

        if (rc) /* coverity[deadcode] */
                mds_mfd_unlink(mfd, 1);

        mds_mfd_put(mfd);
        RETURN(rc);
}

static int mds_open_by_fid(struct ptlrpc_request *req, struct ll_fid *fid,
                           struct mds_body *body, int flags,
                           struct mds_update_record *rec,struct ldlm_reply *rep)
{
        struct mds_obd *mds = mds_req2mds(req);
        struct dentry *dchild;
        char fidname[LL_FID_NAMELEN];
        int fidlen = 0, rc;
        void *handle = NULL;
        ENTRY;

        fidlen = ll_fid2str(fidname, fid->id, fid->generation);
        dchild = ll_lookup_one_len(fidname, mds->mds_pending_dir, fidlen);
        if (IS_ERR(dchild)) {
                rc = PTR_ERR(dchild);
                CERROR("error looking up %s in PENDING: rc = %d\n",fidname, rc);
                RETURN(rc);
        }

        if (dchild->d_inode != NULL) {
                mds_inode_set_orphan(dchild->d_inode);
                CWARN("Orphan %s found and opened in PENDING directory\n",
                       fidname);
        } else {
                l_dput(dchild);

                /* We didn't find it in PENDING so it isn't an orphan.  See
                 * if it was a regular inode that was previously created. */
                dchild = mds_fid2dentry(mds, fid, NULL);
                if (IS_ERR(dchild))
                        RETURN(PTR_ERR(dchild));
        }

        mds_pack_inode2fid(&body->fid1, dchild->d_inode);
        mds_pack_inode2body(body, dchild->d_inode);
        intent_set_disposition(rep, DISP_LOOKUP_EXECD);
        intent_set_disposition(rep, DISP_LOOKUP_POS);

        rc = mds_finish_open(req, dchild, body, flags, &handle, rec, rep, NULL);
        rc = mds_finish_transno(mds, dchild->d_inode, handle,
                                req, rc, rep ? rep->lock_policy_res1 : 0, 0);
        /* XXX what do we do here if mds_finish_transno itself failed? */

        l_dput(dchild);
        RETURN(rc);
}

int mds_pin(struct ptlrpc_request *req, int offset)
{
        struct obd_device *obd = req->rq_export->exp_obd;
        struct mds_body *reqbody, *repbody;
        struct lvfs_run_ctxt saved;
        int rc, size[2] = { sizeof(struct ptlrpc_body), sizeof(*repbody) };
        ENTRY;

        reqbody = lustre_msg_buf(req->rq_reqmsg, offset, sizeof(*reqbody));

        rc = lustre_pack_reply(req, 2, size, NULL);
        if (rc)
                RETURN(rc);

        repbody = lustre_msg_buf(req->rq_repmsg, REPLY_REC_OFF,
                                 sizeof(*repbody));

        push_ctxt(&saved, &obd->obd_lvfs_ctxt, NULL);
        rc = mds_open_by_fid(req, &reqbody->fid1, repbody, reqbody->flags, NULL,
                             NULL);
        pop_ctxt(&saved, &obd->obd_lvfs_ctxt, NULL);

        RETURN(rc);
}

/*  Get an internal lock on the inode number (but not generation) to sync
 *  new inode creation with inode unlink (bug 2029).  If child_lockh is NULL
 *  we just get the lock as a barrier to wait for other holders of this lock,
 *  and drop it right away again. */
int mds_lock_new_child(struct obd_device *obd, struct inode *inode,
                       struct lustre_handle *child_lockh)
{
        struct ldlm_res_id child_res_id = { .name = { inode->i_ino, 0, 1, 0 } };
        struct lustre_handle lockh;
        int lock_flags = LDLM_FL_ATOMIC_CB;
        int rc;

        if (child_lockh == NULL)
                child_lockh = &lockh;

        rc = ldlm_cli_enqueue_local(obd->obd_namespace, &child_res_id,
                                    LDLM_PLAIN, NULL, LCK_EX, &lock_flags,
                                    ldlm_blocking_ast, ldlm_completion_ast,
                                    NULL, NULL, 0, NULL, child_lockh);
        if (rc != ELDLM_OK)
                CERROR("ldlm_cli_enqueue_local: %d\n", rc);
        else if (child_lockh == &lockh)
                ldlm_lock_decref(child_lockh, LCK_EX);

        RETURN(rc);
}

int mds_open(struct mds_update_record *rec, int offset,
             struct ptlrpc_request *req, struct lustre_handle *child_lockh)
{
        /* XXX ALLOCATE _something_ - 464 bytes on stack here */
        struct obd_device *obd = req->rq_export->exp_obd;
        struct mds_obd *mds = mds_req2mds(req);
        struct ldlm_reply *rep = NULL;
        struct mds_body *body = NULL;
        struct dentry *dchild = NULL, *dparent = NULL;
        struct mds_export_data *med;
        struct lustre_handle parent_lockh;
        int rc = 0, cleanup_phase = 0, acc_mode, created = 0;
        int parent_mode = LCK_CR;
        void *handle = NULL;
        struct lvfs_dentry_params dp = LVFS_DENTRY_PARAMS_INIT;
        unsigned int qcids[MAXQUOTAS] = { current->fsuid, current->fsgid };
        unsigned int qpids[MAXQUOTAS] = { 0, 0 };
        int child_mode = LCK_CR;
        /* Always returning LOOKUP lock if open succesful to guard
           dentry on client. */
        ldlm_policy_data_t policy = {.l_inodebits={MDS_INODELOCK_LOOKUP}};
        struct ldlm_res_id child_res_id = { .name = {0}};
        int lock_flags = 0;
        int rec_pending = 0;
        unsigned int gid = current->fsgid;
        ENTRY;

        mds_counter_incr(req->rq_export, LPROC_MDS_OPEN);

        OBD_FAIL_TIMEOUT(OBD_FAIL_MDS_PAUSE_OPEN | OBD_FAIL_ONCE,
                         (obd_timeout + 1) / 4);

        CLASSERT(MAXQUOTAS < 4);
        if (offset == DLM_INTENT_REC_OFF) { /* intent */
                rep = lustre_msg_buf(req->rq_repmsg, DLM_LOCKREPLY_OFF,
                                     sizeof(*rep));
                body = lustre_msg_buf(req->rq_repmsg, DLM_REPLY_REC_OFF,
                                      sizeof(*body));
        } else if (offset == REQ_REC_OFF) { /* non-intent reint */
                body = lustre_msg_buf(req->rq_repmsg, REPLY_REC_OFF,
                                      sizeof(*body));
                LBUG(); /* XXX: not supported yet? */
        } else {
                body = NULL;
                LBUG();
        }

        MDS_CHECK_RESENT(req, reconstruct_open(rec, offset, req, child_lockh));

        /* Step 0: If we are passed a fid, then we assume the client already
         * opened this file and is only replaying the RPC, so we open the
         * inode by fid (at some large expense in security). */
        /*XXX liblustre use mds_open_by_fid to implement LL_IOC_LOV_SETSTRIPE */
        if (((lustre_msg_get_flags(req->rq_reqmsg) & MSG_REPLAY) ||
             (req->rq_export->exp_libclient && rec->ur_flags&MDS_OPEN_HAS_EA))&&
            !(rec->ur_flags & MDS_OPEN_JOIN_FILE)) {
                if (rec->ur_fid2->id == 0) {
                        struct ldlm_lock *lock = ldlm_handle2lock(child_lockh);
                        if (lock) {
                                LDLM_ERROR(lock, "fid2 not set on open replay");
                                LDLM_LOCK_PUT(lock);
                        }
                        DEBUG_REQ(D_ERROR, req, "fid2 not set on open replay");
                        RETURN(-EFAULT);
                }

                rc = mds_open_by_fid(req, rec->ur_fid2, body, rec->ur_flags,
                                     rec, rep);
                if (rc != -ENOENT) {
                        if (req->rq_export->exp_libclient &&
                            rec->ur_flags & MDS_OPEN_HAS_EA)
                                RETURN(0);

                        RETURN(rc);
                }