update cin.po, goog xlat update xx.po
[goodguy/history.git] / cinelerra-5.1 / cinelerra / loadbalance.C
1
2 /*
3  * CINELERRA
4  * Copyright (C) 2008 Adam Williams <broadcast at earthling dot net>
5  * 
6  * This program is free software; you can redistribute it and/or modify
7  * it under the terms of the GNU General Public License as published by
8  * the Free Software Foundation; either version 2 of the License, or
9  * (at your option) any later version.
10  * 
11  * This program is distributed in the hope that it will be useful,
12  * but WITHOUT ANY WARRANTY; without even the implied warranty of
13  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
14  * GNU General Public License for more details.
15  * 
16  * You should have received a copy of the GNU General Public License
17  * along with this program; if not, write to the Free Software
18  * Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA
19  * 
20  */
21
22 #include "condition.h"
23 #include "mutex.h"
24 #include "loadbalance.h"
25
26
27
28
29 LoadPackage::LoadPackage()
30 {
31         completion_lock = new Condition(0, "LoadPackage::completion_lock");
32 }
33 LoadPackage::~LoadPackage()
34 {
35         delete completion_lock;
36 }
37
38
39
40
41
42
43
44
45
46 LoadClient::LoadClient(LoadServer *server)
47  : Thread(1, 0, 0)
48 {
49         Thread::set_synchronous(1);
50         this->server = server;
51         done = 0;
52         package_number = 0;
53         input_lock = new Condition(0, "LoadClient::input_lock");
54         completion_lock = new Condition(0, "LoadClient::completion_lock");
55 }
56
57 LoadClient::LoadClient()
58  : Thread(1, 0, 0)
59 {
60         Thread::set_synchronous(1);
61         server = 0;
62         done = 0;
63         package_number = 0;
64         input_lock = new Condition(0, "LoadClient::input_lock");
65         completion_lock = new Condition(0, "LoadClient::completion_lock");
66 }
67
68 LoadClient::~LoadClient()
69 {
70         done = 1;
71         input_lock->unlock();
72         Thread::join();
73         delete input_lock;
74         delete completion_lock;
75 }
76
77 int LoadClient::get_package_number()
78 {
79         return package_number;
80 }
81
82 LoadServer* LoadClient::get_server()
83 {
84         return server;
85 }
86
87
88 void LoadClient::run()
89 {
90         while(!done)
91         {
92                 input_lock->lock("LoadClient::run");
93
94                 if(!done)
95                 {
96 // Read packet
97                         LoadPackage *package;
98                         
99                         
100                         server->client_lock->lock("LoadClient::run");
101                         if(server->current_package < server->total_packages)
102                         {
103                                 package_number = server->current_package;
104                                 package = server->packages[server->current_package++];
105                                 server->client_lock->unlock();
106                                 input_lock->unlock();
107
108                                 process_package(package);
109
110                                 package->completion_lock->unlock();
111                         }
112                         else
113                         {
114                                 server->client_lock->unlock();
115                                 completion_lock->unlock();
116                         }
117                 }
118         }
119 }
120
121 void LoadClient::run_single()
122 {
123         if(server->total_packages)
124         {
125                 for(int i = 0; i < server->total_packages; i++)
126                 {
127                         process_package(server->packages[i]);
128                 }
129         }
130 }
131
132 void LoadClient::process_package(LoadPackage *package)
133 {
134         printf("LoadClient::process_package\n");
135 }
136
137
138
139
140
141 LoadServer::LoadServer(int total_clients, int total_packages)
142 {
143         if(total_clients <= 0)
144                 printf("LoadServer::LoadServer total_clients == %d\n", total_clients);
145         this->total_clients = total_clients;
146         this->total_packages = total_packages;
147         current_package = 0;
148         clients = 0;
149         packages = 0;
150         client_lock = new Mutex("LoadServer::client_lock");
151         is_single = 0;
152         single_client = 0;
153 }
154
155 LoadServer::~LoadServer()
156 {
157         delete_clients();
158         delete_packages();
159         delete client_lock;
160 }
161
162 void LoadServer::delete_clients()
163 {
164         if(clients)
165         {
166                 for(int i = 0; i < total_clients; i++)
167                         delete clients[i];
168                 delete [] clients;
169         }
170
171         if(single_client) delete single_client;
172
173         clients = 0;
174         single_client = 0;
175 }
176
177 void LoadServer::delete_packages()
178 {
179         if(packages)
180         {
181                 for(int i = 0; i < total_packages; i++)
182                         delete packages[i];
183                 delete [] packages;
184         }
185         packages = 0;
186 }
187
188 void LoadServer::set_package_count(int total_packages)
189 {
190         delete_packages();
191         this->total_packages = total_packages;
192         create_packages();
193 }
194
195
196 void LoadServer::create_clients()
197 {
198         if(!is_single && !clients)
199         {
200                 clients = new LoadClient*[total_clients];
201                 for(int i = 0; i < total_clients; i++)
202                 {
203                         clients[i] = new_client();
204                         clients[i]->server = this;
205                         clients[i]->start();
206                 }
207         }
208
209         if(is_single && !single_client)
210         {
211                 single_client = new_client();
212                 single_client->server = this;
213         }
214 }
215
216 void LoadServer::create_packages()
217 {
218         if(!packages)
219         {
220                 packages = new LoadPackage*[total_packages];
221                 for(int i = 0; i < total_packages; i++)
222                         packages[i] = new_package();
223         }
224 }
225
226 LoadPackage* LoadServer::get_package(int number)
227 {
228         return packages[number];
229 }
230
231 LoadClient* LoadServer::get_client(int number)
232 {
233         return clients[number];
234 }
235
236 int LoadServer::get_total_packages()
237 {
238 //      if(is_single) return 1;
239         return total_packages;
240 }
241
242 int LoadServer::get_total_clients()
243 {
244         if(is_single) return 1;
245         return total_clients;
246 }
247
248 void LoadServer::process_packages()
249 {
250         if(total_clients == 1)
251         {
252                 process_single();
253                 return;
254         }
255
256         is_single = 0;
257         create_clients();
258         create_packages();
259         
260         
261         
262 // Set up packages
263         init_packages();
264
265         current_package = 0;
266 // Start all clients
267         for(int i = 0; i < total_clients; i++)
268         {
269                 clients[i]->input_lock->unlock();
270         }
271         
272 // Wait for packages to get finished
273         for(int i = 0; i < total_packages; i++)
274         {
275                 packages[i]->completion_lock->lock("LoadServer::process_packages 1");
276         }
277
278 // Wait for clients to finish before allowing changes to packages
279         for(int i = 0; i < total_clients; i++)
280         {
281                 clients[i]->completion_lock->lock("LoadServer::process_packages 2");
282         }
283 }
284
285 void LoadServer::process_single()
286 {
287         is_single = 1;
288         create_clients();
289         create_packages();
290         init_packages();
291         current_package = 0;
292         single_client->run_single();
293 }
294