void ObjectDataIStream::addDataPacket( Command& command )
{
    LB_TS_THREAD( _thread );
    LBASSERT( !isReady( ));

    const ObjectDataPacket* packet = command.get< ObjectDataPacket >();
#ifndef NDEBUG
    if( _commands.empty( ))
    {
        LBASSERTINFO( packet->sequence == 0, packet );
    }
    else
    {
        const ObjectDataPacket* previous = 
            _commands.back()->get< ObjectDataPacket >();
        LBASSERTINFO( packet->sequence == previous->sequence+1, 
                      packet->sequence << ", " << previous->sequence );
        LBASSERT( packet->version == previous->version );
    }
#endif

    command.retain();
    _commands.push_back( &command );
    if( packet->last )
        _setReady();
}

void ObjectDataIStream::addDataCommand( ObjectDataICommand command )
{
    LB_TS_THREAD( _thread );
    LBASSERT( !isReady( ));

#ifndef NDEBUG
    const uint128_t& version = command.getVersion();
    const uint32_t sequence = command.getSequence();

    if( _commands.empty( ))
    {
        LBASSERTINFO( sequence == 0, sequence << " in " << command );
    }
    else
    {
        ObjectDataICommand previous( _commands.back() );
        const uint128_t& previousVersion = previous.getVersion();
        const uint32_t previousSequence = previous.getSequence();
        LBASSERTINFO( sequence == previousSequence+1,
                      sequence << ", " << previousSequence );
        LBASSERT( version == previousVersion );
    }
#endif

    _commands.push_back( command );
    if( command.isLast( ))
        _setReady();
}

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;
}

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 );
}

void FullMasterCM::_checkConsistency() const
{
#ifndef NDEBUG
    LBASSERT( !_instanceDatas.empty( ));
    LBASSERT( _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;
        LBASSERT( data->os.getVersion() != VERSION_NONE );
        LBASSERTINFO( data->os.getVersion() == version,
                      data->os.getVersion() << " != " << version );
        if( data != _instanceDatas.front( ))
        {
            LBASSERTINFO( data->commitCount + _nVersions >= _commitCount,
                          data->commitCount << ", " << _commitCount << " [" <<
                          _nVersions << "]" );
        }
        --version;
    }
#endif
}

uint32_t ObjectStore::mapNB( Object* object, const uint128_t& id,
                             const uint128_t& version, NodePtr master )
{
    LB_TS_NOT_THREAD( _receiverThread );
    LBLOG( LOG_OBJECTS )
        << "Mapping " << lunchbox::className( object ) << " to id " << id
        << " version " << version << std::endl;
    LBASSERT( object );
    LBASSERTINFO( id.isUUID(), id );

    if( !master )
        master = _localNode->connectObjectMaster( id );

    if( !master || !master->isReachable( ))
    {
        LBWARN << "Mapping of object " << id << " failed, invalid master node"
               << std::endl;
        return LB_UNDEFINED_UINT32;
    }

    if( !object || !id.isUUID( ))
    {
        LBWARN << "Invalid object " << object << " or id " << id << std::endl;
        return LB_UNDEFINED_UINT32;
    }

    const bool isAttached = object->isAttached();
    const bool isMaster = object->isMaster();
    LBASSERTINFO( !isAttached, *object );
    LBASSERT( !isMaster ) ;
    if( isAttached || isMaster )
    {
        LBWARN << "Invalid object state: attached " << isAttached << " master "
               << isMaster << std::endl;
        return LB_UNDEFINED_UINT32;
    }

    const uint32_t request = _localNode->registerRequest( object );
    uint128_t minCachedVersion = VERSION_HEAD;
    uint128_t maxCachedVersion = VERSION_NONE;
    uint32_t masterInstanceID = 0;
    const bool useCache = _checkInstanceCache( id, minCachedVersion,
                                               maxCachedVersion,
                                               masterInstanceID );
    object->notifyAttach();
    master->send( CMD_NODE_MAP_OBJECT )
        << version << minCachedVersion << maxCachedVersion << id
        << object->getMaxVersions() << request << _genNextID( _instanceIDs )
        << masterInstanceID << useCache;
    return request;
}

BufferPtr BufferCache::alloc( const uint64_t size )
{
    LB_TS_SCOPED( _thread );
    LBASSERTINFO( size >= COMMAND_ALLOCSIZE, size );
    LBASSERTINFO( size < LB_BIT48,
                  "Out-of-sync network stream: buffer size " << size << "?" );

    BufferPtr buffer = _impl->newBuffer();
    LBASSERT( buffer->getRefCount() == 1 );

    buffer->reserve( size );
    buffer->resize( 0 );
    return buffer;
}

Referenced::~Referenced()
{
    LBASSERT( !_hasBeenDeleted );
    _hasBeenDeleted = true;
    LBASSERTINFO( _refCount == 0,
        "Deleting object with ref count " << _refCount );
}

uint128_t& uint128_t::operator = ( const std::string& from )
{
    if( from.empty( ))
    {
        _high = 0;
        _low = 0;
        return *this;
    }

    char* next = 0;
    _high = ::strtoull( from.c_str(), &next, 16 );
    LBASSERT( next != from.c_str( ));

    if( *next == '\0' ) // short representation, high was 0
    {
        _low = _high;
        _high = 0;
    }
    else
    {
        if( strncmp( next, "\\058" /* utf-8 ':' */, 4 ) == 0 )
            next += 4;
        else
        {
            LBASSERTINFO( *next == ':', from << ", " << next );
            ++next;
        }
        _low = ::strtoull( next, 0, 16 );
    }
    return *this;
}

bool DataIStreamQueue::addDataPacket( const uint128_t& key, Command& command )
{
    LB_TS_THREAD( _thread );
    LBASSERTINFO( _pending.size() < 100, "More than 100 pending commits");

    ObjectDataIStream* istream = 0;
    PendingStreams::iterator i = _pending.find( key );
    if( i == _pending.end( ))
        istream = _iStreamCache.alloc();
    else
        istream = i->second;

    istream->addDataPacket( command );
    if( istream->isReady( ))
    {
        if( i != _pending.end( ))
            _pending.erase( i );

        _queued.push( QueuedStream( key, istream ));
        //LBLOG( LOG_OBJECTS ) << "Queued commit " << key << std::endl;
        return true;
    }

    if( i == _pending.end( ))
    {
        _pending[ key ] = istream;
        //LBLOG( LOG_OBJECTS ) << "New incomplete commit " << key << std::endl;
        return false;
    }

    //LBLOG(LOG_OBJECTS) << "Add data to incomplete commit " << key <<std::endl;
    return false;
}

bool DataIStreamQueue::addDataCommand(const uint128_t& key, ICommand& command)
{
    LB_TS_THREAD(_thread);
    LBASSERTINFO(_pending.size() < 100, "More than 100 pending commits");

    ObjectDataIStream* is = 0;
    PendingStreams::iterator i = _pending.find(key);
    if (i == _pending.end())
        is = _iStreamCache.alloc();
    else
        is = i->second;

    is->addDataCommand(command);
    if (is->isReady())
    {
        if (i != _pending.end())
            _pending.erase(i);

        _queued.push(QueuedStream(key, is));
        return true;
    }

    if (i == _pending.end())
    {
        _pending[key] = is;
        return false;
    }

    return false;
}

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;
}

void MonitorEqualizer::_updateZoomAndOffset()
{
    const Compound* compound( getCompound( ));
    const PixelViewport& pvp( compound->getInheritPixelViewport( ));

    const Frames& inputFrames( compound->getInputFrames( ));
    const size_t size( inputFrames.size( ));
    LBASSERTINFO( size == _outputFrames.size(),
                  size << " != " << _outputFrames.size( ));
    LBASSERT( size == _viewports.size( ));

    for( size_t i = 0; i < size; ++i )
    {
        Frame* frame = inputFrames[ i ];
        Frame* outputFrame = _outputFrames[ i ];
        if( !outputFrame )
            continue;

        const Compound* srcCompound = outputFrame->getCompound();
        const Viewport& viewport = _viewports[ i ];

        // compute and apply input frame offset
        const int32_t offsetX = int32_t( float( pvp.w ) * viewport.x );
        const int32_t offsetY = int32_t( float( pvp.h ) * viewport.y );
        frame->setNativeOffset( Vector2i( offsetX, offsetY ));

        // compute and apply output frame zoom
        const int32_t width   = int32_t( float( pvp.w ) * viewport.w );
        const int32_t height  = int32_t( float( pvp.h ) * viewport.h ) ;
        const PixelViewport& srcPVP( srcCompound->getInheritPixelViewport( ));
        const Zoom zoom( float( width ) / float( srcPVP.w ),
                         float( height ) / float( srcPVP.h ));
        outputFrame->setNativeZoom( zoom );
    }
}

void ROITracker::updateDelay( const PixelViewports& pvps,
                              const uint8_t* ticket )
{
    LBASSERT( _needsUpdate );
    LBASSERTINFO( ticket == _ticket, "Wrong ticket" );

    if( ticket != _ticket )
    {
        LBERROR << "Wrong ticket" << std::endl;
        return;
    }

    uint32_t totalAreaFound = 0;
    for( uint32_t i = 0; i < pvps.size(); i++ )
        totalAreaFound += pvps[ i ].getArea();

    Area& area = (*_curFrame)[ _lastStage ].areas.back();
    if( totalAreaFound < area.pvp.getArea()*4/5 )
    {
        // ROI cutted enough, reset failure statistics
        area.lastSkip = 0;
    }else
    {
        // disable ROI for next frames, if it was failing before, 
        // increase number of frames to skip
        area.lastSkip = LB_MIN( area.lastSkip*2 + 1, 64 );
        area.skip     = area.lastSkip;
    }
    _needsUpdate = false;
}

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 ));
}

bool Node::removeConnectionDescription( ConnectionDescriptionPtr cd )
{
    LBASSERTINFO( isClosed(), *this );
    if( !isClosed( ))
        return false;
    return _removeConnectionDescription( cd );
}

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 Node::_cmdFrameStart( co::Command& command )
{
    LB_TS_THREAD( _nodeThread );
    const NodeFrameStartPacket* packet = 
        command.get<NodeFrameStartPacket>();
    LBVERB << "handle node frame start " << packet << std::endl;

    const uint32_t frameNumber = packet->frameNumber;
    LBASSERT( _currentFrame == frameNumber-1 );

    LBLOG( LOG_TASKS ) << "----- Begin Frame ----- " << frameNumber
                       << std::endl;

    Config* config = getConfig();
    
    if( packet->configVersion != co::VERSION_INVALID )
        config->sync( packet->configVersion );
    sync( packet->version );

    config->_frameStart();
    frameStart( packet->frameID, frameNumber );

    LBASSERTINFO( _currentFrame >= frameNumber, 
                  "Node::frameStart() did not start frame " << frameNumber );
    return true;
}

bool Node::_cmdFrameStart( co::ICommand& cmd )
{
    LB_TS_THREAD( _nodeThread );

    co::ObjectICommand command( cmd );
    const uint128_t& version = command.read< uint128_t >();
    const uint128_t& configVersion = command.read< uint128_t >();
    const uint128_t& frameID = command.read< uint128_t >();
    const uint32_t frameNumber = command.read< uint32_t >();

    LBVERB << "handle node frame start " << command << " frame " << frameNumber
           << " id " << frameID << std::endl;

    LBASSERT( _impl->currentFrame == frameNumber-1 );

    LBLOG( LOG_TASKS ) << "----- Begin Frame ----- " << frameNumber
                       << std::endl;

    Config* config = getConfig();

    if( configVersion != co::VERSION_INVALID )
        config->sync( configVersion );
    sync( version );

    config->_frameStart();
    frameStart( frameID, frameNumber );

    LBASSERTINFO( _impl->currentFrame >= frameNumber,
                  "Node::frameStart() did not start frame " << frameNumber );
    return true;
}

void ObjectStore::_attach( Object* object, const uint128_t& id,
                           const uint32_t inInstanceID )
{
    LBASSERT( object );
    LB_TS_THREAD( _receiverThread );

    uint32_t instanceID = inInstanceID;
    if( inInstanceID == CO_INSTANCE_INVALID )
        instanceID = _genNextID( _instanceIDs );

    object->attach( id, instanceID );

    {
        lunchbox::ScopedFastWrite mutex( _objects );
        Objects& objects = _objects.data[ id ];
        LBASSERTINFO( !object->isMaster() || objects.empty(),
                      "Attaching master " << *object << ", " <<
                      objects.size() << " attached objects with same ID, " <<
                      "first is " << ( objects[0]->isMaster() ? "master " :
                                       "slave " ) << *objects[0] );
        objects.push_back( object );
    }

    _localNode->flushCommands(); // redispatch pending commands

    LBLOG( LOG_OBJECTS ) << "attached " << *object << " @"
                         << static_cast< void* >( object ) << std::endl;
}