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