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