pack_generic.c

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

/* -*- mode: c; c-basic-offset: 8; indent-tabs-mode: nil; -*-
 * vim:expandtab:shiftwidth=8:tabstop=8:
 *
 *  Copyright (C) 2001-2003 Cluster File Systems, Inc.
 *   Author: Peter J. Braam <[e-mail]>
 *   Author: Phil Schwan <[e-mail]>
 *   Author: Eric Barton <[e-mail]>
 *
 *   This file is part of the Lustre file system, http://www.lustre.org
 *   Lustre is a trademark of Cluster File Systems, Inc.
 *
 *   You may have signed or agreed to another license before downloading
 *   this software.  If so, you are bound by the terms and conditions
 *   of that agreement, and the following does not apply to you.  See the
 *   LICENSE file included with this distribution for more information.
 *
 *   If you did not agree to a different license, then this copy of Lustre
 *   is open source software; you can redistribute it and/or modify it
 *   under the terms of version 2 of the GNU General Public License as
 *   published by the Free Software Foundation.
 *
 *   In either case, Lustre is distributed in the hope that it will be
 *   useful, but WITHOUT ANY WARRANTY; without even the implied warranty
 *   of MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 *   license text for more details.
 *
 * (Un)packing of OST requests
 *
 */

#define DEBUG_SUBSYSTEM S_RPC
#ifndef __KERNEL__
# include <liblustre.h>
#endif

#include <libcfs/libcfs.h>

#include <obd_support.h>
#include <obd_class.h>
#include <lustre_net.h>

#if LUSTRE_VERSION_CODE > OBD_OCD_VERSION(1,8,0,0)
#error "lustre_msg_v1 has been deprecated since 1.6.0, please remove it"
#elif LUSTRE_VERSION_CODE > OBD_OCD_VERSION(1,6,50,0)
#warning "lustre_msg_v1 has been deprecated since 1.6.0, consider removing it"
#endif

static inline int lustre_msg_hdr_size_v1(int count)
{
        return size_round(offsetof(struct lustre_msg_v1, lm_buflens[count]));
}

static inline int lustre_msg_hdr_size_v2(int count)
{
        return size_round(offsetof(struct lustre_msg_v2, lm_buflens[count]));
}

int lustre_msg_swabbed(struct lustre_msg *msg)
{
        return (msg->lm_magic == LUSTRE_MSG_MAGIC_V1_SWABBED) ||
               (msg->lm_magic == LUSTRE_MSG_MAGIC_V2_SWABBED);
}

static inline int
lustre_msg_check_version_v2(struct lustre_msg_v2 *msg, __u32 version)
{
        __u32 ver = lustre_msg_get_version(msg);
        return (ver & LUSTRE_VERSION_MASK) != version;
}

int lustre_msg_check_version(struct lustre_msg *msg, __u32 version)
{
        switch (msg->lm_magic) {
        case LUSTRE_MSG_MAGIC_V1:
        case LUSTRE_MSG_MAGIC_V1_SWABBED:
                return 0;
        case LUSTRE_MSG_MAGIC_V2:
        case LUSTRE_MSG_MAGIC_V2_SWABBED:
                return lustre_msg_check_version_v2(msg, version);
        default:
                LASSERTF(0, "incorrect message magic: %08x\n", msg->lm_magic);
                return -EINVAL;
        }
}

/* early reply size */
int lustre_msg_early_size() {
        static int size = 0;
        if (!size) 
                size = lustre_msg_size(LUSTRE_MSG_MAGIC_V2, 1, NULL);
        return size;
}
EXPORT_SYMBOL(lustre_msg_early_size);

static inline int lustre_msg_size_v1(int count, int *lengths)
{
        int size;
        int i;

        LASSERT(count >= 0);
        size = lustre_msg_hdr_size_v1(count);
        for (i = 0; i < count; i++)
                size += size_round(lengths[i]);

        return size;
}

static inline int lustre_msg_size_v2(int count, int *lengths)
{
        int size;
        int i;

        size = lustre_msg_hdr_size_v2(count);
        for (i = 0; i < count; i++)
                size += size_round(lengths[i]);

        return size;
}

/* This returns the size of the buffer that is required to hold a lustre_msg
 * with the given sub-buffer lengths.
 * NOTE: this should only be used for NEW requests, and should always be
 *       in the form of a v2 request.  If this is a connection to a v1
 *       target then the first buffer will be stripped because the ptlrpc
 *       data is part of the lustre_msg_v1 header. b=14043 */
int lustre_msg_size(__u32 magic, int count, int *lens)
{
        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 (magic) {
        case LUSTRE_MSG_MAGIC_V1:
                return lustre_msg_size_v1(count - 1, lens + 1);
        case LUSTRE_MSG_MAGIC_V2:
                return lustre_msg_size_v2(count, lens);
        default:
                CERROR("incorrect message magic: %08x\n", magic);
                return 0;
        }
}

/* This is used to determine the size of a buffer that was already packed
 * and will correctly handle the different message formats. */
int lustre_packed_msg_size(struct lustre_msg *msg)
{
        switch (msg->lm_magic) {
        case LUSTRE_MSG_MAGIC_V1: {
                struct lustre_msg_v1 *v1_msg = (struct lustre_msg_v1 *)msg;
                return lustre_msg_size_v1(v1_msg->lm_bufcount,
                                          v1_msg->lm_buflens);
        }
        case LUSTRE_MSG_MAGIC_V2:
                return lustre_msg_size_v2(msg->lm_bufcount, msg->lm_buflens);
        default:
                CERROR("incorrect message magic: %08x\n", msg->lm_magic);
                return 0;
        }
}

static void
lustre_init_msg_v1(void *m, int count, int *lens, char **bufs)
{
        struct lustre_msg_v1 *msg = (struct lustre_msg_v1 *)m;
        char *ptr;
        int i;

        LASSERT(count >= 0);
        msg->lm_magic = LUSTRE_MSG_MAGIC_V1;
        msg->lm_version = PTLRPC_MSG_VERSION;
        msg->lm_bufcount = count;

        for (i = 0; i < count; i++)
                msg->lm_buflens[i] = lens[i];

        if (bufs == NULL)
                return;

        ptr = (char *)msg + lustre_msg_hdr_size_v1(count);
        for (i = 0; i < count; i++) {
                char *tmp = bufs[i];
                LOGL(tmp, lens[i], ptr);
        }
}

static void
lustre_init_msg_v2(struct lustre_msg_v2 *msg, int count, int *lens, char **bufs)
{
        char *ptr;
        int i;

        msg->lm_bufcount = count;
        /* XXX: lm_secflvr uninitialized here */
        msg->lm_magic = LUSTRE_MSG_MAGIC_V2;

        for (i = 0; i < count; i++)
                msg->lm_buflens[i] = lens[i];

        if (bufs == NULL)
                return;

        ptr = (char *)msg + lustre_msg_hdr_size_v2(count);
        for (i = 0; i < count; i++) {
                char *tmp = bufs[i];
                LOGL(tmp, lens[i], ptr);
        }
}

static int lustre_pack_request_v1(struct ptlrpc_request *req,
                                  int count, int *lens, char **bufs)
{
        int reqlen;

        reqlen = lustre_msg_size_v1(count, lens);

        /* See if we got it from prealloc pool */
        if (req->rq_reqmsg) {
                /* Cannot return error here, that would create
                   infinite loop in ptlrpc_prep_req_pool */
                /* In this case ptlrpc_prep_req_from_pool sets req->rq_reqlen
                   to maximum size that would fit into this preallocated
                   request */
                LASSERTF(req->rq_reqlen >= reqlen, "req->rq_reqlen %d, "
                                                   "reqlen %d\n",req->rq_reqlen,
                                                    reqlen);
                memset(req->rq_reqmsg, 0, reqlen);
        } else {
                OBD_ALLOC(req->rq_reqmsg, reqlen);
                if (req->rq_reqmsg == NULL) {
                        CERROR("alloc reqmsg (len %d) failed\n", reqlen);
                        return -ENOMEM;
                }
        }

        req->rq_reqlen = reqlen;

        lustre_init_msg_v1(req->rq_reqmsg, count, lens, bufs);
        return 0;
}

static int lustre_pack_request_v2(struct ptlrpc_request *req,
                                  int count, int *lens, char **bufs)
{
        int reqlen;

        reqlen = lustre_msg_size_v2(count, lens);

        /* See if we got it from prealloc pool */
        if (req->rq_reqmsg) {
                /* Cannot return error here, that would create
                   infinite loop in ptlrpc_prep_req_pool */
                /* In this case ptlrpc_prep_req_from_pool sets req->rq_reqlen
                   to maximum size that would fit into this preallocated
                   request */
                LASSERTF(req->rq_reqlen >= reqlen, "req->rq_reqlen %d, "
                                                   "reqlen %d\n",req->rq_reqlen,
                                                    reqlen);
                memset(req->rq_reqmsg, 0, reqlen);
        } else {
                OBD_ALLOC(req->rq_reqmsg, reqlen);
                if (req->rq_reqmsg == NULL) {
                        CERROR("alloc reqmsg (len %d) failed\n", reqlen);
                        return -ENOMEM;
                }
        }

        req->rq_reqlen = reqlen;

        lustre_init_msg_v2(req->rq_reqmsg, count, lens, bufs);
        lustre_msg_add_version(req->rq_reqmsg, PTLRPC_MSG_VERSION);
        lustre_set_req_swabbed(req, MSG_PTLRPC_BODY_OFF);

        return 0;
}

int lustre_pack_request(struct ptlrpc_request *req, __u32 magic, int count,
                        int *lens, char **bufs)
{
        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 (magic) {
        case LUSTRE_MSG_MAGIC_V1:
                return lustre_pack_request_v1(req, count - 1, lens + 1,
                                              bufs ? bufs + 1 : NULL);
        case LUSTRE_MSG_MAGIC_V2:
                return lustre_pack_request_v2(req, count, lens, bufs);
        default:
                LASSERTF(0, "incorrect message magic: %08x\n", magic);
                return -EINVAL;
        }
}

#if RS_DEBUG
CFS_LIST_HEAD(ptlrpc_rs_debug_lru);
spinlock_t ptlrpc_rs_debug_lock;

#define PTLRPC_RS_DEBUG_LRU_ADD(rs)                                     \
do {                                                                    \
        spin_lock(&ptlrpc_rs_debug_lock);                               \
        list_add_tail(&(rs)->rs_debug_list, &ptlrpc_rs_debug_lru);      \
        spin_unlock(&ptlrpc_rs_debug_lock);                             \
} while (0)

#define PTLRPC_RS_DEBUG_LRU_DEL(rs)             \
do {                                            \
        spin_lock(&ptlrpc_rs_debug_lock);       \
        list_del(&(rs)->rs_debug_list);         \
        spin_unlock(&ptlrpc_rs_debug_lock);     \
} while (0)
#else
# define PTLRPC_RS_DEBUG_LRU_ADD(rs) do {} while(0)
# define PTLRPC_RS_DEBUG_LRU_DEL(rs) do {} while(0)
#endif

static struct ptlrpc_reply_state *lustre_get_emerg_rs(struct ptlrpc_service *svc,
                                                      int size)
{
        struct ptlrpc_reply_state *rs = NULL;

        spin_lock(&svc->srv_lock);
        /* See if we have anything in a pool, and wait if nothing */
        while (list_empty(&svc->srv_free_rs_list)) {
                struct l_wait_info lwi;
                int rc;
                spin_unlock(&svc->srv_lock);
                /* If we cannot get anything for some long time, we better
                   bail out instead of waiting infinitely */
                lwi = LWI_TIMEOUT(cfs_time_seconds(10), NULL, NULL);
                rc = l_wait_event(svc->srv_free_rs_waitq,
                                  !list_empty(&svc->srv_free_rs_list), &lwi);
                if (rc)
                        goto out;
                spin_lock(&svc->srv_lock);
        }
        
        rs = list_entry(svc->srv_free_rs_list.next, struct ptlrpc_reply_state,
                        rs_list);
        list_del(&rs->rs_list);
        spin_unlock(&svc->srv_lock);
        LASSERT(rs);
        LASSERTF(svc->srv_max_reply_size > size, "Want %d, prealloc %d\n", size,
                 svc->srv_max_reply_size);
        memset(rs, 0, size);
        rs->rs_prealloc = 1;
out:
        return rs;
}

static int lustre_pack_reply_v1(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_v1(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;

        lustre_init_msg_v1(rs->rs_msg, count, lens, bufs);