2 Copyright (C) 2012-2014 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"
38 #include "player_video_frame.h"
42 #define LOG_GENERAL(...) _film->log()->log (String::compose (__VA_ARGS__), Log::TYPE_GENERAL);
43 #define LOG_ERROR(...) _film->log()->log (String::compose (__VA_ARGS__), Log::TYPE_ERROR);
44 #define LOG_TIMING(...) _film->log()->microsecond_log (String::compose (__VA_ARGS__), Log::TYPE_TIMING);
48 using std::stringstream;
54 using boost::shared_ptr;
55 using boost::weak_ptr;
56 using boost::optional;
57 using boost::scoped_array;
59 int const Encoder::_history_size = 25;
61 /** @param f Film that we are encoding */
62 Encoder::Encoder (shared_ptr<const Film> f, weak_ptr<Job> j)
65 , _video_frames_out (0)
68 _have_a_real_frame[EYES_BOTH] = false;
69 _have_a_real_frame[EYES_LEFT] = false;
70 _have_a_real_frame[EYES_RIGHT] = false;
78 /** Add a worker thread for a each thread on a remote server. Caller must hold
79 * a lock on _mutex, or know that one is not currently required to
80 * safely modify _threads.
83 Encoder::add_worker_threads (ServerDescription d)
85 LOG_GENERAL (N_("Adding %1 worker threads for remote %2"), d.host_name ());
86 for (int i = 0; i < d.threads(); ++i) {
87 _threads.push_back (new boost::thread (boost::bind (&Encoder::encoder_thread, this, d)));
92 Encoder::process_begin ()
94 for (int i = 0; i < Config::instance()->num_local_encoding_threads (); ++i) {
95 _threads.push_back (new boost::thread (boost::bind (&Encoder::encoder_thread, this, optional<ServerDescription> ())));
98 _writer.reset (new Writer (_film, _job));
99 ServerFinder::instance()->connect (boost::bind (&Encoder::server_found, this, _1));
103 Encoder::process_end ()
105 boost::mutex::scoped_lock lock (_mutex);
107 LOG_GENERAL (N_("Clearing queue of %1"), _queue.size ());
109 /* Keep waking workers until the queue is empty */
110 while (!_queue.empty ()) {
111 _empty_condition.notify_all ();
112 _full_condition.wait (lock);
117 terminate_threads ();
119 LOG_GENERAL (N_("Mopping up %1"), _queue.size());
121 /* The following sequence of events can occur in the above code:
122 1. a remote worker takes the last image off the queue
123 2. the loop above terminates
124 3. the remote worker fails to encode the image and puts it back on the queue
125 4. the remote worker is then terminated by terminate_threads
127 So just mop up anything left in the queue here.
130 for (list<shared_ptr<DCPVideoFrame> >::iterator i = _queue.begin(); i != _queue.end(); ++i) {
131 LOG_GENERAL (N_("Encode left-over frame %1"), (*i)->index ());
133 _writer->write ((*i)->encode_locally(), (*i)->index (), (*i)->eyes ());
135 } catch (std::exception& e) {
136 LOG_ERROR (N_("Local encode failed (%1)"), e.what ());
144 /** @return an estimate of the current number of frames we are encoding per second,
148 Encoder::current_encoding_rate () const
150 boost::mutex::scoped_lock lock (_state_mutex);
151 if (int (_time_history.size()) < _history_size) {
156 gettimeofday (&now, 0);
158 return _history_size / (seconds (now) - seconds (_time_history.back ()));
161 /** @return Number of video frames that have been sent out */
163 Encoder::video_frames_out () const
165 boost::mutex::scoped_lock (_state_mutex);
166 return _video_frames_out;
169 /** Should be called when a frame has been encoded successfully.
170 * @param n Source frame index.
173 Encoder::frame_done ()
175 boost::mutex::scoped_lock lock (_state_mutex);
178 gettimeofday (&tv, 0);
179 _time_history.push_front (tv);
180 if (int (_time_history.size()) > _history_size) {
181 _time_history.pop_back ();
186 Encoder::process_video (shared_ptr<PlayerVideoFrame> pvf, bool same)
190 boost::mutex::scoped_lock lock (_mutex);
192 /* XXX: discard 3D here if required */
194 /* Wait until the queue has gone down a bit */
195 while (_queue.size() >= _threads.size() * 2 && !_terminate) {
196 LOG_TIMING ("decoder sleeps with queue of %1", _queue.size());
197 _full_condition.wait (lock);
198 LOG_TIMING ("decoder wakes with queue of %1", _queue.size());
206 /* Re-throw any exception raised by one of our threads. If more
207 than one has thrown an exception, only one will be rethrown, I think;
208 but then, if that happens something has gone badly wrong.
212 if (_writer->can_fake_write (_video_frames_out)) {
213 _writer->fake_write (_video_frames_out, pvf->eyes ());
214 _have_a_real_frame[pvf->eyes()] = false;
216 } else if (same && _have_a_real_frame[pvf->eyes()]) {
217 /* Use the last frame that we encoded. */
218 _writer->repeat (_video_frames_out, pvf->eyes());
221 /* Queue this new frame for encoding */
222 LOG_TIMING ("adding to queue of %1", _queue.size ());
223 _queue.push_back (shared_ptr<DCPVideoFrame> (
225 pvf, _video_frames_out, _film->video_frame_rate(),
226 _film->j2k_bandwidth(), _film->resolution(), _film->log()
230 /* The queue might not be empty any more, so notify anything which is
233 _empty_condition.notify_all ();
234 _have_a_real_frame[pvf->eyes()] = true;
237 if (pvf->eyes() != EYES_LEFT) {
243 Encoder::process_audio (shared_ptr<const AudioBuffers> data)
245 _writer->write (data);
249 Encoder::terminate_threads ()
252 boost::mutex::scoped_lock lock (_mutex);
254 _full_condition.notify_all ();
255 _empty_condition.notify_all ();
258 for (list<boost::thread *>::iterator i = _threads.begin(); i != _threads.end(); ++i) {
259 if ((*i)->joinable ()) {
269 Encoder::encoder_thread (optional<ServerDescription> server)
272 /* Number of seconds that we currently wait between attempts
273 to connect to the server; not relevant for localhost
276 int remote_backoff = 0;
280 LOG_TIMING ("[%1] encoder thread sleeps", boost::this_thread::get_id());
281 boost::mutex::scoped_lock lock (_mutex);
282 while (_queue.empty () && !_terminate) {
283 _empty_condition.wait (lock);
290 LOG_TIMING ("[%1] encoder thread wakes with queue of %2", boost::this_thread::get_id(), _queue.size());
291 shared_ptr<DCPVideoFrame> vf = _queue.front ();
292 LOG_TIMING ("[%1] encoder thread pops frame %2 (%3) from queue", boost::this_thread::get_id(), vf->index(), vf->eyes ());
297 shared_ptr<EncodedData> encoded;
301 encoded = vf->encode_remotely (server.get ());
303 if (remote_backoff > 0) {
304 LOG_GENERAL ("%1 was lost, but now she is found; removing backoff", server->host_name ());
307 /* This job succeeded, so remove any backoff */
310 } catch (std::exception& e) {
311 if (remote_backoff < 60) {
313 remote_backoff += 10;
316 N_("Remote encode of %1 on %2 failed (%3); thread sleeping for %4s"),
317 vf->index(), server->host_name(), e.what(), remote_backoff
323 LOG_TIMING ("[%1] encoder thread begins local encode of %2", boost::this_thread::get_id(), vf->index());
324 encoded = vf->encode_locally ();
325 LOG_TIMING ("[%1] encoder thread finishes local encode of %2", boost::this_thread::get_id(), vf->index());
326 } catch (std::exception& e) {
327 LOG_ERROR (N_("Local encode failed (%1)"), e.what ());
332 _writer->write (encoded, vf->index (), vf->eyes ());
336 LOG_GENERAL (N_("[%1] Encoder thread pushes frame %2 back onto queue after failure"), boost::this_thread::get_id(), vf->index());
337 _queue.push_front (vf);
341 if (remote_backoff > 0) {
342 dcpomatic_sleep (remote_backoff);
345 /* The queue might not be full any more, so notify anything that is waiting on that */
347 _full_condition.notify_all ();
356 Encoder::server_found (ServerDescription s)
358 add_worker_threads (s);