c++ - FlatBuffers 的多语言集成问题
问题描述
问题
当我尝试访问包含从一个客户端流式传输到服务器的消息的缓冲区(使用创建的)的一部分时,生成的服务器文件经常出现崩溃(panic: runtime error: slice bounds out of range
) 。
此问题仅出现在客户端和服务器的集成中。自己测试时,客户端和服务端都使用成功,服务端没有crash
知道:
- 发送和接收是一样的,(排除通信方式有问题)
- 发送的数据之前格式正确它被放入'并发送。
这可能是什么原因造成的?flatbuffers
flatbuffers
flatbuffers
// see the boundary tests below
byte[]
flatbuffers
buffer
问题背景和细节
我有一个c++
使用. go
_ 客户端和服务器都具有自动边界测试,
以确认它们每个都“正确”地使用了它们应该使用的方式。(即客户端在发送之前创建缓冲区,服务器在访问之前接收)
这些测试有效。我们正在使用FlatBuffers
flatbuffers
FlatBuffers-v1.10.0
我的问题是,当它们一起使用时,服务器访问缓冲区时总是出现以下错误:
panic: runtime error: slice bounds out of range
goroutine 19 [running]:
github.com/google/flatbuffers/go.(*Table).GetVOffsetT(...)
/home/.../go/github.com/google/flatbuffers/go/table.go:134
github.com/google/flatbuffers/go.(*Table).OffsetT(0x4000045c68, 0x4000000004, 0x4000160008)
/home/.../go/github.com/google/flatbuffers/go/table.go:16 +0xf0
github.com/PhantomIntelligence/Server/lib/Protocol/Stream.(*StreamedData).Id(0x4000045c68, 0x4000045c68)
/home/.../go/github.com/PhantomIntelligence/Server/lib/Protocol/Stream/SteamedData.go:30 +0x2c
github.com/PhantomIntelligence/Server/dataAccess/conversion/flatBuffers.ConvertStreamMessage(0x4000015a000, 0xa7c, 0xe00, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0)
/home/.../go/github.com/PhantomIntelligence/Server/dataAccess/conversion/flatBuffers/deserialization.go:55 +0x1fc
... // the rest is irrelevant for this question's purpose
导致崩溃的数据
当服务器尝试访问它以使用它时,崩溃总是发生在服务器接收到的第一个缓冲区上。更具体地说,它总是在尝试访问ContainerData
7th 的(0th、12th 或 15th)时发生Container
。
知道对于这个用例,我们发送一个带有 16 个容器的StreamedData,每个容器包含 16 个ContainerData
代码详情
请注意,我原来的实现只保留了结构。我不允许透露我们操作的数据的性质。因此,对象名称可能看起来很奇怪,但结构是相关部分。
架构编译
c++
客户端在编译期间运行它:
flatc -o lib/ --no-js-exports --no-prefix --scoped-enums --cpp schemas/*.fbs
go
服务器在编译期间运行它:
flatc -o lib/ --no-js-exports --no-prefix --scoped-enums --go schemas/*.fbs
模式
客户端和服务器都使用以下模式:
请注意,此摘录不是完整的架构,只是与此问题相关的部分。
// file: stream_streamed_data.fbs
namespace Protocol.Stream;
table ContainerData {
id:uint16 = 0;
a:uint16 = 0;
b:uint16 = 0;
c:int16 = 0;
d:int16 = 0;
e:uint8 = 0;
}
table Container {
id:uint16 = 65535;
data:[ContainerData];
}
table StreamedData {
id:uint16 = 0;
containers:[Container];
}
----
// file: stream_payload.fbs
include "stream_streamed_data.fbs";
/* some more includes ... */
namespace Protocol.Stream;
union PayloadContent { StreamedData, ..., Request, Result }
table Payload {
sensor_id:string;
content:PayloadContent;
}
----
// file: header.fbs
namespace Protocol;
table Header {
source_id:string;
destination_id:string;
timestamp:string;
}
----
// file: protocol_message.fbs
include "headers.fbs";
include "stream_payload.fbs";
/* some more includes ... */
namespace Protocol;
union Content { Stream.Payload, ..., Other.Payload }
table Message {
header:Header;
content:Content;
}
root_type Message;
客户端必须流式传输Protocol.Message
包含 的 s ,而这些sProtocol.Stream.Payload
本身包含一个Protocol.Stream.StreamedMessage
.
客户端实现
#ifndef CLIENT_FLATBUFFERSCONVERTER_HPP
#define CLIENT_FLATBUFFERSCONVERTER_HPP
#include <flatbuffers/flatbuffers.h>
#include <lib/protocol/protocol_message_generated.h>
#include <chrono>
#include <iostream>
namespace ServerCommunication {
enum class ProtocolStructure : uint8_t {
NONE = 0,
/* some more types ... */
SENSOR_MESSAGE = 12,
/* some more types ... */
UNRECOGNIZED_TYPE = // A NUMBER,
MIN = NONE,
MAX = UNRECOGNIZED_TYPE
};
typedef uint8_t* FlatBuffersBytes;
struct BytesToSend {
FlatBuffersBytes data;
size_t dataLength;
};
namespace BufferSize {
size_t const STREAMED_MESSAGE = 2560;
}
template<class T>
class FlatBuffersConverter {
public:
explicit FlatBuffersConverter() = delete;
~FlatBuffersConverter() noexcept = delete;
FlatBuffersConverter(FlatBuffersConverter const& other) = delete;
FlatBuffersConverter(FlatBuffersConverter&& other) noexcept = delete;
FlatBuffersConverter& operator=(FlatBuffersConverter const& other)& = delete;
FlatBuffersConverter& operator=(FlatBuffersConverter&& other)& noexcept = delete;
static BytesToSend const convertSensorMessage(typename T::Message&& message) noexcept;
private:
static std::string const generateSerializationTimestamp() noexcept;
};
template<class T>
BytesToSend const FlatBuffersConverter<T>::convertStreamMessage(typename T::StreamMessage&& message) noexcept {
flatbuffers::FlatBufferBuilder builder(BufferSize::STREAM_MESSAGE);
auto streamIdFromMessage = message.getStreamId();
auto streamId = builder.CreateString(streamIdFromMessage.c_str(), streamIdFromMessage.size());
auto source = streamId;
auto destination = builder.CreateString("Server");
auto timestamp = builder.CreateString(generateSerializationTimestamp());
auto header = Protocol::CreateHeader(
builder,
source,
destination,
timestamp);
auto containersFromStreamMessage = message.getContainers();
std::vector<flatbuffers::Offset<Protocol::Stream::Container>> containerVector;
auto containerIterator = containersFromStreamMessage->begin();
auto containerEnd = containersFromStreamMessage->end();
for (; containerIterator != containerEnd; ++containerIterator) {
auto dataFromContainer = containerIterator->getData();
std::vector<flatbuffers::Offset<GatewayProtocol::Stream::ContainerData>> containerDataVector;
auto containerDataIterator = dataFromContainer->begin();
auto containerDataEnd = dataFromContainer->end();
for (; containerDataIterator != containerDataEnd; ++containerDataIterator) {
auto track = Protocol::Stream::CreateContainerData(
builder,
containerDataIterator->id,
containerDataIterator->a,
containerDataIterator->b,
containerDataIterator->c,
containerDataIterator->d,
containerDataIterator->e);
containerDataVector.push_back(containerData);
}
auto containerDataFBVector = builder.CreateVector(containerDataVector);
auto container = Protocol::Stream::CreateContainer(
builder,
containerIterator->id,
containerDataFBVector);
containerVector.push_back(container);
}
auto containers = builder.CreateVector(containerVector);
auto streamMessageContent = Protocol::Stream::CreateStreamedData(
builder,
message.messageId,
containers);
auto streamPayload = Protocol::Stream::CreatePayload(
builder,
streamId,
Protocol::Stream::PayloadContent::StreamedData,
streamMessageContent.Union());
auto convertedMessage = Protocol::CreateMessage(
builder,
header,
Protocol::Content::Stream_Payload,
sensorPayload.Union());
builder.Finish(convertedMessage);
auto size = builder.GetSize();
auto data = builder.GetBufferPointer();
BytesToSend bytesToSend{data, size};
return bytesToSend;
}
template<class T>
std::string const FlatBuffersConverter<T>::generateSerializationTimestamp() noexcept {
std::size_t const ARBITRARY_BIG_ENOUGH_SIZE = 128;
auto timestamp = std::chrono::high_resolution_clock::now();
auto time_tTimestamp = std::chrono::system_clock::to_time_t(timestamp);
auto utcTime = gmtime(&time_tTimestamp);
char charArrayTime[ARBITRARY_BIG_ENOUGH_SIZE];
auto numberOfCharacterWritten = strftime(charArrayTime, sizeof(charArrayTime), "%D %T", utcTime);
std::string formattedTime(std::begin(charArrayTime), std::begin(charArrayTime) + numberOfCharacterWritten);
return formattedTime;
}
}
#endif //CLIENT_FLATBUFFERSCONVERTER_HPP
客户端边界测试
#ifndef CLIENT_SERVERCOMMUNICATORTEST_CPP
#define CLIENT_SERVERCOMMUNICATORTEST_CPP
#include <gtest/gtest.h>
/* some other include */
#include "spirit-sensor-gateway/server-communication/WebSocketServerCommunicationStrategy.hpp"
#include "test/utilities/stub/WebSocketServerStub.h" // <--- Receives and accesses the buffer
using ServerCommunication::WebSocketServerCommunicationStrategy;
using Stub::WebSocketServerStub;
using TestFunctions::DataTestUtil;
class WebSocketServerCommunicatorTest : public ::testing::Test {
protected:
};
TEST_F(WebSocketServerCommunicatorTest, given_aNumberOfRandomStreamDataMessage_when_send_then_theDataIsPutInFlatBuffersAndReceivedByTheServer) {
auto numberOfMessageToSend = 10;
WebSocketServerStub webSocketServerStub;
WebSocketServerCommunicationStrategy<Sensor::Test::ServerLike::Structures> webSocketServerCommunicationStrategy;
webSocketServerStub.run();
webSocketServerCommunicationStrategy.openConnection(webSocketServerStub.getAddress());
ServerStructuresLists::StreamDataMessages sentStreamDataMessages;
for (auto i = 0; i < numberOfMessageToSend; i++) {
auto streamDataMessage = DataTestUtil::createRandomStreamDataMessage();
auto streamDataMessageCopy = DataModel::StreamDataMessage(streamDataMessage);
sentStreamDataMessages.push_back(std::move(streamDataMessageCopy));
webSocketServerCommunicationStrategy.sendMessage(std::move(streamDataMessage));
}
auto numberOfReceivedMessages = webSocketServerStub.getNumberOfMessageReceived();
while(numberOfMessageToSend != numberOfReceivedMessages) {
std::this_thread::sleep_for(std::chrono::milliseconds(50));
std::this_thread::yield();
numberOfReceivedMessages = webSocketServerStub.getNumberOfMessageReceived();
}
webSocketServerCommunicationStrategy.closeConnection();
auto receivedStreamDataMessages = webSocketServerStub.getStreamDataMessages();
auto sameMessageSentAndReceived = sentStreamDataMessages.size() == receivedStreamDataMessages.size();
for (auto streamDataMessageIndex = 0;
sameMessageSentAndReceived && streamDataMessageIndex < sentStreamDataMessages.size(); ++streamDataMessageIndex) {
sameMessageSentAndReceived = sameMessageSentAndReceived &&
sentStreamDataMessages.front() == receivedStreamDataMessages.front();
sentStreamDataMessages.pop_front();
receivedStreamDataMessages.pop_front();
}
if (!sameMessageSentAndReceived) {
while (!sentStreamDataMessages.empty() && !receivedStreamDataMessages.empty()) {
std::cout << "Sent: " << sentStreamDataMessages.front().toString() << std::endl;
std::cout << "Received: " << receivedStreamDataMessages.front().toString() << std::endl;
sentStreamDataMessages.pop_front();
receivedStreamDataMessages.pop_front();
}
}
ASSERT_TRUE(sameMessageSentAndReceived);
}
#endif //CLIENT_SERVERCOMMUNICATORTEST_CPP
Client的 ServerStub FlatBuffer 转换函数
#include "ServerStubFlatBuffersConverter.h"
using Stub::ServerFlatBuffersConverter;
using ServerCommunication::ProtocolStructure;
ServerFlatBuffersConverter::StreamDataMessage
ServerFlatBuffersConverter::convertToStreamDataMessage(const ServerCommunication::FlatBuffersBytes flatBuffersBytes) {
// the identification step has been done at this point, we know the []byte holds a StreamedData
auto message = Protocol::GetMessage(flatBuffersBytes);
auto streamDataMessagePayload = message->content_as_Stream_Payload();
auto streamedData = streamDataMessagePayload->content_as_StreamedData();
auto messageId = streamedData->id();
auto streamId = flatbuffers::GetString(streamDataMessagePayload->stream_id());
auto ContainersBuffer = streamedData->containers();
std::vector<DataModel::MessageContainer> containers;
for (flatbuffers::uoffset_t containerIndex = 0;
containerIndex < ContainersBuffer->Length();
++containerIndex) {
auto containerFromBuffer = ContainersBuffer->Get(containerIndex);
auto containerId = containerFromBuffer->id();
auto containerDatasBuffer = containerFromBuffer->data();
std::vector<DataModel::ContainerData> containerDatas;
for (auto dataIterator = containerDatasBuffer->begin();
dataIterator != containerDatasBuffer->end();
++dataIterator) {
auto data = DataModel::ContainerData{
dataIterator->id(),
dataIterator->a(),
dataIterator->b(),
dataIterator->c(),
dataIterator->d(),
dataIterator->e()
};
containerDatas.push_back(data);
}
auto container = DataModel::MessageContainer(containerId, containerDatas);
containers.push_back(container);
}
StreamDataMessage message(streamId, messageId, containers);
return message;
}
服务器实现
package flatBuffers
import (
"github.com/PhantomIntelligence/Server/domain/protocol"
"github.com/PhantomIntelligence/Server/lib/Protocol"
"github.com/PhantomIntelligence/Server/lib/Protocol/Stream"
"github.com/google/flatbuffers/go"
)
type GatewayMessageType = byte
const (
NONE = 0
/* some more types ... */
SENSOR_MESSAGE = 12
/* some more types ... */
UNRECOGNIZED_TYPE = // A NUMBER
)
func ConvertStreamMessage(messageBytes []byte) protocol.StreamMessage {
// the identification step has been done at this point, we know the []byte holds a StreamedData
var protocolMessageFlatBuffersTable = new(flatbuffers.Table)
var protocolMessageContentFlatBuffersTable = new(flatbuffers.Table)
var clientMessageOffset = Protocol.GetRootAsMessage(messageBytes, 0)
var header = new(Protocol.Header)
clientMessageOffset.Header(header)
clientMessageOffset.Content(protocolMessageFlatBuffersTable)
var messageTimestampString = string(header.Timestamp())
var streamedPayload = new(Stream.Payload)
streamedPayload.Init(protocolMessageFlatBuffersTable.Bytes, protocolMessageFlatBuffersTable.Pos)
streamedPayload.Content(protocolMessageContentFlatBuffersTable)
var streamIdFromClient = string(streamedPayload.StreamId())
var streamedDataFromClient = new(Stream.StreamedData)
streamedDataFromClient.Init(protocolMessageContentFlatBuffersTable.Bytes, protocolMessageContentFlatBuffersTable.Pos)
var numberOfContainers = streamedDataFromClient.ContainersLength()
var containers []protocol.Container
for containerIndex := 0; containerIndex < numberOfContainers; containerIndex++ {
var containerFromStream = new(Stream.Container)
streamedDataFromClient.Containers(containerFromStream, containerIndex)
var numberOfContainerDatas = containerFromStream.ContainerDatasLength()
var datas []protocol.ContainerData
for containerDataIndex := 0; containerDataIndex < numberOfContainerDatas; containerDataIndex++ {
var dataFromContainer = new(Stream.ContainerData)
containerFromStream.Data(dataFromContainer, dataIndex)
datas = append(datas, protocol.ContainerData{
Id: protocol.IdType(dataFromContainer.Id()), // <--- This line crashes ! always @ containerIndex = 6, containerDataIndex = 0, 12 or 15
A: protocol.AType(dataFromContainer.A()),
B: protocol.BType(dataFromContainer.B()),
C: protocol.CType(dataFromContainer.C()),
D: protocol.DType(dataFromContainer.D()),
E: protocol.EType(dataFromContainer.E()),
})
}
containers = append(containers, protocol.Container{
ContainerId: protocol.ContainerIdType(containerFromStream.Id()),
ContainerDatas: datas,
})
}
var streamedMessage = protocol.StreamedMessage{
StreamId: protocol.SensorIdType(streamIdFromClient),
MessageId: protocol.MessageIdType(streamedDataFromClient.Id()),
Containers: containers,
}
return streamedMessage
}
服务器边界测试
这个测试通过,我们也有一个类似的测试,发送n 个随机数据而不是 1,它也通过了
package receptionFromGateway_test
import (
"/* some more imports */
"github.com/PhantomIntelligence/Server/test/utilities/clientStub"
"os"
"runtime"
"testing"
"time"
)
func TestFlatBuffersReceptionAndAccessFromClient(test *testing.T) {
defer os.RemoveAll("./logs")
test.Run("given_aStreamedDataMessageSentFromClientStub"+
"_when_receivedAndAccessedByServer"+
"_then_streamedDataMessageIntegrityIsConserved", func(subTest *testing.T) {
sentStreamedDataMessage := utilities.GenerateRandomStreamedDataMessage(16, 16) // 16 container, 16 data each
deserializer := serialization.NewFlatBufferDeserializationFilter()
pipe := dataFlow.NewPipe(deserializer)
procedure := dataFlowMock.NewProcedurePassToPipeThenSave(pipe)
pipeline := dataFlow.NewPipeline(procedure)
client := clientStub.NewWebSocketCommunicator() // <-- this calls `convertStreamMessageToFlatBuffers` written below
server := serving.NewServer()
server.Router.Mediator.Pipeline = pipeline
go server.Serve(":3591")
runtime.Gosched()
time.Sleep(50 * time.Millisecond)
client.Start()
client.Send(sentStreamedDataMessage)
runtime.Gosched()
time.Sleep(50 * time.Millisecond)
client.Stop()
pipeline.GetProducingPipe().TerminateProcess()
var receivedStreamedDataMessage = pipeline.GetProducingPipe().Filter.(*dataFlowMock.FilterSaveObjectReceived).ObjectReceived
utilities.AssertEqual(subTest, receivedStreamedDataMessage, sentStreamedDataMessage)
})
}
Server ClientStub的序列化函数
package client
Stub
import (
"github.com/PhantomIntelligence/Server/dataAccess/conversion/flatBuffers"
"github.com/PhantomIntelligence/Server/domain/protocol"
"github.com/PhantomIntelligence/Server/lib/Protocol"
"github.com/PhantomIntelligence/Server/lib/Protocol/Stream"
"github.com/google/flatbuffers/go"
)
const (
streamedDataMessageInitialSize = 2560
)
func convertStreamMessageToFlatBuffers(message protocol.StreamMessage) []byte {
builder := flatbuffers.NewBuilder(streamedDataMessageInitialSize)
var streamIdOffset = builder.CreateString(string(message.StreamId))
var headerOffset = createFlatBufferHeaders(builder, streamIdOffset)
var numberOfContainers int
var containerOffsets = make(map[int]flatbuffers.UOffsetT)
for containerIndex, container := range message.Containers {
var numberOfData int
var containerDataOffsets = make(map[int]flatbuffers.UOffsetT)
for containerIndex, container := range container.ContainerDatas {
Stream.ContainerDataStart(builder)
Stream.ContainerDataAddId(builder, uint16(container.ContainerDataId))
Stream.ContainerDataAddA(builder, uint16(container.A))
Stream.ContainerDataAddB(builder, uint16(container.B))
Stream.ContainerDataAddC(builder, int16(container.C))
Stream.ContainerDataAddD(builder, int16(container.D))
Stream.ContainerDataAddE(builder, byte(container.E))
containerDataOffset := Stream.ContainerDataEnd(builder)
containerDataOffsets[containerIndex] = containerDataOffset
numberOfData = containerIndex + 1
}
Stream.ContainerStartDataVector(builder, numberOfData)
// FlatBuffer UOffsetT vector requires to be filled by pre-pending backward the offsets
for dataOffsetIndex := numberOfData - 1; dataOffsetIndex >= 0; dataOffsetIndex-- {
builder.PrependUOffsetT(containerDataOffsets[dataOffsetIndex])
}
var dataOffsetVector = builder.EndVector(numberOfData)
Stream.ContainerStart(builder)
Stream.ContainerAddId(builder, uint16(container.ContainerId))
Stream.ContainerAddData(builder, dataOffsetVector)
containerOffset := Stream.ContainerEnd(builder)
containerOffsets[containerIndex] = containerOffset
numberOfContainers = containerIndex + 1
}
Stream.StreamedDataStartContainersVector(builder, numberOfContainers)
// FlatBuffer UOffsetT vector requires to be filled by pre-pending backward the offsets
for containerOffsetIndex := numberOfContainers - 1; containerOffsetIndex >= 0; containerOffsetIndex-- {
builder.PrependUOffsetT(containerOffsets[containerOffsetIndex])
}
var containerOffsetVector = builder.EndVector(numberOfContainers)
Stream.StreamedDataStart(builder)
Stream.StreamedDataAddId(builder, uint16(message.MessageId))
Stream.StreamedDataAddContainers(builder, containerOffsetVector)
var streamedDataMessageOffset = Stream.StreamedDataEnd(builder)
Stream.PayloadStart(builder)
Stream.PayloadAddSensorId(builder, streamIdOffset)
Stream.PayloadAddContentType(builder, Stream.PayloadContentStreamedData)
Stream.PayloadAddContent(builder, streamedDataMessageOffset)
var streamPayloadOffset = Stream.PayloadEnd(builder)
Protocol.MessageStart(builder)
Protocol.MessageAddHeader(builder, headerOffset)
Protocol.MessageAddContentType(builder, Protocol.ContentStream_Payload)
Protocol.MessageAddContent(builder, streamPayloadOffset)
clientStreamMessageOffset := Protocol.MessageEnd(builder)
builder.Finish(clientStreamMessageOffset)
flatBuffersBytes := builder.Bytes[builder.Head():]
return flatBuffersBytes
}
func createFlatBufferHeaders(builder *flatbuffers.Builder, sensorIdOffset flatbuffers.UOffsetT) flatbuffers.UOffsetT {
destinationIdOffset := builder.CreateString("Server")
offset := flatBuffers.CreateHeaderOffset(builder, destinationIdOffset, sensorIdOffset)
return offset
}
解决方案
不确定这是否是整个问题,但快速浏览一下convertStreamMessage
就会发现一个主要问题:这FlatBufferBuilder
是一个超出范围的局部变量,就像您返回一个指向其正在被破坏的内部数据的原始指针一样。
您要么想确保FlatBufferBuilder
数据的发送比发送数据的寿命长,要么FlatBufferBuilder::Release()
改为用作返回值。
另请注意,在调试这些问题时,您可以使用 C++ 验证程序(或任何读取数据的代码)在发送数据之前检查数据,这会更早发现这个问题。
推荐阅读
- r - 如何摆脱`deparse(substitute())`中的数据框名称?
- javascript - 如何模拟条码扫描事件
- javascript - 如何使用 MSAL acquireTokenRedirect() 获取令牌?
- python - 特定日期的 Django DATETIME_FORMAT 不正确(按一年)
- c# - AuthorizationHeader 不再显示在招摇文档中?
- javascript - Reactjs 通过页面和组件发送数据
- python - 记录和导出 conf 到其他模块
- css - Menu.Item 中的 Antd 自动换行
- python - 如何修复“由于 EnvironmentError 无法安装软件包:[WinError 123]”?
- python - python:创建一个读取无效 YAML 文件的测试