问题
尝试访问缓冲区的一部分时，flatbuffers 生成的服务器文件反复发生崩溃( panic :运行时错误: slice 边界超出范围)(使用 flatbuffers 创建)，其中包含从一个客户端流式传输到服务器的消息。
这个问题只出现在客户端和服务器端集成的情况下。单独测试时，客户端和服务器都成功使用flatbuffers，并且服务器没有发生崩溃//请参阅下面的边界测试
知道:
- 发送和接收的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
导致崩溃的数据
当服务器尝试访问它以使用它时，崩溃总是发生在服务器收到的第一个缓冲区上。更具体地说，当尝试访问第 7 个 Container 的(第 0 个、第 12 个或第 15 个)ContainerData 时，总会发生这种情况。
Know that for this use case, we send a StreamedData with 16 Containers, each holding 16 ContainerData
代码详细信息
Note that only the structure is remains from my original implementation. I am not allowed to divulgate the nature of the data we manipulate. Thus the object names might seem weird, but the structure is the relevant part.
架构编译
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
架构
客户端和服务器都使用以下架构:
Please note that this extract is not the complete schema, only the part relevant for this question.
// 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.Stream.Payload的Protocol.Message，而这些Protocol.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
客户端的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)
})
}
服务器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++ 验证程序(或任何读取数据的代码)在发送数据之前检查数据，这样可以更早地捕获此问题。
关于c++ - FlatBuffers 的多语言集成问题，我们在Stack Overflow上找到一个类似的问题： https://stackoverflow.com/questions/53181435/