StreamSplitter.cpp revision 8dc55396fc9bc425b5e2c82e76a38080f2a655ff 
1/*
2 * Copyright 2014 The Android Open Source Project
3 *
4 * Licensed under the Apache License, Version 2.0 (the "License");
5 * you may not use this file except in compliance with the License.
6 * You may obtain a copy of the License at
7 *
8 *      http://www.apache.org/licenses/LICENSE-2.0
9 *
10 * Unless required by applicable law or agreed to in writing, software
11 * distributed under the License is distributed on an "AS IS" BASIS,
12 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13 * See the License for the specific language governing permissions and
14 * limitations under the License.
15 */
16
17#include <inttypes.h>
18
19#define LOG_TAG "StreamSplitter"
20#define ATRACE_TAG ATRACE_TAG_GRAPHICS
21//#define LOG_NDEBUG 0
22
23#include <gui/IGraphicBufferConsumer.h>
24#include <gui/IGraphicBufferProducer.h>
25#include <gui/StreamSplitter.h>
26
27#include <ui/GraphicBuffer.h>
28
29#include <binder/ProcessState.h>
30
31#include <utils/Trace.h>
32
33namespace android {
34
35status_t StreamSplitter::createSplitter(
36        const sp<IGraphicBufferConsumer>& inputQueue,
37        sp<StreamSplitter>* outSplitter) {
38    if (inputQueue == NULL) {
39        ALOGE("createSplitter: inputQueue must not be NULL");
40        return BAD_VALUE;
41    }
42    if (outSplitter == NULL) {
43        ALOGE("createSplitter: outSplitter must not be NULL");
44        return BAD_VALUE;
45    }
46
47    sp<StreamSplitter> splitter(new StreamSplitter(inputQueue));
48    status_t status = splitter->mInput->consumerConnect(splitter, false);
49    if (status == NO_ERROR) {
50        splitter->mInput->setConsumerName(String8("StreamSplitter"));
51        *outSplitter = splitter;
52    }
53    return status;
54}
55
56StreamSplitter::StreamSplitter(const sp<IGraphicBufferConsumer>& inputQueue)
57      : mIsAbandoned(false), mMutex(), mReleaseCondition(),
58        mOutstandingBuffers(0), mInput(inputQueue), mOutputs(), mBuffers() {}
59
60StreamSplitter::~StreamSplitter() {
61    mInput->consumerDisconnect();
62    Vector<sp<IGraphicBufferProducer> >::iterator output = mOutputs.begin();
63    for (; output != mOutputs.end(); ++output) {
64        (*output)->disconnect(NATIVE_WINDOW_API_CPU);
65    }
66
67    if (mBuffers.size() > 0) {
68        ALOGE("%zu buffers still being tracked", mBuffers.size());
69    }
70}
71
72status_t StreamSplitter::addOutput(
73        const sp<IGraphicBufferProducer>& outputQueue) {
74    if (outputQueue == NULL) {
75        ALOGE("addOutput: outputQueue must not be NULL");
76        return BAD_VALUE;
77    }
78
79    Mutex::Autolock lock(mMutex);
80
81    IGraphicBufferProducer::QueueBufferOutput queueBufferOutput;
82    sp<OutputListener> listener(new OutputListener(this, outputQueue));
83    outputQueue->asBinder()->linkToDeath(listener);
84    status_t status = outputQueue->connect(listener, NATIVE_WINDOW_API_CPU,
85            /* producerControlledByApp */ false, &queueBufferOutput);
86    if (status != NO_ERROR) {
87        ALOGE("addOutput: failed to connect (%d)", status);
88        return status;
89    }
90
91    mOutputs.push_back(outputQueue);
92
93    return NO_ERROR;
94}
95
96void StreamSplitter::setName(const String8 &name) {
97    Mutex::Autolock lock(mMutex);
98    mInput->setConsumerName(name);
99}
100
101void StreamSplitter::onFrameAvailable(const BufferItem& /* item */) {
102    ATRACE_CALL();
103    Mutex::Autolock lock(mMutex);
104
105    // The current policy is that if any one consumer is consuming buffers too
106    // slowly, the splitter will stall the rest of the outputs by not acquiring
107    // any more buffers from the input. This will cause back pressure on the
108    // input queue, slowing down its producer.
109
110    // If there are too many outstanding buffers, we block until a buffer is
111    // released back to the input in onBufferReleased
112    while (mOutstandingBuffers >= MAX_OUTSTANDING_BUFFERS) {
113        mReleaseCondition.wait(mMutex);
114
115        // If the splitter is abandoned while we are waiting, the release
116        // condition variable will be broadcast, and we should just return
117        // without attempting to do anything more (since the input queue will
118        // also be abandoned).
119        if (mIsAbandoned) {
120            return;
121        }
122    }
123    ++mOutstandingBuffers;
124
125    // Acquire and detach the buffer from the input
126    IGraphicBufferConsumer::BufferItem bufferItem;
127    status_t status = mInput->acquireBuffer(&bufferItem, /* presentWhen */ 0);
128    LOG_ALWAYS_FATAL_IF(status != NO_ERROR,
129            "acquiring buffer from input failed (%d)", status);
130
131    ALOGV("acquired buffer %#" PRIx64 " from input",
132            bufferItem.mGraphicBuffer->getId());
133
134    status = mInput->detachBuffer(bufferItem.mBuf);
135    LOG_ALWAYS_FATAL_IF(status != NO_ERROR,
136            "detaching buffer from input failed (%d)", status);
137
138    // Initialize our reference count for this buffer
139    mBuffers.add(bufferItem.mGraphicBuffer->getId(),
140            new BufferTracker(bufferItem.mGraphicBuffer));
141
142    IGraphicBufferProducer::QueueBufferInput queueInput(
143            bufferItem.mTimestamp, bufferItem.mIsAutoTimestamp,
144            bufferItem.mCrop, bufferItem.mScalingMode,
145            bufferItem.mTransform, bufferItem.mIsDroppable,
146            bufferItem.mFence);
147
148    // Attach and queue the buffer to each of the outputs
149    Vector<sp<IGraphicBufferProducer> >::iterator output = mOutputs.begin();
150    for (; output != mOutputs.end(); ++output) {
151        int slot;
152        status = (*output)->attachBuffer(&slot, bufferItem.mGraphicBuffer);
153        if (status == NO_INIT) {
154            // If we just discovered that this output has been abandoned, note
155            // that, increment the release count so that we still release this
156            // buffer eventually, and move on to the next output
157            onAbandonedLocked();
158            mBuffers.editValueFor(bufferItem.mGraphicBuffer->getId())->
159                    incrementReleaseCountLocked();
160            continue;
161        } else {
162            LOG_ALWAYS_FATAL_IF(status != NO_ERROR,
163                    "attaching buffer to output failed (%d)", status);
164        }
165
166        IGraphicBufferProducer::QueueBufferOutput queueOutput;
167        status = (*output)->queueBuffer(slot, queueInput, &queueOutput);
168        if (status == NO_INIT) {
169            // If we just discovered that this output has been abandoned, note
170            // that, increment the release count so that we still release this
171            // buffer eventually, and move on to the next output
172            onAbandonedLocked();
173            mBuffers.editValueFor(bufferItem.mGraphicBuffer->getId())->
174                    incrementReleaseCountLocked();
175            continue;
176        } else {
177            LOG_ALWAYS_FATAL_IF(status != NO_ERROR,
178                    "queueing buffer to output failed (%d)", status);
179        }
180
181        ALOGV("queued buffer %#" PRIx64 " to output %p",
182                bufferItem.mGraphicBuffer->getId(), output->get());
183    }
184}
185
186void StreamSplitter::onBufferReleasedByOutput(
187        const sp<IGraphicBufferProducer>& from) {
188    ATRACE_CALL();
189    Mutex::Autolock lock(mMutex);
190
191    sp<GraphicBuffer> buffer;
192    sp<Fence> fence;
193    status_t status = from->detachNextBuffer(&buffer, &fence);
194    if (status == NO_INIT) {
195        // If we just discovered that this output has been abandoned, note that,
196        // but we can't do anything else, since buffer is invalid
197        onAbandonedLocked();
198        return;
199    } else {
200        LOG_ALWAYS_FATAL_IF(status != NO_ERROR,
201                "detaching buffer from output failed (%d)", status);
202    }
203
204    ALOGV("detached buffer %#" PRIx64 " from output %p",
205          buffer->getId(), from.get());
206
207    const sp<BufferTracker>& tracker = mBuffers.editValueFor(buffer->getId());
208
209    // Merge the release fence of the incoming buffer so that the fence we send
210    // back to the input includes all of the outputs' fences
211    tracker->mergeFence(fence);
212
213    // Check to see if this is the last outstanding reference to this buffer
214    size_t releaseCount = tracker->incrementReleaseCountLocked();
215    ALOGV("buffer %#" PRIx64 " reference count %zu (of %zu)", buffer->getId(),
216            releaseCount, mOutputs.size());
217    if (releaseCount < mOutputs.size()) {
218        return;
219    }
220
221    // If we've been abandoned, we can't return the buffer to the input, so just
222    // stop tracking it and move on
223    if (mIsAbandoned) {
224        mBuffers.removeItem(buffer->getId());
225        return;
226    }
227
228    // Attach and release the buffer back to the input
229    int consumerSlot;
230    status = mInput->attachBuffer(&consumerSlot, tracker->getBuffer());
231    LOG_ALWAYS_FATAL_IF(status != NO_ERROR,
232            "attaching buffer to input failed (%d)", status);
233
234    status = mInput->releaseBuffer(consumerSlot, /* frameNumber */ 0,
235            EGL_NO_DISPLAY, EGL_NO_SYNC_KHR, tracker->getMergedFence());
236    LOG_ALWAYS_FATAL_IF(status != NO_ERROR,
237            "releasing buffer to input failed (%d)", status);
238
239    ALOGV("released buffer %#" PRIx64 " to input", buffer->getId());
240
241    // We no longer need to track the buffer once it has been returned to the
242    // input
243    mBuffers.removeItem(buffer->getId());
244
245    // Notify any waiting onFrameAvailable calls
246    --mOutstandingBuffers;
247    mReleaseCondition.signal();
248}
249
250void StreamSplitter::onAbandonedLocked() {
251    ALOGE("one of my outputs has abandoned me");
252    if (!mIsAbandoned) {
253        mInput->consumerDisconnect();
254    }
255    mIsAbandoned = true;
256    mReleaseCondition.broadcast();
257}
258
259StreamSplitter::OutputListener::OutputListener(
260        const sp<StreamSplitter>& splitter,
261        const sp<IGraphicBufferProducer>& output)
262      : mSplitter(splitter), mOutput(output) {}
263
264StreamSplitter::OutputListener::~OutputListener() {}
265
266void StreamSplitter::OutputListener::onBufferReleased() {
267    mSplitter->onBufferReleasedByOutput(mOutput);
268}
269
270void StreamSplitter::OutputListener::binderDied(const wp<IBinder>& /* who */) {
271    Mutex::Autolock lock(mSplitter->mMutex);
272    mSplitter->onAbandonedLocked();
273}
274
275StreamSplitter::BufferTracker::BufferTracker(const sp<GraphicBuffer>& buffer)
276      : mBuffer(buffer), mMergedFence(Fence::NO_FENCE), mReleaseCount(0) {}
277
278StreamSplitter::BufferTracker::~BufferTracker() {}
279
280void StreamSplitter::BufferTracker::mergeFence(const sp<Fence>& with) {
281    mMergedFence = Fence::merge(String8("StreamSplitter"), mMergedFence, with);
282}
283
284} // namespace android
285