Martino Ferrari
0ae35d11ff
Add a FileWriter-compatible per-source binary recorder: native-type encode and TypeDescriptor mapping, FileWriter header serialization, subset/quantized/ ACCUMULATE row serialization, and push-thread double-buffer flush with size-cap rotation, keep-N pruning, fdatasync cadence, disk-free and staging-overflow guards. Covered by 12 GTests; full suite (83) green. Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
440 lines
15 KiB
C++
440 lines
15 KiB
C++
/**
|
|
* @file BinaryRecorderGTest.cpp
|
|
* @brief Unit tests for the StreamHub BinaryRecorder.
|
|
*
|
|
* @copyright Copyright 2015 F4E | European Joint Undertaking for ITER and
|
|
* the Development of Fusion Energy ('Fusion for Energy').
|
|
* Licensed under the EUPL, Version 1.1 or - as soon they will be approved
|
|
* by the European Commission - subsequent versions of the EUPL (the "Licence")
|
|
* You may not use this work except in compliance with the Licence.
|
|
* You may obtain a copy of the Licence at: http://ec.europa.eu/idabc/eupl
|
|
*
|
|
* @warning Unless required by applicable law or agreed to in writing,
|
|
* software distributed under the Licence is distributed on an "AS IS"
|
|
* basis, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express
|
|
* or implied. See the Licence permissions and limitations under the Licence.
|
|
*/
|
|
|
|
#include "BinaryRecorder.h"
|
|
#include "TypeDescriptor.h"
|
|
#include "UDPSProtocol.h"
|
|
|
|
#include <gtest/gtest.h>
|
|
#include <cstring>
|
|
#include <cstdio>
|
|
#include <cstdlib>
|
|
#include <cmath>
|
|
#include <vector>
|
|
#include <string>
|
|
#include <dirent.h>
|
|
#include <unistd.h>
|
|
|
|
using StreamHub::BinaryRecorder;
|
|
using StreamHub::RecorderConfig;
|
|
using MARTe::uint8;
|
|
using MARTe::uint16;
|
|
using MARTe::uint32;
|
|
using MARTe::uint64;
|
|
using MARTe::float64;
|
|
|
|
/*---------------------------------------------------------------------------*/
|
|
/* Helpers */
|
|
/*---------------------------------------------------------------------------*/
|
|
|
|
static MARTe::UDPSSignalDescriptor MakeDesc(const char *name, uint8 tc,
|
|
uint32 nElems, uint8 quant = 0u,
|
|
double rmin = 0.0, double rmax = 0.0) {
|
|
MARTe::UDPSSignalDescriptor d;
|
|
memset(&d, 0, sizeof(d));
|
|
strncpy(d.name, name, MARTe::UDPS_MAX_SIGNAL_NAME - 1u);
|
|
d.typeCode = tc;
|
|
d.quantType = quant;
|
|
d.numDimensions = (nElems > 1u) ? 1u : 0u;
|
|
d.numRows = nElems;
|
|
d.numCols = 1u;
|
|
d.rangeMin = rmin;
|
|
d.rangeMax = rmax;
|
|
d.timeMode = MARTe::UDPS_TIMEMODE_PACKET;
|
|
d.timeSignalIdx = MARTe::UDPS_NO_TIME_SIGNAL;
|
|
return d;
|
|
}
|
|
|
|
static RecorderConfig MakeCfg(const char *dir) {
|
|
RecorderConfig c;
|
|
memset(&c, 0, sizeof(c));
|
|
c.enabled = true;
|
|
c.autoStart = true;
|
|
strncpy(c.directory, dir, sizeof(c.directory) - 1u);
|
|
c.maxFileBytes = 1024u * 1024u;
|
|
c.keepFiles = 8u;
|
|
c.stagingBytes = 1024u * 1024u;
|
|
c.flushIntervalSec = 5u;
|
|
c.minDiskFreeMB = 0u;
|
|
strncpy(c.signals, "all", sizeof(c.signals) - 1u);
|
|
return c;
|
|
}
|
|
|
|
static std::string MakeTempDir() {
|
|
char templ[] = "/tmp/streamhub_rec_XXXXXX";
|
|
char *p = mkdtemp(templ);
|
|
return std::string(p != NULL ? p : "/tmp");
|
|
}
|
|
|
|
static std::vector<uint8> ReadFileBytes(const std::string &path) {
|
|
std::vector<uint8> out;
|
|
FILE *f = fopen(path.c_str(), "rb");
|
|
if (f == NULL) { return out; }
|
|
fseek(f, 0, SEEK_END);
|
|
long n = ftell(f);
|
|
fseek(f, 0, SEEK_SET);
|
|
if (n > 0) {
|
|
out.resize(static_cast<size_t>(n));
|
|
size_t rd = fread(&out[0], 1u, static_cast<size_t>(n), f);
|
|
out.resize(rd);
|
|
}
|
|
fclose(f);
|
|
return out;
|
|
}
|
|
|
|
static std::vector<std::string> ListBinFiles(const std::string &dir) {
|
|
std::vector<std::string> out;
|
|
DIR *d = opendir(dir.c_str());
|
|
if (d == NULL) { return out; }
|
|
struct dirent *e;
|
|
while ((e = readdir(d)) != NULL) {
|
|
std::string nm = e->d_name;
|
|
if (nm.size() > 4u && nm.substr(nm.size() - 4u) == ".bin") {
|
|
out.push_back(dir + "/" + nm);
|
|
}
|
|
}
|
|
closedir(d);
|
|
return out;
|
|
}
|
|
|
|
static uint16 ReadU16(const std::vector<uint8> &b, size_t off) {
|
|
uint16 v = 0u; memcpy(&v, &b[off], 2u); return v;
|
|
}
|
|
static uint32 ReadU32(const std::vector<uint8> &b, size_t off) {
|
|
uint32 v = 0u; memcpy(&v, &b[off], 4u); return v;
|
|
}
|
|
|
|
/*---------------------------------------------------------------------------*/
|
|
/* Static helper tests */
|
|
/*---------------------------------------------------------------------------*/
|
|
|
|
TEST(BinaryRecorderGTest, TypeCodeToDescriptorAll) {
|
|
EXPECT_EQ(MARTe::UnsignedInteger32Bit.all,
|
|
BinaryRecorder::TypeCodeToDescriptorAll(MARTe::UDPS_TYPECODE_UINT32));
|
|
EXPECT_EQ(MARTe::Float32Bit.all,
|
|
BinaryRecorder::TypeCodeToDescriptorAll(MARTe::UDPS_TYPECODE_FLOAT32));
|
|
EXPECT_EQ(MARTe::Float64Bit.all,
|
|
BinaryRecorder::TypeCodeToDescriptorAll(MARTe::UDPS_TYPECODE_FLOAT64));
|
|
EXPECT_EQ(MARTe::UnsignedInteger64Bit.all,
|
|
BinaryRecorder::TypeCodeToDescriptorAll(MARTe::UDPS_TYPECODE_UINT64));
|
|
EXPECT_EQ(MARTe::SignedInteger16Bit.all,
|
|
BinaryRecorder::TypeCodeToDescriptorAll(MARTe::UDPS_TYPECODE_INT16));
|
|
}
|
|
|
|
TEST(BinaryRecorderGTest, EncodeNativeFloat32) {
|
|
uint8 buf[8] = {0};
|
|
uint32 n = BinaryRecorder::EncodeNative(MARTe::UDPS_TYPECODE_FLOAT32, 1.5, buf);
|
|
ASSERT_EQ(4u, n);
|
|
float expect = 1.5f;
|
|
EXPECT_EQ(0, memcmp(buf, &expect, 4u));
|
|
}
|
|
|
|
TEST(BinaryRecorderGTest, EncodeNativeUint32) {
|
|
uint8 buf[8] = {0};
|
|
uint32 n = BinaryRecorder::EncodeNative(MARTe::UDPS_TYPECODE_UINT32, 7.0, buf);
|
|
ASSERT_EQ(4u, n);
|
|
uint32 expect = 7u;
|
|
EXPECT_EQ(0, memcmp(buf, &expect, 4u));
|
|
}
|
|
|
|
TEST(BinaryRecorderGTest, EncodeNativeInt16) {
|
|
uint8 buf[8] = {0};
|
|
uint32 n = BinaryRecorder::EncodeNative(MARTe::UDPS_TYPECODE_INT16, -3.0, buf);
|
|
ASSERT_EQ(2u, n);
|
|
MARTe::int16 expect = -3;
|
|
EXPECT_EQ(0, memcmp(buf, &expect, 2u));
|
|
}
|
|
|
|
/*---------------------------------------------------------------------------*/
|
|
/* Header bytes */
|
|
/*---------------------------------------------------------------------------*/
|
|
|
|
TEST(BinaryRecorderGTest, HeaderBytes) {
|
|
std::string dir = MakeTempDir();
|
|
RecorderConfig cfg = MakeCfg(dir.c_str());
|
|
BinaryRecorder rec;
|
|
rec.Init(cfg, "S0");
|
|
|
|
MARTe::UDPSSignalDescriptor descs[2] = {
|
|
MakeDesc("Sine", MARTe::UDPS_TYPECODE_FLOAT32, 1u),
|
|
MakeDesc("Time", MARTe::UDPS_TYPECODE_UINT32, 4u)
|
|
};
|
|
bool mask[2] = { true, true };
|
|
rec.Configure(descs, 2u, mask);
|
|
rec.FlushTick(0u);
|
|
|
|
bool recording; char path[768]; uint64 bw, rw, dr, fm;
|
|
rec.GetInfo(recording, path, sizeof(path), bw, rw, dr, fm);
|
|
std::vector<uint8> b = ReadFileBytes(path);
|
|
ASSERT_GE(b.size(), 4u + 2u * 38u);
|
|
|
|
EXPECT_EQ(2u, ReadU32(b, 0));
|
|
/* descriptor 0 */
|
|
EXPECT_EQ(MARTe::Float32Bit.all, ReadU16(b, 4));
|
|
EXPECT_STREQ("Sine", reinterpret_cast<const char *>(&b[6]));
|
|
EXPECT_EQ(1u, ReadU32(b, 38));
|
|
/* descriptor 1 */
|
|
EXPECT_EQ(MARTe::UnsignedInteger32Bit.all, ReadU16(b, 42));
|
|
EXPECT_STREQ("Time", reinterpret_cast<const char *>(&b[44]));
|
|
EXPECT_EQ(4u, ReadU32(b, 76));
|
|
}
|
|
|
|
/*---------------------------------------------------------------------------*/
|
|
/* Row serialization */
|
|
/*---------------------------------------------------------------------------*/
|
|
|
|
TEST(BinaryRecorderGTest, RowStrictUnquantized) {
|
|
std::string dir = MakeTempDir();
|
|
BinaryRecorder rec;
|
|
rec.Init(MakeCfg(dir.c_str()), "S0");
|
|
|
|
MARTe::UDPSSignalDescriptor descs[2] = {
|
|
MakeDesc("A", MARTe::UDPS_TYPECODE_FLOAT32, 1u),
|
|
MakeDesc("B", MARTe::UDPS_TYPECODE_UINT32, 2u)
|
|
};
|
|
bool mask[2] = { true, true };
|
|
rec.Configure(descs, 2u, mask);
|
|
|
|
uint8 payload[12];
|
|
float a = 2.5f; uint32 b0 = 10u, b1 = 20u;
|
|
memcpy(payload + 0, &a, 4u);
|
|
memcpy(payload + 4, &b0, 4u);
|
|
memcpy(payload + 8, &b1, 4u);
|
|
uint32 sigOff[2] = { 0u, 4u };
|
|
uint32 sigElems[2] = { 1u, 2u };
|
|
rec.CapturePacket(payload, sigOff, sigElems, MARTe::UDPS_PUBLISH_STRICT, 1u);
|
|
rec.FlushTick(0u);
|
|
|
|
bool recording; char path[768]; uint64 bw, rw, dr, fm;
|
|
rec.GetInfo(recording, path, sizeof(path), bw, rw, dr, fm);
|
|
std::vector<uint8> b = ReadFileBytes(path);
|
|
const size_t hdr = 4u + 2u * 38u;
|
|
ASSERT_EQ(hdr + 12u, b.size());
|
|
float ra; uint32 rb0, rb1;
|
|
memcpy(&ra, &b[hdr + 0], 4u);
|
|
memcpy(&rb0, &b[hdr + 4], 4u);
|
|
memcpy(&rb1, &b[hdr + 8], 4u);
|
|
EXPECT_FLOAT_EQ(2.5f, ra);
|
|
EXPECT_EQ(10u, rb0);
|
|
EXPECT_EQ(20u, rb1);
|
|
EXPECT_EQ(1u, rw);
|
|
}
|
|
|
|
TEST(BinaryRecorderGTest, RowAccumulateExpansion) {
|
|
std::string dir = MakeTempDir();
|
|
BinaryRecorder rec;
|
|
rec.Init(MakeCfg(dir.c_str()), "S0");
|
|
|
|
MARTe::UDPSSignalDescriptor descs[2] = {
|
|
MakeDesc("S", MARTe::UDPS_TYPECODE_FLOAT32, 1u),
|
|
MakeDesc("C", MARTe::UDPS_TYPECODE_UINT32, 1u)
|
|
};
|
|
bool mask[2] = { true, true };
|
|
rec.Configure(descs, 2u, mask);
|
|
|
|
/* Wire: S accumulated x3 (3 floats), C scalar (1 uint32). */
|
|
uint8 payload[16];
|
|
float s[3] = { 1.0f, 2.0f, 3.0f };
|
|
uint32 c = 9u;
|
|
memcpy(payload + 0, s, 12u);
|
|
memcpy(payload + 12, &c, 4u);
|
|
uint32 sigOff[2] = { 0u, 12u };
|
|
uint32 sigElems[2] = { 3u, 1u };
|
|
rec.CapturePacket(payload, sigOff, sigElems, MARTe::UDPS_PUBLISH_ACCUMULATE, 3u);
|
|
rec.FlushTick(0u);
|
|
|
|
bool recording; char path[768]; uint64 bw, rw, dr, fm;
|
|
rec.GetInfo(recording, path, sizeof(path), bw, rw, dr, fm);
|
|
std::vector<uint8> b = ReadFileBytes(path);
|
|
const size_t hdr = 4u + 2u * 38u;
|
|
const size_t rowBytes = 8u;
|
|
ASSERT_EQ(hdr + 3u * rowBytes, b.size());
|
|
for (uint32 r = 0u; r < 3u; r++) {
|
|
float sr; uint32 cr;
|
|
memcpy(&sr, &b[hdr + r * rowBytes + 0], 4u);
|
|
memcpy(&cr, &b[hdr + r * rowBytes + 4], 4u);
|
|
EXPECT_FLOAT_EQ(static_cast<float>(r + 1u), sr);
|
|
EXPECT_EQ(9u, cr);
|
|
}
|
|
EXPECT_EQ(3u, rw);
|
|
}
|
|
|
|
TEST(BinaryRecorderGTest, SubsetSelection) {
|
|
std::string dir = MakeTempDir();
|
|
BinaryRecorder rec;
|
|
rec.Init(MakeCfg(dir.c_str()), "S0");
|
|
|
|
MARTe::UDPSSignalDescriptor descs[3] = {
|
|
MakeDesc("A", MARTe::UDPS_TYPECODE_FLOAT32, 1u),
|
|
MakeDesc("B", MARTe::UDPS_TYPECODE_FLOAT32, 1u),
|
|
MakeDesc("C", MARTe::UDPS_TYPECODE_UINT32, 1u)
|
|
};
|
|
bool mask[3] = { true, false, true };
|
|
rec.Configure(descs, 3u, mask);
|
|
|
|
uint8 payload[12];
|
|
float a = 1.0f, bb = 2.0f; uint32 c = 3u;
|
|
memcpy(payload + 0, &a, 4u);
|
|
memcpy(payload + 4, &bb, 4u);
|
|
memcpy(payload + 8, &c, 4u);
|
|
uint32 sigOff[3] = { 0u, 4u, 8u };
|
|
uint32 sigElems[3] = { 1u, 1u, 1u };
|
|
rec.CapturePacket(payload, sigOff, sigElems, MARTe::UDPS_PUBLISH_STRICT, 1u);
|
|
rec.FlushTick(0u);
|
|
|
|
bool recording; char path[768]; uint64 bw, rw, dr, fm;
|
|
rec.GetInfo(recording, path, sizeof(path), bw, rw, dr, fm);
|
|
std::vector<uint8> b = ReadFileBytes(path);
|
|
/* header has 2 signals, A and C */
|
|
EXPECT_EQ(2u, ReadU32(b, 0));
|
|
EXPECT_STREQ("A", reinterpret_cast<const char *>(&b[6]));
|
|
EXPECT_STREQ("C", reinterpret_cast<const char *>(&b[44]));
|
|
const size_t hdr = 4u + 2u * 38u;
|
|
ASSERT_EQ(hdr + 8u, b.size());
|
|
float ra; uint32 rc;
|
|
memcpy(&ra, &b[hdr + 0], 4u);
|
|
memcpy(&rc, &b[hdr + 4], 4u);
|
|
EXPECT_FLOAT_EQ(1.0f, ra);
|
|
EXPECT_EQ(3u, rc);
|
|
}
|
|
|
|
TEST(BinaryRecorderGTest, QuantizedReconstruction) {
|
|
std::string dir = MakeTempDir();
|
|
BinaryRecorder rec;
|
|
rec.Init(MakeCfg(dir.c_str()), "S0");
|
|
|
|
/* float32 signal quantized to uint16 over [0,10]. */
|
|
MARTe::UDPSSignalDescriptor descs[1] = {
|
|
MakeDesc("Q", MARTe::UDPS_TYPECODE_FLOAT32, 1u,
|
|
MARTe::UDPS_QUANT_UINT16, 0.0, 10.0)
|
|
};
|
|
bool mask[1] = { true };
|
|
rec.Configure(descs, 1u, mask);
|
|
|
|
uint8 payload[2];
|
|
uint16 q = 32767u;
|
|
memcpy(payload, &q, 2u);
|
|
uint32 sigOff[1] = { 0u };
|
|
uint32 sigElems[1] = { 1u };
|
|
rec.CapturePacket(payload, sigOff, sigElems, MARTe::UDPS_PUBLISH_STRICT, 1u);
|
|
rec.FlushTick(0u);
|
|
|
|
bool recording; char path[768]; uint64 bw, rw, dr, fm;
|
|
rec.GetInfo(recording, path, sizeof(path), bw, rw, dr, fm);
|
|
std::vector<uint8> b = ReadFileBytes(path);
|
|
const size_t hdr = 4u + 38u;
|
|
ASSERT_EQ(hdr + 4u, b.size());
|
|
float rv;
|
|
memcpy(&rv, &b[hdr], 4u);
|
|
double expect = 0.0 + (32767.0 / 65535.0) * 10.0;
|
|
EXPECT_NEAR(expect, static_cast<double>(rv), 1e-3);
|
|
}
|
|
|
|
/*---------------------------------------------------------------------------*/
|
|
/* Flush / rotation / guards */
|
|
/*---------------------------------------------------------------------------*/
|
|
|
|
TEST(BinaryRecorderGTest, RotationKeepN) {
|
|
std::string dir = MakeTempDir();
|
|
RecorderConfig cfg = MakeCfg(dir.c_str());
|
|
cfg.maxFileBytes = 200u;
|
|
cfg.keepFiles = 2u;
|
|
BinaryRecorder rec;
|
|
rec.Init(cfg, "S0");
|
|
|
|
MARTe::UDPSSignalDescriptor descs[2] = {
|
|
MakeDesc("A", MARTe::UDPS_TYPECODE_FLOAT32, 1u),
|
|
MakeDesc("B", MARTe::UDPS_TYPECODE_UINT32, 2u)
|
|
};
|
|
bool mask[2] = { true, true };
|
|
rec.Configure(descs, 2u, mask);
|
|
|
|
uint8 payload[12];
|
|
float a = 1.0f; uint32 b0 = 1u, b1 = 2u;
|
|
memcpy(payload + 0, &a, 4u);
|
|
memcpy(payload + 4, &b0, 4u);
|
|
memcpy(payload + 8, &b1, 4u);
|
|
uint32 sigOff[2] = { 0u, 4u };
|
|
uint32 sigElems[2] = { 1u, 2u };
|
|
|
|
for (uint32 i = 0u; i < 50u; i++) {
|
|
rec.CapturePacket(payload, sigOff, sigElems, MARTe::UDPS_PUBLISH_STRICT, 1u);
|
|
rec.FlushTick(0u);
|
|
}
|
|
|
|
std::vector<std::string> files = ListBinFiles(dir);
|
|
EXPECT_LE(files.size(), 2u);
|
|
EXPECT_GE(files.size(), 1u);
|
|
for (size_t i = 0u; i < files.size(); i++) {
|
|
std::vector<uint8> b = ReadFileBytes(files[i]);
|
|
ASSERT_GE(b.size(), 4u);
|
|
EXPECT_EQ(2u, ReadU32(b, 0));
|
|
}
|
|
}
|
|
|
|
TEST(BinaryRecorderGTest, DiskGuardStops) {
|
|
std::string dir = MakeTempDir();
|
|
RecorderConfig cfg = MakeCfg(dir.c_str());
|
|
cfg.minDiskFreeMB = 0xFFFFFFFFu; /* impossibly large → always low */
|
|
BinaryRecorder rec;
|
|
rec.Init(cfg, "S0");
|
|
|
|
MARTe::UDPSSignalDescriptor descs[1] = {
|
|
MakeDesc("A", MARTe::UDPS_TYPECODE_FLOAT32, 1u)
|
|
};
|
|
bool mask[1] = { true };
|
|
rec.Configure(descs, 1u, mask);
|
|
rec.FlushTick(0u);
|
|
|
|
bool recording; char path[768]; uint64 bw, rw, dr, fm;
|
|
rec.GetInfo(recording, path, sizeof(path), bw, rw, dr, fm);
|
|
EXPECT_FALSE(recording);
|
|
EXPECT_EQ(0u, ListBinFiles(dir).size());
|
|
}
|
|
|
|
TEST(BinaryRecorderGTest, OverflowDropsRows) {
|
|
std::string dir = MakeTempDir();
|
|
RecorderConfig cfg = MakeCfg(dir.c_str());
|
|
cfg.stagingBytes = 4096u; /* small staging, no flush → overflow */
|
|
BinaryRecorder rec;
|
|
rec.Init(cfg, "S0");
|
|
|
|
MARTe::UDPSSignalDescriptor descs[2] = {
|
|
MakeDesc("A", MARTe::UDPS_TYPECODE_FLOAT32, 1u),
|
|
MakeDesc("B", MARTe::UDPS_TYPECODE_UINT32, 2u)
|
|
};
|
|
bool mask[2] = { true, true };
|
|
rec.Configure(descs, 2u, mask);
|
|
|
|
uint8 payload[12];
|
|
float a = 1.0f; uint32 b0 = 1u, b1 = 2u;
|
|
memcpy(payload + 0, &a, 4u);
|
|
memcpy(payload + 4, &b0, 4u);
|
|
memcpy(payload + 8, &b1, 4u);
|
|
uint32 sigOff[2] = { 0u, 4u };
|
|
uint32 sigElems[2] = { 1u, 2u };
|
|
|
|
/* Capture far more rows than the 4096-byte buffer can hold (no flush). */
|
|
for (uint32 i = 0u; i < 500u; i++) {
|
|
rec.CapturePacket(payload, sigOff, sigElems, MARTe::UDPS_PUBLISH_STRICT, 1u);
|
|
}
|
|
|
|
bool recording; char path[768]; uint64 bw, rw, dr, fm;
|
|
rec.GetInfo(recording, path, sizeof(path), bw, rw, dr, fm);
|
|
EXPECT_GT(dr, 0u);
|
|
}
|