Line data    Source code

       1             : 
       2             : /* Copyright (c) 2011-2012, Stefan Eilemann <eile@equalizergraphics.com>
       3             :  *               2011-2012, Daniel Nachbaur <danielnachbaur@gmail.com>
       4             :  *
       5             :  * This file is part of Collage <https://github.com/Eyescale/Collage>
       6             :  *
       7             :  * This library is free software; you can redistribute it and/or modify it under
       8             :  * the terms of the GNU Lesser General Public License version 2.1 as published
       9             :  * by the Free Software Foundation.
      10             :  *
      11             :  * This library is distributed in the hope that it will be useful, but WITHOUT
      12             :  * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS
      13             :  * FOR A PARTICULAR PURPOSE.  See the GNU Lesser General Public License for more
      14             :  * details.
      15             :  *
      16             :  * You should have received a copy of the GNU Lesser General Public License
      17             :  * along with this library; if not, write to the Free Software Foundation, Inc.,
      18             :  * 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
      19             :  */
      20             : 
      21             : #include "dataIStreamQueue.h"
      22             : 
      23             : #include "objectDataICommand.h"
      24             : #include "objectDataIStream.h"
      25             : 
      26             : namespace co
      27             : {
      28          63 : DataIStreamQueue::DataIStreamQueue()
      29          63 : {}
      30             : 
      31         122 : DataIStreamQueue::~DataIStreamQueue()
      32             : {
      33          61 :     LBASSERTINFO( _pending.empty(), "Incomplete commits pending" );
      34          61 :     LBASSERTINFO( _queued.isEmpty(), _queued.getSize() << " unapplied commits" )
      35             : 
      36          61 :     for( PendingStreamsCIter i = _pending.begin(); i != _pending.end(); ++i )
      37           0 :         delete i->second;
      38          61 :     _pending.clear();
      39             : 
      40          61 :     QueuedStream stream;
      41         122 :     while( _queued.tryPop( stream ))
      42           0 :         delete stream.second;
      43          61 : }
      44             : 
      45           0 : ObjectDataIStream* DataIStreamQueue::tryPop()
      46             : {
      47           0 :     QueuedStream stream( uint128_t(), (ObjectDataIStream*)0 );
      48           0 :     _queued.tryPop( stream );
      49           0 :     return stream.second;
      50             : }
      51             : 
      52           4 : ObjectDataIStream* DataIStreamQueue::pull( const uint128_t& key )
      53             : {
      54           4 :     ObjectDataIStream* is = 0;
      55           4 :     QueuedStreams unusedStreams;
      56          12 :     while( !is )
      57             :     {
      58           4 :         QueuedStream candidate = _queued.pop();
      59           4 :         if( candidate.first == key )
      60           4 :             is = candidate.second;
      61             :         else
      62           0 :             unusedStreams.push_back( candidate );
      63             :     }
      64             : 
      65           4 :     _queued.pushFront( unusedStreams );
      66           4 :     return is;
      67             : }
      68             : 
      69           4 : void DataIStreamQueue::recycle( ObjectDataIStream* stream )
      70             : {
      71             : #ifdef CO_AGGRESSIVE_CACHING
      72             :     stream->reset();
      73             :     _iStreamCache.release( stream );
      74             : #else
      75           4 :     delete stream;
      76             : #endif
      77           4 : }
      78             : 
      79           6 : bool DataIStreamQueue::addDataCommand( const uint128_t& key, ICommand& command )
      80             : {
      81           6 :     LB_TS_THREAD( _thread );
      82           6 :     LBASSERTINFO( _pending.size() < 100, "More than 100 pending commits");
      83             : 
      84           6 :     ObjectDataIStream* istream = 0;
      85           6 :     PendingStreams::iterator i = _pending.find( key );
      86           6 :     if( i == _pending.end( ))
      87           4 :         istream = _iStreamCache.alloc();
      88             :     else
      89           2 :         istream = i->second;
      90             : 
      91           6 :     istream->addDataCommand( command );
      92           6 :     if( istream->isReady( ))
      93             :     {
      94           4 :         if( i != _pending.end( ))
      95           2 :             _pending.erase( i );
      96             : 
      97           4 :         _queued.push( QueuedStream( key, istream ));
      98             :         //LBLOG( LOG_OBJECTS ) << "Queued commit " << key << std::endl;
      99           4 :         return true;
     100             :     }
     101             : 
     102           2 :     if( i == _pending.end( ))
     103             :     {
     104           2 :         _pending[ key ] = istream;
     105             :         //LBLOG( LOG_OBJECTS ) << "New incomplete commit " << key << std::endl;
     106           2 :         return false;
     107             :     }
     108             : 
     109             :     //LBLOG(LOG_OBJECTS) << "Add data to incomplete commit " << key <<std::endl;
     110           0 :     return false;
     111             : }
     112             : 
     113          63 : }