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