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