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"
42 using std::stringstream;
48 using boost::shared_ptr;
49 using boost::weak_ptr;
50 using boost::optional;
51 using boost::scoped_array;
53 int const Encoder::_history_size = 25;
55 /** @param f Film that we are encoding */
56 Encoder::Encoder (shared_ptr<const Film> f, weak_ptr<Job> j)
59 , _video_frames_out (0)
62 _have_a_real_frame[EYES_BOTH] = false;
63 _have_a_real_frame[EYES_LEFT] = false;
64 _have_a_real_frame[EYES_RIGHT] = false;
75 /** Add a worker thread for a each thread on a remote server. Caller must hold
76 * a lock on _mutex, or know that one is not currently required to
77 * safely modify _threads.
80 Encoder::add_worker_threads (ServerDescription d)
82 for (int i = 0; i < d.threads(); ++i) {
83 _threads.push_back (new boost::thread (boost::bind (&Encoder::encoder_thread, this, d)));
88 Encoder::process_begin ()
90 for (int i = 0; i < Config::instance()->num_local_encoding_threads (); ++i) {
91 _threads.push_back (new boost::thread (boost::bind (&Encoder::encoder_thread, this, optional<ServerDescription> ())));
94 _writer.reset (new Writer (_film, _job));
95 ServerFinder::instance()->connect (boost::bind (&Encoder::server_found, this, _1));
100 Encoder::process_end ()
102 boost::mutex::scoped_lock lock (_mutex);
104 _film->log()->log (String::compose (N_("Clearing queue of %1"), _queue.size ()));
106 /* Keep waking workers until the queue is empty */
107 while (!_queue.empty ()) {
108 _film->log()->log (String::compose (N_("Waking with %1"), _queue.size ()), Log::VERBOSE);
109 _condition.notify_all ();
110 _condition.wait (lock);
115 terminate_threads ();
117 _film->log()->log (String::compose (N_("Mopping up %1"), _queue.size()));
119 /* The following sequence of events can occur in the above code:
120 1. a remote worker takes the last image off the queue
121 2. the loop above terminates
122 3. the remote worker fails to encode the image and puts it back on the queue
123 4. the remote worker is then terminated by terminate_threads
125 So just mop up anything left in the queue here.
128 for (list<shared_ptr<DCPVideoFrame> >::iterator i = _queue.begin(); i != _queue.end(); ++i) {
129 _film->log()->log (String::compose (N_("Encode left-over frame %1"), (*i)->frame ()));
131 _writer->write ((*i)->encode_locally(), (*i)->frame (), (*i)->eyes ());
133 } catch (std::exception& e) {
134 _film->log()->log (String::compose (N_("Local encode failed (%1)"), e.what ()));
142 /** @return an estimate of the current number of frames we are encoding per second,
146 Encoder::current_encoding_rate () const
148 boost::mutex::scoped_lock lock (_state_mutex);
149 if (int (_time_history.size()) < _history_size) {
154 gettimeofday (&now, 0);
156 return _history_size / (seconds (now) - seconds (_time_history.back ()));
159 /** @return Number of video frames that have been sent out */
161 Encoder::video_frames_out () const
163 boost::mutex::scoped_lock (_state_mutex);
164 return _video_frames_out;
167 /** Should be called when a frame has been encoded successfully.
168 * @param n Source frame index.
171 Encoder::frame_done ()
173 boost::mutex::scoped_lock lock (_state_mutex);
176 gettimeofday (&tv, 0);
177 _time_history.push_front (tv);
178 if (int (_time_history.size()) > _history_size) {
179 _time_history.pop_back ();
184 Encoder::process_video (shared_ptr<const Image> image, Eyes eyes, ColourConversion conversion, bool same)
186 boost::mutex::scoped_lock lock (_mutex);
188 /* XXX: discard 3D here if required */
190 /* Wait until the queue has gone down a bit */
191 while (_queue.size() >= _threads.size() * 2 && !_terminate) {
192 TIMING ("decoder sleeps with queue of %1", _queue.size());
193 _condition.wait (lock);
194 TIMING ("decoder wakes with queue of %1", _queue.size());
201 if (_writer->thrown ()) {
205 if (_writer->can_fake_write (_video_frames_out)) {
206 _writer->fake_write (_video_frames_out, eyes);
207 _have_a_real_frame[eyes] = false;
209 } else if (same && _have_a_real_frame[eyes]) {
210 /* Use the last frame that we encoded. */
211 _writer->repeat (_video_frames_out, eyes);
214 /* Queue this new frame for encoding */
215 TIMING ("adding to queue of %1", _queue.size ());
216 _queue.push_back (shared_ptr<DCPVideoFrame> (
218 image, _video_frames_out, eyes, conversion, _film->video_frame_rate(),
219 _film->j2k_bandwidth(), _film->log()
223 _condition.notify_all ();
224 _have_a_real_frame[eyes] = true;
227 if (eyes != EYES_LEFT) {
233 Encoder::process_audio (shared_ptr<const AudioBuffers> data)
235 _writer->write (data);
239 Encoder::terminate_threads ()
242 boost::mutex::scoped_lock lock (_mutex);
244 _condition.notify_all ();
247 for (list<boost::thread *>::iterator i = _threads.begin(); i != _threads.end(); ++i) {
248 if ((*i)->joinable ()) {
258 Encoder::encoder_thread (optional<ServerDescription> server)
260 /* Number of seconds that we currently wait between attempts
261 to connect to the server; not relevant for localhost
264 int remote_backoff = 0;
268 TIMING ("encoder thread %1 sleeps", boost::this_thread::get_id());
269 boost::mutex::scoped_lock lock (_mutex);
270 while (_queue.empty () && !_terminate) {
271 _condition.wait (lock);
278 TIMING ("encoder thread %1 wakes with queue of %2", boost::this_thread::get_id(), _queue.size());
279 shared_ptr<DCPVideoFrame> vf = _queue.front ();
280 TIMING ("encoder thread %1 pops frame %2 (%3) from queue", boost::this_thread::get_id(), vf->frame(), vf->eyes ());
285 shared_ptr<EncodedData> encoded;
289 encoded = vf->encode_remotely (server.get ());
291 if (remote_backoff > 0) {
292 _film->log()->log (String::compose (N_("%1 was lost, but now she is found; removing backoff"), server->host_name ()));
295 /* This job succeeded, so remove any backoff */
298 } catch (std::exception& e) {
299 if (remote_backoff < 60) {
301 remote_backoff += 10;
305 N_("Remote encode of %1 on %2 failed (%3); thread sleeping for %4s"),
306 vf->frame(), server->host_name(), e.what(), remote_backoff)
312 TIMING ("encoder thread %1 begins local encode of %2", boost::this_thread::get_id(), vf->frame());
313 encoded = vf->encode_locally ();
314 TIMING ("encoder thread %1 finishes local encode of %2", boost::this_thread::get_id(), vf->frame());
315 } catch (std::exception& e) {
316 _film->log()->log (String::compose (N_("Local encode failed (%1)"), e.what ()));
321 _writer->write (encoded, vf->frame (), vf->eyes ());
326 String::compose (N_("Encoder thread %1 pushes frame %2 back onto queue after failure"), boost::this_thread::get_id(), vf->frame())
328 _queue.push_front (vf);
332 if (remote_backoff > 0) {
333 dcpomatic_sleep (remote_backoff);
337 _condition.notify_all ();
342 Encoder::server_found (ServerDescription s)
344 add_worker_threads (s);