mdc_locks.c

lustre 1.6.5 source code

        /* As not all attributes are kept under update lock, e.g. 
           owner/group/acls are under lookup lock, we need both 
           ibits for GETATTR. */
        policy.l_inodebits.bits = (it->it_op == IT_GETATTR) ?
                MDS_INODELOCK_UPDATE | MDS_INODELOCK_LOOKUP :
                MDS_INODELOCK_LOOKUP;

        mode = ldlm_lock_match(exp->exp_obd->obd_namespace,
                               LDLM_FL_BLOCK_GRANTED, &res_id, LDLM_IBITS,
                               &policy, LCK_CR|LCK_CW|LCK_PR|LCK_PW, &lockh);
        if (mode) {
                memcpy(&it->d.lustre.it_lock_handle, &lockh, sizeof(lockh));
                it->d.lustre.it_lock_mode = mode;
        }

        return !!mode;
}
EXPORT_SYMBOL(mdc_revalidate_lock);

static int mdc_finish_intent_lock(struct obd_export *exp,
                                  struct ptlrpc_request *req,
                                  struct mdc_op_data *data,
                                  struct lookup_intent *it,
                                  struct lustre_handle *lockh)
{
        struct mds_body *mds_body;
        struct lustre_handle old_lock;
        struct ldlm_lock *lock;
        int rc;
        ENTRY;

        LASSERT(req != NULL);
        LASSERT(req != LP_POISON);
        LASSERT(req->rq_repmsg != LP_POISON);

        if (!it_disposition(it, DISP_IT_EXECD)) {
                /* The server failed before it even started executing the
                 * intent, i.e. because it couldn't unpack the request. */
                LASSERT(it->d.lustre.it_status != 0);
                RETURN(it->d.lustre.it_status);
        }
        rc = it_open_error(DISP_IT_EXECD, it);
        if (rc)
                RETURN(rc);

        mds_body = lustre_msg_buf(req->rq_repmsg, DLM_REPLY_REC_OFF,
                                  sizeof(*mds_body));
        /* mdc_enqueue checked */
        LASSERT(mds_body != NULL);
        /* mdc_enqueue swabbed */
        LASSERT(lustre_rep_swabbed(req, DLM_REPLY_REC_OFF));

        /* If we were revalidating a fid/name pair, mark the intent in
         * case we fail and get called again from lookup */
        if (data->fid2.id && (it->it_op != IT_GETATTR)) {
                it_set_disposition(it, DISP_ENQ_COMPLETE);
                /* Also: did we find the same inode? */
                if (memcmp(&data->fid2, &mds_body->fid1, sizeof(data->fid2))) 
                        RETURN(-ESTALE);
        }

        rc = it_open_error(DISP_LOOKUP_EXECD, it);
        if (rc)
                RETURN(rc);

        /* keep requests around for the multiple phases of the call
         * this shows the DISP_XX must guarantee we make it into the call
         */
        if (!it_disposition(it, DISP_ENQ_CREATE_REF) &&
            it_disposition(it, DISP_OPEN_CREATE) &&
            !it_open_error(DISP_OPEN_CREATE, it)) {
                it_set_disposition(it, DISP_ENQ_CREATE_REF);
                ptlrpc_request_addref(req); /* balanced in ll_create_node */
        }
        if (!it_disposition(it, DISP_ENQ_OPEN_REF) &&
            it_disposition(it, DISP_OPEN_OPEN) &&
            !it_open_error(DISP_OPEN_OPEN, it)) {
                it_set_disposition(it, DISP_ENQ_OPEN_REF);
                ptlrpc_request_addref(req); /* balanced in ll_file_open */
                /* BUG 11546 - eviction in the middle of open rpc processing */
                OBD_FAIL_TIMEOUT(OBD_FAIL_MDC_ENQUEUE_PAUSE, obd_timeout);
        }

        if (it->it_op & IT_CREAT) {
                /* XXX this belongs in ll_create_it */
        } else if (it->it_op == IT_OPEN) {
                LASSERT(!it_disposition(it, DISP_OPEN_CREATE));
        } else {
                LASSERT(it->it_op & (IT_GETATTR | IT_LOOKUP));
        }

        /* If we already have a matching lock, then cancel the new
         * one.  We have to set the data here instead of in
         * mdc_enqueue, because we need to use the child's inode as
         * the l_ast_data to match, and that's not available until
         * intent_finish has performed the iget().) */
        lock = ldlm_handle2lock(lockh);
        if (lock) {
                ldlm_policy_data_t policy = lock->l_policy_data;

                LDLM_DEBUG(lock, "matching against this");
                LDLM_LOCK_PUT(lock);
                memcpy(&old_lock, lockh, sizeof(*lockh));
                if (ldlm_lock_match(NULL, LDLM_FL_BLOCK_GRANTED, NULL,
                                    LDLM_IBITS, &policy, LCK_NL, &old_lock)) {
                        ldlm_lock_decref_and_cancel(lockh,
                                                    it->d.lustre.it_lock_mode);
                        memcpy(lockh, &old_lock, sizeof(old_lock));
                        memcpy(&it->d.lustre.it_lock_handle, lockh,
                               sizeof(*lockh));
                }
        }

        CDEBUG(D_DENTRY,"D_IT dentry %.*s intent: %s status %d disp %x rc %d\n",
               data->namelen, data->name, ldlm_it2str(it->it_op),
               it->d.lustre.it_status, it->d.lustre.it_disposition, rc);
        RETURN(rc);
}

/* 
 * This long block is all about fixing up the lock and request state
 * so that it is correct as of the moment _before_ the operation was
 * applied; that way, the VFS will think that everything is normal and
 * call Lustre's regular VFS methods.
 *
 * If we're performing a creation, that means that unless the creation
 * failed with EEXIST, we should fake up a negative dentry.
 *
 * For everything else, we want to lookup to succeed.
 *
 * One additional note: if CREATE or OPEN succeeded, we add an extra
 * reference to the request because we need to keep it around until
 * ll_create/ll_open gets called.
 *
 * The server will return to us, in it_disposition, an indication of
 * exactly what d.lustre.it_status refers to.
 *
 * If DISP_OPEN_OPEN is set, then d.lustre.it_status refers to the open() call,
 * otherwise if DISP_OPEN_CREATE is set, then it status is the
 * creation failure mode.  In either case, one of DISP_LOOKUP_NEG or
 * DISP_LOOKUP_POS will be set, indicating whether the child lookup
 * was successful.
 *
 * Else, if DISP_LOOKUP_EXECD then d.lustre.it_status is the rc of the
 * child lookup.
 */
int mdc_intent_lock(struct obd_export *exp, struct mdc_op_data *op_data,
                    void *lmm, int lmmsize, struct lookup_intent *it,
                    int lookup_flags, struct ptlrpc_request **reqp,
                    ldlm_blocking_callback cb_blocking, int extra_lock_flags)
{
        struct lustre_handle lockh;
        int rc;
        ENTRY;

        LASSERT(it);

        CDEBUG(D_DLMTRACE,"name: %.*s in inode "LPU64", intent: %s flags %#o\n",
               op_data->namelen, op_data->name, op_data->fid1.id,
               ldlm_it2str(it->it_op), it->it_flags);

        if (op_data->fid2.id &&
            (it->it_op == IT_LOOKUP || it->it_op == IT_GETATTR)) {
                rc = mdc_revalidate_lock(exp, it, &op_data->fid2);
                /* Only return failure if it was not GETATTR by cfid
                   (from inode_revalidate) */
                if (rc || op_data->namelen != 0)
                        RETURN(rc);
        }

        /* lookup_it may be called only after revalidate_it has run, because
         * revalidate_it cannot return errors, only zero.  Returning zero causes
         * this call to lookup, which *can* return an error.
         *
         * We only want to execute the request associated with the intent one
         * time, however, so don't send the request again.  Instead, skip past
         * this and use the request from revalidate.  In this case, revalidate
         * never dropped its reference, so the refcounts are all OK */
        if (!it_disposition(it, DISP_ENQ_COMPLETE)) {
                struct ldlm_enqueue_info einfo =
                        { LDLM_IBITS, it_to_lock_mode(it), cb_blocking,
                          ldlm_completion_ast, NULL, NULL };

                rc = mdc_enqueue(exp, &einfo, it, op_data, &lockh,
                                 lmm, lmmsize, extra_lock_flags);
                if (rc < 0)
                        RETURN(rc);
                memcpy(&it->d.lustre.it_lock_handle, &lockh, sizeof(lockh));
        } else if (!op_data->fid2.id) {
                /* DISP_ENQ_COMPLETE set means there is extra reference on
                 * request referenced from this intent, saved for subsequent
                 * lookup.  This path is executed when we proceed to this
                 * lookup, so we clear DISP_ENQ_COMPLETE */
                it_clear_disposition(it, DISP_ENQ_COMPLETE);
        }

        *reqp = it->d.lustre.it_data;
        rc = mdc_finish_intent_lock(exp, *reqp, op_data, it, &lockh);

        RETURN(rc);
}
EXPORT_SYMBOL(mdc_intent_lock);

static int mdc_intent_getattr_async_interpret(struct ptlrpc_request *req,
                                              void *unused, int rc)
{
        struct mdc_enqueue_args  *ma;
        struct md_enqueue_info   *minfo;
        struct ldlm_enqueue_info *einfo;
        struct obd_export        *exp;
        struct lookup_intent     *it;
        struct lustre_handle     *lockh;
        struct obd_device        *obddev;
        int                       flags = LDLM_FL_HAS_INTENT;
        ENTRY;

        ma = (struct mdc_enqueue_args *)&req->rq_async_args;
        minfo = ma->ma_mi;
        einfo = ma->ma_ei;

        exp   = minfo->mi_exp;
        it    = &minfo->mi_it;
        lockh = &minfo->mi_lockh;

        obddev = class_exp2obd(exp);

        mdc_exit_request(&obddev->u.cli);
        if (OBD_FAIL_CHECK(OBD_FAIL_MDC_GETATTR_ENQUEUE))
                rc = -ETIMEDOUT;

        rc = ldlm_cli_enqueue_fini(exp, req, einfo->ei_type, 1, einfo->ei_mode,
                                   &flags, NULL, 0, NULL, lockh, rc);
        if (rc < 0) {
                CERROR("ldlm_cli_enqueue_fini: %d\n", rc);
                mdc_clear_replay_flag(req, rc);
                GOTO(out, rc);
        }

        rc = mdc_finish_enqueue(exp, req, einfo, it, lockh, rc);
        if (rc)
                GOTO(out, rc);

        memcpy(&it->d.lustre.it_lock_handle, lockh, sizeof(*lockh));

        rc = mdc_finish_intent_lock(exp, req, &minfo->mi_data, it, lockh);
        GOTO(out, rc);
out:
        OBD_FREE_PTR(einfo);
        minfo->mi_cb(exp, req, minfo, rc);

        return 0;
}

int mdc_intent_getattr_async(struct obd_export *exp,
                             struct md_enqueue_info *minfo,
                             struct ldlm_enqueue_info *einfo)
{
        struct mdc_op_data      *op_data = &minfo->mi_data;
        struct lookup_intent    *it = &minfo->mi_it;
        struct ptlrpc_request   *req;
        struct obd_device       *obddev = class_exp2obd(exp);
        struct ldlm_res_id       res_id = {
                                        .name = {op_data->fid1.id,
                                                 op_data->fid1.generation}
                                 };
        ldlm_policy_data_t       policy = {
                                        .l_inodebits = { MDS_INODELOCK_LOOKUP }
                                 };
        struct mdc_enqueue_args *aa;
        int                      rc;
        int                      flags = LDLM_FL_HAS_INTENT;
        ENTRY;

        CDEBUG(D_DLMTRACE,"name: %.*s in inode "LPU64", intent: %s flags %#o\n",
               op_data->namelen, op_data->name, op_data->fid1.id,
               ldlm_it2str(it->it_op), it->it_flags);

        req = mdc_intent_lookup_pack(exp, it, op_data);
        if (!req)
                RETURN(-ENOMEM);

        mdc_enter_request(&obddev->u.cli);
        rc = ldlm_cli_enqueue(exp, &req, einfo, res_id, &policy, &flags, NULL,
                              0, NULL, &minfo->mi_lockh, 1);
        if (rc < 0) {
                mdc_exit_request(&obddev->u.cli);
                RETURN(rc);
        }

        CLASSERT(sizeof(*aa) < sizeof(req->rq_async_args));
        aa = (struct mdc_enqueue_args *)&req->rq_async_args;
        aa->ma_mi = minfo;
        aa->ma_ei = einfo;
        req->rq_interpret_reply = mdc_intent_getattr_async_interpret;
        ptlrpcd_add_req(req);

        RETURN(0);
}
EXPORT_SYMBOL(mdc_intent_getattr_async);