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 namespace android { 35 36 status_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 57 StreamSplitter::StreamSplitter(const sp<IGraphicBufferConsumer>& inputQueue) 58 : mIsAbandoned(false), mMutex(), mReleaseCondition(), 59 mOutstandingBuffers(0), mInput(inputQueue), mOutputs(), mBuffers() {} 60 61 StreamSplitter::~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 73 status_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 97 void StreamSplitter::setName(const String8 &name) { 98 Mutex::Autolock lock(mMutex); 99 mInput->setConsumerName(name); 100 } 101 102 void 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.mSlot); 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.mFence); 148 149 // Attach and queue the buffer to each of the outputs 150 Vector<sp<IGraphicBufferProducer> >::iterator output = mOutputs.begin(); 151 for (; output != mOutputs.end(); ++output) { 152 int slot; 153 status = (*output)->attachBuffer(&slot, bufferItem.mGraphicBuffer); 154 if (status == NO_INIT) { 155 // If we just discovered that this output has been abandoned, note 156 // that, increment the release count so that we still release this 157 // buffer eventually, and move on to the next output 158 onAbandonedLocked(); 159 mBuffers.editValueFor(bufferItem.mGraphicBuffer->getId())-> 160 incrementReleaseCountLocked(); 161 continue; 162 } else { 163 LOG_ALWAYS_FATAL_IF(status != NO_ERROR, 164 "attaching buffer to output failed (%d)", status); 165 } 166 167 IGraphicBufferProducer::QueueBufferOutput queueOutput; 168 status = (*output)->queueBuffer(slot, queueInput, &queueOutput); 169 if (status == NO_INIT) { 170 // If we just discovered that this output has been abandoned, note 171 // that, increment the release count so that we still release this 172 // buffer eventually, and move on to the next output 173 onAbandonedLocked(); 174 mBuffers.editValueFor(bufferItem.mGraphicBuffer->getId())-> 175 incrementReleaseCountLocked(); 176 continue; 177 } else { 178 LOG_ALWAYS_FATAL_IF(status != NO_ERROR, 179 "queueing buffer to output failed (%d)", status); 180 } 181 182 ALOGV("queued buffer %#" PRIx64 " to output %p", 183 bufferItem.mGraphicBuffer->getId(), output->get()); 184 } 185 } 186 187 void StreamSplitter::onBufferReleasedByOutput( 188 const sp<IGraphicBufferProducer>& from) { 189 ATRACE_CALL(); 190 Mutex::Autolock lock(mMutex); 191 192 sp<GraphicBuffer> buffer; 193 sp<Fence> fence; 194 status_t status = from->detachNextBuffer(&buffer, &fence); 195 if (status == NO_INIT) { 196 // If we just discovered that this output has been abandoned, note that, 197 // but we can't do anything else, since buffer is invalid 198 onAbandonedLocked(); 199 return; 200 } else { 201 LOG_ALWAYS_FATAL_IF(status != NO_ERROR, 202 "detaching buffer from output failed (%d)", status); 203 } 204 205 ALOGV("detached buffer %#" PRIx64 " from output %p", 206 buffer->getId(), from.get()); 207 208 const sp<BufferTracker>& tracker = mBuffers.editValueFor(buffer->getId()); 209 210 // Merge the release fence of the incoming buffer so that the fence we send 211 // back to the input includes all of the outputs' fences 212 tracker->mergeFence(fence); 213 214 // Check to see if this is the last outstanding reference to this buffer 215 size_t releaseCount = tracker->incrementReleaseCountLocked(); 216 ALOGV("buffer %#" PRIx64 " reference count %zu (of %zu)", buffer->getId(), 217 releaseCount, mOutputs.size()); 218 if (releaseCount < mOutputs.size()) { 219 return; 220 } 221 222 // If we've been abandoned, we can't return the buffer to the input, so just 223 // stop tracking it and move on 224 if (mIsAbandoned) { 225 mBuffers.removeItem(buffer->getId()); 226 return; 227 } 228 229 // Attach and release the buffer back to the input 230 int consumerSlot; 231 status = mInput->attachBuffer(&consumerSlot, tracker->getBuffer()); 232 LOG_ALWAYS_FATAL_IF(status != NO_ERROR, 233 "attaching buffer to input failed (%d)", status); 234 235 status = mInput->releaseBuffer(consumerSlot, /* frameNumber */ 0, 236 EGL_NO_DISPLAY, EGL_NO_SYNC_KHR, tracker->getMergedFence()); 237 LOG_ALWAYS_FATAL_IF(status != NO_ERROR, 238 "releasing buffer to input failed (%d)", status); 239 240 ALOGV("released buffer %#" PRIx64 " to input", buffer->getId()); 241 242 // We no longer need to track the buffer once it has been returned to the 243 // input 244 mBuffers.removeItem(buffer->getId()); 245 246 // Notify any waiting onFrameAvailable calls 247 --mOutstandingBuffers; 248 mReleaseCondition.signal(); 249 } 250 251 void StreamSplitter::onAbandonedLocked() { 252 ALOGE("one of my outputs has abandoned me"); 253 if (!mIsAbandoned) { 254 mInput->consumerDisconnect(); 255 } 256 mIsAbandoned = true; 257 mReleaseCondition.broadcast(); 258 } 259 260 StreamSplitter::OutputListener::OutputListener( 261 const sp<StreamSplitter>& splitter, 262 const sp<IGraphicBufferProducer>& output) 263 : mSplitter(splitter), mOutput(output) {} 264 265 StreamSplitter::OutputListener::~OutputListener() {} 266 267 void StreamSplitter::OutputListener::onBufferReleased() { 268 mSplitter->onBufferReleasedByOutput(mOutput); 269 } 270 271 void StreamSplitter::OutputListener::binderDied(const wp<IBinder>& /* who */) { 272 Mutex::Autolock lock(mSplitter->mMutex); 273 mSplitter->onAbandonedLocked(); 274 } 275 276 StreamSplitter::BufferTracker::BufferTracker(const sp<GraphicBuffer>& buffer) 277 : mBuffer(buffer), mMergedFence(Fence::NO_FENCE), mReleaseCount(0) {} 278 279 StreamSplitter::BufferTracker::~BufferTracker() {} 280 281 void StreamSplitter::BufferTracker::mergeFence(const sp<Fence>& with) { 282 mMergedFence = Fence::merge(String8("StreamSplitter"), mMergedFence, with); 283 } 284 285 } // namespace android 286
