[移动开发]Android CCodec (三) Input Buffer流程分析

1、基本流程分析

上篇我们分析了Codec2的创建过程，主要从MediaCodec出发，到CodecClient的的创建流程(Android Codec2初始化流程),今天我们来分析Android App是怎么将数据从APP传送到Codec2的解码器的,我们以VpX的软件解码源码为例，解析一下软件解码过程中Input Buffer数据流通过程。
首先我们都知道CCdoec是继承CodecBase的，数据从Android API送到MediaCodc，之后调用MediaCodec的接口。
调用如下接口函数:

status_t MediaCodec::queueInputBuffer(
        size_t index,
        size_t offset,
        size_t size,
        int64_t presentationTimeUs,
        uint32_t flags,
        AString *errorDetailMsg) {
    if (errorDetailMsg != NULL) {
        errorDetailMsg->clear();
    }
    sp<AMessage> msg = new AMessage(kWhatQueueInputBuffer, this);
    msg->setSize("index", index);
    msg->setSize("offset", offset);
    msg->setSize("size", size);
    msg->setInt64("timeUs", presentationTimeUs);
    msg->setInt32("flags", flags);
    msg->setPointer("errorDetailMsg", errorDetailMsg);
    sp<AMessage> response;
    return PostAndAwaitResponse(msg, &response);
}

这是发送消息到MediaCodec::onMessageReceived中进行处理。在MediaCodec::onMessageReceived中调用的是handleLeftover和onQueueInputBuffer
不过在handleLeftover则继续调用onQueueInputBuffer处理方法。

 case kWhatQueueInputBuffer:
        {
            sp<AReplyToken> replyID;
            CHECK(msg->senderAwaitsResponse(&replyID));

            if (!isExecuting()) {
                PostReplyWithError(replyID, INVALID_OPERATION);
                break;
            } else if (mFlags & kFlagStickyError) {
                PostReplyWithError(replyID, getStickyError());
                break;
            }

            status_t err = UNKNOWN_ERROR;
            if (!mLeftover.empty()) {
                mLeftover.push_back(msg);
                size_t index;
                msg->findSize("index", &index);
                err = handleLeftover(index);
            } else {
                err = onQueueInputBuffer(msg);
            }

            PostReplyWithError(replyID, err);
            break;
        }

在MediaCodec::onQueueInputBuffer中，就调用了

status_t MediaCodec::onQueueInputBuffer(const sp<AMessage> &msg) {
    //此处省略不重要的代码
    if (c2Buffer || memory) {
        sp<AMessage> tunings;
        CHECK(msg->findMessage("tunings", &tunings));
        onSetParameters(tunings);

        status_t err = OK;
        if (c2Buffer) { //拿到C2Buffer数据
            err = mBufferChannel->attachBuffer(c2Buffer, buffer);
        }
    }

  if (hasCryptoOrDescrambler() && !c2Buffer && !memory) {
      //Sucure数据送入Buffer数据
    err = mBufferChannel->queueSecureInputBuffer(
            buffer,
            (mFlags & kFlagIsSecure),
            key,
            iv,
            mode,
            pattern,
            subSamples,
            numSubSamples,
            errorDetailMsg);
  } else {
    //送入非Secure数据
    err = mBufferChannel->queueInputBuffer(buffer);
  }
  return err;
}

众所周知，mBufferChannel是std::shared_ptr mBufferChannel;的指针，之前在初始化的时候，也就是 mBufferChannel = mCodec->getBufferChannel(); 设置的。当前平台支持Codec2，则getBufferChannel返回的是CCodecBufferChannel的对象指针。所以，这时候就进入CCodec2的buffer轮转中了，现在继续深入到CCodecBufferChannel中看看，里面是个什么奇形怪状、姹紫嫣红的景象。
CCodecBufferChannel继续追溯，我们可以看到调用了 queueInputBufferInternal 。

status_t CCodecBufferChannel::queueInputBuffer(const sp<MediaCodecBuffer> &buffer) {
    QueueGuard guard(mSync);
    if (!guard.isRunning()) {
        ALOGD("[%s] No more buffers should be queued at current state.", mName);
        return -ENOSYS;
    }
    return queueInputBufferInternal(buffer);
}

在 queueInputBufferInternal 中，进行数据的解析和封装。将数据封装到C2Work的结构中，方便在CCodec架构里面的传递。具体代码如下：

status_t CCodecBufferChannel::queueInputBufferInternal(
        sp<MediaCodecBuffer> buffer,
        std::shared_ptr<C2LinearBlock> encryptedBlock,
        size_t blockSize) {
    int64_t timeUs;
    int32_t flags = 0;
    int32_t tmp = 0;
    bool eos = false;
    bool tunnelFirstFrame = false;
    if (buffer->meta()->findInt32("eos", &tmp) && tmp) {
        eos = true;
        mInputMetEos = true;
        ALOGV("[%s] input EOS", mName);
    }
    if (buffer->meta()->findInt32("csd", &tmp) && tmp) {
        flags |= C2FrameData::FLAG_CODEC_CONFIG;
    }
    if (buffer->meta()->findInt32("tunnel-first-frame", &tmp) && tmp) {
        tunnelFirstFrame = true;
    }
    ALOGV("[%s] queueInputBuffer: buffer->size() = %zu", mName, buffer->size());
    std::list<std::unique_ptr<C2Work>> items;
    std::unique_ptr<C2Work> work(new C2Work);
    work->input.ordinal.timestamp = timeUs; //时间戳
    work->input.ordinal.frameIndex = mFrameIndex++; //帧数? 待定
    work->input.ordinal.customOrdinal = timeUs;
    work->input.buffers.clear();

    sp<Codec2Buffer> copy;
    bool usesFrameReassembler = false;
    if (buffer->size() > 0u) {
        Mutexed<Input>::Locked input(mInput);
        std::shared_ptr<C2Buffer> c2buffer;
        if (!input->buffers->releaseBuffer(buffer, &c2buffer, false)) {
            return -ENOENT;
        }
        // TODO: we want to delay copying buffers.
        if (input->extraBuffers.numComponentBuffers() < input->numExtraSlots) {
            copy = input->buffers->cloneAndReleaseBuffer(buffer);
            if (copy != nullptr) {
                (void)input->extraBuffers.assignSlot(copy);
                if (!input->extraBuffers.releaseSlot(copy, &c2buffer, false)) {
                    return UNKNOWN_ERROR;
                }
                bool released = input->buffers->releaseBuffer(buffer, nullptr, true);
                ALOGV("[%s] queueInputBuffer: buffer copied; %sreleased",
                      mName, released ? "" : "not ");
                buffer.clear();
            } else {
                ALOGW("[%s] queueInputBuffer: failed to copy a buffer; this may cause input "
                      "buffer starvation on component.", mName);
            }
        }
        if (input->frameReassembler) {
            usesFrameReassembler = true;
            input->frameReassembler.process(buffer, &items);
        } else {
            int32_t cvo = 0;
            if (buffer->meta()->findInt32("cvo", &cvo)) {
                int32_t rotation = cvo % 360;
                // change rotation to counter-clock wise.
                rotation = ((rotation <= 0) ? 0 : 360) - rotation;

                Mutexed<OutputSurface>::Locked output(mOutputSurface);
                uint64_t frameIndex = work->input.ordinal.frameIndex.peeku();
                output->rotation[frameIndex] = rotation;
            }
            work->input.buffers.push_back(c2buffer);
            if (encryptedBlock) {
                work->input.infoBuffers.emplace_back(C2InfoBuffer::CreateLinearBuffer(
                        kParamIndexEncryptedBuffer,
                        encryptedBlock->share(0, blockSize, C2Fence())));
            }
        }
    } else if (eos) {
        flags |= C2FrameData::FLAG_END_OF_STREAM;
    }
    if (usesFrameReassembler) {
        if (!items.empty()) {
            items.front()->input.configUpdate = std::move(mParamsToBeSet);
            mFrameIndex = (items.back()->input.ordinal.frameIndex + 1).peek();
        }
    } else {
        work->input.flags = (C2FrameData::flags_t)flags;
        // TODO: fill info's

        work->input.configUpdate = std::move(mParamsToBeSet);
        if (tunnelFirstFrame) {
            C2StreamTunnelHoldRender::input tunnelHoldRender{
                0u /* stream */,
                C2_TRUE /* value */
            };
            work->input.configUpdate.push_back(C2Param::Copy(tunnelHoldRender));
        }
        work->worklets.clear();
        work->worklets.emplace_back(new C2Worklet);

        items.push_back(std::move(work));

        eos = eos && buffer->size() > 0u;
    }
    if (eos) {
        work.reset(new C2Work);
        work->input.ordinal.timestamp = timeUs;
        work->input.ordinal.frameIndex = mFrameIndex++;
        // WORKAROUND: keep client timestamp in customOrdinal
        work->input.ordinal.customOrdinal = timeUs;
        work->input.buffers.clear();
        work->input.flags = C2FrameData::FLAG_END_OF_STREAM;
        work->worklets.emplace_back(new C2Worklet);
        items.push_back(std::move(work));
    }
    c2_status_t err = C2_OK;
    if (!items.empty()) {
        {
            Mutexed<PipelineWatcher>::Locked watcher(mPipelineWatcher);
            PipelineWatcher::Clock::time_point now = PipelineWatcher::Clock::now();
            for (const std::unique_ptr<C2Work> &work : items) {
                watcher->onWorkQueued(
                        work->input.ordinal.frameIndex.peeku(),
                        std::vector(work->input.buffers),
                        now);
            }
        }
        err = mComponent->queue(&items);
    }
    if (err != C2_OK) {
        Mutexed<PipelineWatcher>::Locked watcher(mPipelineWatcher);
        for (const std::unique_ptr<C2Work> &work : items) {
            watcher->onWorkDone(work->input.ordinal.frameIndex.peeku());
        }
    } else {
        Mutexed<Input>::Locked input(mInput);
        bool released = false;
        if (buffer) {
            released = input->buffers->releaseBuffer(buffer, nullptr, true);
        } else if (copy) {
            released = input->extraBuffers.releaseSlot(copy, nullptr, true);
        }
        ALOGV("[%s] queueInputBuffer: buffer%s %sreleased",
              mName, (buffer == nullptr) ? "(copy)" : "", released ? "" : "not ");
    }
    feedInputBufferIfAvailableInternal();
    return err;
}

在上面的queueInputBufferInternal注意到有一个mComponent->queue(&items);的调用。而 std::shared_ptrCodec2Client::Component mComponent ; 指向的是组件。
其是在如下Code中设置的。

void CCodecBufferChannel::setComponent(
        const std::shared_ptr<Codec2Client::Component> &component) {
    mComponent = component;
    mComponentName = component->getName() + StringPrintf("#%d", int(uintptr_t(component.get()) % 997));
    mName = mComponentName.c_str();
}

Codec2Client::Component是组件对象。其构造函数如下：
Codec2Client::Component::Component(const sp<Base1_2>& base)
      : Configurable{
            [base]() -> sp<IConfigurable> {
                Return<sp<IComponentInterface>> transResult1 =
                        base->getInterface();
                if (!transResult1.isOk()) {
                    return nullptr;
                }
                Return<sp<IConfigurable>> transResult2 =
                        static_cast<sp<IComponentInterface>>(transResult1)->
                        getConfigurable();
                return transResult2.isOk() ?
                        static_cast<sp<IConfigurable>>(transResult2) :
                        nullptr;
            }()
        },
        mBase1_0{base},
        mBase1_1{base},
        mBase1_2{base},
        mBufferPoolSender{std::make_unique<BufferPoolSender>()},
        mOutputBufferQueue{std::make_unique<OutputBufferQueue>()} {
}

struct IComponent : public ::android::hardware::media::c2::V1_0::IComponent;

因为Component中有定义一个 std::shared_ptr mComponent;的成员变量。所以在调用Component::queue时调用的其实是C2Component里面的接口函数，

// Methods from ::android::hardware::media::c2::V1_1::IComponent
Return<Status> Component::queue(const WorkBundle& workBundle) {
    std::list<std::unique_ptr<C2Work>> c2works;

    if (!objcpy(&c2works, workBundle)) {
        return Status::CORRUPTED;
    }

    // Register input buffers.
    for (const std::unique_ptr<C2Work>& work : c2works) {
        if (work) {
            InputBufferManager::
                    registerFrameData(mListener, work->input);
        }
    }

    return static_cast<Status>(mComponent->queue_nb(&c2works));
}

Return<void> Component::flush(flush_cb _hidl_cb) {
    std::list<std::unique_ptr<C2Work>> c2flushedWorks;
    c2_status_t c2res = mComponent->flush_sm(
            C2Component::FLUSH_COMPONENT,
            &c2flushedWorks);

    // Unregister input buffers.
    for (const std::unique_ptr<C2Work>& work : c2flushedWorks) {
        if (work) {
            if (work->worklets.empty()
                    || !work->worklets.back()
                    || (work->worklets.back()->output.flags &
                        C2FrameData::FLAG_INCOMPLETE) == 0) {
                InputBufferManager::
                        unregisterFrameData(mListener, work->input);
            }
        }
    }

    WorkBundle flushedWorkBundle;
    Status res = static_cast<Status>(c2res);
    beginTransferBufferQueueBlocks(c2flushedWorks, true);
    if (c2res == C2_OK) {
        if (!objcpy(&flushedWorkBundle, c2flushedWorks, &mBufferPoolSender)) {
            res = Status::CORRUPTED;
        }
    }
    _hidl_cb(res, flushedWorkBundle);
    endTransferBufferQueueBlocks(c2flushedWorks, true, true);
    return Void();
}

Return<Status> Component::drain(bool withEos) {
    return static_cast<Status>(mComponent->drain_nb(withEos ?
            C2Component::DRAIN_COMPONENT_WITH_EOS :
            C2Component::DRAIN_COMPONENT_NO_EOS));
}

2、软件解码Vpx组件分析

下面针对软件解码封装接口进行分析，首先在 frameworks/av/media/codec2/components 目录下创建各个格式解码的基础类SimpleC2Component和SimpleC2Interface，接着各个格式具体的解码实现都是继承他们两个。
Base里面是有关SimpleC2Component和SimpleC2Interface的定义。如下所示：

class SimpleC2Component
        : public C2Component, public std::enable_shared_from_this<SimpleC2Component> {
public:
  explicit SimpleC2Component(
          const std::shared_ptr<C2ComponentInterface> &intf);
  virtual ~SimpleC2Component();
  // C2Component
  // From C2Component
  virtual c2_status_t setListener_vb(
          const std::shared_ptr<Listener> &listener, c2_blocking_t mayBlock) override;
  virtual c2_status_t queue_nb(std::list<std::unique_ptr<C2Work>>* const items) override;
  virtual c2_status_t announce_nb(const std::vector<C2WorkOutline> &items) override;
  virtual c2_status_t flush_sm(
          flush_mode_t mode, std::list<std::unique_ptr<C2Work>>* const flushedWork) override;
  virtual c2_status_t drain_nb(drain_mode_t mode) override;
  virtual c2_status_t start() override;
  virtual c2_status_t stop() override;
  virtual c2_status_t reset() override;
  virtual c2_status_t release() override;
  virtual std::shared_ptr<C2ComponentInterface> intf() override;
  bool processQueue();
protected:
    virtual c2_status_t onInit() = 0;
    virtual c2_status_t onStop() = 0;
    virtual void onReset() = 0;
    virtual void onRelease() = 0;
    virtual c2_status_t onFlush_sm() = 0;
    virtual void process(
            const std::unique_ptr<C2Work> &work,
            const std::shared_ptr<C2BlockPool> &pool) = 0;
    virtual c2_status_t drain(
            uint32_t drainMode,
            const std::shared_ptr<C2BlockPool> &pool) = 0;
    void finish(uint64_t frameIndex, std::function<void(const std::unique_ptr<C2Work> &)> fillWork);
    void cloneAndSend(
            uint64_t frameIndex,
            const std::unique_ptr<C2Work> &currentWork,
            std::function<void(const std::unique_ptr<C2Work> &)> fillWork);
    std::shared_ptr<C2Buffer> createLinearBuffer(
            const std::shared_ptr<C2LinearBlock> &block, size_t offset, size_t size);
    std::shared_ptr<C2Buffer> createGraphicBuffer(
            const std::shared_ptr<C2GraphicBlock> &block,
            const C2Rect &crop);
    static constexpr uint32_t NO_DRAIN = ~0u;
    C2ReadView mDummyReadView;

private:
    const std::shared_ptr<C2ComponentInterface> mIntf;
    class WorkHandler : public AHandler {
    public:
        enum {
            kWhatProcess,
            kWhatInit,
            kWhatStart,
            kWhatStop,
            kWhatReset,
            kWhatRelease,
        };
        WorkHandler();
        ~WorkHandler() override = default;
        void setComponent(const std::shared_ptr<SimpleC2Component> &thiz);
    protected:
        void onMessageReceived(const sp<AMessage> &msg) override;
    private:
        std::weak_ptr<SimpleC2Component> mThiz;
        bool mRunning;
    };
    enum {
        UNINITIALIZED,
        STOPPED,
        RUNNING,
    };

    struct ExecState {
        ExecState() : mState(UNINITIALIZED) {}
        int mState;
        std::shared_ptr<C2Component::Listener> mListener;
    };

    Mutexed<ExecState> mExecState;
    sp<ALooper> mLooper;
    sp<WorkHandler> mHandler;
    class WorkQueue {
    public:
        typedef std::unordered_map<uint64_t, std::unique_ptr<C2Work>> PendingWork;
        inline WorkQueue() : mFlush(false), mGeneration(0ul) {}
        inline uint64_t generation() const { return mGeneration; }
        inline void incGeneration() { ++mGeneration; mFlush = true; }
        std::unique_ptr<C2Work> pop_front();
        void push_back(std::unique_ptr<C2Work> work);
        bool empty() const;
        uint32_t drainMode() const;
        void markDrain(uint32_t drainMode);
        inline bool popPendingFlush(){}
        void clear();
        PendingWork &pending() { return mPendingWork; }
    private:
        struct Entry {
            std::unique_ptr<C2Work> work;
            uint32_t drainMode;
        };

        bool mFlush;
        uint64_t mGeneration;
        std::list<Entry> mQueue;
        PendingWork mPendingWork;
    };
    Mutexed<WorkQueue> mWorkQueue;
    class BlockingBlockPool;
    std::shared_ptr<BlockingBlockPool> mOutputBlockPool;

    SimpleC2Component() = delete;
};

在SimpleC2Component主要关注的是其内部定义了WorkHandler和WorkQueue结构，WorkHandler是模仿ALoop-AHandler-AMessage的消息队列模型的， WorkHandler也是直接继承了WorkHandler。
WorkQueue是下面子类解码和上层交互数据的封装。

class SimpleC2Interface : public C2ComponentInterface {
public:
    SimpleC2Interface(const char *name, c2_node_id_t id, const std::shared_ptr<T> &impl)
        : mName(name),
          mId(id),
          mImpl(impl) {
    }

    ~SimpleC2Interface() override = default;

    // From C2ComponentInterface
    C2String getName() const override { return mName; }
    c2_node_id_t getId() const override { return mId; }
    c2_status_t query_vb(
            const std::vector<C2Param*> &stackParams,
            const std::vector<C2Param::Index> &heapParamIndices,
            c2_blocking_t mayBlock,
            std::vector<std::unique_ptr<C2Param>>* const heapParams) const override {
        return mImpl->query(stackParams, heapParamIndices, mayBlock, heapParams);
    }
    c2_status_t config_vb(
            const std::vector<C2Param*> &params,
            c2_blocking_t mayBlock,
            std::vector<std::unique_ptr<C2SettingResult>>* const failures) override {
        return mImpl->config(params, mayBlock, failures);
    }
    c2_status_t createTunnel_sm(c2_node_id_t) override { return C2_OMITTED; }
    c2_status_t releaseTunnel_sm(c2_node_id_t) override { return C2_OMITTED; }
    c2_status_t querySupportedParams_nb(
            std::vector<std::shared_ptr<C2ParamDescriptor>> * const params) const override {
        return mImpl->querySupportedParams(params);
    }
    c2_status_t querySupportedValues_vb(
            std::vector<C2FieldSupportedValuesQuery> &fields,
            c2_blocking_t mayBlock) const override {
        return mImpl->querySupportedValues(fields, mayBlock);
    }

private:
    C2String mName;
    const c2_node_id_t mId;
    const std::shared_ptr<T> mImpl;
};

我们看下google Vpx的软件解码实现。如下：

struct C2SoftVpxDec : public SimpleC2Component {
    class IntfImpl;

    C2SoftVpxDec(const char* name, c2_node_id_t id,
              const std::shared_ptr<IntfImpl>& intfImpl);
    virtual ~C2SoftVpxDec();

    // From SimpleC2Component
    c2_status_t onInit() override;
    c2_status_t onStop() override;
    void onReset() override;
    void onRelease() override;
    c2_status_t onFlush_sm() override;
    void process(
            const std::unique_ptr<C2Work> &work,
            const std::shared_ptr<C2BlockPool> &pool) override;
    c2_status_t drain(
            uint32_t drainMode,
            const std::shared_ptr<C2BlockPool> &pool) override;
 private:
    enum {
        MODE_VP8,
        MODE_VP9,
    } mMode;

    struct ConversionQueue;

    class ConverterThread : public Thread {
    public:
        explicit ConverterThread(
                const std::shared_ptr<Mutexed<ConversionQueue>> &queue);
        ~ConverterThread() override = default;
        bool threadLoop() override;

    private:
        std::shared_ptr<Mutexed<ConversionQueue>> mQueue;
    };

    std::shared_ptr<IntfImpl> mIntf;
    vpx_codec_ctx_t *mCodecCtx;
    bool mFrameParallelMode;  // Frame parallel is only supported by VP9 decoder.

    uint32_t mWidth;
    uint32_t mHeight;
    bool mSignalledOutputEos;
    bool mSignalledError;

    int mCoreCount;
    struct ConversionQueue {
        std::list<std::function<void()>> entries;
        Condition cond;
        size_t numPending{0u};
    };
    std::shared_ptr<Mutexed<ConversionQueue>> mQueue;
    std::vector<sp<ConverterThread>> mConverterThreads;

    status_t initDecoder();
    status_t destroyDecoder();
    void finishWork(uint64_t index, const std::unique_ptr<C2Work> &work,
                    const std::shared_ptr<C2GraphicBlock> &block);
    status_t outputBuffer(
            const std::shared_ptr<C2BlockPool> &pool,
            const std::unique_ptr<C2Work> &work);
    c2_status_t drainInternal(
            uint32_t drainMode,
            const std::shared_ptr<C2BlockPool> &pool,
            const std::unique_ptr<C2Work> &work);

    C2_DO_NOT_COPY(C2SoftVpxDec);
};

下面我们针对WorkQueue的进行分析，
具体代码如下:

class WorkQueue {
    public:
        typedef std::unordered_map<uint64_t, std::unique_ptr<C2Work>> PendingWork;

        inline WorkQueue() : mFlush(false), mGeneration(0ul) {}

        inline uint64_t generation() const { return mGeneration; }
        inline void incGeneration() { ++mGeneration; mFlush = true; }

        std::unique_ptr<C2Work> pop_front();
        void push_back(std::unique_ptr<C2Work> work);
        bool empty() const;
        uint32_t drainMode() const;
        void markDrain(uint32_t drainMode);
        inline bool popPendingFlush() {
            bool flush = mFlush;
            mFlush = false;
            return flush;
        }
        void clear();
        PendingWork &pending() { return mPendingWork; }

    private:
        struct Entry {
            std::unique_ptr<C2Work> work;
            uint32_t drainMode;
        };

        bool mFlush;
        uint64_t mGeneration;
        std::list<Entry> mQueue; //封装解码数据的链表。
        PendingWork mPendingWork;
};

在queue_nb中，就不停的像mWorkQueue中传送数据的指针。之后发一个kWhatProcess的消息。

c2_status_t SimpleC2Component::queue_nb(std::list<std::unique_ptr<C2Work>> * const items) {
    {
        Mutexed<ExecState>::Locked state(mExecState);
        if (state->mState != RUNNING) {
            return C2_BAD_STATE;
        }
    }
    bool queueWasEmpty = false;
    {
        Mutexed<WorkQueue>::Locked queue(mWorkQueue);
        queueWasEmpty = queue->empty();
        while (!items->empty()) {
            queue->push_back(std::move(items->front()));
            items->pop_front();
        }
    }
    if (queueWasEmpty) {
        (new AMessage(WorkHandler::kWhatProcess, mHandler))->post();
    }
    return C2_OK;
}

void SimpleC2Component::WorkQueue::push_back(std::unique_ptr<C2Work> work) {
    mQueue.push_back({ std::move(work), NO_DRAIN });
}

这样就接受了上面queue下来的数据了。之后发送kWhatProcess消息到SimpleC2Component中进行处理在接受到kWhatProcess的消息后，开始调用processQueue这个函数处理数据，我们注意到processQueue有一个返回值，这个是表示队列中是否还有数据。如果有则继续发送kWhatProcess消息，继续处理数据。

void SimpleC2Component::WorkHandler::onMessageReceived(const sp<AMessage> &msg) {
    switch (msg->what()) {
        case kWhatProcess: {
            if (mRunning) {
                if (thiz->processQueue()) {
                    (new AMessage(kWhatProcess, this))->post();
                }
            } else {
                ALOGV("Ignore process message as we're not running");
            }
            break;
        }
    }
}

在bool SimpleC2Component::processQueue()调用子类的process处理送下来的需要解码的数据，其中process的参数是C2Work和对应的C2BlockPool。
我们可以追踪下Vp9的处理，在 process中work->input.buffers[0]->data().linearBlocks().front().map().get() 是拿到具体的解码数据的封装，rView.data() 是具体的解码数据，而 rView.capacity() 是解码数据的大小。vpx_codec_decode是google集成的VpX的解码库的API，感兴趣的可以去追一下。到此，从上层数据送数据到软件解码的流程就分析完了。下一篇分析当解码完之后数据怎么回到上层显示。

void C2SoftVpxDec::process(
        const std::unique_ptr<C2Work> &work,
        const std::shared_ptr<C2BlockPool> &pool) {
    // Initialize output work
    work->result = C2_OK;
    work->workletsProcessed = 0u;
    work->worklets.front()->output.configUpdate.clear();
    work->worklets.front()->output.flags = work->input.flags;

    if (mSignalledError || mSignalledOutputEos) {
        work->result = C2_BAD_VALUE;
        return;
    }

    size_t inOffset = 0u;
    size_t inSize = 0u;
    C2ReadView rView = mDummyReadView;
    if (!work->input.buffers.empty()) {
        rView = work->input.buffers[0]->data().linearBlocks().front().map().get();
        inSize = rView.capacity();
        if (inSize && rView.error()) {
            ALOGE("read view map failed %d", rView.error());
            work->result = C2_CORRUPTED;
            return;
        }
    }
    bool codecConfig = ((work->input.flags & C2FrameData::FLAG_CODEC_CONFIG) !=0);
    bool eos = ((work->input.flags & C2FrameData::FLAG_END_OF_STREAM) != 0);
    ALOGV("in buffer attr. size %zu timestamp %d frameindex %d, flags %x",
          inSize, (int)work->input.ordinal.timestamp.peeku(),
          (int)work->input.ordinal.frameIndex.peeku(), work->input.flags);
    if (codecConfig) {
        // Ignore CSD buffer for VP9.
        if (mMode == MODE_VP9) {
            fillEmptyWork(work);
            return;
        } else {
            ALOGW("WARNING: Got CSD buffer for VP8. Continue");
        }
    }

    if (inSize) {
        uint8_t *bitstream = const_cast<uint8_t *>(rView.data() + inOffset);
        vpx_codec_err_t err = vpx_codec_decode(
                mCodecCtx, bitstream, inSize, &work->input.ordinal.frameIndex, 0);
        if (err != VPX_CODEC_OK) {
            ALOGE("on2 decoder failed to decode frame. err: %d", err);
            mSignalledError = true;
            work->workletsProcessed = 1u;
            work->result = C2_CORRUPTED;
            return;
        }
    }
    status_t err = outputBuffer(pool, work);
    if (err == NOT_ENOUGH_DATA) {
        if (inSize > 0) {
            ALOGV("Maybe non-display frame at %lld.",
                  work->input.ordinal.frameIndex.peekll());
            // send the work back with empty buffer.
            inSize = 0;
        }
    } else if (err != OK) {
        ALOGD("Error while getting the output frame out");
        // work->result would be already filled; do fillEmptyWork() below to
        // send the work back.
        inSize = 0;
    }

    if (eos) {
        drainInternal(DRAIN_COMPONENT_WITH_EOS, pool, work);
        mSignalledOutputEos = true;
    } else if (!inSize) {
        fillEmptyWork(work);
    }
}

3、调用流程图

4、总结

后续有其他更正继续补充。