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