TapeRecorder.h

Go to the documentation of this file.

/*
 * Copyright (C) 2012 by
 *   MetraLabs GmbH (MLAB), GERMANY
 * and
 *   Neuroinformatics and Cognitive Robotics Labs (NICR) at TU Ilmenau, GERMANY
 * All rights reserved.
 *
 * Contact: info@mira-project.org
 *
 * Commercial Usage:
 *   Licensees holding valid commercial licenses may use this file in
 *   accordance with the commercial license agreement provided with the
 *   software or, alternatively, in accordance with the terms contained in
 *   a written agreement between you and MLAB or NICR.
 *
 * GNU General Public License Usage:
 *   Alternatively, this file may be used under the terms of the GNU
 *   General Public License version 3.0 as published by the Free Software
 *   Foundation and appearing in the file LICENSE.GPL3 included in the
 *   packaging of this file. Please review the following information to
 *   ensure the GNU General Public License version 3.0 requirements will be
 *   met: http://www.gnu.org/copyleft/gpl.html.
 *   Alternatively you may (at your option) use any later version of the GNU
 *   General Public License if such license has been publicly approved by
 *   MLAB and NICR (or its successors, if any).
 *
 * IN NO EVENT SHALL "MLAB" OR "NICR" BE LIABLE TO ANY PARTY FOR DIRECT,
 * INDIRECT, SPECIAL, INCIDENTAL, OR CONSEQUENTIAL DAMAGES ARISING OUT OF
 * THE USE OF THIS SOFTWARE AND ITS DOCUMENTATION, EVEN IF "MLAB" OR
 * "NICR" HAS BEEN ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
 *
 * "MLAB" AND "NICR" SPECIFICALLY DISCLAIM ANY WARRANTIES, INCLUDING,
 * BUT NOT LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND
 * FITNESS FOR A PARTICULAR PURPOSE. THE SOFTWARE PROVIDED HEREUNDER IS
 * ON AN "AS IS" BASIS, AND "MLAB" AND "NICR" HAVE NO OBLIGATION TO
 * PROVIDE MAINTENANCE, SUPPORT, UPDATES, ENHANCEMENTS OR MODIFICATIONS.
 */

#ifndef _MIRA_TAPERECORDER_H_
#define _MIRA_TAPERECORDER_H_

#include <serialization/Serialization.h>
#include <fw/Framework.h>
#include <fw/Tape.h>
#include <fw/PerformanceStatistics.h>

namespace mira {


class RecordedChannelInfo
{
public:

    RecordedChannelInfo() :
        compressionLevel(0),
        interval(Duration::microseconds(0)),
        avoidDataLoss(false) {}

    template <typename Reflector>
    void reflect(Reflector& r)
    {
        serialization::VersionType v = MIRA_REFLECT_VERSION(r, 2);
        r.member("Codecs", codecs, "The codecs used for this channel",
                 std::list<BinarySerializerCodecPtr>());
        if (v < 2) {
            bool compress;
            r.member("Compress", compress, "Should the data be stored compressed", false);
            compressionLevel = (compress ? -1 : 0);
        } else {
            r.member("CompressionLevel", compressionLevel,
                     "Compression level (0=none, 1..9=best speed to best compression, -1=default)", 0);
        }
        r.member("Interval", interval, "Stores only data in that interval (if 0 stores all data)",
                 Duration::microseconds(0));
        r.member("AvoidDataLoss", avoidDataLoss,
                 "Avoid data loss while recording this channel by reading all data from a channel since last recording",
                 false);
    }

    std::list<BinarySerializerCodecPtr> codecs;
    int compressionLevel;
    Duration interval;
    bool avoidDataLoss;
};

typedef std::map<std::string, RecordedChannelInfo> RecordedChannelInfoMap;

class TapeRecorderConfig
{
public:

    template <typename Reflector>
    void reflect(Reflector& r)
    {
        r.member("Channels", channels, "The channels to be recorded");
        r.member("RecordTransformations", recordTransforms,
                 "Record all transformation channels?", false);
        r.member("MaxMessageBlockSize", maxMessageBlockSize,
                 "The maximum size of a message block in bytes",
                 64*1024*1024); // default size is 64MB
        r.member("Filename", filename, "The filename for the output tape");
        r.member("TimestampFilename", timestampFilename,
                 "Automatically extend filename by timestamp?", false);
    }

    RecordedChannelInfoMap channels;
    bool recordTransforms = false;
    uint32 maxMessageBlockSize = 64*1024*1024;
    Path filename;
    bool timestampFilename = false;
};
class MIRA_FRAMEWORK_EXPORT TapeRecorder
{
public:

    class ChannelInfo : public RecordedChannelInfo
    {
    public:
        ChannelInfo() :
            seqID(0), recordCounter(0) {}
        uint32 seqID;
        Time expected;
        uint64 recordCounter;
        TypeMetaPtr meta;
        MetaTypeDatabase metaDB;
        PerformanceStatistics stats;
    };

    typedef std::map<std::string, ChannelInfo> ChannelMap;

public:
    TapeRecorder(const Duration& initialWaitForDataDuration = Duration::seconds(0));

    virtual ~TapeRecorder();

    template <typename Reflector>
    void reflect(Reflector& r)
    {
        r.property("MaxMessageBlockSize", mMaxMessageBlockSize,
                   "The maximum size of a message block in bytes");
        r.property("WarnOfDataLoss", mWarnDataLoss,
                   "Warn when missed update is detected for a channel with AvoidDataLoss", false);
    }

    MIRA_DEPRECATED("Please use int compressionLevel(=0/-1) instead of bool compress(=false/true)",
    void addChannel(const std::string& channelID, 
                    const std::list<BinarySerializerCodecPtr>& codecs,
                    bool compress, Duration interval = Duration::microseconds(0), bool avoidDataLoss = false))
    {
        addChannel(channelID, codecs, (compress ? -1 : 0), interval, avoidDataLoss);
    }


    void addChannel(const std::string& channelID,
                    const std::list<BinarySerializerCodecPtr>& codecs = std::list<BinarySerializerCodecPtr>(),
                    int compressionLevel = 0,
                    Duration interval = Duration::microseconds(0), bool avoidDataLoss = false);

    void addChannel(const std::string& channelID,
                    const RecordedChannelInfo& info);

    void record(const Path& file);

    void record(const TapeRecorderConfig& config);

    void stop(bool finishQueue = true);

    size_t storageQueueSize();

    void processStorageQueue(int numEntries = 1);

    void setMaxMessageBlockSize(uint32 maxMessageBlockSize)
    {
        mMaxMessageBlockSize = maxMessageBlockSize;
    }

    void setWarnOfDataLoss(bool enable)
    {
        mWarnDataLoss = enable;
    }

    bool getWarnOfDataLoss()
    {
        return mWarnDataLoss;
    }

    bool isRecording() const
    {
        return mIsRecording;
    }

    const PerformanceStatistics& getStats(const std::string& name);

protected:

    // normal callback for data changes (writes data to tape)
    bool valueChanged(ChannelRead<void> value, bool forceStorage = false);
    void intervalValueChanged(ChannelReadInterval<void> values, bool forceStorage = false);

    void addAllTransformationChannels();
    Path timestampFilename(const Path& filenameBase);

private:

    struct MessageToStore
    {
        StampedHeader header;
        std::string channelID;
        std::string typeName;
        Time enqueueTime;
        Buffer<uint8> buf;
        ChannelInfo* info;
    };

    void storageThread();
    void storeMessage(MessageToStore& msg);

protected:

    Authority mAuthority;
    uint32 mMaxMessageBlockSize;
    bool mWarnDataLoss;
    Tape mTape;
    bool mIsRecording;
    bool mFirstCallbackMessage;
    ChannelMap mChannels;
    Duration mInitialWaitForDataDuration;

    boost::thread mStorageThread;
    boost::condition_variable mStorageThreadCondition;
    boost::mutex mStorageMutex;
    bool mStorageThreadStarted;

    std::deque<MessageToStore> mStorageQueue;
    volatile bool mStorageThreadTerminationRequest;
};


}

#endif