modules/core/src/thread.cpp

   1 /*
   2  * Copyright (c) 2011 Samsung Electronics Co., Ltd All Rights Reserved
   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  * @file        thread.cpp
  18  * @author      Przemyslaw Dobrowolski (p.dobrowolsk@samsung.com)
  19  * @version     1.0
  20  * @brief       This file is the implementation file of thread
  21  */
  22 #include <stddef.h>
  23 #include <dpl/thread.h>
  24 #include <dpl/log/wrt_log.h>
  25 #include <sys/time.h>
  26 #include <algorithm>
  27 #include <dpl/assert.h>
  28 #include <errno.h>
  29 #include <time.h>
  30 #include <string.h>
  31
  32 namespace // anonymous
  33 {
  34 static const size_t NANOSECONDS_PER_SECOND =
  35     static_cast<uint64_t>(1000 * 1000 * 1000);
  36
  37 static const size_t NANOSECONDS_PER_MILISECOND =
  38     static_cast<uint64_t>(1000 * 1000);
  39
  40 static const size_t NANOSECONDS_PER_MICROSECOND =
  41     static_cast<uint64_t>(1000);
  42
  43 static const std::thread::id g_mainThread = std::this_thread::get_id();
  44
  45 class ThreadSpecific
  46 {
  47   public:
  48     pthread_key_t threadSpecific;
  49
  50     ThreadSpecific() :
  51         threadSpecific(0)
  52     {
  53         threadSpecific = 0;
  54         pthread_key_create(&threadSpecific, NULL);
  55     }
  56
  57     virtual ~ThreadSpecific()
  58     {
  59         pthread_key_delete(threadSpecific);
  60     }
  61 };
  62
  63 static ThreadSpecific g_threadSpecific;
  64 } // namespace anonymous
  65
  66 namespace DPL {
  67 bool g_TLSforMainCreated = false;
  68
  69 Thread::Thread() :
  70     m_thread(),
  71     m_abandon(false),
  72     m_running(false),
  73     m_directInvoke(false)
  74 {}
  75
  76 Thread::~Thread()
  77 {
  78     // Ensure that we quit thread
  79     // Always wait thread by yourself; if thread is still running
  80     // this may be sometimes very bad. When derived, some resources
  81     // may leak or be doubly freed
  82     Quit();
  83
  84     // Remove any remainig events
  85     // Thread proc is surely not running now
  86     for (InternalEventList::iterator iterator = m_eventList.begin();
  87          iterator != m_eventList.end();
  88          ++iterator)
  89     {
  90         iterator->eventDeleteProc(iterator->event, iterator->userParam);
  91     }
  92
  93     m_eventList.clear();
  94 }
  95
  96 bool Thread::IsMainThread()
  97 {
  98     return (std::this_thread::get_id() == g_mainThread);
  99 }
 100
 101 Thread *Thread::GetCurrentThread()
 102 {
 103     if (std::this_thread::get_id() == g_mainThread) {
 104         return NULL;
 105     }
 106
 107     void *threadSpecific = pthread_getspecific(g_threadSpecific.threadSpecific);
 108
 109     // Is this a managed thread ?
 110     if (threadSpecific == NULL) {
 111         Throw(Exception::UnmanagedThread);
 112     }
 113
 114     return static_cast<Thread *>(threadSpecific);
 115 }
 116
 117 void *Thread::StaticThreadEntry(void *param)
 118 {
 119     char *errstr = NULL;
 120     char errbuf[512] = {0,};
 121     WrtLogD("Entered static thread entry");
 122
 123     // Retrieve context
 124     Thread *This = static_cast<Thread *>(param);
 125     Assert(This != NULL);
 126
 127     // Set thread specific
 128     int result = pthread_setspecific(g_threadSpecific.threadSpecific, This);
 129
 130     if (result != 0) {
 131 #ifdef _GNU_SOURCE
 132         errstr = strerror_r(result, errbuf, sizeof(errbuf));
 133 #else
 134         strerror_r(result, errbuf, sizeof(errbuf));
 135         errstr = errbuf;
 136 #endif
 137         WrtLogE("Failed to set threadSpecific. Error: %s", errstr);
 138     }
 139
 140     // Enter thread proc
 141     // Do not allow exceptions to hit pthread core
 142     UNHANDLED_EXCEPTION_HANDLER_BEGIN
 143     {
 144         This->ThreadEntry();
 145     }
 146     UNHANDLED_EXCEPTION_HANDLER_END
 147
 148     // Critical section
 149     {
 150         // Leave running state
 151         std::lock_guard<std::mutex> lock(This->m_stateMutex);
 152
 153         This->m_running = false;
 154
 155         // Abandon thread
 156         if (This->m_abandon) {
 157             WrtLogD("Thread was abandoned");
 158             This->m_thread.detach();
 159         } else {
 160             WrtLogD("Thread is joinable");
 161         }
 162     }
 163
 164     return NULL;
 165 }
 166
 167 int Thread::ThreadEntry()
 168 {
 169     WrtLogD("Entered default thread entry");
 170     return Exec();
 171 }
 172
 173 void Thread::ProcessEvents()
 174 {
 175     WrtLogD("Processing events");
 176
 177     // Steal current event list
 178     InternalEventList stolenEvents;
 179
 180     // Enter event list critical section
 181     {
 182         std::lock_guard<std::mutex> lock(m_eventMutex);
 183         m_eventList.swap(stolenEvents);
 184         m_eventInvoker.Reset();
 185     }
 186
 187     // Process event list
 188     WrtLogD("Stolen %u internal events", stolenEvents.size());
 189
 190     for (InternalEventList::iterator iterator = stolenEvents.begin();
 191          iterator != stolenEvents.end();
 192          ++iterator)
 193     {
 194         // Dispatch immediate event
 195         iterator->eventDispatchProc(iterator->event, iterator->userParam);
 196
 197         // Delete event
 198         iterator->eventDeleteProc(iterator->event, iterator->userParam);
 199     }
 200 }
 201
 202 void Thread::ProcessTimedEvents()
 203 {
 204     // Critical section on timed events mutex
 205     {
 206         std::lock_guard<std::mutex> lock(m_timedEventMutex);
 207
 208         // Get current time
 209         unsigned long currentTimeMiliseconds = GetCurrentTimeMiliseconds();
 210
 211         // Info
 212         WrtLogD("Processing timed events. Time now: %lu ms", currentTimeMiliseconds);
 213
 214         // All timed events are sorted chronologically
 215         // Emit timed out events
 216         while (!m_timedEventVector.empty() &&
 217                currentTimeMiliseconds >=
 218                m_timedEventVector.begin()->registerTimeMiliseconds +
 219                m_timedEventVector.begin()->dueTimeMiliseconds)
 220         {
 221             // Info
 222             WrtLogD("Transforming timed event into immediate event. Absolute due time: %lu ms",
 223                     (m_timedEventVector.begin()->registerTimeMiliseconds +
 224                      m_timedEventVector.begin()->dueTimeMiliseconds));
 225
 226             // Emit immediate event
 227             PushEvent(m_timedEventVector.begin()->event,
 228                       m_timedEventVector.begin()->eventDispatchProc,
 229                       m_timedEventVector.begin()->eventDeleteProc,
 230                       m_timedEventVector.begin()->userParam);
 231
 232             // Remove timed eventand fix heap
 233             std::pop_heap(m_timedEventVector.begin(), m_timedEventVector.end());
 234             m_timedEventVector.pop_back();
 235         }
 236     }
 237 }
 238
 239 unsigned long Thread::GetCurrentTimeMiliseconds() const
 240 {
 241     timeval tv;
 242     gettimeofday(&tv, NULL);
 243     return static_cast<unsigned long>(tv.tv_sec) * 1000 +
 244            static_cast<unsigned long>(tv.tv_usec) / 1000;
 245 }
 246
 247 int Thread::Exec()
 248 {
 249     WrtLogD("Executing thread event processing");
 250
 251     const std::size_t MIN_HANDLE_LIST_SIZE = 4;
 252
 253     // Start processing of events
 254     WaitableHandleListEx handleList;
 255
 256     // index 0: Quit waitable event handle
 257     handleList.push_back(std::make_pair(m_quitEvent.GetHandle(), WaitMode::Read));
 258
 259     // index 1: Event occurred event handle
 260     handleList.push_back(std::make_pair(m_eventInvoker.GetHandle(),
 261                                         WaitMode::Read));
 262
 263     // index 2: Timed event occurred event handle
 264     handleList.push_back(std::make_pair(m_timedEventInvoker.GetHandle(),
 265                                         WaitMode::Read));
 266
 267     // index 3: Waitable handle watch support invoker
 268     handleList.push_back(std::make_pair(WaitableHandleWatchSupport::
 269                                             WaitableInvokerHandle(),
 270                                         WaitMode::Read));
 271
 272     //
 273     // Watch list might have been initialized before threaded started
 274     // Need to fill waitable event watch list in this case
 275     //
 276     {
 277         WaitableHandleListEx waitableHandleWatchHandles =
 278             WaitableHandleWatchSupport::WaitableWatcherHandles();
 279         std::copy(
 280             waitableHandleWatchHandles.begin(),
 281             waitableHandleWatchHandles.end(), std::back_inserter(handleList));
 282     }
 283
 284     // Quit flag
 285     bool quit = false;
 286
 287     while (!quit) {
 288         // Retrieve minimum wait time, according to timed events list
 289         unsigned long minimumWaitTime;
 290
 291         // Critical section on timed events mutex
 292         {
 293             std::lock_guard<std::mutex> lock(m_timedEventMutex);
 294
 295             if (!m_timedEventVector.empty()) {
 296                 unsigned long currentTimeMiliseconds =
 297                     GetCurrentTimeMiliseconds();
 298                 unsigned long destinationTimeMiliseconds =
 299                     m_timedEventVector.begin()->registerTimeMiliseconds +
 300                     m_timedEventVector.begin()->dueTimeMiliseconds;
 301
 302                 // Are we already late with timed event ?
 303                 if (currentTimeMiliseconds > destinationTimeMiliseconds) {
 304                     minimumWaitTime = 0;
 305                 } else {
 306                     minimumWaitTime = destinationTimeMiliseconds -
 307                         currentTimeMiliseconds;
 308                 }
 309             } else {
 310                 minimumWaitTime = 0xFFFFFFFF; // Infinity
 311             }
 312         }
 313
 314         // Info
 315         WrtLogD("Thread loop minimum wait time: %lu ms", minimumWaitTime);
 316
 317         // Do thread waiting
 318         WaitableHandleIndexList waitableHandleIndexList =
 319             WaitForMultipleHandles(handleList, minimumWaitTime);
 320
 321         if (waitableHandleIndexList.empty()) {
 322             // Timeout occurred. Process timed events.
 323             WrtLogD("Timed event list elapsed invoker");
 324             ProcessTimedEvents();
 325             continue;
 326         }
 327
 328         // Go through each index
 329         for (WaitableHandleIndexList::const_iterator
 330              waitableHandleIndexIterator = waitableHandleIndexList.begin();
 331              waitableHandleIndexIterator != waitableHandleIndexList.end();
 332              ++waitableHandleIndexIterator)
 333         {
 334             size_t index = *waitableHandleIndexIterator;
 335
 336             WrtLogD("Event loop triggered with index: %u", index);
 337
 338             switch (index) {
 339             case 0:
 340                 // Quit waitable event handle
 341                 quit = true;
 342                 break;
 343
 344             case 1:
 345                 // Event occurred event handle
 346                 ProcessEvents();
 347
 348                 // Handle direct invoker
 349                 if (m_directInvoke) {
 350                     m_directInvoke = false;
 351
 352                     WrtLogD("Handling direct invoker");
 353
 354                     // Update list
 355                     while (handleList.size() > MIN_HANDLE_LIST_SIZE) {
 356                         handleList.pop_back();
 357                     }
 358
 359                     // Insert current waitable event handles instead
 360                     {
 361                         WaitableHandleListEx waitableHandleWatchHandles =
 362                             WaitableHandleWatchSupport::WaitableWatcherHandles();
 363                         std::copy(
 364                             waitableHandleWatchHandles.begin(),
 365                             waitableHandleWatchHandles.end(),
 366                             std::back_inserter(handleList));
 367                     }
 368                 }
 369
 370                 // Done
 371                 break;
 372
 373             case 2:
 374                 // Timed event list changed
 375                 WrtLogD("Timed event list changed invoker");
 376                 ProcessTimedEvents();
 377
 378                 // Reset timed event invoker
 379                 m_timedEventInvoker.Reset();
 380
 381                 // Done
 382                 break;
 383
 384             case 3:
 385                 // Waitable handle watch support invoker
 386                 WrtLogD("Waitable handle watch invoker event occurred");
 387
 388                 // First, remove all previous handles
 389                 while (handleList.size() > MIN_HANDLE_LIST_SIZE) {
 390                     handleList.pop_back();
 391                 }
 392
 393                 // Insert current waitable event handles instead
 394                 {
 395                     WaitableHandleListEx waitableHandleWatchHandles =
 396                         WaitableHandleWatchSupport::WaitableWatcherHandles();
 397                     std::copy(
 398                         waitableHandleWatchHandles.begin(),
 399                         waitableHandleWatchHandles.end(),
 400                         std::back_inserter(handleList));
 401                 }
 402
 403                 // Handle invoker in waitable watch support
 404                 WaitableHandleWatchSupport::InvokerFinished();
 405
 406                 WrtLogD("Waitable handle watch invoker event handled");
 407
 408                 // Done
 409                 break;
 410
 411             default:
 412                 // Waitable event watch list
 413                 WrtLogD("Waitable handle watch event occurred");
 414
 415                 // Handle event in waitable handle watch
 416                 {
 417                     std::pair<WaitableHandle,
 418                               WaitMode::Type> handle = handleList[index];
 419                     WaitableHandleWatchSupport::HandleWatcher(handle.first,
 420                                                               handle.second);
 421                 }
 422
 423                 if (m_directInvoke) {
 424                     m_directInvoke = false;
 425
 426                     WrtLogD("Handling direct invoker");
 427
 428                     // Update list
 429                     while (handleList.size() > MIN_HANDLE_LIST_SIZE) {
 430                         handleList.pop_back();
 431                     }
 432
 433                     // Insert current waitable event handles instead
 434                     {
 435                         WaitableHandleListEx waitableHandleWatchHandles =
 436                             WaitableHandleWatchSupport::
 437                                 WaitableWatcherHandles();
 438                         std::copy(waitableHandleWatchHandles.begin(),
 439                                   waitableHandleWatchHandles.end(),
 440                                   std::back_inserter(handleList));
 441                     }
 442                 }
 443
 444                 WrtLogD("Waitable handle watch event handled");
 445
 446                 // Done
 447                 break;
 448             }
 449         }
 450     }
 451
 452     WrtLogD("Leaving thread event processing");
 453     return 0;
 454 }
 455
 456 void Thread::Run()
 457 {
 458     WrtLogD("Running thread");
 459
 460     // Critical section
 461     {
 462         std::lock_guard<std::mutex> lock(m_stateMutex);
 463
 464         if (m_running) {
 465             return;
 466         }
 467
 468         try{
 469             m_thread = std::thread(StaticThreadEntry,this);
 470         }catch(std::system_error e){
 471             Throw(Exception::RunFailed);
 472         }
 473
 474         // At default, we abandon thread
 475         m_abandon = true;
 476
 477         // Enter running state
 478         m_running = true;
 479     }
 480
 481     WrtLogD("Thread run");
 482 }
 483
 484 void Thread::Quit()
 485 {
 486     // Critical section
 487     {
 488         std::lock_guard<std::mutex> lock(m_stateMutex);
 489
 490         // Is thread running ?
 491         if (!m_running) {
 492             return;
 493         }
 494
 495         WrtLogD("Quitting thread...");
 496
 497         // Do not abandon thread, we will join
 498         m_abandon = false;
 499
 500         // Singal quit waitable event
 501         m_quitEvent.Signal();
 502     }
 503
 504     try{
 505         m_thread.join();
 506     }catch(std::system_error e){
 507         Throw(Exception::QuitFailed);
 508     }
 509
 510     WrtLogD("Thread quit");
 511 }
 512
 513 void Thread::PushEvent(void *event,
 514                        EventDispatchProc eventDispatchProc,
 515                        EventDeleteProc eventDeleteProc,
 516                        void *userParam)
 517 {
 518     // Enter event list critical section
 519     std::lock_guard<std::mutex> lock(m_eventMutex);
 520
 521     // Push new event
 522     m_eventList.push_back(InternalEvent(event, userParam, eventDispatchProc,
 523                                         eventDeleteProc));
 524
 525     // Trigger invoker
 526     m_eventInvoker.Signal();
 527
 528     WrtLogD("Event pushed and invoker signaled");
 529 }
 530
 531 void Thread::PushTimedEvent(void *event,
 532                             double dueTimeSeconds,
 533                             EventDispatchProc eventDispatchProc,
 534                             EventDeleteProc eventDeleteProc,
 535                             void *userParam)
 536 {
 537     // Check for developer errors
 538     Assert(dueTimeSeconds >= 0.0);
 539
 540     // Enter timed event list critical section
 541     std::lock_guard<std::mutex> lock(m_timedEventMutex);
 542
 543     // Get current time
 544     unsigned long currentTimeMiliseconds = GetCurrentTimeMiliseconds();
 545
 546     // Convert to miliseconds
 547     unsigned long dueTimeMiliseconds =
 548         static_cast<unsigned long>(1000.0 * dueTimeSeconds);
 549
 550     // Push new timed event
 551     m_timedEventVector.push_back(InternalTimedEvent(event, userParam,
 552                                                     dueTimeMiliseconds,
 553                                                     currentTimeMiliseconds,
 554                                                     eventDispatchProc,
 555                                                     eventDeleteProc));
 556
 557     // Heapify timed events
 558     std::make_heap(m_timedEventVector.begin(), m_timedEventVector.end());
 559
 560     // Trigger invoker
 561     m_timedEventInvoker.Signal();
 562
 563     WrtLogD("Timed event pushed and invoker signaled: "
 564         "due time: %lu ms, absolute due time: %lu ms",
 565         dueTimeMiliseconds, currentTimeMiliseconds + dueTimeMiliseconds);
 566 }
 567
 568 Thread *Thread::GetInvokerThread()
 569 {
 570     return this;
 571 }
 572
 573 void Thread::HandleDirectInvoker()
 574 {
 575     // We must be in ProcessEvents call stack
 576     // Mark that situation to handle direct invoker
 577     m_directInvoke = true;
 578 }
 579
 580 void Thread::Sleep(uint64_t seconds)
 581 {
 582     NanoSleep(seconds * NANOSECONDS_PER_SECOND);
 583 }
 584
 585 void Thread::MiliSleep(uint64_t miliseconds)
 586 {
 587     NanoSleep(miliseconds * NANOSECONDS_PER_MILISECOND);
 588 }
 589
 590 void Thread::MicroSleep(uint64_t microseconds)
 591 {
 592     NanoSleep(microseconds * NANOSECONDS_PER_MICROSECOND);
 593 }
 594
 595 void Thread::NanoSleep(uint64_t nanoseconds)
 596 {
 597     timespec requestedTime = {
 598         static_cast<time_t>(
 599             nanoseconds / NANOSECONDS_PER_SECOND),
 600
 601         static_cast<long>(
 602             nanoseconds % NANOSECONDS_PER_SECOND)
 603     };
 604
 605     timespec remainingTime;
 606
 607     for (;;) {
 608         if (nanosleep(&requestedTime, &remainingTime) == 0) {
 609             break;
 610         }
 611
 612         int error = errno;
 613         Assert(error == EINTR);
 614
 615         requestedTime = remainingTime;
 616     }
 617 }
 618 } // namespace DPL