cinelerra-5.0/cinelerra/renderfarm.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 "asset.h"
  23 #include "bcsignals.h"
  24 #include "brender.h"
  25 #include "clip.h"
  26 #include "condition.h"
  27 #include "bchash.h"
  28 #include "edl.h"
  29 #include "filesystem.h"
  30 #include "filexml.h"
  31 #include "format.inc"
  32 #include "language.h"
  33 #include "mutex.h"
  34 #include "mwindow.h"
  35 #include "packagedispatcher.h"
  36 #include "preferences.h"
  37 #include "render.h"
  38 #include "renderfarm.h"
  39 #include "renderfarmclient.h"
  40 #include "bctimer.h"
  41 #include "transportque.h"
  42
  43
  44 #include <arpa/inet.h>
  45 #include <errno.h>
  46 #include <fcntl.h>
  47 #include <netdb.h>
  48 #include <netinet/in.h>
  49 #include <signal.h>
  50 #include <stdio.h>
  51 #include <string.h>
  52 #include <sys/socket.h>
  53 #include <sys/types.h>
  54 #include <sys/un.h>
  55 #include <unistd.h>
  56
  57
  58
  59
  60 RenderFarmServer::RenderFarmServer(
  61         MWindow *mwindow,
  62         PackageDispatcher *packages,
  63         Preferences *preferences,
  64         int use_local_rate,
  65         int *result_return,
  66         int64_t *total_return,
  67         Mutex *total_return_lock,
  68         Asset *default_asset,
  69         EDL *edl,
  70         BRender *brender)
  71 {
  72         this->mwindow = mwindow;
  73         this->packages = packages;
  74         this->preferences = preferences;
  75         this->use_local_rate = use_local_rate;
  76         this->result_return = result_return;
  77         this->total_return = total_return;
  78         this->total_return_lock = total_return_lock;
  79         this->default_asset = default_asset;
  80         this->edl = edl;
  81         this->brender = brender;
  82         client_lock = new Mutex("RenderFarmServer::client_lock");
  83 }
  84
  85 RenderFarmServer::~RenderFarmServer()
  86 {
  87         clients.remove_all_objects();
  88         delete client_lock;
  89 }
  90
  91 // Open connections to clients.
  92 int RenderFarmServer::start_clients()
  93 {
  94         int result = 0;
  95
  96         for(int i = 0; i < preferences->get_enabled_nodes() && !result; i++)
  97         {
  98                 client_lock->lock("RenderFarmServer::start_clients");
  99                 RenderFarmServerThread *client = new RenderFarmServerThread(this,
 100                         i);
 101                 clients.append(client);
 102
 103                 result = client->start_loop();
 104                 client_lock->unlock();
 105         }
 106
 107         return result;
 108 }
 109
 110 // The render farm must wait for all the clients to finish.
 111 int RenderFarmServer::wait_clients()
 112 {
 113 //printf("RenderFarmServer::wait_clients 1\n");
 114         clients.remove_all_objects();
 115 //printf("RenderFarmServer::wait_clients 2\n");
 116         return 0;
 117 }
 118
 119
 120
 121
 122
 123
 124
 125
 126
 127
 128
 129
 130 // Waits for requests from every client.
 131 // Joins when the client is finished.
 132 RenderFarmServerThread::RenderFarmServerThread(RenderFarmServer *server,
 133         int number)
 134  : Thread(1, 0, 0)
 135 {
 136         this->server = server;
 137         this->number = number;
 138         socket_fd = -1;
 139         frames_per_second = 0;
 140         watchdog = 0;
 141         buffer = 0;
 142         datagram = 0;
 143         Thread::set_synchronous(1);
 144 }
 145
 146
 147
 148 RenderFarmServerThread::~RenderFarmServerThread()
 149 {
 150 //printf("RenderFarmServerThread::~RenderFarmServerThread 1 %p\n", this);
 151         Thread::join();
 152 //printf("RenderFarmServerThread::~RenderFarmServerThread 1\n");
 153         if(socket_fd >= 0) close(socket_fd);
 154         if(watchdog) delete watchdog;
 155         if(buffer) delete [] buffer;
 156         if(datagram) delete [] datagram;
 157 //printf("RenderFarmServerThread::~RenderFarmServerThread 2\n");
 158 }
 159
 160
 161 int RenderFarmServerThread::open_client(const char *hostname, int port)
 162 {
 163         int socket_fd = -1;
 164         int result = 0;
 165
 166 // Open file for master node
 167         if(hostname[0] == '/')
 168         {
 169                 if((socket_fd = socket(PF_UNIX, SOCK_STREAM, 0)) < 0)
 170                 {
 171                         perror(_("RenderFarmServerThread::start_loop: socket\n"));
 172                         result = 1;
 173                 }
 174                 else
 175                 {
 176                         struct sockaddr_un addr;
 177                         addr.sun_family = AF_FILE;
 178                         strcpy(addr.sun_path, hostname);
 179                         int size = (offsetof(struct sockaddr_un, sun_path) +
 180                                 strlen(hostname) + 1);
 181
 182 // The master node is always created by BRender.  Keep trying for 30 seconds.
 183
 184 #define ATTEMPT_DELAY 100000
 185                         int done = 0;
 186                         int attempt = 0;
 187
 188                         do
 189                         {
 190                                 if(connect(socket_fd, (struct sockaddr*)&addr, size) < 0)
 191                                 {
 192                                         attempt++;
 193                                         if(attempt > 30000000 / ATTEMPT_DELAY)
 194                                         {
 195                                                 fprintf(stderr, _("RenderFarmServerThread::open_client: %s: %s\n"),
 196                                                         hostname,
 197                                                         strerror(errno));
 198                                                 result = 1;
 199                                         }
 200                                         else
 201                                                 usleep(ATTEMPT_DELAY);
 202                                 }
 203                                 else
 204                                         done = 1;
 205                         }while(!result && !done);
 206                 }
 207         }
 208         else
 209 // Open socket
 210         {
 211                 if((socket_fd = socket(PF_INET, SOCK_STREAM, 0)) < 0)
 212                 {
 213                         perror(_("RenderFarmServerThread::start_loop: socket"));
 214                         result = 1;
 215                 }
 216                 else
 217                 {
 218 // Open port
 219                         struct sockaddr_in addr;
 220                         struct hostent *hostinfo;
 221                         addr.sin_family = AF_INET;
 222                         addr.sin_port = htons(port);
 223                         hostinfo = gethostbyname(hostname);
 224                         if(hostinfo == NULL)
 225                 {
 226                         fprintf(stderr, _("RenderFarmServerThread::open_client: unknown host %s.\n"),
 227                                         hostname);
 228                         result = 1;
 229                 }
 230                         else
 231                         {
 232                                 addr.sin_addr = *(struct in_addr *) hostinfo->h_addr;
 233
 234                                 if(connect(socket_fd, (struct sockaddr*)&addr, sizeof(addr)) < 0)
 235                                 {
 236                                         fprintf(stderr, _("RenderFarmServerThread::open_client: %s: %s\n"),
 237                                                 hostname,
 238                                                 strerror(errno));
 239                                         result = 1;
 240                                 }
 241                         }
 242                 }
 243         }
 244
 245         if(result) socket_fd = -1;
 246
 247         return socket_fd;
 248 }
 249
 250 int RenderFarmServerThread::start_loop()
 251 {
 252         int result = 0;
 253
 254         socket_fd = open_client(server->preferences->get_node_hostname(number),
 255                 server->preferences->get_node_port(number));
 256
 257         if(socket_fd < 0) result = 1;
 258
 259         if(!result)
 260         {
 261                 watchdog = new RenderFarmWatchdog(this, 0);
 262                 watchdog->start();
 263         }
 264
 265         if(!result) Thread::start();
 266
 267         return result;
 268 }
 269
 270
 271
 272
 273
 274
 275
 276
 277
 278
 279
 280 int64_t RenderFarmServerThread::read_int64(int *error)
 281 {
 282         int temp = 0;
 283         if(!error) error = &temp;
 284
 285         unsigned char data[sizeof(int64_t)];
 286         *error = (read_socket((char*)data, sizeof(int64_t)) !=
 287                 sizeof(int64_t));
 288
 289 // Make it return 1 if error so it can be used to read a result code from the
 290 // server.
 291         int64_t result = 1;
 292         if(!*error)
 293         {
 294                 result = (((int64_t)data[0]) << 56) |
 295                         (((uint64_t)data[1]) << 48) |
 296                         (((uint64_t)data[2]) << 40) |
 297                         (((uint64_t)data[3]) << 32) |
 298                         (((uint64_t)data[4]) << 24) |
 299                         (((uint64_t)data[5]) << 16) |
 300                         (((uint64_t)data[6]) << 8)  |
 301                         data[7];
 302         }
 303         return result;
 304 }
 305
 306 int RenderFarmServerThread::write_int64(int64_t value)
 307 {
 308         unsigned char data[sizeof(int64_t)];
 309         data[0] = (value >> 56) & 0xff;
 310         data[1] = (value >> 48) & 0xff;
 311         data[2] = (value >> 40) & 0xff;
 312         data[3] = (value >> 32) & 0xff;
 313         data[4] = (value >> 24) & 0xff;
 314         data[5] = (value >> 16) & 0xff;
 315         data[6] = (value >> 8) & 0xff;
 316         data[7] = value & 0xff;
 317         return (write_socket((char*)data, sizeof(int64_t)) !=
 318                 sizeof(int64_t));
 319 }
 320
 321
 322
 323 int RenderFarmServerThread::read_socket(char *data, int len)
 324 {
 325         int bytes_read = 0;
 326         int offset = 0;
 327 //printf("RenderFarmServerThread::read_socket 1\n");
 328         watchdog->begin_request();
 329         while(len > 0 && bytes_read >= 0)
 330         {
 331                 enable_cancel();
 332                 bytes_read = read(socket_fd, data + offset, len);
 333                 disable_cancel();
 334                 if(bytes_read > 0)
 335                 {
 336                         len -= bytes_read;
 337                         offset += bytes_read;
 338                 }
 339                 else
 340                 if(bytes_read < 0)
 341                         break;
 342         }
 343         watchdog->end_request();
 344 //printf("RenderFarmServerThread::read_socket 10\n");
 345
 346         return offset;
 347 }
 348
 349 int RenderFarmServerThread::write_socket(char *data, int len)
 350 {
 351 //printf("RenderFarmServerThread::write_socket 1\n");
 352         int result = write(socket_fd, data, len);
 353 //printf("RenderFarmServerThread::write_socket 10\n");
 354
 355         return result;
 356 }
 357
 358 void RenderFarmServerThread::reallocate_buffer(int size)
 359 {
 360         if(buffer && buffer_allocated < size)
 361         {
 362                 delete [] buffer;
 363                 buffer = 0;
 364         }
 365
 366         if(!buffer && size)
 367         {
 368                 buffer = new unsigned char[size];
 369                 buffer_allocated = size;
 370         }
 371 }
 372
 373 void RenderFarmServerThread::run()
 374 {
 375 // Wait for requests
 376         unsigned char header[5];
 377         int done = 0;
 378         int bytes_read = 0;
 379
 380
 381         buffer = 0;
 382         buffer_allocated = 0;
 383 //      fs_server = new RenderFarmFSServer(this);
 384 //      fs_server->initialize();
 385
 386
 387 // Send command to run package renderer.
 388         write_int64(RENDERFARM_PACKAGES);
 389
 390
 391
 392         while(!done)
 393         {
 394
 395 // Wait for requests.
 396 // Requests consist of request ID's and accompanying buffers.
 397 // Get request ID.
 398                 bytes_read = read_socket((char*)header, 5);
 399 //printf("RenderFarmServerThread::run 1\n");
 400                 if(bytes_read != 5)
 401                 {
 402                         done = 1;
 403                         continue;
 404                 }
 405
 406                 int request_id = header[0];
 407                 int64_t request_size = (((u_int32_t)header[1]) << 24) |
 408                                                         (((u_int32_t)header[2]) << 16) |
 409                                                         (((u_int32_t)header[3]) << 8)  |
 410                                                         (u_int32_t)header[4];
 411
 412                 reallocate_buffer(request_size);
 413
 414 // Get accompanying buffer
 415                 bytes_read = read_socket((char*)buffer, request_size);
 416
 417 //printf("RenderFarmServerThread::run 2 %d " _LD " %d\n", request_id, request_size, bytes_read);
 418                 if(bytes_read != request_size)
 419                 {
 420                         done = 1;
 421                         continue;
 422                 }
 423 //printf("RenderFarmServerThread::run 3\n");
 424
 425                 switch(request_id)
 426                 {
 427                         case RENDERFARM_PREFERENCES:
 428                                 send_preferences();
 429                                 break;
 430
 431                         case RENDERFARM_ASSET:
 432                                 send_asset();
 433                                 break;
 434
 435                         case RENDERFARM_EDL:
 436                                 send_edl();
 437                                 break;
 438
 439                         case RENDERFARM_PACKAGE:
 440                                 send_package(buffer);
 441                                 break;
 442
 443                         case RENDERFARM_PROGRESS:
 444                                 set_progress(buffer);
 445                                 break;
 446
 447                         case RENDERFARM_SET_RESULT:
 448                                 set_result(buffer);
 449                                 break;
 450
 451                         case RENDERFARM_SET_VMAP:
 452                                 set_video_map(buffer);
 453                                 break;
 454
 455                         case RENDERFARM_GET_RESULT:
 456                                 get_result();
 457                                 break;
 458
 459                         case RENDERFARM_DONE:
 460 //printf("RenderFarmServerThread::run 10\n");
 461                                 done = 1;
 462                                 break;
 463
 464                         case RENDERFARM_KEEPALIVE:
 465                                 break;
 466
 467                         default:
 468 //                              if(!fs_server->handle_request(request_id, request_size, (unsigned char*)buffer))
 469                                 {
 470                                         printf(_("RenderFarmServerThread::run: unknown request %02x\n"), request_id);
 471                                 }
 472                                 break;
 473                 }
 474 //printf("RenderFarmServerThread::run 10 %d " _LD "\n", request_id, request_size);
 475         }
 476
 477 // Don't let watchdog kill the entire renderfarm when a client finishes
 478 // normally.
 479         if(watchdog)
 480         {
 481 //printf("RenderFarmServerThread::run 20\n");
 482                 delete watchdog;
 483                 watchdog = 0;
 484         }
 485
 486 //      delete fs_server;
 487 }
 488
 489 int RenderFarmServerThread::write_string(char *string)
 490 {
 491         int i, len;
 492         i = 0;
 493
 494         len = strlen(string) + 1;
 495         datagram = new char[len + 4];
 496         STORE_INT32(len);
 497         memcpy(datagram + i, string, len);
 498         write_socket((char*)datagram, len + 4);
 499 //printf("RenderFarmServerThread::write_string %02x%02x%02x%02x\n",
 500 //      datagram[0], datagram[1], datagram[2], datagram[3]);
 501
 502         delete [] datagram;
 503         datagram = 0;
 504         return 0;
 505 }
 506
 507 void RenderFarmServerThread::send_preferences()
 508 {
 509         BC_Hash defaults;
 510         char *string;
 511
 512         server->preferences->save_defaults(&defaults);
 513         defaults.save_string(string);
 514         write_string(string);
 515         free(string);
 516 }
 517
 518 void RenderFarmServerThread::send_asset()
 519 {
 520         BC_Hash defaults;
 521         char *string1;
 522
 523 // The asset must be sent in two segments.
 524 // One segment is stored in the EDL and contains decoding information.
 525 // One segment is stored in the asset and contains encoding information.
 526         server->default_asset->save_defaults(&defaults,
 527                 0,
 528                 1,
 529                 1,
 530                 1,
 531                 1,
 532                 1);
 533         defaults.save_string(string1);
 534         FileXML file;
 535         server->default_asset->write(&file, 0, 0);
 536         file.terminate_string();
 537
 538         write_string(string1);
 539         write_string(file.string());
 540         free(string1);
 541 }
 542
 543
 544 void RenderFarmServerThread::send_edl()
 545 {
 546         FileXML file;
 547
 548 // Save the XML
 549         server->edl->save_xml(&file,
 550                 0,
 551                 0,
 552                 0);
 553         file.terminate_string();
 554 //printf("RenderFarmServerThread::send_edl\n%s\n\n", file.string);
 555
 556         write_string(file.string());
 557 //printf("RenderFarmServerThread::send_edl 2\n");
 558 }
 559
 560
 561 void RenderFarmServerThread::send_package(unsigned char *buffer)
 562 {
 563         this->frames_per_second = (double)((((u_int32_t)buffer[0]) << 24) |
 564                 (((u_int32_t)buffer[1]) << 16) |
 565                 (((u_int32_t)buffer[2]) << 8)  |
 566                 ((u_int32_t)buffer[3])) /
 567                 65536.0;
 568
 569 //printf("RenderFarmServerThread::send_package 1 %f\n", frames_per_second);
 570         RenderPackage *package =
 571                 server->packages->get_package(frames_per_second,
 572                         number,
 573                         server->use_local_rate);
 574
 575 //printf("RenderFarmServerThread::send_package 2\n");
 576         datagram = new char[BCTEXTLEN];
 577
 578 // No more packages
 579         if(!package)
 580         {
 581 //printf("RenderFarmServerThread::send_package 1\n");
 582                 datagram[0] = datagram[1] = datagram[2] = datagram[3] = 0;
 583                 write_socket(datagram, 4);
 584         }
 585         else
 586 // Encode package
 587         {
 588 //printf("RenderFarmServerThread::send_package 10\n");
 589                 int i = 4;
 590                 strcpy(&datagram[i], package->path);
 591                 i += strlen(package->path);
 592                 datagram[i++] = 0;
 593
 594                 STORE_INT32(package->audio_start);
 595                 STORE_INT32(package->audio_end);
 596                 STORE_INT32(package->video_start);
 597                 STORE_INT32(package->video_end);
 598                 int use_brender = (server->brender ? 1 : 0);
 599                 STORE_INT32(use_brender);
 600
 601                 int len = i;
 602                 i = 0;
 603                 STORE_INT32(len - 4);
 604
 605                 write_socket(datagram, len);
 606         }
 607         delete [] datagram;
 608         datagram = 0;
 609 }
 610
 611
 612 void RenderFarmServerThread::set_progress(unsigned char *buffer)
 613 {
 614         server->total_return_lock->lock("RenderFarmServerThread::set_progress");
 615         *server->total_return += (int64_t)(((u_int32_t)buffer[0]) << 24) |
 616                 (((u_int32_t)buffer[1]) << 16) |
 617                 (((u_int32_t)buffer[2]) << 8)  |
 618                 ((u_int32_t)buffer[3]);
 619         frames_per_second = (double)((((u_int32_t)buffer[4]) << 24) |
 620                 (((u_int32_t)buffer[5]) << 16) |
 621                 (((u_int32_t)buffer[6]) << 8)  |
 622                 ((u_int32_t)buffer[7])) /
 623                 65536.0;
 624         server->total_return_lock->unlock();
 625
 626         server->preferences->set_rate(frames_per_second, number);
 627
 628 // This locks the preferences
 629         if(server->mwindow) server->mwindow->preferences->copy_rates_from(
 630                 server->preferences);
 631 }
 632
 633 int RenderFarmServerThread::set_video_map(unsigned char *buffer)
 634 {
 635         if(server->brender)
 636         {
 637                 server->brender->set_video_map((int64_t)(((u_int32_t)buffer[0]) << 24) |
 638                                                         (((u_int32_t)buffer[1]) << 16) |
 639                                                         (((u_int32_t)buffer[2]) << 8)  |
 640                                                         ((u_int32_t)buffer[3]),
 641                                                         (int64_t)(((u_int32_t)buffer[4]) << 24) |
 642                                                         (((u_int32_t)buffer[5]) << 16) |
 643                                                         (((u_int32_t)buffer[6]) << 8)  |
 644                                                         ((u_int32_t)buffer[7]));
 645                 char return_value[1];
 646                 return_value[0] = 0;
 647                 write_socket(return_value, 1);
 648                 return 0;
 649         }
 650         return 1;
 651 }
 652
 653
 654 void RenderFarmServerThread::set_result(unsigned char *buffer)
 655 {
 656 //printf("RenderFarmServerThread::set_result %p\n", buffer);
 657         if(!*server->result_return)
 658                 *server->result_return = buffer[0];
 659 }
 660
 661
 662 void RenderFarmServerThread::get_result()
 663 {
 664         unsigned char data[1];
 665         data[0] = *server->result_return;
 666         write_socket((char*)data, 1);
 667 }
 668
 669
 670
 671
 672
 673
 674
 675
 676
 677
 678
 679
 680
 681
 682 RenderFarmWatchdog::RenderFarmWatchdog(
 683         RenderFarmServerThread *server,
 684         RenderFarmClientThread *client)
 685  : Thread(1, 0, 0)
 686 {
 687         this->server = server;
 688         this->client = client;
 689         next_request = new Condition(0, "RenderFarmWatchdog::next_request", 0);
 690         request_complete = new Condition(0, "RenderFarmWatchdog::request_complete", 0);
 691         done = 0;
 692 }
 693
 694 RenderFarmWatchdog::~RenderFarmWatchdog()
 695 {
 696         done = 1;
 697         next_request->unlock();
 698         request_complete->unlock();
 699         join();
 700         delete next_request;
 701         delete request_complete;
 702 }
 703
 704 void RenderFarmWatchdog::begin_request()
 705 {
 706         next_request->unlock();
 707 }
 708
 709 void RenderFarmWatchdog::end_request()
 710 {
 711         request_complete->unlock();
 712 }
 713
 714 void RenderFarmWatchdog::run()
 715 {
 716         while(!done)
 717         {
 718                 next_request->lock("RenderFarmWatchdog::run");
 719
 720                 int result = request_complete->timed_lock(RENDERFARM_TIMEOUT * 1000000,
 721                         "RenderFarmWatchdog::run");
 722 //printf("RenderFarmWatchdog::run 1 %d\n", result);
 723
 724                 if(result)
 725                 {
 726                         if(client)
 727                         {
 728                                 printf("RenderFarmWatchdog::run 1 killing client pid %d\n", client->pid);
 729                                 kill(client->pid, SIGKILL);
 730                         }
 731                         else
 732                         if(server)
 733                         {
 734                                 printf("RenderFarmWatchdog::run 1 killing server thread %p\n", server);
 735                                 server->cancel();
 736                                 unsigned char buffer[4];
 737                                 buffer[0] = 1;
 738                                 server->set_result(buffer);
 739                         }
 740
 741                         done = 1;
 742                 }
 743         }
 744 }
 745
 746
 747
 748