rspConnection.h

/* Copyright (c) 2009, Cedric Stalder <[email protected]> 
 *               2009-2010, Stefan Eilemann <[email protected]>
 *
 * 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_RSPCONNECTION_H
#define EQNET_RSPCONNECTION_H

#include <eq/net/connection.h>

#ifdef EQ_USE_BOOST
#include <eq/net/connectionSet.h> // member
#include <eq/net/types.h>
#include "eventConnection.h" // member

#include <eq/base/base.h>
#include <eq/base/buffer.h>  // member
#include <eq/base/lfQueue.h> // member
#include <eq/base/mtQueue.h> // member

#include <boost/asio.hpp>

namespace eq
{
namespace net
{
    class ConnectionDescription;

    class RSPConnection : public Connection
    {
    public:
        RSPConnection();
        
        virtual bool listen();
        void close();

        bool connect(){ return listen(); }

        virtual void acceptNB(){ EQASSERT( _state == STATE_LISTENING ); }

        virtual ConnectionPtr acceptSync();
        virtual void readNB( void*, const uint64_t ) {/* NOP */}
        virtual int64_t readSync( void* buffer, const uint64_t bytes,
                                  const bool ignored );
        virtual int64_t write( const void* buffer, const uint64_t bytes );

        int64_t getSendRate() const { return _sendRate; }

        uint16_t getID() const { return _id; }
        
        virtual Notifier getNotifier() const { return _event->getNotifier(); }

    protected:
        virtual ~RSPConnection();
    
    private:
        typedef base::RefPtr< RSPConnection > RSPConnectionPtr;

        class Thread : public base::Thread
        {
        public: 
            Thread( RSPConnectionPtr connection )
                : _connection( connection ){}
            virtual ~Thread(){ _connection = 0; }
        protected:
            virtual void run();
            virtual bool init() { return _connection->_initThread(); }
            
        private:
            RSPConnectionPtr _connection;
        };

        enum DatagramType
        { 
            DATA,      
            ACKREQ,    
            NACK,      
            ACK,       
            ID_HELLO,  
            ID_DENY,   
            ID_CONFIRM,
            ID_EXIT,   
            COUNTNODE  
        };
        
        struct DatagramNode
        {
            uint16_t type;
            uint16_t connectionID;
        };

        struct DatagramCount
        {
            uint16_t type;
            uint16_t clientID;
            uint16_t numConnections;
        };

        struct DatagramAckRequest
        {
            uint16_t type;
            uint16_t writerID;
            uint16_t sequence;
        };
        
        struct Nack
        {
            uint16_t start;
            uint16_t end;
        };

#       define EQ_RSP_MAX_NACKS 300 // fits in a single IP frame

        struct DatagramNack
        {
            void set( uint16_t rID, uint16_t wID, uint16_t n )
            {
                type       = NACK;
                readerID   = rID; 
                writerID   = wID;   
                count      = n;
            }

            uint16_t       type;
            uint16_t       readerID;
            uint16_t       writerID;
            uint16_t       count;       
            Nack           nacks[ EQ_RSP_MAX_NACKS ];
        };

        struct DatagramAck
        {
            uint16_t        type;
            uint16_t        readerID;
            uint16_t        writerID;
            uint16_t        sequence;
        };

        struct DatagramData
        {
            uint16_t    type;
            uint16_t    size;
            uint16_t    writerID;
            uint16_t    sequence;
        };

        typedef std::vector< RSPConnectionPtr > RSPConnections;

        RSPConnectionPtr _parent;
        RSPConnections _children;

        // a link for all connection in the connecting state 
        RSPConnections _childrenConnecting;

        uint16_t _id; 
        bool     _idAccepted;
        int32_t  _mtu;
        int32_t  _ackFreq;
        uint32_t _payloadSize;
        int32_t  _timeouts;

        typedef base::RefPtr< EventConnection > EventConnectionPtr;
        EventConnectionPtr _event;

        boost::asio::io_service        _ioService;
        boost::asio::ip::udp::socket*  _read;
        boost::asio::ip::udp::socket*  _write;
        boost::asio::ip::udp::endpoint _readAddr;
        boost::asio::deadline_timer    _timeout;
        boost::asio::deadline_timer    _wakeup;
        
        eq::base::Clock _clock;
        uint64_t        _maxBucketSize;
        size_t          _bucketSize;
        int64_t         _sendRate;

        Thread*          _thread;
        base::Lock       _mutexConnection;
        base::Lock       _mutexEvent;
        uint16_t         _acked;        // sequence ID of last confirmed ack

        typedef base::Bufferb Buffer;
        typedef std::vector< Buffer* > Buffers;

        Buffers _buffers;                   

        base::LFQueue< Buffer* > _threadBuffers;
        base::MTQueue< Buffer* > _appBuffers;

        Buffer _recvBuffer;                      
        std::deque< Buffer* > _recvBuffers;      

        Buffer* _readBuffer;                     
        uint64_t _readBufferPos;                 

        uint16_t _sequence; 
        std::deque< Buffer* > _writeBuffers;    

        typedef std::deque< Nack > RepeatQueue;
        RepeatQueue _repeatQueue; 

        void _close();
        uint16_t _buildNewID();
        
        void _processOutgoing();
        void _writeData();
        void _repeatData();
        void _finishWriteQueue( const uint16_t sequence );

        bool _handleData( Buffer& buffer );
        bool _handleAck( const DatagramAck* ack );
        bool _handleNack( const DatagramNack* nack );
        bool _handleAckRequest( const DatagramAckRequest* ackRequest );

        bool _handleCountNode();

        Buffer* _newDataBuffer( Buffer& inBuffer );
        void _pushDataBuffer( Buffer* buffer );

        /* Run the reader thread */
        void _runThread();

        /* init the reader thread */
        bool _initThread();
        /* Make all buffers available for reading */
        void initBuffers();
        /* handle data about the comunication state */ 
        void _handlePacket( const boost::system::error_code& error,
                            const size_t bytes );
        void _handleConnectedData( const void* data );
        void _handleInitData( const void* data );
        void _handleAcceptIDData( const void* data );

        /* handle timeout about the comunication state */
        void _handleTimeout( const boost::system::error_code& error );
        void _handleConnectedTimeout( );
        void _handleInitTimeout( );
        void _handleAcceptIDTimeout( );

        RSPConnectionPtr _findConnection( const uint16_t id );
        
        void _waitWritable( const uint64_t bytes );

        void _sendDatagramCountNode();

        void _addRepeat( const Nack* nacks, const uint16_t num );

        void _sendSimpleDatagram( DatagramType type, uint16_t id );
        
        void _sendAckRequest();

        void _sendAck( const uint16_t writerID, const uint16_t sequence );
        
        void _sendNack( const uint16_t toWriterID, const Nack* nacks,
                        const uint16_t num );
        
        void _checkNewID( const uint16_t id );

        /* add a new connection detected in the multicast network */
        bool _addNewConnection( const uint16_t id );
        void _removeConnection( const uint16_t id );

        void _setTimeout( const int32_t timeOut );
        void _postWakeup();
        void _asyncReceiveFrom();
        bool _isWriting()
            { return !_threadBuffers.isEmpty() || !_writeBuffers.empty( ); }
    };

    std::ostream& operator << ( std::ostream&, const RSPConnection& );
}
}

#endif //EQNET_RSPCONNECTION_H
#endif //EQ_USE_BOOST