Mercurial > hg > orthanc-client

comparison OrthancCppClient/ThreadedCommandProcessor.cpp @ 1:fd402e53d263

Find changesets by keywords (author, files, the commit message), revision number or hash, or revset expression.
new files
author Sebastien Jodogne <s.jodogne@gmail.com>
date Mon, 01 Jun 2015 11:12:20 +0200
parents
children d5027f9f676a
comparison
equal deleted inserted replaced
0:ebc1e38ef615 1:fd402e53d263
1 /**
2 * Orthanc - A Lightweight, RESTful DICOM Store
3 * Copyright (C) 2012-2015 Sebastien Jodogne, Medical Physics
4 * Department, University Hospital of Liege, Belgium
5 *
6 * This program is free software: you can redistribute it and/or
7 * modify it under the terms of the GNU General Public License as
8 * published by the Free Software Foundation, either version 3 of the
9 * License, or (at your option) any later version.
10 *
11 * In addition, as a special exception, the copyright holders of this
12 * program give permission to link the code of its release with the
13 * OpenSSL project's "OpenSSL" library (or with modified versions of it
14 * that use the same license as the "OpenSSL" library), and distribute
15 * the linked executables. You must obey the GNU General Public License
16 * in all respects for all of the code used other than "OpenSSL". If you
17 * modify file(s) with this exception, you may extend this exception to
18 * your version of the file(s), but you are not obligated to do so. If
19 * you do not wish to do so, delete this exception statement from your
20 * version. If you delete this exception statement from all source files
21 * in the program, then also delete it here.
22 *
23 * This program is distributed in the hope that it will be useful, but
24 * WITHOUT ANY WARRANTY; without even the implied warranty of
25 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
26 * General Public License for more details.
27 *
28 * You should have received a copy of the GNU General Public License
29 * along with this program. If not, see <http://www.gnu.org/licenses/>.
30 **/
31
32
33 #include "../PrecompiledHeaders.h"
34 #include "ThreadedCommandProcessor.h"
35
36 #include "../OrthancException.h"
37
38 namespace Orthanc
39 {
40 static const int32_t TIMEOUT = 10;
41
42
43 void ThreadedCommandProcessor::Processor(ThreadedCommandProcessor* that)
44 {
45 while (!that->done_)
46 {
47 std::auto_ptr<IDynamicObject> command(that->queue_.Dequeue(TIMEOUT));
48
49 if (command.get() != NULL)
50 {
51 bool success = false;
52
53 try
54 {
55 if (that->success_)
56 {
57 // No command has failed so far
58
59 if (that->cancel_)
60 {
61 // The commands have been canceled. Skip the execution
62 // of this command, yet mark it as succeeded.
63 success = true;
64 }
65 else
66 {
67 success = dynamic_cast<ICommand&>(*command).Execute();
68 }
69 }
70 else
71 {
72 // A command has already failed. Skip the execution of this command.
73 }
74 }
75 catch (OrthancException)
76 {
77 }
78
79 {
80 boost::mutex::scoped_lock lock(that->mutex_);
81 assert(that->remainingCommands_ > 0);
82 that->remainingCommands_--;
83
84 if (!success)
85 {
86 if (!that->cancel_ && that->listener_ && that->success_)
87 {
88 // This is the first command that fails
89 that->listener_->SignalFailure();
90 }
91
92 that->success_ = false;
93 }
94 else
95 {
96 if (!that->cancel_ && that->listener_)
97 {
98 if (that->remainingCommands_ == 0)
99 {
100 that->listener_->SignalSuccess(that->totalCommands_);
101 }
102 else
103 {
104 that->listener_->SignalProgress(that->totalCommands_ - that->remainingCommands_,
105 that->totalCommands_);
106 }
107 }
108 }
109
110 that->processedCommand_.notify_all();
111 }
112 }
113 }
114 }
115
116
117 ThreadedCommandProcessor::ThreadedCommandProcessor(unsigned int numThreads)
118 {
119 if (numThreads < 1)
120 {
121 throw OrthancException(ErrorCode_ParameterOutOfRange);
122 }
123
124 listener_ = NULL;
125 success_ = true;
126 done_ = false;
127 cancel_ = false;
128 threads_.resize(numThreads);
129 remainingCommands_ = 0;
130 totalCommands_ = 0;
131
132 for (unsigned int i = 0; i < numThreads; i++)
133 {
134 threads_[i] = new boost::thread(Processor, this);
135 }
136 }
137
138
139 ThreadedCommandProcessor::~ThreadedCommandProcessor()
140 {
141 done_ = true;
142
143 for (unsigned int i = 0; i < threads_.size(); i++)
144 {
145 boost::thread* t = threads_[i];
146
147 if (t != NULL)
148 {
149 if (t->joinable())
150 {
151 t->join();
152 }
153
154 delete t;
155 }
156 }
157 }
158
159
160 void ThreadedCommandProcessor::Post(ICommand* command)
161 {
162 if (command == NULL)
163 {
164 throw OrthancException(ErrorCode_ParameterOutOfRange);
165 }
166
167 boost::mutex::scoped_lock lock(mutex_);
168 queue_.Enqueue(command);
169 remainingCommands_++;
170 totalCommands_++;
171 }
172
173
174 bool ThreadedCommandProcessor::Join()
175 {
176 boost::mutex::scoped_lock lock(mutex_);
177
178 while (remainingCommands_ != 0)
179 {
180 processedCommand_.wait(lock);
181 }
182
183 if (cancel_ && listener_)
184 {
185 listener_->SignalCancel();
186 }
187
188 // Reset the sequence counters for subsequent commands
189 bool hasSucceeded = success_;
190 success_ = true;
191 totalCommands_ = 0;
192 cancel_ = false;
193
194 return hasSucceeded;
195 }
196
197
198 void ThreadedCommandProcessor::Cancel()
199 {
200 boost::mutex::scoped_lock lock(mutex_);
201
202 cancel_ = true;
203 }
204
205
206 void ThreadedCommandProcessor::SetListener(IListener& listener)
207 {
208 boost::mutex::scoped_lock lock(mutex_);
209 listener_ = &listener;
210 }
211 }