BufferQueue: Add StreamSplitter

Adds a StreamSplitter class, that takes one IGraphicBufferConsumer
interface and multiple IGraphicBufferProducer interfaces and
implements a one-to-many broadcast of GraphicBuffers (while managing
fences correctly).

Change-Id: I38ecdf3e311ac521bc781c30dde0cc382a4376a3

include/gui/IGraphicBufferConsumer.h

@@ -27,6 +27,9 @@
 #include <binder/IInterface.h>
 #include <ui/Rect.h>
 
+#include <EGL/egl.h>
+#include <EGL/eglext.h>
+
 namespace android {
 // ----------------------------------------------------------------------------
 

include/gui/StreamSplitter.h

@@ -0,0 +1,184 @@
+/*
+ * Copyright 2014 The Android Open Source Project
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+#ifndef ANDROID_GUI_STREAMSPLITTER_H
+#define ANDROID_GUI_STREAMSPLITTER_H
+
+#include <gui/IConsumerListener.h>
+#include <gui/IProducerListener.h>
+
+#include <utils/Condition.h>
+#include <utils/KeyedVector.h>
+#include <utils/Mutex.h>
+#include <utils/StrongPointer.h>
+
+namespace android {
+
+class GraphicBuffer;
+class IGraphicBufferConsumer;
+class IGraphicBufferProducer;
+
+// StreamSplitter is an autonomous class that manages one input BufferQueue
+// and multiple output BufferQueues. By using the buffer attach and detach logic
+// in BufferQueue, it is able to present the illusion of a single split
+// BufferQueue, where each buffer queued to the input is available to be
+// acquired by each of the outputs, and is able to be dequeued by the input
+// again only once all of the outputs have released it.
+class StreamSplitter : public BnConsumerListener {
+public:
+    // createSplitter creates a new splitter, outSplitter, using inputQueue as
+    // the input BufferQueue. Output BufferQueues must be added using addOutput
+    // before queueing any buffers to the input.
+    //
+    // A return value other than NO_ERROR means that an error has occurred and
+    // outSplitter has not been modified. BAD_VALUE is returned if inputQueue or
+    // outSplitter is NULL. See IGraphicBufferConsumer::consumerConnect for
+    // explanations of other error codes.
+    static status_t createSplitter(const sp<IGraphicBufferConsumer>& inputQueue,
+            sp<StreamSplitter>* outSplitter);
+
+    // addOutput adds an output BufferQueue to the splitter. The splitter
+    // connects to outputQueue as a CPU producer, and any buffers queued
+    // to the input will be queued to each output. It is assumed that all of the
+    // outputs are added before any buffers are queued on the input. If any
+    // output is abandoned by its consumer, the splitter will abandon its input
+    // queue (see onAbandoned).
+    //
+    // A return value other than NO_ERROR means that an error has occurred and
+    // outputQueue has not been added to the splitter. BAD_VALUE is returned if
+    // outputQueue is NULL. See IGraphicBufferProducer::connect for explanations
+    // of other error codes.
+    status_t addOutput(const sp<IGraphicBufferProducer>& outputQueue);
+
+    // setName sets the consumer name of the input queue
+    void setName(const String8& name);
+
+private:
+    // From IConsumerListener
+    //
+    // During this callback, we store some tracking information, detach the
+    // buffer from the input, and attach it to each of the outputs. This call
+    // can block if there are too many outstanding buffers. If it blocks, it
+    // will resume when onBufferReleasedByOutput releases a buffer back to the
+    // input.
+    virtual void onFrameAvailable();
+
+    // From IConsumerListener
+    // We don't care about released buffers because we detach each buffer as
+    // soon as we acquire it. See the comment for onBufferReleased below for
+    // some clarifying notes about the name.
+    virtual void onBuffersReleased() {}
+
+    // From IConsumerListener
+    // We don't care about sideband streams, since we won't be splitting them
+    virtual void onSidebandStreamChanged() {}
+
+    // This is the implementation of the onBufferReleased callback from
+    // IProducerListener. It gets called from an OutputListener (see below), and
+    // 'from' is which producer interface from which the callback was received.
+    //
+    // During this callback, we detach the buffer from the output queue that
+    // generated the callback, update our state tracking to see if this is the
+    // last output releasing the buffer, and if so, release it to the input.
+    // If we release the buffer to the input, we allow a blocked
+    // onFrameAvailable call to proceed.
+    void onBufferReleasedByOutput(const sp<IGraphicBufferProducer>& from);
+
+    // When this is called, the splitter disconnects from (i.e., abandons) its
+    // input queue and signals any waiting onFrameAvailable calls to wake up.
+    // It still processes callbacks from other outputs, but only detaches their
+    // buffers so they can continue operating until they run out of buffers to
+    // acquire. This must be called with mMutex locked.
+    void onAbandonedLocked();
+
+    // This is a thin wrapper class that lets us determine which BufferQueue
+    // the IProducerListener::onBufferReleased callback is associated with. We
+    // create one of these per output BufferQueue, and then pass the producer
+    // into onBufferReleasedByOutput above.
+    class OutputListener : public BnProducerListener,
+                           public IBinder::DeathRecipient {
+    public:
+        OutputListener(const sp<StreamSplitter>& splitter,
+                const sp<IGraphicBufferProducer>& output);
+        virtual ~OutputListener();
+
+        // From IProducerListener
+        virtual void onBufferReleased();
+
+        // From IBinder::DeathRecipient
+        virtual void binderDied(const wp<IBinder>& who);
+
+    private:
+        sp<StreamSplitter> mSplitter;
+        sp<IGraphicBufferProducer> mOutput;
+    };
+
+    class BufferTracker : public LightRefBase<BufferTracker> {
+    public:
+        BufferTracker(const sp<GraphicBuffer>& buffer);
+
+        const sp<GraphicBuffer>& getBuffer() const { return mBuffer; }
+        const sp<Fence>& getMergedFence() const { return mMergedFence; }
+
+        void mergeFence(const sp<Fence>& with);
+
+        // Returns the new value
+        // Only called while mMutex is held
+        size_t incrementReleaseCountLocked() { return ++mReleaseCount; }
+
+    private:
+        // Only destroy through LightRefBase
+        friend LightRefBase<BufferTracker>;
+        ~BufferTracker();
+
+        // Disallow copying
+        BufferTracker(const BufferTracker& other);
+        BufferTracker& operator=(const BufferTracker& other);
+
+        sp<GraphicBuffer> mBuffer; // One instance that holds this native handle
+        sp<Fence> mMergedFence;
+        size_t mReleaseCount;
+    };
+
+    // Only called from createSplitter
+    StreamSplitter(const sp<IGraphicBufferConsumer>& inputQueue);
+
+    // Must be accessed through RefBase
+    virtual ~StreamSplitter();
+
+    static const int MAX_OUTSTANDING_BUFFERS = 2;
+
+    // mIsAbandoned is set to true when an output dies. Once the StreamSplitter
+    // has been abandoned, it will continue to detach buffers from other
+    // outputs, but it will disconnect from the input and not attempt to
+    // communicate with it further.
+    bool mIsAbandoned;
+
+    Mutex mMutex;
+    Condition mReleaseCondition;
+    int mOutstandingBuffers;
+    sp<IGraphicBufferConsumer> mInput;
+    Vector<sp<IGraphicBufferProducer> > mOutputs;
+
+    // Map of GraphicBuffer IDs (GraphicBuffer::getId()) to buffer tracking
+    // objects (which are mostly for counting how many outputs have released the
+    // buffer, but also contain merged release fences).
+    KeyedVector<uint64_t, sp<BufferTracker> > mBuffers;
+};
+
+} // namespace android
+
+#endif

libs/gui/Android.mk

@@ -30,6 +30,7 @@ LOCAL_SRC_FILES:= \
 	Sensor.cpp \
 	SensorEventQueue.cpp \
 	SensorManager.cpp \
+	StreamSplitter.cpp \
 	Surface.cpp \
 	SurfaceControl.cpp \
 	SurfaceComposerClient.cpp \

libs/gui/BufferQueueConsumer.cpp

@@ -243,11 +243,27 @@ status_t BufferQueueConsumer::attachBuffer(int* outSlot,
     mSlots[*outSlot].mGraphicBuffer = buffer;
     mSlots[*outSlot].mBufferState = BufferSlot::ACQUIRED;
     mSlots[*outSlot].mAttachedByConsumer = true;
-    mSlots[*outSlot].mAcquireCalled = true;
     mSlots[*outSlot].mNeedsCleanupOnRelease = false;
     mSlots[*outSlot].mFence = Fence::NO_FENCE;
     mSlots[*outSlot].mFrameNumber = 0;
 
+    // mAcquireCalled tells BufferQueue that it doesn't need to send a valid
+    // GraphicBuffer pointer on the next acquireBuffer call, which decreases
+    // Binder traffic by not un/flattening the GraphicBuffer. However, it
+    // requires that the consumer maintain a cached copy of the slot <--> buffer
+    // mappings, which is why the consumer doesn't need the valid pointer on
+    // acquire.
+    //
+    // The StreamSplitter is one of the primary users of the attach/detach
+    // logic, and while it is running, all buffers it acquires are immediately
+    // detached, and all buffers it eventually releases are ones that were
+    // attached (as opposed to having been obtained from acquireBuffer), so it
+    // doesn't make sense to maintain the slot/buffer mappings, which would
+    // become invalid for every buffer during detach/attach. By setting this to
+    // false, the valid GraphicBuffer pointer will always be sent with acquire
+    // for attached buffers.
+    mSlots[*outSlot].mAcquireCalled = false;
+
     return NO_ERROR;
 }
 

libs/gui/BufferQueueCore.cpp

@@ -191,7 +191,7 @@ void BufferQueueCore::freeBufferLocked(int slot) {
         mSlots[slot].mNeedsCleanupOnRelease = true;
     }
     mSlots[slot].mBufferState = BufferSlot::FREE;
-    mSlots[slot].mFrameNumber = 0;
+    mSlots[slot].mFrameNumber = UINT32_MAX;
     mSlots[slot].mAcquireCalled = false;
 
     // Destroy fence as BufferQueue now takes ownership

libs/gui/IGraphicBufferConsumer.cpp

@@ -14,12 +14,6 @@
  * limitations under the License.
  */
 
-#define EGL_EGLEXT_PROTOTYPES
-
-#include <EGL/egl.h>
-#include <EGL/eglext.h>
-
-
 #include <stdint.h>
 #include <sys/types.h>
 

libs/gui/StreamSplitter.cpp

@@ -0,0 +1,281 @@
+/*
+ * Copyright 2014 The Android Open Source Project
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+#define LOG_TAG "StreamSplitter"
+#define ATRACE_TAG ATRACE_TAG_GRAPHICS
+//#define LOG_NDEBUG 0
+
+#include <gui/IGraphicBufferConsumer.h>
+#include <gui/IGraphicBufferProducer.h>
+#include <gui/StreamSplitter.h>
+
+#include <ui/GraphicBuffer.h>
+
+#include <binder/ProcessState.h>
+
+#include <utils/Trace.h>
+
+namespace android {
+
+status_t StreamSplitter::createSplitter(
+        const sp<IGraphicBufferConsumer>& inputQueue,
+        sp<StreamSplitter>* outSplitter) {
+    if (inputQueue == NULL) {
+        ALOGE("createSplitter: inputQueue must not be NULL");
+        return BAD_VALUE;
+    }
+    if (outSplitter == NULL) {
+        ALOGE("createSplitter: outSplitter must not be NULL");
+        return BAD_VALUE;
+    }
+
+    sp<StreamSplitter> splitter(new StreamSplitter(inputQueue));
+    status_t status = splitter->mInput->consumerConnect(splitter, false);
+    if (status == NO_ERROR) {
+        splitter->mInput->setConsumerName(String8("StreamSplitter"));
+        *outSplitter = splitter;
+    }
+    return status;
+}
+
+StreamSplitter::StreamSplitter(const sp<IGraphicBufferConsumer>& inputQueue)
+      : mIsAbandoned(false), mMutex(), mReleaseCondition(),
+        mOutstandingBuffers(0), mInput(inputQueue), mOutputs(), mBuffers() {}
+
+StreamSplitter::~StreamSplitter() {
+    mInput->consumerDisconnect();
+    Vector<sp<IGraphicBufferProducer> >::iterator output = mOutputs.begin();
+    for (; output != mOutputs.end(); ++output) {
+        (*output)->disconnect(NATIVE_WINDOW_API_CPU);
+    }
+
+    if (mBuffers.size() > 0) {
+        ALOGE("%d buffers still being tracked", mBuffers.size());
+    }
+}
+
+status_t StreamSplitter::addOutput(
+        const sp<IGraphicBufferProducer>& outputQueue) {
+    if (outputQueue == NULL) {
+        ALOGE("addOutput: outputQueue must not be NULL");
+        return BAD_VALUE;
+    }
+
+    Mutex::Autolock lock(mMutex);
+
+    IGraphicBufferProducer::QueueBufferOutput queueBufferOutput;
+    sp<OutputListener> listener(new OutputListener(this, outputQueue));
+    outputQueue->asBinder()->linkToDeath(listener);
+    status_t status = outputQueue->connect(listener, NATIVE_WINDOW_API_CPU,
+            /* producerControlledByApp */ false, &queueBufferOutput);
+    if (status != NO_ERROR) {
+        ALOGE("addOutput: failed to connect (%d)", status);
+        return status;
+    }
+
+    mOutputs.push_back(outputQueue);
+
+    return NO_ERROR;
+}
+
+void StreamSplitter::setName(const String8 &name) {
+    Mutex::Autolock lock(mMutex);
+    mInput->setConsumerName(name);
+}
+
+void StreamSplitter::onFrameAvailable() {
+    ATRACE_CALL();
+    Mutex::Autolock lock(mMutex);
+
+    // The current policy is that if any one consumer is consuming buffers too
+    // slowly, the splitter will stall the rest of the outputs by not acquiring
+    // any more buffers from the input. This will cause back pressure on the
+    // input queue, slowing down its producer.
+
+    // If there are too many outstanding buffers, we block until a buffer is
+    // released back to the input in onBufferReleased
+    while (mOutstandingBuffers >= MAX_OUTSTANDING_BUFFERS) {
+        mReleaseCondition.wait(mMutex);
+
+        // If the splitter is abandoned while we are waiting, the release
+        // condition variable will be broadcast, and we should just return
+        // without attempting to do anything more (since the input queue will
+        // also be abandoned).
+        if (mIsAbandoned) {
+            return;
+        }
+    }
+    ++mOutstandingBuffers;
+
+    // Acquire and detach the buffer from the input
+    IGraphicBufferConsumer::BufferItem bufferItem;
+    status_t status = mInput->acquireBuffer(&bufferItem, /* presentWhen */ 0);
+    LOG_ALWAYS_FATAL_IF(status != NO_ERROR,
+            "acquiring buffer from input failed (%d)", status);
+
+    ALOGV("acquired buffer %#llx from input",
+            bufferItem.mGraphicBuffer->getId());
+
+    status = mInput->detachBuffer(bufferItem.mBuf);
+    LOG_ALWAYS_FATAL_IF(status != NO_ERROR,
+            "detaching buffer from input failed (%d)", status);
+
+    // Initialize our reference count for this buffer
+    mBuffers.add(bufferItem.mGraphicBuffer->getId(),
+            new BufferTracker(bufferItem.mGraphicBuffer));
+
+    IGraphicBufferProducer::QueueBufferInput queueInput(
+            bufferItem.mTimestamp, bufferItem.mIsAutoTimestamp,
+            bufferItem.mCrop, bufferItem.mScalingMode,
+            bufferItem.mTransform, bufferItem.mIsDroppable,
+            bufferItem.mFence);
+
+    // Attach and queue the buffer to each of the outputs
+    Vector<sp<IGraphicBufferProducer> >::iterator output = mOutputs.begin();
+    for (; output != mOutputs.end(); ++output) {
+        int slot;
+        status = (*output)->attachBuffer(&slot, bufferItem.mGraphicBuffer);
+        if (status == NO_INIT) {
+            // If we just discovered that this output has been abandoned, note
+            // that, increment the release count so that we still release this
+            // buffer eventually, and move on to the next output
+            onAbandonedLocked();
+            mBuffers.editValueFor(bufferItem.mGraphicBuffer->getId())->
+                    incrementReleaseCountLocked();
+            continue;
+        } else {
+            LOG_ALWAYS_FATAL_IF(status != NO_ERROR,
+                    "attaching buffer to output failed (%d)", status);
+        }
+
+        IGraphicBufferProducer::QueueBufferOutput queueOutput;
+        status = (*output)->queueBuffer(slot, queueInput, &queueOutput);
+        if (status == NO_INIT) {
+            // If we just discovered that this output has been abandoned, note
+            // that, increment the release count so that we still release this
+            // buffer eventually, and move on to the next output
+            onAbandonedLocked();
+            mBuffers.editValueFor(bufferItem.mGraphicBuffer->getId())->
+                    incrementReleaseCountLocked();
+            continue;
+        } else {
+            LOG_ALWAYS_FATAL_IF(status != NO_ERROR,
+                    "queueing buffer to output failed (%d)", status);
+        }
+
+        ALOGV("queued buffer %#llx to output %p",
+                bufferItem.mGraphicBuffer->getId(), output->get());
+    }
+}
+
+void StreamSplitter::onBufferReleasedByOutput(
+        const sp<IGraphicBufferProducer>& from) {
+    ATRACE_CALL();
+    Mutex::Autolock lock(mMutex);
+
+    sp<GraphicBuffer> buffer;
+    sp<Fence> fence;
+    status_t status = from->detachNextBuffer(&buffer, &fence);
+    if (status == NO_INIT) {
+        // If we just discovered that this output has been abandoned, note that,
+        // but we can't do anything else, since buffer is invalid
+        onAbandonedLocked();
+        return;
+    } else {
+        LOG_ALWAYS_FATAL_IF(status != NO_ERROR,
+                "detaching buffer from output failed (%d)", status);
+    }
+
+    ALOGV("detached buffer %#llx from output %p", buffer->getId(), from.get());
+
+    const sp<BufferTracker>& tracker = mBuffers.editValueFor(buffer->getId());
+
+    // Merge the release fence of the incoming buffer so that the fence we send
+    // back to the input includes all of the outputs' fences
+    tracker->mergeFence(fence);
+
+    // Check to see if this is the last outstanding reference to this buffer
+    size_t releaseCount = tracker->incrementReleaseCountLocked();
+    ALOGV("buffer %#llx reference count %d (of %d)", buffer->getId(),
+            releaseCount, mOutputs.size());
+    if (releaseCount < mOutputs.size()) {
+        return;
+    }
+
+    // If we've been abandoned, we can't return the buffer to the input, so just
+    // stop tracking it and move on
+    if (mIsAbandoned) {
+        mBuffers.removeItem(buffer->getId());
+        return;
+    }
+
+    // Attach and release the buffer back to the input
+    int consumerSlot;
+    status = mInput->attachBuffer(&consumerSlot, tracker->getBuffer());
+    LOG_ALWAYS_FATAL_IF(status != NO_ERROR,
+            "attaching buffer to input failed (%d)", status);
+
+    status = mInput->releaseBuffer(consumerSlot, /* frameNumber */ 0,
+            EGL_NO_DISPLAY, EGL_NO_SYNC_KHR, tracker->getMergedFence());
+    LOG_ALWAYS_FATAL_IF(status != NO_ERROR,
+            "releasing buffer to input failed (%d)", status);
+
+    ALOGV("released buffer %#llx to input", buffer->getId());
+
+    // We no longer need to track the buffer once it has been returned to the
+    // input
+    mBuffers.removeItem(buffer->getId());
+
+    // Notify any waiting onFrameAvailable calls
+    --mOutstandingBuffers;
+    mReleaseCondition.signal();
+}
+
+void StreamSplitter::onAbandonedLocked() {
+    ALOGE("one of my outputs has abandoned me");
+    if (!mIsAbandoned) {
+        mInput->consumerDisconnect();
+    }
+    mIsAbandoned = true;
+    mReleaseCondition.broadcast();
+}
+
+StreamSplitter::OutputListener::OutputListener(
+        const sp<StreamSplitter>& splitter,
+        const sp<IGraphicBufferProducer>& output)
+      : mSplitter(splitter), mOutput(output) {}
+
+StreamSplitter::OutputListener::~OutputListener() {}
+
+void StreamSplitter::OutputListener::onBufferReleased() {
+    mSplitter->onBufferReleasedByOutput(mOutput);
+}
+
+void StreamSplitter::OutputListener::binderDied(const wp<IBinder>& /* who */) {
+    Mutex::Autolock lock(mSplitter->mMutex);
+    mSplitter->onAbandonedLocked();
+}
+
+StreamSplitter::BufferTracker::BufferTracker(const sp<GraphicBuffer>& buffer)
+      : mBuffer(buffer), mMergedFence(Fence::NO_FENCE), mReleaseCount(0) {}
+
+StreamSplitter::BufferTracker::~BufferTracker() {}
+
+void StreamSplitter::BufferTracker::mergeFence(const sp<Fence>& with) {
+    mMergedFence = Fence::merge(String8("StreamSplitter"), mMergedFence, with);
+}
+
+} // namespace android

libs/gui/tests/Android.mk

@@ -14,6 +14,7 @@ LOCAL_SRC_FILES := \
     IGraphicBufferProducer_test.cpp \
     MultiTextureConsumer_test.cpp \
     SRGB_test.cpp \
+    StreamSplitter_test.cpp \
     SurfaceTextureClient_test.cpp \
     SurfaceTextureFBO_test.cpp \
     SurfaceTextureGLThreadToGL_test.cpp \

libs/gui/tests/BufferQueue_test.cpp

@@ -301,7 +301,7 @@ TEST_F(BufferQueueTest, DetachAndReattachOnConsumerSide) {
     ASSERT_EQ(BAD_VALUE, mConsumer->attachBuffer(&newSlot, NULL));
     ASSERT_EQ(OK, mConsumer->attachBuffer(&newSlot, item.mGraphicBuffer));
 
-    ASSERT_EQ(OK, mConsumer->releaseBuffer(item.mBuf, 0, EGL_NO_DISPLAY,
+    ASSERT_EQ(OK, mConsumer->releaseBuffer(newSlot, 0, EGL_NO_DISPLAY,
             EGL_NO_SYNC_KHR, Fence::NO_FENCE));
 
     ASSERT_EQ(IGraphicBufferProducer::BUFFER_NEEDS_REALLOCATION,

libs/gui/tests/StreamSplitter_test.cpp

@@ -0,0 +1,246 @@
+/*
+ * Copyright 2014 The Android Open Source Project
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+#define LOG_TAG "StreamSplitter_test"
+//#define LOG_NDEBUG 0
+
+#include <gui/BufferQueue.h>
+#include <gui/IConsumerListener.h>
+#include <gui/ISurfaceComposer.h>
+#include <gui/StreamSplitter.h>
+#include <private/gui/ComposerService.h>
+
+#include <gtest/gtest.h>
+
+namespace android {
+
+class StreamSplitterTest : public ::testing::Test {
+
+protected:
+    StreamSplitterTest() {
+        const ::testing::TestInfo* const testInfo =
+            ::testing::UnitTest::GetInstance()->current_test_info();
+        ALOGV("Begin test: %s.%s", testInfo->test_case_name(),
+                testInfo->name());
+    }
+
+    ~StreamSplitterTest() {
+        const ::testing::TestInfo* const testInfo =
+            ::testing::UnitTest::GetInstance()->current_test_info();
+        ALOGV("End test:   %s.%s", testInfo->test_case_name(),
+                testInfo->name());
+    }
+};
+
+struct DummyListener : public BnConsumerListener {
+    virtual void onFrameAvailable() {}
+    virtual void onBuffersReleased() {}
+    virtual void onSidebandStreamChanged() {}
+};
+
+class CountedAllocator : public BnGraphicBufferAlloc {
+public:
+    CountedAllocator() : mAllocCount(0) {
+        sp<ISurfaceComposer> composer(ComposerService::getComposerService());
+        mAllocator = composer->createGraphicBufferAlloc();
+    }
+
+    virtual ~CountedAllocator() {}
+
+    virtual sp<GraphicBuffer> createGraphicBuffer(uint32_t w, uint32_t h,
+            PixelFormat format, uint32_t usage, status_t* error) {
+        ++mAllocCount;
+        sp<GraphicBuffer> buffer = mAllocator->createGraphicBuffer(w, h, format,
+                usage, error);
+        return buffer;
+    }
+
+    int getAllocCount() const { return mAllocCount; }
+
+private:
+    sp<IGraphicBufferAlloc> mAllocator;
+    int mAllocCount;
+};
+
+TEST_F(StreamSplitterTest, OneInputOneOutput) {
+    sp<CountedAllocator> allocator(new CountedAllocator);
+
+    sp<IGraphicBufferProducer> inputProducer;
+    sp<IGraphicBufferConsumer> inputConsumer;
+    BufferQueue::createBufferQueue(&inputProducer, &inputConsumer, allocator);
+
+    sp<IGraphicBufferProducer> outputProducer;
+    sp<IGraphicBufferConsumer> outputConsumer;
+    BufferQueue::createBufferQueue(&outputProducer, &outputConsumer, allocator);
+    ASSERT_EQ(OK, outputConsumer->consumerConnect(new DummyListener, false));
+
+    sp<StreamSplitter> splitter;
+    status_t status = StreamSplitter::createSplitter(inputConsumer, &splitter);
+    ASSERT_EQ(OK, status);
+    ASSERT_EQ(OK, splitter->addOutput(outputProducer));
+
+    IGraphicBufferProducer::QueueBufferOutput qbOutput;
+    ASSERT_EQ(OK, inputProducer->connect(new DummyProducerListener,
+            NATIVE_WINDOW_API_CPU, false, &qbOutput));
+
+    int slot;
+    sp<Fence> fence;
+    sp<GraphicBuffer> buffer;
+    ASSERT_EQ(IGraphicBufferProducer::BUFFER_NEEDS_REALLOCATION,
+            inputProducer->dequeueBuffer(&slot, &fence, false, 0, 0, 0,
+                    GRALLOC_USAGE_SW_WRITE_OFTEN));
+    ASSERT_EQ(OK, inputProducer->requestBuffer(slot, &buffer));
+
+    uint32_t* dataIn;
+    ASSERT_EQ(OK, buffer->lock(GraphicBuffer::USAGE_SW_WRITE_OFTEN,
+            reinterpret_cast<void**>(&dataIn)));
+    *dataIn = 0x12345678;
+    ASSERT_EQ(OK, buffer->unlock());
+
+    IGraphicBufferProducer::QueueBufferInput qbInput(0, false,
+            Rect(0, 0, 1, 1), NATIVE_WINDOW_SCALING_MODE_FREEZE, 0, false,
+            Fence::NO_FENCE);
+    ASSERT_EQ(OK, inputProducer->queueBuffer(slot, qbInput, &qbOutput));
+
+    IGraphicBufferConsumer::BufferItem item;
+    ASSERT_EQ(OK, outputConsumer->acquireBuffer(&item, 0));
+
+    uint32_t* dataOut;
+    ASSERT_EQ(OK, item.mGraphicBuffer->lock(GraphicBuffer::USAGE_SW_READ_OFTEN,
+            reinterpret_cast<void**>(&dataOut)));
+    ASSERT_EQ(*dataOut, 0x12345678);
+    ASSERT_EQ(OK, item.mGraphicBuffer->unlock());
+
+    ASSERT_EQ(OK, outputConsumer->releaseBuffer(item.mBuf, item.mFrameNumber,
+            EGL_NO_DISPLAY, EGL_NO_SYNC_KHR, Fence::NO_FENCE));
+
+    ASSERT_EQ(IGraphicBufferProducer::BUFFER_NEEDS_REALLOCATION,
+            inputProducer->dequeueBuffer(&slot, &fence, false, 0, 0, 0,
+                    GRALLOC_USAGE_SW_WRITE_OFTEN));
+
+    ASSERT_EQ(1, allocator->getAllocCount());
+}
+
+TEST_F(StreamSplitterTest, OneInputMultipleOutputs) {
+    const int NUM_OUTPUTS = 4;
+    sp<CountedAllocator> allocator(new CountedAllocator);
+
+    sp<IGraphicBufferProducer> inputProducer;
+    sp<IGraphicBufferConsumer> inputConsumer;
+    BufferQueue::createBufferQueue(&inputProducer, &inputConsumer, allocator);
+
+    sp<IGraphicBufferProducer> outputProducers[NUM_OUTPUTS] = {};
+    sp<IGraphicBufferConsumer> outputConsumers[NUM_OUTPUTS] = {};
+    for (int output = 0; output < NUM_OUTPUTS; ++output) {
+        BufferQueue::createBufferQueue(&outputProducers[output],
+                &outputConsumers[output], allocator);
+        ASSERT_EQ(OK, outputConsumers[output]->consumerConnect(
+                    new DummyListener, false));
+    }
+
+    sp<StreamSplitter> splitter;
+    status_t status = StreamSplitter::createSplitter(inputConsumer, &splitter);
+    ASSERT_EQ(OK, status);
+    for (int output = 0; output < NUM_OUTPUTS; ++output) {
+        ASSERT_EQ(OK, splitter->addOutput(outputProducers[output]));
+    }
+
+    IGraphicBufferProducer::QueueBufferOutput qbOutput;
+    ASSERT_EQ(OK, inputProducer->connect(new DummyProducerListener,
+            NATIVE_WINDOW_API_CPU, false, &qbOutput));
+
+    int slot;
+    sp<Fence> fence;
+    sp<GraphicBuffer> buffer;
+    ASSERT_EQ(IGraphicBufferProducer::BUFFER_NEEDS_REALLOCATION,
+            inputProducer->dequeueBuffer(&slot, &fence, false, 0, 0, 0,
+                    GRALLOC_USAGE_SW_WRITE_OFTEN));
+    ASSERT_EQ(OK, inputProducer->requestBuffer(slot, &buffer));
+
+    uint32_t* dataIn;
+    ASSERT_EQ(OK, buffer->lock(GraphicBuffer::USAGE_SW_WRITE_OFTEN,
+            reinterpret_cast<void**>(&dataIn)));
+    *dataIn = 0x12345678;
+    ASSERT_EQ(OK, buffer->unlock());
+
+    IGraphicBufferProducer::QueueBufferInput qbInput(0, false,
+            Rect(0, 0, 1, 1), NATIVE_WINDOW_SCALING_MODE_FREEZE, 0, false,
+            Fence::NO_FENCE);
+    ASSERT_EQ(OK, inputProducer->queueBuffer(slot, qbInput, &qbOutput));
+
+    for (int output = 0; output < NUM_OUTPUTS; ++output) {
+        IGraphicBufferConsumer::BufferItem item;
+        ASSERT_EQ(OK, outputConsumers[output]->acquireBuffer(&item, 0));
+
+        uint32_t* dataOut;
+        ASSERT_EQ(OK, item.mGraphicBuffer->lock(GraphicBuffer::USAGE_SW_READ_OFTEN,
+                    reinterpret_cast<void**>(&dataOut)));
+        ASSERT_EQ(*dataOut, 0x12345678);
+        ASSERT_EQ(OK, item.mGraphicBuffer->unlock());
+
+        ASSERT_EQ(OK, outputConsumers[output]->releaseBuffer(item.mBuf,
+                    item.mFrameNumber, EGL_NO_DISPLAY, EGL_NO_SYNC_KHR,
+                    Fence::NO_FENCE));
+    }
+
+    ASSERT_EQ(IGraphicBufferProducer::BUFFER_NEEDS_REALLOCATION,
+            inputProducer->dequeueBuffer(&slot, &fence, false, 0, 0, 0,
+                    GRALLOC_USAGE_SW_WRITE_OFTEN));
+
+    ASSERT_EQ(1, allocator->getAllocCount());
+}
+
+TEST_F(StreamSplitterTest, OutputAbandonment) {
+    sp<IGraphicBufferProducer> inputProducer;
+    sp<IGraphicBufferConsumer> inputConsumer;
+    BufferQueue::createBufferQueue(&inputProducer, &inputConsumer);
+
+    sp<IGraphicBufferProducer> outputProducer;
+    sp<IGraphicBufferConsumer> outputConsumer;
+    BufferQueue::createBufferQueue(&outputProducer, &outputConsumer);
+    ASSERT_EQ(OK, outputConsumer->consumerConnect(new DummyListener, false));
+
+    sp<StreamSplitter> splitter;
+    status_t status = StreamSplitter::createSplitter(inputConsumer, &splitter);
+    ASSERT_EQ(OK, status);
+    ASSERT_EQ(OK, splitter->addOutput(outputProducer));
+
+    IGraphicBufferProducer::QueueBufferOutput qbOutput;
+    ASSERT_EQ(OK, inputProducer->connect(new DummyProducerListener,
+            NATIVE_WINDOW_API_CPU, false, &qbOutput));
+
+    int slot;
+    sp<Fence> fence;
+    sp<GraphicBuffer> buffer;
+    ASSERT_EQ(IGraphicBufferProducer::BUFFER_NEEDS_REALLOCATION,
+            inputProducer->dequeueBuffer(&slot, &fence, false, 0, 0, 0,
+                    GRALLOC_USAGE_SW_WRITE_OFTEN));
+    ASSERT_EQ(OK, inputProducer->requestBuffer(slot, &buffer));
+
+    // Abandon the output
+    outputConsumer->consumerDisconnect();
+
+    IGraphicBufferProducer::QueueBufferInput qbInput(0, false,
+            Rect(0, 0, 1, 1), NATIVE_WINDOW_SCALING_MODE_FREEZE, 0, false,
+            Fence::NO_FENCE);
+    ASSERT_EQ(OK, inputProducer->queueBuffer(slot, qbInput, &qbOutput));
+
+    // Input should be abandoned
+    ASSERT_EQ(NO_INIT, inputProducer->dequeueBuffer(&slot, &fence, false, 0, 0,
+            0, GRALLOC_USAGE_SW_WRITE_OFTEN));
+}
+
+} // namespace android