ossimMultiThreadSequencer.cpp

Go to the documentation of this file.

//**************************************************************************************************
//                          OSSIM -- Open Source Software Image Map
//
// LICENSE: See top level LICENSE.txt file.
//
// AUTHOR: Oscar Kramer
//
// 
//**************************************************************************************************
//  $Id$

#include <ossim/parallel/ossimMultiThreadSequencer.h>
#include <ossim/parallel/ossimMtDebug.h>
#include <ossim/base/ossimIrect.h>
#include <ossim/base/ossimTimer.h>
static const ossim_uint32 DEFAULT_MAX_TILE_CACHE_FACTOR = 8; // Must be > 1

ossimMtDebug* ossimMtDebug::m_instance = NULL;

//*************************************************************************************************
// Job's start method performs actual getTile in a thread on cloned chain and saves the result
// in the sequencer's results cache.
//*************************************************************************************************
void ossimMultiThreadSequencer::ossimGetTileJob::run()
{
   running();
   if (m_sequencer.d_debugEnabled)
   {
      ostringstream s1;
      s1<<"THREAD #"<<m_chainID<<" -- Starting tile/job #"<<m_tileID;
      m_sequencer.print(s1);
   }

   // Figure out the rect for this tile. Only process if rect is valid:
   ossimIrect tileRect;
   if (m_sequencer.getTileRect(m_tileID, tileRect))
   {
      // Perform the getTile and save the result:
      ossimRefPtr<ossimImageData> tile = 0;
      ossimImageSource* source = m_sequencer.m_inputChain->getClone(m_chainID);
      double dt = ossimTimer::instance()->time_s(); //###

      if (source != NULL)
         tile = source->getTile(tileRect);
      if (!tile.valid())
      {
         tile = m_sequencer.theBlankTile;
         tile->setImageRectangle(tileRect);
      }
      dt = ossimTimer::instance()->time_s() - dt; //###

      // Give the sequencer the tile. Execution may pause here while waiting for space to free up
      // if the cache is full.
      m_sequencer.setTileInCache(m_tileID, (ossimImageData*)tile->dup(), m_chainID, dt);
   }

   // Unblock the main thread which might be blocked waiting for jobs to finish:
   m_sequencer.m_getTileBlock.release();

   // Queue the next job using this job's freed-up image chain:
   if (t_launchNewJob)
      m_sequencer.nextJob(m_chainID);

   finished();
   
   if (m_sequencer.d_debugEnabled)
   {
      ostringstream s2;
      s2<<"THREAD #"<<m_chainID<<" -- Finished tile/job #"<<m_tileID;
      m_sequencer.print(s2);
   }
}

//*************************************************************************************************
// Constructor
//*************************************************************************************************
ossimMultiThreadSequencer::ossimMultiThreadSequencer(ossimImageSource* input, 
                                                     ossim_uint32 num_threads,
                                                     ossimObject* owner)
   : ossimImageSourceSequencer(input, owner),
   d_maxCacheUsed(0),
   d_cacheEmptyCount(0), 
   d_idleTime1(0.0),     
   d_idleTime2(0.0),     
   d_idleTime3(0.0),     
   d_idleTime4(0.0),     
   d_idleTime5(0.0),     
   d_idleTime6(0.0),
   d_jobGetTileT(0.0),
   m_inputChain(0),
   m_jobMtQueue(0),
   m_numThreads (num_threads),
   m_callback(std::make_shared<ossimGetTileCallback>()),
   m_nextTileID (0),
   m_tileCache(),                       
   m_maxCacheSize (DEFAULT_MAX_TILE_CACHE_FACTOR * num_threads),
   m_maxTileCacheFactor (DEFAULT_MAX_TILE_CACHE_FACTOR),
   m_cacheMutex(),
   m_jobMutex(),
   m_totalNumberOfTiles(0),
   m_getTileBlock(),
   m_nextJobBlock(),
   d_printMutex(),
   d_timerMutex(),                                 
   d_debugEnabled(false),
   d_timedBlocksDt(0),
   d_timeMetricsEnabled(false),
   d_useSharedHandlers(true),
   d_cacheTileSize(1024),
   d_useCache(true),
   d_t1(0.0)                              
{
   //###### DEBUG ############
   ossimMtDebug* mt_debug = ossimMtDebug::instance();
   if (mt_debug->maxTileCacheSize != 0)
      m_maxCacheSize =  mt_debug->maxTileCacheSize;
   d_debugEnabled = mt_debug->seqDebugEnabled;
   d_timedBlocksDt = mt_debug->seqTimedBlocksDt;
   d_timeMetricsEnabled = mt_debug->seqMetricsEnabled;
   //###### END DEBUG ############

   // The base-class' initialize() method should have been called by the base class constructor
   // unless somebody moved it!
   m_nextJobBlock.release();
   m_getTileBlock.release();
   ossimTimer::instance()->setStartTick();
}

//*************************************************************************************************
// Destructor
//*************************************************************************************************
ossimMultiThreadSequencer::~ossimMultiThreadSequencer()
{
   m_inputChain = 0; 
   m_jobMtQueue = 0;
   m_callback.reset();
}

//*************************************************************************************************
//*************************************************************************************************
void ossimMultiThreadSequencer::setToStartOfSequence()
{
   // Reset important indices:
   theCurrentTileNumber = 0;
   m_nextTileID = 0;
   m_totalNumberOfTiles = theNumberOfTilesHorizontal * theNumberOfTilesVertical;

   if (theInputConnection ==  NULL)
      return;

   // Check if this param was already set externally. Query the system capability if not:
   if (m_numThreads == 0)
   {
      m_numThreads = 2 * ossim::getNumberOfThreads();
      m_maxCacheSize = m_maxTileCacheFactor * m_numThreads;
   }

   // Adapt the input source to be an ossimImageChainMtAdaptor since we can only work
   // with this type:
   m_inputChain = dynamic_cast<ossimImageChainMtAdaptor*>(theInputConnection);
   if (m_inputChain.valid())
   {
      m_inputChain->setNumberOfThreads(m_numThreads);
      m_inputChain->setUseSharedHandlers(d_useSharedHandlers);
      m_inputChain->setCacheTileSize(d_cacheTileSize);
      m_inputChain->setUseCache(d_useCache);
   }
   else
   {
      // Need to adapt input. First, is it a chain?
      ossimImageChain* chain = dynamic_cast<ossimImageChain*>(theInputConnection);
      if (chain == NULL)
      {
         // The input is just a common image source. Make it a chain:
         chain = new ossimImageChain;
         chain->add(theInputConnection);
      }

      // This instantiation creates a set of cloned image chains, one per thread, that will be
      // accessed in parallel for the getTile() operation:
      m_inputChain = new ossimImageChainMtAdaptor(chain, m_numThreads, d_useSharedHandlers, d_useCache, d_cacheTileSize);
   }

   // Set the output of the chain to be this sequencer:
   m_inputChain->disconnectAllOutputs();
   //connectMyInputTo(m_inputChain.get());
   //setAreaOfInterest(m_inputChain->getBoundingRect());

   for (ossim_uint32 i=0; i<m_numThreads; ++i)
   {
      std::shared_ptr<ossimGetTileJob> job = std::make_shared<ossimGetTileJob>(m_nextTileID++, i, *this);
      job->setCallback(m_callback);
      job->t_launchNewJob = false;
      job->start();
   }

   // Set up the job queue and fill it with first N jobs:
   ossim_uint32 num_jobs_to_launch =  min<ossim_uint32>(m_numThreads, m_totalNumberOfTiles);
   std::shared_ptr<ossimJobQueue> jobQueue = std::make_shared<ossimJobQueue>();
   for (ossim_uint32 chain_id=0; chain_id<num_jobs_to_launch; ++chain_id)
   {
      if (d_debugEnabled)
      {
         ostringstream s;
         s<<"setToStartOfSequence() -- Creating tile/job #"<<m_nextTileID;
         print(s);
      }

      std::shared_ptr<ossimGetTileJob> job = std::make_shared<ossimGetTileJob>(m_nextTileID++, chain_id, *this);
      job->setCallback(m_callback);
      jobQueue->add(job, false);
   }

   // Initialize the multi-thread queue. Note the setJobQueue is done after construction as it was 
   // crashing do to jobs being launched during init:
   m_jobMtQueue = std::make_shared<ossimJobMultiThreadQueue>(nullptr, num_jobs_to_launch);
   m_jobMtQueue->setJobQueue(jobQueue);
}


//*************************************************************************************************
//*************************************************************************************************
ossimRefPtr<ossimImageData> ossimMultiThreadSequencer::getNextTile(ossim_uint32 /*resLevel*/)
{
   if (!m_inputChain.valid())
      return NULL;

   // May need to initiate the threaded sequencing if not already done:
   if (m_nextTileID == 0)
      setToStartOfSequence();

   // Terminate with null return if done:
   ossimRefPtr<ossimImageData> tile = 0;
   if (theCurrentTileNumber >= m_totalNumberOfTiles)
   {
      return tile;
   }

   // May need to wait until the corresponding job is finished if the tile is not in the cache:
   TileCache::iterator tile_iter = m_tileCache.begin();
   while (!tile.valid()) 
   {
      // If the tile is not yet copied into the cache, it means the job is still running. Let's 
      // block this thread and let the getTile jobs unlock as they finish. We'll exit this loop
      // when the job of interest finishes.
      if (d_timeMetricsEnabled)
         d_t1 = ossimTimer::instance()->time_s(); 
      m_cacheMutex.lock();
      if (d_timeMetricsEnabled)
         d_idleTime1 += ossimTimer::instance()->time_s() - d_t1; 

      // RP - Just grab the first tile for better performance, because order does not matter, we need
      // to process them all
      tile_iter = m_tileCache.begin(); //.find(theCurrentTileNumber);
      m_cacheMutex.unlock();

      if (tile_iter == m_tileCache.end())
      {
         if (d_debugEnabled)
         {
            ostringstream s1;
            s1<<"getNextTile() -- Waiting on tile #"<<theCurrentTileNumber;
            m_cacheMutex.lock();
            s1<<"\n   cache size = "<<m_tileCache.size();
            TileCache::iterator iter = m_tileCache.begin();
            while(iter != m_tileCache.end())
            {
               s1<<"\n   cache.tile_id = "<<iter->first;
               iter++;
            }
            m_cacheMutex.unlock();
            print(s1);
         }

         if (d_timedBlocksDt > 0)
            m_getTileBlock.block(d_timedBlocksDt); 
         else
         {
            m_getTileBlock.reset();
            if (d_timeMetricsEnabled)
               d_t1 = ossimTimer::instance()->time_s(); 
            m_getTileBlock.block();
            if (d_timeMetricsEnabled)
               d_idleTime2 += ossimTimer::instance()->time_s() - d_t1; 
         }
      }
      else
      {
         // A valid tile was found. Need to assign the output tile and free up the reference in the
         // cache:
         if (d_debugEnabled)
         {
            ostringstream s2;
            s2<<"getNextTile() -- Copying tile #"<<theCurrentTileNumber<<".  Cache size: "<<m_tileCache.size();
            print(s2);
         }
         tile = tile_iter->second;
         
         if (d_timeMetricsEnabled)
            d_t1 = ossimTimer::instance()->time_s(); 
         m_cacheMutex.lock();
         m_tileCache.erase(tile_iter);
         m_cacheMutex.unlock();
         if (d_timeMetricsEnabled)
            d_idleTime3 += ossimTimer::instance()->time_s() - d_t1; 

         if (m_tileCache.empty()) 
            ++d_cacheEmptyCount; 
         m_nextJobBlock.release(); // nextJob() may be blocked until cache space is freed
      }
   }

   // Advance the caller-requested tile ID. This is different from the last threaded getTile()'s
   // tile index maintained in m_nextTileID and advanced in initNextJob():
   ++theCurrentTileNumber;
   return tile;
}

//*************************************************************************************************
// Specifies number of thread to support. Default behavior (if this method is never called) is
// query the system for number of cores available.
//*************************************************************************************************
void ossimMultiThreadSequencer::setNumberOfThreads(ossim_uint32 num_threads)
{
   m_numThreads = num_threads;
   m_maxCacheSize = m_maxTileCacheFactor * m_numThreads;

   if (m_inputChain.valid())
      m_inputChain->setNumberOfThreads(num_threads);

   if (m_jobMtQueue && m_jobMtQueue->hasJobsToProcess())
      m_jobMtQueue->getJobQueue()->clear();

   m_nextTileID = 0; // effectively resets this sequencer
}

void ossimMultiThreadSequencer::setUseSharedHandlers(bool use_shared_handlers)
{
   d_useSharedHandlers = use_shared_handlers;

   if (m_inputChain.valid())
      m_inputChain->setUseSharedHandlers(use_shared_handlers);
}

void ossimMultiThreadSequencer::setCacheTileSize(ossim_uint32 cache_tile_size)
{
   d_cacheTileSize = cache_tile_size;

   if (m_inputChain.valid())
      m_inputChain->setCacheTileSize(cache_tile_size);
}

void ossimMultiThreadSequencer::setUseCache(bool use_cache)
{
   d_useCache = use_cache;

   if (m_inputChain.valid())
      m_inputChain->setUseCache(use_cache);
}

//*************************************************************************************************
//*************************************************************************************************
void ossimMultiThreadSequencer::setTileInCache(ossim_uint32 tile_id, 
                                               ossimImageData* tile, 
                                               ossim_uint32 chain_id,
                                               double dt)
{
   if (d_timeMetricsEnabled)
      d_t1 = ossimTimer::instance()->time_s(); 
   std::lock_guard<std::mutex> lock(m_cacheMutex);
   if (d_timeMetricsEnabled)
      d_idleTime4 += ossimTimer::instance()->time_s() - d_t1; 

   d_jobGetTileT += dt;

   m_tileCache[tile_id] = tile;
   if (d_debugEnabled)
   {
      ostringstream s2;
      s2<<"THREAD #"<<chain_id<<" -- setTileInCache() Wrote tile #"<<tile_id;
      print(s2);
   }
   if (d_maxCacheUsed < m_tileCache.size())
      d_maxCacheUsed = (ossim_uint32) m_tileCache.size();
}

//*************************************************************************************************
// Queues up the next getTile job if cache is not full. This is called as soon as the job
// handling the corresponding chain ID is finished.
//*************************************************************************************************
void ossimMultiThreadSequencer::nextJob(ossim_uint32 chain_id)
{
   // Check for end of sequence:
   if (m_nextTileID >= m_totalNumberOfTiles)
      return;

   while (((ossim_uint32) m_tileCache.size()) >= m_maxCacheSize)
   {
      if (d_debugEnabled)
      {
         m_cacheMutex.lock();
         TileCache::const_iterator iter = m_tileCache.begin();
         ostringstream s1;
         s1<<"THREAD #"<<chain_id<<" -- nextJob() Waiting on cache before queuing tile/job #"
            <<m_nextTileID<<"using chain #"<<chain_id<<". Cache size: "<<m_tileCache.size();
         while(iter != m_tileCache.end())
         {
            s1<<"\n   cache.tile_id = "<<iter->first;
            iter++;
         }
         m_cacheMutex.unlock();
         print(s1);
      }

      if (d_timedBlocksDt > 0)
         m_nextJobBlock.block(d_timedBlocksDt);
      else
      {
         m_nextJobBlock.reset();
         if (d_timeMetricsEnabled)
            d_t1 = ossimTimer::instance()->time_s(); 
         m_nextJobBlock.block();
         if (d_timeMetricsEnabled)
         d_idleTime5 += ossimTimer::instance()->time_s() - d_t1; 
      }
   }

   if (d_debugEnabled)
   {
      ostringstream s2;
      s2<<"THREAD #"<<chain_id<<" -- nextJob() Queuing tile/job #"<<m_nextTileID;
      print(s2);
   }

   // Job queue will receive pointer into ossimRefPtr so no leak here:
   if (d_timeMetricsEnabled)
      d_t1 = ossimTimer::instance()->time_s(); 
   std::lock_guard<std::mutex> lock(m_jobMutex);
   if (d_timeMetricsEnabled)
      d_idleTime6 += ossimTimer::instance()->time_s() - d_t1; 

   std::shared_ptr<ossimGetTileJob> job = std::make_shared<ossimGetTileJob>(m_nextTileID++, chain_id, *this);
   job->setCallback(m_callback);
   m_jobMtQueue->getJobQueue()->add(job);
}

//*************************************************************************************************
// For Debugging
//*************************************************************************************************
void ossimMultiThreadSequencer::print(ostringstream& msg) const
{
   std::lock_guard<std::mutex> lock(d_printMutex);
   cerr << msg.str() << endl;
}
double ossimMultiThreadSequencer::handlerGetTileT() 
{
   if (m_inputChain->m_sharedHandlers.empty())
      return -1.0;
   ossimRefPtr<ossimImageHandlerMtAdaptor> ha = m_inputChain->m_sharedHandlers[0].get();
   if (ha.valid())
      return ha->d_getTileT;
   return -1;
}

bool ossimMultiThreadSequencer::loadState(const ossimKeywordlist& kwl, const char* prefix)
{
   const char* lookup;
   lookup = kwl.find(prefix, "num_threads");
   if(lookup)
   {
      ossim_uint32 num_threads = ossimString(lookup).toUInt32();
      setNumberOfThreads(num_threads);
   }
   lookup = kwl.find(prefix, "use_shared_handlers");
   if(lookup)
   {
      bool use_shared_handlers = ossimString(lookup).toBool();
      setUseSharedHandlers(use_shared_handlers);
   }
   lookup = kwl.find(prefix, "cache_tile_size");
   if(lookup)
   {
      ossim_uint32 cache_tile_size = ossimString(lookup).toUInt32();
      setCacheTileSize(cache_tile_size);
   }
   lookup = kwl.find(prefix, "use_cache");

   if(lookup)
   {
      bool use_cache = ossimString(lookup).toBool();
      setUseCache(use_cache);
   }

   bool status = ossimImageSourceSequencer::loadState(kwl, prefix);

   return status;
}