4 * Copyright (C) 2008 Adam Williams <broadcast at earthling dot net>
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.
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.
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
23 #include "bcsignals.h"
26 #include "condition.h"
29 #include "filesystem.h"
34 #include "packagedispatcher.h"
35 #include "preferences.h"
37 #include "renderfarm.h"
38 #include "renderfarmclient.h"
40 #include "transportque.h"
43 #include <arpa/inet.h>
47 #include <netinet/in.h>
51 #include <sys/socket.h>
52 #include <sys/types.h>
58 #define AF_FILE AF_LOCAL
62 RenderFarmServer::RenderFarmServer(
64 PackageDispatcher *packages,
65 Preferences *preferences,
68 int64_t *total_return,
69 Mutex *total_return_lock,
74 this->mwindow = mwindow;
75 this->packages = packages;
76 this->preferences = preferences;
77 this->use_local_rate = use_local_rate;
78 this->result_return = result_return;
79 this->total_return = total_return;
80 this->total_return_lock = total_return_lock;
81 this->default_asset = default_asset;
83 this->brender = brender;
84 client_lock = new Mutex("RenderFarmServer::client_lock");
87 RenderFarmServer::~RenderFarmServer()
89 clients.remove_all_objects();
93 // Open connections to clients.
94 int RenderFarmServer::start_clients()
98 for( int i=0; i<preferences->get_enabled_nodes() && !result; ++i ) {
99 client_lock->lock("RenderFarmServer::start_clients");
100 RenderFarmServerThread *client = new RenderFarmServerThread(this, i);
101 clients.append(client);
103 result = client->start_loop();
104 client_lock->unlock();
110 // The render farm must wait for all the clients to finish.
111 int RenderFarmServer::wait_clients()
113 //printf("RenderFarmServer::wait_clients 1\n");
114 clients.remove_all_objects();
115 //printf("RenderFarmServer::wait_clients 2\n");
119 int RenderFarmServer::active_clients()
122 for( int i=0; i<clients.size(); ++i )
123 if( clients[i]->running() ) ++n;
127 // Waits for requests from every client.
128 // Joins when the client is finished.
129 RenderFarmServerThread::RenderFarmServerThread(RenderFarmServer *server,
133 this->server = server;
134 this->number = number;
136 frames_per_second = 0;
140 Thread::set_synchronous(1);
145 RenderFarmServerThread::~RenderFarmServerThread()
148 if( socket_fd >= 0 ) close(socket_fd);
155 int RenderFarmServerThread::open_client(const char *hostname, int port)
160 // Open file for master node
161 if( hostname[0] == '/' ) {
162 if( (socket_fd = socket(PF_UNIX, SOCK_STREAM, 0)) < 0 ) {
163 perror(_("RenderFarmServerThread::open_client: socket\n"));
167 struct sockaddr_un addr;
168 addr.sun_family = AF_FILE;
169 strcpy(addr.sun_path, hostname);
170 int size = (offsetof(struct sockaddr_un, sun_path) +
171 strlen(hostname) + 1);
173 // The master node is always created by BRender. Keep trying for 30 seconds.
175 #define ATTEMPT_DELAY 100000
181 if( connect(socket_fd, (struct sockaddr*)&addr, size) < 0 ) {
183 if( attempt > 30000000 / ATTEMPT_DELAY ) {
184 fprintf(stderr, _("RenderFarmServerThread::open_client: %s: %s\n"),
190 usleep(ATTEMPT_DELAY);
194 }while(!result && !done);
200 if( (socket_fd = socket(PF_INET, SOCK_STREAM, 0)) < 0 ) {
201 perror(_("RenderFarmServerThread::start_loop: socket"));
206 struct sockaddr_in addr;
207 struct hostent *hostinfo;
208 addr.sin_family = AF_INET;
209 addr.sin_port = htons(port);
210 hostinfo = gethostbyname(hostname);
211 if( hostinfo == NULL ) {
212 fprintf(stderr, _("RenderFarmServerThread::open_client: unknown host %s.\n"),
217 addr.sin_addr = *(struct in_addr *) hostinfo->h_addr;
219 if( connect(socket_fd, (struct sockaddr*)&addr, sizeof(addr)) < 0 ) {
220 fprintf(stderr, _("RenderFarmServerThread::open_client: %s: %s\n"),
229 if( result ) socket_fd = -1;
234 int RenderFarmServerThread::start_loop()
237 socket_fd = open_client(server->preferences->get_node_hostname(number),
238 server->preferences->get_node_port(number));
239 if( socket_fd < 0 ) result = 1;
242 int watchdog_timeout = server->preferences->renderfarm_watchdog_timeout;
243 if( watchdog_timeout > 0 ) {
244 watchdog = new RenderFarmWatchdog(watchdog_timeout, this, 0);
249 if( !result ) Thread::start();
264 int64_t RenderFarmServerThread::read_int64(int *error)
267 if( !error ) error = &temp;
269 unsigned char data[sizeof(int64_t)];
270 *error = (read_socket((char*)data, sizeof(int64_t)) !=
273 // Make it return 1 if error so it can be used to read a result code from the
277 result = (((int64_t)data[0]) << 56) |
278 (((uint64_t)data[1]) << 48) |
279 (((uint64_t)data[2]) << 40) |
280 (((uint64_t)data[3]) << 32) |
281 (((uint64_t)data[4]) << 24) |
282 (((uint64_t)data[5]) << 16) |
283 (((uint64_t)data[6]) << 8) |
289 int RenderFarmServerThread::write_int64(int64_t value)
291 unsigned char data[sizeof(int64_t)];
292 data[0] = (value >> 56) & 0xff;
293 data[1] = (value >> 48) & 0xff;
294 data[2] = (value >> 40) & 0xff;
295 data[3] = (value >> 32) & 0xff;
296 data[4] = (value >> 24) & 0xff;
297 data[5] = (value >> 16) & 0xff;
298 data[6] = (value >> 8) & 0xff;
299 data[7] = value & 0xff;
300 return (write_socket((char*)data, sizeof(int64_t)) !=
306 int RenderFarmServerThread::read_socket(char *data, int len)
310 //printf("RenderFarmServerThread::read_socket 1\n");
312 watchdog->begin_request();
313 while(len > 0 && bytes_read >= 0)
316 bytes_read = read(socket_fd, data + offset, len);
318 if( bytes_read > 0 ) {
320 offset += bytes_read;
327 watchdog->end_request();
328 //printf("RenderFarmServerThread::read_socket 10\n");
333 int RenderFarmServerThread::write_socket(char *data, int len)
335 //printf("RenderFarmServerThread::write_socket 1\n");
336 int result = write(socket_fd, data, len);
337 //printf("RenderFarmServerThread::write_socket 10\n");
342 void RenderFarmServerThread::reallocate_buffer(int size)
344 if( buffer && buffer_allocated < size ) {
349 if( !buffer && size ) {
350 buffer = new unsigned char[size];
351 buffer_allocated = size;
355 void RenderFarmServerThread::run()
358 unsigned char header[5];
364 buffer_allocated = 0;
365 // fs_server = new RenderFarmFSServer(this);
366 // fs_server->initialize();
369 // Send command to run package renderer.
370 write_int64(RENDERFARM_PACKAGES);
376 // Wait for requests.
377 // Requests consist of request ID's and accompanying buffers.
379 bytes_read = read_socket((char*)header, 5);
380 //printf("RenderFarmServerThread::run 1\n");
381 if( bytes_read != 5 ) {
386 int request_id = header[0];
387 int64_t request_size = (((u_int32_t)header[1]) << 24) |
388 (((u_int32_t)header[2]) << 16) |
389 (((u_int32_t)header[3]) << 8) |
390 (u_int32_t)header[4];
392 reallocate_buffer(request_size);
394 // Get accompanying buffer
395 bytes_read = read_socket((char*)buffer, request_size);
397 //printf("RenderFarmServerThread::run 2 %d %jd %d\n", request_id, request_size, bytes_read);
398 if( bytes_read != request_size ) {
402 //printf("RenderFarmServerThread::run 3\n");
404 switch( request_id ) {
405 case RENDERFARM_PREFERENCES:
409 case RENDERFARM_ASSET:
417 case RENDERFARM_PACKAGE:
418 send_package(buffer);
421 case RENDERFARM_PROGRESS:
422 set_progress(buffer);
425 case RENDERFARM_SET_RESULT:
429 case RENDERFARM_SET_VMAP:
430 set_video_map(buffer);
433 case RENDERFARM_GET_RESULT:
437 case RENDERFARM_DONE:
438 //printf("RenderFarmServerThread::run 10\n");
442 case RENDERFARM_KEEPALIVE:
446 // if( fs_server->handle_request(request_id, request_size, (unsigned char*)buffer) ) break;
447 printf(_("RenderFarmServerThread::run: unknown request %02x\n"), request_id);
450 //printf("RenderFarmServerThread::run 10 %d %jd\n", request_id, request_size);
452 //printf("RenderFarmServerThread::run 20\n");
454 // Don't let watchdog kill the entire renderfarm when a client finishes normally.
455 delete watchdog; watchdog = 0;
459 void RenderFarmServerThread::write_string(char *string)
464 len = strlen(string) + 1;
465 datagram = new char[len + 4];
467 memcpy(datagram + i, string, len);
468 write_socket((char*)datagram, len + 4);
469 //printf("RenderFarmServerThread::write_string %02x%02x%02x%02x\n",
470 // datagram[0], datagram[1], datagram[2], datagram[3]);
476 void RenderFarmServerThread::send_preferences()
481 server->preferences->save_defaults(&defaults);
482 defaults.save_string(string);
483 write_string(string);
487 void RenderFarmServerThread::send_asset()
492 // The asset must be sent in two segments.
493 // One segment is stored in the EDL and contains decoding information.
494 // One segment is stored in the asset and contains encoding information.
495 server->default_asset->save_defaults(&defaults, 0, 1, 1, 1, 1, 1);
496 defaults.save_string(string1);
498 server->default_asset->write(&file, 0, 0);
499 file.terminate_string();
501 write_string(string1);
502 write_string(file.string());
507 void RenderFarmServerThread::send_edl()
512 server->edl->save_xml(&file, 0);
513 file.terminate_string();
514 //printf("RenderFarmServerThread::send_edl\n%s\n\n", file.string);
516 write_string(file.string());
517 //printf("RenderFarmServerThread::send_edl 2\n");
521 void RenderFarmServerThread::send_package(unsigned char *buffer)
523 this->frames_per_second = (double)((((u_int32_t)buffer[0]) << 24) |
524 (((u_int32_t)buffer[1]) << 16) |
525 (((u_int32_t)buffer[2]) << 8) |
526 ((u_int32_t)buffer[3])) /
529 //printf("RenderFarmServerThread::send_package 1 %f\n", frames_per_second);
530 RenderPackage *package =
531 server->packages->get_package(frames_per_second,
533 server->use_local_rate);
535 //printf("RenderFarmServerThread::send_package 2\n");
536 datagram = new char[BCTEXTLEN];
540 //printf("RenderFarmServerThread::send_package 1\n");
541 datagram[0] = datagram[1] = datagram[2] = datagram[3] = 0;
542 write_socket(datagram, 4);
547 //printf("RenderFarmServerThread::send_package 10\n");
549 strcpy(&datagram[i], package->path);
550 i += strlen(package->path);
553 STORE_INT32(package->audio_start);
554 STORE_INT32(package->audio_end);
555 STORE_INT32(package->video_start);
556 STORE_INT32(package->video_end);
557 int use_brender = (server->brender ? 1 : 0);
558 STORE_INT32(use_brender);
559 STORE_INT32(package->audio_do);
560 STORE_INT32(package->video_do);
564 STORE_INT32(len - 4);
566 write_socket(datagram, len);
573 void RenderFarmServerThread::set_progress(unsigned char *buffer)
575 server->total_return_lock->lock("RenderFarmServerThread::set_progress");
576 *server->total_return += (int64_t)(((u_int32_t)buffer[0]) << 24) |
577 (((u_int32_t)buffer[1]) << 16) |
578 (((u_int32_t)buffer[2]) << 8) |
579 ((u_int32_t)buffer[3]);
580 frames_per_second = (double)((((u_int32_t)buffer[4]) << 24) |
581 (((u_int32_t)buffer[5]) << 16) |
582 (((u_int32_t)buffer[6]) << 8) |
583 ((u_int32_t)buffer[7])) /
585 server->total_return_lock->unlock();
587 server->preferences->set_rate(frames_per_second, number);
589 // This locks the preferences
590 if( server->mwindow ) server->mwindow->preferences->copy_rates_from(
591 server->preferences);
594 int RenderFarmServerThread::set_video_map(unsigned char *buffer)
596 if( server->brender ) {
597 server->brender->set_video_map((int64_t)(((u_int32_t)buffer[0]) << 24) |
598 (((u_int32_t)buffer[1]) << 16) |
599 (((u_int32_t)buffer[2]) << 8) |
600 ((u_int32_t)buffer[3]),
601 (int64_t)(((u_int32_t)buffer[4]) << 24) |
602 (((u_int32_t)buffer[5]) << 16) |
603 (((u_int32_t)buffer[6]) << 8) |
604 ((u_int32_t)buffer[7]));
605 char return_value[1];
607 write_socket(return_value, 1);
614 void RenderFarmServerThread::set_result(unsigned char *buffer)
616 //printf("RenderFarmServerThread::set_result %p\n", buffer);
617 if( !*server->result_return )
618 *server->result_return = buffer[0];
622 void RenderFarmServerThread::get_result()
624 unsigned char data[1];
625 data[0] = *server->result_return;
626 write_socket((char*)data, 1);
642 RenderFarmWatchdog::RenderFarmWatchdog(int timeout_secs,
643 RenderFarmServerThread *server,
644 RenderFarmClientThread *client)
647 this->timeout_usecs = timeout_secs * 1000000;
648 this->server = server;
649 this->client = client;
650 next_request = new Condition(0, "RenderFarmWatchdog::next_request", 0);
651 request_complete = new Condition(0, "RenderFarmWatchdog::request_complete", 0);
655 RenderFarmWatchdog::~RenderFarmWatchdog()
658 next_request->unlock();
659 request_complete->unlock();
662 delete request_complete;
665 void RenderFarmWatchdog::begin_request()
667 next_request->unlock();
670 void RenderFarmWatchdog::end_request()
672 request_complete->unlock();
675 void RenderFarmWatchdog::run()
678 next_request->lock("RenderFarmWatchdog::run");
679 int result = request_complete->timed_lock(timeout_usecs, "RenderFarmWatchdog::run");
680 //printf("RenderFarmWatchdog::run 1 %d\n", result);
684 printf("RenderFarmWatchdog::run 1 killing client pid %d\n", client->pid);
685 kill(client->pid, SIGKILL);
688 printf("RenderFarmWatchdog::run 1 killing server thread %p\n", server);
690 unsigned char buffer[4];
692 server->set_result(buffer);