localNode.h

/* Copyright (c) 2005-2012, Stefan Eilemann <eile@equalizergraphics.com>
 *                    2010, Cedric Stalder <cedric.stalder@gmail.com> 
 *
 * 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/base/requestHandler.h> // base class

#include <co/commandCache.h>    // member
#include <co/commandQueue.h>    // member
#include <co/connectionSet.h>   // member
#include <co/objectVersion.h>   // used in inline method
#include <co/worker.h>          // member

#include <co/base/clock.h>          // member
#include <co/base/hash.h>           // member
#include <co/base/lockable.h>       // member
#include <co/base/spinLock.h>       // member
#include <co/base/types.h>          // member

namespace co
{
    class ObjectStore;

    class LocalNode : public base::RequestHandler, public Node
    {
    public:
        CO_API LocalNode( );
        CO_API virtual ~LocalNode( );

        typedef NodePtr SendToken; 

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

        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 virtual bool disconnect( NodePtr node );

        CO_API void disableInstanceCache();

        CO_API void expireInstanceData( const int64_t age );

        CO_API void enableSendOnRegister();
        
        CO_API void disableSendOnRegister(); 
        
        CO_API bool registerObject( Object* object );

        CO_API virtual void deregisterObject( Object* object );

        CO_API bool mapObject( Object* object, const base::UUID& id, 
                               const uint128_t& version = VERSION_OLDEST );

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

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

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

        CO_API bool mapObjectSync( const uint32_t requestID );

        CO_API void unmapObject( Object* object );

        CO_API void releaseObject( Object* object );

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

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

        CO_API NodePtr getNode( const NodeID& id ) const;

        void getNodes( Nodes& nodes, const bool addSelf = true ) const;

        CO_API SendToken acquireSendToken( NodePtr toNode );
        CO_API void releaseSendToken( SendToken& token );

        CommandQueue* getCommandThreadQueue()
            { return _commandThread->getWorkerQueue(); }

        bool inCommandThread() const  { return _commandThread->isCurrent(); }

        int64_t getTime64(){ return _clock.getTime64(); }

        CO_API void addListener( ConnectionPtr connection );

        CO_API uint32_t removeListenerNB( ConnectionPtr connection );

        void flushCommands() { _incoming.interrupt(); }

        Command& cloneCommand( Command& command )
            { return _commandCache.clone( command ); }

        CO_API Command& allocCommand( const uint64_t size );

        CO_API bool dispatchCommand( Command& command );

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

        CO_API void ping( NodePtr remoteNode );

        CO_API bool pingIdleNodes();

    protected:
        CO_API bool connect( NodePtr node, ConnectionPtr connection );

        virtual void notifyDisconnect( NodePtr node ) { }

    private:
        typedef std::list< Command* > CommandList;

        CommandList  _pendingCommands;

        CommandCache _commandCache;

        bool _hasSendToken;
        uint64_t _lastTokenTime; 
        std::deque< Command* > _sendTokenQueue; 

        ObjectStore* _objectStore;

        base::Lock _connectMutex;
    
        typedef base::RefPtrHash< Connection, NodePtr > ConnectionNodeHash;
        typedef ConnectionNodeHash::const_iterator ConnectionNodeHashCIter;
        ConnectionNodeHash _connectionNodes; // read and write: recv only

        typedef stde::hash_map< uint128_t, NodePtr > NodeHash;
        typedef NodeHash::const_iterator NodeHashCIter;
        // r: all, w: recv
        base::Lockable< NodeHash, base::SpinLock > _nodes; 

        ConnectionSet _incoming;

        base::Clock _clock;
    
        class ReceiverThread : public base::Thread
        {
        public:
            ReceiverThread( LocalNode* localNode ) : _localNode( localNode ){}
            virtual bool init()
                {
                    setName( std::string("Rcv ") + base::className(_localNode));
                    return _localNode->_commandThread->start();
                }
            virtual void run(){ _localNode->_runReceiverThread(); }

        private:
            LocalNode* const _localNode;
        };
        ReceiverThread* _receiverThread;

        bool _connectSelf();
        void _connectMulticast( NodePtr node );

        void _cleanup();
        CO_API void _addConnection( ConnectionPtr connection );
        void _removeConnection( ConnectionPtr connection );

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

        bool _inReceiverThread() const { return _receiverThread->isCurrent(); }

        void _receiverThreadStart() { _receiverThread->start(); }

        void _runReceiverThread();
        void   _handleConnect();
        void   _handleDisconnect();
        bool   _handleData();

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

        class CommandThread : public Worker
        {
        public:
            CommandThread( LocalNode* localNode ) : _localNode( localNode ){}

        protected:
            virtual bool init();
            virtual bool stopRunning() { return _localNode->isClosed(); }
            virtual bool notifyIdle();

        private:
            LocalNode* const _localNode;
        };
        CommandThread* _commandThread;

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

        bool _cmdAckRequest( Command& packet );
        bool _cmdStopRcv( Command& command );
        bool _cmdStopCmd( Command& command );
        bool _cmdConnect( Command& command );
        bool _cmdConnectReply( Command& command );
        bool _cmdConnectAck( Command& command );
        bool _cmdID( Command& command );
        bool _cmdDisconnect( Command& command );
        bool _cmdGetNodeData( Command& command );
        bool _cmdGetNodeDataReply( Command& command );
        bool _cmdAcquireSendToken( Command& command );
        bool _cmdAcquireSendTokenReply( Command& command );
        bool _cmdReleaseSendToken( Command& command );
        bool _cmdAddListener( Command& command );
        bool _cmdRemoveListener( Command& command );
        bool _cmdPing( Command& command );
        bool _cmdDiscard( Command& ) { return true; }

        EQ_TS_VAR( _cmdThread );
        EQ_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