Line data    Source code

       1              : /*
       2              :  * If not stated otherwise in this file or this component's LICENSE file the
       3              :  * following copyright and licenses apply:
       4              :  *
       5              :  * Copyright 2022 Sky UK
       6              :  *
       7              :  * Licensed under the Apache License, Version 2.0 (the "License");
       8              :  * you may not use this file except in compliance with the License.
       9              :  * You may obtain a copy of the License at
      10              :  *
      11              :  * http://www.apache.org/licenses/LICENSE-2.0
      12              :  *
      13              :  * Unless required by applicable law or agreed to in writing, software
      14              :  * distributed under the License is distributed on an "AS IS" BASIS,
      15              :  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
      16              :  * See the License for the specific language governing permissions and
      17              :  * limitations under the License.
      18              :  */
      19              : 
      20              : #include "EventThread.h"
      21              : #include "RialtoCommonLogging.h"
      22              : 
      23              : #include <pthread.h>
      24              : #include <semaphore.h>
      25              : #include <unistd.h>
      26              : 
      27              : namespace firebolt::rialto::common
      28              : {
      29            6 : std::shared_ptr<IEventThreadFactory> IEventThreadFactory::createFactory()
      30              : {
      31            6 :     std::shared_ptr<IEventThreadFactory> factory;
      32              : 
      33              :     try
      34              :     {
      35            6 :         factory = std::make_shared<EventThreadFactory>();
      36              :     }
      37            0 :     catch (const std::exception &e)
      38              :     {
      39            0 :         RIALTO_COMMON_LOG_ERROR("Failed to create the event thread factory, reason: %s", e.what());
      40              :     }
      41              : 
      42            6 :     return factory;
      43              : }
      44              : 
      45            6 : std::unique_ptr<IEventThread> EventThreadFactory::createEventThread(std::string threadName) const
      46              : {
      47            6 :     return std::make_unique<EventThread>(threadName);
      48              : }
      49              : 
      50            6 : EventThread::EventThread(std::string threadName) : m_kThreadName(std::move(threadName)), m_shutdown(false)
      51              : {
      52            6 :     m_thread = std::thread(&EventThread::threadExecutor, this);
      53              : }
      54              : 
      55           18 : EventThread::~EventThread()
      56              : {
      57              :     {
      58            6 :         std::lock_guard<std::mutex> locker(m_lock);
      59            6 :         m_shutdown = true;
      60            6 :         m_cond.notify_all();
      61              :     }
      62              : 
      63            6 :     if (m_thread.joinable())
      64            6 :         m_thread.join();
      65           12 : }
      66              : 
      67            6 : void EventThread::threadExecutor()
      68              : {
      69            6 :     if (!m_kThreadName.empty())
      70              :     {
      71            6 :         pthread_setname_np(pthread_self(), m_kThreadName.c_str());
      72              :     }
      73              : 
      74            6 :     std::unique_lock<std::mutex> locker(m_lock);
      75              : 
      76              :     while (true)
      77              :     {
      78           28 :         while (!m_shutdown && m_funcs.empty())
      79            7 :             m_cond.wait(locker);
      80              : 
      81           21 :         if (m_shutdown)
      82            6 :             break;
      83              : 
      84           15 :         std::function<void()> func = std::move(m_funcs.front());
      85           15 :         m_funcs.pop_front();
      86              : 
      87           15 :         m_lock.unlock();
      88              : 
      89           15 :         if (func)
      90           15 :             func();
      91              : 
      92           15 :         m_lock.lock();
      93              :     }
      94            6 : }
      95              : 
      96            2 : void EventThread::flush()
      97              : {
      98              :     sem_t semaphore;
      99            2 :     sem_init(&semaphore, 0, 0);
     100              : 
     101              :     // add a simple function to release the semaphore in the context of the event thread
     102            2 :     addImpl(
     103            4 :         [sem = &semaphore]()
     104              :         {
     105            2 :             if (sem_post(sem) != 0)
     106            0 :                 RIALTO_COMMON_LOG_SYS_ERROR(errno, "failed to signal semaphore");
     107            2 :         });
     108              : 
     109              :     // wait for the above call to unblock the semaphore
     110            2 :     TEMP_FAILURE_RETRY(sem_wait(&semaphore));
     111              : }
     112              : 
     113           15 : void EventThread::addImpl(std::function<void()> &&func)
     114              : {
     115           15 :     std::lock_guard<std::mutex> locker(m_lock);
     116           15 :     m_funcs.emplace_back(std::move(func));
     117           15 :     m_cond.notify_all();
     118              : }
     119              : 
     120              : }; // namespace firebolt::rialto::common