stagefright: add support for batching OMX events

Bug: 20503131
Change-Id: I762c419ed1245f8b83fb1f6bf61e5557213ca07b

4 files changed, 37 insertions, 21 deletions

diff --git a/media/libstagefright/omx/OMX.cpp b/media/libstagefright/omx/OMX.cpp
index 76217ec..e94adbd 100644
--- a/media/libstagefright/omx/OMX.cpp
+++ b/media/libstagefright/omx/OMX.cpp
@@ -61,7 +61,11 @@ private:
 struct OMX::CallbackDispatcher : public RefBase {
     CallbackDispatcher(OMXNodeInstance *owner);
 
-    void post(const omx_message &msg);
+    // Posts |msg| to the listener's queue. If |realTime| is true, the listener thread is notified
+    // that a new message is available on the queue. Otherwise, the message stays on the queue, but
+    // the listener is not notified of it. It will process this message when a subsequent message
+    // is posted with |realTime| set to true.
+    void post(const omx_message &msg, bool realTime = true);
 
     bool loop();
 
@@ -74,11 +78,11 @@ private:
     OMXNodeInstance *mOwner;
     bool mDone;
     Condition mQueueChanged;
-    List<omx_message> mQueue;
+    std::list<omx_message> mQueue;
 
     sp<CallbackDispatcherThread> mThread;
 
-    void dispatch(const omx_message &msg);
+    void dispatch(std::list<omx_message> &messages);
 
     CallbackDispatcher(const CallbackDispatcher &);
     CallbackDispatcher &operator=(const CallbackDispatcher &);
@@ -109,24 +113,26 @@ OMX::CallbackDispatcher::~CallbackDispatcher() {
     }
 }
 
-void OMX::CallbackDispatcher::post(const omx_message &msg) {
+void OMX::CallbackDispatcher::post(const omx_message &msg, bool realTime) {
     Mutex::Autolock autoLock(mLock);
 
     mQueue.push_back(msg);
-    mQueueChanged.signal();
+    if (realTime) {
+        mQueueChanged.signal();
+    }
 }
 
-void OMX::CallbackDispatcher::dispatch(const omx_message &msg) {
+void OMX::CallbackDispatcher::dispatch(std::list<omx_message> &messages) {
     if (mOwner == NULL) {
         ALOGV("Would have dispatched a message to a node that's already gone.");
         return;
     }
-    mOwner->onMessage(msg);
+    mOwner->onMessages(messages);
 }
 
 bool OMX::CallbackDispatcher::loop() {
     for (;;) {
-        omx_message msg;
+        std::list<omx_message> messages;
 
         {
             Mutex::Autolock autoLock(mLock);
@@ -138,11 +144,10 @@ bool OMX::CallbackDispatcher::loop() {
                 break;
             }
 
-            msg = *mQueue.begin();
-            mQueue.erase(mQueue.begin());
+            messages.swap(mQueue);
         }
 
-        dispatch(msg);
+        dispatch(messages);
     }
 
     return false;
diff --git a/media/libstagefright/omx/OMXNodeInstance.cpp b/media/libstagefright/omx/OMXNodeInstance.cpp
index 9e399f9..7e92da8 100644
--- a/media/libstagefright/omx/OMXNodeInstance.cpp
+++ b/media/libstagefright/omx/OMXNodeInstance.cpp
@@ -1357,7 +1357,7 @@ status_t OMXNodeInstance::setInternalOption(
     }
 }
 
-void OMXNodeInstance::onMessage(const omx_message &msg) {
+bool OMXNodeInstance::handleMessage(omx_message &msg) {
     const sp<GraphicBufferSource>& bufferSource(getGraphicBufferSource());
 
     if (msg.type == omx_message::FILL_BUFFER_DONE) {
@@ -1384,10 +1384,7 @@ void OMXNodeInstance::onMessage(const omx_message &msg) {
             // fix up the buffer info (especially timestamp) if needed
             bufferSource->codecBufferFilled(buffer);
 
-            omx_message newMsg = msg;
-            newMsg.u.extended_buffer_data.timestamp = buffer->nTimeStamp;
-            mObserver->onMessage(newMsg);
-            return;
+            msg.u.extended_buffer_data.timestamp = buffer->nTimeStamp;
         }
     } else if (msg.type == omx_message::EMPTY_BUFFER_DONE) {
         OMX_BUFFERHEADERTYPE *buffer =
@@ -1408,11 +1405,23 @@ void OMXNodeInstance::onMessage(const omx_message &msg) {
             // know that anyone asked to have the buffer emptied and will
             // be very confused.
             bufferSource->codecBufferEmptied(buffer, msg.fenceFd);
-            return;
+            return true;
+        }
+    }
+
+    return false;
+}
+
+void OMXNodeInstance::onMessages(std::list<omx_message> &messages) {
+    for (std::list<omx_message>::iterator it = messages.begin(); it != messages.end(); ) {
+        if (handleMessage(*it)) {
+            messages.erase(it++);
+        } else {
+            ++it;
         }
     }
 
-    mObserver->onMessage(msg);
+    mObserver->onMessages(messages);
 }
 
 void OMXNodeInstance::onObserverDied(OMXMaster *master) {
diff --git a/media/libstagefright/omx/tests/OMXHarness.cpp b/media/libstagefright/omx/tests/OMXHarness.cpp
index 294b2ed..644b6ed 100644
--- a/media/libstagefright/omx/tests/OMXHarness.cpp
+++ b/media/libstagefright/omx/tests/OMXHarness.cpp
@@ -64,9 +64,11 @@ status_t Harness::initOMX() {
     return mOMX != 0 ? OK : NO_INIT;
 }
 
-void Harness::onMessage(const omx_message &msg) {
+void Harness::onMessages(const std::list<omx_message> &messages) {
     Mutex::Autolock autoLock(mLock);
-    mMessageQueue.push_back(msg);
+    for (std::list<omx_message>::const_iterator it = messages.cbegin(); it != messages.cend(); ) {
+        mMessageQueue.push_back(*it++);
+    }
     mMessageAddedCondition.signal();
 }
 
diff --git a/media/libstagefright/omx/tests/OMXHarness.h b/media/libstagefright/omx/tests/OMXHarness.h
index bb8fd0c..1ebf3aa 100644
--- a/media/libstagefright/omx/tests/OMXHarness.h
+++ b/media/libstagefright/omx/tests/OMXHarness.h
@@ -74,7 +74,7 @@ struct Harness : public BnOMXObserver {
 
     status_t testAll();
 
-    virtual void onMessage(const omx_message &msg);
+    virtual void onMessages(const std::list<omx_message> &messages);
 
 protected:
     virtual ~Harness();