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