//----------------------------------------------------------------------
// Connecting a node
//----------------------------------------------------------------------
NodePtr LocalNode::connect( const NodeID& nodeID )
{
    EQASSERT( nodeID != NodeID::ZERO );
    EQASSERT( _state == STATE_LISTENING );

    Nodes nodes;
    getNodes( nodes );

    for( Nodes::const_iterator i = nodes.begin(); i != nodes.end(); ++i )
    {
        NodePtr peer = *i;
        if( peer->getNodeID() == nodeID && peer->isConnected( )) // early out
            return peer;
    }

    for( Nodes::const_iterator i = nodes.begin(); i != nodes.end(); ++i )
    {
        NodePtr peer = *i;
        NodePtr node = _connect( nodeID, peer );
        if( node.isValid( ))
            return node;
    }
        
    EQWARN << "Node " << nodeID << " connection failed" << std::endl;
    return 0;
}

void FullMasterCM::_checkConsistency() const
{
#ifndef NDEBUG
    EQASSERT( !_instanceDatas.empty( ));
    EQASSERT( _object->isAttached() );

    if( _version == VERSION_NONE )
        return;

    uint128_t version = _version;
    for( InstanceDataDeque::const_reverse_iterator i = _instanceDatas.rbegin();
         i != _instanceDatas.rend(); ++i )
    {
        const InstanceData* data = *i;
        EQASSERT( data->os.getVersion() != VERSION_NONE );
        EQASSERTINFO( data->os.getVersion() == version,
                      data->os.getVersion() << " != " << version );
        if( data != _instanceDatas.front( ))
        {
            EQASSERTINFO( data->commitCount + _nVersions >= _commitCount,
                          data->commitCount << ", " << _commitCount << " [" <<
                          _nVersions << "]" );
        }
        --version;
    }
#endif
}

void LocalNode::_redispatchCommands()
{
    bool changes = true;
    while( changes && !_pendingCommands.empty( ))
    {
        changes = false;

        for( CommandList::iterator i = _pendingCommands.begin();
             i != _pendingCommands.end(); ++i )
        {
            Command* command = *i;
            EQASSERT( command->isValid( ));

            if( dispatchCommand( *command ))
            {
                _pendingCommands.erase( i );
                command->release();
                changes = true;
                break;
            }
        }
    }

#ifndef NDEBUG
    if( !_pendingCommands.empty( ))
        EQVERB << _pendingCommands.size() << " undispatched commands" 
               << std::endl;
    EQASSERT( _pendingCommands.size() < 200 );
#endif
}

uint128_t FullMasterCM::commit( const uint32_t incarnation )
{
    Mutex mutex( _slaves );
#if 0
    EQLOG( LOG_OBJECTS ) << "commit v" << _version << " " << command 
                         << std::endl;
#endif
    EQASSERT( _version != VERSION_NONE );

    _updateCommitCount( incarnation );
    
    if( _object->isDirty( ))
    {
        InstanceData* instanceData = _newInstanceData();

        instanceData->os.enableCommit( _version + 1, *_slaves );
        _object->getInstanceData( instanceData->os );
        instanceData->os.disable();

        if( instanceData->os.hasSentData( ))
        {
            ++_version;
            EQASSERT( _version != VERSION_NONE );
#if 0
            EQINFO << "Committed v" << _version << "@" << _commitCount
                   << ", id " << _object->getID() << std::endl;
#endif
            _addInstanceData( instanceData );
        }
        else
            _instanceDataCache.push_back( instanceData );
    }
    _obsolete();
    return _version;
}

void Window::setXDrawable( XID drawable )
{
    if( _xDrawable == drawable )
        return;

    if( _xDrawable )
        exitEventHandler();

    _xDrawable = drawable;

    if( !drawable )
        return;

    initEventHandler();

    // query pixel viewport of window
    EQASSERT( _xDisplay );

    switch( getIAttribute( eq::Window::IATTR_HINT_DRAWABLE ))
    {
        case PBUFFER:
        {
            unsigned width = 0;
            unsigned height = 0;
            glXQueryDrawable( _xDisplay, drawable, GLX_WIDTH,  &width );
            glXQueryDrawable( _xDisplay, drawable, GLX_HEIGHT, &height );

            getWindow()->setPixelViewport( 
                PixelViewport( 0, 0, int32_t( width ), int32_t( height )));
            break;
        }
        case WINDOW:
        {
            XWindowAttributes wa;
            XGetWindowAttributes( _xDisplay, drawable, &wa );
    
            // position is relative to parent: translate to absolute coords
            ::Window root, parent, *children;
            unsigned nChildren;
    
            XQueryTree( _xDisplay, drawable, &root, &parent, &children,
                        &nChildren );
            if( children != 0 )
                XFree( children );

            int x,y;
            ::Window childReturn;
            XTranslateCoordinates( _xDisplay, parent, root, wa.x, wa.y, &x, &y,
                                   &childReturn );

            getWindow()->setPixelViewport( PixelViewport( x, y,
                                                      wa.width, wa.height ));
            break;
        }
        default:
            EQUNIMPLEMENTED;
        case FBO:
            EQASSERT( getWindow()->getPixelViewport().hasArea( ));
    }
}

void Config::setupMessagePump( Pipe* pipe )
{
    const bool isThreaded = pipe->isThreaded();
    const WindowSystem windowSystem = pipe->getWindowSystem();

    if( isThreaded && windowSystem.getName() != "AGL" )
        return;

    // called from pipe threads - but only during init
    static co::base::Lock _lock;
    co::base::ScopedWrite mutex( _lock );

    if( _eventQueue.getMessagePump( )) // Already done
        return;

    MessagePump* pump = pipe->createMessagePump();
    _eventQueue.setMessagePump( pump );

    ClientPtr client = getClient();
    CommandQueue* queue = EQSAFECAST( CommandQueue*, 
                                      client->getMainThreadQueue( ));
    EQASSERT( queue );
    EQASSERT( !queue->getMessagePump( ));

    queue->setMessagePump( pump );
}

void Config::deregisterObject( co::Object* object )
{
    EQASSERT( object )
    EQASSERT( object->isMaster( ));

    if( !object->isAttached( )) // not registered
        return;

    const uint32_t latency = getLatency();
    ClientPtr client = getClient();
    if( latency == 0 || !_running || !object->isBuffered( )) // OPT
    {
        client->deregisterObject( object );
        return;
    }

    // Keep a distributed object latency frames.
    // Replaces the object with a dummy proxy object using the
    // existing master change manager.
    ConfigSwapObjectPacket packet;
    packet.requestID = getLocalNode()->registerRequest();
    packet.object = object;

    send( client, packet );
    client->waitRequest( packet.requestID );
}

bool Pipe::configInit( const eq::uint128_t& initID )
{
    if( !eq::Pipe::configInit( initID ))
        return false;

    Config*         config      = static_cast<Config*>( getConfig( ));
    const InitData& initData    = config->getInitData();
    const co::base::UUID  frameDataID = initData.getFrameDataID();

    const bool mapped = config->mapObject( &_frameData, frameDataID );
    EQASSERT( mapped );

    const std::string& filename = initData.getFilename();
    const uint32_t precision = initData.getPrecision();
    EQINFO << "Loading model " << filename << std::endl;

    _renderer = new Renderer( filename.c_str(), precision );
    EQASSERT( _renderer );

    if( !_renderer->loadHeader( initData.getBrightness(), initData.getAlpha( )))
    {
        setError( ERROR_EVOLVE_LOADMODEL_FAILED );
        EQWARN << getError() << ": " << filename << ".vhf" << std::endl;
        delete _renderer;
        _renderer = 0;
        return false;
    }

    return mapped;
}

void ObjectStore::_detachObject( Object* object )
{
    // check also _cmdUnmapObject when modifying!
    EQASSERT( object );
    EQ_TS_THREAD( _receiverThread );

    if( !object->isAttached() )
        return;

    const base::UUID& id = object->getID();

    EQASSERT( _objects->find( id ) != _objects->end( ));
    EQLOG( LOG_OBJECTS ) << "Detach " << *object << std::endl;

    Objects& objects = _objects.data[ id ];
    Objects::iterator i = find( objects.begin(),objects.end(), object );
    EQASSERT( i != objects.end( ));

    {
        base::ScopedMutex< base::SpinLock > mutex( _objects );
        objects.erase( i );
        if( objects.empty( ))
            _objects->erase( id );
    }

    EQASSERT( object->getInstanceID() != EQ_INSTANCE_INVALID );
    object->detach();
    return;
}

void Connection::setDescription( ConnectionDescriptionPtr description )
{
    EQASSERT( description.isValid( ));
    EQASSERTINFO( _description->type == description->type,
                  "Wrong connection type in description" );
    _description = description;
    EQASSERT( description->bandwidth > 0 );
}

void InstanceCache::_deleteStream( ObjectDataIStream* stream )
{
    EQASSERT( stream->isReady( ));
    EQASSERT( _size >= stream->getDataSize( ));

    _size -= stream->getDataSize();
    delete stream;
}

void DataOStream::_resend()
{
    EQASSERT( !_enabled );
    EQASSERT( !_connections.empty( ));
    EQASSERT( _save );
    
    _compress( _buffer.getData(), _dataSize, STATE_COMPLETE );
    sendData( _buffer.getData(), _dataSize, true );
}

void Window::removeChild( const co::base::UUID& id )
{
    EQASSERT( getConfig()->isRunning( ));

    Channel* channel = _findChannel( id );
    EQASSERT( channel );
    if( channel )
        channel->postDelete();
}

void LoadEqualizer::_updateNode( Node* node )
{
    Node* left = node->left;
    Node* right = node->right;
    
    EQASSERT( left );
    EQASSERT( right );

    _update( left );
    _update( right );

    node->resources = left->resources + right->resources;

    if( left->resources == 0.f )
    {
        node->maxSize    = right->maxSize;
        node->boundary2i = right->boundary2i;
        node->boundaryf  = right->boundaryf;
    }
    else if( right->resources == 0.f )
    {
        node->maxSize = left->maxSize;
        node->boundary2i = left->boundary2i;
        node->boundaryf = left->boundaryf;
    }
    else
    {
        switch( node->mode )
        {
        case MODE_VERTICAL:
            node->maxSize.x() = left->maxSize.x() + right->maxSize.x();  
            node->maxSize.y() = EQ_MIN( left->maxSize.y(), right->maxSize.y());
            node->boundary2i.x() = left->boundary2i.x()+ right->boundary2i.x();
            node->boundary2i.y() = EQ_MAX( left->boundary2i.y(), 
                                           right->boundary2i.y());
            node->boundaryf = EQ_MAX( left->boundaryf, right->boundaryf );
            break;
        case MODE_HORIZONTAL:
            node->maxSize.x() = EQ_MIN( left->maxSize.x(), right->maxSize.x());
            node->maxSize.y() = left->maxSize.y() + right->maxSize.y(); 
            node->boundary2i.x() = EQ_MAX( left->boundary2i.x(), 
                                           right->boundary2i.x() );
            node->boundary2i.y() = left->boundary2i.y()+ right->boundary2i.y();
            node->boundaryf = EQ_MAX( left->boundaryf, right->boundaryf );
            break;
        case MODE_DB:
            node->boundary2i.x() = EQ_MAX( left->boundary2i.x(), 
                                           right->boundary2i.x() );
            node->boundary2i.y() = EQ_MAX( left->boundary2i.y(), 
                                           right->boundary2i.y() );
            node->boundaryf = left->boundaryf + right->boundaryf;
            break;
        default:
            EQUNIMPLEMENTED;
        }
    }
}

void Pipe::removeChild( const co::base::UUID& id )
{
    EQASSERT( getConfig()->isRunning( ));

    Window* window = _findWindow( id );
    EQASSERT( window );
    if( window )
        window->postDelete();
}

void QueueSlave::applyInstanceData( co::DataIStream& is )
{
    uint128_t masterNodeID;
    is >> _impl->masterInstanceID >> masterNodeID;

    EQASSERT( masterNodeID != NodeID::ZERO );
    EQASSERT( !_impl->master );
    LocalNodePtr localNode = getLocalNode();
    _impl->master = localNode->connect( masterNodeID );
}

void Display::discoverLocal( Config* config )
{
    Node* node = config->findAppNode();
    EQASSERT( node );
    if( !node )
        return;

    const Pipes& pipes = node->getPipes();
    EQASSERT( !pipes.empty( ));
    if( pipes.empty( ))
        return;

    Pipe* pipe = pipes.front();
    Window* window = new Window( pipe );
    window->setViewport( Viewport( .25f, .2f, .5f, .5f ));
    window->setName( pipe->getName() + " window" );
    window->setIAttribute( Window::IATTR_PLANES_STENCIL, 1 );

    Channel* channel = new Channel( window );
    channel->setName( pipe->getName() + " channel" );
    Observer* observer = new Observer( config );

    const PixelViewport& pvp = pipe->getPixelViewport();
    Wall wall;
    if( pvp.isValid( ))
        wall.resizeHorizontalToAR( float( pvp.w ) / float( pvp.h ));

    Canvas* canvas = new Canvas( config );
    canvas->setWall( wall );

    Segment* segment = new Segment( canvas );
    segment->setChannel( channel );

    Strings names;
    names.push_back( EQ_SERVER_CONFIG_LAYOUT_2D_DYNAMIC  );
    names.push_back( EQ_SERVER_CONFIG_LAYOUT_SIMPLE );
    names.push_back( EQ_SERVER_CONFIG_LAYOUT_DB_DS );
    names.push_back( EQ_SERVER_CONFIG_LAYOUT_DB_STATIC );
    names.push_back( EQ_SERVER_CONFIG_LAYOUT_DB_DYNAMIC );
    names.push_back( EQ_SERVER_CONFIG_LAYOUT_2D_STATIC );

    for( StringsCIter i = names.begin(); i != names.end(); ++i )
    {
        Layout* layout = new Layout( config );
        layout->setName( *i );

        View* view = new View( layout );
        view->setObserver( observer );
        view->setWall( wall );

        canvas->addLayout( layout );
    }

    config->activateCanvas( canvas );
}

bool Channel::syncConfigExit()
{
    EQASSERT( _state == STATE_EXITING || _state == STATE_EXIT_SUCCESS || 
              _state == STATE_EXIT_FAILED );
    
    _state.waitNE( STATE_EXITING );
    const bool success = ( _state == STATE_EXIT_SUCCESS );
    EQASSERT( success || _state == STATE_EXIT_FAILED );

    _state = isActive() ? STATE_FAILED : STATE_STOPPED;
    return success;
}

void Channel::deactivate()
{ 
    Window* window = getWindow();
    EQASSERT( _active != 0 );
    EQASSERT( window );

    --_active; 
    window->deactivate(); 

    EQLOG( LOG_VIEW ) << "deactivate: " << _active << " " << (void*)this 
                      << std::endl;
}