stagefright: don't use ALooperRoster mutex for reply handling

Change replyID-s from uint32_t to an object
Move reply handling into the loopers (to reuse a common mutex)

Bug: 19607784
Change-Id: Iaa035b846c424c5687ed17ce1079b325e86c54be

5 files changed, 67 insertions, 21 deletions

diff --git a/include/media/stagefright/MediaCodec.h b/include/media/stagefright/MediaCodec.h
index d448097..8241e19 100644
--- a/include/media/stagefright/MediaCodec.h
+++ b/include/media/stagefright/MediaCodec.h
@@ -27,6 +27,7 @@ namespace android {
 
 struct ABuffer;
 struct AMessage;
+struct AReplyToken;
 struct AString;
 struct CodecBase;
 struct ICrypto;
@@ -222,7 +223,7 @@ private:
     sp<ALooper> mCodecLooper;
     sp<CodecBase> mCodec;
     AString mComponentName;
-    uint32_t mReplyID;
+    sp<AReplyToken> mReplyID;
     uint32_t mFlags;
     status_t mStickyError;
     sp<Surface> mNativeWindow;
@@ -249,10 +250,10 @@ private:
     Vector<BufferInfo> mPortBuffers[2];
 
     int32_t mDequeueInputTimeoutGeneration;
-    uint32_t mDequeueInputReplyID;
+    sp<AReplyToken> mDequeueInputReplyID;
 
     int32_t mDequeueOutputTimeoutGeneration;
-    uint32_t mDequeueOutputReplyID;
+    sp<AReplyToken> mDequeueOutputReplyID;
 
     sp<ICrypto> mCrypto;
 
@@ -267,7 +268,7 @@ private:
     static status_t PostAndAwaitResponse(
             const sp<AMessage> &msg, sp<AMessage> *response);
 
-    static void PostReplyWithError(int32_t replyID, int32_t err);
+    static void PostReplyWithError(const sp<AReplyToken> &replyID, int32_t err);
 
     status_t init(const AString &name, bool nameIsType, bool encoder);
 
@@ -283,8 +284,8 @@ private:
             size_t portIndex, size_t index,
             sp<ABuffer> *buffer, sp<AMessage> *format);
 
-    bool handleDequeueInputBuffer(uint32_t replyID, bool newRequest = false);
-    bool handleDequeueOutputBuffer(uint32_t replyID, bool newRequest = false);
+    bool handleDequeueInputBuffer(const sp<AReplyToken> &replyID, bool newRequest = false);
+    bool handleDequeueOutputBuffer(const sp<AReplyToken> &replyID, bool newRequest = false);
     void cancelPendingDequeueOperations();
 
     void extractCSD(const sp<AMessage> &format);
diff --git a/include/media/stagefright/MediaCodecSource.h b/include/media/stagefright/MediaCodecSource.h
index 0970b2b..7b8f59d 100644
--- a/include/media/stagefright/MediaCodecSource.h
+++ b/include/media/stagefright/MediaCodecSource.h
@@ -25,6 +25,7 @@ namespace android {
 
 class ALooper;
 class AMessage;
+struct AReplyToken;
 class IGraphicBufferProducer;
 class MediaCodec;
 class MetaData;
@@ -99,7 +100,7 @@ private:
     sp<Puller> mPuller;
     sp<MediaCodec> mEncoder;
     uint32_t mFlags;
-    List<uint32_t> mStopReplyIDQueue;
+    List<sp<AReplyToken>> mStopReplyIDQueue;
     bool mIsVideo;
     bool mStarted;
     bool mStopping;
diff --git a/include/media/stagefright/foundation/ALooper.h b/include/media/stagefright/foundation/ALooper.h
index 150cdba..09c469b 100644
--- a/include/media/stagefright/foundation/ALooper.h
+++ b/include/media/stagefright/foundation/ALooper.h
@@ -30,6 +30,7 @@ namespace android {
 
 struct AHandler;
 struct AMessage;
+struct AReplyToken;
 
 struct ALooper : public RefBase {
     typedef int32_t event_id;
@@ -79,7 +80,27 @@ private:
     sp<LooperThread> mThread;
     bool mRunningLocally;
 
+    // use a separate lock for reply handling, as it is always on another thread
+    // use a central lock, however, to avoid creating a mutex for each reply
+    Mutex mRepliesLock;
+    Condition mRepliesCondition;
+
+    // START --- methods used only by AMessage
+
+    // posts a message on this looper with the given timeout
     void post(const sp<AMessage> &msg, int64_t delayUs);
+
+    // creates a reply token to be used with this looper
+    sp<AReplyToken> createReplyToken();
+    // waits for a response for the reply token.  If status is OK, the response
+    // is stored into the supplied variable.  Otherwise, it is unchanged.
+    status_t awaitResponse(const sp<AReplyToken> &replyToken, sp<AMessage> *response);
+    // posts a reply for a reply token.  If the reply could be successfully posted,
+    // it returns OK. Otherwise, it returns an error value.
+    status_t postReply(const sp<AReplyToken> &replyToken, const sp<AMessage> &msg);
+
+    // END --- methods used only by AMessage
+
     bool loop();
 
     DISALLOW_EVIL_CONSTRUCTORS(ALooper);
diff --git a/include/media/stagefright/foundation/ALooperRoster.h b/include/media/stagefright/foundation/ALooperRoster.h
index 96bce6b..9912455 100644
--- a/include/media/stagefright/foundation/ALooperRoster.h
+++ b/include/media/stagefright/foundation/ALooperRoster.h
@@ -33,11 +33,6 @@ struct ALooperRoster {
     void unregisterHandler(ALooper::handler_id handlerID);
     void unregisterStaleHandlers();
 
-    status_t postAndAwaitResponse(
-            const sp<AMessage> &msg, sp<AMessage> *response);
-
-    void postReply(uint32_t replyID, const sp<AMessage> &reply);
-
     void dump(int fd, const Vector<String16>& args);
 
 private:
@@ -49,10 +44,6 @@ private:
     Mutex mLock;
     KeyedVector<ALooper::handler_id, HandlerInfo> mHandlers;
     ALooper::handler_id mNextHandlerID;
-    uint32_t mNextReplyID;
-    Condition mRepliesCondition;
-
-    KeyedVector<uint32_t, sp<AMessage> > mReplies;
 
     DISALLOW_EVIL_CONSTRUCTORS(ALooperRoster);
 };
diff --git a/include/media/stagefright/foundation/AMessage.h b/include/media/stagefright/foundation/AMessage.h
index ab3279d..4c6bd21 100644
--- a/include/media/stagefright/foundation/AMessage.h
+++ b/include/media/stagefright/foundation/AMessage.h
@@ -30,6 +30,34 @@ struct AHandler;
 struct AString;
 struct Parcel;
 
+struct AReplyToken : public RefBase {
+    AReplyToken(const sp<ALooper> &looper)
+        : mLooper(looper),
+          mReplied(false) {
+    }
+
+private:
+    friend struct AMessage;
+    friend struct ALooper;
+    wp<ALooper> mLooper;
+    sp<AMessage> mReply;
+    bool mReplied;
+
+    sp<ALooper> getLooper() const {
+        return mLooper.promote();
+    }
+    // if reply is not set, returns false; otherwise, it retrieves the reply and returns true
+    bool retrieveReply(sp<AMessage> *reply) {
+        if (mReplied) {
+            *reply = mReply;
+            mReply.clear();
+        }
+        return mReplied;
+    }
+    // sets the reply for this token. returns OK or error
+    status_t setReply(const sp<AMessage> &reply);
+};
+
 struct AMessage : public RefBase {
     AMessage();
     AMessage(uint32_t what, const sp<const AHandler> &handler);
@@ -84,11 +112,15 @@ struct AMessage : public RefBase {
     status_t postAndAwaitResponse(sp<AMessage> *response);
 
     // If this returns true, the sender of this message is synchronously
-    // awaiting a response, the "replyID" can be used to send the response
-    // via "postReply" below.
-    bool senderAwaitsResponse(uint32_t *replyID) const;
-
-    void postReply(uint32_t replyID);
+    // awaiting a response and the reply token is consumed from the message
+    // and stored into replyID. The reply token must be used to send the response
+    // using "postReply" below.
+    bool senderAwaitsResponse(sp<AReplyToken> *replyID);
+
+    // Posts the message as a response to a reply token.  A reply token can
+    // only be used once. Returns OK if the response could be posted; otherwise,
+    // an error.
+    status_t postReply(const sp<AReplyToken> &replyID);
 
     // Performs a deep-copy of "this", contained messages are in turn "dup'ed".
     // Warning: RefBase items, i.e. "objects" are _not_ copied but only have