src/framework/client/handle-ext.cpp

   1 /*
   2  *  Copyright (c) 2016 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        handle-ext.cpp
  18  * @author      Kyungwook Tak (k.tak@samsung.com)
  19  * @version     1.0
  20  * @brief       handle with async request extension
  21  */
  22 #include "client/handle-ext.h"
  23
  24 #include <algorithm>
  25
  26 #include "client/utils.h"
  27 #include "common/dispatcher.h"
  28 #include "common/audit/logger.h"
  29
  30 namespace Csr {
  31 namespace Client {
  32
  33 HandleExt::HandleExt(std::shared_ptr<Context> &&context) :
  34         Handle(std::move(context)),
  35         m_stop(false)
  36 {
  37 }
  38
  39 HandleExt::~HandleExt()
  40 {
  41         DEBUG("Destroying extended handle... join all workers...");
  42         eraseJoinableIf();
  43 }
  44
  45 void HandleExt::stop()
  46 {
  47         DEBUG("Stop & join all workers...");
  48         m_stop = true;
  49         eraseJoinableIf();
  50 }
  51
  52 bool HandleExt::isStopped() const
  53 {
  54         return m_stop.load();
  55 }
  56
  57 void HandleExt::eraseJoinableIf(std::function<bool(const WorkerMapPair &)> pred)
  58 {
  59         std::unique_lock<std::mutex> l(m_mutex);
  60         DEBUG("clean joinable workers! current worker map size: " <<
  61                   m_workerMap.size());
  62         auto it = m_workerMap.begin();
  63
  64         while (it != m_workerMap.end()) {
  65                 DEBUG("Worker map traversing to erase! current iter tid: " << it->first);
  66
  67                 if (!it->second.t.joinable())
  68                         throw std::logic_error(FORMAT("All workers should be joinable "
  69                                                                                   "but it isn't. tid: " << it->first));
  70
  71                 if (!pred(*it)) {
  72                         ++it;
  73                         continue;
  74                 }
  75
  76                 DEBUG("Joining worker! tid:" << it->first);
  77                 l.unlock();
  78                 it->second.t.join(); // release lock for worker who calls done()
  79                 l.lock();
  80                 DEBUG("Joined worker! tid:" << it->first);
  81                 it = m_workerMap.erase(it);
  82         }
  83 }
  84
  85 void HandleExt::done()
  86 {
  87         std::lock_guard<std::mutex> l(m_mutex);
  88         auto it = m_workerMap.find(std::this_thread::get_id());
  89
  90         if (it == m_workerMap.end())
  91                 throw std::logic_error(FORMAT("worker done but it's not registered in map. "
  92                                                                           "tid: " << std::this_thread::get_id()));
  93
  94         it->second.isDone = true;
  95 }
  96
  97 void HandleExt::dispatchAsync(const Task &f)
  98 {
  99         eraseJoinableIf([](const WorkerMapPair & pair) {
 100                 return pair.second.isDone.load();
 101         });
 102         // TODO: how to handle exceptions in workers
 103         std::thread t([this, f] {
 104                 DEBUG("client async thread dispatched! tid: " << std::this_thread::get_id());
 105
 106                 f();
 107                 done();
 108
 109                 DEBUG("client async thread done! tid: " << std::this_thread::get_id());
 110         });
 111         {
 112                 std::lock_guard<std::mutex> l(m_mutex);
 113                 m_workerMap.emplace(t.get_id(), std::move(t));
 114         }
 115 }
 116
 117 HandleExt::Worker::Worker() : isDone(false)
 118 {
 119         DEBUG("Worker default constructor called");
 120 }
 121
 122 HandleExt::Worker::Worker(std::thread &&_t) :
 123         isDone(false),
 124         t(std::forward<std::thread>(_t))
 125 {
 126 }
 127
 128 HandleExt::Worker::Worker(HandleExt::Worker &&other) :
 129         isDone(other.isDone.load()),
 130         t(std::move(other.t))
 131 {
 132 }
 133
 134 HandleExt::Worker &HandleExt::Worker::operator=(HandleExt::Worker &&other)
 135 {
 136         isDone = other.isDone.load();
 137         t = std::move(other.t);
 138         return *this;
 139 }
 140
 141 } // namespace Client
 142 } // namespace Csr