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 36 namespace android { 37 38 status_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 59 StreamSplitter::StreamSplitter(const sp<IGraphicBufferConsumer>& inputQueue) 60 : mIsAbandoned(false), mMutex(), mReleaseCondition(), 61 mOutstandingBuffers(0), mInput(inputQueue), mOutputs(), mBuffers() {} 62 63 StreamSplitter::~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 75 status_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 99 void StreamSplitter::setName(const String8 &name) { 100 Mutex::Autolock lock(mMutex); 101 mInput->setConsumerName(name); 102 } 103 104 void 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 189 void 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 253 void 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 262 StreamSplitter::OutputListener::OutputListener( 263 const sp<StreamSplitter>& splitter, 264 const sp<IGraphicBufferProducer>& output) 265 : mSplitter(splitter), mOutput(output) {} 266 267 StreamSplitter::OutputListener::~OutputListener() {} 268 269 void StreamSplitter::OutputListener::onBufferReleased() { 270 mSplitter->onBufferReleasedByOutput(mOutput); 271 } 272 273 void StreamSplitter::OutputListener::binderDied(const wp<IBinder>& /* who */) { 274 Mutex::Autolock lock(mSplitter->mMutex); 275 mSplitter->onAbandonedLocked(); 276 } 277 278 StreamSplitter::BufferTracker::BufferTracker(const sp<GraphicBuffer>& buffer) 279 : mBuffer(buffer), mMergedFence(Fence::NO_FENCE), mReleaseCount(0) {} 280 281 StreamSplitter::BufferTracker::~BufferTracker() {} 282 283 void StreamSplitter::BufferTracker::mergeFence(const sp<Fence>& with) { 284 mMergedFence = Fence::merge(String8("StreamSplitter"), mMergedFence, with); 285 } 286 287 } // namespace android 288
