platform-abstractions/slp/resource-loader/resource-thread-base.cpp

   1 /*
   2  * Copyright (c) 2014 Samsung Electronics Co., Ltd.
   3  *
   4  * Licensed under the Apache License, Version 2.0 (the "License");
   5  * you may not use this file except in compliance with the License.
   6  * You may obtain a copy of the License at
   7  *
   8  * http://www.apache.org/licenses/LICENSE-2.0
   9  *
  10  * Unless required by applicable law or agreed to in writing, software
  11  * distributed under the License is distributed on an "AS IS" BASIS,
  12  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  13  * See the License for the specific language governing permissions and
  14  * limitations under the License.
  15  *
  16  */
  17
  18 #include <dali/integration-api/debug.h>
  19 #include "resource-thread-base.h"
  20 #include "slp-logging.h"
  21 #include "atomics.h"
  22
  23 using namespace Dali::Integration;
  24 using boost::mutex;
  25 using boost::unique_lock;
  26 using boost::scoped_ptr;
  27
  28 namespace Dali
  29 {
  30
  31 // Initial values for the members tracking which resources have been cancelled.
  32 // They start out with different values so that if the first load executed is
  33 // synchronous, it won't be erroneously cancelled.
  34 const Integration::ResourceId NO_REQUEST_IN_FLIGHT = Integration::ResourceId(0) - 1;
  35 const Integration::ResourceId NO_REQUEST_CANCELLED = Integration::ResourceId(0) - 2;
  36
  37 namespace SlpPlatform
  38 {
  39
  40 namespace
  41 {
  42 const char * const IDLE_PRIORITY_ENVIRONMENT_VARIABLE_NAME = "DALI_RESOURCE_THREAD_IDLE_PRIORITY"; ///@Todo Move this to somewhere that other environment variables are declared and document it there.
  43 } // unnamed namespace
  44
  45 /** Thrown by InterruptionPoint() to abort a request early. */
  46 class CancelRequestException {};
  47
  48 ResourceThreadBase::ResourceThreadBase( ResourceLoader& resourceLoader ) :
  49   mResourceLoader( resourceLoader ),
  50   mCurrentRequestId( NO_REQUEST_IN_FLIGHT ),
  51   mCancelRequestId( NO_REQUEST_CANCELLED ),
  52   mPaused( false )
  53 {
  54 #if defined(DEBUG_ENABLED)
  55   mLogFilter = Debug::Filter::New(Debug::Concise, false, "LOG_RESOURCE_THREAD_BASE");
  56 #endif
  57
  58   mThread = new boost::thread(boost::bind(&ResourceThreadBase::ThreadLoop, this));
  59 }
  60
  61 ResourceThreadBase::~ResourceThreadBase()
  62 {
  63   TerminateThread();
  64
  65 #if defined(DEBUG_ENABLED)
  66   delete mLogFilter;
  67 #endif
  68 }
  69
  70 void ResourceThreadBase::TerminateThread()
  71 {
  72   if (mThread)
  73   {
  74     // wake thread
  75     mCondition.notify_all();
  76     // wait for thread to exit
  77     mThread->join();
  78     // delete thread instance
  79     delete mThread;
  80     // mark thread terminated
  81     mThread = NULL;
  82   }
  83 }
  84
  85 void ResourceThreadBase::AddRequest(const ResourceRequest& request, const RequestType type)
  86 {
  87   bool wasEmpty = false;
  88   bool wasPaused = false;
  89
  90   {
  91     // Lock while adding to the request queue
  92     unique_lock<mutex> lock( mMutex );
  93
  94     wasEmpty = mQueue.empty();
  95     wasPaused = mPaused;
  96
  97     mQueue.push_back( std::make_pair(request, type) );
  98   }
  99
 100   if( wasEmpty && !wasPaused )
 101   {
 102     // Wake-up the thread
 103     mCondition.notify_all();
 104   }
 105 }
 106
 107 // Called from outer thread.
 108 void ResourceThreadBase::CancelRequest( const Integration::ResourceId resourceId )
 109 {
 110   bool found = false;
 111   DALI_LOG_INFO( mLogFilter, Debug::Verbose, "%s: %u.\n", __FUNCTION__, unsigned(resourceId) );
 112
 113   // Eliminate the cancelled request from the request queue if it is in there:
 114   {
 115     // Lock while searching and removing from the request queue:
 116     unique_lock<mutex> lock( mMutex );
 117
 118     for( RequestQueueIter iterator = mQueue.begin();
 119          iterator != mQueue.end();
 120          ++iterator )
 121     {
 122       if( ((*iterator).first).GetId() == resourceId )
 123       {
 124         iterator = mQueue.erase( iterator );
 125         found = true;
 126         break;
 127       }
 128     }
 129   }
 130
 131   // Remember the cancelled id for the worker thread to poll at one of its points
 132   // of interruption:
 133   if( !found )
 134   {
 135     Dali::Internal::AtomicWriteToCacheableAlignedAddress( &mCancelRequestId, resourceId );
 136     DALI_LOG_INFO( mLogFilter, Debug::Concise, "%s: Cancelling in-flight resource (%u).\n", __FUNCTION__, unsigned(resourceId) );
 137   }
 138 }
 139
 140 // Called from worker thread.
 141 void ResourceThreadBase::InterruptionPoint() const
 142 {
 143   const Integration::ResourceId cancelled = Dali::Internal::AtomicReadFromCacheableAlignedAddress( &mCancelRequestId );
 144   const Integration::ResourceId current = mCurrentRequestId;
 145
 146   if( current == cancelled )
 147   {
 148     DALI_LOG_INFO( mLogFilter, Debug::Concise, "%s: Cancelled in-flight resource (%u).\n", __FUNCTION__, unsigned(cancelled) );
 149     throw CancelRequestException();
 150   }
 151 }
 152
 153 void ResourceThreadBase::Pause()
 154 {
 155   unique_lock<mutex> lock( mMutex );
 156   mPaused = true;
 157 }
 158
 159 void ResourceThreadBase::Resume()
 160 {
 161   // Clear the paused flag and if we weren't running already, also wake up the background thread:
 162   bool wasPaused = false;
 163   {
 164     unique_lock<mutex> lock( mMutex );
 165     wasPaused = mPaused;
 166     mPaused = false;
 167   }
 168
 169   // If we were paused, wake up the background thread and give it a
 170   // chance to do some work:
 171   if( wasPaused )
 172   {
 173     mCondition.notify_all();
 174   }
 175 }
 176
 177 //----------------- Called from separate thread (mThread) -----------------
 178
 179 void ResourceThreadBase::ThreadLoop()
 180 {
 181   // TODO: Use Environment Options
 182   const char* threadPriorityIdleRequired = std::getenv( IDLE_PRIORITY_ENVIRONMENT_VARIABLE_NAME );
 183   if( threadPriorityIdleRequired )
 184   {
 185     // if the parameter exists then set up an idle priority for this thread
 186     struct sched_param sp;
 187     sp.sched_priority = 0;
 188     sched_setscheduler(0, SCHED_IDLE, &sp);
 189     ///@ToDo: change to the corresponding Pthreads call, not this POSIX.1-2001 one with a Linux-specific argument (SCHED_IDLE): int pthread_setschedparam(pthread_t thread, int policy, const struct sched_param *param);, as specified in the docs for sched_setscheduler(): http://man7.org/linux/man-pages/man2/sched_setscheduler.2.html
 190   }
 191
 192   InstallLogging();
 193
 194   while( !mResourceLoader.IsTerminating() )
 195   {
 196     try
 197     {
 198       WaitForRequests();
 199
 200       if ( !mResourceLoader.IsTerminating() )
 201       {
 202         ProcessNextRequest();
 203       }
 204     }
 205
 206     catch( CancelRequestException& ex )
 207     {
 208       // No problem: a derived class deliberately threw to abort an in-flight request
 209       // that was cancelled.
 210       DALI_LOG_INFO( mLogFilter, Debug::Concise, "%s: Caught cancellation exception for resource (%u).\n", __FUNCTION__, unsigned(mCurrentRequestId) );
 211       CancelRequestException* disableUnusedVarWarning = &ex;
 212       ex = *disableUnusedVarWarning;
 213     }
 214
 215     // Catch all exceptions to avoid killing the process, and log the error:
 216     catch( std::exception& ex )
 217     {
 218       const char * const what = ex.what();
 219       DALI_LOG_ERROR( "std::exception caught in resource thread. Aborting request with id %u because of std::exception with reason, \"%s\".\n", unsigned(mCurrentRequestId), what ? what : "null" );
 220     }
 221     catch( Dali::DaliException& ex )
 222     {
 223       // Probably a failed assert-always:
 224       DALI_LOG_ERROR( "DaliException caught in resource thread. Aborting request with id %u. Location: \"%s\". Condition: \"%s\".\n", unsigned(mCurrentRequestId), ex.mLocation.c_str(), ex.mCondition.c_str() );
 225     }
 226     catch( ... )
 227     {
 228       DALI_LOG_ERROR( "Unknown exception caught in resource thread. Aborting request with id %u.\n", unsigned(mCurrentRequestId) );
 229     }
 230   }
 231 }
 232
 233 void ResourceThreadBase::WaitForRequests()
 234 {
 235   unique_lock<mutex> lock( mMutex );
 236
 237   if( mQueue.empty() || mPaused == true )
 238   {
 239     // Waiting for a wake up from resource loader control thread
 240     // This will be to process a new request or terminate
 241     mCondition.wait(lock);
 242   }
 243 }
 244
 245 void ResourceThreadBase::ProcessNextRequest()
 246 {
 247   ResourceRequest* request(NULL);
 248   RequestType type(RequestLoad);
 249
 250   {
 251     // lock the queue and extract the next request
 252     unique_lock<mutex> lock(mMutex);
 253
 254     if (!mQueue.empty())
 255     {
 256       const RequestInfo & front = mQueue.front();
 257       request = new ResourceRequest( front.first );
 258       type = front.second;
 259       mCurrentRequestId = front.first.GetId();
 260       mQueue.pop_front();
 261     }
 262   } // unlock the queue
 263
 264   // process request outside of lock
 265   if ( NULL != request )
 266   {
 267     std::auto_ptr<ResourceRequest> deleter( request );
 268     switch( type )
 269     {
 270       case RequestLoad:
 271       {
 272         Load(*request);
 273       }
 274       break;
 275
 276       case RequestDecode:
 277       {
 278         Decode(*request);
 279       }
 280       break;
 281
 282       case RequestSave:
 283       {
 284         Save(*request);
 285       }
 286       break;
 287     }
 288   }
 289 }
 290
 291 void ResourceThreadBase::InstallLogging()
 292 {
 293   // resource loading thread will send its logs to SLP Platform's LogMessage handler.
 294   Dali::Integration::Log::InstallLogFunction(Dali::SlpPlatform::LogMessage);
 295 }
 296
 297 void ResourceThreadBase::UninstallLogging()
 298 {
 299   // uninstall it on resource loading thread.
 300   Dali::Integration::Log::UninstallLogFunction();
 301 }
 302
 303 void ResourceThreadBase::Decode(const Integration::ResourceRequest& request)
 304 {
 305   DALI_LOG_TRACE_METHOD(mLogFilter);
 306   DALI_LOG_WARNING("Resource Decoding from a memory buffer not supported for this type.");
 307   ///! If you need this for a subclassed thread, look to ResourceThreadImage::Decode() for an example implementation.
 308 }
 309
 310 } // namespace SlpPlatform
 311
 312 } // namespace Dali