如何从消息队列接收动态长度数据？

Question

我必须使用 SysV 消息队列为大学项目发送和接收动态数据。

数据的长度在单独的消息中传输，因此size 是已知的。

这就是我尝试接收数据的方式。我不得不承认我不是 C++ 专家，尤其是在内存分配方面。

罢工>

struct {
    long mtype;
    char *mdata;
} msg;

msg.mdata = (char *)malloc(size * sizeof(char));

msgrcv(MSGQ_ID, &msg, size, MSG_ID, 0);

问题似乎是malloc 调用，但我不知道如何正确执行此操作。

编辑

我尝试在消息队列周围的 OO 包装器中使用某种 read 方法。我想将消息队列中的数据读入char[] 或std::string。我现在看起来（简化）是这样的。

bool Wrapper::read(char *data, int length)
{
    struct Message {
        long mtype;
        std::string mdata;
    };

    Message msg;
    msg.mdata = std::string(size, '\0');

    if(msgrcv(MSGQ_ID, &msg, size, MSG_ID, 0) < 0)
    {
        return false;
    }

    memcpy(data, msg.mdata.c_str(), msg.mdata.size());

    return true;
}

我得到的只是分段错误或完全损坏的数据（尽管这些数据有时包含我想要的）。

Answer 1

这里是an example for SyS。我希望它会有所帮助。

您使用 malloc 的方式似乎是正确的，但在为 IPC 分配内存时应该非常小心。您应该检查其他进程如何管理内存（字节对齐、大小、平台...）

在您的代码中，mtype 的用途是什么？您收到的大小是否考虑了这个 mtype？还是只有mdata的大小？

更新：mtype 是消息的一部分吗？

如果是这样：

msgsize = size * sizeof(char) + sizeof(long)

pmsg = malloc(msgsize);

msgrcv(MSGQ_ID, pmsg, msgsize, MSQ_ID, 0);

如果没有

msg.data = (char *)malloc(size * sizeof(char));

msgrcv(MSGQ_ID, msg.data, size, MSQ_ID, 0);

mtype 分配在堆栈上，而数据分配在堆上。如果 msgreceive 在给定的指针上做一种 memcpy 会造成一些麻烦。

Answer 2

您不能将指向包含std::string 成员的结构的指针传递给msgrcv，这违反了接口协定。

传递给msgrcv 的第二个参数需要指向一个有足够空间的缓冲区来存储struct { long mtype; char mdata[size]; }; 形式的“普通”C 结构，其中size 是msgrcv 的第三个参数。

不幸的是，由于可能的对齐问题，确定此缓冲区的大小可能取决于size，但您必须假设它不在提供此类接口的系统上。您可以使用标准的offsetof 宏来帮助确定此大小。

由于vector 连续存储其组件，一旦您知道缓冲区的大小，您可以调整char 的vector 的大小并使用它来保存缓冲区。使用vector 可以免除您手动使用free 或delete[] 缓冲区的义务。

你需要做这样的事情。

std::string RecvMessage()
{
    extern size_t size; // maximum size, should be a parameter??
    extern int MSGQ_ID; // message queue id, should be a parameter??
    extern long MSG_ID; // message type, should be a parameter??

    // ugly struct hack required by msgrcv
    struct RawMessage {
        long mtype;
        char mdata[1];
    };

    size_t data_offset = offsetof(RawMessage, mdata);

    // Allocate a buffer of the correct size for message
    std::vector<char> msgbuf(size + data_offset);

    ssize_t bytes_read;

    // Read raw message
    if((bytes_read = msgrcv(MSGQ_ID, &msgbuf[0], size, MSG_ID, 0)) < 0)
    {
        throw MsgRecvFailedException();
    }

    // a string encapsulates the data and the size, why not just return one
    return std::string(msgbuf.begin() + data_offset, msgbuf.begin() + data_offset + bytes_read);
}

反之，您只需按照 msgsnd 接口的要求将数据打包到一个struct hack 兼容的数据数组中。正如其他人指出的那样，它不是一个好的接口，但掩盖了实现定义的行为和对齐问题，这样的事情应该可以工作。

例如

void SendMessage(const std::string& data)
{
    extern int MSGQ_ID; // message queue id, should be a parameter??
    extern long MSG_ID; // message type, should be a parameter??

    // ugly struct hack required by msgsnd
    struct RawMessage {
        long mtype;
        char mdata[1];
    };

    size_t data_offset = offsetof(RawMessage, mdata);

    // Allocate a buffer of the required size for message
    std::vector<char> msgbuf(data.size() + data_offset);

    long mtype = MSG_ID;
    const char* mtypeptr = reinterpret_cast<char*>(&mtype);

    std::copy(mtypeptr, mtypeptr + sizeof mtype, &msgbuf[0]);
    std::copy(data.begin(), data.end(), &msgbuf[data_offset]);

    int result = msgsnd(MSGQ_ID, &msgbuf[0], msgbuf.size(), 0);
    if (result != 0)
    {
        throw MsgSendFailedException();
    }
}

Answer 3

您似乎可以随意混合 C 和 C++，所以我也会这样做。 请注意，您可能应该完全用 C 编写函数，将其放在自己的翻译单元中，然后从 C++ 调用它，但这应该可以解决您的直接问题。 看来您困惑的根源可以说是 msgrcv 的 API 设计不佳。您不能将 msg 结构的数据成员作为指针或引用或除原始内存块之外的任何其他内容 - msgrcv 会将数据直接放入结构中。

#include #include #include /* 调用者必须确保数据至少指向长度字符 */ bool read_msg( char *data, int length ) { 布尔状态 = 假； 结构味精{ 长 mtype; 字符数据[1]； } *m = (struct msg *)malloc(length + offsetof(struct msg, data)); 如果（米！= NULL）{ 如果（msgrcv（MSGQ_ID，米，长度，MSQ_ID，0）==长度）{ memcpy（数据，m->数据，长度）； 状态=真； } } 免费（米）； 返回状态； }

Answer 4

根本原因 分段错误的原因是 memcpy 尝试复制大于结构大小的消息。您的结构只有一个长（4 个字节）和一个 std:string 的空间。字符串可以是可变大小，但仅在用作字符串时（并且它会自动为此分配内存并维护一个指针）。 std:string 变量看起来像

struct {
  unsigned int stringlength;
  char *pString;
}

您的代码复制到 msg，在这个（简化的）示例中只有 8 个字节长。然后你用数据覆盖指针，这后来被解释为一个内存位置。

解决方案 一种解决方案是分配一个足够大的内存块来容纳标头 (mtype) 和 size 字节的消息。由于它只是临时空间，我建议为此使用堆栈变量。所以

 struct msg {
        long mtype;
        char data[ size ];
    } 

如果 size 是全局常量，我建议将其替换为更独特和更有意义的东西。请注意 size 以字节表示；如果您需要发送 size 个字符的 std::string，则需要至少增加 sizeof(std::string)+1 的数组长度（对于结构开销和 \0特点）;更好一点。

Answer 5

分段错误或数据损坏的原因是您为msgrcv() 和msgsnd() 函数使用了不正确的消息结构。这些函数需要使用以下结构：

struct msgbuf {
    long mtype;     /* message type, must be > 0 */
    char mtext[1];  /* message data */
};

但是您的Message 结构包含一个在堆上分配内存的std::string，但msgbuf 结构需要一个从mtext[0] 开始的内存区域。 mtext 成员不是指针，而是结构内的 char 数组。由于大小未知，数组被声明为 1 大小的数组，但实际上调用者应提供更大的缓冲区。这种技术也用于一些 Windows API。

为了能够接收消息，你需要一个缓冲区

static const unsigned int BufferSize = 256;
char* buffer = new char[ sizeof(long) + BufferSize ];
msgbuf* msg = reinterpret_cast< msgbuf* >( buffer );

那么这个msg可以传递给msgrcv()。并且不要忘记delete[] 缓冲区。

还要注意缓冲区中接收的文本不是以空值结尾的，除非在发送消息时明确写入 0 字符。这样的非空终止文本可以用于构造 std::string 从它 - 例如std::string( msg->mtext, len )或在缓冲区中为0保留一个字节，并在接收到msgrcv()返回的长度的消息后写入

您可以使用 std::vector 代替普通的 char 数组，它保证项目将存储在连续的内存范围中。以下代码可用于消息接收循环，如果消息不适合，它会自动增加缓冲区大小。

static const unsigned int InitialBufferSize = 32;
std::vector<char> buffer( InitialBufferSize );
msgbuf* msg = new (&buffer[0]) msgbuf();
// reserve space in the vector for the mtype member of msgbuf structure
std::vector<char>::size_type available = buffer.size() - sizeof(long);

for(;;)
{
    const ssize_t count = ::msgrcv( MSGQ_ID, msg, available, 0, 0 );
    if( -1 == count )
    {
        if( E2BIG == errno )
        {
            buffer.resize( buffer.size() * 2 );
            msg = new (&buffer[0]) msgbuf();
            available = buffer.size() - sizeof(long);
            continue;
        }
        perror( "Failed to read message from queue" );
        break;
    }

    // handle the received message
    ...
}