2 Copyright (C) 2012 Carl Hetherington <cth@carlh.net>
4 This program is free software; you can redistribute it and/or modify
5 it under the terms of the GNU General Public License as published by
6 the Free Software Foundation; either version 2 of the License, or
7 (at your option) any later version.
9 This program is distributed in the hope that it will be useful,
10 but WITHOUT ANY WARRANTY; without even the implied warranty of
11 MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
12 GNU General Public License for more details.
14 You should have received a copy of the GNU General Public License
15 along with this program; if not, write to the Free Software
16 Foundation, Inc., 675 Mass Ave, Cambridge, MA 02139, USA.
27 #ifdef DVDOMATIC_POSIX
28 #include <sys/types.h>
29 #include <sys/socket.h>
30 #include <netinet/in.h>
32 #include <boost/algorithm/string.hpp>
33 #include <boost/thread.hpp>
34 #include <boost/thread/mutex.hpp>
35 #include <boost/thread/condition.hpp>
37 #include "dcp_video_frame.h"
38 #include "exceptions.h"
48 using namespace boost;
50 static vector<thread *> worker_threads;
52 static std::list<int> queue;
53 static mutex worker_mutex;
54 static condition worker_condition;
55 static Log log_ ("servomatic.log");
60 SocketReader reader (fd);
63 reader.read_indefinite ((uint8_t *) buffer, sizeof (buffer));
64 reader.consume (strlen (buffer) + 1);
66 stringstream s (buffer);
70 if (command != "encode") {
81 float frames_per_second;
86 s >> in_size.width >> in_size.height
88 >> out_size.width >> out_size.height
97 PixelFormat pixel_format = (PixelFormat) pixel_format_int;
98 Scaler const * scaler = Scaler::from_id (scaler_id);
99 if (post_process == "none") {
103 shared_ptr<SimpleImage> image (new SimpleImage (pixel_format, in_size));
105 for (int i = 0; i < image->components(); ++i) {
108 image->set_line_size (i, line_size);
111 for (int i = 0; i < image->components(); ++i) {
112 reader.read_definite_and_consume (image->data()[i], image->line_size()[i] * image->lines(i));
116 image->hash ("Image for encoding (as received by server)");
119 DCPVideoFrame dcp_video_frame (image, out_size, padding, scaler, frame, frames_per_second, post_process, colour_lut_index, j2k_bandwidth, &log_);
120 shared_ptr<EncodedData> encoded = dcp_video_frame.encode_locally ();
124 encoded->hash ("Encoded image (as made by server and as sent back)");
135 mutex::scoped_lock lock (worker_mutex);
136 while (queue.empty ()) {
137 worker_condition.wait (lock);
140 int fd = queue.front ();
147 struct timeval start;
148 gettimeofday (&start, 0);
151 frame = process (fd);
152 } catch (std::exception& e) {
153 cerr << "Error: " << e.what() << "\n";
162 gettimeofday (&end, 0);
163 cout << "Encoded frame " << frame << " in " << (seconds (end) - seconds (start)) << "\n";
166 worker_condition.notify_all ();
173 Scaler::setup_scalers ();
175 int const num_threads = Config::instance()->num_local_encoding_threads ();
177 for (int i = 0; i < num_threads; ++i) {
178 worker_threads.push_back (new thread (worker_thread));
181 int fd = socket (AF_INET, SOCK_STREAM, 0);
183 throw NetworkError ("could not open socket");
187 setsockopt (fd, SOL_SOCKET, SO_REUSEADDR, &o, sizeof (o));
192 setsockopt (fd, SOL_SOCKET, SO_RCVTIMEO, (void *) &tv, sizeof (tv));
193 setsockopt (fd, SOL_SOCKET, SO_SNDTIMEO, (void *) &tv, sizeof (tv));
195 struct sockaddr_in server_address;
196 memset (&server_address, 0, sizeof (server_address));
197 server_address.sin_family = AF_INET;
198 server_address.sin_addr.s_addr = INADDR_ANY;
199 server_address.sin_port = htons (Config::instance()->server_port ());
200 if (::bind (fd, (struct sockaddr *) &server_address, sizeof (server_address)) < 0) {
202 s << "could not bind to port " << Config::instance()->server_port() << " (" << strerror (errno) << ")";
203 throw NetworkError (s.str());
206 listen (fd, BACKLOG);
209 struct sockaddr_in client_address;
210 socklen_t client_length = sizeof (client_address);
211 int new_fd = accept (fd, (struct sockaddr *) &client_address, &client_length);
213 if (errno != EAGAIN && errno != EWOULDBLOCK) {
214 throw NetworkError ("accept failed");
220 mutex::scoped_lock lock (worker_mutex);
222 /* Wait until the queue has gone down a bit */
223 while (int (queue.size()) >= num_threads * 2) {
224 worker_condition.wait (lock);
230 setsockopt (new_fd, SOL_SOCKET, SO_RCVTIMEO, (void *) &tv, sizeof (tv));
231 setsockopt (new_fd, SOL_SOCKET, SO_SNDTIMEO, (void *) &tv, sizeof (tv));
233 queue.push_back (new_fd);
234 worker_condition.notify_all ();