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