This patch enables each omx instance to have a separate message dispatcher, and

thus eliminates the sharing of the message dispatches between omx instances.

If the omx audio and video encoders share the same dispatcher, when the audio read
blocks in the AudioSource, the message dispatcher thread gets blocked. As a result,
the message for the omx video encoder can not be dispatched, hence gets blocked too.
If the blocking time is long enough, the video frame rate decreases significantly.
This is the case when we read 2048 bytes by default each time. Reading smaller blocks
of audio data helps mitigate the above-mentioned problem, but it is not an ideal
solution for two reasons: a) it is not efficient, and thus can cause a lot of overhead
passing buffers between the frame work and the omx encoders; b) also, the audio record
thread can overflow as a result, which lead to the loss of recorded audio frames.

This patch affects both authoring engine and the playback engine.

Change-Id: I26dfde7ac46c8752cf1793ce1bfcc7be7724580e

2 files changed, 22 insertions, 16 deletions

diff --git a/media/libstagefright/include/OMX.h b/media/libstagefright/include/OMX.h
index ea131e8..c99da59 100644
--- a/media/libstagefright/include/OMX.h
+++ b/media/libstagefright/include/OMX.h
@@ -102,7 +102,7 @@ public:
             OMX_IN OMX_U32 nData1,
             OMX_IN OMX_U32 nData2,
             OMX_IN OMX_PTR pEventData);
-        
+
     OMX_ERRORTYPE OnEmptyBufferDone(
             node_id node, OMX_IN OMX_BUFFERHEADERTYPE *pBuffer);
 
@@ -115,20 +115,19 @@ protected:
     virtual ~OMX();
 
 private:
-    Mutex mLock;
-
-    OMXMaster *mMaster;
-
     struct CallbackDispatcher;
-    sp<CallbackDispatcher> mDispatcher;
 
+    Mutex mLock;
+    OMXMaster *mMaster;
     int32_t mNodeCounter;
 
     KeyedVector<wp<IBinder>, OMXNodeInstance *> mLiveNodes;
     KeyedVector<node_id, OMXNodeInstance *> mNodeIDToInstance;
+    KeyedVector<node_id, sp<CallbackDispatcher> > mDispatchers;
 
     node_id makeNodeID(OMXNodeInstance *instance);
     OMXNodeInstance *findInstance(node_id node);
+    sp<CallbackDispatcher> findDispatcher(node_id node);
 
     void invalidateNodeID_l(node_id node);
 
diff --git a/media/libstagefright/omx/OMX.cpp b/media/libstagefright/omx/OMX.cpp
index 132e31b..ad5b0f9 100644
--- a/media/libstagefright/omx/OMX.cpp
+++ b/media/libstagefright/omx/OMX.cpp
@@ -43,7 +43,7 @@ namespace android {
 ////////////////////////////////////////////////////////////////////////////////
 
 struct OMX::CallbackDispatcher : public RefBase {
-    CallbackDispatcher(OMX *owner);
+    CallbackDispatcher(OMXNodeInstance *owner);
 
     void post(const omx_message &msg);
 
@@ -53,7 +53,7 @@ protected:
 private:
     Mutex mLock;
 
-    OMX *mOwner;
+    OMXNodeInstance *mOwner;
     bool mDone;
     Condition mQueueChanged;
     List<omx_message> mQueue;
@@ -69,7 +69,7 @@ private:
     CallbackDispatcher &operator=(const CallbackDispatcher &);
 };
 
-OMX::CallbackDispatcher::CallbackDispatcher(OMX *owner)
+OMX::CallbackDispatcher::CallbackDispatcher(OMXNodeInstance *owner)
     : mOwner(owner),
       mDone(false) {
     pthread_attr_t attr;
@@ -101,12 +101,11 @@ void OMX::CallbackDispatcher::post(const omx_message &msg) {
 }
 
 void OMX::CallbackDispatcher::dispatch(const omx_message &msg) {
-    OMXNodeInstance *instance = mOwner->findInstance(msg.node);
-    if (instance == NULL) {
+    if (mOwner == NULL) {
         LOGV("Would have dispatched a message to a node that's already gone.");
         return;
     }
-    instance->onMessage(msg);
+    mOwner->onMessage(msg);
 }
 
 // static
@@ -145,7 +144,6 @@ void OMX::CallbackDispatcher::threadEntry() {
 
 OMX::OMX()
     : mMaster(new OMXMaster),
-      mDispatcher(new CallbackDispatcher(this)),
       mNodeCounter(0) {
 }
 
@@ -226,6 +224,7 @@ status_t OMX::allocateNode(
     }
 
     *node = makeNodeID(instance);
+    mDispatchers.add(*node, new CallbackDispatcher(instance));
 
     instance->setHandle(*node, handle);
 
@@ -341,7 +340,7 @@ OMX_ERRORTYPE OMX::OnEvent(
     msg.u.event_data.data1 = nData1;
     msg.u.event_data.data2 = nData2;
 
-    mDispatcher->post(msg);
+    findDispatcher(node)->post(msg);
 
     return OMX_ErrorNone;
 }
@@ -355,7 +354,7 @@ OMX_ERRORTYPE OMX::OnEmptyBufferDone(
     msg.node = node;
     msg.u.buffer_data.buffer = pBuffer;
 
-    mDispatcher->post(msg);
+    findDispatcher(node)->post(msg);
 
     return OMX_ErrorNone;
 }
@@ -375,7 +374,7 @@ OMX_ERRORTYPE OMX::OnFillBufferDone(
     msg.u.extended_buffer_data.platform_private = pBuffer->pPlatformPrivate;
     msg.u.extended_buffer_data.data_ptr = pBuffer->pBuffer;
 
-    mDispatcher->post(msg);
+    findDispatcher(node)->post(msg);
 
     return OMX_ErrorNone;
 }
@@ -397,6 +396,14 @@ OMXNodeInstance *OMX::findInstance(node_id node) {
     return index < 0 ? NULL : mNodeIDToInstance.valueAt(index);
 }
 
+sp<OMX::CallbackDispatcher> OMX::findDispatcher(node_id node) {
+    Mutex::Autolock autoLock(mLock);
+
+    ssize_t index = mDispatchers.indexOfKey(node);
+
+    return index < 0 ? NULL : mDispatchers.valueAt(index);
+}
+
 void OMX::invalidateNodeID(node_id node) {
     Mutex::Autolock autoLock(mLock);
     invalidateNodeID_l(node);