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;
73 /** Add a worker thread for a each thread on a remote server. Caller must hold
74 * a lock on _mutex, or know that one is not currently required to
75 * safely modify _threads.
78 Encoder::add_worker_threads (ServerDescription d)
80 for (int i = 0; i < d.threads(); ++i) {
81 _threads.push_back (new boost::thread (boost::bind (&Encoder::encoder_thread, this, d)));
86 Encoder::process_begin ()
88 for (int i = 0; i < Config::instance()->num_local_encoding_threads (); ++i) {
89 _threads.push_back (new boost::thread (boost::bind (&Encoder::encoder_thread, this, optional<ServerDescription> ())));
92 _writer.reset (new Writer (_film, _job));
93 ServerFinder::instance()->connect (boost::bind (&Encoder::server_found, this, _1));
97 Encoder::process_end ()
99 boost::mutex::scoped_lock lock (_mutex);
101 _film->log()->log (String::compose (N_("Clearing queue of %1"), _queue.size ()));
103 /* Keep waking workers until the queue is empty */
104 while (!_queue.empty ()) {
105 _film->log()->log (String::compose (N_("Waking with %1"), _queue.size ()), Log::VERBOSE);
106 _condition.notify_all ();
107 _condition.wait (lock);
112 terminate_threads ();
114 _film->log()->log (String::compose (N_("Mopping up %1"), _queue.size()));
116 /* The following sequence of events can occur in the above code:
117 1. a remote worker takes the last image off the queue
118 2. the loop above terminates
119 3. the remote worker fails to encode the image and puts it back on the queue
120 4. the remote worker is then terminated by terminate_threads
122 So just mop up anything left in the queue here.
125 for (list<shared_ptr<DCPVideoFrame> >::iterator i = _queue.begin(); i != _queue.end(); ++i) {
126 _film->log()->log (String::compose (N_("Encode left-over frame %1"), (*i)->frame ()));
128 _writer->write ((*i)->encode_locally(), (*i)->frame (), (*i)->eyes ());
130 } catch (std::exception& e) {
131 _film->log()->log (String::compose (N_("Local encode failed (%1)"), e.what ()));
139 /** @return an estimate of the current number of frames we are encoding per second,
143 Encoder::current_encoding_rate () const
145 boost::mutex::scoped_lock lock (_state_mutex);
146 if (int (_time_history.size()) < _history_size) {
151 gettimeofday (&now, 0);
153 return _history_size / (seconds (now) - seconds (_time_history.back ()));
156 /** @return Number of video frames that have been sent out */
158 Encoder::video_frames_out () const
160 boost::mutex::scoped_lock (_state_mutex);
161 return _video_frames_out;
164 /** Should be called when a frame has been encoded successfully.
165 * @param n Source frame index.
168 Encoder::frame_done ()
170 boost::mutex::scoped_lock lock (_state_mutex);
173 gettimeofday (&tv, 0);
174 _time_history.push_front (tv);
175 if (int (_time_history.size()) > _history_size) {
176 _time_history.pop_back ();
181 Encoder::process_video (shared_ptr<PlayerImage> image, Eyes eyes, ColourConversion conversion, bool same)
185 boost::mutex::scoped_lock lock (_mutex);
187 /* XXX: discard 3D here if required */
189 /* Wait until the queue has gone down a bit */
190 while (_queue.size() >= _threads.size() * 2 && !_terminate) {
191 TIMING ("decoder sleeps with queue of %1", _queue.size());
192 _condition.wait (lock);
193 TIMING ("decoder wakes with queue of %1", _queue.size());
201 /* Re-throw any exception raised by one of our threads. If more
202 than one has thrown an exception, only one will be rethrown, I think;
203 but then, if that happens something has gone badly wrong.
207 if (_writer->can_fake_write (_video_frames_out)) {
208 _writer->fake_write (_video_frames_out, eyes);
209 _have_a_real_frame[eyes] = false;
211 } else if (same && _have_a_real_frame[eyes]) {
212 /* Use the last frame that we encoded. */
213 _writer->repeat (_video_frames_out, eyes);
216 /* Queue this new frame for encoding */
217 TIMING ("adding to queue of %1", _queue.size ());
218 _queue.push_back (shared_ptr<DCPVideoFrame> (
220 image->image(), _video_frames_out, eyes, conversion, _film->video_frame_rate(),
221 _film->j2k_bandwidth(), _film->resolution(), _film->log()
225 _condition.notify_all ();
226 _have_a_real_frame[eyes] = true;
229 if (eyes != EYES_LEFT) {
235 Encoder::process_audio (shared_ptr<const AudioBuffers> data)
237 _writer->write (data);
241 Encoder::terminate_threads ()
244 boost::mutex::scoped_lock lock (_mutex);
246 _condition.notify_all ();
249 for (list<boost::thread *>::iterator i = _threads.begin(); i != _threads.end(); ++i) {
250 if ((*i)->joinable ()) {
260 Encoder::encoder_thread (optional<ServerDescription> server)
263 /* Number of seconds that we currently wait between attempts
264 to connect to the server; not relevant for localhost
267 int remote_backoff = 0;
271 TIMING ("encoder thread %1 sleeps", boost::this_thread::get_id());
272 boost::mutex::scoped_lock lock (_mutex);
273 while (_queue.empty () && !_terminate) {
274 _condition.wait (lock);
281 TIMING ("encoder thread %1 wakes with queue of %2", boost::this_thread::get_id(), _queue.size());
282 shared_ptr<DCPVideoFrame> vf = _queue.front ();
283 TIMING ("encoder thread %1 pops frame %2 (%3) from queue", boost::this_thread::get_id(), vf->frame(), vf->eyes ());
288 shared_ptr<EncodedData> encoded;
292 encoded = vf->encode_remotely (server.get ());
294 if (remote_backoff > 0) {
295 _film->log()->log (String::compose (N_("%1 was lost, but now she is found; removing backoff"), server->host_name ()));
298 /* This job succeeded, so remove any backoff */
301 } catch (std::exception& e) {
302 if (remote_backoff < 60) {
304 remote_backoff += 10;
308 N_("Remote encode of %1 on %2 failed (%3); thread sleeping for %4s"),
309 vf->frame(), server->host_name(), e.what(), remote_backoff)
315 TIMING ("encoder thread %1 begins local encode of %2", boost::this_thread::get_id(), vf->frame());
316 encoded = vf->encode_locally ();
317 TIMING ("encoder thread %1 finishes local encode of %2", boost::this_thread::get_id(), vf->frame());
318 } catch (std::exception& e) {
319 _film->log()->log (String::compose (N_("Local encode failed (%1)"), e.what ()));
324 _writer->write (encoded, vf->frame (), vf->eyes ());
329 String::compose (N_("Encoder thread %1 pushes frame %2 back onto queue after failure"), boost::this_thread::get_id(), vf->frame())
331 _queue.push_front (vf);
335 if (remote_backoff > 0) {
336 dcpomatic_sleep (remote_backoff);
340 _condition.notify_all ();
349 Encoder::server_found (ServerDescription s)
351 add_worker_threads (s);