在这篇文章中，我们将了解一位软件工程师在一个项目中的经验，该项目需要在内存受限的嵌入式系统上使用协议缓冲区。

每天，我处理 定制软件开发 并参与各个行业的项目。我专注于 在嵌入式系统中使用 Modern C++ 以及使用 Qt 构建应用程序。在这里，我将与您分享我在 内存受限的嵌入式系统上需要协议缓冲区的项目中的经验。让我们来看看！

想象这样一种情况，几个人见面，每个人都说不同的语言。为了相互理解，他们开始使用团队中每个人都能理解的语言。然后，每个人想要说些什么都必须将他们的想法（通常是他们的母语）翻译成小组的语言。

然后我们可以说他们每个人都在群体的语言和特定的母语之间执行某种形式的编码和解码信息。

如果我们将单个语言更改为编程语言，将组语言更改为 Protocol Buffers 消息语言，我们将获得 Protocol Buffers 的优势之一；也就是说，能够以整个组都知道的特定编程语言创建消息，并将其翻译成只有特定组成员知道的语言的形式。

除了独立于语言和平台以及编码和解码数据的能力之外，Protocol Buffers 还可以快速有效地完成它。

根据 Wikipedia的说法，“Google 广泛使用协议缓冲区来存储和交换各种结构化信息。该方法作为自定义远程过程调用（RPC 或远程过程调用）系统的基础，该系统用于几乎所有的交互谷歌的机器通信。”

协议缓冲区徽标

协议缓冲区消息语言

正如 Google 所说， “协议缓冲区使您能够定义一次数据的结构化方式（以 .proto 文件的形式），然后您可以使用特殊生成的源代码轻松编写和读取结构化数据往返于各种数据流并使用各种语言。”

协议缓冲区背后的想法

Protocol Buffers 语言指南继续说：“首先，让我们看一个非常简单的示例。假设您要定义一种Person消息格式，其中每个人都有姓名、年龄和电子邮件。这是.proto用于定义此消息类型的文件：

原始缓冲区

// person.proto

syntax = "proto3";

message Person {
  string name = 1;
  int32 age = 2;
  string email = 3;
}

该文件的第一行指定您正在使用proto3语法。”

Person消息定义指定了三个字段（名称/值对），每个字段用于您希望包含在此类消息中的每条数据。该字段具有 a name、 atype,和 a field number。

拥有.proto文件后，您可以生成特定语言的源代码：例如，C++，使用称为protocol compiler.aka的特殊编译器protoc。

协议编译器使用可视化

生成的文件包含语言原生结构来操作消息 Let's call it API。

API为您提供所有必要的类和方法set以及retrieve数据以及serialization to字节parsing from流的方法。序列化和解析是在后台处理的。

对于 C++，生成的文件包含Person类和所有必要的方法来处理底层数据。例如：

C++

void clear_name();
const ::std::string &name() const;
void set_name(const ::std::string &value);
void set_name(const char *value);

此外，Person类继承方法从google::protobuf::Message流中序列化或反序列化（解析）：

C++

// Serialization:
bool SerializeToOstream(std::ostream* output) const;
bool SerializePartialToOstream(std::ostream* output) const;
// Deserialization:
bool ParseFromIstream(std::istream* input);
bool ParsePartialFromIstream(std::istream* input);

在 Protobuf-C 中使用自定义分配器进行静态分配

如果您正在编写一个完全静态分配的系统，那么您可能使用的是 C 而不是 C++。在这里，您将了解如何编写使用静态分配的缓冲区而不是动态分配的内存的自定义分配器。

背后的想法

默认情况下Protobuf-C，解包时通过调用动态分配内存malloc()。有时它是no-go某些嵌入式系统或资源受限系统中的一个选项。

Protobuf-C使您能够提供custom allocator- 替代 malloc() 和 free() 函数 - 在本例中，serial_alloc().

malloc() 和 serial_alloc() 行为之间的区别

在这个例子中，我们将实现自定义malloc()和free()函数，并在自定义分配器中使用它们serial_allocator，这将把数据放入由 Protobuf-C 库静态分配的连续内存块中。

malloc()如何和工作之间的区别serial_alloc()如下图所示。

堆上 malloc() 分配的表示

静态缓冲区上 serial_alloc() “分配”的表示

通常，malloc() 在堆上“随机”分配内存，导致内存碎片整理。我们的自定义 serial_alloc() 按顺序在静态分配的内存上“分配”内存，这导致没有堆使用和内存碎片整理。

环境设置

本文中显示的代码在 Ubuntu 22.04 LTS 上进行了测试。

要安装protoc-c编译器Protocol Buffers C Runtime，只需运行：

sudo apt install libprotobuf-c-dev protobuf-c-compiler

通过运行检查它是否有效：

protoc-c --version

那应该返回已安装的版本：

纯文本

protobuf-c 1.3.3
libprotoc 3.12.4

如果您需要源代码或需要从源代码构建，请参阅GitHub 存储库。

信息

在此示例中，在文件中创建了一个简单的MessageProtobuf 消息。message.proto

原始缓冲区

syntax = "proto3";

message Message
{
    bool flag = 1;
    float value = 2;
}

代码生成

要生成代码，只需运行：

protoc-c -I=. --c_out=. message.proto

这将生成两个文件：message.pb-ch和message.pb-cc。

程序编译

要使用生成的代码编译您的 C 程序并链接到protobuf-c library，您只需运行：

gcc -Wall -Wextra -Wpedantic main.c message.pb-c.c -lprotobuf-c -o protobuf-c-custom_allocator

代码概述

一般来说，代码使用 Protobuf-C 对静态缓冲区进行序列化/编码/打包pack_buffer，然后对另一个静态缓冲区进行反序列化/解码/解包，out：

C

#include "message.pb-ch"br#include <stdbool.h>br#include <stdio.h>br#include <string.h>brbr静态 uint8_t  pack_buffer [ 100 ]；brbr主函数 （）br{br    留言 ；_br    message__init ( & in );brbr    在. 标志 = 真；br    在. 值 =  1.234f ;brbr    // 序列化：br    message__pack ( & in , pack_buffer );brbr    // 反序列化：br    unpacked_message_wrapper 出来；br    消息*  outPtr  =  unpack_to_message_wrapper_from_buffer ( message__get_packed_size ( & in ), pack_buffer , & out );brbr    如果（空 ！=  outPtr）br    {br        断言（in.flag == out.message.flag）；_ _ _ _ _  br        断言（in.value == out.message.value）；_ _  _ _ _ brbr        断言（in.flag == outPtr - > flag  ）； br        断言（in.value == outPtr - > value  ）； br    }br    别的br    {br        printf ( "错误: 解压到串行缓冲区失败！可能 MAX_UNPACKED_MESSAGE_LENGTH 太小或请求的大小不正确。\n" );br    }brbr    返回 0 ;br}

在unpack_to_message_wrapper_from_buffer()中，我们创建对象并用和函数ProtobufCAllocator填充它（替换和）。然后，我们通过调用和传递来解包消息：serial_alloc()serial_free()malloc()free()message__unpackserial_allocator

C

Message* unpack_to_message_wrapper_from_buffer(const size_t packed_message_length, const uint8_t* buffer, unpacked_message_wrapper* wrapper)
{
    wrapper->next_free_index = 0;

    // Here is the trick: We pass `wrapper` (not wrapper.buffer) as `allocator_data`, to track number of allocations in `serial_alloc()`.
    ProtobufCAllocator serial_allocator = {.alloc = serial_alloc, .free = serial_free, .allocator_data = wrapper};

    return message__unpack(&serial_allocator, packed_message_length, buffer);
}

malloc()-Based 与serial_alloc-Based 方法的比较

您可以在下面找到默认 Protobuf-C 行为（malloc()基于 -）和使用自定义分配器的自定义行为之间的比较：

Protobuf-C 默认行为 →使用动态内存分配：

C

tatic uint8_t buffer[SOME_BIG_ENOUGH_SIZE];

...

// NULL in this context means -> use malloc():
Message* parsed = message__unpack(NULL, packed_size, bufer);
// dynamic memory allocation occurred above

...

// somewhere below memory must be freed:
free(me)

Protobuf-C 使用自定义分配器 →不使用动态内存分配：

C

// statically allocated buffer inside some wrapper around the unpacked proto message:
typedef struct
{
    uint8_t buffer[SOME_BIG_ENOUGH_SIZE];
    ...
} unpacked_message_wrapper;

...

// malloc and free functions replacements:
static void* serial_alloc(void* allocator_data, size_t size) { ... }
static void serial_free(void* allocator_data, void* ignored) { ... }

...

ProtobufCAllocator serial_allocator = { .alloc = serial_alloc,
                                        .free = serial_free,
                                        .allocator_data = wrapper};

// now, instead of NULL we pass serial_allocator:
if (NULL == message__unpack(&serial_allocator, packed_message_length, input_buffer))
{
    printf("Unpack to serial buffer failed!\n");
}

最有趣的部分是unpacked_message_wrapper结构和serial_alloc() serial_free()实现，下面将对其进行解释。

围绕原型消息进行结构

unpacked_message_wrapperstruct 只是一个简单的 proto 包装器，并且在 union 中Message足够大，可以存储解压缩的数据并跟踪该缓冲区中的已用空间：buffernext_free_index

C

#define MAX_UNPACKED_MESSAGE_LENGTH 100

typedef struct
{
    size_t next_free_index;
    union
    {
        uint8_t buffer[MAX_UNPACKED_MESSAGE_LENGTH];
        Message message;   // Replace `Message` with your own type - generated from your own .proto message
    };
} unpacked_message_wrapper;

对象的大小Message不会改变它的大小，但Message可以是一个扩展的 .proto（请参阅本文的“提示和技巧”部分），例如包含重复字段，这通常涉及多个malloc()调用。所以你可能需要更多的尺寸，而不是 Message它本身的尺寸。为了实现这一点，buffer和message成员在一个工会中。

MAX_UNPACKED_MESSAGE_LENGTH必须足够大以适应最坏的情况。有关更多信息，请查看“提示和技巧”部分。

该结构的目的是将预定义的内存缓冲区unpacked_message_wrapper保存在一个位置，并跟踪该缓冲区上的“分配”。

实施serial_alloc()和serial_free()

的签名serial_alloc()遵循以下ProtobufCAllocator要求：

C

static void* serial_alloc(void* allocator_data, size_t size)

serial_alloc()size在上分配请求allocator_data，然后递增next_free_index到下一个字边界的开头（这是一种优化，将连续的数据块对齐到下一个字边界）。size在解析/解码数据时来自 Protobuf-C 内部。

C

static void* serial_alloc(void* allocator_data, size_t size)
{
    void* ptr_to_memory_block = NULL;

    unpacked_message_wrapper* const wrapper = (unpacked_message_wrapper*)allocator_data;

    // Optimization: Align to next word boundary.
    const size_t temp_index = wrapper->next_free_index + ((size + sizeof(int)) & ~(sizeof(int)));

    if ((size > 0) && (temp_index <= MAX_UNPACKED_MESSAGE_LENGTH))
    {
        ptr_to_memory_block = (void*)&wrapper->buffer[wrapper->next_free_index];

        wrapper->next_free_index = temp_index;
    }

    return ptr_to_memory_block;
}

第一次调用时serial_alloc()，它设置next_free_index为分配的大小并返回指向缓冲区开头的指针：

在第二次调用时，它重新计算next_free_index值并将地址返回到下一个数据块：

在第三次通话中：

该serial_free()函数将使用的缓冲区空间设置为零：

C

static void serial_free(void* allocator_data, void* ignored)
{
    (void)ignored;

    unpacked_message_wrapper* wrapper = (unpacked_message_wrapper*)allocator_data;
    wrapper->next_free_index = 0;
}

当serial_free()被调用时，它通过设置为零来“释放”所有内存next_free_index，以便可以重用缓冲区：

测试实施

该实现在Valgrind. 要在 Valgrind 类型下运行程序：

valgrind ./protobuf-c-custom_allocator

在生成的报告中，您将看到未进行任何分配：

纯文本

==3977== Memcheck, a memory error detector
==3977== Copyright (C) 2002-2017, and GNU GPL'd, by Julian Seward et al.
==3977== Using Valgrind-3.18.1 and LibVEX; rerun with -h for copyright info
==3977== Command: ./protobuf-c-custom_allocator
==3977== 
==3977== 
==3977== HEAP SUMMARY:
==3977==     in use at exit: 0 bytes in 0 blocks
==3977==   total heap usage: 0 allocs, 0 frees, 0 bytes allocated
==3977== 
==3977== All heap blocks were freed -- no leaks are possible
==3977== 
==3977== For lists of detected and suppressed errors, rerun with: -s
==3977== ERROR SUMMARY: 0 errors from 0 contexts (suppressed: 0 from 0)

概括

要编写和使用自定义分配器，您必须：

1. 确定您希望如何存储数据的方式。
2. 围绕您的 proto 消息类型编写某种包装器，并提供适当alloc()的和free()功能替换。
3. 创建包装器的对象。
4. 创建ProtobufCAllocator使用你的alloc()和free()替换的，然后在x__unpack()函数中使用分配器。

在这个例子中，我们：

• 决定将数据存储在连续的、静态分配的内存块中 (1)
• unpacked_message_wrapper作为 protoMessage和的包装器编写，并为and :和(2)buffer提供替换alloc()free()serial_alloc()serial_free()
• 在堆栈上创建了一个对象unpacked_message_wrapper，我们将其命名为out(3)
• 在unpack_to_message_wrapper_from_buffer中，我们创建并用andProtobufCAllocator填充它并传递给函数 (4)。serial_alloc()serial_free()message__unpack

技巧和窍门

如果您在一个内存非常有限的系统上工作并且每个字节都是黄金，那么您可以确定需要多大MAX_UNPACKED_MESSAGE_LENGTH。为此，您可以首先为MAX_UNPACKED_MESSAGE_LENGTH. 然后在 中serial_alloc，您需要添加一些工具：

C

static void* serial_alloc(void* allocator_data, size_t size)
{
    static int call_counter = 0;
    static size_t needed_space_counter = 0;
    needed_space_counter += ((size + sizeof(int)) & ~(sizeof(int)));

    printf("serial_alloc() called for: %d time. Needed space for worst case scenario is = %ld\n", ++call_counter, needed_space_counter);

    ...

对于这个示例案例，我们得到：

纯文本

serial_alloc called for: 1 time. The needed space for the worst-case scenario is = 32

当 .proto 消息变得更加复杂时，事情会变得很困难。让我们在 .proto 消息中添加一个新字段：

原始缓冲区

syntax = "proto3";

message Message
{
    bool flag = 1;
    float value = 2;
    repeated string names = 3; // added field, type repeated means "dynamic array"
}

然后我们在消息中添加新条目：

C

int main()
{
    Message in;
    message__init(&in);

    in.flag = true;
    in.value = 1.234f;
    const char name1[] = "Let's";
    const char name2[] = "Solve";
    const char name3[] = "It";
    const char* names[] = {name1, name2, name3};
    in.names = (char**)names;
    in.n_names = 3;

    // Serialization:
    message__pack(&in, pack_buffer);
    
    ...

我们将在输出中看到这一点：

纯文本

serial_alloc() called for: 1 time. Needed space for worst case scenario is = 48
serial_alloc() called for: 2 time. Needed space for worst case scenario is = 72
serial_alloc() called for: 3 time. Needed space for worst case scenario is = 82
serial_alloc() called for: 4 time. Needed space for worst case scenario is = 92
serial_alloc() called for: 5 time. Needed space for worst case scenario is = 95

所以现在我们知道，这是我们最坏的情况。我们至少需要一个95字节宽的缓冲区。

在现实世界中，您通常希望设置大于 95 的空间，除非您 100% 确定并彻底测试过。