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