Fix potential partial DASH segments during live packaging

Packager uses ThreadedIO to write media segments and manifest /
playlists. There was a possibility that media segments write being
delayed and scheduled after updating manifest / playlists.

This CL fixes the race condition.

Also added a note on how segments can be synced to cloud storage to
avoid the race condition during file sync.

Also added a live WebM test.

Fixes #386.

Change-Id: Icf9c38cdec715fa3dc2836eab1511131e129fe41

docs/source/tutorials/live.rst

@@ -41,6 +41,24 @@ Here are some examples.
     `preserved_segments_outside_live_window` option in
     :doc:`/options/dash_options` or :doc:`/options/hls_options` for details.
 
+.. note::
+
+    Shaka Packager ensures all segments referenced in DASH manifest / HLS
+    playlists are available, by updating the manifest / playlists only after a
+    segment is completed.
+
+    However, if content is not served directly from packaging output location,
+    extra care must be taken outside of packager to avoid updating manifest /
+    playlists without updating media segments.
+
+    Here is an example flow that avoids potential race condition. The following
+    steps should be done SERIALLY AND IN ORDER in every sync loop when uploading
+    manifest / playlists and media segments to content server:
+
+      1. Upload manifest / playlists under different names
+      2. Upload / Sync media segments
+      3. Rename uploaded manifest / playlists back to the original names
+
 .. include:: /options/udp_file_options.rst
 
 .. include:: /options/segment_template_formatting.rst

packager/app/test/packager_test.py

@@ -1241,6 +1241,15 @@ class PackagerFunctionalTest(PackagerAppTest):
             'http://foo.bar/check_me_for_the_date_header'))
     self._CheckTestResults('live-profile')
 
+  def testPackageLiveProfileWithWebM(self):
+    self.assertPackageSuccess(
+        self._GetStreams(
+            ['audio', 'video'],
+            segmented=True,
+            output_format='webm',
+            test_files=['bear-640x360.webm']), self._GetFlags())
+    self._CheckTestResults('live-profile-with-webm')
+
   def testPackageLiveStaticProfile(self):
     self.assertPackageSuccess(
         self._GetStreams(['audio', 'video'], segmented=True),

packager/app/test/testdata/live-profile-with-webm/output.mpd

@@ -0,0 +1,28 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--Generated with https://github.com/google/shaka-packager version <tag>-<hash>-<test>-->
+<MPD xmlns="urn:mpeg:dash:schema:mpd:2011" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns:xlink="http://www.w3.org/1999/xlink" xsi:schemaLocation="urn:mpeg:dash:schema:mpd:2011 DASH-MPD.xsd" xmlns:cenc="urn:mpeg:cenc:2013" profiles="urn:mpeg:dash:profile:isoff-live:2011" minBufferTime="PT2S" type="dynamic" publishTime="some_time" availabilityStartTime="some_time" minimumUpdatePeriod="PT5S" timeShiftBufferDepth="PT1800S">
+  <Period id="0" start="PT0S">
+    <AdaptationSet id="0" contentType="audio" segmentAlignment="true">
+      <Representation id="0" bandwidth="83969" codecs="vorbis" mimeType="audio/webm" audioSamplingRate="44100">
+        <AudioChannelConfiguration schemeIdUri="urn:mpeg:dash:23003:3:audio_channel_configuration:2011" value="2"/>
+        <SegmentTemplate timescale="1000000" initialization="bear-640x360-audio-init.webm" media="bear-640x360-audio-$Number$.webm" startNumber="1">
+          <SegmentTimeline>
+            <S t="0" d="1014000"/>
+            <S t="1014000" d="1002000"/>
+            <S t="2016000" d="755000"/>
+          </SegmentTimeline>
+        </SegmentTemplate>
+      </Representation>
+    </AdaptationSet>
+    <AdaptationSet id="1" contentType="video" width="640" height="360" frameRate="1000000/33000" segmentAlignment="true" par="16:9">
+      <Representation id="1" bandwidth="472384" codecs="vp8" mimeType="video/webm" sar="1:1">
+        <SegmentTemplate timescale="1000000" initialization="bear-640x360-video-init.webm" media="bear-640x360-video-$Number$.webm" startNumber="1">
+          <SegmentTimeline>
+            <S t="26000" d="1001000" r="1"/>
+            <S t="2028000" d="734000"/>
+          </SegmentTimeline>
+        </SegmentTemplate>
+      </Representation>
+    </AdaptationSet>
+  </Period>
+</MPD>

packager/media/formats/mp4/multi_segment_segmenter.cc

@@ -18,6 +18,7 @@
 #include "packager/media/event/muxer_listener.h"
 #include "packager/media/formats/mp4/box_definitions.h"
 #include "packager/media/formats/mp4/key_frame_info.h"
+#include "packager/status_macros.h"
 
 namespace shaka {
 namespace media {
@@ -61,10 +62,9 @@ Status MultiSegmentSegmenter::DoInitialize() {
 
 Status MultiSegmentSegmenter::DoFinalize() {
   // Update init segment with media duration set.
-  Status status = WriteInitSegment();
+  RETURN_IF_ERROR(WriteInitSegment());
-  if (status.ok())
+  SetComplete();
-    SetComplete();
+  return Status::OK;
-  return status;
 }
 
 Status MultiSegmentSegmenter::DoFinalizeSegment() {
@@ -180,21 +180,25 @@ Status MultiSegmentSegmenter::WriteSegment() {
   const size_t segment_size = segment_header_size + fragment_buffer()->Size();
   DCHECK_NE(segment_size, 0u);
 
-  Status status = buffer->WriteToFile(file.get());
+  RETURN_IF_ERROR(buffer->WriteToFile(file.get()));
-  if (status.ok()) {
+  if (muxer_listener()) {
-    if (muxer_listener()) {
+    for (const KeyFrameInfo& key_frame_info : key_frame_infos()) {
-      for (const KeyFrameInfo& key_frame_info : key_frame_infos()) {
+      muxer_listener()->OnKeyFrame(
-        muxer_listener()->OnKeyFrame(
+          key_frame_info.timestamp,
-            key_frame_info.timestamp,
+          segment_header_size + key_frame_info.start_byte_offset,
-            segment_header_size + key_frame_info.start_byte_offset,
+          key_frame_info.size);
-            key_frame_info.size);
-      }
     }
-    status = fragment_buffer()->WriteToFile(file.get());
   }
+  RETURN_IF_ERROR(fragment_buffer()->WriteToFile(file.get()));
 
-  if (!status.ok())
+  // Close the file, which also does flushing, to make sure the file is written
-    return status;
+  // before manifest is updated.
+  if (!file.release()->Close()) {
+    return Status(
+        error::FILE_FAILURE,
+        "Cannot close file " + file_name +
+            ", possibly file permission issue or running out of disk space.");
+  }
 
   uint64_t segment_duration = 0;
   // ISO/IEC 23009-1:2012: the value shall be identical to sum of the the

packager/media/formats/webm/multi_segment_segmenter.cc

@@ -10,6 +10,7 @@
 #include "packager/media/base/muxer_util.h"
 #include "packager/media/base/stream_info.h"
 #include "packager/media/event/muxer_listener.h"
+#include "packager/status_macros.h"
 #include "packager/third_party/libwebm/src/mkvmuxer.hpp"
 
 namespace shaka {
@@ -25,17 +26,21 @@ Status MultiSegmentSegmenter::FinalizeSegment(uint64_t start_timestamp,
                                               uint64_t duration_timestamp,
                                               bool is_subsegment) {
   CHECK(cluster());
-  Status status = Segmenter::FinalizeSegment(start_timestamp,
+  RETURN_IF_ERROR(Segmenter::FinalizeSegment(
-                                             duration_timestamp, is_subsegment);
+      start_timestamp, duration_timestamp, is_subsegment));
-  if (!status.ok())
-    return status;
   if (!cluster()->Finalize())
     return Status(error::FILE_FAILURE, "Error finalizing segment.");
+
   if (!is_subsegment) {
+    const std::string segment_name = writer_->file()->file_name();
+    // Close the file, which also does flushing, to make sure the file is
+    // written before manifest is updated.
+    RETURN_IF_ERROR(writer_->Close());
+
     if (muxer_listener()) {
       const uint64_t size = cluster()->Size();
-      muxer_listener()->OnNewSegment(writer_->file()->file_name(),
+      muxer_listener()->OnNewSegment(segment_name, start_timestamp,
-                                     start_timestamp, duration_timestamp, size);
+                                     duration_timestamp, size);
     }
     VLOG(1) << "WEBM file '" << writer_->file()->file_name() << "' finalized.";
   }
@@ -66,7 +71,7 @@ Status MultiSegmentSegmenter::DoInitialize() {
 }
 
 Status MultiSegmentSegmenter::DoFinalize() {
-  return writer_->Close();
+  return Status::OK;
 }
 
 Status MultiSegmentSegmenter::NewSegment(uint64_t start_timestamp,