localNode.h

/* Copyright (c) 2005-2011, 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/base/hash.h>           // member
#include <co/base/lockable.h>       // member
#include <co/base/spinLock.h>       // member
#include <co/base/types.h>          // member

#pragma warning(push)
#pragma warning(disable: 4190)
extern "C" EQSERVER_EXPORT co::ConnectionPtr eqsStartLocalServer( const
                                                                 std::string& );
extern "C" EQSERVER_EXPORT void eqsJoinLocalServer();
#pragma warning(pop)

namespace co
{
    class ObjectStore;

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

        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 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 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 void acquireSendToken( NodePtr toNode );
        CO_API void releaseSendToken( NodePtr toNode );

        virtual CommandQueue* getCommandThreadQueue() 
            { return &_commandThreadQueue; }

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

        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 );

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

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

        CommandList  _pendingCommands;
    
        CommandCache _commandCache;

        CommandQueue _commandThreadQueue;        
    
        bool _hasSendToken;
        std::deque< Command* > _sendTokenQueue;

        ObjectStore* _objectStore;

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

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

        ConnectionSet _incoming;
    
        friend EQSERVER_EXPORT 
        co::ConnectionPtr (::eqsStartLocalServer( const std::string& ));

        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 );

        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 base::Thread
        {
        public:
            CommandThread( LocalNode* localNode ) : _localNode( localNode ){}
            virtual bool init()
                {
                    setName( std::string("Cmd ") + base::className(_localNode));
                    return true;
                }
            virtual void run(){ _localNode->_runCommandThread(); }
        private:
            LocalNode* const _localNode;
        };
        CommandThread* _commandThread;

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

        bool _cmdAckRequest( Command& packet );
        bool _cmdStop( 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 );

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