src/lib/server.cc

   1 /*
   2     Copyright (C) 2012 Carl Hetherington <cth@carlh.net>
   3
   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.
   8
   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.
  13
  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.
  17
  18 */
  19
  20 /** @file src/server.cc
  21  *  @brief Class to describe a server to which we can send
  22  *  encoding work, and a class to implement such a server.
  23  */
  24
  25 #include <string>
  26 #include <vector>
  27 #include <sstream>
  28 #include <boost/algorithm/string.hpp>
  29 #include "server.h"
  30 #include "util.h"
  31 #include "scaler.h"
  32 #include "image.h"
  33 #include "dcp_video_frame.h"
  34 #include "config.h"
  35
  36 using namespace std;
  37 using namespace boost;
  38
  39 /** Create a server description from a string of metadata returned from as_metadata().
  40  *  @param v Metadata.
  41  *  @return ServerDescription, or 0.
  42  */
  43 ServerDescription *
  44 ServerDescription::create_from_metadata (string v)
  45 {
  46         vector<string> b;
  47         split (b, v, is_any_of (" "));
  48
  49         if (b.size() != 2) {
  50                 return 0;
  51         }
  52
  53         return new ServerDescription (b[0], atoi (b[1].c_str ()));
  54 }
  55
  56 /** @return Description of this server as text */
  57 string
  58 ServerDescription::as_metadata () const
  59 {
  60         stringstream s;
  61         s << _host_name << " " << _threads;
  62         return s.str ();
  63 }
  64
  65 Server::Server ()
  66         : _log ("servomatic.log")
  67 {
  68
  69 }
  70
  71 int
  72 Server::process (shared_ptr<asio::ip::tcp::socket> socket)
  73 {
  74         SocketReader reader (socket);
  75
  76         char buffer[128];
  77         reader.read_indefinite ((uint8_t *) buffer, sizeof (buffer));
  78         reader.consume (strlen (buffer) + 1);
  79
  80         stringstream s (buffer);
  81
  82         string command;
  83         s >> command;
  84         if (command != "encode") {
  85                 return -1;
  86         }
  87
  88         Size in_size;
  89         int pixel_format_int;
  90         Size out_size;
  91         int padding;
  92         string scaler_id;
  93         int frame;
  94         float frames_per_second;
  95         string post_process;
  96         int colour_lut_index;
  97         int j2k_bandwidth;
  98
  99         s >> in_size.width >> in_size.height
 100           >> pixel_format_int
 101           >> out_size.width >> out_size.height
 102           >> padding
 103           >> scaler_id
 104           >> frame
 105           >> frames_per_second
 106           >> post_process
 107           >> colour_lut_index
 108           >> j2k_bandwidth;
 109
 110         PixelFormat pixel_format = (PixelFormat) pixel_format_int;
 111         Scaler const * scaler = Scaler::from_id (scaler_id);
 112         if (post_process == "none") {
 113                 post_process = "";
 114         }
 115
 116         shared_ptr<SimpleImage> image (new SimpleImage (pixel_format, in_size));
 117
 118         for (int i = 0; i < image->components(); ++i) {
 119                 int line_size;
 120                 s >> line_size;
 121                 image->set_line_size (i, line_size);
 122         }
 123
 124         for (int i = 0; i < image->components(); ++i) {
 125                 reader.read_definite_and_consume (image->data()[i], image->line_size()[i] * image->lines(i));
 126         }
 127
 128 #ifdef DEBUG_HASH
 129         image->hash ("Image for encoding (as received by server)");
 130 #endif
 131
 132         DCPVideoFrame dcp_video_frame (image, out_size, padding, scaler, frame, frames_per_second, post_process, colour_lut_index, j2k_bandwidth, &_log);
 133         shared_ptr<EncodedData> encoded = dcp_video_frame.encode_locally ();
 134         encoded->send (socket);
 135
 136 #ifdef DEBUG_HASH
 137         encoded->hash ("Encoded image (as made by server and as sent back)");
 138 #endif
 139
 140         return frame;
 141 }
 142
 143 void
 144 Server::worker_thread ()
 145 {
 146         while (1) {
 147                 mutex::scoped_lock lock (_worker_mutex);
 148                 while (_queue.empty ()) {
 149                         _worker_condition.wait (lock);
 150                 }
 151
 152                 shared_ptr<asio::ip::tcp::socket> socket = _queue.front ();
 153                 _queue.pop_front ();
 154
 155                 lock.unlock ();
 156
 157                 int frame = -1;
 158
 159                 struct timeval start;
 160                 gettimeofday (&start, 0);
 161
 162                 try {
 163                         frame = process (socket);
 164                 } catch (std::exception& e) {
 165                         cerr << "Error: " << e.what() << "\n";
 166                 }
 167
 168                 socket.reset ();
 169
 170                 lock.lock ();
 171
 172                 if (frame >= 0) {
 173                         struct timeval end;
 174                         gettimeofday (&end, 0);
 175                         cout << "Encoded frame " << frame << " in " << (seconds (end) - seconds (start)) << "\n";
 176                 }
 177
 178                 _worker_condition.notify_all ();
 179         }
 180 }
 181
 182 void
 183 Server::run ()
 184 {
 185         int const num_threads = Config::instance()->num_local_encoding_threads ();
 186
 187         for (int i = 0; i < num_threads; ++i) {
 188                 _worker_threads.push_back (new thread (bind (&Server::worker_thread, this)));
 189         }
 190
 191         asio::io_service io_service;
 192         asio::ip::tcp::acceptor acceptor (io_service, asio::ip::tcp::endpoint (asio::ip::tcp::v4(), Config::instance()->server_port ()));
 193         while (1) {
 194                 shared_ptr<asio::ip::tcp::socket> socket (new asio::ip::tcp::socket (io_service));
 195                 acceptor.accept (*socket);
 196
 197                 mutex::scoped_lock lock (_worker_mutex);
 198
 199                 /* Wait until the queue has gone down a bit */
 200                 while (int (_queue.size()) >= num_threads * 2) {
 201                         _worker_condition.wait (lock);
 202                 }
 203
 204                 _queue.push_back (socket);
 205                 _worker_condition.notify_all ();
 206         }
 207 }