2 Copyright (C) 2012-2015 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"
37 #include <libcxml/cxml.h>
38 #include <boost/lambda/lambda.hpp>
39 #include <boost/foreach.hpp>
44 #define LOG_GENERAL(...) _film->log()->log (String::compose (__VA_ARGS__), Log::TYPE_GENERAL);
45 #define LOG_ERROR(...) _film->log()->log (String::compose (__VA_ARGS__), Log::TYPE_ERROR);
46 #define LOG_TIMING(...) _film->log()->microsecond_log (String::compose (__VA_ARGS__), Log::TYPE_TIMING);
55 using boost::shared_ptr;
56 using boost::weak_ptr;
57 using boost::optional;
58 using boost::scoped_array;
60 int const Encoder::_history_size = 25;
62 /** @param f Film that we are encoding */
63 Encoder::Encoder (shared_ptr<const Film> film, weak_ptr<Job> j, shared_ptr<Writer> writer)
66 , _video_frames_enqueued (0)
70 servers_list_changed ();
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.threads(), 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)));
90 _writer->set_encoder_threads (_threads.size ());
96 for (int i = 0; i < Config::instance()->num_local_encoding_threads (); ++i) {
97 _threads.push_back (new boost::thread (boost::bind (&Encoder::encoder_thread, this, optional<ServerDescription> ())));
100 _writer->set_encoder_threads (_threads.size ());
102 if (!ServerFinder::instance()->disabled ()) {
103 _server_found_connection = ServerFinder::instance()->ServersListChanged.connect (boost::bind (&Encoder::servers_list_changed, this));
110 boost::mutex::scoped_lock lock (_mutex);
112 LOG_GENERAL (N_("Clearing queue of %1"), _queue.size ());
114 /* Keep waking workers until the queue is empty */
115 while (!_queue.empty ()) {
116 _empty_condition.notify_all ();
117 _full_condition.wait (lock);
122 terminate_threads ();
124 LOG_GENERAL (N_("Mopping up %1"), _queue.size());
126 /* The following sequence of events can occur in the above code:
127 1. a remote worker takes the last image off the queue
128 2. the loop above terminates
129 3. the remote worker fails to encode the image and puts it back on the queue
130 4. the remote worker is then terminated by terminate_threads
132 So just mop up anything left in the queue here.
135 for (list<shared_ptr<DCPVideo> >::iterator i = _queue.begin(); i != _queue.end(); ++i) {
136 LOG_GENERAL (N_("Encode left-over frame %1"), (*i)->index ());
139 (*i)->encode_locally (boost::bind (&Log::dcp_log, _film->log().get(), _1, _2)),
144 } catch (std::exception& e) {
145 LOG_ERROR (N_("Local encode failed (%1)"), e.what ());
150 /** @return an estimate of the current number of frames we are encoding per second,
154 Encoder::current_encoding_rate () const
156 boost::mutex::scoped_lock lock (_state_mutex);
157 if (int (_time_history.size()) < _history_size) {
162 gettimeofday (&now, 0);
164 return _history_size / (seconds (now) - seconds (_time_history.back ()));
167 /** @return Number of video frames that have been sent out */
169 Encoder::video_frames_out () const
171 boost::mutex::scoped_lock (_state_mutex);
172 return _video_frames_enqueued;
175 /** Should be called when a frame has been encoded successfully.
176 * @param n Source frame index.
179 Encoder::frame_done ()
181 boost::mutex::scoped_lock lock (_state_mutex);
184 gettimeofday (&tv, 0);
185 _time_history.push_front (tv);
186 if (int (_time_history.size()) > _history_size) {
187 _time_history.pop_back ();
191 /** Called in order, so each time this is called the supplied frame is the one
192 * after the previous one.
195 Encoder::enqueue (shared_ptr<PlayerVideo> pv)
199 boost::mutex::scoped_lock lock (_mutex);
201 /* XXX: discard 3D here if required */
203 /* Wait until the queue has gone down a bit */
204 while (_queue.size() >= _threads.size() * 2 && !_terminate) {
205 LOG_TIMING ("decoder sleeps with queue of %1", _queue.size());
206 _full_condition.wait (lock);
207 LOG_TIMING ("decoder wakes with queue of %1", _queue.size());
215 /* Re-throw any exception raised by one of our threads. If more
216 than one has thrown an exception, only one will be rethrown, I think;
217 but then, if that happens something has gone badly wrong.
221 if (_writer->can_fake_write (_video_frames_enqueued)) {
222 /* We can fake-write this frame */
223 _writer->fake_write (_video_frames_enqueued, pv->eyes ());
225 } else if (pv->has_j2k ()) {
226 /* This frame already has JPEG2000 data, so just write it */
227 _writer->write (pv->j2k(), _video_frames_enqueued, pv->eyes ());
228 } else if (_last_player_video && pv->same (_last_player_video)) {
229 _writer->repeat (_video_frames_enqueued, pv->eyes ());
231 /* Queue this new frame for encoding */
232 LOG_TIMING ("adding to queue of %1", _queue.size ());
233 _queue.push_back (shared_ptr<DCPVideo> (
236 _video_frames_enqueued,
237 _film->video_frame_rate(),
238 _film->j2k_bandwidth(),
244 /* The queue might not be empty any more, so notify anything which is
247 _empty_condition.notify_all ();
250 if (pv->eyes() != EYES_LEFT) {
251 ++_video_frames_enqueued;
254 _last_player_video = pv;
258 Encoder::terminate_threads ()
261 boost::mutex::scoped_lock lock (_mutex);
263 _full_condition.notify_all ();
264 _empty_condition.notify_all ();
267 for (list<boost::thread *>::iterator i = _threads.begin(); i != _threads.end(); ++i) {
268 if ((*i)->joinable ()) {
279 Encoder::encoder_thread (optional<ServerDescription> server)
282 /* Number of seconds that we currently wait between attempts
283 to connect to the server; not relevant for localhost
286 int remote_backoff = 0;
290 LOG_TIMING ("[%1] encoder thread sleeps", boost::this_thread::get_id());
291 boost::mutex::scoped_lock lock (_mutex);
292 while (_queue.empty () && !_terminate) {
293 _empty_condition.wait (lock);
300 LOG_TIMING ("[%1] encoder thread wakes with queue of %2", boost::this_thread::get_id(), _queue.size());
301 shared_ptr<DCPVideo> vf = _queue.front ();
302 LOG_TIMING ("[%1] encoder thread pops frame %2 (%3) from queue", boost::this_thread::get_id(), vf->index(), vf->eyes ());
307 optional<Data> encoded;
309 /* We need to encode this input */
312 encoded = vf->encode_remotely (server.get ());
314 if (remote_backoff > 0) {
315 LOG_GENERAL ("%1 was lost, but now she is found; removing backoff", server->host_name ());
318 /* This job succeeded, so remove any backoff */
321 } catch (std::exception& e) {
322 if (remote_backoff < 60) {
324 remote_backoff += 10;
327 N_("Remote encode of %1 on %2 failed (%3); thread sleeping for %4s"),
328 vf->index(), server->host_name(), e.what(), remote_backoff
334 LOG_TIMING ("[%1] encoder thread begins local encode of %2", boost::this_thread::get_id(), vf->index());
335 encoded = vf->encode_locally (boost::bind (&Log::dcp_log, _film->log().get(), _1, _2));
336 LOG_TIMING ("[%1] encoder thread finishes local encode of %2", boost::this_thread::get_id(), vf->index());
337 } catch (std::exception& e) {
338 LOG_ERROR (N_("Local encode failed (%1)"), e.what ());
343 _writer->write (encoded.get(), vf->index (), vf->eyes ());
347 LOG_GENERAL (N_("[%1] Encoder thread pushes frame %2 back onto queue after failure"), boost::this_thread::get_id(), vf->index());
348 _queue.push_front (vf);
352 if (remote_backoff > 0) {
353 dcpomatic_sleep (remote_backoff);
356 /* The queue might not be full any more, so notify anything that is waiting on that */
358 _full_condition.notify_all ();
367 Encoder::servers_list_changed ()
369 terminate_threads ();
370 BOOST_FOREACH (ServerDescription i, ServerFinder::instance()->servers ()) {
371 add_worker_threads (i);