ChannelBuffer.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_CHANNELBUFFER_H_
#define _MIRA_CHANNELBUFFER_H_

#include <string>
#include <list>
#include <type_traits>

#ifndef Q_MOC_RUN
#include <boost/noncopyable.hpp>
#include <boost/thread/mutex.hpp>
#include <boost/thread/shared_mutex.hpp>
#endif

#include <error/LoggingCore.h>

#include <factory/TypeId.h>

#include <serialization/BinarySerializer.h>
#include <serialization/JSONSerializer.h>
#include <serialization/XMLSerializer.h>
#include <serialization/MetaSerializer.h>

#include <utils/Time.h>
#include <utils/Stamped.h>
#include <platform/Typename.h>

#include <math/Eigen.h>

#include <fw/FrameworkExports.h>

namespace mira {


template <typename TargetType>
struct ChannelBufferPromoterCommon;

template <typename TargetType>
struct ChannelBufferPromoter;


enum SlotQueryMode
{
    OLDER_SLOT = 0,

    NEWER_SLOT,

    NEAREST_SLOT
};

enum IntervalFillMode
{
    PREFER_OLDER = 0,
    PREFER_NEWER
};


class MIRA_FRAMEWORK_EXPORT ChannelBufferBase : boost::noncopyable
{
    template <typename TargetType>
    friend struct ChannelBufferPromoterCommon;

    template <typename TargetType>
    friend struct ChannelBufferPromoter;

public:

    struct Slot;

    struct ListItem : boost::noncopyable
    {
        ListItem() {
            // creates empty list
            next = (Slot*)this;
            prev = (Slot*)this;
        }

        // both will be locked by mutex in ChannelBuffer
        Slot* next;
        Slot* prev;
    };

    struct Slot : public ListItem
    {
        boost::shared_mutex lock; // locks the following data

        boost::mutex serializedValueLock;

        const Time* timestampPtr;

        Buffer<uint8> serializedValue;

        uint32 flags;

        const Time& timestamp() const {
            assert(timestampPtr!=NULL); // pointer is set in ChannelBuffers allocateSlot
            return *timestampPtr;
        }

        virtual ~Slot() = default;
        virtual std::unique_ptr<Slot> clone() = 0;
    };

public:

    ChannelBufferBase() :
        mStorageDuration(Duration::seconds(0)),
        mAutoIncreaseStorageDuration(true),
        mMinSlots(0),
        mMaxSlots(100),
        mSize(0)
    {
    }

    virtual ~ChannelBufferBase();

public:

    void cloneParameters(const ChannelBufferBase& other)
    {
        mMinSlots = other.mMinSlots;
        mMaxSlots = other.mMaxSlots;
        mStorageDuration = other.mStorageDuration;
        mAutoIncreaseStorageDuration = other.mAutoIncreaseStorageDuration;
    }

public:

    void setStorageDuration(const Duration& storageDuration);


    void setAutoIncreaseStorageDuration(bool increase);

    void setMinSlots(std::size_t minSlots);


    void setMaxSlots(std::size_t maxSlots);

public:

    std::size_t getSize() const;

    std::size_t getMaxSlots() const;

    std::size_t getMinSlots() const;

    Duration getStorageDuration() const;

    bool isAutoIncreasingStorageDuration() const;


public:

    virtual int getTypeId() const = 0;

    virtual bool isTyped() const = 0;

    virtual bool isPolymorphic() const = 0;

    virtual Typename getTypename() const = 0;

    virtual void setTypename(const Typename& name) = 0;

    virtual TypeMetaPtr createTypeMeta(Slot* s, MetaTypeDatabase& ioDB) = 0;

    TypeMetaPtr getTypeMeta() const { return mTypeMeta; }

    void setTypeMeta(TypeMetaPtr meta) { mTypeMeta = meta; }

    virtual void fixateType() = 0;

    virtual Slot* allocateSlot() = 0;

    virtual void freeSlot(Slot* s) = 0;

    virtual StampedHeader& getStampedHeader(Slot* s) = 0;

    virtual const Buffer<uint8>& readSerializedValue(Slot* s) = 0;

    virtual const Buffer<uint8>& readSerializedValue(Slot* s,
                                                     uint8 formatVersion, bool orLower) = 0;

    virtual Buffer<uint8> readSerializedValue(Slot* s,
                                              std::list<BinarySerializerCodecPtr>& codecs) = 0;

    virtual Buffer<uint8> readSerializedValue(Slot* s,
                                              std::list<BinarySerializerCodecPtr>& codecs,
                                              uint8 formatVersion, bool orLower) = 0;

    virtual void writeSerializedValue(Slot* s, Buffer<uint8> data) = 0;

    virtual void readJSON(Slot* s, JSONValue& oValue) = 0;

    virtual void readJSON(Slot* s, JSONValue& oValue, JSONSerializer& serializer) = 0;

    virtual void writeJSON(Slot* s, const JSONValue& value) = 0;

    virtual void writeJSON(Slot* s, JSONDeserializer& value) = 0;

    virtual void writeXML(Slot* s, const XMLDom::const_iterator& node) = 0;

public:

    // write access

    Slot* requestWriteSlot();

    bool finishWriteSlot(Slot* n, bool* dropped=NULL);

    void discardWriteSlot(Slot* n);


    // read access

    Slot* readNewestSlot();

    Slot* readSlotAtTime(const Time& timestamp, SlotQueryMode mode);

    void readInterval(const Time& timestamp, std::size_t nrSlots,
                      std::size_t olderSlots, std::size_t newerSlots,
                      IntervalFillMode fillMode, std::list<Slot*>& oSlots);

    void readInterval(const Time& from, const Time& to, std::list<Slot*>& oSlots);

public:

    template<typename TargetType>
    ChannelBufferBase* promote();

private:

    Slot* findFirstOlderSlot(const Time& timestamp, Slot* start=NULL);

private:

    void splice(Slot* first, Slot* last, Slot* position) const;

    void splice(Slot* item, Slot* position) const {
        splice(item,item,position);
    }

    static bool isEmpty(const ListItem& list) { return list.next == &list; }
    static Slot* begin(ListItem& list) { return list.next; }
    static Slot* last(ListItem& list)  { return list.prev; }
    static Slot* end(ListItem& list)   { return (Slot*)&list; }

    void deleteSlots(ListItem& list);

protected:

    void clear();

private:

    void resetSlot(Slot* s);

    // parameters:
    Duration  mStorageDuration;
    bool mAutoIncreaseStorageDuration;
    std::size_t mMinSlots;
    std::size_t mMaxSlots;


    // mutex that locks all actions when we change our lists
    mutable boost::mutex mMutex;

    std::size_t mSize; // current number of slots we are using

    ListItem mBuffer;

    ListItem mWaitingOrFree;

    ListItem mWriting;

    TypeMetaPtr mTypeMeta;

private:

    std::string dbgDumpList(ListItem& list, bool brief);

public:

    void dbgDump(const std::string& prefix, bool brief=true);

    static void dbgCheckListIntegrity(const ListItem& list);
    void dbgCheckIntegrity();
};


template <typename T>
class TypedChannelBufferBase : public ChannelBufferBase
{
public:

    typedef Stamped<T> ValueType;

public:

    virtual ~TypedChannelBufferBase() { this->clear(); }

public:

    struct Slot : public ChannelBufferBase::Slot {
        ValueType data;
        EIGEN_MAKE_ALIGNED_OPERATOR_NEW

        std::unique_ptr<ChannelBufferBase::Slot> clone() override {
            auto slot = std::make_unique<Slot>();
            slot->serializedValue = this->serializedValue;
            slot->timestampPtr = &(static_cast<StampedHeader&>(slot->data).timestamp);
            slot->flags = this->flags;
            slot->data = this->data;
            return slot;
        }
    };

protected:

    Slot* castSlot(ChannelBufferBase::Slot* s)
    {
        // cast abstract ChannelBufferBase::Slot to TypedChannelBufferBase<T>::Slot.
        // this is safe since all slots were created by us in allocateSlot()
        // as TypedChannelBufferBase<T>::Slots
        return static_cast<Slot*>(s);
    }

public:
    virtual int getTypeId() const { return typeId<T>(); }
    virtual bool isTyped() const { return true; }
    virtual bool isPolymorphic() const {return false; }
    virtual Typename getTypename() const { return typeName<T>(); }
    virtual void setTypename(const Typename& name) {
        MIRA_THROW(XRuntime, "Cannot set Typename for typed channels");
    }
    virtual TypeMetaPtr createTypeMeta(ChannelBufferBase::Slot* s, MetaTypeDatabase& ioDB) {
        MetaSerializer ms(ioDB);
        Slot* slot = castSlot(s);
        TypeMetaPtr meta = ms.addMeta(slot->data.internalValueRep());
        return meta;
    }

    virtual void fixateType() {} // doesn't do anything
    virtual ChannelBufferBase::Slot* allocateSlot() {
        Slot* slot = new Slot;
        slot->timestampPtr = &(static_cast<StampedHeader&>(slot->data).timestamp); // "connect" the timestamp ptr
        return slot;
    }
    virtual void freeSlot(ChannelBufferBase::Slot* s) {
        // casting s to our slot type ensures that the correct destructor is called
        Slot* slot = castSlot(s);
        delete slot;
    }

public:

    virtual StampedHeader& getStampedHeader(ChannelBufferBase::Slot* s) {
        // we can safely cast here, since s is an object of our Slot type
        Slot* slot = static_cast<Slot*>(s);
        return stampedHeader(slot->data);
    }

    virtual const Buffer<uint8>& readSerializedValue(ChannelBufferBase::Slot* s)
    {
        // lock the serialization mutex to protect serializedValueLock
        // (we or someone else may write it here)
        boost::mutex::scoped_lock lock(s->serializedValueLock);

        // no serialized data yet, so serialize it now (lazy evaluation)
        if(s->serializedValue.empty()) {
            // we can safely cast here, since s is an object of our Slot type
            Slot* slot = static_cast<Slot*>(s);

            // serialize the data
            BinaryBufferSerializer bs(&s->serializedValue);
            bs.serialize(slot->data.internalValueRep(), false);
        }

        return s->serializedValue;
    }

    
    virtual const Buffer<uint8>& readSerializedValue(ChannelBufferBase::Slot* s,
                                                     uint8 formatVersion, bool orLower)
    {
        uint8 serializerVersion = BinaryBufferSerializer::getSerializerFormatVersion();

        if (formatVersion == serializerVersion)
            return readSerializedValue(s);

        if (formatVersion == 0) {
            // lock the serialization mutex to protect serializedValueLock
            // (we or someone else may write it here)
            boost::mutex::scoped_lock lock(s->serializedValueLock);

            // no serialized data yet, so serialize it now (lazy evaluation)
            if(s->serializedValue.empty()) {
                // we can safely cast here, since s is an object of our Slot type
                Slot* slot = static_cast<Slot*>(s);

                // serialize the data
                BinaryBufferSerializerLegacy bs(&s->serializedValue);
                bs.serialize(slot->data.internalValueRep(), false);
            }

            return s->serializedValue;
        }

        MIRA_THROW(XIO, "Requested serialized data of binary format version " << (int)formatVersion
                     << ". Only implemented for versions 0, " << (int)serializerVersion << ".");
    }

    virtual Buffer<uint8> readSerializedValue(ChannelBufferBase::Slot* s,
                                              std::list<BinarySerializerCodecPtr>& codecs)
    {
        Buffer<uint8> serializedValue;

        // we can safely cast here, since s is an object of our Slot type
        Slot* slot = static_cast<Slot*>(s);

        BinaryBufferSerializer bs(&serializedValue);

        // register codecs
        foreach(BinarySerializerCodecPtr codec, codecs)
            bs.registerCodec(codec);

        // serialize the data
        bs.serialize(slot->data.internalValueRep(),false);

        return std::move(serializedValue);
    }

    virtual Buffer<uint8> readSerializedValue(ChannelBufferBase::Slot* s,
                                              std::list<BinarySerializerCodecPtr>& codecs,
                                              uint8 formatVersion, bool orLower)
    {
        uint8 serializerVersion = BinaryBufferSerializer::getSerializerFormatVersion();

        if (formatVersion == serializerVersion)
            return readSerializedValue(s, codecs);

        if (formatVersion == 0) {
            Buffer<uint8> serializedValue;

            // we can safely cast here, since s is an object of our Slot type
            Slot* slot = static_cast<Slot*>(s);

            BinaryBufferSerializerLegacy bs(&serializedValue);

            // register codecs
            foreach(BinarySerializerCodecPtr codec, codecs)
                bs.registerCodec(codec);

            // serialize the data
            bs.serialize(slot->data.internalValueRep(),false);

            return std::move(serializedValue);
        }

        MIRA_THROW(XIO, "Requested serialized data of binary format version " << (int)formatVersion
                     << ". Only implemented for versions 0, " << (int)serializerVersion << ".");
    }

    virtual void writeSerializedValue(ChannelBufferBase::Slot* s, Buffer<uint8> data)
    {
        s->serializedValue = std::move(data);

        // we can safely cast here, since s is an object of our Slot type
        Slot* slot = static_cast<Slot*>(s);

        // deserialize into our data
        BinaryBufferDeserializer bs(&s->serializedValue);

        {
            boost::mutex::scoped_lock lock(mCodecsMutex);
            bs.setCodecs(mCodecs);
        }
        bs.deserialize(slot->data.internalValueRep(), false);
        {
            boost::mutex::scoped_lock lock(mCodecsMutex);
            mCodecs = bs.getCodecs();
        }
    }

    virtual void readJSON(ChannelBufferBase::Slot* s, JSONValue& oValue)
    {
        // we can safely cast here, since s is an object of our Slot type
        Slot* slot = static_cast<Slot*>(s);

        JSONSerializer json;
        oValue = json.serialize(slot->data.internalValueRep());
    }

    virtual void readJSON(ChannelBufferBase::Slot* s, JSONValue& oValue,
                          JSONSerializer& serializer)
    {
        // we can safely cast here, since s is an object of our Slot type
        Slot* slot = static_cast<Slot*>(s);

        oValue = serializer.serialize(slot->data.internalValueRep());
    }

    virtual void writeJSON(ChannelBufferBase::Slot* s, const JSONValue& value)
    {
        // we can safely cast here, since s is an object of our Slot type
        Slot* slot = static_cast<Slot*>(s);

        JSONDeserializer json(value);
        json.deserialize(slot->data.internalValueRep());
    }

    virtual void writeJSON(ChannelBufferBase::Slot* s, JSONDeserializer& deserializer)
    {
        // we can safely cast here, since s is an object of our Slot type
        Slot* slot = static_cast<Slot*>(s);

        deserializer.deserialize(slot->data.internalValueRep());
    }

    void writeXML(ChannelBufferBase::Slot* s, const XMLDom::const_iterator& node) override
    {
        // we can safely cast here, since s is an object of our Slot type
        Slot* slot = static_cast<Slot*>(s);
        XMLDeserializer xml(node);
        const std::string rootTag = *node;
        xml.deserializeFromNode(rootTag.c_str(), slot->data.internalValueRep());
    }

private:

    BinaryBufferDeserializer::CodecsMap mCodecs;
    boost::mutex mCodecsMutex;
};



template <typename T>
class ChannelBuffer : public TypedChannelBufferBase<T>
{};


template <>
class MIRA_FRAMEWORK_EXPORT ChannelBuffer<void> : public ChannelBufferBase
{
public:

    typedef StampedHeader ValueType;

public:

    struct Slot : public ChannelBufferBase::Slot {
        ValueType header;
        EIGEN_MAKE_ALIGNED_OPERATOR_NEW

        std::unique_ptr<ChannelBufferBase::Slot> clone() override {
            auto slot = std::make_unique<Slot>();
            slot->serializedValue = this->serializedValue;
            slot->timestampPtr = &slot->header.timestamp;
            slot->flags = this->flags;
            slot->header = this->header;
            return slot;
        }
    };

protected:

    Slot* castSlot(ChannelBufferBase::Slot* s)
    {
        return static_cast<Slot*>(s);
    }

public:

    ChannelBuffer() : mTypename("") {}

    virtual ~ChannelBuffer() { this->clear(); }


public:
    virtual int getTypeId() const { return -1; }
    virtual bool isTyped() const { return false; }
    virtual bool isPolymorphic() const {return false; }
    virtual Typename getTypename() const { return mTypename; }
    virtual void setTypename(const Typename& name);
    virtual TypeMetaPtr createTypeMeta(ChannelBufferBase::Slot* s, MetaTypeDatabase& ioDB) {
        MIRA_THROW(XRuntime, "Cannot create type meta for untyped channels");
        // keep compiler happy
        return TypeMetaPtr();
    }
    virtual void fixateType() {} // doesn't do anything

    virtual ChannelBufferBase::Slot* allocateSlot() {
        Slot* slot = new Slot;
        slot->timestampPtr = &slot->header.timestamp; // "connect" the timestamp ptr
        return slot;
    }

    virtual void freeSlot(ChannelBufferBase::Slot* s) {
        // casting s to our slot type ensures that the correct destructor is called
        Slot* slot = castSlot(s);
        delete slot;
    }

public:

    virtual StampedHeader& getStampedHeader(ChannelBufferBase::Slot* s) {
        // casting here is not safe, if we actually are a typed buffer,
        // however, if we are, then the TypedChannelBuffer::getStampedHeader()
        // of that class would have been called instead of us.
        Slot* slot = static_cast<Slot*>(s);
        return slot->header;
    }

    virtual const Buffer<uint8>& readSerializedValue(ChannelBufferBase::Slot* s)
    {
        return s->serializedValue;
    }

    virtual const Buffer<uint8>& readSerializedValue(ChannelBufferBase::Slot* s,
                                                     uint8 formatVersion, bool orLower)
    {
        BinaryBufferIstream stream(&(s->serializedValue));
        uint8 dataVersion = BinaryBufferDeserializer::getDataFormatVersion(stream);
        if ((!orLower && (dataVersion != formatVersion)) || (dataVersion > formatVersion)) {
            MIRA_THROW(XIO, "Untyped channel contains serialized data of version " << (int)dataVersion
                            << ", cannot return data of version " << (int)formatVersion << ".");
        }

        return readSerializedValue(s);
    }


    virtual Buffer<uint8> readSerializedValue(ChannelBufferBase::Slot* s,
                                              std::list<BinarySerializerCodecPtr>& codecs)
    {
        // we must ignore the codecs
        return s->serializedValue;
    }

    virtual Buffer<uint8> readSerializedValue(ChannelBufferBase::Slot* s,
                                              std::list<BinarySerializerCodecPtr>& codecs,
                                              uint8 formatVersion, bool orLower)
    {
        // we must ignore the codecs
        return readSerializedValue(s, formatVersion, orLower);
    }

    virtual void writeSerializedValue(ChannelBufferBase::Slot* s, Buffer<uint8> data)
    {
        s->serializedValue = std::move(data);
    }

    virtual void readJSON(ChannelBufferBase::Slot* s, JSONValue& oValue);
    virtual void readJSON(ChannelBufferBase::Slot* s, JSONValue& oValue, JSONSerializer& serializer);

    virtual void writeJSON(ChannelBufferBase::Slot* s, const JSONValue& value);
    virtual void writeJSON(ChannelBufferBase::Slot* s, JSONDeserializer& deserializer);
    void writeXML(ChannelBufferBase::Slot* s, const XMLDom::const_iterator& node) override;


private:

    Typename mTypename; 
};


class MIRA_FRAMEWORK_EXPORT PolymorphicChannelBuffer : public TypedChannelBufferBase<Object*>
{
    typedef TypedChannelBufferBase<Object*> Base;

public:

    PolymorphicChannelBuffer() :
        mMostDerivedClass(NULL),
        mFixatedClass(NULL) {}

public:
    virtual bool isPolymorphic() const {return true; }

    virtual int getTypeId() const;

    virtual Typename getTypename() const;

    virtual ChannelBufferBase::Slot* allocateSlot();

    virtual void freeSlot(ChannelBufferBase::Slot* s);

    virtual void fixateType();

    virtual void writeSerializedValue(ChannelBufferBase::Slot* s, Buffer<uint8> data);

    void promote(const Class* promotionClass);

protected:

    const Class* mMostDerivedClass;
    const Class* mFixatedClass;
};

#ifndef MIRA_WINDOWS
// declare template extern to speed up compilation time by several seconds
extern template class TypedChannelBufferBase<Object*>;
#endif


template <typename T>
class ChannelBuffer<T*> : public PolymorphicChannelBuffer
{
    static_assert(std::is_base_of<mira::Object,T>::value,
                  "In channels you can only use pointers of polymorphic "
                  "classes that are derived from mira::Object. Pointers to "
                  "other classes cannot be stored in a channel.");
public:
    typedef Stamped<T*> ValueType;

public:
    ChannelBuffer() {
        mMostDerivedClass = &T::CLASS();
    }
};

// Channel buffer promotion code

template <typename U>
struct ChannelBufferPromoterCommon
{
    static ChannelBufferBase* promoteUntyped(ChannelBufferBase* buffer)
    {
        assert(!buffer->isTyped() &&
               "Can promote untyped channels to typed channels only");

        if(!buffer->getTypename().empty() && buffer->getTypename()!=typeName<U>())
            MIRA_THROW(XBadCast, "Invalid promotion from untyped to typed "
                       "ChannelBuffer. Typename does not match. ('" <<
                       buffer->getTypename() << "' != '" << typeName<U>() << "'");

        typedef ChannelBufferBase::Slot Slot;

        // create a new typed buffer
        ChannelBuffer<U>* typedBuffer = new ChannelBuffer<U>;

        // copy all parameters
        typedBuffer->cloneParameters(*buffer);

        // copy all the serialized data from the untyped buffer into the
        // typed buffer and deserialize it into the value

        // now lock the whole buffer for reading ...
        for(Slot* s=ChannelBufferBase::begin(buffer->mBuffer);
                s!=ChannelBufferBase::end(buffer->mBuffer); s=s->next)
            s->lock.lock_shared();

        // ... and deserialize the content of each slot into the new buffer
        for(Slot* s=ChannelBufferBase::begin(buffer->mBuffer);
                s!=ChannelBufferBase::end(buffer->mBuffer); s=s->next)
        {
            Slot* n = typedBuffer->requestWriteSlot(); // obtain a new slot

            // ... copy the headers
            typedBuffer->getStampedHeader(n) = buffer->getStampedHeader(s);

            // ... and write the value into the new slot (this will deserialize the data there)
            typedBuffer->writeSerializedValue(n, s->serializedValue);

            typedBuffer->finishWriteSlot(n);
        }

        // free the read lock from above ...
        for(Slot* s=ChannelBufferBase::begin(buffer->mBuffer);
            s!=ChannelBufferBase::end(buffer->mBuffer); s=s->next)
            s->lock.unlock_shared();

        return typedBuffer;
    }
};


template <typename U>
struct ChannelBufferPromoter : public ChannelBufferPromoterCommon<U>
{
    typedef ChannelBufferPromoterCommon<U> Base;
    static ChannelBufferBase* promote(ChannelBufferBase* buffer)
    {
        if(buffer->isTyped())
            return buffer;  // we are promoted already, do nothing
        return Base::promoteUntyped(buffer);
    }
};


template <typename U>
struct ChannelBufferPromoter<U*>
{
    typedef ChannelBufferPromoterCommon<U> Base;
    static ChannelBufferBase* promote(ChannelBufferBase* buffer)
    {
        // if untyped, promote to typed first
        if(!buffer->isTyped())
            buffer = Base::promoteUntyped(buffer);

        PolymorphicChannelBuffer* polymorphicBuffer = dynamic_cast<PolymorphicChannelBuffer*>(buffer);

        assert(polymorphicBuffer!=NULL &&
               "We should never reach here if the buffer to promote is not "
               "polymorphic");

        // get class of U
        const Class* classU = &U::CLASS();

        // promote us (this will do error checking and might throw XBadCast)
        polymorphicBuffer->promote(classU);

        return polymorphicBuffer;
    }
};


template<typename U>
ChannelBufferBase* ChannelBufferBase::promote()
{
    return ChannelBufferPromoter<U>::promote(this);
}


}

#endif