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.
29 #include "dcp_video.h"
33 #include "server_finder.h"
35 #include "player_video.h"
36 #include <libcxml/cxml.h>
37 #include <boost/lambda/lambda.hpp>
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);
53 using boost::shared_ptr;
54 using boost::weak_ptr;
55 using boost::optional;
56 using boost::scoped_array;
58 int const Encoder::_history_size = 25;
60 /** @param f Film that we are encoding */
61 Encoder::Encoder (shared_ptr<const Film> f, weak_ptr<Job> j, shared_ptr<Writer> writer)
64 , _video_frames_out (0)
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 LOG_GENERAL (N_("Adding %1 worker threads for remote %2"), d.threads(), d.host_name ());
84 for (int i = 0; i < d.threads(); ++i) {
85 _threads.push_back (new boost::thread (boost::bind (&Encoder::encoder_thread, this, d)));
92 for (int i = 0; i < Config::instance()->num_local_encoding_threads (); ++i) {
93 _threads.push_back (new boost::thread (boost::bind (&Encoder::encoder_thread, this, optional<ServerDescription> ())));
96 ServerFinder::instance()->connect (boost::bind (&Encoder::server_found, this, _1));
102 boost::mutex::scoped_lock lock (_mutex);
104 LOG_GENERAL (N_("Clearing queue of %1"), _queue.size ());
106 /* Keep waking workers until the queue is empty */
107 while (!_queue.empty ()) {
108 _empty_condition.notify_all ();
109 _full_condition.wait (lock);
114 terminate_threads ();
116 LOG_GENERAL (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<DCPVideo> >::iterator i = _queue.begin(); i != _queue.end(); ++i) {
128 LOG_GENERAL (N_("Encode left-over frame %1"), (*i)->index ());
130 _writer->write ((*i)->encode_locally(), (*i)->index (), (*i)->eyes ());
132 } catch (std::exception& e) {
133 LOG_ERROR (N_("Local encode failed (%1)"), e.what ());
138 /** @return an estimate of the current number of frames we are encoding per second,
142 Encoder::current_encoding_rate () const
144 boost::mutex::scoped_lock lock (_state_mutex);
145 if (int (_time_history.size()) < _history_size) {
150 gettimeofday (&now, 0);
152 return _history_size / (seconds (now) - seconds (_time_history.back ()));
155 /** @return Number of video frames that have been sent out */
157 Encoder::video_frames_out () const
159 boost::mutex::scoped_lock (_state_mutex);
160 return _video_frames_out;
163 /** Should be called when a frame has been encoded successfully.
164 * @param n Source frame index.
167 Encoder::frame_done ()
169 boost::mutex::scoped_lock lock (_state_mutex);
172 gettimeofday (&tv, 0);
173 _time_history.push_front (tv);
174 if (int (_time_history.size()) > _history_size) {
175 _time_history.pop_back ();
179 /** Called in order, so each time this is called the supplied frame is the one
180 * after the previous one.
183 Encoder::enqueue (shared_ptr<PlayerVideo> pv)
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 LOG_TIMING ("decoder sleeps with queue of %1", _queue.size());
194 _full_condition.wait (lock);
195 LOG_TIMING ("decoder wakes with queue of %1", _queue.size());
203 /* Re-throw any exception raised by one of our threads. If more
204 than one has thrown an exception, only one will be rethrown, I think;
205 but then, if that happens something has gone badly wrong.
209 if (_writer->can_fake_write (_video_frames_out)) {
210 /* We can fake-write this frame */
211 _writer->fake_write (_video_frames_out, pv->eyes ());
213 } else if (pv->has_j2k ()) {
214 /* This frame already has JPEG2000 data, so just write it */
215 _writer->write (pv->j2k(), _video_frames_out, pv->eyes ());
217 /* Queue this new frame for encoding */
218 LOG_TIMING ("adding to queue of %1", _queue.size ());
219 _queue.push_back (shared_ptr<DCPVideo> (
223 _film->video_frame_rate(),
224 _film->j2k_bandwidth(),
226 _film->burn_subtitles(),
231 /* The queue might not be empty any more, so notify anything which is
234 _empty_condition.notify_all ();
237 if (pv->eyes() != EYES_LEFT) {
243 Encoder::terminate_threads ()
246 boost::mutex::scoped_lock lock (_mutex);
248 _full_condition.notify_all ();
249 _empty_condition.notify_all ();
252 for (list<boost::thread *>::iterator i = _threads.begin(); i != _threads.end(); ++i) {
253 if ((*i)->joinable ()) {
263 Encoder::encoder_thread (optional<ServerDescription> server)
266 /* Number of seconds that we currently wait between attempts
267 to connect to the server; not relevant for localhost
270 int remote_backoff = 0;
271 shared_ptr<DCPVideo> last_dcp_video;
272 shared_ptr<EncodedData> last_encoded;
276 LOG_TIMING ("[%1] encoder thread sleeps", boost::this_thread::get_id());
277 boost::mutex::scoped_lock lock (_mutex);
278 while (_queue.empty () && !_terminate) {
279 _empty_condition.wait (lock);
286 LOG_TIMING ("[%1] encoder thread wakes with queue of %2", boost::this_thread::get_id(), _queue.size());
287 shared_ptr<DCPVideo> vf = _queue.front ();
288 LOG_TIMING ("[%1] encoder thread pops frame %2 (%3) from queue", boost::this_thread::get_id(), vf->index(), vf->eyes ());
293 shared_ptr<EncodedData> encoded;
295 if (last_dcp_video && vf->same (last_dcp_video)) {
296 /* We already have encoded data for the same input as this one, so take a short-cut */
297 encoded = last_encoded;
299 /* We need to encode this input */
302 encoded = vf->encode_remotely (server.get ());
304 if (remote_backoff > 0) {
305 LOG_GENERAL ("%1 was lost, but now she is found; removing backoff", server->host_name ());
308 /* This job succeeded, so remove any backoff */
311 } catch (std::exception& e) {
312 if (remote_backoff < 60) {
314 remote_backoff += 10;
317 N_("Remote encode of %1 on %2 failed (%3); thread sleeping for %4s"),
318 vf->index(), server->host_name(), e.what(), remote_backoff
324 LOG_TIMING ("[%1] encoder thread begins local encode of %2", boost::this_thread::get_id(), vf->index());
325 encoded = vf->encode_locally ();
326 LOG_TIMING ("[%1] encoder thread finishes local encode of %2", boost::this_thread::get_id(), vf->index());
327 } catch (std::exception& e) {
328 LOG_ERROR (N_("Local encode failed (%1)"), e.what ());
334 last_encoded = encoded;
337 _writer->write (encoded, vf->index (), vf->eyes ());
341 LOG_GENERAL (N_("[%1] Encoder thread pushes frame %2 back onto queue after failure"), boost::this_thread::get_id(), vf->index());
342 _queue.push_front (vf);
346 if (remote_backoff > 0) {
347 dcpomatic_sleep (remote_backoff);
350 /* The queue might not be full any more, so notify anything that is waiting on that */
352 _full_condition.notify_all ();
361 Encoder::server_found (ServerDescription s)
363 add_worker_threads (s);