hvirtual/cinelerra/renderfarm.C

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