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
  27 namespace Dali
  28 {
  29
  30 // Initial values for the members tracking which resources have been cancelled.
  31 // They start out with different values so that if the first load executed is
  32 // synchronous, it won't be erroneously cancelled.
  33 const Integration::ResourceId NO_REQUEST_IN_FLIGHT = Integration::ResourceId(0) - 1;
  34 const Integration::ResourceId NO_REQUEST_CANCELLED = Integration::ResourceId(0) - 2;
  35
  36 namespace SlpPlatform
  37 {
  38
  39 namespace
  40 {
  41 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.
  42 } // unnamed namespace
  43
  44 /** Thrown by InterruptionPoint() to abort a request early. */
  45 class CancelRequestException {};
  46
  47 ResourceThreadBase::ResourceThreadBase( ResourceLoader& resourceLoader ) :
  48   mResourceLoader( resourceLoader ),
  49   mCurrentRequestId( NO_REQUEST_IN_FLIGHT ),
  50   mCancelRequestId( NO_REQUEST_CANCELLED ),
  51   mPaused( false )
  52 {
  53 #if defined(DEBUG_ENABLED)
  54   mLogFilter = Debug::Filter::New(Debug::Concise, false, "LOG_RESOURCE_THREAD_BASE");
  55 #endif
  56
  57   mThread = new boost::thread(boost::bind(&ResourceThreadBase::ThreadLoop, this));
  58 }
  59
  60 ResourceThreadBase::~ResourceThreadBase()
  61 {
  62   TerminateThread();
  63
  64 #if defined(DEBUG_ENABLED)
  65   delete mLogFilter;
  66 #endif
  67 }
  68
  69 void ResourceThreadBase::TerminateThread()
  70 {
  71   if (mThread)
  72   {
  73     // wake thread
  74     mCondition.notify_all();
  75     // wait for thread to exit
  76     mThread->join();
  77     // delete thread instance
  78     delete mThread;
  79     // mark thread terminated
  80     mThread = NULL;
  81   }
  82 }
  83
  84 void ResourceThreadBase::AddRequest(const ResourceRequest& request, const RequestType type)
  85 {
  86   bool wasEmpty = false;
  87   bool wasPaused = false;
  88
  89   {
  90     // Lock while adding to the request queue
  91     unique_lock<mutex> lock( mMutex );
  92
  93     wasEmpty = mQueue.empty();
  94     wasPaused = mPaused;
  95
  96     mQueue.push_back( std::make_pair(request, type) );
  97   }
  98
  99   if( wasEmpty && !wasPaused )
 100   {
 101     // Wake-up the thread
 102     mCondition.notify_all();
 103   }
 104 }
 105
 106 // Called from outer thread.
 107 void ResourceThreadBase::CancelRequest( const Integration::ResourceId resourceId )
 108 {
 109   bool found = false;
 110   DALI_LOG_INFO( mLogFilter, Debug::Verbose, "%s: %u.\n", __FUNCTION__, unsigned(resourceId) );
 111
 112   // Eliminate the cancelled request from the request queue if it is in there:
 113   {
 114     // Lock while searching and removing from the request queue:
 115     unique_lock<mutex> lock( mMutex );
 116
 117     for( RequestQueueIter iterator = mQueue.begin();
 118          iterator != mQueue.end();
 119          ++iterator )
 120     {
 121       if( ((*iterator).first).GetId() == resourceId )
 122       {
 123         iterator = mQueue.erase( iterator );
 124         found = true;
 125         break;
 126       }
 127     }
 128   }
 129
 130   // Remember the cancelled id for the worker thread to poll at one of its points
 131   // of interruption:
 132   if( !found )
 133   {
 134     Dali::Internal::AtomicWriteToCacheableAlignedAddress( &mCancelRequestId, resourceId );
 135     DALI_LOG_INFO( mLogFilter, Debug::Concise, "%s: Cancelling in-flight resource (%u).\n", __FUNCTION__, unsigned(resourceId) );
 136   }
 137 }
 138
 139 // Called from worker thread.
 140 void ResourceThreadBase::InterruptionPoint() const
 141 {
 142   const Integration::ResourceId cancelled = Dali::Internal::AtomicReadFromCacheableAlignedAddress( &mCancelRequestId );
 143   const Integration::ResourceId current = mCurrentRequestId;
 144
 145   if( current == cancelled )
 146   {
 147     DALI_LOG_INFO( mLogFilter, Debug::Concise, "%s: Cancelled in-flight resource (%u).\n", __FUNCTION__, unsigned(cancelled) );
 148     throw CancelRequestException();
 149   }
 150 }
 151
 152 void ResourceThreadBase::Pause()
 153 {
 154   unique_lock<mutex> lock( mMutex );
 155   mPaused = true;
 156 }
 157
 158 void ResourceThreadBase::Resume()
 159 {
 160   // Clear the paused flag and if we weren't running already, also wake up the background thread:
 161   bool wasPaused = false;
 162   {
 163     unique_lock<mutex> lock( mMutex );
 164     wasPaused = mPaused;
 165     mPaused = false;
 166   }
 167
 168   // If we were paused, wake up the background thread and give it a
 169   // chance to do some work:
 170   if( wasPaused )
 171   {
 172     mCondition.notify_all();
 173   }
 174 }
 175
 176 //----------------- Called from separate thread (mThread) -----------------
 177
 178 void ResourceThreadBase::ThreadLoop()
 179 {
 180   // TODO: Use Environment Options
 181   const char* threadPriorityIdleRequired = std::getenv( IDLE_PRIORITY_ENVIRONMENT_VARIABLE_NAME );
 182   if( threadPriorityIdleRequired )
 183   {
 184     // if the parameter exists then set up an idle priority for this thread
 185     struct sched_param sp;
 186     sp.sched_priority = 0;
 187     sched_setscheduler(0, SCHED_IDLE, &sp);
 188     ///@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
 189   }
 190
 191   InstallLogging();
 192
 193   while( !mResourceLoader.IsTerminating() )
 194   {
 195     try
 196     {
 197       WaitForRequests();
 198
 199       if ( !mResourceLoader.IsTerminating() )
 200       {
 201         ProcessNextRequest();
 202       }
 203     }
 204
 205     catch( CancelRequestException& ex )
 206     {
 207       // No problem: a derived class deliberately threw to abort an in-flight request
 208       // that was cancelled.
 209       DALI_LOG_INFO( mLogFilter, Debug::Concise, "%s: Caught cancellation exception for resource (%u).\n", __FUNCTION__, unsigned(mCurrentRequestId) );
 210       CancelRequestException* disableUnusedVarWarning = &ex;
 211       ex = *disableUnusedVarWarning;
 212     }
 213
 214     // Catch all exceptions to avoid killing the process, and log the error:
 215     catch( std::exception& ex )
 216     {
 217       const char * const what = ex.what();
 218       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" );
 219     }
 220     catch( Dali::DaliException& ex )
 221     {
 222       // Probably a failed assert-always:
 223       DALI_LOG_ERROR( "DaliException caught in resource thread. Aborting request with id %u. Location: \"%s\". Condition: \"%s\".\n", unsigned(mCurrentRequestId), ex.location, ex.condition );
 224     }
 225     catch( ... )
 226     {
 227       DALI_LOG_ERROR( "Unknown exception caught in resource thread. Aborting request with id %u.\n", unsigned(mCurrentRequestId) );
 228     }
 229   }
 230 }
 231
 232 void ResourceThreadBase::WaitForRequests()
 233 {
 234   unique_lock<mutex> lock( mMutex );
 235
 236   if( mQueue.empty() || mPaused == true )
 237   {
 238     // Waiting for a wake up from resource loader control thread
 239     // This will be to process a new request or terminate
 240     mCondition.wait(lock);
 241   }
 242 }
 243
 244 void ResourceThreadBase::ProcessNextRequest()
 245 {
 246   ResourceRequest* request(NULL);
 247   RequestType type(RequestLoad);
 248
 249   {
 250     // lock the queue and extract the next request
 251     unique_lock<mutex> lock(mMutex);
 252
 253     if (!mQueue.empty())
 254     {
 255       const RequestInfo & front = mQueue.front();
 256       request = new ResourceRequest( front.first );
 257       type = front.second;
 258       mCurrentRequestId = front.first.GetId();
 259       mQueue.pop_front();
 260     }
 261   } // unlock the queue
 262
 263   // process request outside of lock
 264   if ( NULL != request )
 265   {
 266     std::auto_ptr<ResourceRequest> deleter( request );
 267     switch( type )
 268     {
 269       case RequestLoad:
 270       {
 271         Load(*request);
 272       }
 273       break;
 274
 275       case RequestDownload:
 276       {
 277         Download(*request);
 278       }
 279       break;
 280
 281       case RequestDecode:
 282       {
 283         Decode(*request);
 284       }
 285       break;
 286
 287       case RequestSave:
 288       {
 289         Save(*request);
 290       }
 291       break;
 292     }
 293   }
 294 }
 295
 296 void ResourceThreadBase::InstallLogging()
 297 {
 298   // resource loading thread will send its logs to SLP Platform's LogMessage handler.
 299   Dali::Integration::Log::InstallLogFunction(Dali::SlpPlatform::LogMessage);
 300 }
 301
 302 void ResourceThreadBase::UninstallLogging()
 303 {
 304   // uninstall it on resource loading thread.
 305   Dali::Integration::Log::UninstallLogFunction();
 306 }
 307
 308 void ResourceThreadBase::Download(const Integration::ResourceRequest& request)
 309 {
 310   DALI_LOG_TRACE_METHOD(mLogFilter);
 311   DALI_LOG_WARNING("Resource Downloading from a remote server not supported for this type.");
 312   ///! If you need this for a subclassed thread, look to ResourceThreadImage::Download() for an example implementation.
 313 }
 314
 315 void ResourceThreadBase::Decode(const Integration::ResourceRequest& request)
 316 {
 317   DALI_LOG_TRACE_METHOD(mLogFilter);
 318   DALI_LOG_WARNING("Resource Decoding from a memory buffer not supported for this type.");
 319   ///! If you need this for a subclassed thread, look to ResourceThreadImage::Decode() for an example implementation.
 320 }
 321
 322 } // namespace SlpPlatform
 323
 324 } // namespace Dali