Mercurial > hg > orthanc
comparison Core/JobsEngine/Operations/SequenceOfOperationsJob.cpp @ 2603:988936118354 jobs
reorganization
author | Sebastien Jodogne <s.jodogne@gmail.com> |
---|---|
date | Fri, 18 May 2018 17:02:25 +0200 |
parents | |
children | 25225f0b4f33 |
comparison
equal
deleted
inserted
replaced
2602:c25f1a52acbc | 2603:988936118354 |
---|---|
1 /** | |
2 * Orthanc - A Lightweight, RESTful DICOM Store | |
3 * Copyright (C) 2012-2016 Sebastien Jodogne, Medical Physics | |
4 * Department, University Hospital of Liege, Belgium | |
5 * Copyright (C) 2017-2018 Osimis S.A., Belgium | |
6 * | |
7 * This program is free software: you can redistribute it and/or | |
8 * modify it under the terms of the GNU General Public License as | |
9 * published by the Free Software Foundation, either version 3 of the | |
10 * License, or (at your option) any later version. | |
11 * | |
12 * In addition, as a special exception, the copyright holders of this | |
13 * program give permission to link the code of its release with the | |
14 * OpenSSL project's "OpenSSL" library (or with modified versions of it | |
15 * that use the same license as the "OpenSSL" library), and distribute | |
16 * the linked executables. You must obey the GNU General Public License | |
17 * in all respects for all of the code used other than "OpenSSL". If you | |
18 * modify file(s) with this exception, you may extend this exception to | |
19 * your version of the file(s), but you are not obligated to do so. If | |
20 * you do not wish to do so, delete this exception statement from your | |
21 * version. If you delete this exception statement from all source files | |
22 * in the program, then also delete it here. | |
23 * | |
24 * This program is distributed in the hope that it will be useful, but | |
25 * WITHOUT ANY WARRANTY; without even the implied warranty of | |
26 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU | |
27 * General Public License for more details. | |
28 * | |
29 * You should have received a copy of the GNU General Public License | |
30 * along with this program. If not, see <http://www.gnu.org/licenses/>. | |
31 **/ | |
32 | |
33 | |
34 #include "../../PrecompiledHeaders.h" | |
35 #include "SequenceOfOperationsJob.h" | |
36 | |
37 #include "../../Logging.h" | |
38 #include "../../OrthancException.h" | |
39 | |
40 namespace Orthanc | |
41 { | |
42 class SequenceOfOperationsJob::Operation : public boost::noncopyable | |
43 { | |
44 private: | |
45 JobOperationValues originalInputs_; | |
46 JobOperationValues workInputs_; | |
47 std::auto_ptr<IJobOperation> operation_; | |
48 std::list<Operation*> nextOperations_; | |
49 size_t currentInput_; | |
50 | |
51 public: | |
52 Operation(IJobOperation* operation) : | |
53 operation_(operation), | |
54 currentInput_(0) | |
55 { | |
56 if (operation == NULL) | |
57 { | |
58 throw OrthancException(ErrorCode_NullPointer); | |
59 } | |
60 } | |
61 | |
62 void AddOriginalInput(const JobOperationValue& value) | |
63 { | |
64 if (currentInput_ != 0) | |
65 { | |
66 // Cannot add input after processing has started | |
67 throw OrthancException(ErrorCode_BadSequenceOfCalls); | |
68 } | |
69 else | |
70 { | |
71 originalInputs_.Append(value.Clone()); | |
72 } | |
73 } | |
74 | |
75 const JobOperationValues& GetOriginalInputs() const | |
76 { | |
77 return originalInputs_; | |
78 } | |
79 | |
80 void Reset() | |
81 { | |
82 workInputs_.Clear(); | |
83 currentInput_ = 0; | |
84 } | |
85 | |
86 void AddNextOperation(Operation& other) | |
87 { | |
88 if (currentInput_ != 0) | |
89 { | |
90 // Cannot add input after processing has started | |
91 throw OrthancException(ErrorCode_BadSequenceOfCalls); | |
92 } | |
93 else | |
94 { | |
95 nextOperations_.push_back(&other); | |
96 } | |
97 } | |
98 | |
99 bool IsDone() const | |
100 { | |
101 return currentInput_ >= originalInputs_.GetSize() + workInputs_.GetSize(); | |
102 } | |
103 | |
104 void Step() | |
105 { | |
106 if (IsDone()) | |
107 { | |
108 throw OrthancException(ErrorCode_BadSequenceOfCalls); | |
109 } | |
110 | |
111 const JobOperationValue* input; | |
112 | |
113 if (currentInput_ < originalInputs_.GetSize()) | |
114 { | |
115 input = &originalInputs_.GetValue(currentInput_); | |
116 } | |
117 else | |
118 { | |
119 input = &workInputs_.GetValue(currentInput_ - originalInputs_.GetSize()); | |
120 } | |
121 | |
122 JobOperationValues outputs; | |
123 operation_->Apply(outputs, *input); | |
124 | |
125 if (!nextOperations_.empty()) | |
126 { | |
127 std::list<Operation*>::iterator first = nextOperations_.begin(); | |
128 outputs.Move((*first)->workInputs_); | |
129 | |
130 std::list<Operation*>::iterator current = first; | |
131 ++current; | |
132 | |
133 while (current != nextOperations_.end()) | |
134 { | |
135 (*first)->workInputs_.Copy((*current)->workInputs_); | |
136 ++current; | |
137 } | |
138 } | |
139 | |
140 currentInput_ += 1; | |
141 } | |
142 }; | |
143 | |
144 | |
145 // Invoked from constructors | |
146 void SequenceOfOperationsJob::Setup() | |
147 { | |
148 done_ = false; | |
149 current_ = 0; | |
150 trailingTimeout_ = boost::posix_time::milliseconds(1000); | |
151 } | |
152 | |
153 | |
154 SequenceOfOperationsJob::~SequenceOfOperationsJob() | |
155 { | |
156 for (size_t i = 0; i < operations_.size(); i++) | |
157 { | |
158 if (operations_[i] != NULL) | |
159 { | |
160 delete operations_[i]; | |
161 } | |
162 } | |
163 } | |
164 | |
165 | |
166 void SequenceOfOperationsJob::Register(IObserver& observer) | |
167 { | |
168 boost::mutex::scoped_lock lock(mutex_); | |
169 observers_.push_back(&observer); | |
170 } | |
171 | |
172 | |
173 void SequenceOfOperationsJob::Lock::SetTrailingOperationTimeout(unsigned int timeout) | |
174 { | |
175 that_.trailingTimeout_ = boost::posix_time::milliseconds(timeout); | |
176 } | |
177 | |
178 | |
179 size_t SequenceOfOperationsJob::Lock::AddOperation(IJobOperation* operation) | |
180 { | |
181 if (IsDone()) | |
182 { | |
183 throw OrthancException(ErrorCode_BadSequenceOfCalls); | |
184 } | |
185 | |
186 that_.operations_.push_back(new Operation(operation)); | |
187 that_.operationAdded_.notify_one(); | |
188 | |
189 return that_.operations_.size() - 1; | |
190 } | |
191 | |
192 | |
193 void SequenceOfOperationsJob::Lock::AddInput(size_t index, | |
194 const JobOperationValue& value) | |
195 { | |
196 if (IsDone()) | |
197 { | |
198 throw OrthancException(ErrorCode_BadSequenceOfCalls); | |
199 } | |
200 else if (index >= that_.operations_.size() || | |
201 index < that_.current_) | |
202 { | |
203 throw OrthancException(ErrorCode_ParameterOutOfRange); | |
204 } | |
205 else | |
206 { | |
207 that_.operations_[index]->AddOriginalInput(value); | |
208 } | |
209 } | |
210 | |
211 | |
212 void SequenceOfOperationsJob::Lock::Connect(size_t input, | |
213 size_t output) | |
214 { | |
215 if (IsDone()) | |
216 { | |
217 throw OrthancException(ErrorCode_BadSequenceOfCalls); | |
218 } | |
219 else if (input >= output || | |
220 input >= that_.operations_.size() || | |
221 output >= that_.operations_.size() || | |
222 input < that_.current_ || | |
223 output < that_.current_) | |
224 { | |
225 throw OrthancException(ErrorCode_ParameterOutOfRange); | |
226 } | |
227 else | |
228 { | |
229 Operation& a = *that_.operations_[input]; | |
230 Operation& b = *that_.operations_[output]; | |
231 a.AddNextOperation(b); | |
232 } | |
233 } | |
234 | |
235 | |
236 JobStepResult SequenceOfOperationsJob::ExecuteStep() | |
237 { | |
238 boost::mutex::scoped_lock lock(mutex_); | |
239 | |
240 if (current_ == operations_.size()) | |
241 { | |
242 LOG(INFO) << "Executing the trailing timeout in the sequence of operations"; | |
243 operationAdded_.timed_wait(lock, trailingTimeout_); | |
244 | |
245 if (current_ == operations_.size()) | |
246 { | |
247 // No operation was added during the trailing timeout: The | |
248 // job is over | |
249 LOG(INFO) << "The sequence of operations is over"; | |
250 done_ = true; | |
251 | |
252 for (std::list<IObserver*>::iterator it = observers_.begin(); | |
253 it != observers_.end(); ++it) | |
254 { | |
255 (*it)->SignalDone(*this); | |
256 } | |
257 | |
258 return JobStepResult::Success(); | |
259 } | |
260 else | |
261 { | |
262 LOG(INFO) << "New operation added to the sequence of operations"; | |
263 } | |
264 } | |
265 | |
266 assert(current_ < operations_.size()); | |
267 | |
268 while (current_ < operations_.size() && | |
269 operations_[current_]->IsDone()) | |
270 { | |
271 current_++; | |
272 } | |
273 | |
274 if (current_ < operations_.size()) | |
275 { | |
276 operations_[current_]->Step(); | |
277 } | |
278 | |
279 return JobStepResult::Continue(); | |
280 } | |
281 | |
282 | |
283 void SequenceOfOperationsJob::SignalResubmit() | |
284 { | |
285 boost::mutex::scoped_lock lock(mutex_); | |
286 | |
287 current_ = 0; | |
288 done_ = false; | |
289 | |
290 for (size_t i = 0; i < operations_.size(); i++) | |
291 { | |
292 operations_[i]->Reset(); | |
293 } | |
294 } | |
295 | |
296 | |
297 float SequenceOfOperationsJob::GetProgress() | |
298 { | |
299 boost::mutex::scoped_lock lock(mutex_); | |
300 | |
301 return (static_cast<float>(current_) / | |
302 static_cast<float>(operations_.size() + 1)); | |
303 } | |
304 | |
305 | |
306 void SequenceOfOperationsJob::GetPublicContent(Json::Value& value) | |
307 { | |
308 boost::mutex::scoped_lock lock(mutex_); | |
309 | |
310 value["CountOperations"] = static_cast<unsigned int>(operations_.size()); | |
311 } | |
312 } |