RemoteConnection.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_REMOTECONNECTION_H_
#define _MIRA_REMOTECONNECTION_H_

#ifndef Q_MOC_RUN
#include <boost/array.hpp>
#include <boost/make_shared.hpp>
#endif

#include <utils/UUID.h>
#include <communication/IOService.h>
#include <error/Exceptions.h>

#include <rpc/RPCManager.h>

#include <fw/RemoteAuthority.h>
#include <fw/RemoteConnectionPool.h>

#include <fw/AuthorityDescription.h>
#include <fw/Channel.h>
#include <fw/FrameworkMessage.h>
#include <fw/ServiceLevel.h>

namespace mira {


class Authority;
class MicroUnit;

struct KnownFramework
{
    KnownFramework() :
        address(""),
        keep(true),
        forcePTP(false),
        binaryFormatVersion(BinaryBufferSerializer::getSerializerFormatVersion()),
        monitorOnly(false),
        lastConnectionTry(Time::unixEpoch()) {}

    template<typename Reflector>
    void reflect(Reflector& r)
    {
        r.member("Address", address, "");
        r.roproperty("Address", address, "The address IP:port");

        r.member("KeepConnected", keep, "", true);
        r.roproperty("KeepConnected", keep, "Whether to reconnect after a lost connection");

        r.member("ForcePTP", forcePTP, "", false);
        r.roproperty("ForcePTP", forcePTP, "Whether to force PTP time sync for this connection");

        r.member("BinaryFormatVersion", binaryFormatVersion, "",
                 BinaryBufferSerializer::getSerializerFormatVersion());
        r.roproperty("BinaryFormatVersion", binaryFormatVersion,
                     "Binary format version used by the remote framework");

        r.member("MonitorOnly", monitorOnly, "", false);
        r.roproperty("MonitorOnly", monitorOnly,
                     "Monitor-only connections do not publish local channels, "
                     "services and authorities to remote framework");

        r.member("LastConnectionTry", lastConnectionTry, "",
                 serialization::IgnoreMissing());
        r.roproperty("LastConnectionTry", lastConnectionTry,
                     "Last time tried to establish connection (UTC)");
    }

    void validate() const;
    boost::tuple<std::string, uint16> getHostPort() const;

    std::string address;

    bool keep;

    bool forcePTP;

    uint8 binaryFormatVersion;

    bool monitorOnly;

    Time lastConnectionTry;
};

template <typename SerializerTag>
class IsTransparentSerializable<KnownFramework, SerializerTag> : public std::true_type {};


class TimeOffsetCompensation
{
public:
    TimeOffsetCompensation(const TimeOffsetCompensation& other)
        : mCompensationInterval(other.mCompensationInterval),
          mTargetOffset(other.mTargetOffset),
          mStartOffset(other.mStartOffset),
          mStartTime(other.mStartTime) {}

    explicit TimeOffsetCompensation(const Duration& compensationInterval = Duration::seconds(10))
        : mCompensationInterval(compensationInterval.totalMicroseconds()),
          mTargetOffset(Duration::invalid()),
          mStartOffset(Duration::invalid()),
          mStartTime(Time::invalid()) {}

    template<typename Reflector>
    void reflect(Reflector& r)
    {
        r.property("CompensationInterval",
                   getter(&TimeOffsetCompensation::getCompensationInterval, this),
                   setter(&TimeOffsetCompensation::setCompensationInterval, this),
                   "Interval to reach target offset");
        r.roproperty("TargetOffset", mTargetOffset, "Target offset");
        r.roproperty("StartOffset", mStartOffset, "Start offset");
        r.roproperty("StartTime", mStartTime, "Time when start offset was valid");
        r.roproperty("CurrentOffset",
                     getter(&TimeOffsetCompensation::currentOffset, this),
                     "Current offset (interpolated between start and target offset)");
    }

public:
    bool isInitialized()const { return mTargetOffset.isValid(); }

public:
    Duration getCompensationInterval() const;
    void setCompensationInterval(const Duration& interval);

    void setTargetOffset(const Duration& target, const Time& localTs);

public:
    Duration queryLocalOffset(const Time& localTs) const;

    Duration queryRemoteOffset(const Time& remoteTs) const;

public:
    // assignment operator uses copy constructor (through by-value argument)
    TimeOffsetCompensation& operator=(TimeOffsetCompensation other) {
        using std::swap;
        swap(mCompensationInterval, other.mCompensationInterval);
        swap(mTargetOffset, other.mTargetOffset);
        swap(mStartOffset, other.mStartOffset);
        swap(mStartTime, other.mStartTime);
        return *this;
    }

protected:
    Duration currentOffset() const { return queryLocalOffset(Time::now()); }

    Duration offset(const Time& ts, bool local) const;

public:
    int mCompensationInterval; // total microseconds
    Duration mTargetOffset;
    Duration mStartOffset;
    Time mStartTime;

    mutable boost::mutex mMutex;
};


class RemoteConnection : public Object
{
    MIRA_ABSTRACT_OBJECT(RemoteConnection)
public:

    typedef std::map<std::string, Typename> ChannelTypeMap;
    typedef std::set<std::string> StringSet;

    struct SendData {
        uint8 metaVersion;
        Time lastData;
    };
    typedef std::map<std::string, SendData> ChannelSendMap;

    typedef std::set<std::string> MetaSet;
    typedef std::list<AuthorityDescription> AuthorityDescriptions;


    class RPCRemoteFinishHandler : public RPCManager::RemoteFinishHandler
    {
    public:
        RPCRemoteFinishHandler(RemoteConnection* iConnection) :
            mConnection(iConnection)
        {}

        virtual void onRPCfinished(Buffer<uint8>&& answer);

        void setConnection(RemoteConnection* iConnection);

    protected:

        boost::mutex mConnectionMutex;
        RemoteConnection* mConnection;
    };


    class RPCRemoteRequestHandler : public RPCManager::RemoteRequestHandler
    {
    public:
        RPCRemoteRequestHandler(RemoteConnection* iConnection) :
            mConnection(iConnection)
        {}

        virtual void onRPCrequested(Buffer<uint8>&& request);

        void setConnection(RemoteConnection* iConnection);

    protected:

        boost::mutex mConnectionMutex;
        RemoteConnection* mConnection;
    };


protected:
    RemoteConnection();

    RemoteConnection(boost::asio::io_service& service);

public:
    virtual ~RemoteConnection();

    template<typename Reflector>
    void reflect(Reflector& r)
    {
        r.roproperty("RemoteFrameworkName", frameworkID, "ID/Name of the remote framework");
        r.roproperty("RemoteProtocolVersion",
                     getter<std::string>( [&]()->std::string {
                         return MakeString() << MIRA_MAJOR_VERSION(remoteVersion) << "."
                                             << MIRA_MINOR_VERSION(remoteVersion); }),
                     "Protocol version of the remote framework");
        r.roproperty("Connected", synchronizedTime, "Time when the connection was established");
        r.roproperty("PTPSync", getter(&RemoteConnection::isPTPSyncEnabled, this),
                     "Synchronization of clocks via PTP enabled?");
        r.roproperty("PingTimeout", getter(&RemoteConnection::isPingTimeoutEnabled, this),
                     "Ping timeout enabled?");
        r.property("TimeSynch", clockOffset, "PTP time synchronization");
    }

    boost::asio::ip::tcp::socket& getSocket()
    {
        return mSocket;
    }

    virtual void start();

    void stop();

    void onConnect(bool enablePTPTimeSync, bool enablePingTimeout);

    bool isPTPSyncEnabled()
    {
        return mEnablePTPSync;
    }

    bool isPingTimeoutEnabled()
    {
        return mEnablePingTimeout;
    }

    void startTimeSync();


    bool isSynchronized() const
    {
        return synchronizedTime.isValid();
    }

    virtual void publishChannels(const ChannelTypeMap& channels) = 0;

    void unpublishChannel(const std::string& channel);

    virtual void subscribeChannel(const std::string& channelID, const ServiceLevel& serviceLevel) = 0;

    virtual void publishAuthorities(const AuthorityDescriptions& authorities) = 0;

    virtual void unpublishAuthorities(const AuthorityDescriptions& authorities) = 0;

    void migrateUnit(const std::string& id);

    bool hasAuthority(const std::string& id) const;

    virtual void publishServices(const StringSet& services) = 0;

    virtual void unpublishServices(const StringSet& services) = 0;

    template <typename BufferSequence>
    void write(const BufferSequence& buffers)
    {
        boost::mutex::scoped_lock lock(mWriteMutex);
        if (mStopped)
            return;
        try
        {
            std::size_t bytes = boost::asio::write(mSocket, buffers, boost::asio::transfer_all());
            updateOutgoingStats(bytes);
        }
        catch (boost::system::system_error& e)
        {
            // let the derived class decide what to do with the error
            onWriteError(e);
        }
    }

    UUID getRemoteID() const
    {
        return remoteID;
    }

    std::string getGlobalID() const
    {
        return authority->getGlobalID();
    }

    void setAuthority(std::unique_ptr<Authority> auth)
    {
        authority = std::move(auth);
    }

    const std::string& getFrameworkID() const
    {
        return frameworkID;
    }

    const KnownFramework& getAddress() const
    {
        return address;
    }

    void setAddress(const KnownFramework& addr)
    {
        address = addr;
    }

    void setAutoReconnect(bool autoReconnect)
    {
        address.keep = autoReconnect;
    }

protected:

    virtual void onDisconnect() {}

    virtual void onWriteError(boost::system::system_error& e) {}

    void syncTime();

    void sendPTP();

    void ping();

    bool hasPingTimeout() const;

    void writeMessage(FrameworkMessageType msg) {
        FrameworkMessageHeader header(0, msg);
        boost::array<boost::asio::const_buffer, 1> buffers =
        {{
            header.asBuffer(),
        }};
        write(buffers);
    }

    template <typename Arg, typename... Args>
    void writeMessage(FrameworkMessageType msg, Arg&& arg, Args&&... args)
    {
        BinaryBufferOstream::buffer_type buffer;
        BinaryBufferOstream os(&buffer);
        writeArgsToStream(os, std::forward<Arg>(arg), std::forward<Args>(args)...);
        writeMessageFromBuffer(msg, buffer);
    }

    template <typename BufferType>
    void writeMessageFromBuffer(FrameworkMessageType msg, const BufferType& buffer) {
        FrameworkMessageHeader header(buffer.size(), msg);
        boost::array<boost::asio::const_buffer, 2> buffers =
        {{
            header.asBuffer(),
            boost::asio::buffer(buffer.data(), buffer.size())
        }};
        write(buffers);
    }

    template <typename DataType>
    void writeMessageFromData(FrameworkMessageType msg, const DataType& data) {
        FrameworkMessageHeader header(sizeof(DataType), msg);
        boost::array<boost::asio::const_buffer, 2> buffers =
        {{
            header.asBuffer(),
            boost::asio::buffer((void*)&data, sizeof(DataType))
        }};
        write(buffers);
    }

    void queueRPCMessage(FrameworkMessageType msg, Buffer<uint8>&& answer);

    void valueChanged(ChannelRead<void> value, ServiceLevel& serviceLevel);

    void parseMessage();

    KnownFramework address;                  
    UUID remoteID;                           
    std::string frameworkID;                 
    uint32 remoteVersion;                    
    std::unique_ptr<Authority> authority;    
    ChannelSendMap subscriptions;            
    MetaSet sentMetaInformation;             
    StringSet publishedServices;             

    Time synchronizedTime;                    
    bool mEnablePTPSync;                      
    bool mEnablePingTimeout;                  
    TimeOffsetCompensation clockOffset;       

private:

    void init(); // called by constructors to initialize members

    bool isLocal() const;

    static void writeArgsToStream(BinaryBufferOstream& stream) {}

    template <typename Arg, typename... Args>
    static void writeArgsToStream(BinaryBufferOstream& stream, Arg&& arg, Args&&... args)
    {
        stream << std::forward<Arg>(arg);
        writeArgsToStream(stream, std::forward<Args>(args)...);
    }

protected:

    void receivedPTPFollowUp(uint64 timestamp);
    void receivedPTPDelayResponse(uint64 timestamp);
    void receivedPTPDelayRequest(uint64 timestamp);
    void receivedPTPFinish();
    void receivedSubscribeChannelRequest();
    void receivedUnsubscribeChannelRequest(const std::string& channelID);
    void receivedPublishChannelMsg();
    void receivedUnpublishChannelMsg();
    void receivedPublishAuthorityMsg();
    void receivedUnpublishAuthorityMsg();
    virtual void receivedPublishServiceMsg() = 0;
    void receivedUnpublishServiceMsg();
    void receivedWriteChannelMsg();
    virtual void receivedRPCRequestMsg() = 0;
    virtual void receivedRPCResponseMsg() = 0;
    void receivedRequestMigrationMsg();
    void receivedMigrationMsg();
    void receivedMigrationSinkSuccessMsg();
    void receivedMigrationSinkFailureMsg();
    void receivedMigrationFinishedMsg();
    void receivedTypeMetaMsg();
    void receivedChannelMetaMsg();
    void receivedPingMsg();

    // This method must not be pure virtual: When a remote incoming connection
    // times out and will be destroyed, some still pending valueChanged()
    // callbacks will be executed, which will call sendData. At this time, the
    // RemoteIncomingConnection is already destroyed (or will be destroyed) and
    // then sendData will be a pure-virtual function.
    // TODO: that still applies after r9735?
    virtual void sendData(ChannelRead<void> value, ServiceLevel& serviceLevel) {}

    virtual int addBinaryFormatVersion(Buffer<uint8>& data) = 0;
    void synchronizeFrameworks();
    void updateOutgoingStats(std::size_t size);

    bool checkMessageHeader() const;

    void sendConnectDenied(const std::string& msg);

    void sendRPCMessagesThread();
    void processPingThread();
    void checkPingTimeoutThread();
    void sendChannelUpdatesThread();

    IOService mService;
    boost::asio::ip::tcp::socket mSocket;

    boost::condition_variable mRPCMessagesCondition;

    boost::mutex mWriteMutex;
    boost::mutex mStopMutex;

    TimerPtr mSyncTimeTimer;

    boost::thread mSendRPCMessagesThread;
    boost::thread mProcessPingThread;
    boost::thread mCheckPingTimeoutThread;
    boost::thread mSendChannelUpdatesThread;

    Time mHeaderReceived;
    FrameworkMessageHeader mHeader;
    Buffer<uint8> mMessage;

    enum AuthState {
        AUTHSTATE_NONE,
        AUTHSTATE_CONNECTING,
        AUTHSTATE_AUTHENTICATING,
        AUTHSTATE_ACCEPTED,
        AUTHSTATE_DENIED,
    };
    AuthState mAuthState;
    std::string mAuthSignMsg; // random message that is used for strong authentication

    bool mPTPOutgoing;
    Time mPTPSyncLocal;
    Time mPTPSyncRemote;
    Time mPTPDelayLocal;
    Time mPTPDelayRemote;
    Time mLastPTP;

    Time mPingLastSend;
    Time mPingLastReceived;

    boost::shared_ptr<MicroUnit> mMigrationUnit;
    std::string mMigrationNS;
    std::string mMigrationID;
    bool mStopped;

    boost::shared_ptr<RPCRemoteFinishHandler> mRPCFinishHandler;
    boost::shared_ptr<RPCRemoteRequestHandler> mRPCRequestHandler;
    std::map<std::string, boost::shared_ptr<RemoteAuthority>> mRemoteAuthorities;

    typedef std::pair<FrameworkMessageType, Buffer<uint8>> RPCMessage;
    std::list<RPCMessage> mOutgoingRPCMessages;
    boost::mutex mRPCMessagesMutex;

    std::unordered_map<std::string, ServiceLevel> mPendingChannelUpdates;
    boost::mutex mChannelUpdatesMutex;
};

template <uint8 BinaryFormatVersion>
class ConcreteRemoteConnection : public virtual RemoteConnection
{
protected:
    ConcreteRemoteConnection() : RemoteConnection() {}

    ConcreteRemoteConnection(boost::asio::io_service& service) : RemoteConnection(service) {}

    friend void createConcreteRemoteConnectionInstances();

public:
    virtual ~ConcreteRemoteConnection() {};

protected:

    virtual void publishChannels(const ChannelTypeMap& channels);
    virtual void subscribeChannel(const std::string& channelID, const ServiceLevel& serviceLevel);
    virtual void publishAuthorities(const AuthorityDescriptions& authorities);
    virtual void unpublishAuthorities(const AuthorityDescriptions& authorities);
    virtual void publishServicesFiltered(const StringSet& services);
    virtual void publishServices(const StringSet& services);
    virtual void unpublishServicesFiltered(const StringSet& services);
    virtual void unpublishServices(const StringSet& services);
    virtual void sendData(ChannelRead<void> value, ServiceLevel& serviceLevel);
    virtual int addBinaryFormatVersion(Buffer<uint8>& data);

    virtual void receivedPublishServiceMsg();
    virtual void receivedRPCRequestMsg();
    virtual void receivedRPCResponseMsg();

private:

    typedef ConcreteBinarySerializer<BinaryBufferOstream, BinaryFormatVersion> BinarySerializer;
    boost::shared_ptr<BinarySerializer> createBinarySerializer(Buffer<uint8>* buffer)
    {
        return boost::make_shared<BinarySerializer>(buffer);
    }
};


class RemoteIncomingConnection : public ConcreteRemoteConnection<2>
{
    MIRA_OBJECT(RemoteIncomingConnection)

protected:
    RemoteIncomingConnection();
    friend class RemoteConnectionPool;
    friend class ClassFactoryDefaultConstClassBuilder;
public:
    template<typename Reflector>
    void reflect(Reflector& r)
    {
        static const std::string incoming = "Incoming";
        r.roproperty("ConnectDirection", incoming, "Incoming or outgoing connection?");
        r.roproperty("Address", address.address, "Origin address of the connection");
        MIRA_REFLECT_BASE(r, ConcreteRemoteConnection<2>);
    }

    virtual void start();

    boost::asio::ip::tcp::endpoint& getEndpoint() {
        return mPeerEndpoint;
    }

protected:

    virtual void onDisconnect();

protected:

    void handleReadHeader(const boost::system::error_code& error);

    void handleReadMessage(const boost::system::error_code& error);

protected:

    boost::asio::ip::tcp::endpoint mPeerEndpoint;
};


class RemoteOutgoingConnectionBase : public virtual RemoteConnection
{
    MIRA_ABSTRACT_OBJECT(RemoteOutgoingConnectionBase)
protected:
    RemoteOutgoingConnectionBase(const KnownFramework& address);

public:
    template<typename Reflector>
    void reflect(Reflector& r)
    {
        static const std::string outgoing = "Outgoing";
        r.roproperty("ConnectDirection", outgoing, "Incoming or outgoing connection?");
        r.roproperty("ConnectDetails", address, "Active connect details");
        MIRA_REFLECT_BASE(r, RemoteConnection);
    }

    virtual void start();

protected:
    virtual void onDisconnect();

    virtual void onWriteError(boost::system::system_error& e);

protected:

    void handleConnect(const boost::system::error_code& error,
                       boost::asio::ip::tcp::resolver::iterator iterator);

    void handleReadHeader(const boost::system::error_code& error);

    void handleReadMessage(const boost::system::error_code& error);

    std::string mHostName;
    uint16 mPort;
    bool mStopScheduled;
};

template <uint8 BinaryFormatVersion>
class ConcreteRemoteOutgoingConnection : public RemoteOutgoingConnectionBase,
                                         public ConcreteRemoteConnection<BinaryFormatVersion>
{
protected:
    ConcreteRemoteOutgoingConnection(const KnownFramework& address)
        : RemoteOutgoingConnectionBase(address) {}
    friend class RemoteConnectionPool;
    friend class ClassFactoryDefaultConstClassBuilder;
};

typedef ConcreteRemoteOutgoingConnection<2> RemoteOutgoingConnection;
typedef ConcreteRemoteOutgoingConnection<0> RemoteOutgoingConnectionLegacy;


}

#endif