分布式共享内存 [重复]

Question

我当前的系统在 Linux 上运行，不同的任务使用共享内存来访问公共数据（定义为 C 结构）。共享数据的大小约为 100K。

现在，我想在 Windows 上运行主用户界面，同时在 Linux 中保留所有其他任务，我正在寻找共享内存的最佳替代品。 UI的刷新率约为每秒10次。

我怀疑是自己做还是使用第三方解决方案更好。如果我自己做，我的想法是打开一个套接字，然后在客户端（Windows）和服务器（Linux）之间使用某种数据序列化。

对于第三方解决方案，我对选项的数量感到有点不知所措。经过一番搜索，在我看来，正确的解决方案可能是 MPI，但我想在开始使用它之前考虑其他选项：XDR、OpenMP、JSON、DBus、RDMA、memcached、boost 库……有没有人有任何经验和他们中的任何一个？使用此类解决方案从 Windows 访问 Linux 上的共享内存有哪些优缺点？

也许 MPI 或其他第三方解决方案过于复杂，无法像我这样简单地使用，我应该使用自己动手的方法吗？如果我采用该解决方案有什么建议吗？我错过了什么吗？我看错方向了吗？我不想重新发明轮子。

对于文件共享，我正在考虑 Samba。

我的开发是用C和C++完成的，当然解决方案需要在Linux和Windows下编译。

Answer 1

AFAIK（但我不知道 Windows）你不能同时在 Linux 和 Windows 上 share memory（shm_overview(7)）（你可以使用一些网络协议共享数据，这就是 memcached 所做的)。

但是，您可以让 Linux 进程响应来自 Windows 机器的网络消息（但这不是共享内存！）。

您是否考虑过使用 Web 界面？您可以在 Linux 上（添加到您的 Linux 软件）上使用 ajax 技术添加一个 Web 界面，例如使用FastCGI 或像libonion 或wt 这样的HTTP 服务器库。然后，您的 Windows 用户可以使用他们的浏览器与您的程序进行交互（该程序仍将在某些 Linux 计算服务器上运行）。

PS。阅读distributed shared memory 上的维基页面。这意味着与您所要求的不同！另请阅读message passing！

Answer 2

我建议在用户界面和服务之间使用普通的 TCP 套接字，在两者之间传递简单的标记消息帧，最好以独立于体系结构的方式（即具有特定的字节顺序以及整数和浮点类型）。

在服务器端，我会使用辅助进程或线程，它为每个客户端维护共享状态结构的两个本地副本。第一个反映客户端知道的状态，第二个用于定期对共享状态进行快照。当两者不同时，助手将不同的字段发送给客户端。相反，客户端可以向助手发送修改请求。

我不会将 100k 共享数据结构作为单个块发送；这将是非常低效的。此外，假设状态包含多个字段，让客户端发送整个新的 100k 状态将覆盖客户端不打算发送的字段。

我会为状态中的每个字段或原子操作的一组字段使用一条消息。消息结构本身应该非常简单。至少，每条消息都应该以固定大小的长度和类型字段开头，以便接收消息变得微不足道。在不破坏向后兼容性的情况下，保留以后扩展功能/字段的可能性总是一个好主意。

您不需要很多代码来实现上述内容。对于状态中的每种不同类型的字段（char、short、int、long、double、float 以及您可能使用的任何其他类型或结构），您确实需要一些访问器/操纵器代码，并且（以及比较具有模拟用户状态的活动状态）可能是大部分代码。

虽然我通常不建议重新发明轮子，但在这种情况下，我建议您直接在 TCP 之上编写自己的消息传递代码，因为我知道的所有库都可以用于此目的，而不是笨拙或会强加一些我宁愿留给应用程序的设计选择。

如果您愿意，我可以提供一些示例代码，希望能更好地说明这一点。我没有 Windows，但 Qt 确实有一个 QTcpSocket 类，你可以使用它在 Linux 和 Windows 中的工作方式相同，所以不应该有（m）任何差异。事实上，如果你设计好消息结构，你可以用 Python 等其他语言编写 UI，而与服务器本身没有任何区别。

---

承诺的例子如下；为最大长度的帖子道歉。

当特定字段只有一个写入器（可能有很多读取器）并且您希望向写入路径添加最小的开销时，配对计数器非常有用。阅读永远不会阻塞，读者会看到他们是否获得了有效的副本或是否正在进行更新，他们不应该依赖他们看到的值。 （这也允许在异步信号安全上下文中为字段提供可靠的写入语义，其中不保证 pthread 锁定工作。）

此实现在 GCC-4.7 及更高版本上效果最佳，但也应适用于早期的 C 编译器。旧版 __sync 内置程序也适用于 Intel、Pathscale 和 PGI C 编译器。 counter.h：

#ifndef   COUNTER_H
#define   COUNTER_H

/*
 * Atomic generation counter support.
 *
 * A generation counter allows a single writer thread to
 * locklessly modify a value non-atomically, while letting
 * any number of reader threads detect the change. Reader
 * threads can also check if the value they observed is
 * "atomic", or taken during a modification in progress.
 *
 * There is no protection against multiple concurrent writers.
 *
 * If the writer gets canceled or dies during an update,
 * the counter will get garbled. Reinitialize before relying
 * on such a counter.
 *
 * There is no guarantee that a reader will observe an
 * "atomic" value, if the writer thread modifies the value
 * more often than the reader thread can read it.
 *
 * Two typical use cases:
 *
 *  A) A single writer requires minimum overhead/latencies,
 *     whereas readers can poll and retry if necessary
 *
 *  B) Non-atomic value or structure modified in
 *     an interrupt context (async-signal-safe)
 *
 * Initialization:
 *
 *      Counter counter = COUNTER_INITIALIZER;
 *
 *      or
 *
 *      Counter counter;
 *      counter_init(&counter);
 *
 * Write sequence:
 *
 *      counter_acquire(&counter);
 *      (modify or write value)
 *      counter_release(&counter);
 *
 * Read sequence:
 *
 *      unsigned int check;
 *      check = counter_before(&counter);
 *      (obtain a copy of the value)
 *      if (check == counter_after(&counter))
 *          (a copy of the value is atomic)
 *
 * Read sequence with spin-waiting,
 * will spin forever if counter garbled:
 *
 *      unsigned int check;
 *      do {
 *          check = counter_before(&counter);
 *          (obtain a copy of the value)
 *      } while (check != counter_after(&counter));
 *
 * All these are async-signal-safe, and will never block
 * (except for the spinning loop just above, obviously).
*/

typedef struct {
    volatile unsigned int  value[2];
} Counter;
#define COUNTER_INITIALIZER {{0U,0U}}

#if (__GNUC__ > 4) || (__GNUC__ == 4 && __GNUC_MINOR__ >= 7)
/*
 * GCC 4.7 and later provide __atomic built-ins.
 * These are very efficient on x86 and x86-64.
*/

static inline void counter_init(Counter *const counter)
{
    /* This is silly; assignments should suffice. */
    do {
        __atomic_store_n(&(counter->value[1]), 0U, __ATOMIC_SEQ_CST);
        __atomic_store_n(&(counter->value[0]), 0U, __ATOMIC_SEQ_CST);
    } while (__atomic_load_n(&(counter->value[0]), __ATOMIC_SEQ_CST) |
             __atomic_load_n(&(counter->value[1]), __ATOMIC_SEQ_CST));
}

static inline unsigned int counter_before(Counter *const counter)
{
    return __atomic_load_n(&(counter->value[1]), __ATOMIC_SEQ_CST);
}

static inline unsigned int counter_after(Counter *const counter)
{
    return __atomic_load_n(&(counter->value[0]), __ATOMIC_SEQ_CST);
}

static inline unsigned int counter_acquire(Counter *const counter)
{
    return __atomic_fetch_add(&(counter->value[0]), 1U, __ATOMIC_SEQ_CST);
}

static inline unsigned int counter_release(Counter *const counter)
{
    return __atomic_fetch_add(&(counter->value[1]), 1U, __ATOMIC_SEQ_CST);
}

#else
/*
 * Rely on legacy __sync built-ins.
 *
 * Because there is no __sync_fetch(),
 * counter_before() and counter_after() are safe,
 * but not optimal (especially on x86 and x86-64).
*/

static inline void counter_init(Counter *const counter)
{
    /* This is silly; assignments should suffice. */
    do {
        counter->value[0] = 0U;
        counter->value[1] = 0U;
        __sync_synchronize();
    } while (__sync_fetch_and_add(&(counter->value[0]), 0U) |
             __sync_fetch_and_add(&(counter->value[1]), 0U));
}

static inline unsigned int counter_before(Counter *const counter)
{
    return __sync_fetch_and_add(&(counter->value[1]), 0U);
}

static inline unsigned int counter_after(Counter *const counter)
{
    return __sync_fetch_and_add(&(counter->value[0]), 0U);
}

static inline unsigned int counter_acquire(Counter *const counter)
{
    return __sync_fetch_and_add(&(counter->value[0]), 1U);
}

static inline unsigned int counter_release(Counter *const counter)
{
    return __sync_fetch_and_add(&(counter->value[1]), 1U);
}

#endif

#endif /* COUNTER_H */

每个共享状态字段类型都需要定义自己的结构。出于我们的目的，我将只使用Value 结构来描述其中之一。您可以定义所有需要的字段和值类型以满足您的需求。例如，双精度浮点分量的 3D 向量将是

typedef struct {
    double    x;
    double    y;
    double    z;
} Value;

typedef struct {
    Counter   counter;
    Value     value;
} Field;

修改线程使用的地方，例如

void set_field(Field *const field, Value *const value)
{
    counter_acquire(&field->counter);
    field->value = value;
    counter_release(&field->counter);
}

每个读者使用例如维护自己的本地快照

typedef enum {
    UPDATED = 0,
    UNCHANGED = 1,
    BUSY = 2
} Updated;

Updated check_field(Field *const local, const Field *const shared)
{
    Field cache;

    cache.counter[0] = counter_before(&shared->counter);

    /* Local counter check allows forcing an update by
     * simply changing one half of the local counter.
     * If you don't need that, omit the local-only part. */
    if (local->counter[0] == local->counter[1] &&
        cache.counter[0] == local->counter[0])
        return UNCHANGED;

    cache.value = shared->value;

    cache.counter[1] = counter_after(&shared->counter);
    if (cache.counter[0] != cache.counter[1])
        return BUSY;

    *local = cache;

    return UPDATED;
}

注意，上面的cache构成了该值的一个本地副本，因此每个读者只需要维护一份它感兴趣的字段的副本。另外，上面的访问器函数只修改了一个成功的快照local .

---

当使用 pthread 锁定时，pthread_rwlock_t 是一个不错的选择，只要程序员了解多个读取器和写入器的优先级问题。 （see man pthread_rwlock_rdlock 和man pthread_rwlock_wrlock 了解详情。）

就个人而言，我会坚持使用尽可能短的保持时间的互斥锁，以降低整体复杂性。我将使用的受互斥体保护的更改计数值的字段结构可能类似于

typedef struct {
    pthread_mutex_t        mutex;
    Value                  value;
    volatile unsigned int  change;
} Field;

void set_field(Field *const field, const Value *const value)
{
    pthread_mutex_lock(&field->mutex);
    field->value = *value;
#if (__GNUC__ > 4) || (__GNUC__ == 4 && __GNUC_MINOR__ >= 7)
    __atomic_fetch_add(&field->change, 1U, __ATOMIC_SEQ_CST);
#else
    __sync_fetch_and_add(&field->change, 1U);
#endif
    pthread_mutex_unlock(&field->mutex);
}

void get_field(Field *const local, Field *const field)
{
    pthread_mutex_lock(&field->mutex);
    local->value = field->value;
#if (__GNUC__ > 4) || (__GNUC__ == 4 && __GNUC_MINOR__ >= 7)
    local->value = __atomic_fetch_n(&field->change, __ATOMIC_SEQ_CST);
#else
    local->value = __sync_fetch_and_add(&field->change, 0U);
#endif
    pthread_mutex_unlock(&field->mutex);
}

Updated check_field(Field *const local, Field *const shared)
{
#if (__GNUC__ > 4) || (__GNUC__ == 4 && __GNUC_MINOR__ >= 7)
    if (local->change == __atomic_fetch_n(&field->change, __ATOMIC_SEQ_CST))
        return UNCHANGED;
#else
    if (local->change == __sync_fetch_and_add(&field->change, 0U))
        return UNCHANGED;
#endif
    pthread_mutex_lock(&shared->mutex);
    local->value = shared->value;
#if (__GNUC__ > 4) || (__GNUC__ == 4 && __GNUC_MINOR__ >= 7)
    local->change = __atomic_fetch_n(&field->change, __ATOMIC_SEQ_CST);
#else
    local->change = __sync_fetch_and_add(&field->change, 0U);
#endif
    pthread_mutex_unlock(&shared->mutex);
}

请注意，本地副本中的互斥体字段是未使用的，并且可以从本地副本中省略（定义一个合适的Cache等结构类型没有它）。

关于 change 字段的语义很简单：它总是以原子方式访问，因此可以在不持有互斥体的情况下读取。仅在持有互斥锁时才修改（但由于读取器不持有互斥锁，因此即使在那时也必须原子地修改）。

请注意，如果您的服务器仅限于 x86 或 x86-64 架构，您可以使用上述更改的普通访问（例如field->change++），因为unsigned int 访问在这些架构上是原子的;无需使用内置函数。 （虽然增量是非原子的，但读者只会看到旧的或新的unsigned int，所以这也不是问题。）

---

在服务器上，您需要某种“代理”进程或线程，代表远程客户端执行操作，向每个客户端发送字段更改，并可选择从每个客户端接收字段内容以进行更新。

这种代理的有效实施是一个大问题。一个经过深思熟虑的建议将需要有关系统的更详细信息。我想，最简单的可能实现将是一个低优先级线程，简单地连续扫描共享状态，报告每个客户端不知道的任何字段更改，因为它看到它们。如果变化率很高——比如说，整个状态通常每秒有十几个变化——那么这可能是完全可以接受的。如果更改率不同，添加一个全局更改计数器（在修改单个字段时修改，即在 set_field() 中）并在未观察到全局更改时休眠（或至少让步）会更好。

如果有多个远程客户端，我会使用一个代理，它将更改收集到一个队列中，并在队列头部显示状态的缓存副本（之前的更改是导致它的那些）。最初连接的客户端获得整个状态，然后每个排队的更改随之而来。请注意，队列条目不需要表示字段本身，而是发送给每个客户端以更新该字段的消息，并且如果后面的更新替换了已排队的值，则可以删除已排队的值，从而减少需要发送给正在追赶的客户的数据。在我看来，这本身就是一个有趣的话题，并且值得提出一个问题。

---

就像我上面提到的，我个人会在用户界面和服务之间使用一个普通的 TCP 套接字。

关于 TCP 消息有四个主要问题：

1. 字节顺序和数据格式（除非服务器和远程客户端都保证使用相同的硬件架构）

   我首选的解决方案是在发送时使用本机字节顺序（但定义明确的标准类型），而收件人则执行任何字节顺序转换。这要求每一端发送一个初始消息，其中包含预先确定的“原型值”，用于使用的每个整数和浮点类型。

   我假设整数使用没有填充位的二进制补码格式，并且 double 和 float 分别是双精度和单精度 IEEE-754 类型。大多数当前的硬件架构本身就具有这些（尽管字节顺序确实有所不同）。奇怪的架构可以并且确实在软件中模拟这些，通常是通过提供合适的命令行开关或使用一些库。

2. 消息帧（从应用程序的角度来看，TCP 是字节流，而不是数据包流）

   最可靠的选项是为每条消息添加一个固定大小类型字段和一个固定大小长度字段的前缀。 （通常，长度字段表示该消息中有多少附加字节，并将包括发送方可能添加的任何可能的填充字节。）

   接收方将接收 TCP 输入，直到它有足够的缓存用于固定类型和长度部分以检查完整的数据包长度。然后它接收更多数据，直到它有一个完整的 TCP 数据包。最小的接收缓冲区实现类似于

3. 可扩展性

   向后兼容性对于简化更新和升级至关重要。这意味着您可能需要在从服务器到客户端的初始消息中包含某种版本号，并让服务器和客户端都忽略他们不知道的消息。 （这显然需要在每条消息中添加一个长度字段，因为收件人无法推断出他们无法识别的消息的长度。）

   在某些情况下，您可能需要将消息标记为重要的功能，即如果收件人无法识别该消息，则应中止进一步的通信。这最容易通过在类型字段中使用某种标识符或位来实现。例如，PNG file format 与我在这里描述的消息格式非常相似，具有一个四字节（四个 ASCII 字符）类型字段，以及每个消息的四字节（32 位）长度字段（“块”）。如果第一个类型字符是大写ASCII，则表示它是关键字符。

4. 传递初始状态

   将初始状态作为单个块传递可修复整个共享​​状态结构。相反，我建议单独传递每个字段或记录，就像它们是更新一样。可选地，当所有初始字段都已发送后，您可以发送一条消息，通知客户端他们拥有整个状态；这应该让客户端首先接收到完整的状态，然后构建和呈现用户界面，而不必动态调整以适应不同数量的字段。

   是每个客户端都需要完整状态，还是只需要状态的较小子集，这是另一个问题。当远程客户端连接到服务器时，它可能应该包含某种类型的标识符，服务器可以使用它来做出决定。

这里是 messages.h，一个内联实现整数和浮点类型处理和字节顺序检测的头文件：

#ifndef   MESSAGES_H
#define   MESSAGES_H

#include <stdint.h>
#include <string.h>
#include <errno.h>

#if defined(__GNUC__)
static const struct __attribute__((packed)) {
#else
#pragma pack(push,1)
#endif
    const uint64_t  u64;        /* Offset +0 */
    const int64_t   i64;        /*        +8 */
    const double    dbl;        /*       +16 */
    const uint32_t  u32;        /*       +24 */
    const int32_t   i32;        /*       +28 */
    const float     flt;        /*       +32 */
    const uint16_t  u16;        /*       +36 */
    const int16_t   i16;        /*       +38 */
} prototypes = {
    18364758544493064720ULL,    /* u64 */
    -1311768465156298103LL,     /* i64 */
    0.71948481353325643983254167324048466980457305908203125,
    3735928559UL,               /* u32 */
    -195951326L,                /* i32 */
    1.06622731685638427734375f, /* flt */
    51966U,                     /* u16 */
    -7658                       /* i16 */
};
#if !defined(__GNUC__)
#pragma pack(pop)
#endif

/* Add prototype section to a buffer.
*/
size_t add_prototypes(unsigned char *const data, const size_t size)
{
    if (size < sizeof prototypes) {
        errno = ENOSPC;
        return 0;
    } else {
        memcpy(data, &prototypes, sizeof prototypes);
        return sizeof prototypes;
    }
}

/*
 * Byte order manipulation functions.
*/
static void reorder64(void *const dst, const void *const src, const int order)
{
    if (order) {
        uint64_t  value;
        memcpy(&value, src, 8);
        if (order & 1)
            value = ((value >>  8U) & 0x00FF00FF00FF00FFULL)
                  | ((value & 0x00FF00FF00FF00FFULL) <<  8U);
        if (order & 2)
            value = ((value >> 16U) & 0x0000FFFF0000FFFFULL)
                  | ((value & 0x0000FFFF0000FFFFULL) << 16U);
        if (order & 4)
            value = ((value >> 32U) & 0x00000000FFFFFFFFULL)
                  | ((value & 0x00000000FFFFFFFFULL) << 32U);
        memcpy(dst, &value, 8);
    } else
    if (dst != src)
        memmove(dst, src, 8);
}

static void reorder32(void *const dst, const void *const src, const int order)
{
    if (order) {
        uint32_t  value;
        memcpy(&value, src, 4);
        if (order & 1)
            value = ((value >>  8U) & 0x00FF00FFUL)
                  | ((value & 0x00FF00FFUL) <<  8U);
        if (order & 2)
            value = ((value >> 16U) & 0x0000FFFFUL)
                  | ((value & 0x0000FFFFUL) << 16U);
        memcpy(dst, &value, 4);
    } else
    if (dst != src)
        memmove(dst, src, 4);
}

static void reorder16(void *const dst, const void *const src, const int order)
{
    if (order & 1) {
        const unsigned char c[2] = { ((const unsigned char *)src)[0],
                                     ((const unsigned char *)src)[1] };
        ((unsigned char *)dst)[0] = c[1];
        ((unsigned char *)dst)[1] = c[0];
    } else
    if (dst != src)
        memmove(dst, src, 2);
}

/* Detect byte order conversions needed.
 *
 * If the prototypes cannot be supported, returns -1.
 *
 * If prototype record uses native types, returns 0.
 *
 * Otherwise, bits 0..2 are the integer order conversion,
 * and bits 3..5 are the floating-point order conversion.
 * If 'order' is the return value, use
 *      reorderXX(local, remote, order)
 * for integers, and
 *      reorderXX(local, remote, order/8)
 * for floating-point types.
 *
 * For adjusting records sent to server, just do the same,
 * but with order obtained by calling this function with
 * parameters swapped.
*/
int detect_order(const void *const native, const void *const other)
{
    const unsigned char *const  source = other;
    const unsigned char *const  target = native;
    unsigned char               temp[8];
    int                         iorder = 0;
    int                         forder = 0;

    /* Verify the size of the types.
     * C89/C99/C11 says sizeof (char) == 1, but check that too. */
    if (sizeof (double) != 8 ||
        sizeof (int64_t) != 8 ||
        sizeof (uint64_t) != 8 ||
        sizeof (float) != 4 ||
        sizeof (int32_t) != 4 ||
        sizeof (uint32_t) != 4 ||
        sizeof (int16_t) != 2 ||
        sizeof (uint16_t) != 2 ||
        sizeof (unsigned char) != 1 ||
        sizeof prototypes != 40)
        return -1;

    /* Find byte order for the largest floating-point type. */
    while (1) {
        reorder64(temp, source + 16, forder);
        if (!memcmp(temp, target + 16, 8))
            break;
        if (++forder >= 8)
            return -1;
    }

    /* Verify forder works for all other floating-point types. */
    reorder32(temp, source + 32, forder);
    if (memcmp(temp, target + 32, 4))
        return -1;

    /* Find byte order for the largest integer type. */
    while (1) {
        reorder64(temp, source + 0, iorder);
        if (!memcmp(temp, target + 0, 8))
            break;
        if (++iorder >= 8)
            return -1;
    }
    /* Verify iorder works for all other integer types. */
    reorder64(temp, source + 8, iorder);
    if (memcmp(temp, target + 8, 8))
        return -1;
    reorder32(temp, source + 24, iorder);
    if (memcmp(temp, target + 24, 4))
        return -1;
    reorder32(temp, source + 28, iorder);
    if (memcmp(temp, target + 28, 4))
        return -1;
    reorder16(temp, source + 36, iorder);
    if (memcmp(temp, target + 36, 2))
        return -1;
    reorder16(temp, source + 38, iorder);
    if (memcmp(temp, target + 38, 2))
        return -1;

    /* Everything works. */
    return iorder + 8 * forder;
}

/* Verify current architecture is supportable.
 * This could be a compile-time check.
 *
 * (The buffer contains prototypes for network byte order,
 *  and actually returns the order needed to convert from
 *  native to network byte order.)
 *
 * Returns -1 if architecture is not supported,
 * a nonnegative (0 or positive) value if successful.
*/
static int verify_architecture(void)
{
    static const unsigned char network_endian[40] = {
        254U, 220U, 186U, 152U, 118U,  84U,  50U,  16U, /* u64 */
        237U, 203U, 169U, 135U, 238U, 204U, 170U, 137U, /* i64 */
         63U, 231U,   6U,   5U,   4U,   3U,   2U,   1U, /* dbl */
        222U, 173U, 190U, 239U,                         /* u32 */
        244U,  82U,   5U,  34U,                         /* i32 */
         63U, 136U, 122U,  35U,                         /* flt */
        202U, 254U,                                     /* u16 */
        226U,  22U,                                     /* i16 */
    };

    return detect_order(&prototypes, network_endian);
}

#endif /* MESSAGES_H */

这是一个示例函数，发送者可以使用该函数将由 32 位无符号整数标识的 3 分量向量打包成 36 字节消息。这使用以四个字节（“Vec3”）开头的消息帧，后跟 32 位帧长度，然后是 32 位标识符，然后是三个双精度：

size_t add_vector(unsigned char *const data, const size_t size,
                  const uint32_t id,
                  const double x, const double y, const double z)
{
    const uint32_t length = 4 + 4 + 4 + 8 + 8 + 8;

    if (size < (size_t)bytes) {
        errno = ENOSPC;
        return 0;
    }

    /* Message identifier, four bytes */
    buffer[0] = 'V';
    buffer[1] = 'e';
    buffer[2] = 'c';
    buffer[3] = '3';

    /* Length field, uint32_t */
    memcpy(buffer + 4, &length, 4);

    /* Identifier, uint32_t */
    memcpy(buffer + 8, &id, 4);

    /* Vector components */
    memcpy(buffer + 12, &x, 8);
    memcpy(buffer + 20, &y, 8);
    memcpy(buffer + 28, &z, 8);

    return length;
}

就个人而言，我更喜欢较短的 16 位标识符和 16 位长度；这也将最大消息长度限制为 65536 字节，使 65536 字节成为一个很好的读/写缓冲区大小。

接收者可以处理接收到的 TCP 数据流，例如：

static unsigned char *input_data;  /* Data buffer */
static size_t         input_size;  /* Data buffer size */
static unsigned char *input_head;  /* Next byte in buffer */
static unsigned char *input_tail;  /* After last buffered byte */
static int            input_order; /* From detect_order() */

/* Function to handle "Vec3" messages: */
static void handle_vec3(unsigned char *const msg)
{
    uint32_t id;
    double   x, y, z;

    reorder32(&id, msg+8, input_order);

    reorder64(&x, msg+12, input_order/8);
    reorder64(&y, msg+20, input_order/8);
    reorder64(&z, msg+28, input_order/8);

    /* Do something with vector id, x, y, z. */
}

/* Function that tries to consume thus far buffered
 * input data -- typically run once after each
 * successful TCP receive. */
static void consume(void)
{
    while (input_head + 8 < input_tail) {
        uint32_t length;

        /* Get current packet length. */
        reorder32(&length, input_head + 4, input_order);
        if (input_head + length < input_tail)
            break; /* Not all read, yet. */

        /* We have a full packet! */

        /* Handle "Vec3" packet: */
        if (input_head[0] == 'V' &&
            input_head[1] == 'e' &&
            input_head[2] == 'c' &&
            input_head[3] == '3')
            handle_vec3(input_head);

        /* Advance to next packet. */
        input_head += length;
    }

    if (input_head < input_tail) {
        /* Move partial message to start of buffer. */
        if (input_head > input_data) {
            const size_t have = input_head - input_data;
            memmove(input_data, input_head, have);
            input_head = input_data;
            input_tail = input_data + have;
        }
    } else {
        /* Buffer is empty. */
        input_head = input_data;
        input_tail = input_data;
    }
}

问题？