src/lib/tvgTask.cpp

   1 /*
   2  * Copyright (c) 2020 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
  18 #include <tvgTask.h>
  19
  20 #include <deque>
  21
  22 template <typename Task>
  23 class TaskQueue {
  24     using lock_t = std::unique_lock<std::mutex>;
  25     std::deque<Task>        _q;
  26     bool                    _done{false};
  27     std::mutex              _mutex;
  28     std::condition_variable _ready;
  29
  30 public:
  31     bool try_pop(Task &task)
  32     {
  33         lock_t lock{_mutex, std::try_to_lock};
  34         if (!lock || _q.empty()) return false;
  35         task = std::move(_q.front());
  36         _q.pop_front();
  37         return true;
  38     }
  39
  40     bool try_push(Task &&task)
  41     {
  42         {
  43             lock_t lock{_mutex, std::try_to_lock};
  44             if (!lock) return false;
  45             _q.push_back(std::move(task));
  46         }
  47         _ready.notify_one();
  48         return true;
  49     }
  50
  51     void done()
  52     {
  53         {
  54             lock_t lock{_mutex};
  55             _done = true;
  56         }
  57         _ready.notify_all();
  58     }
  59
  60     bool pop(Task &task)
  61     {
  62         lock_t lock{_mutex};
  63         while (_q.empty() && !_done) _ready.wait(lock);
  64         if (_q.empty()) return false;
  65         task = std::move(_q.front());
  66         _q.pop_front();
  67         return true;
  68     }
  69
  70     void push(Task &&task)
  71     {
  72         {
  73             lock_t lock{_mutex};
  74             _q.push_back(std::move(task));
  75         }
  76         _ready.notify_one();
  77     }
  78
  79 };
  80
  81 #include <thread>
  82 #include <vector>
  83
  84 namespace tvg
  85 {
  86
  87 class Executor
  88 {
  89     const unsigned                      _count{std::thread::hardware_concurrency()};
  90     std::vector<std::thread>            _threads;
  91     std::vector<TaskQueue<shared_task>> _q{_count};
  92     std::atomic<unsigned>               _index{0};
  93     void run(unsigned i)
  94     {
  95         // Task Loop
  96         shared_task task;
  97         while (true) {
  98             bool success = false;
  99
 100             for (unsigned n = 0; n != _count * 2; ++n) {
 101                 if (_q[(i + n) % _count].try_pop(task)) {
 102                     success = true;
 103                     break;
 104                 }
 105             }
 106
 107             if (!success && !_q[i].pop(task)) break;
 108
 109             (*task)();
 110         }
 111     }
 112
 113     Executor()
 114     {
 115         for (unsigned n = 0; n != _count; ++n) {
 116             _threads.emplace_back([&, n] { run(n); });
 117         }
 118     }
 119     ~Executor()
 120     {
 121         for (auto &e : _q) e.done();
 122
 123         for (auto &e : _threads) e.join();
 124     }
 125
 126 public:
 127
 128     static Executor& instance() {
 129         static Executor singleton;
 130         return singleton;
 131     }
 132
 133     void post(shared_task task)
 134     {
 135         task->prepare();
 136
 137         auto i = _index++;
 138
 139         for (unsigned n = 0; n != _count; ++n) {
 140             if (_q[(i + n) % _count].try_push(std::move(task))) return;
 141         }
 142
 143         if (_count > 0) {
 144             _q[i % _count].push(std::move(task));
 145         }
 146     }
 147 };
 148
 149 void async(shared_task task)
 150 {
 151     Executor::instance().post(std::move(task));
 152 }
 153
 154 }
 155
 156