Merge changes from topic "inputsurfaceconnection" into main * changes: InputSurface: implement InputSurfaceConnection InputSurface: add FrameQueueThread InputSurface: input buffer work status config update

diff --git a/media/codec2/hal/aidl/Android.bp b/media/codec2/hal/aidl/Android.bp
index c85df825..b2db705 100644
--- a/media/codec2/hal/aidl/Android.bp
+++ b/media/codec2/hal/aidl/Android.bp

@@ -162,6 +162,7 @@
         "InputBufferManager.cpp",
         "ParamTypes.cpp",
         "inputsurface/FrameDropper.cpp",
+        "inputsurface/FrameQueueThread.cpp",
         "inputsurface/InputSurface.cpp",
         "inputsurface/InputSurfaceConnection.cpp",
         "inputsurface/InputSurfaceSource.cpp",
@@ -186,6 +187,7 @@
         "liblog",
         "libnativewindow",
         "libmediandk",
+        "libsfplugin_ccodec_utils",
         "libstagefright_aidl_bufferpool2",
         "libstagefright_bufferpool@2.0.1",
         "libstagefright_foundation",

diff --git a/media/codec2/hal/aidl/include/codec2/aidl/inputsurface/FrameQueueThread.h b/media/codec2/hal/aidl/include/codec2/aidl/inputsurface/FrameQueueThread.h
new file mode 100644
index 0000000..d8af97b
--- /dev/null
+++ b/media/codec2/hal/aidl/include/codec2/aidl/inputsurface/FrameQueueThread.h

@@ -0,0 +1,86 @@
+/*
+ *
+ * Copyright (C) 2025 The Android Open Source Project
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+#pragma once
+
+#include <condition_variable>
+#include <deque>
+#include <thread>
+
+#include <aidl/android/hardware/media/c2/BnInputSink.h>
+#include <utils/Timers.h>
+
+#include <C2Config.h>
+#include <C2Work.h>
+
+namespace aidl::android::hardware::media::c2::implementation {
+
+/**
+ * This class runs a thread which receives encoder frames and queues them
+ * to an encoder component. Frames queued in a specific short duration can be
+ * batched for queueing to an encoder component.
+ */
+class FrameQueueThread {
+public:
+    FrameQueueThread(const std::shared_ptr<IInputSink> &sink);
+
+    ~FrameQueueThread();
+
+    /**
+     * Queue a frame for an encoder.
+     */
+    void queue(std::unique_ptr<C2Work> &&work, int fenceFd);
+
+    /**
+     * Set/update a dataspace for upcoming frames.
+     */
+    void setDataspace(android_dataspace dataspace);
+
+    /**
+     * Set/update thread priority for the frame queueing thread.
+     */
+    void setPriority(int priority);
+
+private:
+    bool mDone = false;
+    std::thread mThread;
+    std::weak_ptr<IInputSink> mSink;
+
+    std::mutex mLock;
+    std::condition_variable mCv;
+    struct Item {
+        Item(std::unique_ptr<C2Work> &&w, int fd) : work(std::move(w)), fenceFd(fd) {}
+
+        void updateConfig(std::deque<std::unique_ptr<C2Param>> &newConfig) {
+            configUpdate = std::move(newConfig);
+        }
+
+        std::unique_ptr<C2Work> work;
+        int fenceFd;
+        std::deque<std::unique_ptr<C2Param>> configUpdate;
+    };
+    std::deque<Item> mItems;
+    std::deque<std::unique_ptr<C2Param>> mConfigUpdate;
+    nsecs_t mLastQueuedTimestampNs = 0;
+
+private:
+    void run();
+
+    void queueItems(std::deque<Item> &items);
+};
+
+}  // namespace aidl::android::hardware::media::c2::implementation

diff --git a/media/codec2/hal/aidl/include/codec2/aidl/inputsurface/InputSurface.h b/media/codec2/hal/aidl/include/codec2/aidl/inputsurface/InputSurface.h
index 8e15778..47063b9 100644
--- a/media/codec2/hal/aidl/include/codec2/aidl/inputsurface/InputSurface.h
+++ b/media/codec2/hal/aidl/include/codec2/aidl/inputsurface/InputSurface.h

@@ -95,7 +95,7 @@
     //
     // Config for current work status w.r.t input buffers
     struct WorkStatusConfig {
-        int32_t mLastDoneIndex = -1;      // Last work done input buffer index
+        uint64_t mLastDoneIndex = UINT64_MAX;      // Last work done buffer frame index
         uint32_t mLastDoneCount = 0;      // # of work done count
         uint64_t mEmptyCount = 0;         // # of input buffers being emptied
     };

diff --git a/media/codec2/hal/aidl/include/codec2/aidl/inputsurface/InputSurfaceConnection.h b/media/codec2/hal/aidl/include/codec2/aidl/inputsurface/InputSurfaceConnection.h
index 7a57f18..61fa83f 100644
--- a/media/codec2/hal/aidl/include/codec2/aidl/inputsurface/InputSurfaceConnection.h
+++ b/media/codec2/hal/aidl/include/codec2/aidl/inputsurface/InputSurfaceConnection.h

@@ -23,12 +23,17 @@
 
 #include <C2.h>
 
+#include <list>
+#include <map>
 #include <memory>
 
 namespace aidl::android::hardware::media::c2::implementation {
 class InputSurfaceSource;
+class FrameQueueThread;
 }
 
+class C2Allocator;
+
 namespace aidl::android::hardware::media::c2::utils {
 
 struct InputSurfaceConnection : public BnInputSurfaceConnection {
@@ -57,12 +62,55 @@
     void dispatchDataSpaceChanged(
             int32_t dataSpace, int32_t aspects, int32_t pixelFormat);
 
+    void release();
+
+    // InputSurface config
+    void setAdjustTimestampGapUs(int32_t gapUs);
+
+    void onInputBufferDone(c2_cntr64_t index);
+
+    void onInputBufferEmptied();
+
 protected:
     virtual ~InputSurfaceConnection() override;
 
 private:
+    c2_status_t mInit;
+    std::atomic<bool> mReleased;
+
     std::weak_ptr<IInputSink> mSink;
-    ::android::sp<c2::implementation::InputSurfaceSource> mSource;
+    ::android::wp<c2::implementation::InputSurfaceSource> mSource;
+    std::shared_ptr<c2::implementation::FrameQueueThread> mQueueThread;
+
+    std::atomic_uint64_t mFrameIndex;
+
+    // WORKAROUND: timestamp adjustment
+
+    // if >0: this is the max timestamp gap, if <0: this is -1 times the fixed timestamp gap
+    // if 0: no timestamp adjustment is made
+    // note that C2OMXNode can be recycled between encoding sessions.
+    int32_t mAdjustTimestampGapUs;
+    bool mFirstInputFrame; // true for first input
+    c2_cntr64_t mPrevInputTimestamp; // input timestamp for previous frame
+    c2_cntr64_t mPrevCodecTimestamp; // adjusted (codec) timestamp for previous frame
+
+    // Tracks the status of buffers
+    struct BuffersTracker {
+        BuffersTracker() = default;
+
+        // For synchronization of data accesses and/or modifications.
+        std::mutex mMutex;
+        // Keeps track of buffers that are used by the component. Maps timestamp -> ID
+        std::map<uint64_t, uint32_t> mIdsInUse;
+        // Keeps track of the buffer IDs that are available after being released from the component.
+        std::list<uint32_t> mAvailableIds;
+    };
+    BuffersTracker mBuffersTracker;
+
+    c2_status_t submitBufferInternal(
+            int32_t bufferId, const AImage *buffer, int64_t timestamp, int fenceFd, bool eos);
+
+    void notifyInputBufferEmptied(int32_t bufferId);
 };
 
 }  // namespace aidl::android::hardware::media::c2::utils

diff --git a/media/codec2/hal/aidl/inputsurface/FrameQueueThread.cpp b/media/codec2/hal/aidl/inputsurface/FrameQueueThread.cpp
new file mode 100644
index 0000000..b6cb8b4
--- /dev/null
+++ b/media/codec2/hal/aidl/inputsurface/FrameQueueThread.cpp

@@ -0,0 +1,137 @@
+/*
+ *
+ * Copyright (C) 2025 The Android Open Source Project
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+//#define LOG_NDEBUG 0
+#define LOG_TAG "Codec2-InputSurface"
+
+#include <sys/types.h>
+
+#include <chrono>
+
+#include <android-base/logging.h>
+#include <codec2/aidl/BufferTypes.h>
+#include <codec2/aidl/inputsurface/FrameQueueThread.h>
+#include <media/stagefright/foundation/ColorUtils.h>
+#include <ui/Fence.h>
+#include <utils/AndroidThreads.h>
+
+#include <Codec2Mapper.h>
+
+namespace aidl::android::hardware::media::c2::implementation {
+
+FrameQueueThread::FrameQueueThread(const std::shared_ptr<IInputSink> &sink)
+        : mSink{sink} {
+    mThread = std::thread(&FrameQueueThread::run, this);
+}
+
+FrameQueueThread::~FrameQueueThread() {
+    {
+        std::unique_lock<std::mutex> l(mLock);
+        mDone = true;
+        mCv.notify_all();
+    }
+    if (mThread.joinable()) {
+        mThread.join();
+    }
+}
+
+void FrameQueueThread::queue(std::unique_ptr<C2Work> &&work, int fenceFd) {
+    {
+        std::unique_lock<std::mutex> l(mLock);
+        mItems.emplace_back(std::move(work), fenceFd);
+        if (!mConfigUpdate.empty()) {
+            mItems.back().updateConfig(mConfigUpdate);
+        }
+    }
+    mCv.notify_all();
+}
+
+void FrameQueueThread::setDataspace(android_dataspace dataspace) {
+    std::unique_lock<std::mutex> l(mLock);
+    ::android::ColorUtils::convertDataSpaceToV0(dataspace);
+    mConfigUpdate.emplace_back(new C2StreamDataSpaceInfo::input(0u, dataspace));
+    int32_t standard;
+    int32_t transfer;
+    int32_t range;
+    ::android::ColorUtils::getColorConfigFromDataSpace(dataspace, &range, &standard, &transfer);
+    std::unique_ptr<C2StreamColorAspectsInfo::input> colorAspects =
+        std::make_unique<C2StreamColorAspectsInfo::input>(0u);
+    if (::android::C2Mapper::map(standard, &colorAspects->primaries, &colorAspects->matrix)
+            && ::android::C2Mapper::map(transfer, &colorAspects->transfer)
+            && ::android::C2Mapper::map(range, &colorAspects->range)) {
+        mConfigUpdate.push_back(std::move(colorAspects));
+    }
+}
+
+void FrameQueueThread::setPriority(int priority) {
+    androidSetThreadPriority(gettid(), priority);
+}
+
+void FrameQueueThread::run() {
+    constexpr nsecs_t kIntervalNs = nsecs_t(10) * 1000 * 1000;  // 10ms
+    constexpr nsecs_t kWaitNs = kIntervalNs * 2;
+
+    std::unique_lock<std::mutex> lock(mLock);
+    while (!mDone) {
+        nsecs_t nowNs = systemTime();
+        nsecs_t diffNs = nowNs - mLastQueuedTimestampNs;
+        if (mItems.empty() || (mLastQueuedTimestampNs != 0 && diffNs < kIntervalNs)) {
+            mCv.wait_for(lock, std::chrono::nanoseconds(kIntervalNs - diffNs));
+            continue;
+        }
+        std::deque<Item> items = std::move(mItems);
+        lock.unlock();
+        queueItems(items);
+        lock.lock();
+        mLastQueuedTimestampNs = nowNs;
+        mCv.wait_for(lock, std::chrono::nanoseconds(kWaitNs));
+    }
+}
+
+void FrameQueueThread::queueItems(std::deque<Item> &items) {
+    std::shared_ptr<IInputSink> sink = mSink.lock();
+    if (!sink) {
+        ALOGE("queueItems: sink is not valid");
+        return;
+    }
+
+    std::list<std::unique_ptr<C2Work>> c2Items;
+    std::vector<int> fenceFds;
+    while (!items.empty()) {
+        c2Items.push_back(std::move(items.front().work));
+        fenceFds.push_back(items.front().fenceFd);
+        for (const std::unique_ptr<C2Param> &param: items.front().configUpdate) {
+            c2Items.back()->input.configUpdate.emplace_back(C2Param::Copy(*param));
+        }
+        items.pop_front();
+    }
+    // TODO: Pass fence if an encoder supports receiving fences
+    // along with a block.
+    for (int fenceFd : fenceFds) {
+        ::android::sp<::android::Fence> fence(new ::android::Fence(fenceFd));
+        fence->waitForever(LOG_TAG);
+    }
+
+    WorkBundle workBundle;
+    if (!utils::ToAidl(&workBundle, c2Items, nullptr)) {
+        ALOGE("queueItems: conversion from C2Work to workBundle failed");
+        return;
+    }
+    sink->queue(workBundle);
+}
+
+}  // namespace aidl::android::hardware::media::c2::implementation

diff --git a/media/codec2/hal/aidl/inputsurface/InputSurface.cpp b/media/codec2/hal/aidl/inputsurface/InputSurface.cpp
index ce694ee..eeb1f0b 100644
--- a/media/codec2/hal/aidl/inputsurface/InputSurface.cpp
+++ b/media/codec2/hal/aidl/inputsurface/InputSurface.cpp

@@ -169,8 +169,8 @@
                 .build());
 
         addParameter(
-                DefineParam(mInputDone, C2_PARAMKEY_LAYER_INDEX)
-                .withDefault(new C2StreamLayerIndexInfo::output(0u, UINT32_MAX))
+                DefineParam(mInputDone, C2_PARAMKEY_OUTPUT_COUNTER)
+                .withDefault(new C2PortConfigCounterTuning::output(UINT64_MAX))
                 .withFields({C2F(mInputDone, value).any()})
                 .withSetter(BasicSetter<decltype(mInputDone)::element_type>)
                 .build());
@@ -178,13 +178,13 @@
                 DefineParam(mInputDoneCount, C2_PARAMKEY_LAYER_INDEX)
                 .withDefault(new C2StreamLayerCountInfo::input(0u, 0))
                 .withFields({C2F(mInputDoneCount, value).any()})
-                .withSetter(InputDoneCountSetter)
+                .withSetter(BasicSetter<decltype(mInputDoneCount)::element_type>)
                 .build());
         addParameter(
                 DefineParam(mEmptyCount, C2_PARAMKEY_LAYER_COUNT)
                 .withDefault(new C2StreamLayerCountInfo::output(0u, 0))
                 .withFields({C2F(mEmptyCount, value).any()})
-                .withSetter(EmptyCountSetter)
+                .withSetter(BasicSetter<decltype(mEmptyCount)::element_type>)
                 .build());
     }
 
@@ -222,11 +222,7 @@
     }
 
     void getWorkStatusConfig(WorkStatusConfig* _Nonnull config) {
-        if (mInputDone->value == UINT32_MAX) {
-            config->mLastDoneIndex = -1;
-        } else {
-            config->mLastDoneIndex = mInputDone->value;
-        }
+        config->mLastDoneIndex = mInputDone->value;
         config->mLastDoneCount = mInputDoneCount->value;
         config->mEmptyCount = mEmptyCount->value;
     }
@@ -251,20 +247,6 @@
         return C2R::Ok();
     }
 
-    static C2R InputDoneCountSetter(bool mayBlock,
-            C2InterfaceHelper::C2P<C2StreamLayerCountInfo::input> &me) {
-        (void)mayBlock;
-        me.set().value = me.v.value + 1;
-        return C2R::Ok();
-    }
-
-    static C2R EmptyCountSetter(bool mayBlock,
-            C2InterfaceHelper::C2P<C2StreamLayerCountInfo::output> &me) {
-        (void)mayBlock;
-        me.set().value = me.v.value + 1;
-        return C2R::Ok();
-    }
-
 private:
     // buffer configuraration
     std::shared_ptr<C2StreamBlockSizeInfo::output> mBlockSize;
@@ -289,7 +271,7 @@
 
     // current work status configuration
     // TODO: remove this and move this to onWorkDone()
-    std::shared_ptr<C2StreamLayerIndexInfo::output> mInputDone;
+    std::shared_ptr<C2PortConfigCounterTuning::output> mInputDone;
     std::shared_ptr<C2StreamLayerCountInfo::input> mInputDoneCount;
     std::shared_ptr<C2StreamLayerCountInfo::output> mEmptyCount;
 };
@@ -409,6 +391,7 @@
 ::ndk::ScopedAStatus InputSurface::connect(
         const std::shared_ptr<IInputSink>& sink,
         std::shared_ptr<IInputSurfaceConnection>* connection) {
+    std::unique_lock<std::mutex> l(mLock);
     mConnection = SharedRefBase::make<InputSurfaceConnection>(sink, mSource);
     *connection = mConnection;
     return ::ndk::ScopedAStatus::ok();
@@ -459,12 +442,12 @@
     if (config.mAdjustedFpsMode != C2TimestampGapAdjustmentStruct::NONE && (
             config.mAdjustedFpsMode != mStreamConfig.mAdjustedFpsMode ||
             config.mAdjustedGapUs != mStreamConfig.mAdjustedGapUs)) {
-        // TODO: configure GapUs to connection
-        // The original codes do not update config, figure out why.
         mStreamConfig.mAdjustedFpsMode = config.mAdjustedFpsMode;
         mStreamConfig.mAdjustedGapUs = config.mAdjustedGapUs;
         fixedModeUpdate = (config.mAdjustedFpsMode == C2TimestampGapAdjustmentStruct::FIXED_GAP);
-        // TODO: update Gap to Connection.
+        if (mConnection) {
+            mConnection->setAdjustTimestampGapUs(mStreamConfig.mAdjustedGapUs);
+        }
     }
     // TRICKY: we do not unset max fps to 0 unless using fixed fps
     if ((config.mMaxFps > 0 || (fixedModeUpdate && config.mMaxFps == -1))
@@ -550,8 +533,22 @@
 }
 
 void InputSurface::updateWorkStatusConfig(WorkStatusConfig &config) {
-    (void)config;
-    // TODO
+    std::unique_lock<std::mutex> l(mLock);
+    if (!mConnection) {
+        ALOGE("work status is updated though there is no connection.");
+        return;
+    }
+    if (mWorkStatusConfig.mLastDoneIndex != config.mLastDoneIndex) {
+        mWorkStatusConfig.mLastDoneIndex = config.mLastDoneIndex;
+        mConnection->onInputBufferDone(mWorkStatusConfig.mLastDoneIndex);
+    }
+    if (mWorkStatusConfig.mLastDoneCount != config.mLastDoneCount) {
+        mWorkStatusConfig.mLastDoneCount = config.mLastDoneCount;
+    }
+    if (mWorkStatusConfig.mEmptyCount != config.mEmptyCount) {
+        mWorkStatusConfig.mEmptyCount = config.mEmptyCount;
+        mConnection->onInputBufferEmptied();
+    }
 }
 
 bool InputSurface::updateConfig(

diff --git a/media/codec2/hal/aidl/inputsurface/InputSurfaceConnection.cpp b/media/codec2/hal/aidl/inputsurface/InputSurfaceConnection.cpp
index 6a95472..1250531 100644
--- a/media/codec2/hal/aidl/inputsurface/InputSurfaceConnection.cpp
+++ b/media/codec2/hal/aidl/inputsurface/InputSurfaceConnection.cpp

@@ -16,54 +16,235 @@
 
 //#define LOG_NDEBUG 0
 #define LOG_TAG "Codec2-InputSurface"
-#include <android-base/logging.h>
 
+#include <android_media_codec.h>
+#include <android-base/logging.h>
+#include <android-base/unique_fd.h>
+
+#include <codec2/aidl/inputsurface/FrameQueueThread.h>
 #include <codec2/aidl/inputsurface/InputSurfaceConnection.h>
 #include <codec2/aidl/inputsurface/InputSurfaceSource.h>
 
+#include <C2AllocatorGralloc.h>
+#include <C2BlockInternal.h>
+
 namespace aidl::android::hardware::media::c2::utils {
 
 InputSurfaceConnection::InputSurfaceConnection(
         const std::shared_ptr<IInputSink>& sink,
         ::android::sp<c2::implementation::InputSurfaceSource> const &source)
-        : mSink{sink}, mSource{source} {
+    : mSink{sink}, mSource{source},
+      mQueueThread{std::make_shared<implementation::FrameQueueThread>(sink)}, mFrameIndex(0),
+      mAdjustTimestampGapUs(0), mFirstInputFrame(true) {
+    auto component = mSink.lock();
+    if (!component) {
+        mInit = C2_NO_INIT;
+        return;
+    }
+    mInit = C2_OK;
 }
 
 InputSurfaceConnection::~InputSurfaceConnection() {
 }
 
 c2_status_t InputSurfaceConnection::status() const {
-    // TODO;
-    return C2_OK;
+    return mInit;
 }
 
 ::ndk::ScopedAStatus InputSurfaceConnection::disconnect() {
+    auto source = mSource.promote();
+    if (!source) {
+        return ::ndk::ScopedAStatus::fromServiceSpecificError(C2_CORRUPTED);
+    }
+    (void)source->stop();
+    (void)source->release();
+
     return ::ndk::ScopedAStatus::ok();
 }
 
 ::ndk::ScopedAStatus InputSurfaceConnection::signalEndOfStream() {
+    auto source = mSource.promote();
+    if (!source) {
+        return ::ndk::ScopedAStatus::fromServiceSpecificError(C2_CORRUPTED);
+    }
+
+    c2_status_t status = source->signalEndOfInputStream();
+    if (status != C2_OK) {
+        return ::ndk::ScopedAStatus::fromServiceSpecificError(status);
+    }
     return ::ndk::ScopedAStatus::ok();
 }
 
 c2_status_t InputSurfaceConnection::submitBuffer(
         int32_t bufferId, const AImage *buffer, int64_t timestamp, int fenceFd) {
-    (void)bufferId;
-    (void)buffer;
-    (void)timestamp;
-    (void)fenceFd;
-    return C2_OK;
+    return submitBufferInternal(bufferId, buffer, timestamp, fenceFd, false);
 }
 
 c2_status_t InputSurfaceConnection::submitEos(int32_t bufferId) {
-    (void)bufferId;
-    return C2_OK;
+    return submitBufferInternal(bufferId, nullptr, 0, -1, true);
 }
 
 void InputSurfaceConnection::dispatchDataSpaceChanged(
             int32_t dataSpace, int32_t aspects, int32_t pixelFormat) {
-    (void)dataSpace;
     (void)aspects;
     (void)pixelFormat;
+    android_dataspace d = (android_dataspace)dataSpace;
+    mQueueThread->setDataspace(d);
+}
+
+void InputSurfaceConnection::setAdjustTimestampGapUs(int32_t gapUs) {
+    mAdjustTimestampGapUs = gapUs;
+}
+
+
+void InputSurfaceConnection::onInputBufferDone(c2_cntr64_t index) {
+    if (::android::media::codec::provider_->input_surface_throttle()) {
+        std::unique_lock<std::mutex> l(mBuffersTracker.mMutex);
+        auto it = mBuffersTracker.mIdsInUse.find(index.peeku());
+        if (it == mBuffersTracker.mIdsInUse.end()) {
+            ALOGV("Untracked input index %llu (maybe already removed)", index.peekull());
+            return;
+        }
+        int32_t bufferId = it->second;
+        (void)mBuffersTracker.mIdsInUse.erase(it);
+        mBuffersTracker.mAvailableIds.push_back(bufferId);
+    } else {
+        {
+            auto source = mSource.promote();
+            if (!source) {
+                return;
+            }
+        }
+        int32_t bufferId = 0;
+        {
+            std::unique_lock<std::mutex> l(mBuffersTracker.mMutex);
+            auto it = mBuffersTracker.mIdsInUse.find(index.peeku());
+            if (it == mBuffersTracker.mIdsInUse.end()) {
+                ALOGV("Untracked input index %llu (maybe already removed)", index.peekull());
+                return;
+            }
+            bufferId = it->second;
+            (void)mBuffersTracker.mIdsInUse.erase(it);
+        }
+        notifyInputBufferEmptied(bufferId);
+    }
+}
+
+void InputSurfaceConnection::onInputBufferEmptied() {
+    if (!::android::media::codec::provider_->input_surface_throttle()) {
+        ALOGE("onInputBufferEmptied should not be called "
+              "when input_surface_throttle is false");
+        return;
+    }
+    {
+        auto source = mSource.promote();
+        if (!source) {
+            return;
+        }
+    }
+    int32_t bufferId = 0;
+    {
+        std::unique_lock<std::mutex> l(mBuffersTracker.mMutex);
+        if (mBuffersTracker.mAvailableIds.empty()) {
+            ALOGV("The codec is ready to take more input buffers "
+                    "but no input buffers are ready yet.");
+            return;
+        }
+        bufferId = mBuffersTracker.mAvailableIds.front();
+        mBuffersTracker.mAvailableIds.pop_front();
+    }
+}
+
+c2_status_t InputSurfaceConnection::submitBufferInternal(
+        int32_t bufferId, const AImage *buffer, int64_t timestamp, int fenceFd, bool eos) {
+    // close fenceFd on returning an error.
+    ::android::base::unique_fd ufd(fenceFd);
+    std::shared_ptr<IInputSink> sink = mSink.lock();
+    if (!sink) {
+        ALOGE("inputsurface does not have valid sink");
+        return C2_BAD_STATE;
+    }
+
+    uint32_t c2Flags = (eos == true) ? C2FrameData::FLAG_END_OF_STREAM : 0;
+    AHardwareBuffer *hwBuffer = nullptr;
+
+    if (buffer) {
+        if (AImage_getHardwareBuffer(buffer, &hwBuffer) != AMEDIA_OK) {
+            ALOGE("cannot get AHardwareBuffer form AImage");
+            return C2_CORRUPTED;
+        }
+    } else if (!eos) {
+        ALOGE("buffer should be submitted, but was nullptr");
+        return C2_BAD_VALUE;
+    }
+
+    std::shared_ptr<C2GraphicBlock> block;
+    if (hwBuffer) {
+        block = _C2BlockFactory::CreateGraphicBlock(hwBuffer);
+    }
+
+    std::unique_ptr<C2Work> work(new C2Work);
+    work->input.flags = (C2FrameData::flags_t)c2Flags;
+    work->input.ordinal.timestamp = timestamp;
+    {
+        work->input.ordinal.customOrdinal = timestamp; // save input timestamp
+        if (mFirstInputFrame) {
+            // grab timestamps on first frame
+            mPrevInputTimestamp = timestamp;
+            mPrevCodecTimestamp = timestamp;
+            mFirstInputFrame = false;
+        } else if (mAdjustTimestampGapUs > 0) {
+            work->input.ordinal.timestamp =
+                mPrevCodecTimestamp
+                        + c2_min((timestamp - mPrevInputTimestamp).peek(), mAdjustTimestampGapUs);
+        } else if (mAdjustTimestampGapUs < 0) {
+            work->input.ordinal.timestamp = mPrevCodecTimestamp - mAdjustTimestampGapUs;
+        }
+        mPrevInputTimestamp = work->input.ordinal.customOrdinal;
+        mPrevCodecTimestamp = work->input.ordinal.timestamp;
+        ALOGV("adjusting %lld to %lld (gap=%lld)",
+              work->input.ordinal.customOrdinal.peekll(),
+              work->input.ordinal.timestamp.peekll(),
+              (long long)mAdjustTimestampGapUs);
+    }
+
+    work->input.ordinal.frameIndex = mFrameIndex++;
+    work->input.buffers.clear();
+    if (block) {
+        std::shared_ptr<C2Buffer> c2Buffer(
+                C2Buffer::CreateGraphicBuffer(block->share(
+                        C2Rect(block->width(), block->height()), ::C2Fence())));
+        work->input.buffers.push_back(c2Buffer);
+        std::shared_ptr<C2StreamHdrStaticInfo::input> staticInfo;
+        std::shared_ptr<C2StreamHdrDynamicMetadataInfo::input> dynamicInfo;
+        ::android::GetHdrMetadataFromGralloc4Handle(
+                block->handle(),
+                &staticInfo,
+                &dynamicInfo);
+        if (staticInfo && *staticInfo) {
+            c2Buffer->setInfo(staticInfo);
+        }
+        if (dynamicInfo && *dynamicInfo) {
+            c2Buffer->setInfo(dynamicInfo);
+        }
+    }
+    work->worklets.clear();
+    work->worklets.emplace_back(new C2Worklet);
+    {
+        std::unique_lock<std::mutex> l(mBuffersTracker.mMutex);
+        mBuffersTracker.mIdsInUse.emplace(work->input.ordinal.frameIndex.peeku(), bufferId);
+    }
+    mQueueThread->queue(std::move(work), ufd.release());
+
+    return C2_OK;
+}
+
+void InputSurfaceConnection::notifyInputBufferEmptied(int32_t bufferId) {
+    auto source = mSource.promote();
+    if (!source) {
+        return;
+    }
+    source->onInputBufferEmptied(bufferId, -1);
 }
 
 }  // namespace aidl::android::hardware::media::c2::utils