uint128_t UnbufferedMasterCM::commit( const uint32_t incarnation )
{
#if 0
    LBLOG( LOG_OBJECTS ) << "commit v" << _version << " " << command
                         << std::endl;
#endif
    if( !_object->isDirty( ))
        return _version;

    _maxVersion.waitGE( _version.low() + 1 );
    Mutex mutex( _slaves );
    if( _slaves->empty( ))
        return _version;

    ObjectDeltaDataOStream os( this );
    os.enableCommit( _version + 1, *_slaves );
    _object->pack( os );
    os.disable();

    if( os.hasSentData( ))
    {
        ++_version;
        LBASSERT( _version != VERSION_NONE );
#if 0
        LBLOG( LOG_OBJECTS ) << "Committed v" << _version << ", id "
                             << _object->getID() << std::endl;
#endif
    }

    return _version;
}

bool Window::_cmdConfigInit( co::Command& command )
{
    const WindowConfigInitPacket* packet =
        command.get<WindowConfigInitPacket>();
    LBLOG( LOG_INIT ) << "TASK window config init " << packet << std::endl;

    WindowConfigInitReplyPacket reply;
    setError( ERROR_NONE );

    if( getPipe()->isRunning( ))
    {
        _state = STATE_INITIALIZING;
        reply.result = configInit( packet->initID );
        if( reply.result )
            _state = STATE_RUNNING;
    }
    else
    {
        setError( ERROR_WINDOW_PIPE_NOTRUNNING );
        reply.result = false;
    }
    LBLOG( LOG_INIT ) << "TASK window config init reply " << &reply <<std::endl;

    commit();
    send( command.getNode(), reply );
    return true;
}

bool Window::_cmdConfigInit( co::Command& cmd )
{
    co::ObjectCommand command( cmd );

    LBLOG( LOG_INIT ) << "TASK window config init " << command << std::endl;

    setError( ERROR_NONE );

    bool result = false;
    if( getPipe()->isRunning( ))
    {
        _state = STATE_INITIALIZING;
        result = configInit( command.get< uint128_t >( ));
        if( result )
            _state = STATE_RUNNING;
    }
    else
        setError( ERROR_WINDOW_PIPE_NOTRUNNING );

    LBLOG( LOG_INIT ) << "TASK window config init reply " << std::endl;

    commit();
    send( command.getNode(), fabric::CMD_WINDOW_CONFIG_INIT_REPLY ) << result;
    return true;
}

void Config::_stopNodes()
{
    // wait for the nodes to stop, destroy entities, disconnect
    Nodes stoppingNodes;
    const Nodes& nodes = getNodes();
    for( Nodes::const_iterator i = nodes.begin(); i != nodes.end(); ++i )
    {
        Node* node = *i;
        const State state = node->getState();
        if( state != STATE_STOPPED && state != STATE_FAILED )
            continue;

        LBASSERT( !node->isActive() || state == STATE_FAILED );
        if( node->isApplicationNode( ))
            continue;

        co::NodePtr netNode = node->getNode();
        if( !netNode ) // already disconnected
            continue;

        LBLOG( LOG_INIT ) << "Exiting node" << std::endl;

        if( state == STATE_FAILED )
            node->setState( STATE_STOPPED );

        stoppingNodes.push_back( node );
        LBASSERT( netNode.isValid( ));

        netNode->send( fabric::CMD_SERVER_DESTROY_CONFIG )
                << getID() << LB_UNDEFINED_UINT32;
        netNode->send( fabric::CMD_CLIENT_EXIT );
    }

    // now wait that the render clients disconnect
    uint32_t nSleeps = 50; // max 5 seconds for all clients
    for( Nodes::const_iterator i = stoppingNodes.begin();
         i != stoppingNodes.end(); ++i )
    {
        Node*        node    = *i;
        co::NodePtr netNode = node->getNode();
        node->setNode( 0 );

        if( nSleeps )
            while( netNode->isConnected() && --nSleeps )
                lunchbox::sleep( 100 ); // ms

        if( netNode->isConnected( ))
        {
            co::LocalNodePtr localNode = getLocalNode();
            LBASSERT( localNode.isValid( ));

            LBWARN << "Forcefully disconnecting exited render client node"
                   << std::endl;
            localNode->disconnect( netNode );
        }

        LBLOG( LOG_INIT ) << "Disconnected node" << std::endl;
    }
}

//===========================================================================
// Operations
//===========================================================================
//---------------------------------------------------------------------------
// init
//---------------------------------------------------------------------------
void Window::configInit( const uint128_t& initID, const uint32_t /*frame*/ )
{
    LBASSERT( !needsDelete( ));
    LBASSERT( _state == STATE_STOPPED );
    _state = STATE_INITIALIZING;

    LBLOG( LOG_INIT ) << "Create Window" << std::endl;
    getPipe()->send( fabric::CMD_PIPE_CREATE_WINDOW ) << getID();

    LBLOG( LOG_INIT ) << "Init Window" << std::endl;
    send( fabric::CMD_WINDOW_CONFIG_INIT ) << initID;
    LBLOG( LOG_TASKS ) << "TASK window configInit  " << " id " << getID()
                       << std::endl;
}

void View::trigger( const Canvas* canvas, const bool active )
{
    const Mode mode = getMode();
    Config* config = getConfig();

    // (De)activate destination compounds for canvas/eye(s)
    for( Channels::const_iterator i = _channels.begin();
         i != _channels.end(); ++i )
    {
        Channel* channel = *i;
        const Canvas* channelCanvas = channel->getCanvas();
        const Layout* canvasLayout = channelCanvas->getActiveLayout();

        if((  canvas && channelCanvas != canvas ) ||
           ( !canvas && canvasLayout  != getLayout( )))
        {
            continue;
        }

        const Segment* segment = channel->getSegment();
        const uint32_t segmentEyes = segment->getEyes();
        const uint32_t eyes = ( mode == MODE_MONO ) ?
                           EYE_CYCLOP & segmentEyes : EYES_STEREO & segmentEyes;
        if( eyes == 0 )
            continue;

        ConfigDestCompoundVisitor visitor( channel, true /*activeOnly*/ );
        config->accept( visitor );

        const Compounds& compounds = visitor.getResult();
        for( Compounds::const_iterator j = compounds.begin();
             j != compounds.end(); ++j )
        {
            Compound* compound = *j;
            if( active )
            {
                compound->activate( eyes );
                LBLOG( LOG_VIEW ) << "Activate " << compound->getName()
                                  << std::endl;
            }
            else
            {
                compound->deactivate( eyes );
                LBLOG( LOG_VIEW ) << "Deactivate " << compound->getName()
                                  << std::endl;
            }
        }
    }
}

bool Pipe::_cmdConfigInit( co::ICommand& cmd )
{
    LB_TS_THREAD( _pipeThread );

    co::ObjectICommand command( cmd );
    const uint128_t initID = command.get< uint128_t >();
    const uint32_t frameNumber = command.get< uint32_t >();

    LBLOG( LOG_INIT ) << "Init pipe " << command << " init id " << initID
                      << " frame " << frameNumber << std::endl;

    if( !isThreaded( ))
    {
        _impl->windowSystem = selectWindowSystem();
        _setupCommandQueue();
    }

    setError( ERROR_NONE );

    Node* node = getNode();
    LBASSERT( node );
    node->waitInitialized();

    bool result = false;
    if( node->isRunning( ))
    {
        _impl->currentFrame  = frameNumber;
        _impl->finishedFrame = frameNumber;
        _impl->unlockedFrame = frameNumber;
        _impl->state = STATE_INITIALIZING;

        result = configInit( initID );

        if( result )
            _impl->state = STATE_RUNNING;
    }
    else
        setError( ERROR_PIPE_NODE_NOTRUNNING );

    LBLOG( LOG_INIT ) << "TASK pipe config init reply result " << result
                      << std::endl;

    co::NodePtr netNode = command.getNode();

    commit();
    send( netNode, fabric::CMD_PIPE_CONFIG_INIT_REPLY ) << result;
    return true;
}

void ObjectStore::_detach( Object* object )
{
    // check also _cmdUnmapObject when modifying!
    LBASSERT( object );
    LB_TS_THREAD( _receiverThread );

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

    const uint128_t& id = object->getID();

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

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

    {
        lunchbox::ScopedFastWrite mutex( _objects );
        objects.erase( i );
        if( objects.empty( ))
            _objects->erase( id );
    }

    LBASSERT( object->getInstanceID() != CO_INSTANCE_INVALID );
    object->detach();
    return;
}

uint128_t VersionedSlaveCM::sync( const uint128_t& v )
{
#if 0
    LBLOG( LOG_OBJECTS ) << "sync to v" << v << ", id " << _object->getID()
                         << "." << _object->getInstanceID() << std::endl;
#endif
    if( _version == v )
        return _version;

    if( v == VERSION_HEAD )
    {
        _syncToHead();
        return _version;
    }

    const uint128_t version = ( v == VERSION_NEXT ) ? _version + 1 : v;
    LBASSERTINFO( version.high() == 0, "Not a master version: " << version )
    LBASSERTINFO( _version <= version,
                  "can't sync to older version of object " << 
                  lunchbox::className( _object ) << " " << _object->getID() <<
                  " (" << _version << ", " << version <<")" );

    while( _version < version )
        _unpackOneVersion( _queuedVersions.pop( ));

    LocalNodePtr node = _object->getLocalNode();
    if( node.isValid( ))
        node->flushCommands();

    return _version;
}

//---------------------------------------------------------------------------
// command handlers
//---------------------------------------------------------------------------
bool VersionedSlaveCM::_cmdData( Command& command )
{
    LB_TS_THREAD( _rcvThread );
    LBASSERT( command.getNode().isValid( ));

    if( !_currentIStream )
        _currentIStream = _iStreamCache.alloc();

    _currentIStream->addDataPacket( command );
    if( _currentIStream->isReady( ))
    {
        const uint128_t& version = _currentIStream->getVersion();
#if 0
        LBLOG( LOG_OBJECTS ) << "v" << version << ", id " << _object->getID()
                             << "." << _object->getInstanceID() << " ready"
                             << std::endl;
#endif
#ifndef NDEBUG
        ObjectDataIStream* debugStream = 0;
        _queuedVersions.getBack( debugStream );
        if ( debugStream )
        {
            LBASSERT( debugStream->getVersion() + 1 == version );
        }
#endif
        _queuedVersions.push( _currentIStream );
        _object->notifyNewHeadVersion( version );
        _currentIStream = 0;
    }
    return true;
}

void VersionedSlaveCM::_unpackOneVersion( ObjectDataIStream* is )
{
    LBASSERT( is );
    LBASSERTINFO( _version == is->getVersion() - 1, "Expected version " 
                  << _version + 1 << ", got " << is->getVersion() << " for "
                  << *_object );

    if( is->hasInstanceData( ))
        _object->applyInstanceData( *is );
    else
        _object->unpack( *is );

    _version = is->getVersion();
    _sendAck();

    LBASSERT( _version != VERSION_INVALID );
    LBASSERT( _version != VERSION_NONE );
    LBASSERTINFO( is->getRemainingBufferSize()==0 && is->nRemainingBuffers()==0,
                  "Object " << typeid( *_object ).name() <<
                  " did not unpack all data" );

#if 0
    LBLOG( LOG_OBJECTS ) << "applied v" << _version << ", id "
                         << _object->getID() << "." << _object->getInstanceID()
                         << std::endl;
#endif
    _releaseStream( is );
}

bool ObjectStore::_finishSync( const uint32_t requestID, Object* object )
{
    if( requestID == LB_UNDEFINED_UINT32 )
        return false;

    void* data = _localNode->getRequestData( requestID );
    if( data == 0 )
        return false;

    ObjectDataIStream* is = LBSAFECAST( ObjectDataIStream*, data );

    bool ok = false;
    _localNode->waitRequest( requestID, ok );

    if( !ok )
    {
        LBWARN << "Object synchronization failed" << std::endl;
        delete is;
        return false;
    }

    is->waitReady();
    object->applyInstanceData( *is );
    LBLOG( LOG_OBJECTS ) << "Synced " << lunchbox::className( object )
                         << std::endl;
    delete is;
    return true;
}

bool Window::_cmdFrameStart( co::ICommand& cmd )
{
    co::ObjectICommand command( cmd );

    LB_TS_THREAD( _pipeThread );

    const uint128_t& version = command.read< uint128_t >();
    const uint128_t& frameID = command.read< uint128_t >();
    const uint32_t frameNumber = command.read< uint32_t >();

    LBLOG( LOG_TASKS ) << "TASK frame start " << getName()
                       << " " << command << " frame " << frameNumber
                       << " id " << frameID << std::endl;

    //_grabFrame( frameNumber ); single-threaded
    sync( version );

    const DrawableConfig& drawableConfig = getDrawableConfig();
    if( drawableConfig.doublebuffered )
        _renderContexts[FRONT].swap( _renderContexts[BACK] );
    _renderContexts[BACK].clear();

    makeCurrent();
    frameStart( frameID, frameNumber );
    return true;
}

void FrameData::setVersion( const uint64_t version )
{
    LBASSERTINFO( _impl->version <= version, _impl->version << " > "
                                                            << version );
    _impl->version = version;
    LBLOG( LOG_ASSEMBLY ) << "New v" << version << std::endl;
}

bool Node::_cmdConfigInit( co::Command& command )
{
    LB_TS_THREAD( _nodeThread );

    const NodeConfigInitPacket* packet = 
        command.get<NodeConfigInitPacket>();
    LBLOG( LOG_INIT ) << "Init node " << packet << std::endl;

    _state = STATE_INITIALIZING;

    _currentFrame  = packet->frameNumber;
    _unlockedFrame = packet->frameNumber;
    _finishedFrame = packet->frameNumber;
    _setAffinity();

    transmitter.start();
    setError( ERROR_NONE );
    NodeConfigInitReplyPacket reply;
    reply.result = configInit( packet->initID );

    if( getIAttribute( IATTR_THREAD_MODEL ) == eq::UNDEFINED )
        setIAttribute( IATTR_THREAD_MODEL, eq::DRAW_SYNC );

    _state = reply.result ? STATE_RUNNING : STATE_INIT_FAILED;

    commit();
    send( command.getNode(), reply );
    return true;
}

uint128_t VersionedMasterCM::sync( const uint128_t& inVersion )
{
    LBASSERTINFO( inVersion.high() != 0 || inVersion == VERSION_NEXT ||
                  inVersion == VERSION_HEAD, inVersion );
#if 0
    LBLOG( LOG_OBJECTS ) << "sync to v" << inVersion << ", id "
                         << _object->getID() << "." << _object->getInstanceID()
                         << std::endl;
#endif

    if( inVersion == VERSION_NEXT )
        return _apply( _slaveCommits.pop( ));

    if( inVersion == VERSION_HEAD )
    {
        uint128_t version = VERSION_NONE;
        for( ObjectDataIStream* is = _slaveCommits.tryPop(); is;
                is = _slaveCommits.tryPop( ))
        {
            version = _apply( is );
        }
        return version;
    }
    // else apply only concrete slave commit

    return _apply( _slaveCommits.pull( inVersion ));
}

void PluginRegistry::init()
{
    // for each directory
    for( StringsCIter i = impl_->directories.begin();
         i != impl_->directories.end(); ++i )
    {
        const std::string& dir = *i;
        LBLOG( LOG_PLUGIN ) << "Searching plugins in " << dir << std::endl;

#ifdef _WIN32
        const Strings& files = searchDirectory( dir, ".*Compressor.*\\.dll" );
        const char DIRSEP = '\\';
#elif defined (Darwin)
        const Strings& files = searchDirectory( dir,
                                                "lib.*Compressor.*\\.dylib" );
        const char DIRSEP = '/';
#else
        const Strings& files = searchDirectory( dir, "lib.*Compressor.*\\.so" );
        const char DIRSEP = '/';
#endif
        // for each file found in the directory
        for( StringsCIter j = files.begin(); j != files.end(); ++j )
        {
            // build path + name of library
            const std::string libraryName = dir.empty() ? *j : dir+DIRSEP+*j;
            addPlugin( libraryName );
        }
    }
}

void LoadEqualizer::_assign( Compound* compound, const Viewport& vp,
                             const Range& range )
{
    LBASSERTINFO( vp == Viewport::FULL || range == Range::ALL,
                  "Mixed 2D/DB load-balancing not implemented" );

    compound->setViewport( vp );
    compound->setRange( range );
    LBLOG( LOG_LB2 ) << compound->getChannel()->getName() << " set " << vp
                     << ", " << range << std::endl;

    if( getDamping() >= 1.f )
        return;

    // save data for later use
    Data data;
    data.vp      = vp;
    data.range   = range;
    data.channel = compound->getChannel();
    data.taskID  = compound->getTaskID();

    const Compound* destCompound = getCompound();
    if( destCompound->getChannel() == compound->getChannel( ))
        data.destTaskID  = destCompound->getTaskID();

    LBASSERT( data.taskID > 0 );

    if( !vp.hasArea() || !range.hasData( )) // will not render
        data.time = 0;

    LBFrameData& frameData = _history.back();
    LBDatas& items = frameData.second;

    items.push_back( data );
}

bool PluginRegistry::addPlugin( const std::string& filename )
{
    Plugin* plugin = new Plugin(); 
    if( !plugin->init( filename ))
    {
        delete plugin;
        return false;
    }

    const CompressorInfos& infos = plugin->getInfos();
    for( Plugins::const_iterator i = _plugins.begin(); i != _plugins.end(); ++i)
    {
        const CompressorInfos& infos2 = (*i)->getInfos();

        // Simple test to avoid loading the same dll twice
        if( infos.front().name == infos2.front().name )
        {
            delete plugin;
            return true;
        }
    }

    _plugins.push_back( plugin );
    LBLOG( LOG_PLUGIN ) << "Found " << plugin->getInfos().size()
                        << " compression engines in " << filename << std::endl;
    return true;
}

bool Node::_cmdFrameDataTransmit( co::ICommand& cmd )
{
    co::ObjectICommand command( cmd );

    const co::ObjectVersion& frameDataVersion =
                                             command.read< co::ObjectVersion >();
    const PixelViewport& pvp = command.read< PixelViewport >();
    const Zoom& zoom = command.read< Zoom >();
    const uint32_t buffers = command.read< uint32_t >();
    const uint32_t frameNumber = command.read< uint32_t >();
    const bool useAlpha = command.read< bool >();
    const uint8_t* data = reinterpret_cast< const uint8_t* >(
                command.getRemainingBuffer( command.getRemainingBufferSize( )));

    LBLOG( LOG_ASSEMBLY )
        << "received image data for " << frameDataVersion << ", buffers "
        << buffers << " pvp " << pvp << std::endl;

    LBASSERT( pvp.isValid( ));

    FrameDataPtr frameData = getFrameData( frameDataVersion );
    LBASSERT( !frameData->isReady() );

    NodeStatistics event( Statistic::NODE_FRAME_DECOMPRESS, this,
                          frameNumber );

    // Note on the const_cast: since the PixelData structure stores non-const
    // pointers, we have to go non-const at some point, even though we do not
    // modify the data.
    LBCHECK( frameData->addImage( frameDataVersion, pvp, zoom, buffers,
                                  useAlpha, const_cast< uint8_t* >( data )));
    return true;
}