mdc_lib.c

lustre 1.6.5 source code

        rec->sa_fsgid = current->fsgid;
        rec->sa_cap = current->cap_effective;
        rec->sa_fid = data->fid1;
        rec->sa_suppgid = -1;

        if (iattr) {
                rec->sa_valid = attr_pack(iattr->ia_valid);
                rec->sa_mode = iattr->ia_mode;
                rec->sa_uid = iattr->ia_uid;
                rec->sa_gid = iattr->ia_gid;
                rec->sa_size = iattr->ia_size;
                rec->sa_atime = LTIME_S(iattr->ia_atime);
                rec->sa_mtime = LTIME_S(iattr->ia_mtime);
                rec->sa_ctime = LTIME_S(iattr->ia_ctime);
                rec->sa_attr_flags =
                               ((struct ll_iattr_struct *)iattr)->ia_attr_flags;
                if ((iattr->ia_valid & ATTR_GID) && in_group_p(iattr->ia_gid))
                        rec->sa_suppgid = iattr->ia_gid;
                else
                        rec->sa_suppgid = data->suppgids[0];
        }

        if (ealen == 0)
                return;

        memcpy(lustre_msg_buf(req->rq_reqmsg, offset + 1, ealen), ea, ealen);

        if (ea2len == 0)
                return;

        memcpy(lustre_msg_buf(req->rq_reqmsg, offset + 2, ea2len), ea2, ea2len);
}

void mdc_unlink_pack(struct ptlrpc_request *req, int offset,
                     struct mdc_op_data *data)
{
        struct mds_rec_unlink *rec;
        char *tmp;

        rec = lustre_msg_buf(req->rq_reqmsg, offset, sizeof (*rec));
        LASSERT (rec != NULL);

        rec->ul_opcode = REINT_UNLINK;
        rec->ul_fsuid = current->fsuid;
        rec->ul_fsgid = current->fsgid;
        rec->ul_cap = current->cap_effective;
        rec->ul_mode = data->create_mode;
        rec->ul_suppgid = data->suppgids[0];
        rec->ul_fid1 = data->fid1;
        rec->ul_fid2 = data->fid2;
        rec->ul_time = data->mod_time;

        tmp = lustre_msg_buf(req->rq_reqmsg, offset + 1, data->namelen + 1);
        LASSERT (tmp != NULL);
        LOGL0(data->name, data->namelen, tmp);
}

void mdc_link_pack(struct ptlrpc_request *req, int offset,
                   struct mdc_op_data *data)
{
        struct mds_rec_link *rec;
        char *tmp;

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

        rec->lk_opcode = REINT_LINK;
        rec->lk_fsuid = current->fsuid;
        rec->lk_fsgid = current->fsgid;
        rec->lk_cap = current->cap_effective;
        rec->lk_suppgid1 = data->suppgids[0];
        rec->lk_suppgid2 = data->suppgids[1];
        rec->lk_fid1 = data->fid1;
        rec->lk_fid2 = data->fid2;
        rec->lk_time = data->mod_time;

        tmp = lustre_msg_buf(req->rq_reqmsg, offset + 1, data->namelen + 1);
        LOGL0(data->name, data->namelen, tmp);
}

void mdc_rename_pack(struct ptlrpc_request *req, int offset,
                     struct mdc_op_data *data,
                     const char *old, int oldlen, const char *new, int newlen)
{
        struct mds_rec_rename *rec;
        char *tmp;

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

        /* XXX do something about time, uid, gid */
        rec->rn_opcode = REINT_RENAME;
        rec->rn_fsuid = current->fsuid;
        rec->rn_fsgid = current->fsgid;
        rec->rn_cap = current->cap_effective;
        rec->rn_suppgid1 = data->suppgids[0];
        rec->rn_suppgid2 = data->suppgids[1];
        rec->rn_fid1 = data->fid1;
        rec->rn_fid2 = data->fid2;
        rec->rn_time = data->mod_time;

        tmp = lustre_msg_buf(req->rq_reqmsg, offset + 1, oldlen + 1);
        LOGL0(old, oldlen, tmp);

        if (new) {
                tmp = lustre_msg_buf(req->rq_reqmsg, offset + 2, newlen + 1);
                LOGL0(new, newlen, tmp);
        }
}

void mdc_getattr_pack(struct ptlrpc_request *req, int offset, __u64 valid,
                      int flags, struct mdc_op_data *data)
{
        struct mds_body *b;
        b = lustre_msg_buf(req->rq_reqmsg, offset, sizeof(*b));

        b->fsuid = current->fsuid;
        b->fsgid = current->fsgid;
        b->capability = current->cap_effective;
        b->valid = valid;
        b->flags = flags | MDS_BFLAG_EXT_FLAGS;
        /* skip MDS_BFLAG_EXT_FLAGS to verify the "client < 1.4.7" case 
         * refer to bug 12848.
         */
        if (OBD_FAIL_CHECK(OBD_FAIL_MDC_OLD_EXT_FLAGS))
                b->flags &= ~MDS_BFLAG_EXT_FLAGS;
        b->suppgid = data->suppgids[0];

        b->fid1 = data->fid1;
        b->fid2 = data->fid2;
        if (data->name) {
                char *tmp;
                tmp = lustre_msg_buf(req->rq_reqmsg, offset + 1,
                                     data->namelen + 1);
                memcpy(tmp, data->name, data->namelen);
                data->name = tmp;
        }
}

void mdc_close_pack(struct ptlrpc_request *req, int offset, struct obdo *oa,
                    __u64 valid, struct obd_client_handle *och)
{
        struct mds_body *body;

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

        mdc_pack_fid(&body->fid1, oa->o_id, 0, oa->o_mode);
        memcpy(&body->handle, &och->och_fh, sizeof(body->handle));
        if (oa->o_valid & OBD_MD_FLATIME) {
                body->atime = oa->o_atime;
                body->valid |= OBD_MD_FLATIME;
        }
        if (oa->o_valid & OBD_MD_FLMTIME) {
                body->mtime = oa->o_mtime;
                body->valid |= OBD_MD_FLMTIME;
        }
        if (oa->o_valid & OBD_MD_FLCTIME) {
                body->ctime = oa->o_ctime;
                body->valid |= OBD_MD_FLCTIME;
        }
        if (oa->o_valid & OBD_MD_FLSIZE) {
                body->size = oa->o_size;
                body->valid |= OBD_MD_FLSIZE;
        }
        if (oa->o_valid & OBD_MD_FLBLOCKS) {
                body->blocks = oa->o_blocks;
                body->valid |= OBD_MD_FLBLOCKS;
        }
        if (oa->o_valid & OBD_MD_FLFLAGS) {
                body->flags = oa->o_flags;
                body->valid |= OBD_MD_FLFLAGS;
        }
}

struct mdc_cache_waiter {       
        struct list_head        mcw_entry;
        wait_queue_head_t       mcw_waitq;
};

static int mdc_req_avail(struct client_obd *cli, struct mdc_cache_waiter *mcw)
{
        int rc;
        ENTRY;
        spin_lock(&cli->cl_loi_list_lock);
        rc = list_empty(&mcw->mcw_entry);
        spin_unlock(&cli->cl_loi_list_lock);
        RETURN(rc);
};

/* We record requests in flight in cli->cl_r_in_flight here.
 * There is only one write rpc possible in mdc anyway. If this to change
 * in the future - the code may need to be revisited. */
void mdc_enter_request(struct client_obd *cli)
{
        struct mdc_cache_waiter mcw;
        struct l_wait_info lwi = { 0 };

        spin_lock(&cli->cl_loi_list_lock);
        if (cli->cl_r_in_flight >= cli->cl_max_rpcs_in_flight) {
                list_add_tail(&mcw.mcw_entry, &cli->cl_cache_waiters);
                init_waitqueue_head(&mcw.mcw_waitq);
                spin_unlock(&cli->cl_loi_list_lock);
                l_wait_event(mcw.mcw_waitq, mdc_req_avail(cli, &mcw), &lwi);
        } else {
                cli->cl_r_in_flight++;
                spin_unlock(&cli->cl_loi_list_lock);
        }
}

void mdc_exit_request(struct client_obd *cli)
{
        struct list_head *l, *tmp;
        struct mdc_cache_waiter *mcw;

        spin_lock(&cli->cl_loi_list_lock);
        cli->cl_r_in_flight--;

        list_for_each_safe(l, tmp, &cli->cl_cache_waiters) {
                if (cli->cl_r_in_flight >= cli->cl_max_rpcs_in_flight) {
                        /* No free request slots anymore */
                        break;
                }

                mcw = list_entry(l, struct mdc_cache_waiter, mcw_entry);
                list_del_init(&mcw->mcw_entry);
                cli->cl_r_in_flight++;
                wake_up(&mcw->mcw_waitq);
        }
        /* Empty waiting list? Decrease reqs in-flight number */

        spin_unlock(&cli->cl_loi_list_lock);
}