ChannelReadWrite.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_CHANNELLOCKEDREADWRITE_H_
#define _MIRA_CHANNELLOCKEDREADWRITE_H_

#ifndef Q_MOC_RUN
#include <boost/shared_ptr.hpp>
#include <boost/thread/shared_mutex.hpp>
#include <boost/thread/locks.hpp>
#endif

#include <utils/IsCheapToCopy.h>

#include <fw/AbstractChannel.h>

namespace mira {


// forward declaration
template<typename T>
class ConcreteChannel;

template<typename T>
class Channel;

// forward declaration
template <typename ConcreteChannelReadWrite>
struct ChannelReadWriteTraits {};


template<typename DerivedContainer>
class ChannelReadWriteShared
{
public:
    typedef typename ChannelReadWriteTraits<DerivedContainer>::ChannelType ChannelType;
    typedef typename ChannelReadWriteTraits<DerivedContainer>::Lock    Lock;
    typedef typename ChannelReadWriteTraits<DerivedContainer>::SlotType Slot;
    typedef typename ChannelReadWriteTraits<DerivedContainer>::ValueType ValueType;

    typedef ChannelType*               ChannelTypePtr;


public:

    ChannelReadWriteShared(ChannelTypePtr iChannel, ChannelBufferBase::Slot* iSlot) :
        channel(iChannel), slot(iSlot), lock(iSlot->lock, boost::adopt_lock) {}

    ~ChannelReadWriteShared() {
        // if our destructor was invoked due to stack unwinding in an exception
        // handling mechanism, we do NOT write the data but discard it. The
        // data is written only if the destructor is called as we got out
        // of scope WITHOUT a raised exception.
#if __cplusplus >= 201703L
        if(std::uncaught_exceptions() == 0)
#else
        if(!std::uncaught_exception()) // deprecated in C++17
#endif
            try {
                DerivedContainer::finish(this); // may throw
            }
            // do not allow exceptions to escape from the destructor,
            // they will terminate the program!
            catch(std::exception& ex) {
                MIRA_LOG_EXCEPTION(ERROR, ex) << "Error in ChannelRead/Write::finish(): ";
            }
            catch(...) {
                MIRA_LOG(ERROR) << "Unknown error in ChannelRead/Write::finish()";
            }
        else
            DerivedContainer::discard(this); // must not throw
    }

public:

    ChannelTypePtr channel;

    ChannelBufferBase::Slot* slot;

    Lock lock;
};

template<typename DerivedContainer>
class ChannelReadCopy
{

public:
    typedef typename ChannelReadWriteTraits<DerivedContainer>::ChannelType ChannelType;
    typedef typename ChannelReadWriteTraits<DerivedContainer>::SlotType Slot;
    typedef typename ChannelReadWriteTraits<DerivedContainer>::ValueType ValueType;

    typedef ChannelType*               ChannelTypePtr;

public:
    ChannelReadCopy(ChannelTypePtr iChannel, ChannelBufferBase::Slot* iSlot) :
        channel(iChannel)
    {
        // it should not be possible to get a wrong type in here
        // without a rework of the channel read mechanism,
        // so we only do a thorough check in debug build
        assert(dynamic_cast<Slot*>(iSlot) != nullptr);

        auto typedSlot = static_cast<Slot*>(iSlot);

        slotHolder = typedSlot->clone();
        slot = slotHolder.get();
        iSlot->lock.unlock_shared(); // the slot is already locked
                                     // we release the lock after taking our own copy
    }

    ChannelTypePtr channel;

public:

    std::unique_ptr<Slot> slotHolder;

    // still need this for the internal interface defined by ChannelReadWriteShared
    ChannelBufferBase::Slot* slot;

private:
    // fake lock mimics locking to adhere to the expected behaviour
    // but does not actually lock anything (since we hold an exclusive copy of the slot)
    struct MockLock
    {
        bool owns_lock() { return locked; }
        void unlock()    { locked = false; };

        bool locked = true;
    };

public:
    MockLock lock;
};


template<typename Derived>
class ChannelReadWriteCommonBase
{
public:

    typedef typename ChannelReadWriteTraits<Derived>::Shared Shared;
    typedef boost::shared_ptr<Shared>       SharedPtr;

    // import some important types from our Shared class
    typedef typename Shared::ChannelType    ChannelType;
    typedef typename Shared::ChannelTypePtr ChannelTypePtr;
    typedef typename Shared::Slot           Slot;
    typedef typename Shared::ValueType      ValueType;

public:

    ChannelReadWriteCommonBase() {}

    ChannelReadWriteCommonBase(ChannelTypePtr channel, ChannelBufferBase::Slot* slot) :
        shared(new Shared(channel, slot) )
    {}

public:

    bool isValid() const {
        // channel must be assigned and data must be locked
        return shared.get()!=nullptr && shared->lock.owns_lock();
    }

public:

    const std::string& getChannelID() {
        static std::string emptyString;
        return shared.get()!=nullptr ? shared->channel->getID() : emptyString;
    }

    std::string getTypename() const {
        return shared.get()!=nullptr ? shared->channel->getTypename() : "";
    }

    TypeMetaPtr getTypeMeta() const {
        return shared.get()!=nullptr ? shared->channel->getTypeMeta() : TypeMetaPtr();
    }

    const Time& getTimestamp() const {
        checkValid();
        return shared->slot->timestamp();
    }

    int getUseCount() const {
        return shared.get()!=nullptr ? shared.use_count() : 0;
    }

public:

    uint32 getFlags() const {
        checkValid();
        return shared->slot->flags;
    }

    void addFlags(uint32 moreFlags) const {
        checkValid();
        shared->slot->flags |= moreFlags;
    }

protected:

    void checkValid() const {
        if(shared.get()==nullptr)
            MIRA_THROW(XAccessViolation, "Trying to access ChannelRead / "
                       "ChannelWrite that was not assigned with valid data");

        if(!shared->lock.owns_lock())
            MIRA_THROW(XAccessViolation, "Trying to access data of ChannelRead "
                       "/ ChannelWrite after losing the lock");
    }

protected:
    // these read and write methods are protected by default. They will be
    // made public in the final ChannelRead/Write classes.

    const Buffer<uint8>& readSerializedValue() {
        checkValid();
        const Buffer<uint8>& res =
                shared->channel->getBuffer()->readSerializedValue(shared->slot);

        // ??? can we ever have an empty vector here ???
        // I think we can't, since:
        //    1. we either have a typed buffer where we can deserialize the
        //       data, if it doesn't exist.
        //    2. If we have an untyped buffer, someone has probably written
        //       some serialized data, otherwise the reader would never get a
        //       ChannelRead object since the channel is empty. He would have
        //       got an exception when trying to obtain the read object.
        assert(!res.empty());
        return res;
    }

    const Buffer<uint8>& readSerializedValue(uint8 formatVersion, bool orLower) {
        checkValid();
        return shared->channel->getBuffer()->readSerializedValue(shared->slot, formatVersion, orLower);
    }

    Buffer<uint8> readSerializedValue(std::list<BinarySerializerCodecPtr>& codecs) {
        checkValid();
        return shared->channel->getBuffer()->readSerializedValue(shared->slot, codecs);
    }

    Buffer<uint8> readSerializedValue(std::list<BinarySerializerCodecPtr>& codecs,
                                      uint8 formatVersion, bool orLower) {
        checkValid();
        return shared->channel->getBuffer()->readSerializedValue(shared->slot, codecs,
                                                                 formatVersion, orLower);
    }

    void readJSON(JSONValue& oValue) {
        checkValid();
        shared->channel->getBuffer()->readJSON(shared->slot, oValue);
    }

    void readJSON(JSONValue& oValue, JSONSerializer& serializer) {
        checkValid();
        shared->channel->getBuffer()->readJSON(shared->slot, oValue, serializer);
    }

protected:

    void writeSerializedValue(Buffer<uint8> data) {
        checkValid();
        shared->channel->getBuffer()->writeSerializedValue(shared->slot, std::move(data));
    }

    void writeJSON(const JSONValue& value) {
        checkValid();
        shared->channel->getBuffer()->writeJSON(shared->slot, value);
    }

    void writeJSON(JSONDeserializer& deserializer) {
        checkValid();
        shared->channel->getBuffer()->writeJSON(shared->slot, deserializer);
    }

    void writeXML(const XMLDom::const_iterator& node)  {
        checkValid();
        shared->channel->getBuffer()->writeXML(shared->slot, node);
    }

protected:
    SharedPtr shared;
};



template<typename Derived, typename T>
class ChannelReadWriteBase : public ChannelReadWriteCommonBase<Derived>
{
public:

    typedef ChannelReadWriteCommonBase<Derived> Base;

    // push down some important types from our base
    typedef typename Base::Shared     Shared;
    typedef typename Base::ChannelTypePtr ChannelTypePtr;
    typedef typename Base::Slot       Slot;
    typedef typename Base::ValueType  ValueType;

public:

    ChannelReadWriteBase() {}

    ChannelReadWriteBase(ChannelTypePtr channel, ChannelBufferBase::Slot* slot) : Base(channel,slot) {}

protected:

    ValueType& internalValue() {
        this->checkValid();
        return static_cast<Slot*>(this->shared->slot)->data;
    }

    const ValueType& internalValue() const {
        this->checkValid();
        return static_cast<Slot*>(this->shared->slot)->data;
    }

};


template<typename Derived>
class ChannelReadWriteBase<Derived, void> : public ChannelReadWriteCommonBase<Derived>
{
public:

    typedef ChannelReadWriteCommonBase<Derived> Base;

    // push down some important types from our base
    typedef typename Base::Shared     Shared;
    typedef typename Base::ChannelTypePtr ChannelTypePtr;
    typedef typename Base::ValueType  ValueType;

public:

    ChannelReadWriteBase() {}

    ChannelReadWriteBase(ChannelTypePtr channel, ChannelBufferBase::Slot* slot) : Base(channel,slot) {}

protected:

    ValueType& internalValue() {
        this->checkValid();
        return this->shared->channel->getBuffer()->getStampedHeader(this->shared->slot);
    }

    const ValueType& internalValue() const {
        this->checkValid();
        return this->shared->channel->getBuffer()->getStampedHeader(this->shared->slot);
    }

};


template<typename T>
class ChannelRead : public ChannelReadWriteBase<ChannelRead<T>,T>
{
protected:

    typedef ChannelReadWriteBase<ChannelRead<T>,T> Base;

    // push down some important types from our base
    typedef typename Base::Shared     Shared;
    typedef typename Base::ChannelTypePtr ChannelTypePtr;
    typedef typename Base::ValueType  ValueType;

public:

    ChannelRead() {}

    ChannelRead(ChannelTypePtr channel, ChannelBufferBase::Slot* slot) : Base(channel, slot) {}

public:

    operator const ValueType&() const { return this->internalValue(); }

    const ValueType& operator*() const { return this->internalValue(); }

    const ValueType* operator->() const { return &this->internalValue(); }

public:
    // make read-methods public

    const Buffer<uint8>& readSerializedValue() {
        return Base::readSerializedValue();
    }

    const Buffer<uint8>& readSerializedValue(uint8 formatVersion, bool orLower) {
        return Base::readSerializedValue(formatVersion, orLower);
    }

    Buffer<uint8> readSerializedValue(std::list<BinarySerializerCodecPtr>& codecs) {
        return Base::readSerializedValue(codecs);
    }

    Buffer<uint8> readSerializedValue(std::list<BinarySerializerCodecPtr>& codecs,
                                      uint8 formatVersion, bool orLower) {
        return Base::readSerializedValue(codecs, formatVersion, orLower);
    }


    void readJSON(JSONValue& oValue) {
        Base::readJSON(oValue);
    }

    void readJSON(JSONValue& oValue, JSONSerializer& serializer) {
        Base::readJSON(oValue, serializer);
    }

public:

    void finish() {
        if(this->isValid())
            finish(this->shared.get());
    }

public:

    Channel<T> getChannel();

protected:

    friend class ChannelReadWriteShared<ChannelRead<T> >;

    // API for ~ChannelReadWriteShared
    static void finish(Shared* shared) {
        assert(shared!=nullptr);
        if(shared->lock.owns_lock())
            shared->lock.unlock();
    }

    static void discard(Shared* shared) {
        // for read objects a "discard" is equal to a "finish"
        finish(shared);
    }

};


template<typename T>
class ChannelWrite : public ChannelReadWriteBase<ChannelWrite<T>, T>
{
protected:
    typedef ChannelReadWriteBase<ChannelWrite<T>, T> Base;

    // push down some important types from our base
    typedef typename Base::Shared     Shared;
    typedef typename Base::ChannelTypePtr ChannelTypePtr;
    typedef typename Base::ValueType  ValueType;


public:

    ChannelWrite() {}

    ChannelWrite(ChannelTypePtr channel, ChannelBufferBase::Slot* slot) : Base(channel, slot) {
        static_cast<StampedHeader&>(this->internalValue()).timestamp = Time::now();
    }

public:

    operator ValueType&() { return this->internalValue(); }

    ValueType& operator*() { return this->internalValue(); }


    /*const ValueType& operator=(const T& value) {
        this->internalValue() = value;
        return this->internalValue();
    }*/

    const ValueType& operator=(const ValueType& value) {
        this->internalValue() = value;
        return this->internalValue();
    }


    ValueType* operator->() { return &this->internalValue(); }

public:
    // make write-methods public

    void writeSerializedValue(Buffer<uint8> data) {
        Base::writeSerializedValue(std::move(data));
    }

    void writeJSON(const JSONValue& value) {
        Base::writeJSON(value);
    }

    void writeJSON(JSONDeserializer& deserializer) {
        Base::writeJSON(deserializer);
    }

    void writeXML(const XMLDom::const_iterator& node) {
        Base::writeXML(node);
    }

public:

    void finish()
    {
        if(this->isValid())
            finish(this->shared.get());
    }

    void discard()
    {
        if(this->isValid())
            discard(this->shared.get());
    }


public:

    Channel<T> getChannel();

protected:
    friend class ChannelReadWriteShared<ChannelWrite<T> >;

    // API for ~ChannelReadWriteShared
    static void finish(Shared* shared)
    {
        assert(shared!=nullptr);
        if(shared->lock.owns_lock()) {
            // release lock object, since it is unlocked by finishWrite()
            shared->lock.release();
            // inform the channel, that the writing was finished
            shared->channel->finishWrite(shared->slot);
        }
    }

    static void discard(Shared* shared)
    {
        assert(shared!=nullptr);
        if(shared->lock.owns_lock()) {
            // release lock object, since it is unlocked by discardWrite()
            shared->lock.release();
            // inform the channel, to discard writing
            shared->channel->discardWrite(shared->slot);
        }
    }
};


template <typename Derived, typename T, bool Copy = IsCheapToCopy<T>::value>
struct ChannelReadSelectCopy
{
    typedef ChannelReadWriteShared<Derived> Shared;
};

template <typename Derived, typename T>
struct ChannelReadSelectCopy<Derived, T, true>
{
    typedef ChannelReadCopy<Derived>        Shared;
};


// typed read and writes

template<typename T>
struct ChannelReadWriteTraits<ChannelRead<T> >
{
    typedef ConcreteChannel<T>  ChannelType;
    typedef typename ChannelBuffer<T>::Slot SlotType;
    typedef typename ChannelBuffer<T>::ValueType ValueType;
    typedef boost::shared_lock<boost::shared_mutex> Lock; // use shared locks for readers
    typedef typename ChannelReadSelectCopy<ChannelRead<T>, T>::Shared Shared;
};

template<typename T>
struct ChannelReadWriteTraits<ChannelWrite<T> >
{
    typedef ConcreteChannel<T>  ChannelType;
    typedef typename ChannelBuffer<T>::Slot SlotType;
    typedef typename ChannelBuffer<T>::ValueType ValueType;
    typedef boost::unique_lock<boost::shared_mutex> Lock; // use unique locks for writers
    typedef ChannelReadWriteShared<ChannelWrite<T>> Shared;
};

// untyped read and writes

template<>
struct ChannelReadWriteTraits<ChannelRead<void> >
{
    typedef AbstractChannel  ChannelType;
    typedef ChannelBuffer<void>::Slot SlotType;
    typedef ChannelBuffer<void>::ValueType ValueType;
    typedef boost::shared_lock<boost::shared_mutex> Lock; // use shared locks for readers
    typedef ChannelReadWriteShared<ChannelRead<void>> Shared;
};

template<>
struct ChannelReadWriteTraits<ChannelWrite<void> >
{
    typedef AbstractChannel  ChannelType;
    typedef ChannelBuffer<void>::Slot SlotType;
    typedef ChannelBuffer<void>::ValueType ValueType;
    typedef boost::unique_lock<boost::shared_mutex> Lock; // use unique locks for writers
    typedef ChannelReadWriteShared<ChannelWrite<void>> Shared;
};


}


#endif