Index: lib/net/packets.h =================================================================== --- lib/net/packets.h (revision 1663) +++ lib/net/packets.h (working copy) @@ -52,6 +52,15 @@ NodePacket(){ datatype = DATATYPE_EQNET_NODE; } }; + struct NodeNopPacket : public NodePacket + { + NodeNopPacket() + { + command = CMD_NODE_NOP; + size = sizeof( NodeNopPacket ); + } + }; + struct NodeStopPacket : public NodePacket { NodeStopPacket() Index: lib/net/commands.h =================================================================== --- lib/net/commands.h (revision 1663) +++ lib/net/commands.h (working copy) @@ -11,6 +11,7 @@ { enum NodeCommand { + CMD_NODE_NOP, CMD_NODE_STOP, CMD_NODE_MESSAGE, CMD_NODE_MAP_SESSION, Index: lib/net/session.cpp =================================================================== --- lib/net/session.cpp (revision 1663) +++ lib/net/session.cpp (working copy) @@ -103,7 +103,7 @@ //--------------------------------------------------------------------------- uint32_t Session::genIDs( const uint32_t range ) { - if( _isMaster && _server->inReceiverThread( )) + if( _isMaster && _server->inCommandThread( )) return _masterPool.genIDs( range ); uint32_t id = _localPool.genIDs( range ); @@ -202,7 +202,7 @@ { EQASSERT( object ); - if( _localNode->inReceiverThread( )) + if( _localNode->inCommandThread( )) { CHECK_THREAD( _receiverThread ); @@ -231,7 +231,7 @@ void Session::detachObject( Object* object ) { EQASSERT( object ); - if( _localNode->inReceiverThread( )) + if( _localNode->inCommandThread( )) { CHECK_THREAD( _receiverThread ); @@ -281,7 +281,7 @@ EQASSERT( object->_id == EQ_ID_INVALID ); EQASSERT( id != EQ_ID_INVALID ); - EQASSERT( !_localNode->inReceiverThread( )); + EQASSERT( !_localNode->inCommandThread( )); RefPtr master; if( !object->isMaster( )) @@ -325,7 +325,7 @@ if( object->getID() == EQ_ID_INVALID ) // not registered return; - EQASSERT( !_localNode->inReceiverThread( )); + EQASSERT( !_localNode->inCommandThread( )); SessionUnmapObjectPacket packet; packet.requestID = _requestHandler.registerRequest( object ); Index: lib/net/node.cpp =================================================================== --- lib/net/node.cpp (revision 1663) +++ lib/net/node.cpp (working copy) @@ -42,7 +42,11 @@ , _launchID( EQ_ID_INVALID ) , _programName( Global::getProgramName( )) , _workDir( Global::getWorkDir( )) + , _receiverThread( this ) + , _commandThread( this ) { + registerCommand( CMD_NODE_NOP, + CommandFunc( this, &Node::_cmdNop )); registerCommand( CMD_NODE_STOP, CommandFunc( this, &Node::_cmdStop )); registerCommand( CMD_NODE_MAP_SESSION, @@ -64,13 +68,11 @@ registerCommand( CMD_NODE_GET_NODE_DATA_REPLY, CommandFunc( this, &Node::_cmdGetNodeDataReply )); - _receiverThread = new ReceiverThread( this ); EQINFO << "New Node @" << (void*)this << " " << _id << endl; } Node::~Node() { - delete _receiverThread; EQINFO << "Delete Node @" << (void*)this << " " << _id << endl; } @@ -195,8 +197,10 @@ _state = STATE_LISTENING; - EQVERB << typeid(*this).name() << " starting recv thread " << endl; - _receiverThread->start(); + EQVERB << typeid(*this).name() << " starting command and receiver thread " + << endl; + _commandThread.start(); + _receiverThread.start(); EQINFO << this << " listening." << endl; return true; @@ -209,8 +213,13 @@ NodeStopPacket packet; send( packet ); - const bool joined = _receiverThread->join(); - EQASSERT( joined ); + + const bool recvJoined = _receiverThread.join(); + EQASSERT( recvJoined ); + + const bool cmdJoined = _commandThread.join(); + EQASSERT( cmdJoined ); + _cleanup(); return true; } @@ -307,7 +316,7 @@ if( !node || _state != STATE_LISTENING || node->_state != STATE_CONNECTED || !node->_connection ) return false; - EQASSERT( !inReceiverThread( )); + EQASSERT( !inCommandThread( )); NodeDisconnectPacket packet; packet.requestID = _requestHandler.registerRequest( node.get( )); @@ -358,7 +367,7 @@ const string& name ) { EQASSERT( isLocal( )); - EQASSERT( !_receiverThread->isCurrent( )); + EQASSERT( !inCommandThread( )); NodeMapSessionPacket packet; packet.requestID = _requestHandler.registerRequest( session ); @@ -484,7 +493,7 @@ //---------------------------------------------------------------------- // receiver thread functions //---------------------------------------------------------------------- -void* Node::_runReceiver() +void* Node::_runReceiverThread() { EQINFO << "Entered receiver thread" << endl; @@ -538,8 +547,15 @@ break; case ConnectionSet::EVENT_INTERRUPT: - _redispatchCommands(); + { + Command command; + NodeNopPacket packet; + command.allocate( this, this, packet.size ); + *(command.getPacket( )) = packet; + + _commandQueue.push( command ); break; + } default: EQUNIMPLEMENTED; @@ -550,24 +566,13 @@ nErrors = 0; } - if( !_pendingCommands.empty( )) - { - EQWARN << _pendingCommands.size() - << " commands rescheduled while leaving receiver thread" << endl; - - for( list::const_iterator i = _pendingCommands.begin(); - i != _pendingCommands.end(); ++i ) - - _commandCache.release( *i ); - - _pendingCommands.clear(); - } - _commandCache.flush(); delete _receivedCommand; _receivedCommand = 0; + EQINFO << _connectionSet.size() << " connections open during receiver exit" + << endl; EQINFO << "Leaving receiver thread" << endl; return EXIT_SUCCESS; } @@ -647,51 +652,99 @@ EQERROR << "Incomplete packet read: " << *_receivedCommand << endl; return false; } - // If no node is associated with the connection, the incoming packet has to - // be a one of the connection initialization packets - EQASSERTINFO( node.isValid() || - ((*_receivedCommand)->datatype == DATATYPE_EQNET_NODE && - ((*_receivedCommand)->command == CMD_NODE_CONNECT || - (*_receivedCommand)->command == CMD_NODE_CONNECT_REPLY )), - *_receivedCommand << " connection " << connection ); - - const CommandResult result = dispatchCommand( *_receivedCommand ); - switch( result ) - { - case COMMAND_ERROR: - EQASSERTINFO( 0, "Error handling " << *_receivedCommand ); - break; + + if( !node.isValid( )) + { + // This is one of the initial packets during the connection + // handshake. Handle them directly in the receiver thread since at this + // point the node is not yet available. + EQASSERTINFO( (*_receivedCommand)->datatype == DATATYPE_EQNET_NODE && + ( (*_receivedCommand)->command == CMD_NODE_CONNECT || + (*_receivedCommand)->command == CMD_NODE_CONNECT_REPLY), + *_receivedCommand << " connection " << connection ); - case COMMAND_REDISPATCH: - case COMMAND_HANDLED: - case COMMAND_DISCARD: - break; - - case COMMAND_PUSH: - if( !pushCommand( *_receivedCommand )) - EQASSERTINFO( 0, "Error handling command packet: pushCommand " - << "failed for " << *_receivedCommand << endl ); - break; + const CommandResult result = dispatchCommand( *_receivedCommand ); + EQASSERT( result == COMMAND_HANDLED ); - default: - EQUNIMPLEMENTED; + return true; } - _redispatchCommands(); + _commandQueue.push( *_receivedCommand ); + return true; +} + + +//---------------------------------------------------------------------- +// command thread functions +//---------------------------------------------------------------------- +void* Node::_runCommandThread() +{ + EQINFO << "Entered command thread" << endl; + + while( _state == STATE_LISTENING ) + { + Command* command = _commandQueue.pop(); + EQASSERT( command->isValid( )); + + const CommandResult result = dispatchCommand( *command ); + switch( result ) + { + case COMMAND_ERROR: + EQASSERTINFO( 0, "Error handling " << *command ); + break; + + case COMMAND_REDISPATCH: + case COMMAND_HANDLED: + case COMMAND_DISCARD: + break; + + case COMMAND_PUSH: + if( !pushCommand( *command )) + EQASSERTINFO( 0, + "Error handling command packet: pushCommand " + << "failed for " << *command << + endl ); + break; + + default: + EQUNIMPLEMENTED; + } - if( result == COMMAND_REDISPATCH ) + _redispatchCommands(); + + if( result == COMMAND_REDISPATCH ) + { + Command* dispCommand = _commandCache.alloc( *command ); + _pendingCommands.push_back( dispCommand ); + +#ifndef NDEBUG + if( (_pendingCommands.size() % 10) == 9 ) + EQINFO << _pendingCommands.size() << " commands pending" <::const_iterator i = _pendingCommands.begin(); + i != _pendingCommands.end(); ++i ) + + _commandCache.release( *i ); + + _pendingCommands.clear(); } - - return true; + + EQINFO << "Leaving command thread" << endl; + return EXIT_SUCCESS; } void Node::_redispatchCommands() { - bool changes = !_pendingCommands.empty(); - while( changes ) + bool changes = true; + while( changes && !_pendingCommands.empty( )) { changes = false; @@ -699,7 +752,8 @@ while( !changes && i != _pendingCommands.end( )) { Command* command = (*i); - + EQASSERT( command->isValid( )); + switch( dispatchCommand( *command )) { case COMMAND_HANDLED: @@ -779,6 +833,8 @@ EQASSERT( _state == STATE_LISTENING ); _state = STATE_STOPPED; + _connectionSet.interrupt(); + return COMMAND_HANDLED; } @@ -923,7 +979,7 @@ CommandResult Node::_cmdConnect( Command& command ) { EQASSERT( !command.getNode().isValid( )); - EQASSERT( inReceiverThread( )); + EQASSERT( _receiverThread.isCurrent( )); const NodeConnectPacket* packet = command.getPacket(); RefPtr connection = _connectionSet.getConnection(); @@ -987,7 +1043,7 @@ CommandResult Node::_cmdConnectReply( Command& command ) { EQASSERT( !command.getNode().isValid( )); - EQASSERT( inReceiverThread( )); + EQASSERT( _receiverThread.isCurrent( )); const NodeConnectReplyPacket* packet = command.getPacket(); @@ -1045,7 +1101,7 @@ CommandResult Node::_cmdDisconnect( Command& command ) { - EQASSERT( inReceiverThread( )); + EQASSERT( inCommandThread( )); const NodeDisconnectPacket* packet = command.getPacket(); @@ -1250,13 +1306,12 @@ if( iter != _nodes.end( )) return iter->second; - // Make sure that only one connection request based solely on the node - // identifier is pending at a given time. Otherwise a node with the same id - // might be instanciated twice in _cmdGetNodeDataReply(). The alternative to - // this mutex is to register connecting nodes with this local node, and - // handle all cases correctly, which is far more complex. Node connection - // only happen a lot during initialization, and are therefore not - // time-critical. + // Make sure that only one connection request based on the node identifier + // is pending at a given time. Otherwise a node with the same id might be + // instanciated twice in _cmdGetNodeDataReply(). The alternative to this + // mutex is to register connecting nodes with this local node, and handle + // all cases correctly, which is far more complex. Node connection only + // happens a lot during initialization, and are therefore not time-critical. ScopedMutex< Lock > mutex( _connectMutex ); iter = _nodes.find( nodeID ); Index: lib/net/commandQueue.h =================================================================== --- lib/net/commandQueue.h (revision 1663) +++ lib/net/commandQueue.h (working copy) @@ -37,18 +37,6 @@ virtual void push( Command& packet ); /** - * Push a command to the front of the queue. - * - * The command's command is incremented by one, which enables the usage - * of the same code for handling the arriving packet and the queued - * command, as they use different commands. - * - * @param node the node sending the packet. - * @param packet the command packet. - */ - virtual void pushFront( Command& packet ); - - /** * Pop a command from the queue. * * The returned packet is valid until the next pop operation. Index: lib/net/connectionSet.h =================================================================== --- lib/net/connectionSet.h (revision 1663) +++ lib/net/connectionSet.h (working copy) @@ -48,6 +48,7 @@ void addConnection( eqBase::RefPtr connection ); bool removeConnection( eqBase::RefPtr connection ); void clear(); + size_t size() const { return _connections.size(); } const ConnectionVector& getConnections() const { return _connections; } Index: lib/net/commandQueue.cpp =================================================================== --- lib/net/commandQueue.cpp (revision 1663) +++ lib/net/commandQueue.cpp (working copy) @@ -39,23 +39,14 @@ void CommandQueue::push( Command& inCommand ) { - _commandCacheLock.set(); - Command* outCommand = _commandCache.alloc( inCommand ); - _commandCacheLock.unset(); - - // Note: REQ must always follow CMD - ++(*outCommand)->command; - _commands.push( outCommand ); -} + EQASSERT( inCommand.isValid( )); -void CommandQueue::pushFront( Command& inCommand ) -{ _commandCacheLock.set(); Command* outCommand = _commandCache.alloc( inCommand ); _commandCacheLock.unset(); - ++(*outCommand)->command; // REQ must always follow CMD - _commands.pushFront( outCommand ); + EQASSERT( outCommand->isValid( )); + _commands.push( outCommand ); } Command* CommandQueue::pop() @@ -70,6 +61,7 @@ } _lastCommand = _commands.pop(); + EQASSERT( _lastCommand->isValid( )); return _lastCommand; } @@ -89,6 +81,7 @@ } _lastCommand = command; + EQASSERT( command->isValid( )); return command; } Index: lib/net/node.h =================================================================== --- lib/net/node.h (revision 1663) +++ lib/net/node.h (working copy) @@ -6,7 +6,7 @@ #define EQNET_NODE_H #include // base class -#include // member +#include // member #include // member #include // member #include // member @@ -445,10 +445,10 @@ virtual bool runClient( const std::string& clientArgs ); /** - * @return true if executed from the receiver thread, - * false if not. + * @return true if executed from the command handler + * thread, false if not. */ - bool inReceiverThread() const { return _receiverThread->isCurrent(); } + bool inCommandThread() const { return _commandThread.isCurrent(); } const NodeID& getNodeID() const { return _id; } @@ -540,6 +540,9 @@ /** The node for each connection. */ eqBase::PtrHash< Connection*, eqBase::RefPtr > _connectionNodes; + /** The receiver->command command queue. */ + CommandQueue _commandQueue; + /** Needed for thread-safety during nodeID-based connect() */ eqBase::Lock _connectMutex; @@ -622,20 +625,38 @@ : _node( node ) {} - virtual void* run(){ return _node->_runReceiver(); } + virtual void* run(){ return _node->_runReceiverThread(); } private: Node* _node; }; - ReceiverThread* _receiverThread; + ReceiverThread _receiverThread; - void* _runReceiver(); + /** The command handler thread. */ + class CommandThread : public eqBase::Thread + { + public: + CommandThread( Node* node ) + : _node( node ) + {} + + virtual void* run(){ return _node->_runCommandThread(); } + + private: + Node* _node; + }; + CommandThread _commandThread; + + void* _runReceiverThread(); void _handleConnect(); void _handleDisconnect(); bool _handleData(); + + void* _runCommandThread(); void _redispatchCommands(); /** The command functions. */ + CommandResult _cmdNop( Command& command ){ return COMMAND_HANDLED; } CommandResult _cmdStop( Command& command ); CommandResult _cmdMapSession( Command& command ); CommandResult _cmdMapSessionReply( Command& command ); Index: lib/client/node.cpp =================================================================== --- lib/client/node.cpp (revision 1663) +++ lib/client/node.cpp (working copy) @@ -311,10 +311,11 @@ //--------------------------------------------------------------------------- eqNet::CommandResult Node::_cmdCreatePipe( eqNet::Command& command ) { - CHECK_THREAD( _recvThread ); const NodeCreatePipePacket* packet = command.getPacket(); EQINFO << "Handle create pipe " << packet << endl; + + CHECK_THREAD( _cmdThread ); EQASSERT( packet->pipeID != EQ_ID_INVALID ); Pipe* pipe = Global::getNodeFactory()->createPipe( this ); @@ -325,11 +326,11 @@ eqNet::CommandResult Node::_cmdDestroyPipe( eqNet::Command& command ) { - CHECK_THREAD( _recvThread ); const NodeDestroyPipePacket* packet = command.getPacket(); EQINFO << "Handle destroy pipe " << packet << endl; + CHECK_THREAD( _cmdThread ); Pipe* pipe = _findPipe( packet->pipeID ); pipe->waitExit(); _config->detachObject( pipe ); @@ -340,11 +341,11 @@ eqNet::CommandResult Node::_reqConfigInit( eqNet::Command& command ) { - CHECK_NOT_THREAD( _recvThread ); const NodeConfigInitPacket* packet = command.getPacket(); EQINFO << "handle node config init " << packet << endl; + CHECK_NOT_THREAD( _cmdThread ); _name = packet->name; _currentFrame = 0; _unlockedFrame = 0; @@ -360,11 +361,11 @@ eqNet::CommandResult Node::_reqConfigExit( eqNet::Command& command ) { - CHECK_NOT_THREAD( _recvThread ); const NodeConfigExitPacket* packet = command.getPacket(); EQINFO << "handle node config exit " << packet << endl; + CHECK_NOT_THREAD( _cmdThread ); for( PipeVector::const_iterator i = _pipes.begin(); i != _pipes.end(); ++i ) { @@ -384,11 +385,11 @@ eqNet::CommandResult Node::_reqFrameStart( eqNet::Command& command ) { - CHECK_NOT_THREAD( _recvThread ); const NodeFrameStartPacket* packet = command.getPacket(); EQVERB << "handle node frame start " << packet << endl; + CHECK_NOT_THREAD( _cmdThread ); EQLOG( LOG_TASKS ) << "----- Begin Frame ----- " << packet->frameNumber << endl; frameStart( packet->frameID, packet->frameNumber ); @@ -401,11 +402,11 @@ eqNet::CommandResult Node::_reqFrameFinish( eqNet::Command& command ) { - CHECK_NOT_THREAD( _recvThread ); const NodeFrameFinishPacket* packet = command.getPacket(); EQVERB << "handle node frame finish " << packet << endl; + CHECK_NOT_THREAD( _cmdThread ); _finishFrame( packet->frameNumber ); frameFinish( packet->frameID, packet->frameNumber ); EQLOG( LOG_TASKS ) << "---- Finished Frame --- " << packet->frameNumber Index: lib/client/client.h =================================================================== --- lib/client/client.h (revision 1663) +++ lib/client/client.h (working copy) @@ -7,6 +7,7 @@ #include // member #include // for TYPE_EQ_CLIENT enum +#include // member #include // base class namespace eq @@ -79,7 +80,7 @@ virtual bool useMessagePump() { return true; } //*} private: - /** The receiver->node command queue. */ + /** The command->node command queue. */ CommandQueue* _commandQueue; bool _running; @@ -92,7 +93,7 @@ /** @sa eqNet::Node::pushCommand */ virtual bool pushCommand( eqNet::Command& command ) - { _commandQueue->push( command ); return true; } + { ++(command->command); _commandQueue->push( command ); return true; } /** The command functions. */ eqNet::CommandResult _reqExit( eqNet::Command& command ); Index: lib/client/commandQueue.h =================================================================== --- lib/client/commandQueue.h (revision 1663) +++ lib/client/commandQueue.h (working copy) @@ -24,9 +24,6 @@ /** @sa eqNet::CommandQueue::push(). */ virtual void push( eqNet::Command& packet ); - /** @sa eqNet::CommandQueue::pushFront(). */ - virtual void pushFront( eqNet::Command& packet ); - /** @sa eqNet::CommandQueue::pop(). */ virtual eqNet::Command* pop(); Index: lib/client/pipe.cpp =================================================================== --- lib/client/pipe.cpp (revision 1663) +++ lib/client/pipe.cpp (working copy) @@ -327,6 +327,7 @@ return eqNet::COMMAND_PUSH; // handled by main thread // else + ++(command->command); // increase command to REQ_ _commandQueue->push( command ); return eqNet::COMMAND_HANDLED; } Index: lib/client/commandQueue.cpp =================================================================== --- lib/client/commandQueue.cpp (revision 1663) +++ lib/client/commandQueue.cpp (working copy) @@ -67,13 +67,6 @@ _messagePump->postWakeup(); } -void CommandQueue::pushFront(eqNet::Command& inCommand) -{ - eqNet::CommandQueue::pushFront(inCommand); - if( _messagePump ) - _messagePump->postWakeup(); -} - eqNet::Command* CommandQueue::pop() { while( true ) Index: lib/client/node.h =================================================================== --- lib/client/node.h (revision 1663) +++ lib/client/node.h (working copy) @@ -278,7 +278,7 @@ eqNet::CommandResult _reqFrameFinish( eqNet::Command& command ); eqNet::CommandResult _reqFrameDrawFinish( eqNet::Command& command ); - CHECK_THREAD_DECLARE( _recvThread ); + CHECK_THREAD_DECLARE( _cmdThread ); }; } Index: server/server.h =================================================================== --- server/server.h (revision 1663) +++ server/server.h (working copy) @@ -1,5 +1,5 @@ -/* Copyright (c) 2005-2007, Stefan Eilemann +/* Copyright (c) 2005-2008, Stefan Eilemann All rights reserved. */ #ifndef EQS_SERVER_H @@ -9,6 +9,7 @@ #include "types.h" #include // for TYPE_EQ_SERVER enum +#include // used in inline method #include // member #include // member #include // base class & eqsStartLocalServer declaration @@ -79,7 +80,7 @@ /** @sa eqNet::Node::pushCommand */ virtual bool pushCommand( eqNet::Command& command ) - { _commandQueue.push( command ); return true; } + { ++(command->command); _commandQueue.push( command ); return true; } private: /** The unique config identifier. */