src/lib/tvgTaskScheduler.cpp

   1 /*
   2  * Copyright (c) 2020-2021 Samsung Electronics Co., Ltd. All rights reserved.
   3
   4  * Permission is hereby granted, free of charge, to any person obtaining a copy
   5  * of this software and associated documentation files (the "Software"), to deal
   6  * in the Software without restriction, including without limitation the rights
   7  * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
   8  * copies of the Software, and to permit persons to whom the Software is
   9  * furnished to do so, subject to the following conditions:
  10
  11  * The above copyright notice and this permission notice shall be included in all
  12  * copies or substantial portions of the Software.
  13
  14  * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
  15  * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
  16  * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
  17  * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
  18  * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
  19  * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
  20  * SOFTWARE.
  21  */
  22 #include <deque>
  23 #include <thread>
  24 #include <vector>
  25 #include <atomic>
  26 #include <condition_variable>
  27 #include "tvgTaskScheduler.h"
  28
  29 /************************************************************************/
  30 /* Internal Class Implementation                                        */
  31 /************************************************************************/
  32
  33 namespace tvg {
  34
  35 struct TaskQueue {
  36     deque<Task*>             taskDeque;
  37     mutex                    mtx;
  38     condition_variable       ready;
  39     bool                     done = false;
  40
  41     bool tryPop(Task** task)
  42     {
  43         unique_lock<mutex> lock{mtx, try_to_lock};
  44         if (!lock || taskDeque.empty()) return false;
  45         *task = taskDeque.front();
  46         taskDeque.pop_front();
  47
  48         return true;
  49     }
  50
  51     bool tryPush(Task* task)
  52     {
  53         {
  54             unique_lock<mutex> lock{mtx, try_to_lock};
  55             if (!lock) return false;
  56             taskDeque.push_back(task);
  57         }
  58
  59         ready.notify_one();
  60
  61         return true;
  62     }
  63
  64     void complete()
  65     {
  66         {
  67             unique_lock<mutex> lock{mtx};
  68             done = true;
  69         }
  70         ready.notify_all();
  71     }
  72
  73     bool pop(Task** task)
  74     {
  75         unique_lock<mutex> lock{mtx};
  76
  77         while (taskDeque.empty() && !done) {
  78             ready.wait(lock);
  79         }
  80
  81         if (taskDeque.empty()) return false;
  82
  83         *task = taskDeque.front();
  84         taskDeque.pop_front();
  85
  86         return true;
  87     }
  88
  89     void push(Task* task)
  90     {
  91         {
  92             unique_lock<mutex> lock{mtx};
  93             taskDeque.push_back(task);
  94         }
  95
  96         ready.notify_one();
  97     }
  98
  99 };
 100
 101
 102 class TaskSchedulerImpl
 103 {
 104 public:
 105     unsigned                       threadCnt;
 106     vector<thread>                 threads;
 107     vector<TaskQueue>              taskQueues;
 108     atomic<unsigned>               idx{0};
 109
 110     TaskSchedulerImpl(unsigned threadCnt) : threadCnt(threadCnt), taskQueues(threadCnt)
 111     {
 112         for (unsigned i = 0; i < threadCnt; ++i) {
 113             threads.emplace_back([&, i] { run(i); });
 114         }
 115     }
 116
 117     ~TaskSchedulerImpl()
 118     {
 119         for (auto& queue : taskQueues) queue.complete();
 120         for (auto& thread : threads) thread.join();
 121     }
 122
 123     void run(unsigned i)
 124     {
 125         Task* task;
 126
 127         //Thread Loop
 128         while (true) {
 129             auto success = false;
 130             for (unsigned x = 0; x < threadCnt * 2; ++x) {
 131                 if (taskQueues[(i + x) % threadCnt].tryPop(&task)) {
 132                     success = true;
 133                     break;
 134                 }
 135             }
 136
 137             if (!success && !taskQueues[i].pop(&task)) break;
 138             (*task)(i);
 139         }
 140     }
 141
 142     void request(Task* task)
 143     {
 144         //Async
 145         if (threadCnt > 0) {
 146             task->prepare();
 147             auto i = idx++;
 148             for (unsigned n = 0; n < threadCnt; ++n) {
 149                 if (taskQueues[(i + n) % threadCnt].tryPush(task)) return;
 150             }
 151             taskQueues[i % threadCnt].push(task);
 152         //Sync
 153         } else {
 154             task->run(0);
 155         }
 156     }
 157 };
 158
 159 }
 160
 161 static TaskSchedulerImpl* inst = nullptr;
 162
 163 /************************************************************************/
 164 /* External Class Implementation                                        */
 165 /************************************************************************/
 166
 167 void TaskScheduler::init(unsigned threads)
 168 {
 169     if (inst) return;
 170     inst = new TaskSchedulerImpl(threads);
 171 }
 172
 173
 174 void TaskScheduler::term()
 175 {
 176     if (!inst) return;
 177     delete(inst);
 178     inst = nullptr;
 179 }
 180
 181
 182 void TaskScheduler::request(Task* task)
 183 {
 184     if (inst) inst->request(task);
 185 }
 186
 187
 188 unsigned TaskScheduler::threads()
 189 {
 190     if (inst) return inst->threadCnt;
 191     return 0;
 192 }