CCodec: refactor pipeline logic
Bug: 123632127
Test: bug repro steps
Test: atest CtsMediaTestCases -- --module-arg CtsMediaTestCases:size:small
Test: atest CtsMediaTestCases -- --module-arg CtsMediaTestCases:include-annotation:android.media.cts.MediaHeavyPresubmitTests
Change-Id: I289f51709dbd675991cd8949cd343c5bf5c6ef5c
diff --git a/media/codec2/sfplugin/CCodecBufferChannel.cpp b/media/codec2/sfplugin/CCodecBufferChannel.cpp
index 8e6a3f8..587f83c 100644
--- a/media/codec2/sfplugin/CCodecBufferChannel.cpp
+++ b/media/codec2/sfplugin/CCodecBufferChannel.cpp
@@ -152,6 +152,11 @@
*/
virtual std::unique_ptr<InputBuffers> toArrayMode(size_t size) = 0;
+ /**
+ * Return number of buffers the client owns.
+ */
+ virtual size_t numClientBuffers() const = 0;
+
protected:
// Pool to obtain blocks for input buffers.
std::shared_ptr<C2BlockPool> mPool;
@@ -508,6 +513,14 @@
mBuffers.clear();
}
+ size_t numClientBuffers() const {
+ return std::count_if(
+ mBuffers.begin(), mBuffers.end(),
+ [](const Entry &entry) {
+ return (entry.clientBuffer != nullptr);
+ });
+ }
+
private:
friend class BuffersArrayImpl;
@@ -693,6 +706,14 @@
}
}
+ size_t numClientBuffers() const {
+ return std::count_if(
+ mBuffers.begin(), mBuffers.end(),
+ [](const Entry &entry) {
+ return entry.ownedByClient;
+ });
+ }
+
private:
std::string mImplName; ///< name for debugging
const char *mName; ///< C-string version of name
@@ -756,6 +777,10 @@
mImpl.flush();
}
+ size_t numClientBuffers() const final {
+ return mImpl.numClientBuffers();
+ }
+
private:
BuffersArrayImpl mImpl;
};
@@ -823,6 +848,10 @@
return std::move(array);
}
+ size_t numClientBuffers() const final {
+ return mImpl.numClientBuffers();
+ }
+
virtual sp<Codec2Buffer> alloc(size_t size) {
C2MemoryUsage usage = { C2MemoryUsage::CPU_READ, C2MemoryUsage::CPU_WRITE };
std::shared_ptr<C2LinearBlock> block;
@@ -967,6 +996,10 @@
return std::move(array);
}
+ size_t numClientBuffers() const final {
+ return mImpl.numClientBuffers();
+ }
+
private:
FlexBuffersImpl mImpl;
std::shared_ptr<C2AllocatorStore> mStore;
@@ -1030,6 +1063,10 @@
return std::move(array);
}
+ size_t numClientBuffers() const final {
+ return mImpl.numClientBuffers();
+ }
+
private:
FlexBuffersImpl mImpl;
std::shared_ptr<LocalBufferPool> mLocalBufferPool;
@@ -1065,6 +1102,10 @@
void getArray(Vector<sp<MediaCodecBuffer>> *array) const final {
array->clear();
}
+
+ size_t numClientBuffers() const final {
+ return 0u;
+ }
};
class OutputBuffersArray : public CCodecBufferChannel::OutputBuffers {
@@ -1422,90 +1463,6 @@
count->value = -1;
}
-// CCodecBufferChannel::PipelineCapacity
-
-CCodecBufferChannel::PipelineCapacity::PipelineCapacity()
- : input(0), component(0),
- mName("<UNKNOWN COMPONENT>") {
-}
-
-void CCodecBufferChannel::PipelineCapacity::initialize(
- int newInput,
- int newComponent,
- const char* newName,
- const char* callerTag) {
- input.store(newInput, std::memory_order_relaxed);
- component.store(newComponent, std::memory_order_relaxed);
- mName = newName;
- ALOGV("[%s] %s -- PipelineCapacity::initialize(): "
- "pipeline availability initialized ==> "
- "input = %d, component = %d",
- mName, callerTag ? callerTag : "*",
- newInput, newComponent);
-}
-
-bool CCodecBufferChannel::PipelineCapacity::allocate(const char* callerTag) {
- int prevInput = input.fetch_sub(1, std::memory_order_relaxed);
- int prevComponent = component.fetch_sub(1, std::memory_order_relaxed);
- if (prevInput > 0 && prevComponent > 0) {
- ALOGV("[%s] %s -- PipelineCapacity::allocate() returns true: "
- "pipeline availability -1 all ==> "
- "input = %d, component = %d",
- mName, callerTag ? callerTag : "*",
- prevInput - 1,
- prevComponent - 1);
- return true;
- }
- input.fetch_add(1, std::memory_order_relaxed);
- component.fetch_add(1, std::memory_order_relaxed);
- ALOGV("[%s] %s -- PipelineCapacity::allocate() returns false: "
- "pipeline availability unchanged ==> "
- "input = %d, component = %d",
- mName, callerTag ? callerTag : "*",
- prevInput,
- prevComponent);
- return false;
-}
-
-void CCodecBufferChannel::PipelineCapacity::free(const char* callerTag) {
- int prevInput = input.fetch_add(1, std::memory_order_relaxed);
- int prevComponent = component.fetch_add(1, std::memory_order_relaxed);
- ALOGV("[%s] %s -- PipelineCapacity::free(): "
- "pipeline availability +1 all ==> "
- "input = %d, component = %d",
- mName, callerTag ? callerTag : "*",
- prevInput + 1,
- prevComponent + 1);
-}
-
-int CCodecBufferChannel::PipelineCapacity::freeInputSlots(
- size_t numDiscardedInputBuffers,
- const char* callerTag) {
- int prevInput = input.fetch_add(numDiscardedInputBuffers,
- std::memory_order_relaxed);
- ALOGV("[%s] %s -- PipelineCapacity::freeInputSlots(%zu): "
- "pipeline availability +%zu input ==> "
- "input = %d, component = %d",
- mName, callerTag ? callerTag : "*",
- numDiscardedInputBuffers,
- numDiscardedInputBuffers,
- prevInput + static_cast<int>(numDiscardedInputBuffers),
- component.load(std::memory_order_relaxed));
- return prevInput + static_cast<int>(numDiscardedInputBuffers);
-}
-
-int CCodecBufferChannel::PipelineCapacity::freeComponentSlot(
- const char* callerTag) {
- int prevComponent = component.fetch_add(1, std::memory_order_relaxed);
- ALOGV("[%s] %s -- PipelineCapacity::freeComponentSlot(): "
- "pipeline availability +1 component ==> "
- "input = %d, component = %d",
- mName, callerTag ? callerTag : "*",
- input.load(std::memory_order_relaxed),
- prevComponent + 1);
- return prevComponent + 1;
-}
-
// CCodecBufferChannel::ReorderStash
CCodecBufferChannel::ReorderStash::ReorderStash() {
@@ -1595,7 +1552,6 @@
mFrameIndex(0u),
mFirstValidFrameIndex(0u),
mMetaMode(MODE_NONE),
- mAvailablePipelineCapacity(),
mInputMetEos(false) {
Mutexed<std::unique_ptr<InputBuffers>>::Locked buffers(mInputBuffers);
buffers->reset(new DummyInputBuffers(""));
@@ -1658,6 +1614,9 @@
work->input.ordinal.customOrdinal = timeUs;
work->input.buffers.clear();
+ uint64_t queuedFrameIndex = work->input.ordinal.frameIndex.peeku();
+ std::vector<std::shared_ptr<C2Buffer>> queuedBuffers;
+
if (buffer->size() > 0u) {
Mutexed<std::unique_ptr<InputBuffers>>::Locked buffers(mInputBuffers);
std::shared_ptr<C2Buffer> c2buffer;
@@ -1665,11 +1624,9 @@
return -ENOENT;
}
work->input.buffers.push_back(c2buffer);
- } else {
- mAvailablePipelineCapacity.freeInputSlots(1, "queueInputBufferInternal");
- if (eos) {
- flags |= C2FrameData::FLAG_END_OF_STREAM;
- }
+ queuedBuffers.push_back(c2buffer);
+ } else if (eos) {
+ flags |= C2FrameData::FLAG_END_OF_STREAM;
}
work->input.flags = (C2FrameData::flags_t)flags;
// TODO: fill info's
@@ -1680,10 +1637,16 @@
std::list<std::unique_ptr<C2Work>> items;
items.push_back(std::move(work));
+ mPipelineWatcher.lock()->onWorkQueued(
+ queuedFrameIndex,
+ std::move(queuedBuffers),
+ PipelineWatcher::Clock::now());
c2_status_t err = mComponent->queue(&items);
+ if (err != C2_OK) {
+ mPipelineWatcher.lock()->onWorkDone(queuedFrameIndex);
+ }
if (err == C2_OK && eos && buffer->size() > 0u) {
- mCCodecCallback->onWorkQueued(false);
work.reset(new C2Work);
work->input.ordinal.timestamp = timeUs;
work->input.ordinal.frameIndex = mFrameIndex++;
@@ -1693,13 +1656,22 @@
work->input.flags = C2FrameData::FLAG_END_OF_STREAM;
work->worklets.emplace_back(new C2Worklet);
+ queuedFrameIndex = work->input.ordinal.frameIndex.peeku();
+ queuedBuffers.clear();
+
items.clear();
items.push_back(std::move(work));
+
+ mPipelineWatcher.lock()->onWorkQueued(
+ queuedFrameIndex,
+ std::move(queuedBuffers),
+ PipelineWatcher::Clock::now());
err = mComponent->queue(&items);
+ if (err != C2_OK) {
+ mPipelineWatcher.lock()->onWorkDone(queuedFrameIndex);
+ }
}
if (err == C2_OK) {
- mCCodecCallback->onWorkQueued(eos);
-
Mutexed<std::unique_ptr<InputBuffers>>::Locked buffers(mInputBuffers);
bool released = (*buffers)->releaseBuffer(buffer, nullptr, true);
ALOGV("[%s] queueInputBuffer: buffer %sreleased", mName, released ? "" : "not ");
@@ -1846,14 +1818,16 @@
void CCodecBufferChannel::feedInputBufferIfAvailableInternal() {
while (!mInputMetEos &&
!mReorderStash.lock()->hasPending() &&
- mAvailablePipelineCapacity.allocate("feedInputBufferIfAvailable")) {
+ !mPipelineWatcher.lock()->pipelineFull()) {
sp<MediaCodecBuffer> inBuffer;
size_t index;
{
Mutexed<std::unique_ptr<InputBuffers>>::Locked buffers(mInputBuffers);
+ if ((*buffers)->numClientBuffers() >= mNumInputSlots) {
+ return;
+ }
if (!(*buffers)->requestNewBuffer(&index, &inBuffer)) {
ALOGV("[%s] no new buffer available", mName);
- mAvailablePipelineCapacity.free("feedInputBufferIfAvailable");
break;
}
}
@@ -2032,15 +2006,12 @@
{
Mutexed<std::unique_ptr<InputBuffers>>::Locked buffers(mInputBuffers);
if (*buffers && (*buffers)->releaseBuffer(buffer, nullptr, true)) {
- buffers.unlock();
released = true;
- mAvailablePipelineCapacity.freeInputSlots(1, "discardBuffer");
}
}
{
Mutexed<std::unique_ptr<OutputBuffers>>::Locked buffers(mOutputBuffers);
if (*buffers && (*buffers)->releaseBuffer(buffer, nullptr)) {
- buffers.unlock();
released = true;
}
}
@@ -2408,10 +2379,14 @@
// about buffers from the previous generation do not interfere with the
// newly initialized pipeline capacity.
- mAvailablePipelineCapacity.initialize(
- mNumInputSlots,
- mNumInputSlots + mNumOutputSlots,
- mName);
+ {
+ Mutexed<PipelineWatcher>::Locked watcher(mPipelineWatcher);
+ watcher->inputDelay(inputDelay ? inputDelay.value : 0)
+ .pipelineDelay(pipelineDelay ? pipelineDelay.value : 0)
+ .outputDelay(outputDelay ? outputDelay.value : 0)
+ .smoothnessFactor(kSmoothnessFactor);
+ watcher->flush();
+ }
mInputMetEos = false;
mSync.start();
@@ -2472,21 +2447,16 @@
buffer->meta()->setInt64("timeUs", 0);
post = false;
}
- if (mAvailablePipelineCapacity.allocate("requestInitialInputBuffers")) {
- if (post) {
- mCallback->onInputBufferAvailable(index, buffer);
- } else {
- toBeQueued.emplace_back(buffer);
- }
+ if (post) {
+ mCallback->onInputBufferAvailable(index, buffer);
} else {
- ALOGD("[%s] pipeline is full while requesting %zu-th input buffer",
- mName, i);
+ toBeQueued.emplace_back(buffer);
}
}
}
for (const sp<MediaCodecBuffer> &buffer : toBeQueued) {
if (queueInputBufferInternal(buffer) != OK) {
- mAvailablePipelineCapacity.freeComponentSlot("requestInitialInputBuffers");
+ ALOGV("[%s] Error while queueing initial buffers", mName);
}
}
return OK;
@@ -2532,28 +2502,25 @@
(*buffers)->flush(flushedWork);
}
mReorderStash.lock()->flush();
+ mPipelineWatcher.lock()->flush();
}
void CCodecBufferChannel::onWorkDone(
std::unique_ptr<C2Work> work, const sp<AMessage> &outputFormat,
- const C2StreamInitDataInfo::output *initData,
- size_t numDiscardedInputBuffers) {
+ const C2StreamInitDataInfo::output *initData) {
if (handleWork(std::move(work), outputFormat, initData)) {
- mAvailablePipelineCapacity.freeInputSlots(numDiscardedInputBuffers,
- "onWorkDone");
feedInputBufferIfAvailable();
}
}
void CCodecBufferChannel::onInputBufferDone(
- const std::shared_ptr<C2Buffer>& buffer) {
+ uint64_t frameIndex, size_t arrayIndex) {
+ std::shared_ptr<C2Buffer> buffer =
+ mPipelineWatcher.lock()->onInputBufferReleased(frameIndex, arrayIndex);
bool newInputSlotAvailable;
{
Mutexed<std::unique_ptr<InputBuffers>>::Locked buffers(mInputBuffers);
newInputSlotAvailable = (*buffers)->expireComponentBuffer(buffer);
- if (newInputSlotAvailable) {
- mAvailablePipelineCapacity.freeInputSlots(1, "onInputBufferDone");
- }
}
if (newInputSlotAvailable) {
feedInputBufferIfAvailable();
@@ -2573,7 +2540,7 @@
if (work->worklets.size() != 1u
|| !work->worklets.front()
|| !(work->worklets.front()->output.flags & C2FrameData::FLAG_INCOMPLETE)) {
- mAvailablePipelineCapacity.freeComponentSlot("handleWork");
+ mPipelineWatcher.lock()->onWorkDone(work->input.ordinal.frameIndex.peeku());
}
if (work->result == C2_NOT_FOUND) {
@@ -2832,6 +2799,10 @@
return OK;
}
+PipelineWatcher::Clock::duration CCodecBufferChannel::elapsed() {
+ return mPipelineWatcher.lock()->elapsed(PipelineWatcher::Clock::now());
+}
+
void CCodecBufferChannel::setMetaMode(MetaMode mode) {
mMetaMode = mode;
}