src/lib/writer.cc

   1 /*
   2     Copyright (C) 2012 Carl Hetherington <cth@carlh.net>
   3
   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.
   8
   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.
  13
  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.
  17
  18 */
  19
  20 #include <fstream>
  21 #include <libdcp/picture_asset.h>
  22 #include <libdcp/sound_asset.h>
  23 #include <libdcp/picture_frame.h>
  24 #include <libdcp/reel.h>
  25 #include "writer.h"
  26 #include "compose.hpp"
  27 #include "film.h"
  28 #include "format.h"
  29 #include "log.h"
  30 #include "dcp_video_frame.h"
  31
  32 using std::make_pair;
  33 using std::pair;
  34 using std::string;
  35 using std::ifstream;
  36 using std::cout;
  37 using boost::shared_ptr;
  38
  39 unsigned int const Writer::_maximum_frames_in_memory = 8;
  40
  41 Writer::Writer (shared_ptr<Film> f)
  42         : _film (f)
  43         , _first_nonexistant_frame (0)
  44         , _thread (0)
  45         , _finish (false)
  46         , _last_written_frame (-1)
  47 {
  48         /* Remove any old DCP */
  49         boost::filesystem::remove_all (_film->dir (_film->dcp_name ()));
  50
  51         check_existing_picture_mxf ();
  52
  53         /* Create our picture asset in a subdirectory, named according to those
  54            film's parameters which affect the video output.  We will hard-link
  55            it into the DCP later.
  56         */
  57
  58         _picture_asset.reset (
  59                 new libdcp::MonoPictureAsset (
  60                         _film->video_mxf_dir (),
  61                         _film->video_mxf_filename (),
  62                         DCPFrameRate (_film->frames_per_second()).frames_per_second,
  63                         _film->format()->dcp_size()
  64                         )
  65                 );
  66
  67         _picture_asset_writer = _picture_asset->start_write (_first_nonexistant_frame > 0);
  68
  69         if (_film->audio_channels() > 0) {
  70                 _sound_asset.reset (
  71                         new libdcp::SoundAsset (
  72                                 _film->dir (_film->dcp_name()),
  73                                 "audio.mxf",
  74                                 DCPFrameRate (_film->frames_per_second()).frames_per_second,
  75                                 _film->audio_channels(),
  76                                 dcp_audio_sample_rate (_film->audio_stream()->sample_rate())
  77                                 )
  78                         );
  79
  80                 _sound_asset_writer = _sound_asset->start_write ();
  81         }
  82
  83         _thread = new boost::thread (boost::bind (&Writer::thread, this));
  84 }
  85
  86 void
  87 Writer::write (shared_ptr<const EncodedData> encoded, int frame)
  88 {
  89         boost::mutex::scoped_lock lock (_mutex);
  90
  91         QueueItem qi;
  92         qi.type = QueueItem::FULL;
  93         qi.encoded = encoded;
  94         qi.frame = frame;
  95         _queue.push_back (qi);
  96
  97         _condition.notify_all ();
  98 }
  99
 100 void
 101 Writer::fake_write (int frame)
 102 {
 103         boost::mutex::scoped_lock lock (_mutex);
 104
 105         ifstream ifi (_film->info_path (frame).c_str());
 106         libdcp::FrameInfo info (ifi);
 107
 108         QueueItem qi;
 109         qi.type = QueueItem::FAKE;
 110         qi.size = info.size;
 111         qi.frame = frame;
 112         _queue.push_back (qi);
 113
 114         _condition.notify_all ();
 115 }
 116
 117 /** This method is not thread safe */
 118 void
 119 Writer::write (shared_ptr<const AudioBuffers> audio)
 120 {
 121         _sound_asset_writer->write (audio->data(), audio->frames());
 122 }
 123
 124 void
 125 Writer::thread ()
 126 {
 127         while (1)
 128         {
 129                 boost::mutex::scoped_lock lock (_mutex);
 130
 131                 while (1) {
 132                         if (_finish ||
 133                             _queue.size() > _maximum_frames_in_memory ||
 134                             (!_queue.empty() && _queue.front().frame == (_last_written_frame + 1))) {
 135
 136                                     break;
 137                             }
 138
 139                             TIMING ("writer sleeps with a queue of %1; %2 pending", _queue.size(), _pending.size());
 140                             _condition.wait (lock);
 141                             TIMING ("writer wakes with a queue of %1", _queue.size());
 142
 143                             _queue.sort ();
 144                 }
 145
 146                 if (_finish && _queue.empty() && _pending.empty()) {
 147                         return;
 148                 }
 149
 150                 /* Write any frames that we can write; i.e. those that are in sequence */
 151                 while (!_queue.empty() && _queue.front().frame == (_last_written_frame + 1)) {
 152                         QueueItem qi = _queue.front ();
 153                         _queue.pop_front ();
 154
 155                         lock.unlock ();
 156                         switch (qi.type) {
 157                         case QueueItem::FULL:
 158                         {
 159                                 _film->log()->log (String::compose ("Writer FULL-writes %1 to MXF", qi.frame));
 160                                 libdcp::FrameInfo const fin = _picture_asset_writer->write (qi.encoded->data(), qi.encoded->size());
 161                                 qi.encoded->write_info (_film, qi.frame, fin);
 162                                 _last_written = qi.encoded;
 163                                 break;
 164                         }
 165                         case QueueItem::FAKE:
 166                                 _film->log()->log (String::compose ("Writer FAKE-writes %1 to MXF", qi.frame));
 167                                 _picture_asset_writer->fake_write (qi.size);
 168                                 _last_written.reset ();
 169                                 break;
 170                         case QueueItem::REPEAT:
 171                         {
 172                                 _film->log()->log (String::compose ("Writer REPEAT-writes %1 to MXF", qi.frame));
 173                                 libdcp::FrameInfo const fin = _picture_asset_writer->write (_last_written->data(), _last_written->size());
 174                                 _last_written->write_info (_film, qi.frame, fin);
 175                                 break;
 176                         }
 177                         }
 178                         lock.lock ();
 179
 180                         ++_last_written_frame;
 181                 }
 182
 183                 while (_queue.size() > _maximum_frames_in_memory) {
 184                         /* Too many frames in memory which can't yet be written to the stream.
 185                            Put some in our pending list (and write FULL queue items' data to disk)
 186                         */
 187
 188                         QueueItem qi = _queue.back ();
 189                         _queue.pop_back ();
 190
 191                         if (qi.type == QueueItem::FULL) {
 192                                 lock.unlock ();
 193                                 _film->log()->log (String::compose ("Writer full (awaiting %1); pushes %2 to disk", _last_written_frame + 1, qi.frame));
 194                                 qi.encoded->write (_film, qi.frame);
 195                                 lock.lock ();
 196                                 qi.encoded.reset ();
 197                         }
 198
 199                         _pending.push_back (qi);
 200                 }
 201
 202                 while (_queue.size() < _maximum_frames_in_memory && !_pending.empty()) {
 203                         /* We have some space in memory.  Fetch some frames back off disk. */
 204
 205                         _pending.sort ();
 206                         QueueItem qi = _pending.front ();
 207
 208                         if (qi.type == QueueItem::FULL) {
 209                                 lock.unlock ();
 210                                 _film->log()->log (String::compose ("Writer pulls %1 back from disk", qi.frame));
 211                                 shared_ptr<const EncodedData> encoded;
 212                                 qi.encoded.reset (new EncodedData (_film->j2c_path (qi.frame, false)));
 213                                 lock.lock ();
 214                         }
 215
 216                         _queue.push_back (qi);
 217                         _pending.remove (qi);
 218                 }
 219         }
 220
 221 }
 222
 223 void
 224 Writer::finish ()
 225 {
 226         if (!_thread) {
 227                 return;
 228         }
 229
 230         boost::mutex::scoped_lock lock (_mutex);
 231         _finish = true;
 232         _condition.notify_all ();
 233         lock.unlock ();
 234
 235         _thread->join ();
 236         delete _thread;
 237         _thread = 0;
 238
 239         _picture_asset_writer->finalize ();
 240
 241         if (_sound_asset_writer) {
 242                 _sound_asset_writer->finalize ();
 243         }
 244
 245         int const frames = _last_written_frame + 1;
 246         int const duration = frames - _film->trim_start() - _film->trim_end();
 247
 248         _film->set_dcp_intrinsic_duration (frames);
 249
 250         _picture_asset->set_entry_point (_film->trim_start ());
 251         _picture_asset->set_duration (duration);
 252
 253         /* Hard-link the video MXF into the DCP */
 254
 255         boost::filesystem::path from;
 256         from /= _film->video_mxf_dir();
 257         from /= _film->video_mxf_filename();
 258
 259         boost::filesystem::path to;
 260         to /= _film->dir (_film->dcp_name());
 261         to /= "video.mxf";
 262
 263         boost::filesystem::create_hard_link (from, to);
 264
 265         /* And update the asset */
 266
 267         _picture_asset->set_directory (_film->dir (_film->dcp_name ()));
 268         _picture_asset->set_file_name ("video.mxf");
 269
 270         if (_sound_asset) {
 271                 _sound_asset->set_entry_point (_film->trim_start ());
 272                 _sound_asset->set_duration (duration);
 273         }
 274
 275         libdcp::DCP dcp (_film->dir (_film->dcp_name()));
 276         DCPFrameRate dfr (_film->frames_per_second ());
 277
 278         shared_ptr<libdcp::CPL> cpl (
 279                 new libdcp::CPL (_film->dir (_film->dcp_name()), _film->dcp_name(), _film->dcp_content_type()->libdcp_kind (), frames, dfr.frames_per_second)
 280                 );
 281
 282         dcp.add_cpl (cpl);
 283
 284         cpl->add_reel (shared_ptr<libdcp::Reel> (new libdcp::Reel (
 285                                                          _picture_asset,
 286                                                          _sound_asset,
 287                                                          shared_ptr<libdcp::SubtitleAsset> ()
 288                                                          )
 289                                ));
 290
 291         dcp.write_xml ();
 292 }
 293
 294 /** Tell the writer that frame `f' should be a repeat of the frame before it */
 295 void
 296 Writer::repeat (int f)
 297 {
 298         boost::mutex::scoped_lock lock (_mutex);
 299
 300         QueueItem qi;
 301         qi.type = QueueItem::REPEAT;
 302         qi.frame = f;
 303
 304         _queue.push_back (qi);
 305
 306         _condition.notify_all ();
 307 }
 308
 309
 310 void
 311 Writer::check_existing_picture_mxf ()
 312 {
 313         /* Try to open the existing MXF */
 314         boost::filesystem::path p;
 315         p /= _film->video_mxf_dir ();
 316         p /= _film->video_mxf_filename ();
 317         FILE* mxf = fopen (p.string().c_str(), "rb");
 318         if (!mxf) {
 319                 return;
 320         }
 321
 322         while (1) {
 323
 324                 /* Read the frame info as written */
 325                 ifstream ifi (_film->info_path (_first_nonexistant_frame).c_str());
 326                 libdcp::FrameInfo info (ifi);
 327
 328                 /* Read the data from the MXF and hash it */
 329                 fseek (mxf, info.offset, SEEK_SET);
 330                 EncodedData data (info.size);
 331                 fread (data.data(), 1, data.size(), mxf);
 332                 string const existing_hash = md5_digest (data.data(), data.size());
 333
 334                 if (existing_hash != info.hash) {
 335                         _film->log()->log (String::compose ("Existing frame %1 failed hash check", _first_nonexistant_frame));
 336                         break;
 337                 }
 338
 339                 _film->log()->log (String::compose ("Have existing frame %1", _first_nonexistant_frame));
 340                 ++_first_nonexistant_frame;
 341         }
 342
 343         fclose (mxf);
 344 }
 345
 346 /** @return true if the fake write succeeded, otherwise false */
 347 bool
 348 Writer::can_fake_write (int frame) const
 349 {
 350         return (frame != 0 && frame < _first_nonexistant_frame);
 351 }
 352
 353
 354 bool
 355 operator< (QueueItem const & a, QueueItem const & b)
 356 {
 357         return a.frame < b.frame;
 358 }
 359
 360 bool
 361 operator== (QueueItem const & a, QueueItem const & b)
 362 {
 363         return a.frame == b.frame;
 364 }