cinelerra/renderfarmclient.C

   1 #include "asset.h"
   2 #include "assets.h"
   3 #include "clip.h"
   4 #include "bchash.h"
   5 #include "dvbtune.h"
   6 #include "edl.h"
   7 #include "filesystem.h"
   8 #include "filexml.h"
   9 #include "language.h"
  10 #include "mutex.h"
  11 #include "mwindow.h"
  12 #include "pluginserver.h"
  13 #include "preferences.h"
  14 #include "renderfarm.h"
  15 #include "renderfarmclient.h"
  16 //#include "renderfarmfsclient.h"
  17 #include "sighandler.h"
  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 <sys/wait.h>
  30 #include <unistd.h>
  31
  32
  33
  34
  35 // The render client waits for connections from the server.
  36 // Then it starts a thread for each connection.
  37 RenderFarmClient::RenderFarmClient(int port,
  38         char *deamon_path,
  39         int nice_value,
  40         char *config_path)
  41 {
  42         this->port = port;
  43         this->deamon_path = deamon_path;
  44         SigHandler *signals = new SigHandler;
  45         signals->initialize();
  46
  47         this_pid = getpid();
  48         nice(nice_value);
  49
  50
  51         MWindow::init_defaults(boot_defaults, config_path);
  52         boot_preferences = new Preferences;
  53         boot_preferences->load_defaults(boot_defaults);
  54         MWindow::init_plugins(boot_preferences, plugindb, 0);
  55 }
  56
  57
  58
  59
  60 RenderFarmClient::~RenderFarmClient()
  61 {
  62 //      delete thread;
  63         delete boot_defaults;
  64         delete boot_preferences;
  65         plugindb->remove_all_objects();
  66         delete plugindb;
  67 }
  68
  69
  70 void RenderFarmClient::main_loop()
  71 {
  72         int socket_fd;
  73
  74
  75
  76 // Open listening port
  77
  78         if(!deamon_path)
  79         {
  80                 struct sockaddr_in addr;
  81
  82                 addr.sin_family = AF_INET;
  83                 addr.sin_port = htons((unsigned short)port);
  84                 addr.sin_addr.s_addr = htonl(INADDR_ANY);
  85
  86                 if((socket_fd = socket(PF_INET, SOCK_STREAM, 0)) < 0)
  87                 {
  88                         perror(_("RenderFarmClient::main_loop: socket"));
  89                         return;
  90                 }
  91
  92                 if(bind(socket_fd,
  93                         (struct sockaddr*)&addr,
  94                         sizeof(addr)) < 0)
  95                 {
  96                         fprintf(stderr,
  97                                 _("RenderFarmClient::main_loop: bind port %d: %s"),
  98                                 port,
  99                                 strerror(errno));
 100                         return;
 101                 }
 102         }
 103         else
 104         {
 105                 struct sockaddr_un addr;
 106                 addr.sun_family = AF_FILE;
 107                 strcpy(addr.sun_path, deamon_path);
 108                 int size = (offsetof(struct sockaddr_un, sun_path) +
 109                         strlen(deamon_path) + 1);
 110
 111                 if((socket_fd = socket(PF_UNIX, SOCK_STREAM, 0)) < 0)
 112                 {
 113                         perror(_("RenderFarmClient::main_loop: socket"));
 114                         return;
 115                 }
 116
 117                 if(bind(socket_fd,
 118                         (struct sockaddr*)&addr,
 119                         size) < 0)
 120                 {
 121                         fprintf(stderr,
 122                                 _("RenderFarmClient::main_loop: bind path %s: %s\n"),
 123                                 deamon_path,
 124                                 strerror(errno));
 125                         return;
 126                 }
 127         }
 128
 129 // Wait for connections
 130         printf("RenderFarmClient::main_loop: client started\n");
 131         while(1)
 132         {
 133                 if(listen(socket_fd, 256) < 0)
 134         {
 135                 perror(_("RenderFarmClient::main_loop: listen"));
 136                 return;
 137         }
 138
 139                 int new_socket_fd;
 140
 141
 142
 143                 if(!deamon_path)
 144                 {
 145                         struct sockaddr_in clientname;
 146                         socklen_t size = sizeof(clientname);
 147                 if((new_socket_fd = accept(socket_fd,
 148                                 (struct sockaddr*)&clientname,
 149                                                 &size)) < 0)
 150                 {
 151                         perror(_("RenderFarmClient::main_loop: accept"));
 152                         return;
 153                 }
 154                         else
 155                         {
 156 //printf("RenderFarmClient::main_loop: Session started from %s\n", inet_ntoa(clientname.sin_addr));
 157                                 RenderFarmClientThread *thread =
 158                                         new RenderFarmClientThread(this);
 159                                 thread->main_loop(new_socket_fd);
 160                         }
 161                 }
 162                 else
 163                 {
 164                         struct sockaddr_un clientname;
 165                         socklen_t size = sizeof(clientname);
 166                 if((new_socket_fd = accept(socket_fd,
 167                                 (struct sockaddr*)&clientname,
 168                                                 &size)) < 0)
 169                 {
 170                         perror(_("RenderFarmClient::main_loop: accept"));
 171                         return;
 172                 }
 173                         else
 174                         {
 175 //printf("RenderFarmClient::main_loop: Session started from %s\n", clientname.sun_path);
 176                                 RenderFarmClientThread *thread =
 177                                         new RenderFarmClientThread(this);
 178                                 thread->main_loop(new_socket_fd);
 179                         }
 180                 }
 181         }
 182 }
 183
 184 void RenderFarmClient::kill_client()
 185 {
 186 printf("RenderFarmClient::kill_client 1\n");
 187         if(deamon_path)
 188         {
 189 printf("RenderFarmClient::kill_client 2\n");
 190                 remove(deamon_path);
 191                 kill(this_pid, SIGKILL);
 192         }
 193 }
 194
 195
 196
 197
 198
 199
 200
 201
 202
 203
 204
 205 // The thread requests jobs from the server until the job table is empty
 206 // or the server reports an error.  This thread must poll the server
 207 // after every frame for the error status.
 208 // Detaches when finished.
 209 RenderFarmClientThread::RenderFarmClientThread(RenderFarmClient *client)
 210  : Thread(0, 0, 1)
 211 {
 212         this->client = client;
 213         frames_per_second = 0;
 214         Thread::set_synchronous(0);
 215 //      fs_client = 0;
 216         mutex_lock = new Mutex("RenderFarmClientThread::mutex_lock");
 217         watchdog = 0;
 218         keep_alive = 0;
 219 }
 220
 221 RenderFarmClientThread::~RenderFarmClientThread()
 222 {
 223 //      if(fs_client) delete fs_client;
 224         delete mutex_lock;
 225         delete watchdog;
 226         delete keep_alive;
 227 }
 228
 229
 230 int RenderFarmClientThread::send_request_header(int request,
 231         int len)
 232 {
 233         unsigned char datagram[5];
 234         datagram[0] = request;
 235
 236         int i = 1;
 237         STORE_INT32(len);
 238 // printf("RenderFarmClientThread::send_request_header %d %02x%02x%02x%02x%02x\n",
 239 // request, datagram[0], datagram[1], datagram[2], datagram[3], datagram[4]);
 240
 241         return (write_socket((char*)datagram, 5) != 5);
 242 }
 243
 244 int RenderFarmClientThread::write_socket(char *data, int len)
 245 {
 246         return write(socket_fd, data, len);
 247 }
 248
 249 int RenderFarmClientThread::read_socket(char *data, int len)
 250 {
 251         int bytes_read = 0;
 252         int offset = 0;
 253         watchdog->begin_request();
 254         while(len > 0 && bytes_read >= 0)
 255         {
 256                 bytes_read = read(socket_fd, data + offset, len);
 257                 if(bytes_read > 0)
 258                 {
 259                         len -= bytes_read;
 260                         offset += bytes_read;
 261                 }
 262                 else
 263                 if(bytes_read < 0)
 264                 {
 265                         break;
 266                 }
 267         }
 268         watchdog->end_request();
 269
 270         return offset;
 271 }
 272
 273 int RenderFarmClientThread::write_int64(int64_t value)
 274 {
 275         unsigned char data[sizeof(int64_t)];
 276         data[0] = (value >> 56) & 0xff;
 277         data[1] = (value >> 48) & 0xff;
 278         data[2] = (value >> 40) & 0xff;
 279         data[3] = (value >> 32) & 0xff;
 280         data[4] = (value >> 24) & 0xff;
 281         data[5] = (value >> 16) & 0xff;
 282         data[6] = (value >> 8) & 0xff;
 283         data[7] = value & 0xff;
 284         return (write_socket((char*)data, sizeof(int64_t)) != sizeof(int64_t));
 285 }
 286
 287 int64_t RenderFarmClientThread::read_int64(int *error)
 288 {
 289         int temp = 0;
 290         if(!error) error = &temp;
 291
 292         unsigned char data[sizeof(int64_t)];
 293         *error = (read_socket((char*)data, sizeof(int64_t)) != sizeof(int64_t));
 294
 295 // Make it return 1 if error so it can be used to read a result code from the
 296 // server.
 297         int64_t result = 1;
 298         if(!*error)
 299         {
 300                 result = (((int64_t)data[0]) << 56) |
 301                         (((uint64_t)data[1]) << 48) |
 302                         (((uint64_t)data[2]) << 40) |
 303                         (((uint64_t)data[3]) << 32) |
 304                         (((uint64_t)data[4]) << 24) |
 305                         (((uint64_t)data[5]) << 16) |
 306                         (((uint64_t)data[6]) << 8)  |
 307                         data[7];
 308         }
 309         return result;
 310 }
 311
 312 void RenderFarmClientThread::read_string(char* &string)
 313 {
 314         unsigned char header[4];
 315         if(read_socket((char*)header, 4) != 4)
 316         {
 317                 string = 0;
 318                 return;
 319         }
 320
 321         int64_t len = (((u_int32_t)header[0]) << 24) |
 322                                 (((u_int32_t)header[1]) << 16) |
 323                                 (((u_int32_t)header[2]) << 8) |
 324                                 ((u_int32_t)header[3]);
 325
 326         if(len)
 327         {
 328                 string = new char[len];
 329                 if(read_socket(string, len) != len)
 330                 {
 331                         delete [] string;
 332                         string = 0;
 333                 }
 334         }
 335         else
 336                 string = 0;
 337
 338 }
 339
 340 void RenderFarmClientThread::abort()
 341 {
 342         send_completion(socket_fd);
 343         close(socket_fd);
 344         exit(1);
 345 }
 346
 347 void RenderFarmClientThread::lock(char *location)
 348 {
 349         mutex_lock->lock(location);
 350 }
 351
 352 void RenderFarmClientThread::unlock()
 353 {
 354         mutex_lock->unlock();
 355 }
 356
 357 void RenderFarmClientThread::get_command(int socket_fd, int *command)
 358 {
 359         unsigned char data[4];
 360         int error;
 361         *command = read_int64(&error);
 362         if(error)
 363         {
 364                 *command = 0;
 365                 return;
 366         }
 367 }
 368
 369
 370 void RenderFarmClientThread::read_preferences(int socket_fd,
 371         Preferences *preferences)
 372 {
 373         lock("RenderFarmClientThread::read_preferences");
 374         send_request_header(RENDERFARM_PREFERENCES,
 375                 0);
 376
 377         char *string;
 378         read_string(string);
 379
 380         BC_Hash defaults;
 381         defaults.load_string((char*)string);
 382         preferences->load_defaults(&defaults);
 383
 384         delete [] string;
 385         unlock();
 386 }
 387
 388
 389
 390 void RenderFarmClientThread::read_asset(int socket_fd, Asset_GC asset)
 391 {
 392         lock("RenderFarmClientThread::read_asset");
 393         send_request_header(RENDERFARM_ASSET,
 394                 0);
 395
 396         char *string1;
 397         char *string2;
 398         read_string(string1);
 399         read_string(string2);
 400
 401
 402
 403         FileXML file;
 404         file.read_from_string((char*)string2);
 405         asset->read(&file);
 406
 407
 408
 409         BC_Hash defaults;
 410         defaults.load_string((char*)string1);
 411         asset->load_defaults(&defaults,
 412                 0,
 413                 1,
 414                 1,
 415                 1,
 416                 1,
 417                 1);
 418
 419         delete [] string1;
 420         delete [] string2;
 421         unlock();
 422 }
 423
 424 void RenderFarmClientThread::read_edl(int socket_fd,
 425         EDL *edl,
 426         Preferences *preferences)
 427 {
 428         lock("RenderFarmClientThread::read_edl");
 429         send_request_header(RENDERFARM_EDL,
 430                 0);
 431
 432         char *string;
 433         read_string(string);
 434
 435
 436         FileXML file;
 437         file.read_from_string((char*)string);
 438         delete [] string;
 439
 440
 441
 442
 443
 444
 445
 446
 447         edl->load_xml(client->plugindb,
 448                 &file,
 449                 LOAD_ALL);
 450
 451
 452         unlock();
 453 }
 454
 455 int RenderFarmClientThread::read_package(int socket_fd, RenderPackage *package)
 456 {
 457         lock("RenderFarmClientThread::read_package");
 458         send_request_header(RENDERFARM_PACKAGE,
 459                 4);
 460
 461         unsigned char datagram[5];
 462         int i = 0;
 463
 464
 465 // Fails if -ieee isn't set.
 466         int64_t fixed = !EQUIV(frames_per_second, 0.0) ?
 467                 (int64_t)(frames_per_second * 65536.0) : 0;
 468         STORE_INT32(fixed);
 469         write_socket((char*)datagram, 4);
 470
 471
 472 //printf("RenderFarmClientThread::read_package 1 %f %ld\n", frames_per_second, fixed);
 473         char *data;
 474         unsigned char *data_ptr;
 475         read_string(data);
 476 //printf("RenderFarmClientThread::read_package 2 %p\n", data);
 477 // Signifies end of session.
 478         if(!data)
 479         {
 480 //              printf(_("RenderFarmClientThread::read_package no output path recieved.\n"));
 481                 unlock();
 482                 return 1;
 483         }
 484
 485 //printf("RenderFarmClientThread::read_package 2\n");
 486
 487
 488         data_ptr = (unsigned char*)data;
 489         strcpy(package->path, data);
 490         data_ptr += strlen(package->path);
 491         data_ptr++;
 492         package->audio_start = READ_INT32(data_ptr);
 493         data_ptr += 4;
 494         package->audio_end = READ_INT32(data_ptr);
 495         data_ptr += 4;
 496         package->video_start = READ_INT32(data_ptr);
 497         data_ptr += 4;
 498         package->video_end = READ_INT32(data_ptr);
 499         data_ptr += 4;
 500         package->use_brender = READ_INT32(data_ptr);
 501         data_ptr += 4;
 502         package->audio_do = READ_INT32(data_ptr);
 503         data_ptr += 4;
 504         package->video_do = READ_INT32(data_ptr);
 505
 506         delete [] data;
 507         unlock();
 508
 509         return 0;
 510 }
 511
 512 int RenderFarmClientThread::send_completion(int socket_fd)
 513 {
 514         lock("RenderFarmClientThread::send_completion");
 515         int result = send_request_header(RENDERFARM_DONE, 0);
 516         unlock();
 517         return result;
 518 }
 519
 520
 521 void RenderFarmClientThread::ping_server()
 522 {
 523         lock("RenderFarmClientThread::ping_server");
 524         send_request_header(RENDERFARM_KEEPALIVE, 0);
 525         unlock();
 526 }
 527
 528
 529
 530 void RenderFarmClientThread::main_loop(int socket_fd)
 531 {
 532         this->socket_fd = socket_fd;
 533
 534         Thread::start();
 535 }
 536
 537 void RenderFarmClientThread::run()
 538 {
 539 // Create new memory space
 540         pid = fork();
 541         if(pid != 0)
 542         {
 543                 int return_value;
 544                 waitpid(pid, &return_value, 0);
 545                 return;
 546         }
 547
 548 // Get the pid of the fork if inside the fork
 549         pid = getpid();
 550
 551
 552
 553         int socket_fd = this->socket_fd;
 554
 555         init_client_keepalive();
 556
 557 // Get command to run
 558         int command;
 559 SET_TRACE
 560         lock("RenderFarmClientThread::run");
 561 SET_TRACE
 562         get_command(socket_fd, &command);
 563 SET_TRACE
 564         unlock();
 565
 566 //printf("RenderFarmClientThread::run command=%d\n", command);
 567
 568 SET_TRACE
 569         switch(command)
 570         {
 571                 case RENDERFARM_TUNER:
 572                         do_tuner(socket_fd);
 573                         break;
 574                 case RENDERFARM_PACKAGES:
 575                         do_packages(socket_fd);
 576                         break;
 577         }
 578
 579         _exit(0);
 580 }
 581
 582
 583 void RenderFarmClientThread::init_client_keepalive()
 584 {
 585         keep_alive = new RenderFarmKeepalive(this);
 586         keep_alive->start();
 587         watchdog = new RenderFarmWatchdog(0, this);
 588         watchdog->start();
 589 }
 590
 591
 592
 593 void RenderFarmClientThread::do_tuner(int socket_fd)
 594 {
 595 // Currently only 1 tuner driver.  Maybe more someday.
 596         DVBTune server(this);
 597         server.main_loop();
 598         ::close(socket_fd);
 599 }
 600
 601
 602 void RenderFarmClientThread::do_packages(int socket_fd)
 603 {
 604
 605         EDL *edl;
 606         RenderPackage *package;
 607         Asset_GC default_asset;
 608         Preferences *preferences;
 609
 610
 611
 612         FarmPackageRenderer package_renderer(this, socket_fd);
 613         int result = 0;
 614
 615
 616
 617 //printf("RenderFarmClientThread::run 2\n");
 618 // Read settings
 619         preferences = new Preferences;
 620         default_asset = Asset_GC(new Asset);
 621         package = new RenderPackage;
 622         edl = new EDL;
 623         edl->create_objects();
 624
 625 //printf("RenderFarmClientThread::run 3\n");
 626
 627
 628
 629
 630
 631
 632
 633         read_preferences(socket_fd, preferences);
 634 //printf("RenderFarmClientThread::run 3\n");
 635         read_asset(socket_fd, default_asset);
 636 //printf("RenderFarmClientThread::run 3\n");
 637         read_edl(socket_fd, edl, preferences);
 638 //edl->dump();
 639
 640
 641
 642
 643
 644
 645
 646 //printf("RenderFarmClientThread::run 4\n");
 647
 648         package_renderer.initialize(0,
 649                         edl,
 650                         preferences,
 651                         default_asset,
 652                         client->plugindb);
 653 //printf("RenderFarmClientThread::run 5\n");
 654
 655 // Read packages
 656         while(1)
 657         {
 658                 result = read_package(socket_fd, package);
 659 //printf("RenderFarmClientThread::run 6 %d\n", result);
 660
 661
 662 // Finished list
 663                 if(result)
 664                 {
 665 //printf("RenderFarmClientThread::run 7\n");
 666
 667                         result = send_completion(socket_fd);
 668                         break;
 669                 }
 670
 671                 Timer timer;
 672                 timer.update();
 673
 674 // Error
 675                 if(package_renderer.render_package(package))
 676                 {
 677 //printf("RenderFarmClientThread::run 8\n");
 678                         result = send_completion(socket_fd);
 679                         break;
 680                 }
 681
 682                 frames_per_second = (double)(package->video_end - package->video_start) /
 683                         ((double)timer.get_difference() / 1000);
 684
 685 //printf("RenderFarmClientThread::run 9\n");
 686
 687
 688
 689         }
 690
 691
 692 //printf("RenderFarmClientThread::run 9\n");
 693 //printf("RenderFarmClientThread::run 10\n");
 694         delete edl;
 695 //printf("RenderFarmClientThread::run 11\n");
 696         delete preferences;
 697 printf(_("RenderFarmClientThread::run: Session finished.\n"));
 698 }
 699
 700
 701
 702
 703
 704
 705
 706
 707
 708 RenderFarmKeepalive::RenderFarmKeepalive(
 709         RenderFarmClientThread *client_thread)
 710  : Thread(1, 0, 0)
 711 {
 712         this->client_thread = client_thread;
 713         done = 0;
 714 }
 715
 716 RenderFarmKeepalive::~RenderFarmKeepalive()
 717 {
 718         done = 1;
 719         cancel();
 720         join();
 721 }
 722
 723
 724 void RenderFarmKeepalive::run()
 725 {
 726         while(!done)
 727         {
 728                 enable_cancel();
 729                 sleep(5);
 730                 disable_cancel();
 731                 if(!done)
 732                 {
 733 // watchdog thread kills this if it gets stuck
 734                         client_thread->ping_server();
 735                 }
 736         }
 737 }
 738
 739
 740
 741
 742
 743
 744
 745
 746
 747
 748
 749
 750
 751
 752
 753 FarmPackageRenderer::FarmPackageRenderer(RenderFarmClientThread *thread,
 754                 int socket_fd)
 755  : PackageRenderer()
 756 {
 757         this->thread = thread;
 758         this->socket_fd = socket_fd;
 759 }
 760
 761
 762
 763 FarmPackageRenderer::~FarmPackageRenderer()
 764 {
 765 }
 766
 767
 768 int FarmPackageRenderer::get_result()
 769 {
 770         thread->lock("FarmPackageRenderer::get_result");
 771         thread->send_request_header(RENDERFARM_GET_RESULT,
 772                 0);
 773         unsigned char data[1];
 774         data[0] = 1;
 775         if(thread->read_socket((char*)data, 1) != 1)
 776         {
 777                 thread->unlock();
 778                 return 1;
 779         }
 780         thread->unlock();
 781         return data[0];
 782 }
 783
 784 void FarmPackageRenderer::set_result(int value)
 785 {
 786         thread->lock("FarmPackageRenderer::set_result");
 787         thread->send_request_header(RENDERFARM_SET_RESULT,
 788                 1);
 789         unsigned char data[1];
 790         data[0] = value;
 791         thread->write_socket((char*)data, 1);
 792         thread->unlock();
 793 }
 794
 795 void FarmPackageRenderer::set_progress(int64_t total_samples)
 796 {
 797         thread->lock("FarmPackageRenderer::set_progress");
 798         thread->send_request_header(RENDERFARM_PROGRESS,
 799                 4);
 800         unsigned char datagram[4];
 801         int i = 0;
 802         STORE_INT32(total_samples);
 803         thread->write_socket((char*)datagram, 4);
 804         thread->unlock();
 805 }
 806
 807 int FarmPackageRenderer::set_video_map(int64_t position, int value)
 808 {
 809         int result = 0;
 810         unsigned char datagram[8];
 811         char return_value[1];
 812         int i = 0;
 813
 814         thread->lock("FarmPackageRenderer::set_video_map");
 815         thread->send_request_header(RENDERFARM_SET_VMAP,
 816                 8);
 817         STORE_INT32(position);
 818         STORE_INT32(value);
 819         thread->write_socket((char*)datagram, 8);
 820
 821 // Get completion since the GUI may be locked for a long time.
 822         if(!thread->read_socket(return_value, 1))
 823         {
 824                 result = 1;
 825         }
 826
 827         thread->unlock();
 828         return result;
 829 }
 830
 831
 832
 833
 834
 835
 836
 837
 838
 839
 840 /*
 841 //      Local Variables:
 842 //      mode: C++
 843 //      c-file-style: "linux"
 844 //      End:
 845 */