hvirtual/cinelerra/renderfarm.C

   1 #include "asset.h"
   2 #include "brender.h"
   3 #include "clip.h"
   4 #include "defaults.h"
   5 #include "edl.h"
   6 #include "filesystem.h"
   7 #include "filexml.h"
   8 #include "language.h"
   9 #include "mutex.h"
  10 //#include "mwindow.h"
  11 #include "packagedispatcher.h"
  12 #include "preferences.h"
  13 #include "render.h"
  14 #include "renderfarm.h"
  15 #include "renderfarmfsserver.h"
  16 #include "bctimer.h"
  17 #include "transportque.h"
  18
  19
  20 #include <arpa/inet.h>
  21 #include <errno.h>
  22 #include <fcntl.h>
  23 #include <netdb.h>
  24 #include <netinet/in.h>
  25 #include <stdio.h>
  26 #include <string.h>
  27 #include <sys/socket.h>
  28 #include <sys/types.h>
  29 #include <sys/un.h>
  30 #include <unistd.h>
  31
  32
  33
  34
  35 RenderFarmServer::RenderFarmServer(ArrayList<PluginServer*> *plugindb,
  36         PackageDispatcher *packages,
  37         Preferences *preferences,
  38         int use_local_rate,
  39         int *result_return,
  40         int64_t *total_return,
  41         Mutex *total_return_lock,
  42         Asset *default_asset,
  43         EDL *edl,
  44         BRender *brender)
  45 {
  46         this->plugindb = plugindb;
  47         this->packages = packages;
  48         this->preferences = preferences;
  49         this->use_local_rate = use_local_rate;
  50         this->result_return = result_return;
  51         this->total_return = total_return;
  52         this->total_return_lock = total_return_lock;
  53         this->default_asset = default_asset;
  54         this->edl = edl;
  55         this->brender = brender;
  56         client_lock = new Mutex("RenderFarmServer::client_lock");
  57 }
  58
  59 RenderFarmServer::~RenderFarmServer()
  60 {
  61         clients.remove_all_objects();
  62         delete client_lock;
  63 }
  64
  65 // Open connections to clients.
  66 int RenderFarmServer::start_clients()
  67 {
  68         int result = 0;
  69
  70         for(int i = 0; i < preferences->get_enabled_nodes() && !result; i++)
  71         {
  72                 client_lock->lock("RenderFarmServer::start_clients");
  73                 RenderFarmServerThread *client = new RenderFarmServerThread(plugindb,
  74                         this,
  75                         i);
  76                 clients.append(client);
  77
  78                 result = client->start_loop();
  79                 client_lock->unlock();
  80 //usleep(100000);
  81 // Fails to connect all without a delay
  82         }
  83
  84         return result;
  85 }
  86
  87 // The render farm must wait for all the clients to finish.
  88 int RenderFarmServer::wait_clients()
  89 {
  90 //printf("RenderFarmServer::wait_clients 1\n");
  91         clients.remove_all_objects();
  92 //printf("RenderFarmServer::wait_clients 2\n");
  93         return 0;
  94 }
  95
  96
  97
  98
  99
 100
 101
 102
 103
 104
 105
 106
 107 // Waits for requests from every client.
 108 // Joins when the client is finished.
 109 RenderFarmServerThread::RenderFarmServerThread(ArrayList<PluginServer*> *plugindb,
 110         RenderFarmServer *server,
 111         int number)
 112  : Thread()
 113 {
 114         this->plugindb = plugindb;
 115         this->server = server;
 116         this->number = number;
 117         socket_fd = -1;
 118         frames_per_second = 0;
 119         Thread::set_synchronous(1);
 120 }
 121
 122
 123
 124 RenderFarmServerThread::~RenderFarmServerThread()
 125 {
 126 //printf("RenderFarmServerThread::~RenderFarmServerThread 1 %p\n", this);
 127         Thread::join();
 128 //printf("RenderFarmServerThread::~RenderFarmServerThread 1\n");
 129         if(socket_fd >= 0) close(socket_fd);
 130 //printf("RenderFarmServerThread::~RenderFarmServerThread 2\n");
 131 }
 132
 133
 134 int RenderFarmServerThread::start_loop()
 135 {
 136         int result = 0;
 137         char *hostname = server->preferences->get_node_hostname(number);
 138 //printf("RenderFarmServerThread::start_loop 1\n");
 139
 140 // Open file for master node
 141         if(hostname[0] == '/')
 142         {
 143                 if((socket_fd = socket(PF_UNIX, SOCK_STREAM, 0)) < 0)
 144                 {
 145                         perror(_("RenderFarmServerThread::start_loop: socket\n"));
 146                         result = 1;
 147                 }
 148                 else
 149                 {
 150                         struct sockaddr_un addr;
 151                         addr.sun_family = AF_FILE;
 152                         strcpy(addr.sun_path, hostname);
 153                         int size = (offsetof(struct sockaddr_un, sun_path) +
 154                                 strlen(hostname) + 1);
 155
 156 // The master node is always created by BRender.  Keep trying for 30 seconds.
 157
 158 #define ATTEMPT_DELAY 100000
 159                         int done = 0;
 160                         int attempt = 0;
 161 //printf("RenderFarmServerThread::start_loop 2 %s\n", hostname);
 162                         do
 163                         {
 164 //printf("RenderFarmServerThread::start_loop 3\n");
 165                                 if(connect(socket_fd, (struct sockaddr*)&addr, size) < 0)
 166                                 {
 167                                         attempt++;
 168                                         if(attempt > 30000000 / ATTEMPT_DELAY)
 169                                         {
 170 //printf("RenderFarmServerThread::start_loop 4 %s\n", hostname);
 171                                                 fprintf(stderr, _("RenderFarmServerThread::start_loop: %s: %s\n"),
 172                                                         hostname,
 173                                                         strerror(errno));
 174                                                 result = 1;
 175                                         }
 176                                         else
 177                                                 usleep(ATTEMPT_DELAY);
 178 //printf("RenderFarmServerThread::start_loop 5 %s\n", hostname);
 179                                 }
 180                                 else
 181                                         done = 1;
 182                         }while(!result && !done);
 183 //printf("RenderFarmServerThread::start_loop 6\n");
 184                 }
 185         }
 186         else
 187 // Open socket
 188         {
 189                 if((socket_fd = socket(PF_INET, SOCK_STREAM, 0)) < 0)
 190                 {
 191                         perror(_("RenderFarmServerThread::start_loop: socket"));
 192                         result = 1;
 193                 }
 194                 else
 195                 {
 196 // Open port
 197                         struct sockaddr_in addr;
 198                         struct hostent *hostinfo;
 199                         addr.sin_family = AF_INET;
 200                         addr.sin_port = htons(server->preferences->get_node_port(number));
 201                         hostinfo = gethostbyname(hostname);
 202                         if(hostinfo == NULL)
 203                 {
 204                         fprintf(stderr, _("RenderFarmServerThread::start_loop: unknown host %s.\n"),
 205                                         server->preferences->get_node_hostname(number));
 206                         result = 1;
 207                 }
 208                         else
 209                         {
 210                                 addr.sin_addr = *(struct in_addr *) hostinfo->h_addr;
 211
 212                                 if(connect(socket_fd, (struct sockaddr*)&addr, sizeof(addr)) < 0)
 213                                 {
 214                                         fprintf(stderr, _("RenderFarmServerThread::start_loop: %s: %s\n"),
 215                                                 server->preferences->get_node_hostname(number),
 216                                                 strerror(errno));
 217                                         result = 1;
 218                                 }
 219                         }
 220                 }
 221         }
 222 //printf("RenderFarmServerThread::start_loop 7\n");
 223
 224         if(!result) Thread::start();
 225
 226         return result;
 227 }
 228
 229
 230 int RenderFarmServerThread::read_socket(int socket_fd, char *data, int len, int timeout)
 231 {
 232         int bytes_read = 0;
 233         int offset = 0;
 234 //timeout = 0;
 235
 236 //printf("RenderFarmServerThread::read_socket 1\n");
 237         while(len > 0 && bytes_read >= 0)
 238         {
 239                 int result = 0;
 240                 if(timeout > 0)
 241                 {
 242                         fd_set read_fds;
 243                         struct timeval tv;
 244
 245                         FD_ZERO(&read_fds);
 246                         FD_SET(socket_fd, &read_fds);
 247                         tv.tv_sec = timeout;
 248                         tv.tv_usec = 0;
 249
 250                         result = select(socket_fd + 1,
 251                                 &read_fds,
 252                                 0,
 253                                 0,
 254                                 &tv);
 255                         FD_ZERO(&read_fds);
 256 //printf("RenderFarmServerThread::read_socket 1 %d\n", result);
 257                 }
 258                 else
 259                         result = 1;
 260
 261                 if(result)
 262                 {
 263                         bytes_read = read(socket_fd, data + offset, len);
 264                         if(bytes_read > 0)
 265                         {
 266                                 len -= bytes_read;
 267                                 offset += bytes_read;
 268                         }
 269                         else
 270                         {
 271 //printf("RenderFarmServerThread::read_socket got 0 len=%d\n", len);
 272                                 break;
 273                         }
 274                 }
 275                 else
 276                 {
 277 printf("RenderFarmServerThread::read_socket timed out. len=%d\n", len);
 278                         break;
 279                 }
 280         }
 281 //printf("RenderFarmServerThread::read_socket 2\n");
 282
 283         return offset;
 284 }
 285
 286 int RenderFarmServerThread::write_socket(int socket_fd, char *data, int len, int timeout)
 287 {
 288         int result = 0;
 289         if(timeout > 0)
 290         {
 291                 fd_set write_fds;
 292                 struct timeval tv;
 293                 FD_ZERO(&write_fds);
 294                 FD_SET(socket_fd, &write_fds);
 295                 tv.tv_sec = timeout;
 296                 tv.tv_usec = 0;
 297 //printf("RenderFarmServerThread::write_socket 1\n");
 298                 result = select(socket_fd + 1,
 299                                 0,
 300                                 &write_fds,
 301                                 0,
 302                                 &tv);
 303 //printf("RenderFarmServerThread::write_socket 2\n");
 304                 FD_ZERO(&write_fds);
 305                 if(!result)
 306                 {
 307 printf("RenderFarmServerThread::write_socket 1 socket timed out. len=%d\n", len);
 308                         return 0;
 309                 }
 310         }
 311
 312         return write(socket_fd, data, len);
 313 }
 314
 315 int RenderFarmServerThread::read_socket(char *data, int len, int timeout)
 316 {
 317         return read_socket(socket_fd, data, len, timeout);
 318 }
 319
 320 int RenderFarmServerThread::write_socket(char *data, int len, int timeout)
 321 {
 322         return write_socket(socket_fd, data, len, timeout);
 323 }
 324
 325 void RenderFarmServerThread::reallocate_buffer(int size)
 326 {
 327         if(buffer && buffer_allocated < size)
 328         {
 329                 delete [] buffer;
 330                 buffer = 0;
 331         }
 332
 333         if(!buffer && size)
 334         {
 335                 buffer = new unsigned char[size];
 336                 buffer_allocated = size;
 337         }
 338 }
 339
 340 void RenderFarmServerThread::run()
 341 {
 342 // Wait for requests
 343         unsigned char header[5];
 344         int done = 0;
 345
 346
 347         buffer = 0;
 348         buffer_allocated = 0;
 349         fs_server = new RenderFarmFSServer(this);
 350         fs_server->initialize();
 351         while(!done)
 352         {
 353 // Wait for requests.
 354 // Requests consist of request ID's and accompanying buffers.
 355 // Get request ID.
 356                 if(read_socket(socket_fd, (char*)header, 5, -1) != 5)
 357                 {
 358                         done = 1;
 359                         continue;
 360                 }
 361
 362                 int request_id = header[0];
 363 //printf("RenderFarmServerThread::run 1 %d\n", request_id);
 364                 int64_t request_size = (((u_int32_t)header[1]) << 24) |
 365                                                         (((u_int32_t)header[2]) << 16) |
 366                                                         (((u_int32_t)header[3]) << 8)  |
 367                                                         (u_int32_t)header[4];
 368
 369                 reallocate_buffer(request_size);
 370
 371 // Get accompanying buffer
 372                 if(read_socket((char*)buffer, request_size, RENDERFARM_TIMEOUT) != request_size)
 373                 {
 374                         done = 1;
 375                         continue;
 376                 }
 377
 378                 switch(request_id)
 379                 {
 380                         case RENDERFARM_PREFERENCES:
 381                                 send_preferences();
 382                                 break;
 383
 384                         case RENDERFARM_ASSET:
 385                                 send_asset();
 386                                 break;
 387
 388                         case RENDERFARM_EDL:
 389                                 send_edl();
 390                                 break;
 391
 392                         case RENDERFARM_PACKAGE:
 393                                 send_package(buffer);
 394                                 break;
 395
 396                         case RENDERFARM_PROGRESS:
 397                                 set_progress(buffer);
 398                                 break;
 399
 400                         case RENDERFARM_SET_RESULT:
 401                                 set_result(buffer);
 402                                 break;
 403
 404                         case RENDERFARM_SET_VMAP:
 405                                 set_video_map(buffer);
 406                                 break;
 407
 408                         case RENDERFARM_GET_RESULT:
 409                                 get_result();
 410                                 break;
 411
 412                         case RENDERFARM_DONE:
 413                                 done = 1;
 414                                 break;
 415
 416
 417                         default:
 418                                 if(!fs_server->handle_request(request_id, request_size, (char*)buffer))
 419                                 {
 420                                         printf(_("RenderFarmServerThread::run: unknown request %02x\n"), request_id);
 421                                 }
 422                                 break;
 423                 }
 424 //printf("RenderFarmServerThread::run 4\n");
 425         }
 426 //printf("RenderFarmServerThread::run 5\n");
 427
 428         if(buffer) delete [] buffer;
 429         delete fs_server;
 430 }
 431
 432 int RenderFarmServerThread::write_string(int socket_fd, char *string)
 433 {
 434         unsigned char *datagram;
 435         int i, len;
 436         i = 0;
 437
 438         len = strlen(string) + 1;
 439         datagram = new unsigned char[len + 4];
 440         STORE_INT32(len);
 441         memcpy(datagram + i, string, len);
 442         write_socket(socket_fd, (char*)datagram, len + 4, RENDERFARM_TIMEOUT);
 443 //printf("RenderFarmServerThread::write_string %02x%02x%02x%02x\n",
 444 //      datagram[0], datagram[1], datagram[2], datagram[3]);
 445
 446         delete [] datagram;
 447 }
 448
 449 void RenderFarmServerThread::send_preferences()
 450 {
 451         Defaults defaults;
 452         char *string;
 453
 454 //printf("RenderFarmServerThread::send_preferences 1\n");
 455         server->preferences->save_defaults(&defaults);
 456         defaults.save_string(string);
 457 //printf("RenderFarmServerThread::send_preferences 2 \n%s\n", string);
 458         write_string(socket_fd, string);
 459
 460         delete [] string;
 461 }
 462
 463 void RenderFarmServerThread::send_asset()
 464 {
 465         FileXML file;
 466 //printf("RenderFarmServerThread::send_asset 1\n");
 467
 468         server->default_asset->write(plugindb,
 469                 &file,
 470                 0,
 471                 0);
 472         file.terminate_string();
 473
 474 //printf("RenderFarmServerThread::send_asset 2\n");
 475         write_string(socket_fd, file.string);
 476 //printf("RenderFarmServerThread::send_asset 3\n");
 477 }
 478
 479
 480 void RenderFarmServerThread::send_edl()
 481 {
 482         FileXML file;
 483
 484 // Save the XML
 485         server->edl->save_xml(plugindb,
 486                 &file,
 487                 0,
 488                 0,
 489                 0);
 490         file.terminate_string();
 491 //printf("RenderFarmServerThread::send_edl\n%s\n\n", file.string);
 492
 493         write_string(socket_fd, file.string);
 494 //printf("RenderFarmServerThread::send_edl 2\n");
 495 }
 496
 497
 498 void RenderFarmServerThread::send_package(unsigned char *buffer)
 499 {
 500         this->frames_per_second = (double)((((u_int32_t)buffer[0]) << 24) |
 501                 (((u_int32_t)buffer[1]) << 16) |
 502                 (((u_int32_t)buffer[2]) << 8)  |
 503                 ((u_int32_t)buffer[3])) /
 504                 65536.0;
 505
 506 //printf("RenderFarmServerThread::send_package 1 %f\n", frames_per_second);
 507         RenderPackage *package =
 508                 server->packages->get_package(frames_per_second,
 509                         number,
 510                         server->use_local_rate);
 511
 512 //printf("RenderFarmServerThread::send_package 2\n");
 513
 514         char datagram[BCTEXTLEN];
 515
 516 // No more packages
 517         if(!package)
 518         {
 519 //printf("RenderFarmServerThread::send_package 1\n");
 520                 datagram[0] = datagram[1] = datagram[2] = datagram[3] = 0;
 521                 write_socket(datagram, 4, RENDERFARM_TIMEOUT);
 522         }
 523         else
 524 // Encode package
 525         {
 526 //printf("RenderFarmServerThread::send_package 10\n");
 527                 int i = 4;
 528                 strcpy(&datagram[i], package->path);
 529                 i += strlen(package->path);
 530                 datagram[i++] = 0;
 531
 532                 STORE_INT32(package->audio_start);
 533                 STORE_INT32(package->audio_end);
 534                 STORE_INT32(package->video_start);
 535                 STORE_INT32(package->video_end);
 536                 int use_brender = (server->brender ? 1 : 0);
 537                 STORE_INT32(use_brender);
 538
 539                 int len = i;
 540                 i = 0;
 541                 STORE_INT32(len - 4);
 542
 543                 write_socket(datagram, len, RENDERFARM_TIMEOUT);
 544         }
 545 }
 546
 547
 548 void RenderFarmServerThread::set_progress(unsigned char *buffer)
 549 {
 550         server->total_return_lock->lock("RenderFarmServerThread::set_progress");
 551         *server->total_return += (int64_t)(((u_int32_t)buffer[0]) << 24) |
 552                                                                                         (((u_int32_t)buffer[1]) << 16) |
 553                                                                                         (((u_int32_t)buffer[2]) << 8)  |
 554                                                                                         ((u_int32_t)buffer[3]);
 555         server->total_return_lock->unlock();
 556 }
 557
 558 void RenderFarmServerThread::set_video_map(unsigned char *buffer)
 559 {
 560         if(server->brender)
 561                 server->brender->set_video_map((int64_t)(((u_int32_t)buffer[0]) << 24) |
 562                                                         (((u_int32_t)buffer[1]) << 16) |
 563                                                         (((u_int32_t)buffer[2]) << 8)  |
 564                                                         ((u_int32_t)buffer[3]),
 565                                                         (int64_t)(((u_int32_t)buffer[4]) << 24) |
 566                                                         (((u_int32_t)buffer[5]) << 16) |
 567                                                         (((u_int32_t)buffer[6]) << 8)  |
 568                                                         ((u_int32_t)buffer[7]));
 569 }
 570
 571
 572 void RenderFarmServerThread::set_result(unsigned char *buffer)
 573 {
 574 //printf("RenderFarmServerThread::set_result %p\n", buffer);
 575         if(!*server->result_return)
 576                 *server->result_return = buffer[0];
 577 }
 578
 579
 580 void RenderFarmServerThread::get_result()
 581 {
 582         unsigned char data[1];
 583         data[0] = *server->result_return;
 584         write_socket((char*)data, 1, RENDERFARM_TIMEOUT);
 585 }
 586
 587