mdc_locks.c

lustre 1.6.5 source code

                mdc_open_pack(req, DLM_INTENT_REC_OFF, data,
                              it->it_create_mode, 0, it->it_flags,
                              lmm, lmmsize);

                ptlrpc_req_set_repsize(req, 5, repsize);
        }
        return req;
}

static struct ptlrpc_request *mdc_intent_unlink_pack(struct obd_export *exp,
                                                     struct lookup_intent *it,
                                                     struct mdc_op_data *data)
{
        struct ptlrpc_request *req;
        struct ldlm_intent *lit;
        struct obd_device *obddev = class_exp2obd(exp);
        int size[5] = { [MSG_PTLRPC_BODY_OFF] = sizeof(struct ptlrpc_body),
                        [DLM_LOCKREQ_OFF]     = sizeof(struct ldlm_request),
                        [DLM_INTENT_IT_OFF]   = sizeof(*lit),
                        [DLM_INTENT_REC_OFF]  = sizeof(struct mds_rec_unlink),
                        [DLM_INTENT_REC_OFF+1]= data->namelen + 1 };
        int repsize[5] = { [MSG_PTLRPC_BODY_OFF] = sizeof(struct ptlrpc_body),
                           [DLM_LOCKREPLY_OFF]   = sizeof(struct ldlm_reply),
                           [DLM_REPLY_REC_OFF]   = sizeof(struct mds_body),
                           [DLM_REPLY_REC_OFF+1] = obddev->u.cli.
                                                        cl_max_mds_easize,
                           [DLM_REPLY_REC_OFF+2] = obddev->u.cli.
                                                        cl_max_mds_cookiesize };

        req = ldlm_prep_enqueue_req(exp, 5, size, NULL, 0);
        if (req) {
                /* pack the intent */
                lit = lustre_msg_buf(req->rq_reqmsg, DLM_INTENT_IT_OFF,
                                     sizeof(*lit));
                lit->opc = (__u64)it->it_op;

                /* pack the intended request */
                mdc_unlink_pack(req, DLM_INTENT_REC_OFF, data);

                ptlrpc_req_set_repsize(req, 5, repsize);
        }
        return req;
}

static struct ptlrpc_request *mdc_intent_lookup_pack(struct obd_export *exp,
                                                     struct lookup_intent *it,
                                                     struct mdc_op_data *data)
{
        struct ptlrpc_request *req;
        struct ldlm_intent *lit;
        struct obd_device *obddev = class_exp2obd(exp);
        int size[5] = { [MSG_PTLRPC_BODY_OFF] = sizeof(struct ptlrpc_body),
                        [DLM_LOCKREQ_OFF]     = sizeof(struct ldlm_request),
                        [DLM_INTENT_IT_OFF]   = sizeof(*lit),
                        [DLM_INTENT_REC_OFF]  = sizeof(struct mds_body),
                        [DLM_INTENT_REC_OFF+1]= data->namelen + 1 };
        int repsize[5] = { [MSG_PTLRPC_BODY_OFF] = sizeof(struct ptlrpc_body),
                           [DLM_LOCKREPLY_OFF]   = sizeof(struct ldlm_reply),
                           [DLM_REPLY_REC_OFF]   = sizeof(struct mds_body),
                           [DLM_REPLY_REC_OFF+1] = obddev->u.cli.
                                                        cl_max_mds_easize,
                           [DLM_REPLY_REC_OFF+2] = LUSTRE_POSIX_ACL_MAX_SIZE };
        obd_valid valid = OBD_MD_FLGETATTR | OBD_MD_FLEASIZE | OBD_MD_FLACL |
                          OBD_MD_FLMODEASIZE | OBD_MD_FLDIREA;

                req = ldlm_prep_enqueue_req(exp, 5, size, NULL, 0);
        if (req) {
                /* pack the intent */
                lit = lustre_msg_buf(req->rq_reqmsg, DLM_INTENT_IT_OFF,
                                     sizeof(*lit));
                lit->opc = (__u64)it->it_op;

                /* pack the intended request */
                mdc_getattr_pack(req, DLM_INTENT_REC_OFF, valid, it->it_flags,
                                 data);
                ptlrpc_req_set_repsize(req, 5, repsize);
        }
        return req;
}

static struct ptlrpc_request *mdc_intent_readdir_pack(struct obd_export *exp)
{
        struct ptlrpc_request *req;
        int size[2] = { [MSG_PTLRPC_BODY_OFF] = sizeof(struct ptlrpc_body),
                        [DLM_LOCKREQ_OFF]     = sizeof(struct ldlm_request) };
        int repsize[2] = { [MSG_PTLRPC_BODY_OFF] = sizeof(struct ptlrpc_body),
                           [DLM_LOCKREPLY_OFF]   = sizeof(struct ldlm_reply) };

        req = ldlm_prep_enqueue_req(exp, 2, size, NULL, 0);
        if (req)
                ptlrpc_req_set_repsize(req, 2, repsize);
        return req;
}

static int mdc_finish_enqueue(struct obd_export *exp,
                              struct ptlrpc_request *req,
                              struct ldlm_enqueue_info *einfo,
                              struct lookup_intent *it,
                              struct lustre_handle *lockh,
                              int rc)
{
        struct ldlm_request *lockreq;
        struct ldlm_reply *lockrep;
        ENTRY;

        LASSERT(rc >= 0);
        /* Similarly, if we're going to replay this request, we don't want to
         * actually get a lock, just perform the intent. */
        if (req->rq_transno || req->rq_replay) {
                lockreq = lustre_msg_buf(req->rq_reqmsg, DLM_LOCKREQ_OFF,
                                         sizeof(*lockreq));
                lockreq->lock_flags |= LDLM_FL_INTENT_ONLY;
        }

        if (rc == ELDLM_LOCK_ABORTED) {
                einfo->ei_mode = 0;
                memset(lockh, 0, sizeof(*lockh));
                rc = 0;
        } else { /* rc = 0 */
                struct ldlm_lock *lock = ldlm_handle2lock(lockh);
                LASSERT(lock);

                /* If the server gave us back a different lock mode, we should
                 * fix up our variables. */
                if (lock->l_req_mode != einfo->ei_mode) {
                        ldlm_lock_addref(lockh, lock->l_req_mode);
                        ldlm_lock_decref(lockh, einfo->ei_mode);
                        einfo->ei_mode = lock->l_req_mode;
                }
                LDLM_LOCK_PUT(lock);
        }

        lockrep = lustre_msg_buf(req->rq_repmsg, DLM_LOCKREPLY_OFF,
                                 sizeof(*lockrep));
        LASSERT(lockrep != NULL);                 /* checked by ldlm_cli_enqueue() */
        /* swabbed by ldlm_cli_enqueue() */
        LASSERT(lustre_rep_swabbed(req, DLM_LOCKREPLY_OFF));

        it->d.lustre.it_disposition = (int)lockrep->lock_policy_res1;
        it->d.lustre.it_status = (int)lockrep->lock_policy_res2;
        it->d.lustre.it_lock_mode = einfo->ei_mode;
        it->d.lustre.it_data = req;

        if (it->d.lustre.it_status < 0 && req->rq_replay)
                mdc_clear_replay_flag(req, it->d.lustre.it_status);

        /* If we're doing an IT_OPEN which did not result in an actual
         * successful open, then we need to remove the bit which saves
         * this request for unconditional replay.
         *
         * It's important that we do this first!  Otherwise we might exit the
         * function without doing so, and try to replay a failed create
         * (bug 3440) */
        if ((it->it_op & IT_OPEN) &&
            req->rq_replay &&
            (!it_disposition(it, DISP_OPEN_OPEN) ||
             it->d.lustre.it_status != 0))
                mdc_clear_replay_flag(req, it->d.lustre.it_status);

        DEBUG_REQ(D_RPCTRACE, req, "op: %d disposition: %x, status: %d",
                  it->it_op,it->d.lustre.it_disposition,it->d.lustre.it_status);

        /* We know what to expect, so we do any byte flipping required here */
        if (it->it_op & (IT_OPEN | IT_UNLINK | IT_LOOKUP | IT_GETATTR)) {
                struct mds_body *body;

                body = lustre_swab_repbuf(req, DLM_REPLY_REC_OFF, sizeof(*body),
                                         lustre_swab_mds_body);
                if (body == NULL) {
                        CERROR ("Can't swab mds_body\n");
                        RETURN (-EPROTO);
                }

                /* If this is a successful OPEN request, we need to set
                   replay handler and data early, so that if replay happens
                   immediately after swabbing below, new reply is swabbed
                   by that handler correctly */
                if (it_disposition(it, DISP_OPEN_OPEN) &&
                    !it_open_error(DISP_OPEN_OPEN, it))
                        mdc_set_open_replay_data(NULL, req);

                if ((body->valid & OBD_MD_FLEASIZE) != 0) {
                        void *eadata;

                        /* The eadata is opaque; just check that it is there.
                         * Eventually, obd_unpackmd() will check the contents */
                        eadata = lustre_swab_repbuf(req, DLM_REPLY_REC_OFF + 1,
                                                    body->eadatasize, NULL);
                        if (eadata == NULL) {
                                CERROR ("Missing/short eadata\n");
                                RETURN (-EPROTO);
                        }
                        if (body->valid & OBD_MD_FLMODEASIZE) {
                                struct obd_device *obddev = class_exp2obd(exp);

                                if (obddev->u.cli.cl_max_mds_easize < 
                                                        body->max_mdsize) {
                                        obddev->u.cli.cl_max_mds_easize = 
                                                body->max_mdsize;
                                        CDEBUG(D_INFO, "maxeasize become %d\n",
                                               body->max_mdsize);
                                }
                                if (obddev->u.cli.cl_max_mds_cookiesize <
                                                        body->max_cookiesize) {
                                        obddev->u.cli.cl_max_mds_cookiesize =
                                                body->max_cookiesize;
                                        CDEBUG(D_INFO, "cookiesize become %d\n",
                                               body->max_cookiesize);
                                }
                        }
                        /* We save the reply LOV EA in case we have to replay
                         * a create for recovery.  If we didn't allocate a
                         * large enough request buffer above we need to
                         * reallocate it here to hold the actual LOV EA. */
                        if (it->it_op & IT_OPEN) {
                                int offset = DLM_INTENT_REC_OFF + 2;
                                void *lmm;

                                if (lustre_msg_buflen(req->rq_reqmsg, offset) <
                                    body->eadatasize)
                                        mdc_realloc_openmsg(req, body);

                                lmm = lustre_msg_buf(req->rq_reqmsg, offset,
                                                     body->eadatasize);
                                if (lmm)
                                        memcpy(lmm, eadata, body->eadatasize);
                        }
                }
        }

        RETURN(rc);
}

/* We always reserve enough space in the reply packet for a stripe MD, because
 * we don't know in advance the file type. */
int mdc_enqueue(struct obd_export *exp, struct ldlm_enqueue_info *einfo,
                struct lookup_intent *it, struct mdc_op_data *data,
                struct lustre_handle *lockh, void *lmm, int lmmsize,
                int extra_lock_flags)
{
        struct ptlrpc_request *req;
        struct obd_device *obddev = class_exp2obd(exp);
        struct ldlm_res_id res_id =
                { .name = {data->fid1.id, data->fid1.generation} };
        ldlm_policy_data_t policy = { .l_inodebits = { MDS_INODELOCK_LOOKUP } };
        int flags = extra_lock_flags | LDLM_FL_HAS_INTENT;
        int rc;
        ENTRY;

        LASSERTF(einfo->ei_type == LDLM_IBITS,"lock type %d\n", einfo->ei_type);
        if (it->it_op & (IT_UNLINK | IT_GETATTR | IT_READDIR))
                policy.l_inodebits.bits = MDS_INODELOCK_UPDATE;

        if (it->it_op & IT_OPEN) {
                req = mdc_intent_open_pack(exp, it, data, lmm, lmmsize);
                if (it->it_flags & O_JOIN_FILE) {
                        policy.l_inodebits.bits = MDS_INODELOCK_UPDATE;
                }
        } else if (it->it_op & IT_UNLINK) {
                req = mdc_intent_unlink_pack(exp, it, data);
        } else if (it->it_op & (IT_GETATTR | IT_LOOKUP)) {
                req = mdc_intent_lookup_pack(exp, it, data);
        } else if (it->it_op == IT_READDIR) {
                req = mdc_intent_readdir_pack(exp);
        } else {
                CERROR("bad it_op %x\n", it->it_op);
                RETURN(-EINVAL);
        }

        if (!req)
                RETURN(-ENOMEM);

         /* It is important to obtain rpc_lock first (if applicable), so that
          * threads that are serialised with rpc_lock are not polluting our
          * rpcs in flight counter */
        mdc_get_rpc_lock(obddev->u.cli.cl_rpc_lock, it);
        mdc_enter_request(&obddev->u.cli);
        rc = ldlm_cli_enqueue(exp, &req, einfo, res_id, &policy, &flags, NULL,
                              0, NULL, lockh, 0);
        mdc_exit_request(&obddev->u.cli);
        mdc_put_rpc_lock(obddev->u.cli.cl_rpc_lock, it);
        if (rc < 0) {
                CERROR("ldlm_cli_enqueue: %d\n", rc);
                mdc_clear_replay_flag(req, rc);
                ptlrpc_req_finished(req);
                RETURN(rc);
        }
        rc = mdc_finish_enqueue(exp, req, einfo, it, lockh, rc);

        RETURN(rc);
}
EXPORT_SYMBOL(mdc_enqueue);

int mdc_revalidate_lock(struct obd_export *exp, struct lookup_intent *it,
                        struct ll_fid *fid)
{
                /* We could just return 1 immediately, but since we should only
                 * be called in revalidate_it if we already have a lock, let's
                 * verify that. */
        struct ldlm_res_id res_id = {.name ={fid->id, fid->generation}};
        struct lustre_handle lockh;
        ldlm_policy_data_t policy;
        ldlm_mode_t mode;