1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
|
//
// Copyright © 2021 Arm Ltd and Contributors. All rights reserved.
// SPDX-License-Identifier: MIT
//
#include <armnn/Threadpool.hpp>
#include <armnn/utility/Timer.hpp>
namespace armnn
{
namespace experimental
{
Threadpool::Threadpool(std::size_t numThreads,
IRuntime* runtimePtr,
std::vector<std::shared_ptr<IWorkingMemHandle>> memHandles)
: m_RuntimePtr(runtimePtr)
{
for (auto i = 0u; i < numThreads; ++i)
{
m_Threads.emplace_back(std::make_unique<std::thread>(&Threadpool::ProcessExecPriorities, this, i));
}
LoadMemHandles(memHandles);
}
void Threadpool::LoadMemHandles(std::vector<std::shared_ptr<IWorkingMemHandle>> memHandles)
{
if (memHandles.size() == 0)
{
throw armnn::RuntimeException("Threadpool::UnloadMemHandles: Size of memHandles vector must be greater than 0");
}
if (memHandles.size() != m_Threads.size())
{
throw armnn::RuntimeException(
"Threadpool::UnloadMemHandles: Size of memHandles vector must match the number of threads");
}
NetworkId networkId = memHandles[0]->GetNetworkId();
for (uint32_t i = 1; i < memHandles.size(); ++i)
{
if (networkId != memHandles[i]->GetNetworkId())
{
throw armnn::RuntimeException(
"Threadpool::UnloadMemHandles: All network ids must be identical in memHandles");
}
}
std::pair<NetworkId, std::vector<std::shared_ptr<IWorkingMemHandle>>> pair {networkId, memHandles};
m_WorkingMemHandleMap.insert(pair);
}
void Threadpool::UnloadMemHandles(NetworkId networkId)
{
if (m_WorkingMemHandleMap.find(networkId) != m_WorkingMemHandleMap.end())
{
m_WorkingMemHandleMap.erase(networkId);
}
else
{
throw armnn::RuntimeException("Threadpool::UnloadMemHandles: Unknown NetworkId");
}
}
void Threadpool::Schedule(NetworkId networkId,
const InputTensors& inputTensors,
const OutputTensors& outputTensors,
const QosExecPriority priority,
std::shared_ptr<IAsyncExecutionCallback> cb)
{
if (m_WorkingMemHandleMap.find(networkId) == m_WorkingMemHandleMap.end())
{
throw armnn::RuntimeException("Threadpool::UnloadMemHandles: Unknown NetworkId");
}
// Group execution parameters so that they can be easily added to the queue
ExecutionTuple groupExecParams = std::make_tuple(networkId, inputTensors, outputTensors, cb);
std::shared_ptr<ExecutionTuple> operation = std::make_shared<ExecutionTuple>(groupExecParams);
// Add a message to the queue and notify the request thread
std::unique_lock<std::mutex> lock(m_ThreadPoolMutex);
switch (priority)
{
case QosExecPriority::High:
m_HighPriorityQueue.push(operation);
break;
case QosExecPriority::Low:
m_LowPriorityQueue.push(operation);
break;
case QosExecPriority::Medium:
default:
m_MediumPriorityQueue.push(operation);
}
m_ThreadPoolEvent.notify_one();
}
void Threadpool::TerminateThreadPool() noexcept
{
{
std::unique_lock<std::mutex> threadPoolLock(m_ThreadPoolMutex);
m_TerminatePool = true;
}
m_ThreadPoolEvent.notify_all();
for (auto &thread : m_Threads)
{
thread->join();
}
}
void Threadpool::ProcessExecPriorities(uint32_t index)
{
int expireRate = EXPIRE_RATE;
int highPriorityCount = 0;
int mediumPriorityCount = 0;
while (true)
{
std::shared_ptr<ExecutionTuple> currentExecInProgress(nullptr);
{
// Wait for a message to be added to the queue
// This is in a separate scope to minimise the lifetime of the lock
std::unique_lock<std::mutex> lock(m_ThreadPoolMutex);
m_ThreadPoolEvent.wait(lock,
[=]
{
return m_TerminatePool || !m_HighPriorityQueue.empty() ||
!m_MediumPriorityQueue.empty() || !m_LowPriorityQueue.empty();
});
if (m_TerminatePool && m_HighPriorityQueue.empty() && m_MediumPriorityQueue.empty() &&
m_LowPriorityQueue.empty())
{
break;
}
// Get the message to process from the front of each queue based on priority from high to low
// Get high priority first if it does not exceed the expire rate
if (!m_HighPriorityQueue.empty() && highPriorityCount < expireRate)
{
currentExecInProgress = m_HighPriorityQueue.front();
m_HighPriorityQueue.pop();
highPriorityCount += 1;
}
// If high priority queue is empty or the count exceeds the expire rate, get medium priority message
else if (!m_MediumPriorityQueue.empty() && mediumPriorityCount < expireRate)
{
currentExecInProgress = m_MediumPriorityQueue.front();
m_MediumPriorityQueue.pop();
mediumPriorityCount += 1;
// Reset high priority count
highPriorityCount = 0;
}
// If medium priority queue is empty or the count exceeds the expire rate, get low priority message
else if (!m_LowPriorityQueue.empty())
{
currentExecInProgress = m_LowPriorityQueue.front();
m_LowPriorityQueue.pop();
// Reset high and medium priority count
highPriorityCount = 0;
mediumPriorityCount = 0;
}
else
{
// Reset high and medium priority count
highPriorityCount = 0;
mediumPriorityCount = 0;
continue;
}
}
// invoke the asynchronous execution method
auto networkId = std::get<0>(*currentExecInProgress);
auto inputTensors = std::get<1>(*currentExecInProgress);
auto outputTensors = std::get<2>(*currentExecInProgress);
auto cb = std::get<3>(*currentExecInProgress);
// Get time at start of inference
HighResolutionClock startTime = armnn::GetTimeNow();
try // executing the inference
{
IWorkingMemHandle& memHandle = *(m_WorkingMemHandleMap.at(networkId))[index];
// Execute and populate the time at end of inference in the callback
m_RuntimePtr->Execute(memHandle, inputTensors, outputTensors) == Status::Success ?
cb->Notify(Status::Success, std::make_pair(startTime, armnn::GetTimeNow())) :
cb->Notify(Status::Failure, std::make_pair(startTime, armnn::GetTimeNow()));
}
catch (const RuntimeException&)
{
cb->Notify(Status::Failure, std::make_pair(startTime, armnn::GetTimeNow()));
}
}
}
} // namespace experimental
} // namespace armnn
|
