GCC Code Coverage Report

Directory: avs_core/
Coverage: low: ≥ 0% medium: ≥ 75.0% high: ≥ 90.0%
Coverage Exec / Excl / Total
Lines: 60.5% 46 / 0 / 76
Functions: 75.0% 6 / 0 / 8
Branches: 33.0% 30 / 0 / 91
core/ThreadPool.cpp
Line Branch Exec Source
1 #include "ThreadPool.h"
2 #include "internal.h"
3 #include <cassert>
4 #include <thread>
5
6 struct ThreadPoolGenericItemData
7 {
8 ThreadWorkerFuncPtr Func;
9 void* Params;
10 AVSPromise* Promise;
11 Device* device;
12 };
13
14 #include "mpmc_bounded_queue.h"
15 typedef mpmc_bounded_queue<ThreadPoolGenericItemData> MessageQueue;
16
17 class ThreadPoolPimpl
18 {
19 public:
20 std::vector<std::thread> Threads;
21 MessageQueue MsgQueue;
22 std::mutex Mutex;
23 std::condition_variable FinishCond;
24 size_t NumRunning;
25
26 453 ThreadPoolPimpl(size_t nThreads) :
27 453 Threads(),
28
1/2
✓ Branch 3 → 4 taken 453 times.
✗ Branch 3 → 7 not taken.
 453 MsgQueue(nThreads * 6)
29 453 {}
30 };
31
32 1812 void ThreadPool::ThreadFunc(size_t thread_id, ThreadPoolPimpl * const _pimpl, InternalEnvironment* env)
33 {
34
1/2
✓ Branch 2 → 3 taken 1811 times.
✗ Branch 2 → 66 not taken.
 1812 auto EnvTLS = env->NewThreadScriptEnvironment((int)thread_id);
35 1811 PInternalEnvironment holder = PInternalEnvironment(EnvTLS);
36
37 while (true)
38 {
39 ThreadPoolGenericItemData data;
40
2/4
✓ Branch 4 → 5 taken 1812 times.
✗ Branch 4 → 63 not taken.
✓ Branch 5 → 6 taken 1812 times.
✗ Branch 5 → 12 not taken.
 1810 if (_pimpl->MsgQueue.pop_back(&data) == false) {
41 // threadpool is canceled
42
1/2
✓ Branch 6 → 7 taken 1812 times.
✗ Branch 6 → 24 not taken.
 1812 std::unique_lock<std::mutex> lock(_pimpl->Mutex);
43
2/2
✓ Branch 7 → 8 taken 453 times.
✓ Branch 7 → 9 taken 1359 times.
 1812 if (--_pimpl->NumRunning == 0) {
44 453 _pimpl->FinishCond.notify_all();
45 }
46 1762 return;
47 1812 }
48
49 ✗ EnvTLS->SetCurrentDevice(data.device);
50 ✗ EnvTLS->GetSupressCaching() = false;
51 ✗ if (data.Promise != NULL)
52 {
53 try
54 {
55 ✗ data.Promise->set_value(data.Func(EnvTLS, data.Params));
56 }
57 ✗ catch (const AvisynthError&)
58 {
59 ✗ data.Promise->set_exception(std::current_exception());
60 ✗ }
61 ✗ catch (const std::exception&)
62 {
63 ✗ data.Promise->set_exception(std::current_exception());
64 ✗ }
65 ✗ catch (...)
66 {
67 ✗ data.Promise->set_exception(std::current_exception());
68 //data.Promise->set_value(AVSValue("An unknown exception was thrown in the thread pool."));
69 ✗ }
70 }
71 else
72 {
73 try
74 {
75 ✗ data.Func(EnvTLS, data.Params);
76 }
77 ✗ catch (...) {}
78 }
79 ✗ } //while
80 1812 }
81
82 453 ThreadPool::ThreadPool(size_t nThreads, size_t nStartId, InternalEnvironment* env) :
83
2/6
✓ Branch 3 → 4 taken 453 times.
✗ Branch 3 → 14 not taken.
✗ Branch 4 → 5 not taken.
✓ Branch 4 → 6 taken 453 times.
✗ Branch 14 → 15 not taken.
✗ Branch 14 → 16 not taken.
 453 _pimpl(new ThreadPoolPimpl(nThreads))
84 {
85
1/2
✓ Branch 6 → 7 taken 453 times.
✗ Branch 6 → 20 not taken.
 453 _pimpl->Threads.reserve(nThreads);
86
87
1/2
✓ Branch 7 → 8 taken 453 times.
✗ Branch 7 → 20 not taken.
 453 std::unique_lock<std::mutex> lock(_pimpl->Mutex);
88
89 // i is used as the thread id. Skip id zero because that is reserved for the main thread.
90 // CUDA: thread id is controled by caller
91
2/2
✓ Branch 11 → 9 taken 1812 times.
✓ Branch 11 → 12 taken 453 times.
 2265 for (size_t i = 0; i < nThreads; ++i)
92
1/2
✓ Branch 9 → 10 taken 1812 times.
✗ Branch 9 → 17 not taken.
 1812 _pimpl->Threads.emplace_back(ThreadFunc, i + nStartId, _pimpl, env);
93
94 453 _pimpl->NumRunning = nThreads;
95 453 }
96
97 ✗ void ThreadPool::QueueJob(ThreadWorkerFuncPtr clb, void* params, InternalEnvironment* env, JobCompletion* tc)
98 {
99 ThreadPoolGenericItemData itemData;
100 ✗ itemData.Func = clb;
101 ✗ itemData.Params = params;
102 ✗ itemData.device = env->GetCurrentDevice();
103
104 ✗ if (tc != NULL)
105 ✗ itemData.Promise = tc->Add();
106 else
107 ✗ itemData.Promise = NULL;
108
109 ✗ if (_pimpl->MsgQueue.push_front(std::move(itemData)) == false) {
110 ✗ throw AvisynthError("Threadpool is cancelled");
111 }
112 ✗ }
113
114 ✗ size_t ThreadPool::NumThreads() const
115 {
116 ✗ return _pimpl->Threads.size();
117 }
118
119 906 std::vector<void*> ThreadPool::Finish()
120 {
121
1/2
✓ Branch 2 → 3 taken 906 times.
✗ Branch 2 → 29 not taken.
 906 std::unique_lock<std::mutex> lock(_pimpl->Mutex);
122
2/2
✓ Branch 3 → 4 taken 453 times.
✓ Branch 3 → 18 taken 453 times.
 906 if (_pimpl->NumRunning > 0) {
123
1/2
✓ Branch 4 → 5 taken 453 times.
✗ Branch 4 → 26 not taken.
 453 _pimpl->MsgQueue.finish();
124
2/2
✓ Branch 8 → 6 taken 453 times.
✓ Branch 8 → 9 taken 453 times.
 906 while (_pimpl->NumRunning > 0)
125 {
126
1/2
✓ Branch 6 → 7 taken 453 times.
✗ Branch 6 → 26 not taken.
 453 _pimpl->FinishCond.wait(lock);
127 }
128
0/2
✗ Branch 23 → 24 not taken.
✗ Branch 23 → 25 not taken.
 453 std::vector<void*> ret;
129 ThreadPoolGenericItemData item;
130
2/4
✓ Branch 13 → 14 taken 453 times.
✗ Branch 13 → 23 not taken.
✗ Branch 14 → 11 not taken.
✓ Branch 14 → 15 taken 453 times.
 453 while (_pimpl->MsgQueue.pop_remain(&item)) {
131 ✗ ret.push_back(item.Params);
132 }
133
1/2
✗ Branch 15 → 16 not taken.
✓ Branch 15 → 17 taken 453 times.
 453 return ret;
134 ✗ }
135 453 return std::vector<void*>();
136 906 }
137
138 453 void ThreadPool::Join()
139 {
140
1/2
✓ Branch 3 → 4 taken 453 times.
✗ Branch 3 → 16 not taken.
 453 if (_pimpl->Threads.size() > 0) {
141
1/2
✓ Branch 4 → 5 taken 453 times.
✗ Branch 4 → 17 not taken.
 453 Finish();
142
2/2
✓ Branch 14 → 7 taken 1812 times.
✓ Branch 14 → 15 taken 453 times.
 2265 for (size_t i = 0; i < _pimpl->Threads.size(); ++i)
143 {
144
1/2
✓ Branch 9 → 10 taken 1812 times.
✗ Branch 9 → 12 not taken.
 1812 if (_pimpl->Threads[i].joinable())
145 1812 _pimpl->Threads[i].join();
146 }
147 453 _pimpl->Threads.clear();
148 }
149 453 }
150
151 453 ThreadPool::~ThreadPool()
152 {
153 453 Finish();
154 453 Join();
155
1/2
✓ Branch 5 → 6 taken 453 times.
✗ Branch 5 → 8 not taken.
 453 delete _pimpl;
156 453 }
157