cinelerra/loadbalance.C

   1 #include "condition.h"
   2 #include "mutex.h"
   3 #include "loadbalance.h"
   4
   5
   6
   7
   8 LoadPackage::LoadPackage()
   9 {
  10         completion_lock = new Condition(0, "LoadPackage::completion_lock");
  11 }
  12 LoadPackage::~LoadPackage()
  13 {
  14         delete completion_lock;
  15 }
  16
  17
  18
  19
  20
  21
  22
  23
  24
  25 LoadClient::LoadClient(LoadServer *server)
  26  : Thread()
  27 {
  28         Thread::set_synchronous(1);
  29         this->server = server;
  30         done = 0;
  31         package_number = 0;
  32         input_lock = new Condition(0, "LoadClient::input_lock");
  33         completion_lock = new Condition(0, "LoadClient::completion_lock");
  34 }
  35
  36 LoadClient::~LoadClient()
  37 {
  38         done = 1;
  39         input_lock->unlock();
  40         Thread::join();
  41         delete input_lock;
  42         delete completion_lock;
  43 }
  44
  45 int LoadClient::get_package_number()
  46 {
  47         return package_number;
  48 }
  49
  50 LoadServer* LoadClient::get_server()
  51 {
  52         return server;
  53 }
  54
  55
  56 void LoadClient::run()
  57 {
  58         while(!done)
  59         {
  60                 input_lock->lock("LoadClient::run");
  61
  62                 if(!done)
  63                 {
  64 // Read packet
  65                         LoadPackage *package;
  66
  67
  68                         server->client_lock->lock("LoadClient::run");
  69                         if(server->current_package < server->total_packages)
  70                         {
  71                                 package_number = server->current_package;
  72                                 package = server->packages[server->current_package++];
  73                                 server->client_lock->unlock();
  74                                 input_lock->unlock();
  75
  76                                 process_package(package);
  77
  78                                 package->completion_lock->unlock();
  79                         }
  80                         else
  81                         {
  82                                 completion_lock->unlock();
  83                                 server->client_lock->unlock();
  84                         }
  85                 }
  86         }
  87 }
  88
  89
  90
  91
  92
  93
  94
  95
  96 LoadServer::LoadServer(int total_clients, int total_packages)
  97 {
  98         if(total_clients <= 0)
  99                 printf("LoadServer::LoadServer total_clients == %d\n", total_clients);
 100         this->total_clients = total_clients;
 101         this->total_packages = total_packages;
 102         current_package = 0;
 103         clients = 0;
 104         packages = 0;
 105         client_lock = new Mutex("LoadServer::client_lock");
 106 }
 107
 108 LoadServer::~LoadServer()
 109 {
 110         delete_clients();
 111         delete_packages();
 112         delete client_lock;
 113 }
 114
 115 void LoadServer::delete_clients()
 116 {
 117         if(clients)
 118         {
 119                 for(int i = 0; i < total_clients; i++)
 120                         delete clients[i];
 121                 delete [] clients;
 122         }
 123         clients = 0;
 124 }
 125
 126 void LoadServer::delete_packages()
 127 {
 128         if(packages)
 129         {
 130                 for(int i = 0; i < total_packages; i++)
 131                         delete packages[i];
 132                 delete [] packages;
 133         }
 134         packages = 0;
 135 }
 136
 137 void LoadServer::set_package_count(int total_packages)
 138 {
 139         delete_packages();
 140         this->total_packages = total_packages;
 141         create_packages();
 142 }
 143
 144
 145 void LoadServer::create_clients()
 146 {
 147         if(!clients)
 148         {
 149                 clients = new LoadClient*[total_clients];
 150                 for(int i = 0; i < total_clients; i++)
 151                 {
 152                         clients[i] = new_client();
 153                         clients[i]->start();
 154                 }
 155         }
 156 }
 157
 158 void LoadServer::create_packages()
 159 {
 160         if(!packages)
 161         {
 162                 packages = new LoadPackage*[total_packages];
 163                 for(int i = 0; i < total_packages; i++)
 164                         packages[i] = new_package();
 165         }
 166 }
 167
 168 LoadPackage* LoadServer::get_package(int number)
 169 {
 170         return packages[number];
 171 }
 172
 173 LoadClient* LoadServer::get_client(int number)
 174 {
 175         return clients[number];
 176 }
 177
 178 int LoadServer::get_total_packages()
 179 {
 180         return total_packages;
 181 }
 182
 183 int LoadServer::get_total_clients()
 184 {
 185         return total_clients;
 186 }
 187
 188 void LoadServer::process_packages()
 189 {
 190         if(!clients) create_clients();
 191         if(!packages) create_packages();
 192
 193
 194
 195 // Set up packages
 196         init_packages();
 197
 198         current_package = 0;
 199 // Start all clients
 200         for(int i = 0; i < total_clients; i++)
 201         {
 202                 clients[i]->input_lock->unlock();
 203         }
 204
 205 // Wait for packages to get finished
 206         for(int i = 0; i < total_packages; i++)
 207         {
 208                 packages[i]->completion_lock->lock("LoadServer::process_packages 1");
 209         }
 210
 211 // Wait for clients to finish before allowing changes to packages
 212         for(int i = 0; i < total_clients; i++)
 213         {
 214                 clients[i]->completion_lock->lock("LoadServer::process_packages 2");
 215         }
 216 }
 217