pack_generic.c

来自「lustre 1.6.5 source code」· C语言 代码 · 共 1,984 行 · 第 1/5 页

        PTLRPC_RS_DEBUG_LRU_ADD(rs);

        RETURN (0);
}

static int lustre_pack_reply_v2(struct ptlrpc_request *req, int count,
                                int *lens, char **bufs, int flags)
{
        struct ptlrpc_reply_state *rs;
        int                        msg_len;
        int                        size;
        ENTRY;

        LASSERT(req->rq_reply_state == NULL);

        if ((flags & LPRFL_EARLY_REPLY) == 0)
                req->rq_packed_final = 1;

        msg_len = lustre_msg_size_v2(count, lens);
        size = sizeof(struct ptlrpc_reply_state) + msg_len;
        OBD_ALLOC(rs, size);
        if (unlikely(rs == NULL)) {
                rs = lustre_get_emerg_rs(req->rq_rqbd->rqbd_service, size);
                if (!rs) 
                        RETURN (-ENOMEM);
        }
        atomic_set(&rs->rs_refcount, 1);        /* 1 ref for rq_reply_state */
        rs->rs_cb_id.cbid_fn = reply_out_callback;
        rs->rs_cb_id.cbid_arg = rs;
        rs->rs_service = req->rq_rqbd->rqbd_service;
        rs->rs_size = size;
        CFS_INIT_LIST_HEAD(&rs->rs_exp_list);
        CFS_INIT_LIST_HEAD(&rs->rs_obd_list);
        rs->rs_msg = (struct lustre_msg *)(rs + 1);

        req->rq_replen = msg_len;
        req->rq_reply_state = rs;
        req->rq_repmsg = rs->rs_msg;

        /* server side, no rq_repbuf */
        lustre_init_msg_v2(rs->rs_msg, count, lens, bufs);
        lustre_msg_add_version(rs->rs_msg, PTLRPC_MSG_VERSION);
        lustre_set_rep_swabbed(req, MSG_PTLRPC_BODY_OFF);

        PTLRPC_RS_DEBUG_LRU_ADD(rs);

        RETURN(0);
}

int lustre_pack_reply_flags(struct ptlrpc_request *req, int count, int *lens,
                            char **bufs, int flags)
{
        int size[] = { sizeof(struct ptlrpc_body) };

        if (!lens) {
                LASSERT(count == 1);
                lens = size;
        }

        LASSERT(count > 0);
        LASSERT(lens[MSG_PTLRPC_BODY_OFF] == sizeof(struct ptlrpc_body));

        switch (req->rq_reqmsg->lm_magic) {
        case LUSTRE_MSG_MAGIC_V1:
        case LUSTRE_MSG_MAGIC_V1_SWABBED:
                return lustre_pack_reply_v1(req, count - 1, lens + 1,
                                            bufs ? bufs + 1 : NULL, flags);
        case LUSTRE_MSG_MAGIC_V2:
        case LUSTRE_MSG_MAGIC_V2_SWABBED:
                return lustre_pack_reply_v2(req, count, lens, bufs, flags);
        default:
                LASSERTF(0, "incorrect message magic: %08x\n",
                         req->rq_reqmsg->lm_magic);
                return -EINVAL;
        }
}

int lustre_pack_reply(struct ptlrpc_request *req, int count, int *lens,
                      char **bufs)
{
        int rc = lustre_pack_reply_flags(req, count, lens, bufs, 0);
        if (rc != 0)
                CERROR("lustre_pack_reply failed: rc=%d size=%d\n", rc,
                       lustre_msg_size(req->rq_reqmsg->lm_magic, count, lens));
        return rc;
}


void *lustre_msg_buf_v1(void *msg, int n, int min_size)
{
        struct lustre_msg_v1 *m = (struct lustre_msg_v1 *)msg;
        int i, offset, buflen, bufcount;

        LASSERT(m != NULL);
        LASSERT(n >= 0);

        bufcount = m->lm_bufcount;
        if (n >= bufcount) {
                CDEBUG(D_INFO, "msg %p buffer[%d] not present (count %d)\n",
                       m, n, bufcount);
                return NULL;
        }

        buflen = m->lm_buflens[n];
        if (buflen < min_size) {
                CERROR("msg %p buffer[%d] size %d too small (required %d)\n",
                       m, n, buflen, min_size);
                LBUG();
                return NULL;
        }

        offset = lustre_msg_hdr_size_v1(bufcount);
        for (i = 0; i < n; i++)
                offset += size_round(m->lm_buflens[i]);

        return (char *)m + offset;
}

void *lustre_msg_buf_v2(struct lustre_msg_v2 *m, int n, int min_size)
{
        int i, offset, buflen, bufcount;

        LASSERT(m != NULL);
        LASSERT(n >= 0);

        bufcount = m->lm_bufcount;
        if (n >= bufcount) {
                CDEBUG(D_INFO, "msg %p buffer[%d] not present (count %d)\n",
                       m, n, bufcount);
                return NULL;
        }

        buflen = m->lm_buflens[n];
        if (buflen < min_size) {
                CERROR("msg %p buffer[%d] size %d too small (required %d)\n",
                       m, n, buflen, min_size);
                return NULL;
        }

        offset = lustre_msg_hdr_size_v2(bufcount);
        for (i = 0; i < n; i++)
                offset += size_round(m->lm_buflens[i]);

        return (char *)m + offset;
}

void *lustre_msg_buf(struct lustre_msg *m, int n, int min_size)
{
        switch (m->lm_magic) {
        case LUSTRE_MSG_MAGIC_V1:
        case LUSTRE_MSG_MAGIC_V1_SWABBED:
                return lustre_msg_buf_v1(m, n - 1, min_size);
        case LUSTRE_MSG_MAGIC_V2:
        case LUSTRE_MSG_MAGIC_V2_SWABBED:
                return lustre_msg_buf_v2(m, n, min_size);
        default:
                LASSERTF(0, "incorrect message magic: %08x\n", m->lm_magic);
                return NULL;
        }
}

void lustre_shrink_reply_v1(struct ptlrpc_request *req, int segment,
                            unsigned int newlen, int move_data)
{
        struct lustre_msg_v1 *msg = (struct lustre_msg_v1 *)req->rq_repmsg;
        char *tail = NULL, *newpos;
        int tail_len = 0, n;

        LASSERT(req->rq_reply_state);
        LASSERT(msg);
        LASSERT(segment >= 0);
        LASSERT(msg->lm_bufcount > segment);
        LASSERT(msg->lm_buflens[segment] >= newlen);

        if (msg->lm_buflens[segment] == newlen)
                return;

        if (move_data && msg->lm_bufcount > segment + 1) {
                tail = lustre_msg_buf_v1(msg, segment + 1, 0);
                for (n = segment + 1; n < msg->lm_bufcount; n++)
                        tail_len += size_round(msg->lm_buflens[n]);
        }

        msg->lm_buflens[segment] = newlen;

        if (tail && tail_len) {
                newpos = lustre_msg_buf_v1(msg, segment + 1, 0);
                LASSERT(newpos <= tail);
                if (newpos != tail)
                        memcpy(newpos, tail, tail_len);
        }

        if (newlen == 0 && msg->lm_bufcount > segment + 1) {
                memmove(&msg->lm_buflens[segment], &msg->lm_buflens[segment + 1],
                        (msg->lm_bufcount - segment - 1) * sizeof(__u32));
                msg->lm_buflens[msg->lm_bufcount - 1] = 0;
        }

        req->rq_replen = lustre_msg_size_v1(msg->lm_bufcount, (int *)msg->lm_buflens);
}

void lustre_shrink_reply_v2(struct ptlrpc_request *req, int segment,
                            unsigned int newlen, int move_data)
{
        struct lustre_msg_v2 *msg = req->rq_repmsg;
        char *tail = NULL, *newpos;
        int tail_len = 0, n;

        LASSERT(req->rq_reply_state);
        LASSERT(msg);
        LASSERT(msg->lm_bufcount > segment);
        LASSERT(msg->lm_buflens[segment] >= newlen);

        if (msg->lm_buflens[segment] == newlen)
                return;

        if (move_data && msg->lm_bufcount > segment + 1) {
                tail = lustre_msg_buf_v2(msg, segment + 1, 0);
                for (n = segment + 1; n < msg->lm_bufcount; n++)
                        tail_len += size_round(msg->lm_buflens[n]);
        }

        msg->lm_buflens[segment] = newlen;

        if (tail && tail_len) {
                newpos = lustre_msg_buf_v2(msg, segment + 1, 0);
                LASSERT(newpos <= tail);
                if (newpos != tail)
                        memcpy(newpos, tail, tail_len);
        }

        if (newlen == 0 && msg->lm_bufcount > segment + 1) {
                memmove(&msg->lm_buflens[segment], &msg->lm_buflens[segment + 1],
                        (msg->lm_bufcount - segment - 1) * sizeof(__u32));
                msg->lm_buflens[msg->lm_bufcount - 1] = 0;
        }

        req->rq_replen = lustre_msg_size_v2(msg->lm_bufcount, (int *)msg->lm_buflens);
}

/*
 * shrink @segment to size @newlen. if @move_data is non-zero, we also move
 * data forward from @segment + 1.
 * 
 * if @newlen == 0, we remove the segment completely, but we still keep the
 * totally bufcount the same to save possible data moving. this will leave a
 * unused segment with size 0 at the tail, but that's ok.
 *
 * CAUTION:
 * + if any buffers higher than @segment has been filled in, must call shrink
 *   with non-zero @move_data.
 * + caller should NOT keep pointers to msg buffers which higher than @segment
 *   after call shrink.
 */
void lustre_shrink_reply(struct ptlrpc_request *req, int segment,
                        unsigned int newlen, int move_data)
{
        switch (req->rq_repmsg->lm_magic) {
        case LUSTRE_MSG_MAGIC_V1:
                lustre_shrink_reply_v1(req, segment - 1, newlen, move_data);
                return;
        case LUSTRE_MSG_MAGIC_V2:
                lustre_shrink_reply_v2(req, segment, newlen, move_data);
                return;
        default:
                LASSERTF(0, "incorrect message magic: %08x\n",
                         req->rq_repmsg->lm_magic);
        }
}

void lustre_free_reply_state(struct ptlrpc_reply_state *rs)
{
        PTLRPC_RS_DEBUG_LRU_DEL(rs);

        LASSERT (atomic_read(&rs->rs_refcount) == 0);
        LASSERT (!rs->rs_difficult || rs->rs_handled);
        LASSERT (!rs->rs_on_net);
        LASSERT (!rs->rs_scheduled);
        LASSERT (rs->rs_export == NULL);
        LASSERT (rs->rs_nlocks == 0);
        LASSERT (list_empty(&rs->rs_exp_list));
        LASSERT (list_empty(&rs->rs_obd_list));

        if (unlikely(rs->rs_prealloc)) {
                struct ptlrpc_service *svc = rs->rs_service;

                spin_lock(&svc->srv_lock);
                list_add(&rs->rs_list,
                         &svc->srv_free_rs_list);
                spin_unlock(&svc->srv_lock);
                cfs_waitq_signal(&svc->srv_free_rs_waitq);
        } else {
                OBD_FREE(rs, rs->rs_size);
        }
}

int lustre_unpack_msg_v1(void *msg, int len)
{
        struct lustre_msg_v1 *m = (struct lustre_msg_v1 *)msg;
        int flipped, required_len, i;
        ENTRY;

        /* Now we know the sender speaks my language. */
        required_len = lustre_msg_hdr_size_v1(0);
        if (len < required_len) {
                /* can't even look inside the message */
                CERROR("message length %d too small for lustre_msg\n", len);
                RETURN(-EINVAL);
        }

        flipped = lustre_msg_swabbed((struct lustre_msg *)m);

        if (flipped) {
                __swab32s(&m->lm_type);
                __swab32s(&m->lm_version);
                __swab32s(&m->lm_opc);
                __swab64s(&m->lm_last_xid);
                __swab64s(&m->lm_last_committed);
                __swab64s(&m->lm_transno);
                __swab32s(&m->lm_status);
                __swab32s(&m->lm_flags);
                __swab32s(&m->lm_conn_cnt);
                __swab32s(&m->lm_bufcount);
        }

        if (m->lm_version != PTLRPC_MSG_VERSION) {
                CERROR("wrong lustre_msg version %08x\n", m->lm_version);
                RETURN(-EINVAL);
        }

        required_len = lustre_msg_hdr_size_v1(m->lm_bufcount);
        if (len < required_len) {
                /* didn't receive all the buffer lengths */
                CERROR("message length %d too small for %d buflens\n",
                        len, m->lm_bufcount);
                RETURN(-EINVAL);
        }

        for (i = 0; i < m->lm_bufcount; i++) {
                if (flipped)
                        __swab32s (&m->lm_buflens[i]);
                required_len += size_round(m->lm_buflens[i]);
        }

        if (len < required_len) {
                CERROR("len: %d, required_len %d\n", len, required_len);
                CERROR("bufcount: %d\n", m->lm_bufcount);
                for (i = 0; i < m->lm_bufcount; i++)
                        CERROR("buffer %d length %d\n", i, m->lm_buflens[i]);
                RETURN(-EINVAL);
        }

        RETURN(0);
}

static int lustre_unpack_msg_v2(struct lustre_msg_v2 *m, int len)
{
        int flipped, required_len, i;

        /* Now we know the sender speaks my language. */
        required_len = lustre_msg_hdr_size_v2(0);
        if (len < required_len) {
                /* can't even look inside the message */
                CERROR("message length %d too small for lustre_msg\n", len);
                RETURN(-EINVAL);
        }

        flipped = lustre_msg_swabbed(m);

        if (flipped) {
                __swab32s(&m->lm_bufcount);
                __swab32s(&m->lm_secflvr);
                __swab32s(&m->lm_repsize);
                __swab32s(&m->lm_cksum);
                __swab32s(&m->lm_flags);
                CLASSERT(offsetof(typeof(*m), lm_padding_2) != 0);
                CLASSERT(offsetof(typeof(*m), lm_padding_3) != 0);
        }

        required_len = lustre_msg_hdr_size_v2(m->lm_bufcount);
        if (len < required_len) {
                /* didn't receive all the buffer lengths */
                CERROR ("message length %d too small for %d buflens\n",
                        len, m->lm_bufcount);
                return -EINVAL;
        }

        for (i = 0; i < m->lm_bufcount; i++) {
                if (flipped)
                        __swab32s(&m->lm_buflens[i]);
                required_len += size_round(m->lm_buflens[i]);
        }

        if (len < required_len) {
                CERROR("len: %d, required_len %d\n", len, required_len);
                CERROR("bufcount: %d\n", m->lm_bufcount);