MARTe-Integrated-Components/Source/Applications/StreamHub/BinaryRecorder.h

/**
 * @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 <string.h>

#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_ */