Home

nevrax.com

Nevrax.org News Mailing-list Documentation CVS Bugs License

Documentation                       Search   buf_server.h Go to the documentation of this file. /* Copyright, 2001 Nevrax Ltd. * * This file is part of NEVRAX NEL. * NEVRAX NEL is free software; you can redistribute it and/or modify * it under the terms of the GNU General Public License as published by * the Free Software Foundation; either version 2, or (at your option) * any later version. * NEVRAX NEL 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 * General Public License for more details. * You should have received a copy of the GNU General Public License * along with NEVRAX NEL; see the file COPYING. If not, write to the * Free Software Foundation, Inc., 59 Temple Place - Suite 330, Boston, * MA 02111-1307, USA. */ #ifndef NL_BUF_SERVER_H #define NL_BUF_SERVER_H #include "nel/misc/types_nl.h" #include "nel/net/buf_net_base.h" #include "nel/net/listen_sock.h" #include "nel/net/buf_sock.h" #include <list> #include <set> // POLL1 (ignore POLL comments) namespace NLNET { class CInetAddress; class CBufServer; class CServerTask { public: virtual ~CServerTask(); void requireExit() { _ExitRequired = true; } #ifdef NL_OS_UNIX void wakeUp(); #endif protected: CServerTask(); bool exitRequired() const { return _ExitRequired; } #ifdef NL_OS_UNIX int _WakeUpPipeHandle [2]; #endif private: volatile bool _ExitRequired; }; class CListenTask : public NLMISC::IRunnable, public CServerTask { public: CListenTask( CBufServer *server ) : CServerTask(), _Server(server) {} void init( uint16 port ); virtual void run(); void close() { _ListenSock.close(); } const CInetAddress& localAddr() { return _ListenSock.localAddr(); } private: CBufServer *_Server; CListenSock _ListenSock; }; typedef std::vector<NLMISC::IThread*> CThreadPool; // Mode: Small server #undef PRESET_BIG_SERVER #ifdef PRESET_BIG_SERVER // Big server #define DEFAULT_STRATEGY SpreadSockets #define DEFAULT_MAX_THREADS 64 #define DEFAULT_MAX_SOCKETS_PER_THREADS 64 #else // Small server #define DEFAULT_STRATEGY FillThreads #define DEFAULT_MAX_THREADS 64 #define DEFAULT_MAX_SOCKETS_PER_THREADS 16 #endif class CBufServer : public CBufNetBase { public: enum TThreadStategy { SpreadSockets, FillThreads }; CBufServer( TThreadStategy strategy=DEFAULT_STRATEGY, uint16 max_threads=DEFAULT_MAX_THREADS, uint16 max_sockets_per_thread=DEFAULT_MAX_SOCKETS_PER_THREADS, bool nodelay=true, bool replaymode=false ); virtual ~CBufServer(); void init( uint16 port ); void disconnect( TSockId hostid, bool quick=false ); void setConnectionCallback( TNetCallback cb, void* arg ) { _ConnectionCallback = cb; _ConnectionCbArg = arg; } //void send( const std::vector<uint8>& buffer, TSockId hostid ); void send( const NLMISC::CMemStream& buffer, TSockId hostid ); bool dataAvailable(); //void receive( std::vector<uint8>& buffer, TSockId* hostid ); void receive( NLMISC::CMemStream& buffer, TSockId* hostid ); void update(); // Returns the size in bytes of the data stored in the send queue. uint32 getSendQueueSize( TSockId destid ); void setTimeFlushTrigger( TSockId destid, sint32 ms ) { nlassert( destid != InvalidSockId ); destid->setTimeFlushTrigger( ms ); } void setSizeFlushTrigger( TSockId destid, sint32 size ) { nlassert( destid != InvalidSockId ); destid->setSizeFlushTrigger( size ); } bool flush( TSockId destid ) { nlassert( destid != InvalidSockId ); return destid->flush(); } const CInetAddress& listenAddress() const { return _ListenTask->localAddr(); } const CInetAddress& hostAddress( TSockId hostid ) { nlassert( hostid != InvalidSockId ); return hostid->Sock->remoteAddr(); } /* uint32 bytesDownloaded() { NLMISC::CSynchronized<uint32>::CAccessor syncbpi ( &_BytesPushedIn ); return syncbpi.value(); } uint32 newBytesDownloaded(); */ uint64 bytesReceived() const { return _BytesPoppedIn; } uint64 newBytesReceived(); uint64 bytesSent() const { return _BytesPushedOut; } uint64 newBytesSent(); uint32 nbConnections() const { return _NbConnections; } protected: friend class CServerBufSock; friend class CListenTask; friend class CServerReceiveTask; bool noDelay() const { return _NoDelay; } void dispatchNewSocket( CServerBufSock *bufsock ); CServerReceiveTask *receiveTask( std::vector<NLMISC::IThread*>::iterator ipt ) { return ((CServerReceiveTask*)((*ipt)->getRunnable())); } /*void pushBufferToHost( const std::vector<uint8>& buffer, TSockId hostid ) { if ( hostid->pushBuffer( buffer ) ) { _BytesPushedOut += buffer.size() + sizeof(TBlockSize); // statistics } }*/ void pushBufferToHost( const NLMISC::CMemStream& buffer, TSockId hostid ) { nlassert( hostid != InvalidSockId ); if ( hostid->pushBuffer( buffer ) ) { _BytesPushedOut += buffer.length() + sizeof(TBlockSize); // statistics } } // Creates a new task and run a new thread for it void addNewThread( CThreadPool& threadpool, CServerBufSock *bufsock ); TNetCallback connectionCallback() const { return _ConnectionCallback; } void* argOfConnectionCallback() const { return _ConnectionCbArg; } /* NLMISC::CSynchronized<uint32>& syncBytesPushedIn() { return _BytesPushedIn; } */ private: bool _NoDelay; TThreadStategy _ThreadStrategy; uint16 _MaxThreads; uint16 _MaxSocketsPerThread; CListenTask *_ListenTask; NLMISC::IThread *_ListenThread; /* Vector of receiving threads. * Thread: thread control * Thread->Runnable: access to the CServerReceiveTask object * Thread->getRunnable()->sock(): access to the socket * The purpose of this list is to delete the objects after use. */ NLMISC::CSynchronized<CThreadPool> _ThreadPool; TNetCallback _ConnectionCallback; void* _ConnectionCbArg; uint64 _BytesPushedOut; uint64 _BytesPoppedIn; uint64 _PrevBytesPoppedIn; uint64 _PrevBytesPushedOut; uint32 _NbConnections; bool _ReplayMode; /* NLMISC::CSynchronized<uint32> _BytesPushedIn; uint32 _PrevBytesPushedIn; */ }; typedef std::set<TSockId> CConnections; /* // Workaround for Internal Compiler Error in debug mode with MSVC6 #ifdef NL_RELEASE typedef CConnections::iterator ItConnection; #else typedef TSockId ItConnection; #endif // Set of iterators to connections (set because we must not add an element twice) typedef std::set<ItConnection> CItConnections; */ #ifdef NL_OS_UNIX enum TPipeWay { PipeRead, PipeWrite }; #endif // POLL2 class CServerReceiveTask : public NLMISC::IRunnable, public CServerTask { public: CServerReceiveTask( CBufServer *server ) : CServerTask(), _Server(server), _Connections("CServerReceiveTask::_Connections"), _RemoveSet("CServerReceiveTask::_RemoveSet") {} virtual void run(); uint numberOfConnections() { uint nb; { NLMISC::CSynchronized<CConnections>::CAccessor connectionssync( &_Connections ); nb = connectionssync.value().size(); } return nb; } void addNewSocket( TSockId sockid ) { //nlnettrace( "CServerReceiveTask::addNewSocket" ); nlassert( sockid != InvalidSockId ); { NLMISC::CSynchronized<CConnections>::CAccessor connectionssync( &_Connections ); connectionssync.value().insert( sockid ); } // POLL3 } // POLL4 void addToRemoveSet( TSockId sockid ) { nlnettrace( "CServerReceiveTask::addToRemoveSet" ); nlassert( sockid != InvalidSockId ); { // Three possibilities : // - The value is inserted into the set. // - The value is already present in the set. // - The set is locked by a receive thread which is removing the closed connections. // When the set gets unlocked, it is empty so the value is inserted. It means the // value could be already in the set before it was cleared. // Note: with a fonction such as tryAcquire(), we could avoid to enter the mutex // when it is already locked // See clearClosedConnections(). NLMISC::CSynchronized<CConnections>::CAccessor removesetsync( &_RemoveSet ); removesetsync.value().insert( sockid ); //nldebug( "LNETL1: ic: %p - RemoveSet.size(): %d", ic, removesetsync.value().size() ); } #ifdef NL_OS_UNIX wakeUp(); #endif } void clearClosedConnections(); CBufServer *server() { return _Server; } friend class CBufServer; private: CBufServer *_Server; /* List of sockets and send buffer. * A TSockId is a pointer to a CBufSock object */ NLMISC::CSynchronized<CConnections> _Connections; // Connections to remove NLMISC::CSynchronized<CConnections> _RemoveSet; // POLL5 }; } // NLNET #endif // NL_BUF_SERVER_H /* End of buf_server.h */