lib/net/node.h

/* Copyright (c) 2005-2009, Stefan Eilemann <eile@equalizergraphics.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 EQNET_NODE_H
#define EQNET_NODE_H

#include <eq/net/dispatcher.h>               // base class
#include <eq/net/commandCache.h>             // member
#include <eq/net/commandQueue.h>             // member
#include <eq/net/connectionSet.h>            // member
#include <eq/net/nodeType.h>                 // for TYPE_EQNET_NODE enum
#include <eq/net/types.h>

#include <eq/base/base.h>
#include <eq/base/perThread.h>
#include <eq/base/requestHandler.h>
#include <eq/base/thread.h>

#include <list>

#pragma warning(disable : 4190)
extern "C" EQSERVER_EXPORT eq::net::ConnectionPtr eqsStartLocalServer( const
                                                                 std::string& );
extern "C" EQSERVER_EXPORT void                   eqsJoinLocalServer();
#pragma warning(default : 4190)

namespace eq
{
namespace net
{
    class Command;
    class ConnectionDescription;
    class Session;

    class Node : public Dispatcher, public base::Referenced
    {
    public:
        enum State 
        {
            STATE_STOPPED,   
            STATE_LAUNCHED,  
            STATE_CONNECTED, 
            STATE_LISTENING  
        };

        EQ_EXPORT Node();

        bool operator == ( const Node* n ) const;

        State getState()    const { return _state; }
        bool  isConnected() const 
            { return (_state == STATE_CONNECTED || _state == STATE_LISTENING); }

        void setAutoLaunch( const bool autoLaunch ) { _autoLaunch = autoLaunch;}

        EQ_EXPORT void setProgramName( const std::string& name );

        EQ_EXPORT void setWorkDir( const std::string& name );

        NodePtr getNode( const NodeID& id ) const;

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

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

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

        EQ_EXPORT virtual bool listen();

        EQ_EXPORT virtual bool stopListening();

        bool connect( NodePtr node, ConnectionPtr connection );

        EQ_EXPORT bool connect( NodePtr node );

        EQ_EXPORT bool initConnect( NodePtr node );

        EQ_EXPORT bool syncConnect( NodePtr node );

        NodePtr connect( const NodeID& nodeID );

        EQ_EXPORT bool disconnect( NodePtr node );


        bool isLocal() const { return (_state==STATE_LISTENING); }

        EQ_EXPORT void addConnectionDescription( ConnectionDescriptionPtr cd );
        
        void removeConnectionDescription( const uint32_t index );

        EQ_EXPORT const ConnectionDescriptionVector& getConnectionDescriptions()
                            const;
        ConnectionPtr getConnection() const { return _connection; }

        ConnectionPtr checkConnection()
            {
                ConnectionPtr connection = _connection;
                if( _state == STATE_CONNECTED || _state == STATE_LISTENING )
                    return _connection;
                return 0;
            }

        bool send( const Packet& packet )
            {
                ConnectionPtr connection = checkConnection();
                if( !connection )
                    return false;
                return connection->send( packet );
            }

        bool send( Packet& packet, const std::string& string )
            {
                ConnectionPtr connection = checkConnection();
                if( !connection )
                    return false;
                return connection->send( packet, string );
            }

        template< class T >
        bool send( Packet& packet, const std::vector<T>& data )
            {
                ConnectionPtr connection = checkConnection();
                if( !connection )
                    return false;
                return connection->send( packet, data );
            }

        bool send( Packet& packet, const void* data, const uint64_t size )
            {
                ConnectionPtr connection = checkConnection();
                if( !connection )
                    return false;
                return connection->send( packet, data, size );
            }

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

        void acquireSendToken( NodePtr toNode );
        void releaseSendToken( NodePtr toNode );

        EQ_EXPORT bool registerSession( Session* session );

        bool deregisterSession( Session* session )
            { return unmapSession( session ); }

        bool mapSession( NodePtr server, Session* session, 
                         const uint32_t id );

        EQ_EXPORT bool unmapSession( Session* session );

        Session* getSession( const uint32_t id );

        bool hasSessions() const { return !_sessions.empty(); }

        EQ_EXPORT virtual bool runClient( const std::string& clientArgs );

        CommandQueue* getCommandThreadQueue() 
            { EQASSERT( isLocal( )); return &_commandThreadQueue; }

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

        const NodeID& getNodeID() const { return _id; }

        EQ_EXPORT std::string serialize() const;
        EQ_EXPORT bool deserialize( std::string& data );

    protected:
        EQ_EXPORT virtual ~Node();

        EQ_EXPORT virtual bool dispatchCommand( Command& command );

        EQ_EXPORT virtual CommandResult invokeCommand( Command& command );

        virtual bool clientLoop() { return true; }

        virtual uint32_t getType() const { return TYPE_EQNET_NODE; }

        EQ_EXPORT virtual NodePtr createNode( const uint32_t type );

        base::RequestHandler _requestHandler;

    private:
        bool _autoLaunch;

        NodeID _id;

        State _state;

        SessionHash _sessions;

        ConnectionPtr _connection;

        ConnectionSet _connectionSet;
        friend eq::net::ConnectionPtr (::eqsStartLocalServer(const
                                                             std::string& ));

        base::UUIDHash< NodePtr > _nodes;

        typedef base::RefPtrHash< Connection, NodePtr > ConnectionNodeHash;
        ConnectionNodeHash _connectionNodes;

        CommandQueue _commandThreadQueue;

        base::Lock _connectMutex;

        uint32_t _launchID;

        base::Clock _launchTimeout;

        CommandList  _pendingCommands;

        CommandCache _commandCache;

        ConnectionDescriptionVector _connectionDescriptions;

        std::string _programName;
        std::string _workDir;

        class ReceiverThread : public base::Thread
        {
        public:
            ReceiverThread( Node* node ) 
                    : _node( node )
                {}
            
            virtual void* run(){ return _node->_runReceiverThread(); }

        private:
            Node* _node;
        };
        ReceiverThread* _receiverThread;

        class CommandThread : public base::Thread
        {
        public:
            CommandThread( Node* node ) 
                    : _node( node )
                {}
            
            virtual void* run(){ return _node->_runCommandThread(); }

        private:
            Node* _node;
        };
        CommandThread* _commandThread;

        bool _hasSendToken;

        bool _connectSelf();
        EQ_EXPORT void _addConnection( ConnectionPtr connection );
        void _removeConnection( ConnectionPtr connection );
        void _cleanup();

        void _dispatchCommand( Command& command );

        bool _launch( NodePtr node, ConnectionDescriptionPtr description );

        std::string _createLaunchCommand( NodePtr node,
                                          ConnectionDescriptionPtr description);
        std::string   _createRemoteCommand( NodePtr node, const char quote );

        NodePtr _findConnectedNode( const char* connectionDescription );

        void _addSession( Session* session, NodePtr server,
                          const uint32_t sessionID );

        void _removeSession( Session* session );

        uint32_t _generateSessionID();

        NodePtr _connect( const NodeID& nodeID, NodePtr server );

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

        void* _runCommandThread();
        void    _redispatchCommands();

        CommandResult _cmdStop( Command& command );
        CommandResult _cmdRegisterSession( Command& command );
        CommandResult _cmdRegisterSessionReply( Command& command );
        CommandResult _cmdMapSession( Command& command );
        CommandResult _cmdMapSessionReply( Command& command );
        CommandResult _cmdUnmapSession( Command& command );
        CommandResult _cmdUnmapSessionReply( Command& command );
        CommandResult _cmdConnect( Command& command );
        CommandResult _cmdConnectReply( Command& command );
        CommandResult _cmdDisconnect( Command& command );
        CommandResult _cmdGetNodeData( Command& command );
        CommandResult _cmdGetNodeDataReply( Command& command );
        CommandResult _cmdAcquireSendToken( Command& command );
        CommandResult _cmdAcquireSendTokenReply( Command& command );
        CommandResult _cmdReleaseSendToken( Command& command );

        CHECK_THREAD_DECLARE( _thread );
    };

    inline std::ostream& operator << ( std::ostream& os, const Node* node )
    {
        if( node )
            os << "node " << node->getNodeID();
        else
            os << "NULL node";
        
        return os;
    }

    EQ_EXPORT std::ostream& operator << ( std::ostream& os, 
                                          const Node::State state );
}
}

#endif // EQNET_NODE_H