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 "compose.hpp"
38 #include <libcxml/cxml.h>
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);
50 using boost::shared_ptr;
51 using boost::weak_ptr;
52 using boost::optional;
54 int const Encoder::_history_size = 25;
56 /** @param f Film that we are encoding */
57 Encoder::Encoder (shared_ptr<const Film> film, weak_ptr<Job> j, shared_ptr<Writer> writer)
60 , _video_frames_enqueued (0)
66 servers_list_changed ();
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 LOG_GENERAL (N_("Adding %1 worker threads for remote %2"), d.threads(), d.host_name ());
82 for (int i = 0; i < d.threads(); ++i) {
83 _threads.push_back (new boost::thread (boost::bind (&Encoder::encoder_thread, this, d)));
86 _writer->set_encoder_threads (_threads.size ());
92 if (!Config::instance()->only_servers_encode ()) {
93 for (int i = 0; i < Config::instance()->num_local_encoding_threads (); ++i) {
94 _threads.push_back (new boost::thread (boost::bind (&Encoder::encoder_thread, this, optional<ServerDescription> ())));
98 _writer->set_encoder_threads (_threads.size ());
100 if (!ServerFinder::instance()->disabled ()) {
101 _server_found_connection = ServerFinder::instance()->ServersListChanged.connect (boost::bind (&Encoder::servers_list_changed, this));
108 boost::mutex::scoped_lock lock (_mutex);
110 LOG_GENERAL (N_("Clearing queue of %1"), _queue.size ());
112 /* Keep waking workers until the queue is empty */
113 while (!_queue.empty ()) {
114 _empty_condition.notify_all ();
115 _full_condition.wait (lock);
120 terminate_threads ();
122 LOG_GENERAL (N_("Mopping up %1"), _queue.size());
124 /* The following sequence of events can occur in the above code:
125 1. a remote worker takes the last image off the queue
126 2. the loop above terminates
127 3. the remote worker fails to encode the image and puts it back on the queue
128 4. the remote worker is then terminated by terminate_threads
130 So just mop up anything left in the queue here.
133 for (list<shared_ptr<DCPVideo> >::iterator i = _queue.begin(); i != _queue.end(); ++i) {
134 LOG_GENERAL (N_("Encode left-over frame %1"), (*i)->index ());
137 (*i)->encode_locally (boost::bind (&Log::dcp_log, _film->log().get(), _1, _2)),
142 } catch (std::exception& e) {
143 LOG_ERROR (N_("Local encode failed (%1)"), e.what ());
148 /** @return an estimate of the current number of frames we are encoding per second,
152 Encoder::current_encoding_rate () const
154 boost::mutex::scoped_lock lock (_state_mutex);
155 if (int (_time_history.size()) < _history_size) {
160 gettimeofday (&now, 0);
162 return _history_size / (seconds (now) - seconds (_time_history.back ()));
165 /** @return Number of video frames that have been sent out */
167 Encoder::video_frames_out () const
169 boost::mutex::scoped_lock (_state_mutex);
170 return _video_frames_enqueued;
173 /** Should be called when a frame has been encoded successfully.
174 * @param n Source frame index.
177 Encoder::frame_done ()
179 boost::mutex::scoped_lock lock (_state_mutex);
182 gettimeofday (&tv, 0);
183 _time_history.push_front (tv);
184 if (int (_time_history.size()) > _history_size) {
185 _time_history.pop_back ();
189 /** Called in order, so each time this is called the supplied frame is the one
190 * after the previous one.
193 Encoder::enqueue (shared_ptr<PlayerVideo> pv)
197 boost::mutex::scoped_lock lock (_mutex);
199 /* XXX: discard 3D here if required */
201 /* Wait until the queue has gone down a bit */
202 while (_queue.size() >= _threads.size() * 2 && !_terminate) {
203 LOG_TIMING ("decoder-sleep queue=%1", _queue.size());
204 _full_condition.wait (lock);
205 LOG_TIMING ("decoder-wake queue=%1", _queue.size());
213 /* Re-throw any exception raised by one of our threads. If more
214 than one has thrown an exception, only one will be rethrown, I think;
215 but then, if that happens something has gone badly wrong.
219 if (_writer->can_fake_write (_video_frames_enqueued)) {
220 /* We can fake-write this frame */
221 _writer->fake_write (_video_frames_enqueued, pv->eyes ());
223 } else if (pv->has_j2k ()) {
224 /* This frame already has JPEG2000 data, so just write it */
225 _writer->write (pv->j2k(), _video_frames_enqueued, pv->eyes ());
226 } else if (_last_player_video && pv->same (_last_player_video)) {
227 _writer->repeat (_video_frames_enqueued, pv->eyes ());
229 /* Queue this new frame for encoding */
230 LOG_TIMING ("add-frame-to-queue queue=%1", _queue.size ());
231 _queue.push_back (shared_ptr<DCPVideo> (
234 _video_frames_enqueued,
235 _film->video_frame_rate(),
236 _film->j2k_bandwidth(),
242 /* The queue might not be empty any more, so notify anything which is
245 _empty_condition.notify_all ();
248 switch (pv->eyes ()) {
250 ++_video_frames_enqueued;
262 if (_left_done && _right_done) {
263 ++_video_frames_enqueued;
264 _left_done = _right_done = false;
267 _last_player_video = pv;
271 Encoder::terminate_threads ()
274 boost::mutex::scoped_lock lock (_mutex);
276 _full_condition.notify_all ();
277 _empty_condition.notify_all ();
280 for (list<boost::thread *>::iterator i = _threads.begin(); i != _threads.end(); ++i) {
281 if ((*i)->joinable ()) {
292 Encoder::encoder_thread (optional<ServerDescription> server)
295 /* Number of seconds that we currently wait between attempts
296 to connect to the server; not relevant for localhost
299 int remote_backoff = 0;
303 LOG_TIMING ("encoder-sleep thread=%1", boost::this_thread::get_id());
304 boost::mutex::scoped_lock lock (_mutex);
305 while (_queue.empty () && !_terminate) {
306 _empty_condition.wait (lock);
313 LOG_TIMING ("encoder-wake thread=%1 queue=%2", boost::this_thread::get_id(), _queue.size());
314 shared_ptr<DCPVideo> vf = _queue.front ();
315 LOG_TIMING ("encoder-pop thread=%1 frame=%2 eyes=%3", boost::this_thread::get_id(), vf->index(), vf->eyes ());
320 optional<Data> encoded;
322 /* We need to encode this input */
325 encoded = vf->encode_remotely (server.get ());
327 if (remote_backoff > 0) {
328 LOG_GENERAL ("%1 was lost, but now she is found; removing backoff", server->host_name ());
331 /* This job succeeded, so remove any backoff */
334 } catch (std::exception& e) {
335 if (remote_backoff < 60) {
337 remote_backoff += 10;
340 N_("Remote encode of %1 on %2 failed (%3); thread sleeping for %4s"),
341 vf->index(), server->host_name(), e.what(), remote_backoff
347 LOG_TIMING ("start-local-encode thread=%1 frame=%2", boost::this_thread::get_id(), vf->index());
348 encoded = vf->encode_locally (boost::bind (&Log::dcp_log, _film->log().get(), _1, _2));
349 LOG_TIMING ("finish-local-encode thread=%1 frame=%2", boost::this_thread::get_id(), vf->index());
350 } catch (std::exception& e) {
351 LOG_ERROR (N_("Local encode failed (%1)"), e.what ());
356 _writer->write (encoded.get(), vf->index (), vf->eyes ());
360 LOG_GENERAL (N_("[%1] Encoder thread pushes frame %2 back onto queue after failure"), boost::this_thread::get_id(), vf->index());
361 _queue.push_front (vf);
365 if (remote_backoff > 0) {
366 dcpomatic_sleep (remote_backoff);
369 /* The queue might not be full any more, so notify anything that is waiting on that */
371 _full_condition.notify_all ();
380 Encoder::servers_list_changed ()
382 terminate_threads ();
383 BOOST_FOREACH (ServerDescription i, ServerFinder::instance()->servers ()) {
384 add_worker_threads (i);