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::optional;
50 using boost::scoped_array;
52 int const Encoder::_history_size = 25;
54 /** @param f Film that we are encoding */
55 Encoder::Encoder (shared_ptr<const Film> f, shared_ptr<Job> j)
58 , _video_frames_out (0)
61 _have_a_real_frame[EYES_BOTH] = false;
62 _have_a_real_frame[EYES_LEFT] = false;
63 _have_a_real_frame[EYES_RIGHT] = false;
74 /** Add a worker thread for a each thread on a remote server. Caller must hold
75 * a lock on _mutex, or know that one is not currently required to
76 * safely modify _threads.
79 Encoder::add_worker_threads (ServerDescription d)
81 for (int i = 0; i < d.threads(); ++i) {
82 _threads.push_back (new boost::thread (boost::bind (&Encoder::encoder_thread, this, d)));
87 Encoder::process_begin ()
89 for (int i = 0; i < Config::instance()->num_local_encoding_threads (); ++i) {
90 _threads.push_back (new boost::thread (boost::bind (&Encoder::encoder_thread, this, optional<ServerDescription> ())));
93 _writer.reset (new Writer (_film, _job));
94 ServerFinder::instance()->connect (boost::bind (&Encoder::server_found, this, _1));
99 Encoder::process_end ()
101 boost::mutex::scoped_lock lock (_mutex);
103 _film->log()->log (String::compose (N_("Clearing queue of %1"), _queue.size ()));
105 /* Keep waking workers until the queue is empty */
106 while (!_queue.empty ()) {
107 _film->log()->log (String::compose (N_("Waking with %1"), _queue.size ()), Log::VERBOSE);
108 _condition.notify_all ();
109 _condition.wait (lock);
114 terminate_threads ();
116 _film->log()->log (String::compose (N_("Mopping up %1"), _queue.size()));
118 /* The following sequence of events can occur in the above code:
119 1. a remote worker takes the last image off the queue
120 2. the loop above terminates
121 3. the remote worker fails to encode the image and puts it back on the queue
122 4. the remote worker is then terminated by terminate_threads
124 So just mop up anything left in the queue here.
127 for (list<shared_ptr<DCPVideoFrame> >::iterator i = _queue.begin(); i != _queue.end(); ++i) {
128 _film->log()->log (String::compose (N_("Encode left-over frame %1"), (*i)->frame ()));
130 _writer->write ((*i)->encode_locally(), (*i)->frame (), (*i)->eyes ());
132 } catch (std::exception& e) {
133 _film->log()->log (String::compose (N_("Local encode failed (%1)"), e.what ()));
141 /** @return an estimate of the current number of frames we are encoding per second,
145 Encoder::current_encoding_rate () const
147 boost::mutex::scoped_lock lock (_state_mutex);
148 if (int (_time_history.size()) < _history_size) {
153 gettimeofday (&now, 0);
155 return _history_size / (seconds (now) - seconds (_time_history.back ()));
158 /** @return Number of video frames that have been sent out */
160 Encoder::video_frames_out () const
162 boost::mutex::scoped_lock (_state_mutex);
163 return _video_frames_out;
166 /** Should be called when a frame has been encoded successfully.
167 * @param n Source frame index.
170 Encoder::frame_done ()
172 boost::mutex::scoped_lock lock (_state_mutex);
175 gettimeofday (&tv, 0);
176 _time_history.push_front (tv);
177 if (int (_time_history.size()) > _history_size) {
178 _time_history.pop_back ();
183 Encoder::process_video (shared_ptr<const Image> 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());
200 if (_writer->thrown ()) {
204 if (_writer->can_fake_write (_video_frames_out)) {
205 _writer->fake_write (_video_frames_out, eyes);
206 _have_a_real_frame[eyes] = false;
208 } else if (same && _have_a_real_frame[eyes]) {
209 /* Use the last frame that we encoded. */
210 _writer->repeat (_video_frames_out, eyes);
213 /* Queue this new frame for encoding */
214 TIMING ("adding to queue of %1", _queue.size ());
215 _queue.push_back (shared_ptr<DCPVideoFrame> (
217 image, _video_frames_out, eyes, conversion, _film->video_frame_rate(),
218 _film->j2k_bandwidth(), _film->log()
222 _condition.notify_all ();
223 _have_a_real_frame[eyes] = true;
226 if (eyes != EYES_LEFT) {
232 Encoder::process_audio (shared_ptr<const AudioBuffers> data)
234 _writer->write (data);
238 Encoder::terminate_threads ()
241 boost::mutex::scoped_lock lock (_mutex);
243 _condition.notify_all ();
246 for (list<boost::thread *>::iterator i = _threads.begin(); i != _threads.end(); ++i) {
247 if ((*i)->joinable ()) {
257 Encoder::encoder_thread (optional<ServerDescription> server)
259 /* Number of seconds that we currently wait between attempts
260 to connect to the server; not relevant for localhost
263 int remote_backoff = 0;
267 TIMING ("encoder thread %1 sleeps", boost::this_thread::get_id());
268 boost::mutex::scoped_lock lock (_mutex);
269 while (_queue.empty () && !_terminate) {
270 _condition.wait (lock);
277 TIMING ("encoder thread %1 wakes with queue of %2", boost::this_thread::get_id(), _queue.size());
278 shared_ptr<DCPVideoFrame> vf = _queue.front ();
279 _film->log()->log (String::compose (N_("Encoder thread %1 pops frame %2 (%3) from queue"), boost::this_thread::get_id(), vf->frame(), vf->eyes ()));
284 shared_ptr<EncodedData> encoded;
288 encoded = vf->encode_remotely (server.get ());
290 if (remote_backoff > 0) {
291 _film->log()->log (String::compose (N_("%1 was lost, but now she is found; removing backoff"), server->host_name ()));
294 /* This job succeeded, so remove any backoff */
297 } catch (std::exception& e) {
298 if (remote_backoff < 60) {
300 remote_backoff += 10;
304 N_("Remote encode of %1 on %2 failed (%3); thread sleeping for %4s"),
305 vf->frame(), server->host_name(), e.what(), remote_backoff)
311 TIMING ("encoder thread %1 begins local encode of %2", boost::this_thread::get_id(), vf->frame());
312 encoded = vf->encode_locally ();
313 TIMING ("encoder thread %1 finishes local encode of %2", boost::this_thread::get_id(), vf->frame());
314 } catch (std::exception& e) {
315 _film->log()->log (String::compose (N_("Local encode failed (%1)"), e.what ()));
320 _writer->write (encoded, vf->frame (), vf->eyes ());
325 String::compose (N_("Encoder thread %1 pushes frame %2 back onto queue after failure"), boost::this_thread::get_id(), vf->frame())
327 _queue.push_front (vf);
331 if (remote_backoff > 0) {
332 dcpomatic_sleep (remote_backoff);
336 _condition.notify_all ();
341 Encoder::server_found (ServerDescription s)
343 add_worker_threads (s);