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.
20 /** @file src/encoder.h
21 * @brief Parent class for classes which can encode video and audio frames.
25 #include <boost/lambda/lambda.hpp>
26 #include <libcxml/cxml.h>
32 #include "dcp_video_frame.h"
36 #include "server_finder.h"
43 using std::stringstream;
49 using boost::shared_ptr;
50 using boost::weak_ptr;
51 using boost::optional;
52 using boost::scoped_array;
54 int const Encoder::_history_size = 25;
56 /** @param f Film that we are encoding */
57 Encoder::Encoder (shared_ptr<const Film> f, weak_ptr<Job> j)
60 , _video_frames_out (0)
63 _have_a_real_frame[EYES_BOTH] = false;
64 _have_a_real_frame[EYES_LEFT] = false;
65 _have_a_real_frame[EYES_RIGHT] = false;
76 /** Add a worker thread for a each thread on a remote server. Caller must hold
77 * a lock on _mutex, or know that one is not currently required to
78 * safely modify _threads.
81 Encoder::add_worker_threads (ServerDescription d)
83 for (int i = 0; i < d.threads(); ++i) {
84 _threads.push_back (new boost::thread (boost::bind (&Encoder::encoder_thread, this, d)));
89 Encoder::process_begin ()
91 for (int i = 0; i < Config::instance()->num_local_encoding_threads (); ++i) {
92 _threads.push_back (new boost::thread (boost::bind (&Encoder::encoder_thread, this, optional<ServerDescription> ())));
95 _writer.reset (new Writer (_film, _job));
96 ServerFinder::instance()->connect (boost::bind (&Encoder::server_found, this, _1));
101 Encoder::process_end ()
103 boost::mutex::scoped_lock lock (_mutex);
105 _film->log()->log (String::compose (N_("Clearing queue of %1"), _queue.size ()));
107 /* Keep waking workers until the queue is empty */
108 while (!_queue.empty ()) {
109 _film->log()->log (String::compose (N_("Waking with %1"), _queue.size ()), Log::VERBOSE);
110 _condition.notify_all ();
111 _condition.wait (lock);
116 terminate_threads ();
118 _film->log()->log (String::compose (N_("Mopping up %1"), _queue.size()));
120 /* The following sequence of events can occur in the above code:
121 1. a remote worker takes the last image off the queue
122 2. the loop above terminates
123 3. the remote worker fails to encode the image and puts it back on the queue
124 4. the remote worker is then terminated by terminate_threads
126 So just mop up anything left in the queue here.
129 for (list<shared_ptr<DCPVideoFrame> >::iterator i = _queue.begin(); i != _queue.end(); ++i) {
130 _film->log()->log (String::compose (N_("Encode left-over frame %1"), (*i)->frame ()));
132 _writer->write ((*i)->encode_locally(), (*i)->frame (), (*i)->eyes ());
134 } catch (std::exception& e) {
135 _film->log()->log (String::compose (N_("Local encode failed (%1)"), e.what ()));
143 /** @return an estimate of the current number of frames we are encoding per second,
147 Encoder::current_encoding_rate () const
149 boost::mutex::scoped_lock lock (_state_mutex);
150 if (int (_time_history.size()) < _history_size) {
155 gettimeofday (&now, 0);
157 return _history_size / (seconds (now) - seconds (_time_history.back ()));
160 /** @return Number of video frames that have been sent out */
162 Encoder::video_frames_out () const
164 boost::mutex::scoped_lock (_state_mutex);
165 return _video_frames_out;
168 /** Should be called when a frame has been encoded successfully.
169 * @param n Source frame index.
172 Encoder::frame_done ()
174 boost::mutex::scoped_lock lock (_state_mutex);
177 gettimeofday (&tv, 0);
178 _time_history.push_front (tv);
179 if (int (_time_history.size()) > _history_size) {
180 _time_history.pop_back ();
185 Encoder::process_video (shared_ptr<PlayerImage> image, Eyes eyes, ColourConversion conversion, bool same)
187 boost::mutex::scoped_lock lock (_mutex);
189 /* XXX: discard 3D here if required */
191 /* Wait until the queue has gone down a bit */
192 while (_queue.size() >= _threads.size() * 2 && !_terminate) {
193 TIMING ("decoder sleeps with queue of %1", _queue.size());
194 _condition.wait (lock);
195 TIMING ("decoder wakes with queue of %1", _queue.size());
202 if (_writer->thrown ()) {
206 if (_writer->can_fake_write (_video_frames_out)) {
207 _writer->fake_write (_video_frames_out, eyes);
208 _have_a_real_frame[eyes] = false;
210 } else if (same && _have_a_real_frame[eyes]) {
211 /* Use the last frame that we encoded. */
212 _writer->repeat (_video_frames_out, eyes);
215 /* Queue this new frame for encoding */
216 TIMING ("adding to queue of %1", _queue.size ());
217 _queue.push_back (shared_ptr<DCPVideoFrame> (
219 image->image(PIX_FMT_RGB24, false), _video_frames_out, eyes, conversion, _film->video_frame_rate(),
220 _film->j2k_bandwidth(), _film->log()
224 _condition.notify_all ();
225 _have_a_real_frame[eyes] = true;
228 if (eyes != EYES_LEFT) {
234 Encoder::process_audio (shared_ptr<const AudioBuffers> data)
236 _writer->write (data);
240 Encoder::terminate_threads ()
243 boost::mutex::scoped_lock lock (_mutex);
245 _condition.notify_all ();
248 for (list<boost::thread *>::iterator i = _threads.begin(); i != _threads.end(); ++i) {
249 if ((*i)->joinable ()) {
259 Encoder::encoder_thread (optional<ServerDescription> server)
261 /* Number of seconds that we currently wait between attempts
262 to connect to the server; not relevant for localhost
265 int remote_backoff = 0;
269 TIMING ("encoder thread %1 sleeps", boost::this_thread::get_id());
270 boost::mutex::scoped_lock lock (_mutex);
271 while (_queue.empty () && !_terminate) {
272 _condition.wait (lock);
279 TIMING ("encoder thread %1 wakes with queue of %2", boost::this_thread::get_id(), _queue.size());
280 shared_ptr<DCPVideoFrame> vf = _queue.front ();
281 TIMING ("encoder thread %1 pops frame %2 (%3) from queue", boost::this_thread::get_id(), vf->frame(), vf->eyes ());
286 shared_ptr<EncodedData> encoded;
290 encoded = vf->encode_remotely (server.get ());
292 if (remote_backoff > 0) {
293 _film->log()->log (String::compose (N_("%1 was lost, but now she is found; removing backoff"), server->host_name ()));
296 /* This job succeeded, so remove any backoff */
299 } catch (std::exception& e) {
300 if (remote_backoff < 60) {
302 remote_backoff += 10;
306 N_("Remote encode of %1 on %2 failed (%3); thread sleeping for %4s"),
307 vf->frame(), server->host_name(), e.what(), remote_backoff)
313 TIMING ("encoder thread %1 begins local encode of %2", boost::this_thread::get_id(), vf->frame());
314 encoded = vf->encode_locally ();
315 TIMING ("encoder thread %1 finishes local encode of %2", boost::this_thread::get_id(), vf->frame());
316 } catch (std::exception& e) {
317 _film->log()->log (String::compose (N_("Local encode failed (%1)"), e.what ()));
322 _writer->write (encoded, vf->frame (), vf->eyes ());
327 String::compose (N_("Encoder thread %1 pushes frame %2 back onto queue after failure"), boost::this_thread::get_id(), vf->frame())
329 _queue.push_front (vf);
333 if (remote_backoff > 0) {
334 dcpomatic_sleep (remote_backoff);
338 _condition.notify_all ();
343 Encoder::server_found (ServerDescription s)
345 add_worker_threads (s);