diff --git a/Source/Applications/StreamHub/BinaryRecorder.cpp b/Source/Applications/StreamHub/BinaryRecorder.cpp new file mode 100644 index 0000000..e207c8d --- /dev/null +++ b/Source/Applications/StreamHub/BinaryRecorder.cpp @@ -0,0 +1,565 @@ +/** + * @file BinaryRecorder.cpp + * @brief Per-source binary recorder implementation. + * + * @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 "AdvancedErrorManagement.h" +#include "TypeDescriptor.h" + +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + +namespace StreamHub { + +/*---------------------------------------------------------------------------*/ +/* Static helpers */ +/*---------------------------------------------------------------------------*/ + +uint16 BinaryRecorder::TypeCodeToDescriptorAll(uint8 udpsTypeCode) { + switch (udpsTypeCode) { + case MARTe::UDPS_TYPECODE_UINT8: return MARTe::UnsignedInteger8Bit.all; + case MARTe::UDPS_TYPECODE_INT8: return MARTe::SignedInteger8Bit.all; + case MARTe::UDPS_TYPECODE_UINT16: return MARTe::UnsignedInteger16Bit.all; + case MARTe::UDPS_TYPECODE_INT16: return MARTe::SignedInteger16Bit.all; + case MARTe::UDPS_TYPECODE_UINT32: return MARTe::UnsignedInteger32Bit.all; + case MARTe::UDPS_TYPECODE_INT32: return MARTe::SignedInteger32Bit.all; + case MARTe::UDPS_TYPECODE_UINT64: return MARTe::UnsignedInteger64Bit.all; + case MARTe::UDPS_TYPECODE_INT64: return MARTe::SignedInteger64Bit.all; + case MARTe::UDPS_TYPECODE_FLOAT32: return MARTe::Float32Bit.all; + case MARTe::UDPS_TYPECODE_FLOAT64: return MARTe::Float64Bit.all; + default: return MARTe::Float64Bit.all; + } +} + +uint32 BinaryRecorder::EncodeNative(uint8 typeCode, float64 v, uint8 *dst) { + switch (typeCode) { + case MARTe::UDPS_TYPECODE_UINT8: { + uint8 x = static_cast(v); memcpy(dst, &x, 1u); return 1u; + } + case MARTe::UDPS_TYPECODE_INT8: { + MARTe::int8 x = static_cast(v); memcpy(dst, &x, 1u); return 1u; + } + case MARTe::UDPS_TYPECODE_UINT16: { + uint16 x = static_cast(v); memcpy(dst, &x, 2u); return 2u; + } + case MARTe::UDPS_TYPECODE_INT16: { + MARTe::int16 x = static_cast(v); memcpy(dst, &x, 2u); return 2u; + } + case MARTe::UDPS_TYPECODE_UINT32: { + uint32 x = static_cast(v); memcpy(dst, &x, 4u); return 4u; + } + case MARTe::UDPS_TYPECODE_INT32: { + MARTe::int32 x = static_cast(v); memcpy(dst, &x, 4u); return 4u; + } + case MARTe::UDPS_TYPECODE_UINT64: { + uint64 x = static_cast(v); memcpy(dst, &x, 8u); return 8u; + } + case MARTe::UDPS_TYPECODE_INT64: { + MARTe::int64 x = static_cast(v); memcpy(dst, &x, 8u); return 8u; + } + case MARTe::UDPS_TYPECODE_FLOAT32: { + float x = static_cast(v); memcpy(dst, &x, 4u); return 4u; + } + case MARTe::UDPS_TYPECODE_FLOAT64: { + float64 x = v; memcpy(dst, &x, 8u); return 8u; + } + default: return 0u; + } +} + +/** Local dequantization mirror of UDPSourceSession::DequantizeValue. */ +static float64 RecDequantize(float64 q, uint8 quantType, + float64 rangeMin, float64 rangeMax) { + const float64 range = rangeMax - rangeMin; + switch (quantType) { + case MARTe::UDPS_QUANT_UINT8: return rangeMin + (q / 255.0) * range; + case MARTe::UDPS_QUANT_INT8: return rangeMin + ((q + 127.0) / 254.0) * range; + case MARTe::UDPS_QUANT_UINT16: return rangeMin + (q / 65535.0) * range; + case MARTe::UDPS_QUANT_INT16: return rangeMin + ((q + 32767.0) / 65534.0) * range; + default: return q; + } +} + +/** Read one quantized wire element as a float64 integer value. */ +static float64 RecDecodeQuantWire(const uint8 *ptr, uint8 quantType) { + switch (quantType) { + case MARTe::UDPS_QUANT_UINT8: { + uint8 v = 0u; memcpy(&v, ptr, 1u); return static_cast(v); + } + case MARTe::UDPS_QUANT_INT8: { + MARTe::int8 v = 0; memcpy(&v, ptr, 1u); return static_cast(v); + } + case MARTe::UDPS_QUANT_UINT16: { + uint16 v = 0u; memcpy(&v, ptr, 2u); return static_cast(v); + } + case MARTe::UDPS_QUANT_INT16: { + MARTe::int16 v = 0; memcpy(&v, ptr, 2u); return static_cast(v); + } + default: return 0.0; + } +} + +static uint32 RecQuantWireBytes(uint8 quantType) { + switch (quantType) { + case MARTe::UDPS_QUANT_UINT8: return 1u; + case MARTe::UDPS_QUANT_INT8: return 1u; + case MARTe::UDPS_QUANT_UINT16: return 2u; + case MARTe::UDPS_QUANT_INT16: return 2u; + default: return 0u; + } +} + +/*---------------------------------------------------------------------------*/ +/* Constructor / Destructor */ +/*---------------------------------------------------------------------------*/ + +BinaryRecorder::BinaryRecorder() + : nIncluded_(0u), + rowBytes_(0u), + headerLen_(0u), + layoutReady_(false), + pushHeaderLen_(0u), + pushRowBytes_(0u), + front_(static_cast(0)), + back_(static_cast(0)), + frontLen_(0u), + backLen_(0u), + pendingArm_(false), + pendingDisarm_(false), + pendingReopen_(false), + fd_(-1), + fileOffset_(0u), + lastSyncSec_(0u), + fileSeq_(0u), + armed_(false), + diskLow_(false), + bytesWritten_(0u), + rowsWritten_(0u), + droppedRows_(0u) { + memset(&config_, 0, sizeof(config_)); + sourceId_[0] = '\0'; + curFilePath_[0] = '\0'; + stagingMutex_.Create(); +} + +BinaryRecorder::~BinaryRecorder() { + CloseFile(); + if (front_ != static_cast(0)) { delete[] front_; front_ = static_cast(0); } + if (back_ != static_cast(0)) { delete[] back_; back_ = static_cast(0); } +} + +/*---------------------------------------------------------------------------*/ +/* Init */ +/*---------------------------------------------------------------------------*/ + +void BinaryRecorder::Init(const RecorderConfig &cfg, const char *sourceId) { + config_ = cfg; + if (config_.stagingBytes < 4096u) { config_.stagingBytes = 4096u; } + strncpy(sourceId_, (sourceId != static_cast(0)) ? sourceId : "src", + sizeof(sourceId_) - 1u); + sourceId_[sizeof(sourceId_) - 1u] = '\0'; + + front_ = new uint8[config_.stagingBytes]; + back_ = new uint8[config_.stagingBytes]; + frontLen_ = 0u; + backLen_ = 0u; + + /* Create output directory (best-effort). */ + if (config_.directory[0] != '\0') { + (void) mkdir(config_.directory, 0775); + } + + if (config_.autoStart) { + pendingArm_ = true; + } +} + +/*---------------------------------------------------------------------------*/ +/* Configure (receive thread) */ +/*---------------------------------------------------------------------------*/ + +void BinaryRecorder::Configure(const MARTe::UDPSSignalDescriptor *descs, + uint32 nSigs, const bool *includeMask) { + uint32 inc = 0u; + for (uint32 s = 0u; (s < nSigs) && (inc < kRecMaxSig); s++) { + if ((includeMask != static_cast(0)) && (!includeMask[s])) { + continue; + } + const MARTe::UDPSSignalDescriptor &d = descs[s]; + uint32 nElems = d.numRows * d.numCols; + if (nElems == 0u) { nElems = 1u; } + const uint32 native = MARTe::UDPSTypeCodeByteSize(d.typeCode); + if (native == 0u) { continue; } + const uint32 wire = (d.quantType != MARTe::UDPS_QUANT_NONE) + ? RecQuantWireBytes(d.quantType) + : native; + srcIdx_[inc] = s; + typeCode_[inc] = d.typeCode; + quantType_[inc] = d.quantType; + declaredElems_[inc] = nElems; + nativeBytes_[inc] = native; + wireBytes_[inc] = wire; + rangeMin_[inc] = d.rangeMin; + rangeMax_[inc] = d.rangeMax; + inc++; + } + nIncluded_ = inc; + + /* Build the FileWriter header blob and compute the row size. */ + uint32 pos = 0u; + memcpy(&headerBlob_[pos], &nIncluded_, 4u); pos += 4u; + uint32 row = 0u; + for (uint32 i = 0u; i < nIncluded_; i++) { + const uint16 tcAll = TypeCodeToDescriptorAll(typeCode_[i]); + memcpy(&headerBlob_[pos], &tcAll, 2u); pos += 2u; + /* 32-byte name, null-padded, truncated from the descriptor name. */ + char name32[32]; + memset(name32, 0, sizeof(name32)); + strncpy(name32, descs[srcIdx_[i]].name, sizeof(name32) - 1u); + memcpy(&headerBlob_[pos], name32, 32u); pos += 32u; + memcpy(&headerBlob_[pos], &declaredElems_[i], 4u); pos += 4u; + row += declaredElems_[i] * nativeBytes_[i]; + } + headerLen_ = pos; + rowBytes_ = row; + + (void) stagingMutex_.FastLock(); + layoutReady_ = true; + pendingReopen_ = true; + /* Discard any rows staged under the previous layout: they would belong + * to the file being closed and must not be mixed into the new layout. */ + frontLen_ = 0u; + stagingMutex_.FastUnLock(); +} + +/*---------------------------------------------------------------------------*/ +/* CapturePacket (receive thread) */ +/*---------------------------------------------------------------------------*/ + +void BinaryRecorder::CapturePacket(const uint8 *payload, const uint32 *sigOff, + const uint32 *sigElems, uint8 publishMode, + uint32 numSamples) { + if (!layoutReady_ || (nIncluded_ == 0u) || (rowBytes_ == 0u)) { return; } + const bool accumulate = (publishMode == MARTe::UDPS_PUBLISH_ACCUMULATE); + uint32 numRows = accumulate ? numSamples : 1u; + if (numRows == 0u) { numRows = 1u; } + + const uint32 rowSpan = numRows * rowBytes_; + + (void) stagingMutex_.FastLock(); + if (frontLen_ + rowSpan > config_.stagingBytes) { + droppedRows_ += static_cast(numRows); + stagingMutex_.FastUnLock(); + return; + } + + for (uint32 r = 0u; r < numRows; r++) { + for (uint32 i = 0u; i < nIncluded_; i++) { + const uint32 s = srcIdx_[i]; + const uint8 *base = payload + sigOff[s]; + /* A per-sample (accumulated) scalar carries numSamples wire values; + * a plain scalar carries one value that is repeated across rows. */ + const bool isAccumScalar = accumulate && (declaredElems_[i] == 1u) && + (sigElems[s] > 1u); + const uint32 nElem = declaredElems_[i]; + + if (quantType_[i] == MARTe::UDPS_QUANT_NONE) { + if (isAccumScalar) { + /* Take wire element r for an accumulated scalar. */ + memcpy(&front_[frontLen_], base + r * wireBytes_[i], + nativeBytes_[i]); + frontLen_ += nativeBytes_[i]; + } + else { + /* Straight copy of all elements (wire == native size). */ + const uint32 bytes = nElem * nativeBytes_[i]; + memcpy(&front_[frontLen_], base, bytes); + frontLen_ += bytes; + } + } + else { + /* Quantized: dequantize + re-encode per element. */ + for (uint32 e = 0u; e < nElem; e++) { + const uint32 wireIdx = isAccumScalar ? r : e; + const float64 q = RecDecodeQuantWire( + base + wireIdx * wireBytes_[i], quantType_[i]); + const float64 phys = RecDequantize(q, quantType_[i], + rangeMin_[i], rangeMax_[i]); + frontLen_ += EncodeNative(typeCode_[i], phys, + &front_[frontLen_]); + } + } + } + } + stagingMutex_.FastUnLock(); +} + +/*---------------------------------------------------------------------------*/ +/* FlushTick (push thread) */ +/*---------------------------------------------------------------------------*/ + +void BinaryRecorder::FlushTick(uint32 nowSec) { + bool doArm = false; + bool doDisarm = false; + bool doReopen = false; + + (void) stagingMutex_.FastLock(); + doArm = pendingArm_; pendingArm_ = false; + doDisarm = pendingDisarm_; pendingDisarm_ = false; + doReopen = pendingReopen_; pendingReopen_ = false; + if (doReopen && layoutReady_) { + memcpy(pushHeader_, headerBlob_, headerLen_); + pushHeaderLen_ = headerLen_; + pushRowBytes_ = rowBytes_; + } + /* Swap front -> back. */ + uint8 *tmp = back_; + back_ = front_; + front_ = tmp; + backLen_ = frontLen_; + frontLen_ = 0u; + stagingMutex_.FastUnLock(); + + if (doArm) { armed_ = true; } + + if (doReopen) { + /* Layout changed: close the current file. The freshly swapped back + * buffer holds new-layout rows and is written to the new file below. */ + CloseFile(); + } + + if (doDisarm) { + if ((fd_ >= 0) && (backLen_ > 0u)) { + (void) pwrite(fd_, back_, backLen_, static_cast(fileOffset_)); + fileOffset_ += backLen_; + bytesWritten_ += backLen_; + } + backLen_ = 0u; + armed_ = false; + CloseFile(); + return; + } + + if (!armed_) { + backLen_ = 0u; + return; + } + + /* Disk-free guard. */ + if (!HasSufficientDisk()) { + if (!diskLow_) { + diskLow_ = true; + REPORT_ERROR_STATIC(MARTe::ErrorManagement::Warning, + "BinaryRecorder: free disk below MinDiskFreeMB; recording stopped"); + } + armed_ = false; + backLen_ = 0u; + CloseFile(); + return; + } + diskLow_ = false; + + /* Open a file if none is active. */ + if ((fd_ < 0) && (pushHeaderLen_ > 0u)) { + OpenNewFile(); + } + if (fd_ < 0) { + backLen_ = 0u; + return; + } + + /* Flush staged rows. */ + if (backLen_ > 0u) { + const ssize_t w = pwrite(fd_, back_, backLen_, + static_cast(fileOffset_)); + if (w > 0) { + fileOffset_ += static_cast(w); + bytesWritten_ += static_cast(w); + if (pushRowBytes_ > 0u) { + rowsWritten_ += static_cast(w) / pushRowBytes_; + } + } + backLen_ = 0u; + } + + /* Rotate on size cap. */ + if (fileOffset_ >= static_cast(config_.maxFileBytes)) { + CloseFile(); + OpenNewFile(); + PruneOldFiles(); + } + + /* Periodic durability. */ + if ((fd_ >= 0) && + ((nowSec - lastSyncSec_) >= config_.flushIntervalSec)) { + (void) fdatasync(fd_); + lastSyncSec_ = nowSec; + } +} + +/*---------------------------------------------------------------------------*/ +/* File management (push thread) */ +/*---------------------------------------------------------------------------*/ + +void BinaryRecorder::OpenNewFile() { + struct timeval tv; + (void) gettimeofday(&tv, static_cast(0)); + struct tm tmv; + time_t secs = static_cast(tv.tv_sec); + (void) gmtime_r(&secs, &tmv); + char stamp[32]; + (void) strftime(stamp, sizeof(stamp), "%Y%m%dT%H%M%S", &tmv); + + fileSeq_++; + (void) snprintf(curFilePath_, sizeof(curFilePath_), + "%s/%s_%s_%08u.bin", + config_.directory, sourceId_, stamp, fileSeq_); + + fd_ = open(curFilePath_, O_WRONLY | O_CREAT | O_TRUNC, 0664); + if (fd_ < 0) { + REPORT_ERROR_STATIC(MARTe::ErrorManagement::Warning, + "BinaryRecorder: failed to open recording file"); + fileOffset_ = 0u; + return; + } + (void) pwrite(fd_, pushHeader_, pushHeaderLen_, 0); + fileOffset_ = pushHeaderLen_; +} + +void BinaryRecorder::CloseFile() { + if (fd_ >= 0) { + (void) fdatasync(fd_); + (void) close(fd_); + fd_ = -1; + } + fileOffset_ = 0u; +} + +void BinaryRecorder::PruneOldFiles() { + DIR *dir = opendir(config_.directory); + if (dir == static_cast(0)) { return; } + + char prefix[160]; + (void) snprintf(prefix, sizeof(prefix), "%s_", sourceId_); + const size_t prefixLen = strlen(prefix); + + static const uint32 kMaxList = 4096u; + char (*names)[256] = new char[kMaxList][256]; + uint32 count = 0u; + + struct dirent *ent; + while (((ent = readdir(dir)) != static_cast(0)) && + (count < kMaxList)) { + const char *nm = ent->d_name; + const size_t nlen = strlen(nm); + if (nlen < prefixLen + 4u) { continue; } + if (strncmp(nm, prefix, prefixLen) != 0) { continue; } + if (strcmp(nm + (nlen - 4u), ".bin") != 0) { continue; } + strncpy(names[count], nm, 255u); + names[count][255] = '\0'; + count++; + } + (void) closedir(dir); + + /* Sort ascending (lexicographic == chronological via timestamp+seq). */ + for (uint32 a = 0u; a < count; a++) { + for (uint32 b = a + 1u; b < count; b++) { + if (strcmp(names[a], names[b]) > 0) { + char t[256]; + strncpy(t, names[a], 256u); + strncpy(names[a], names[b], 256u); + strncpy(names[b], t, 256u); + } + } + } + + /* Unlink oldest until at most keepFiles remain. */ + if (count > config_.keepFiles) { + const uint32 toRemove = count - config_.keepFiles; + for (uint32 i = 0u; i < toRemove; i++) { + char full[768]; + (void) snprintf(full, sizeof(full), "%s/%s", + config_.directory, names[i]); + (void) unlink(full); + } + } + + delete[] names; +} + +bool BinaryRecorder::HasSufficientDisk() { + if (config_.minDiskFreeMB == 0u) { return true; } + struct statvfs st; + if (statvfs(config_.directory, &st) != 0) { return true; } + const uint64 freeBytes = static_cast(st.f_bavail) * + static_cast(st.f_frsize); + const uint64 freeMB = freeBytes / (1024u * 1024u); + return freeMB >= static_cast(config_.minDiskFreeMB); +} + +/*---------------------------------------------------------------------------*/ +/* Control + info */ +/*---------------------------------------------------------------------------*/ + +void BinaryRecorder::RequestArm() { + (void) stagingMutex_.FastLock(); + pendingArm_ = true; + pendingDisarm_ = false; + stagingMutex_.FastUnLock(); +} + +void BinaryRecorder::RequestDisarm() { + (void) stagingMutex_.FastLock(); + pendingDisarm_ = true; + pendingArm_ = false; + stagingMutex_.FastUnLock(); +} + +bool BinaryRecorder::IsEnabled() const { + return config_.enabled; +} + +void BinaryRecorder::GetInfo(bool &recording, char *file, uint32 fileSz, + uint64 &bytesWritten, uint64 &rowsWritten, + uint64 &droppedRows, uint64 &freeMB) const { + recording = armed_ && (fd_ >= 0); + if ((file != static_cast(0)) && (fileSz > 0u)) { + strncpy(file, curFilePath_, fileSz - 1u); + file[fileSz - 1u] = '\0'; + } + bytesWritten = bytesWritten_; + rowsWritten = rowsWritten_; + droppedRows = droppedRows_; + + freeMB = 0u; + struct statvfs st; + if (statvfs(config_.directory, &st) == 0) { + const uint64 freeBytes = static_cast(st.f_bavail) * + static_cast(st.f_frsize); + freeMB = freeBytes / (1024u * 1024u); + } +} + +} /* namespace StreamHub */ diff --git a/Source/Applications/StreamHub/BinaryRecorder.h b/Source/Applications/StreamHub/BinaryRecorder.h new file mode 100644 index 0000000..5cb2b61 --- /dev/null +++ b/Source/Applications/StreamHub/BinaryRecorder.h @@ -0,0 +1,164 @@ +/** + * @file BinaryRecorder.h + * @brief Per-source binary recorder writing MARTe2 FileWriter-compatible files. + * + * @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. + * + * @details + * Records incoming signal packets to disk in the MARTe2 FileWriter binary + * format, one rotating file per source. The capture path (CapturePacket, + * Configure) runs on the session receive thread; all file I/O (FlushTick) + * runs on the 30 Hz push thread. Only the staging double-buffer, the header + * blob and a handful of control flags cross threads, guarded by a + * FastPollingMutexSem. + * + * On-disk format (byte-identical to FileWriter for un-quantized signals): + * [uint32 numSigs] + * per signal: [uint16 TypeDescriptor.all][char[32] name][uint32 numElements] + * rows: each row = all included signals concatenated, native little-endian. + */ +#ifndef STREAMHUB_BINARYRECORDER_H_ +#define STREAMHUB_BINARYRECORDER_H_ + +#include + +#include "CompilerTypes.h" +#include "TypeDescriptor.h" +#include "FastPollingMutexSem.h" +#include "UDPSProtocol.h" + +namespace StreamHub { + +using MARTe::uint8; +using MARTe::uint16; +using MARTe::uint32; +using MARTe::uint64; +using MARTe::float64; + +/** Maximum number of signals a recorder can include. */ +static const uint32 kRecMaxSig = 256u; + +/** Configuration for a BinaryRecorder, parsed from the +Recorder cfg block. */ +struct RecorderConfig { + bool enabled; ///< Master enable + bool autoStart; ///< Arm at startup + char directory[512]; ///< Output directory + uint32 maxFileBytes; ///< Roll threshold (MaxFileMB * 1024 * 1024) + uint32 keepFiles; ///< Newest N kept per source + uint32 stagingBytes; ///< Per-buffer hard cap (StagingMB * 1024 * 1024) + uint32 flushIntervalSec; ///< fdatasync cadence + uint32 minDiskFreeMB; ///< Stop-writing guard + char signals[1024]; ///< "all" or comma-separated "src:sig" keys +}; + +class BinaryRecorder { +public: + BinaryRecorder(); + ~BinaryRecorder(); + + /** One-time setup: copy config, source id, create directory. */ + void Init(const RecorderConfig &cfg, const char *sourceId); + + /** + * Receive thread: (re)build the FileWriter header and per-signal row + * layout for the included signals. Triggers a file reopen on the push + * thread so the new layout takes effect with a fresh header. + */ + void Configure(const MARTe::UDPSSignalDescriptor *descs, uint32 nSigs, + const bool *includeMask); + + /** Receive thread: append one packet's rows to the staging buffer. */ + void CapturePacket(const uint8 *payload, const uint32 *sigOff, + const uint32 *sigElems, uint8 publishMode, + uint32 numSamples); + + /** Push thread: open pending file, swap+flush staging, rotate, fsync. */ + void FlushTick(uint32 nowSec); + + void RequestArm(); ///< Any thread: arm recording. + void RequestDisarm(); ///< Any thread: disarm, flush, close. + bool IsEnabled() const; ///< True when the master enable is set. + + /** recInfo snapshot (push thread). */ + void GetInfo(bool &recording, char *file, uint32 fileSz, + uint64 &bytesWritten, uint64 &rowsWritten, + uint64 &droppedRows, uint64 &freeMB) const; + + /** Map a UDPS type code to a MARTe2 TypeDescriptor.all (uint16). */ + static uint16 TypeCodeToDescriptorAll(uint8 udpsTypeCode); + + /** Encode a physical value into native little-endian bytes; returns size. */ + static uint32 EncodeNative(uint8 typeCode, float64 v, uint8 *dst); + +private: + void OpenNewFile(); + void CloseFile(); + void PruneOldFiles(); + bool HasSufficientDisk(); + + /* Configuration */ + RecorderConfig config_; + char sourceId_[128]; + + /* Layout (built on receive thread by Configure) */ + uint32 srcIdx_[kRecMaxSig]; ///< Source signal index of included signal + uint8 typeCode_[kRecMaxSig]; ///< UDPS type code + uint8 quantType_[kRecMaxSig]; ///< UDPS quant code + uint32 declaredElems_[kRecMaxSig];///< Element count per signal + uint32 nativeBytes_[kRecMaxSig]; ///< Native bytes per element + uint32 wireBytes_[kRecMaxSig]; ///< Wire bytes per element + float64 rangeMin_[kRecMaxSig]; + float64 rangeMax_[kRecMaxSig]; + uint32 nIncluded_; + uint32 rowBytes_; + + /* Header blob (receive thread writes, push thread snapshots) */ + uint8 headerBlob_[4u + kRecMaxSig * 38u]; + uint32 headerLen_; + bool layoutReady_; + + /* Push-owned header snapshot (used for file I/O without holding the lock) */ + uint8 pushHeader_[4u + kRecMaxSig * 38u]; + uint32 pushHeaderLen_; + uint32 pushRowBytes_; + + /* Staging double buffer */ + uint8 *front_; + uint8 *back_; + uint32 frontLen_; + uint32 backLen_; + mutable MARTe::FastPollingMutexSem stagingMutex_; + + /* Control flags (guarded by stagingMutex_) */ + bool pendingArm_; + bool pendingDisarm_; + bool pendingReopen_; + + /* Push-thread file state */ + int fd_; + uint64 fileOffset_; + uint32 lastSyncSec_; + uint32 fileSeq_; + char curFilePath_[768]; + + /* State + counters */ + bool armed_; + bool diskLow_; + uint64 bytesWritten_; + uint64 rowsWritten_; + uint64 droppedRows_; +}; + +} /* namespace StreamHub */ + +#endif /* STREAMHUB_BINARYRECORDER_H_ */ diff --git a/Test/Applications/StreamHub/BinaryRecorderGTest.cpp b/Test/Applications/StreamHub/BinaryRecorderGTest.cpp new file mode 100644 index 0000000..f0154ab --- /dev/null +++ b/Test/Applications/StreamHub/BinaryRecorderGTest.cpp @@ -0,0 +1,439 @@ +/** + * @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 +#include +#include +#include +#include +#include +#include +#include +#include + +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 ReadFileBytes(const std::string &path) { + std::vector 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(n)); + size_t rd = fread(&out[0], 1u, static_cast(n), f); + out.resize(rd); + } + fclose(f); + return out; +} + +static std::vector ListBinFiles(const std::string &dir) { + std::vector 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 &b, size_t off) { + uint16 v = 0u; memcpy(&v, &b[off], 2u); return v; +} +static uint32 ReadU32(const std::vector &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 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(&b[6])); + EXPECT_EQ(1u, ReadU32(b, 38)); + /* descriptor 1 */ + EXPECT_EQ(MARTe::UnsignedInteger32Bit.all, ReadU16(b, 42)); + EXPECT_STREQ("Time", reinterpret_cast(&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 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 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(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 b = ReadFileBytes(path); + /* header has 2 signals, A and C */ + EXPECT_EQ(2u, ReadU32(b, 0)); + EXPECT_STREQ("A", reinterpret_cast(&b[6])); + EXPECT_STREQ("C", reinterpret_cast(&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 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(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 files = ListBinFiles(dir); + EXPECT_LE(files.size(), 2u); + EXPECT_GE(files.size(), 1u); + for (size_t i = 0u; i < files.size(); i++) { + std::vector 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); +} diff --git a/Test/Applications/StreamHub/BinaryRecorderSrc.cpp b/Test/Applications/StreamHub/BinaryRecorderSrc.cpp new file mode 100644 index 0000000..1fda4ae --- /dev/null +++ b/Test/Applications/StreamHub/BinaryRecorderSrc.cpp @@ -0,0 +1,19 @@ +/** + * @file BinaryRecorderSrc.cpp + * @brief Compiles the StreamHub BinaryRecorder implementation into the test + * library (StreamHub builds an executable, not a linkable archive). + * + * @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 "../../../Source/Applications/StreamHub/BinaryRecorder.cpp" diff --git a/Test/Applications/StreamHub/Makefile.inc b/Test/Applications/StreamHub/Makefile.inc index 7fc40fd..79eea1e 100644 --- a/Test/Applications/StreamHub/Makefile.inc +++ b/Test/Applications/StreamHub/Makefile.inc @@ -22,7 +22,7 @@ # ############################################################# -OBJSX = TriggerEngineSrc.x SignalRingBufferGTest.x TriggerEngineGTest.x LTTBGTest.x +OBJSX = TriggerEngineSrc.x BinaryRecorderSrc.x SignalRingBufferGTest.x TriggerEngineGTest.x LTTBGTest.x BinaryRecorderGTest.x PACKAGE=Applications ROOT_DIR=../../..