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