localNode.h

/* Copyright (c) 2005-2016, Stefan Eilemann <eile@equalizergraphics.com>
 *                          Cedric Stalder <cedric.stalder@gmail.com>
 *                          Daniel Nachbaur <danielnachbaur@gmail.com>
 *
 * This file is part of Collage <https://github.com/Eyescale/Collage>
 *
 * This library is free software; you can redistribute it and/or modify it under
 * the terms of the GNU Lesser General Public License version 2.1 as published
 * by the Free Software Foundation.
 *
 * This library is distributed in the hope that it will be useful, but WITHOUT
 * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS
 * FOR A PARTICULAR PURPOSE.  See the GNU Lesser General Public License for more
 * details.
 *
 * You should have received a copy of the GNU Lesser General Public License
 * along with this library; if not, write to the Free Software Foundation, Inc.,
 * 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
 */

#ifndef CO_LOCALNODE_H
#define CO_LOCALNODE_H

#include <co/node.h>            // base class
#include <co/objectHandler.h>   // base class
#include <co/objectVersion.h>   // VERSION_FOO used inline
#include <lunchbox/requestHandler.h> // base class

#include <boost/function/function1.hpp>
#include <boost/function/function4.hpp>

namespace co
{
namespace detail { class LocalNode; class ReceiverThread; class CommandThread; }

class LocalNode : public lunchbox::RequestHandler, public Node,
                  public ObjectHandler
{
public:
    enum Counter
    {
        COUNTER_MAP_OBJECT_REMOTE, 
        COUNTER_ALL // must be last
    };

    CO_API explicit LocalNode( const uint32_t type = co::NODETYPE_NODE );

    CO_API virtual bool initLocal( const int argc, char** argv );

    CO_API virtual bool listen();

    CO_API virtual bool close();

    virtual bool exitLocal() { return close(); }

    CO_API bool connect( NodePtr node );

    CO_API NodePtr connect( const NodeID& nodeID );

    CO_API NodePtr connectObjectMaster( const uint128_t& id );

    CO_API virtual bool disconnect( NodePtr node );

    CO_API bool launch( NodePtr node, const std::string& command );

    CO_API NodePtr syncLaunch( const uint128_t& nodeID, int64_t timeout );

    CO_API bool registerObject( Object* object ) override;

    CO_API void deregisterObject( Object* object ) override;

    CO_API f_bool_t mapObject( Object* object, const uint128_t& id,
                               NodePtr master,
                               const uint128_t& version = VERSION_OLDEST );

    f_bool_t mapObject( Object* object, const ObjectVersion& v )
        { return mapObject( object, v.identifier, 0, v.version ); }

    f_bool_t mapObject( Object* object, const uint128_t& id,
                        const uint128_t& version = VERSION_OLDEST )
        { return mapObject( object, id, 0, version ); }

    CO_API uint32_t mapObjectNB( Object* object, const uint128_t& id,
                                 const uint128_t& version = VERSION_OLDEST );

    CO_API uint32_t mapObjectNB( Object* object, const uint128_t& id,
                                 const uint128_t& version,
                                 NodePtr master ) override;

    CO_API bool mapObjectSync( const uint32_t requestID ) override;

    CO_API f_bool_t syncObject( Object* object, const uint128_t& id,
                                NodePtr master,
                         const uint32_t instanceID = CO_INSTANCE_ALL ) override;
    CO_API void unmapObject( Object* object ) override;

    CO_API void disableInstanceCache();

    CO_API void expireInstanceData( const int64_t age );

    CO_API void enableSendOnRegister();

    CO_API void disableSendOnRegister();

    CO_API virtual void objectPush( const uint128_t& groupID,
                                    const uint128_t& objectType,
                                    const uint128_t& objectID,
                                    DataIStream& istream );

    typedef boost::function< void( const uint128_t&, 
                                   const uint128_t&, 
                                   const uint128_t&, 
                                   DataIStream& ) > PushHandler;
    CO_API void registerPushHandler( const uint128_t& groupID,
                                     const PushHandler& handler );


    typedef boost::function< bool( CustomICommand& ) > CommandHandler;

    CO_API bool registerCommandHandler( const uint128_t& command,
                                        const CommandHandler& func,
                                        CommandQueue* queue );

    CO_API void swapObject( Object* oldObject, Object* newObject );

    CO_API NodePtr getNode( const NodeID& id ) const;

    CO_API Nodes getNodes( const bool addSelf = true ) const;

    CO_API CommandQueue* getCommandThreadQueue();

    CO_API bool inCommandThread() const;

    CO_API const Strings& getCommandLine() const;

    CO_API int64_t getTime64() const; 
    CO_API ssize_t getCounter( const Counter counter ) const; 


    CO_API ConnectionPtr addListener( ConnectionDescriptionPtr desc );

    CO_API void addListener( ConnectionPtr connection );

    CO_API void removeListeners( const Connections& connections );

    CO_API void flushCommands();

    CO_API BufferPtr allocBuffer( const uint64_t size );

    CO_API bool dispatchCommand( ICommand& command ) override;


    typedef lunchbox::RefPtr< co::SendToken > SendToken;

    CO_API SendToken acquireSendToken( NodePtr toNode );

    CO_API void releaseSendToken( SendToken token );

    CO_API Zeroconf getZeroconf();

    CO_API void ackRequest( NodePtr node, const uint32_t requestID );

    CO_API void ping( NodePtr remoteNode );

    CO_API bool pingIdleNodes();

    CO_API void setAffinity( const int32_t affinity );

    CO_API void addConnection( ConnectionPtr connection );

protected:
    CO_API ~LocalNode() override;

    CO_API bool connect( NodePtr node, ConnectionPtr connection );

    virtual void notifyConnect( NodePtr ) {}

    virtual void notifyDisconnect( NodePtr ) {}

    CO_API virtual NodePtr createNode( const uint32_t type );

private:
    detail::LocalNode* const _impl;

    friend class detail::ReceiverThread;
    bool _startCommandThread( const int32_t threadID );
    void _runReceiverThread();

    friend class detail::CommandThread;
    bool _notifyCommandThreadIdle();

    void _cleanup();
    void _closeNode( NodePtr node );
    void _addConnection( ConnectionPtr connection );
    void _removeConnection( ConnectionPtr connection );

    lunchbox::Request< void > _removeListener( ConnectionPtr connection );

    uint32_t _connect( NodePtr node );
    NodePtr _connect( const NodeID& nodeID );
    uint32_t _connect( NodePtr node, ConnectionPtr connection );
    NodePtr _connect( const NodeID& nodeID, NodePtr peer );
    NodePtr _connectFromZeroconf( const NodeID& nodeID );
    bool _connectSelf();

    bool _setupPeer( const std::string& setupOpts );

    void _handleConnect();
    void _handleDisconnect();
    bool _handleData();
    BufferPtr _readHead( ConnectionPtr connection );
    ICommand _setupCommand( ConnectionPtr, ConstBufferPtr );
    bool _readTail( ICommand&, BufferPtr, ConnectionPtr );
    void _initService();
    void _exitService();

    friend class ObjectStore;
    template< typename T >
    void _registerCommand( const uint32_t command, const CommandFunc< T >& func,
                           CommandQueue* destinationQueue )
    {
        registerCommand( command, func, destinationQueue );
    }

    void _dispatchCommand( ICommand& command );
    void _redispatchCommands();

    bool _cmdAckRequest( ICommand& command );
    bool _cmdStopRcv( ICommand& command );
    bool _cmdStopCmd( ICommand& command );
    bool _cmdSetAffinity( ICommand& command );
    bool _cmdConnect( ICommand& command );
    bool _cmdConnectReply( ICommand& command );
    bool _cmdConnectAck( ICommand& command );
    bool _cmdID( ICommand& command );
    bool _cmdDisconnect( ICommand& command );
    bool _cmdGetNodeData( ICommand& command );
    bool _cmdGetNodeDataReply( ICommand& command );
    bool _cmdAcquireSendToken( ICommand& command );
    bool _cmdAcquireSendTokenReply( ICommand& command );
    bool _cmdReleaseSendToken( ICommand& command );
    bool _cmdAddListener( ICommand& command );
    bool _cmdRemoveListener( ICommand& command );
    bool _cmdPing( ICommand& command );
    bool _cmdCommand( ICommand& command );
    bool _cmdCommandAsync( ICommand& command );
    bool _cmdAddConnection( ICommand& command );
    bool _cmdDiscard( ICommand& ) { return true; }

    LB_TS_VAR( _cmdThread )
    LB_TS_VAR( _rcvThread )
};

inline std::ostream& operator << ( std::ostream& os, const LocalNode& node )
{
    os << static_cast< const Node& >( node );
    return os;
}
}
#endif // CO_LOCALNODE_H