forked from ossimlabs/ossim
-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathossimJobMultiThreadQueue.cpp
More file actions
126 lines (113 loc) · 3.2 KB
/
Copy pathossimJobMultiThreadQueue.cpp
File metadata and controls
126 lines (113 loc) · 3.2 KB
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
#include <ossim/parallel/ossimJobMultiThreadQueue.h>
ossimJobMultiThreadQueue::ossimJobMultiThreadQueue(std::shared_ptr<ossimJobQueue> q,
ossim_uint32 nThreads)
:m_jobQueue(q?q:std::make_shared<ossimJobQueue>())
{
setNumberOfThreads(nThreads);
}
ossimJobMultiThreadQueue::~ossimJobMultiThreadQueue()
{
cancel();
waitForCompletion();
m_threadQueueList.clear();
}
std::shared_ptr<ossimJobQueue> ossimJobMultiThreadQueue::getJobQueue()
{
std::lock_guard<std::mutex> lock(m_mutex);
return m_jobQueue;
}
const std::shared_ptr<ossimJobQueue> ossimJobMultiThreadQueue::getJobQueue()const
{
std::lock_guard<std::mutex> lock(m_mutex);
return m_jobQueue;
}
void ossimJobMultiThreadQueue::setJobQueue(std::shared_ptr<ossimJobQueue> q)
{
std::lock_guard<std::mutex> lock(m_mutex);
ossim_uint32 idx = 0;
m_jobQueue = q;
for(idx = 0; idx < m_threadQueueList.size(); ++idx)
{
m_threadQueueList[idx]->setJobQueue(m_jobQueue);
}
}
void ossimJobMultiThreadQueue::setNumberOfThreads(ossim_uint32 nThreads)
{
std::lock_guard<std::mutex> lock(m_mutex);
ossim_uint32 idx = 0;
ossim_uint32 queueSize = m_threadQueueList.size();
if(nThreads > queueSize)
{
for(idx = queueSize; idx < nThreads;++idx)
{
std::shared_ptr<ossimJobThreadQueue> threadQueue = std::make_shared<ossimJobThreadQueue>();
threadQueue->setJobQueue(m_jobQueue);
m_threadQueueList.push_back(threadQueue);
}
}
else if(nThreads < queueSize)
{
ThreadQueueList::iterator iter = m_threadQueueList.begin()+nThreads;
while(iter != m_threadQueueList.end())
{
(*iter)->cancel();
iter = m_threadQueueList.erase(iter);
}
}
}
ossim_uint32 ossimJobMultiThreadQueue::getNumberOfThreads() const
{
std::lock_guard<std::mutex> lock(m_mutex);
return static_cast<ossim_uint32>( m_threadQueueList.size() );
}
ossim_uint32 ossimJobMultiThreadQueue::numberOfBusyThreads()const
{
ossim_uint32 result = 0;
std::lock_guard<std::mutex> lock(m_mutex);
ossim_uint32 idx = 0;
ossim_uint32 queueSize = m_threadQueueList.size();
for(idx = 0; idx < queueSize;++idx)
{
if(m_threadQueueList[idx]->isProcessingJob()) ++result;
}
return result;
}
bool ossimJobMultiThreadQueue::areAllThreadsBusy()const
{
std::lock_guard<std::mutex> lock(m_mutex);
ossim_uint32 idx = 0;
ossim_uint32 queueSize = m_threadQueueList.size();
for(idx = 0; idx < queueSize;++idx)
{
if(!m_threadQueueList[idx]->isProcessingJob()) return false;
}
return true;
}
bool ossimJobMultiThreadQueue::hasJobsToProcess()const
{
bool result = false;
{
std::lock_guard<std::mutex> lock(m_mutex);
ossim_uint32 queueSize = m_threadQueueList.size();
ossim_uint32 idx = 0;
for(idx = 0; ((idx<queueSize)&&!result);++idx)
{
result = m_threadQueueList[idx]->hasJobsToProcess();
}
}
return result;
}
void ossimJobMultiThreadQueue::cancel()
{
for(auto thread:m_threadQueueList)
{
thread->cancel();
}
}
void ossimJobMultiThreadQueue::waitForCompletion()
{
for(auto thread:m_threadQueueList)
{
thread->waitForCompletion();
}
}