MSE: Add StreamParser buffer remuxing utility and tests

media/base/stream_parser.cc

Index: media/base/stream_parser.cc
diff --git a/media/base/stream_parser.cc b/media/base/stream_parser.cc
index 12409194fede3cf0fcaa3d1671b7301416457409..b15d328643f06019c01fc54fdce674958c6478c1 100644
--- a/media/base/stream_parser.cc
+++ b/media/base/stream_parser.cc
@@ -4,10 +4,79 @@
 
 #include "media/base/stream_parser.h"
 
+#include "media/base/buffers.h"
+#include "media/base/stream_parser_buffer.h"
+
 namespace media {
 
 StreamParser::StreamParser() {}
 
 StreamParser::~StreamParser() {}
 
+bool StreamParser::MergeBufferQueues(const BufferQueue& audio_buffers,

[xhwang, 2014/01/29 08:04:50]

This function doesn't use any member variable. Does it make sense to declare
this as a standalone helper function instead of a class static function? (I am
not clear about how this function will be used so I am not sure.)

[wolenetz, 2014/02/05 02:49:53]

Yeah, there's really no need other than possibly name-scoping to StreamParser
class. This helper will hopefully not outlive http://crbug.com/338484's fix.
I've removed it from the class.

+                                     const BufferQueue& video_buffers,
+                                     const TextBufferQueueMap& text_buffers,
+                                     BufferQueue* merged_buffers) {
+  DCHECK(merged_buffers);
+
+  // Prepare vector containing pointers to any provided non-empty buffer queues.
+  std::vector<const BufferQueue*> buffer_queues;
+  if (!audio_buffers.empty())
+    buffer_queues.push_back(&audio_buffers);
+  if (!video_buffers.empty())
+    buffer_queues.push_back(&video_buffers);
+  for (TextBufferQueueMap::const_iterator map_itr = text_buffers.begin();
+       map_itr != text_buffers.end();
+       map_itr++) {
+    if (!map_itr->second.empty())
+      buffer_queues.push_back(&(map_itr->second));
+  }
+
+  // Do the merge.
+  return MergeBufferQueues(buffer_queues, merged_buffers);
+}
+
+bool
+StreamParser::MergeBufferQueues(std::vector<const BufferQueue*> buffer_queues,

[xhwang, 2014/01/29 08:04:50]

The implementation is complicated, we may need more comments about what we are
doing.

I haven't reviewed the detail of this function yet. Just OOC, this seems like
merge-sorting three already sorted sequences. Can we use std::merge to help us?

http://www.cplusplus.com/reference/algorithm/merge/

[wolenetz, 2014/02/05 02:49:53]

I've added more documentation now.

We'd need to hierarchically use std::merge for N, not 3, sorted sequences (since
there could be multiple text tracks, each with their own BufferQueue). Also,
std::merge behavior is undefined for non-sorted inputs. Granted, we could
presort, or pre-detect out-of-sort each queue for the latter, but I'm not
convinced the resulting code would be less complex or better performant than the
existing approach. I've also retained the non-critical allowance of pre-existing
content in |merged_buffers| and associated checks around new input buffers' DTS
vs the last buffer, if any, in |merged_buffers|.

This helper is really only a stop-gap to enable parallel work on
1) coded frame processing in SourceState::OnNewBuffers() from a single sorted
mixed-media-typed BufferQueue (another prereq for fixing
http://crbug.com/249422), and
2) refactoring this large-grain merge into fine-grained merges done directly by
the stream parsers (next step for fixing http://crbug.com/338484).

+                                BufferQueue* merged_buffers) {

[xhwang, 2014/01/29 08:04:50]

format looks odd; put bool and function name in the same line?

[wolenetz, 2014/02/05 02:49:53]

Done.

+  if (buffer_queues.empty())
+    return true;
+
+  std::vector<BufferQueue::const_iterator> itrs;
+  for (size_t i = 0; i < buffer_queues.size(); ++i)
+    itrs.push_back(buffer_queues[i]->begin());
+
+  base::TimeDelta last_timestamp = kNoTimestamp();
+  if (!merged_buffers->empty())
+    last_timestamp = merged_buffers->back()->GetDecodeTimestamp();
+
+  while (true) {
+    int index_of_queue_with_next_timestamp = -1;
+    base::TimeDelta next_timestamp = kNoTimestamp();
+
+    for (size_t i = 0; i < buffer_queues.size(); ++i) {
+      if (itrs[i] != buffer_queues[i]->end()) {
+        base::TimeDelta ts = (*itrs[i])->GetDecodeTimestamp();
+
+        if (last_timestamp != kNoTimestamp() && ts < last_timestamp)
+          return false;
+
+        if (ts < next_timestamp || next_timestamp == kNoTimestamp()) {
+          next_timestamp = ts;
+          index_of_queue_with_next_timestamp = i;
+        }
+      }
+    }
+
+    if (index_of_queue_with_next_timestamp == -1)
+      return true;
+
+    scoped_refptr<StreamParserBuffer> buffer =
+        *itrs[index_of_queue_with_next_timestamp];
+    last_timestamp = buffer->GetDecodeTimestamp();
+    merged_buffers->push_back(buffer);
+    ++itrs[index_of_queue_with_next_timestamp];
+  }
+}
+
 }  // namespace media