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");
119 // Waits for requests from every client.
120 // Joins when the client is finished.
121 RenderFarmServerThread::RenderFarmServerThread(RenderFarmServer *server,
125 this->server = server;
126 this->number = number;
128 frames_per_second = 0;
132 Thread::set_synchronous(1);
137 RenderFarmServerThread::~RenderFarmServerThread()
139 //printf("RenderFarmServerThread::~RenderFarmServerThread 1 %p\n", this);
141 //printf("RenderFarmServerThread::~RenderFarmServerThread 1\n");
142 if(socket_fd >= 0) close(socket_fd);
143 if(watchdog) delete watchdog;
144 if(buffer) delete [] buffer;
145 if(datagram) delete [] datagram;
146 //printf("RenderFarmServerThread::~RenderFarmServerThread 2\n");
150 int RenderFarmServerThread::open_client(const char *hostname, int port)
155 // Open file for master node
156 if(hostname[0] == '/')
158 if((socket_fd = socket(PF_UNIX, SOCK_STREAM, 0)) < 0)
160 perror(_("RenderFarmServerThread::start_loop: socket\n"));
165 struct sockaddr_un addr;
166 addr.sun_family = AF_FILE;
167 strcpy(addr.sun_path, hostname);
168 int size = (offsetof(struct sockaddr_un, sun_path) +
169 strlen(hostname) + 1);
171 // The master node is always created by BRender. Keep trying for 30 seconds.
173 #define ATTEMPT_DELAY 100000
179 if(connect(socket_fd, (struct sockaddr*)&addr, size) < 0)
182 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)
202 perror(_("RenderFarmServerThread::start_loop: socket"));
208 struct sockaddr_in addr;
209 struct hostent *hostinfo;
210 addr.sin_family = AF_INET;
211 addr.sin_port = htons(port);
212 hostinfo = gethostbyname(hostname);
215 fprintf(stderr, _("RenderFarmServerThread::open_client: unknown host %s.\n"),
221 addr.sin_addr = *(struct in_addr *) hostinfo->h_addr;
223 if(connect(socket_fd, (struct sockaddr*)&addr, sizeof(addr)) < 0)
225 fprintf(stderr, _("RenderFarmServerThread::open_client: %s: %s\n"),
234 if(result) socket_fd = -1;
239 int RenderFarmServerThread::start_loop()
243 socket_fd = open_client(server->preferences->get_node_hostname(number),
244 server->preferences->get_node_port(number));
246 if(socket_fd < 0) result = 1;
250 watchdog = new RenderFarmWatchdog(this, 0);
254 if(!result) Thread::start();
269 int64_t RenderFarmServerThread::read_int64(int *error)
272 if(!error) error = &temp;
274 unsigned char data[sizeof(int64_t)];
275 *error = (read_socket((char*)data, sizeof(int64_t)) !=
278 // Make it return 1 if error so it can be used to read a result code from the
283 result = (((int64_t)data[0]) << 56) |
284 (((uint64_t)data[1]) << 48) |
285 (((uint64_t)data[2]) << 40) |
286 (((uint64_t)data[3]) << 32) |
287 (((uint64_t)data[4]) << 24) |
288 (((uint64_t)data[5]) << 16) |
289 (((uint64_t)data[6]) << 8) |
295 int RenderFarmServerThread::write_int64(int64_t value)
297 unsigned char data[sizeof(int64_t)];
298 data[0] = (value >> 56) & 0xff;
299 data[1] = (value >> 48) & 0xff;
300 data[2] = (value >> 40) & 0xff;
301 data[3] = (value >> 32) & 0xff;
302 data[4] = (value >> 24) & 0xff;
303 data[5] = (value >> 16) & 0xff;
304 data[6] = (value >> 8) & 0xff;
305 data[7] = value & 0xff;
306 return (write_socket((char*)data, sizeof(int64_t)) !=
312 int RenderFarmServerThread::read_socket(char *data, int len)
316 //printf("RenderFarmServerThread::read_socket 1\n");
317 watchdog->begin_request();
318 while(len > 0 && bytes_read >= 0)
321 bytes_read = read(socket_fd, data + offset, len);
326 offset += bytes_read;
332 watchdog->end_request();
333 //printf("RenderFarmServerThread::read_socket 10\n");
338 int RenderFarmServerThread::write_socket(char *data, int len)
340 //printf("RenderFarmServerThread::write_socket 1\n");
341 int result = write(socket_fd, data, len);
342 //printf("RenderFarmServerThread::write_socket 10\n");
347 void RenderFarmServerThread::reallocate_buffer(int size)
349 if(buffer && buffer_allocated < size)
357 buffer = new unsigned char[size];
358 buffer_allocated = size;
362 void RenderFarmServerThread::run()
365 unsigned char header[5];
371 buffer_allocated = 0;
372 // fs_server = new RenderFarmFSServer(this);
373 // fs_server->initialize();
376 // Send command to run package renderer.
377 write_int64(RENDERFARM_PACKAGES);
384 // Wait for requests.
385 // Requests consist of request ID's and accompanying buffers.
387 bytes_read = read_socket((char*)header, 5);
388 //printf("RenderFarmServerThread::run 1\n");
395 int request_id = header[0];
396 int64_t request_size = (((u_int32_t)header[1]) << 24) |
397 (((u_int32_t)header[2]) << 16) |
398 (((u_int32_t)header[3]) << 8) |
399 (u_int32_t)header[4];
401 reallocate_buffer(request_size);
403 // Get accompanying buffer
404 bytes_read = read_socket((char*)buffer, request_size);
406 //printf("RenderFarmServerThread::run 2 %d %jd %d\n", request_id, request_size, bytes_read);
407 if(bytes_read != request_size)
412 //printf("RenderFarmServerThread::run 3\n");
416 case RENDERFARM_PREFERENCES:
420 case RENDERFARM_ASSET:
428 case RENDERFARM_PACKAGE:
429 send_package(buffer);
432 case RENDERFARM_PROGRESS:
433 set_progress(buffer);
436 case RENDERFARM_SET_RESULT:
440 case RENDERFARM_SET_VMAP:
441 set_video_map(buffer);
444 case RENDERFARM_GET_RESULT:
448 case RENDERFARM_DONE:
449 //printf("RenderFarmServerThread::run 10\n");
453 case RENDERFARM_KEEPALIVE:
457 // if(!fs_server->handle_request(request_id, request_size, (unsigned char*)buffer))
459 printf(_("RenderFarmServerThread::run: unknown request %02x\n"), request_id);
463 //printf("RenderFarmServerThread::run 10 %d %jd\n", request_id, request_size);
466 // Don't let watchdog kill the entire renderfarm when a client finishes
470 //printf("RenderFarmServerThread::run 20\n");
478 void RenderFarmServerThread::write_string(char *string)
483 len = strlen(string) + 1;
484 datagram = new char[len + 4];
486 memcpy(datagram + i, string, len);
487 write_socket((char*)datagram, len + 4);
488 //printf("RenderFarmServerThread::write_string %02x%02x%02x%02x\n",
489 // datagram[0], datagram[1], datagram[2], datagram[3]);
495 void RenderFarmServerThread::send_preferences()
500 server->preferences->save_defaults(&defaults);
501 defaults.save_string(string);
502 write_string(string);
506 void RenderFarmServerThread::send_asset()
511 // The asset must be sent in two segments.
512 // One segment is stored in the EDL and contains decoding information.
513 // One segment is stored in the asset and contains encoding information.
514 server->default_asset->save_defaults(&defaults,
521 defaults.save_string(string1);
523 server->default_asset->write(&file, 0, 0);
524 file.terminate_string();
526 write_string(string1);
527 write_string(file.string());
532 void RenderFarmServerThread::send_edl()
537 server->edl->save_xml(&file,
541 file.terminate_string();
542 //printf("RenderFarmServerThread::send_edl\n%s\n\n", file.string);
544 write_string(file.string());
545 //printf("RenderFarmServerThread::send_edl 2\n");
549 void RenderFarmServerThread::send_package(unsigned char *buffer)
551 this->frames_per_second = (double)((((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])) /
557 //printf("RenderFarmServerThread::send_package 1 %f\n", frames_per_second);
558 RenderPackage *package =
559 server->packages->get_package(frames_per_second,
561 server->use_local_rate);
563 //printf("RenderFarmServerThread::send_package 2\n");
564 datagram = new char[BCTEXTLEN];
569 //printf("RenderFarmServerThread::send_package 1\n");
570 datagram[0] = datagram[1] = datagram[2] = datagram[3] = 0;
571 write_socket(datagram, 4);
576 //printf("RenderFarmServerThread::send_package 10\n");
578 strcpy(&datagram[i], package->path);
579 i += strlen(package->path);
582 STORE_INT32(package->audio_start);
583 STORE_INT32(package->audio_end);
584 STORE_INT32(package->video_start);
585 STORE_INT32(package->video_end);
586 int use_brender = (server->brender ? 1 : 0);
587 STORE_INT32(use_brender);
588 STORE_INT32(package->audio_do);
589 STORE_INT32(package->video_do);
593 STORE_INT32(len - 4);
595 write_socket(datagram, len);
602 void RenderFarmServerThread::set_progress(unsigned char *buffer)
604 server->total_return_lock->lock("RenderFarmServerThread::set_progress");
605 *server->total_return += (int64_t)(((u_int32_t)buffer[0]) << 24) |
606 (((u_int32_t)buffer[1]) << 16) |
607 (((u_int32_t)buffer[2]) << 8) |
608 ((u_int32_t)buffer[3]);
609 frames_per_second = (double)((((u_int32_t)buffer[4]) << 24) |
610 (((u_int32_t)buffer[5]) << 16) |
611 (((u_int32_t)buffer[6]) << 8) |
612 ((u_int32_t)buffer[7])) /
614 server->total_return_lock->unlock();
616 server->preferences->set_rate(frames_per_second, number);
618 // This locks the preferences
619 if(server->mwindow) server->mwindow->preferences->copy_rates_from(
620 server->preferences);
623 int RenderFarmServerThread::set_video_map(unsigned char *buffer)
627 server->brender->set_video_map((int64_t)(((u_int32_t)buffer[0]) << 24) |
628 (((u_int32_t)buffer[1]) << 16) |
629 (((u_int32_t)buffer[2]) << 8) |
630 ((u_int32_t)buffer[3]),
631 (int64_t)(((u_int32_t)buffer[4]) << 24) |
632 (((u_int32_t)buffer[5]) << 16) |
633 (((u_int32_t)buffer[6]) << 8) |
634 ((u_int32_t)buffer[7]));
635 char return_value[1];
637 write_socket(return_value, 1);
644 void RenderFarmServerThread::set_result(unsigned char *buffer)
646 //printf("RenderFarmServerThread::set_result %p\n", buffer);
647 if(!*server->result_return)
648 *server->result_return = buffer[0];
652 void RenderFarmServerThread::get_result()
654 unsigned char data[1];
655 data[0] = *server->result_return;
656 write_socket((char*)data, 1);
672 RenderFarmWatchdog::RenderFarmWatchdog(
673 RenderFarmServerThread *server,
674 RenderFarmClientThread *client)
677 this->server = server;
678 this->client = client;
679 next_request = new Condition(0, "RenderFarmWatchdog::next_request", 0);
680 request_complete = new Condition(0, "RenderFarmWatchdog::request_complete", 0);
684 RenderFarmWatchdog::~RenderFarmWatchdog()
687 next_request->unlock();
688 request_complete->unlock();
691 delete request_complete;
694 void RenderFarmWatchdog::begin_request()
696 next_request->unlock();
699 void RenderFarmWatchdog::end_request()
701 request_complete->unlock();
704 void RenderFarmWatchdog::run()
708 next_request->lock("RenderFarmWatchdog::run");
710 int result = request_complete->timed_lock(RENDERFARM_TIMEOUT * 1000000,
711 "RenderFarmWatchdog::run");
712 //printf("RenderFarmWatchdog::run 1 %d\n", result);
718 printf("RenderFarmWatchdog::run 1 killing client pid %d\n", client->pid);
719 kill(client->pid, SIGKILL);
724 printf("RenderFarmWatchdog::run 1 killing server thread %p\n", server);
726 unsigned char buffer[4];
728 server->set_result(buffer);