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>
59 RenderFarmServer::RenderFarmServer(
61 PackageDispatcher *packages,
62 Preferences *preferences,
65 int64_t *total_return,
66 Mutex *total_return_lock,
71 this->mwindow = mwindow;
72 this->packages = packages;
73 this->preferences = preferences;
74 this->use_local_rate = use_local_rate;
75 this->result_return = result_return;
76 this->total_return = total_return;
77 this->total_return_lock = total_return_lock;
78 this->default_asset = default_asset;
80 this->brender = brender;
81 client_lock = new Mutex("RenderFarmServer::client_lock");
84 RenderFarmServer::~RenderFarmServer()
86 clients.remove_all_objects();
90 // Open connections to clients.
91 int RenderFarmServer::start_clients()
95 for(int i = 0; i < preferences->get_enabled_nodes() && !result; i++)
97 client_lock->lock("RenderFarmServer::start_clients");
98 RenderFarmServerThread *client = new RenderFarmServerThread(this,
100 clients.append(client);
102 result = client->start_loop();
103 client_lock->unlock();
109 // The render farm must wait for all the clients to finish.
110 int RenderFarmServer::wait_clients()
112 //printf("RenderFarmServer::wait_clients 1\n");
113 clients.remove_all_objects();
114 //printf("RenderFarmServer::wait_clients 2\n");
118 int RenderFarmServer::active_clients()
121 for( int i=0; i<clients.size(); ++i )
122 if( clients[i]->running() ) ++n;
126 // Waits for requests from every client.
127 // Joins when the client is finished.
128 RenderFarmServerThread::RenderFarmServerThread(RenderFarmServer *server,
132 this->server = server;
133 this->number = number;
135 frames_per_second = 0;
139 Thread::set_synchronous(1);
144 RenderFarmServerThread::~RenderFarmServerThread()
146 //printf("RenderFarmServerThread::~RenderFarmServerThread 1 %p\n", this);
148 //printf("RenderFarmServerThread::~RenderFarmServerThread 1\n");
149 if(socket_fd >= 0) close(socket_fd);
150 if(watchdog) delete watchdog;
151 if(buffer) delete [] buffer;
152 if(datagram) delete [] datagram;
153 //printf("RenderFarmServerThread::~RenderFarmServerThread 2\n");
157 int RenderFarmServerThread::open_client(const char *hostname, int port)
162 // Open file for master node
163 if(hostname[0] == '/')
165 if((socket_fd = socket(PF_UNIX, SOCK_STREAM, 0)) < 0)
167 perror(_("RenderFarmServerThread::start_loop: socket\n"));
172 struct sockaddr_un addr;
173 addr.sun_family = AF_FILE;
174 strcpy(addr.sun_path, hostname);
175 int size = (offsetof(struct sockaddr_un, sun_path) +
176 strlen(hostname) + 1);
178 // The master node is always created by BRender. Keep trying for 30 seconds.
180 #define ATTEMPT_DELAY 100000
186 if(connect(socket_fd, (struct sockaddr*)&addr, size) < 0)
189 if(attempt > 30000000 / ATTEMPT_DELAY)
191 fprintf(stderr, _("RenderFarmServerThread::open_client: %s: %s\n"),
197 usleep(ATTEMPT_DELAY);
201 }while(!result && !done);
207 if((socket_fd = socket(PF_INET, SOCK_STREAM, 0)) < 0)
209 perror(_("RenderFarmServerThread::start_loop: socket"));
215 struct sockaddr_in addr;
216 struct hostent *hostinfo;
217 addr.sin_family = AF_INET;
218 addr.sin_port = htons(port);
219 hostinfo = gethostbyname(hostname);
222 fprintf(stderr, _("RenderFarmServerThread::open_client: unknown host %s.\n"),
228 addr.sin_addr = *(struct in_addr *) hostinfo->h_addr;
230 if(connect(socket_fd, (struct sockaddr*)&addr, sizeof(addr)) < 0)
232 fprintf(stderr, _("RenderFarmServerThread::open_client: %s: %s\n"),
241 if(result) socket_fd = -1;
246 int RenderFarmServerThread::start_loop()
250 socket_fd = open_client(server->preferences->get_node_hostname(number),
251 server->preferences->get_node_port(number));
253 if(socket_fd < 0) result = 1;
257 watchdog = new RenderFarmWatchdog(this, 0);
261 if(!result) Thread::start();
276 int64_t RenderFarmServerThread::read_int64(int *error)
279 if(!error) error = &temp;
281 unsigned char data[sizeof(int64_t)];
282 *error = (read_socket((char*)data, sizeof(int64_t)) !=
285 // Make it return 1 if error so it can be used to read a result code from the
290 result = (((int64_t)data[0]) << 56) |
291 (((uint64_t)data[1]) << 48) |
292 (((uint64_t)data[2]) << 40) |
293 (((uint64_t)data[3]) << 32) |
294 (((uint64_t)data[4]) << 24) |
295 (((uint64_t)data[5]) << 16) |
296 (((uint64_t)data[6]) << 8) |
302 int RenderFarmServerThread::write_int64(int64_t value)
304 unsigned char data[sizeof(int64_t)];
305 data[0] = (value >> 56) & 0xff;
306 data[1] = (value >> 48) & 0xff;
307 data[2] = (value >> 40) & 0xff;
308 data[3] = (value >> 32) & 0xff;
309 data[4] = (value >> 24) & 0xff;
310 data[5] = (value >> 16) & 0xff;
311 data[6] = (value >> 8) & 0xff;
312 data[7] = value & 0xff;
313 return (write_socket((char*)data, sizeof(int64_t)) !=
319 int RenderFarmServerThread::read_socket(char *data, int len)
323 //printf("RenderFarmServerThread::read_socket 1\n");
324 watchdog->begin_request();
325 while(len > 0 && bytes_read >= 0)
328 bytes_read = read(socket_fd, data + offset, len);
333 offset += bytes_read;
339 watchdog->end_request();
340 //printf("RenderFarmServerThread::read_socket 10\n");
345 int RenderFarmServerThread::write_socket(char *data, int len)
347 //printf("RenderFarmServerThread::write_socket 1\n");
348 int result = write(socket_fd, data, len);
349 //printf("RenderFarmServerThread::write_socket 10\n");
354 void RenderFarmServerThread::reallocate_buffer(int size)
356 if(buffer && buffer_allocated < size)
364 buffer = new unsigned char[size];
365 buffer_allocated = size;
369 void RenderFarmServerThread::run()
372 unsigned char header[5];
378 buffer_allocated = 0;
379 // fs_server = new RenderFarmFSServer(this);
380 // fs_server->initialize();
383 // Send command to run package renderer.
384 write_int64(RENDERFARM_PACKAGES);
391 // Wait for requests.
392 // Requests consist of request ID's and accompanying buffers.
394 bytes_read = read_socket((char*)header, 5);
395 //printf("RenderFarmServerThread::run 1\n");
402 int request_id = header[0];
403 int64_t request_size = (((u_int32_t)header[1]) << 24) |
404 (((u_int32_t)header[2]) << 16) |
405 (((u_int32_t)header[3]) << 8) |
406 (u_int32_t)header[4];
408 reallocate_buffer(request_size);
410 // Get accompanying buffer
411 bytes_read = read_socket((char*)buffer, request_size);
413 //printf("RenderFarmServerThread::run 2 %d %jd %d\n", request_id, request_size, bytes_read);
414 if(bytes_read != request_size)
419 //printf("RenderFarmServerThread::run 3\n");
423 case RENDERFARM_PREFERENCES:
427 case RENDERFARM_ASSET:
435 case RENDERFARM_PACKAGE:
436 send_package(buffer);
439 case RENDERFARM_PROGRESS:
440 set_progress(buffer);
443 case RENDERFARM_SET_RESULT:
447 case RENDERFARM_SET_VMAP:
448 set_video_map(buffer);
451 case RENDERFARM_GET_RESULT:
455 case RENDERFARM_DONE:
456 //printf("RenderFarmServerThread::run 10\n");
460 case RENDERFARM_KEEPALIVE:
464 // if(!fs_server->handle_request(request_id, request_size, (unsigned char*)buffer))
466 printf(_("RenderFarmServerThread::run: unknown request %02x\n"), request_id);
470 //printf("RenderFarmServerThread::run 10 %d %jd\n", request_id, request_size);
473 // Don't let watchdog kill the entire renderfarm when a client finishes
477 //printf("RenderFarmServerThread::run 20\n");
485 void RenderFarmServerThread::write_string(char *string)
490 len = strlen(string) + 1;
491 datagram = new char[len + 4];
493 memcpy(datagram + i, string, len);
494 write_socket((char*)datagram, len + 4);
495 //printf("RenderFarmServerThread::write_string %02x%02x%02x%02x\n",
496 // datagram[0], datagram[1], datagram[2], datagram[3]);
502 void RenderFarmServerThread::send_preferences()
507 server->preferences->save_defaults(&defaults);
508 defaults.save_string(string);
509 write_string(string);
513 void RenderFarmServerThread::send_asset()
518 // The asset must be sent in two segments.
519 // One segment is stored in the EDL and contains decoding information.
520 // One segment is stored in the asset and contains encoding information.
521 server->default_asset->save_defaults(&defaults,
528 defaults.save_string(string1);
530 server->default_asset->write(&file, 0, 0);
531 file.terminate_string();
533 write_string(string1);
534 write_string(file.string());
539 void RenderFarmServerThread::send_edl()
544 server->edl->save_xml(&file,
548 file.terminate_string();
549 //printf("RenderFarmServerThread::send_edl\n%s\n\n", file.string);
551 write_string(file.string());
552 //printf("RenderFarmServerThread::send_edl 2\n");
556 void RenderFarmServerThread::send_package(unsigned char *buffer)
558 this->frames_per_second = (double)((((u_int32_t)buffer[0]) << 24) |
559 (((u_int32_t)buffer[1]) << 16) |
560 (((u_int32_t)buffer[2]) << 8) |
561 ((u_int32_t)buffer[3])) /
564 //printf("RenderFarmServerThread::send_package 1 %f\n", frames_per_second);
565 RenderPackage *package =
566 server->packages->get_package(frames_per_second,
568 server->use_local_rate);
570 //printf("RenderFarmServerThread::send_package 2\n");
571 datagram = new char[BCTEXTLEN];
576 //printf("RenderFarmServerThread::send_package 1\n");
577 datagram[0] = datagram[1] = datagram[2] = datagram[3] = 0;
578 write_socket(datagram, 4);
583 //printf("RenderFarmServerThread::send_package 10\n");
585 strcpy(&datagram[i], package->path);
586 i += strlen(package->path);
589 STORE_INT32(package->audio_start);
590 STORE_INT32(package->audio_end);
591 STORE_INT32(package->video_start);
592 STORE_INT32(package->video_end);
593 int use_brender = (server->brender ? 1 : 0);
594 STORE_INT32(use_brender);
595 STORE_INT32(package->audio_do);
596 STORE_INT32(package->video_do);
600 STORE_INT32(len - 4);
602 write_socket(datagram, len);
609 void RenderFarmServerThread::set_progress(unsigned char *buffer)
611 server->total_return_lock->lock("RenderFarmServerThread::set_progress");
612 *server->total_return += (int64_t)(((u_int32_t)buffer[0]) << 24) |
613 (((u_int32_t)buffer[1]) << 16) |
614 (((u_int32_t)buffer[2]) << 8) |
615 ((u_int32_t)buffer[3]);
616 frames_per_second = (double)((((u_int32_t)buffer[4]) << 24) |
617 (((u_int32_t)buffer[5]) << 16) |
618 (((u_int32_t)buffer[6]) << 8) |
619 ((u_int32_t)buffer[7])) /
621 server->total_return_lock->unlock();
623 server->preferences->set_rate(frames_per_second, number);
625 // This locks the preferences
626 if(server->mwindow) server->mwindow->preferences->copy_rates_from(
627 server->preferences);
630 int RenderFarmServerThread::set_video_map(unsigned char *buffer)
634 server->brender->set_video_map((int64_t)(((u_int32_t)buffer[0]) << 24) |
635 (((u_int32_t)buffer[1]) << 16) |
636 (((u_int32_t)buffer[2]) << 8) |
637 ((u_int32_t)buffer[3]),
638 (int64_t)(((u_int32_t)buffer[4]) << 24) |
639 (((u_int32_t)buffer[5]) << 16) |
640 (((u_int32_t)buffer[6]) << 8) |
641 ((u_int32_t)buffer[7]));
642 char return_value[1];
644 write_socket(return_value, 1);
651 void RenderFarmServerThread::set_result(unsigned char *buffer)
653 //printf("RenderFarmServerThread::set_result %p\n", buffer);
654 if(!*server->result_return)
655 *server->result_return = buffer[0];
659 void RenderFarmServerThread::get_result()
661 unsigned char data[1];
662 data[0] = *server->result_return;
663 write_socket((char*)data, 1);
679 RenderFarmWatchdog::RenderFarmWatchdog(
680 RenderFarmServerThread *server,
681 RenderFarmClientThread *client)
684 this->server = server;
685 this->client = client;
686 next_request = new Condition(0, "RenderFarmWatchdog::next_request", 0);
687 request_complete = new Condition(0, "RenderFarmWatchdog::request_complete", 0);
691 RenderFarmWatchdog::~RenderFarmWatchdog()
694 next_request->unlock();
695 request_complete->unlock();
698 delete request_complete;
701 void RenderFarmWatchdog::begin_request()
703 next_request->unlock();
706 void RenderFarmWatchdog::end_request()
708 request_complete->unlock();
711 void RenderFarmWatchdog::run()
715 next_request->lock("RenderFarmWatchdog::run");
717 int result = request_complete->timed_lock(RENDERFARM_TIMEOUT * 1000000,
718 "RenderFarmWatchdog::run");
719 //printf("RenderFarmWatchdog::run 1 %d\n", result);
725 printf("RenderFarmWatchdog::run 1 killing client pid %d\n", client->pid);
726 kill(client->pid, SIGKILL);
731 printf("RenderFarmWatchdog::run 1 killing server thread %p\n", server);
733 unsigned char buffer[4];
735 server->set_result(buffer);