14. Android MultiMedia框架完全解析 - NuPlayerDecoder与MediaCodec的交互

前言

上一篇文章中详细分析了MediaCodec,以及由它向下的内容,但是在MediaCodec外面包裹的是一层NuPlayerDecoder,这里就看看它们两者之间是如何沟通的。

从理论上来讲,既然NuPlayerDecoder包裹在MediaCodec外层,所以它相对于MediaCodec也可以理解为App,它调用MediaCodec的API来完成一些任务。下面就详细看看这个流程

1. 解码顺序的启动过程

实际的解码是从setRenderer开始,在NuPlayer::onStart()函数中,

mVideoDecoder->setRenderer(mRenderer);

到NuPlayer::DecoderBase::setRenderer()函数中,发送kWhatSetRenderer这个msg,然后跳到onSetRenderer()函数中,

void NuPlayer::Decoder::onSetRenderer(const sp<Renderer> &renderer) {
    bool hadNoRenderer = (mRenderer == NULL);
    mRenderer = renderer;
    if (hadNoRenderer && mRenderer != NULL) {
        // this means that the widevine legacy source is ready
        onRequestInputBuffers();
    }
}

在onSetRenderer函数中,调用到onRequestInputBuffers()函数,这个函数挺奇葩的,来看一下它的执行逻辑:

void NuPlayer::DecoderBase::onRequestInputBuffers() {
    if (mRequestInputBuffersPending) {
        return;
    }
 
    // doRequestBuffers() return true if we should request more data
    if (doRequestBuffers()) {
        mRequestInputBuffersPending = true;
 
        sp<AMessage> msg = new AMessage(kWhatRequestInputBuffers, this);
        msg->post(2 * 1000ll);
    }
}
 
void NuPlayer::DecoderBase::onMessageReceived(const sp<AMessage> &msg) {
	case kWhatRequestInputBuffers:
        {
            mRequestInputBuffersPending = false;
            onRequestInputBuffers();
            break;
        }

看到这个代码逻辑了没,在onRequestInputBuffers函数中,会去发送kWhatRequestInputBuffers这个msg,而在msg的async处理函数中,还会继续去调用onRequestInputBuffers函数,就这样循环下去了,嗯哼???还有这操作?

那唯一能阻止这个循环操作的,就是onRequestInputBuffers函数中的if (doRequestBuffers())判断语句了,只有它判断为0,才会终止这个循环操作。

那么再来看这个doRequestBuffers()函数,它实际上是一个while循环,当需要更多的数据时,这个函数返回true :

/*
 * returns true if we should request more data
 */
bool NuPlayer::Decoder::doRequestBuffers() {
    // mRenderer is only NULL if we have a legacy widevine source that
    // is not yet ready. In this case we must not fetch input.
    if (isDiscontinuityPending() || mRenderer == NULL) {
        return false;
    }
    status_t err = OK;
    while (err == OK && !mDequeuedInputBuffers.empty()) {
        size_t bufferIx = *mDequeuedInputBuffers.begin();
        sp<AMessage> msg = new AMessage();
        msg->setSize("buffer-ix", bufferIx);
        err = fetchInputData(msg);  //取一个输入buffer
        if (err != OK && err != ERROR_END_OF_STREAM) {
            // if EOS, need to queue EOS buffer
            break;
        }
        mDequeuedInputBuffers.erase(mDequeuedInputBuffers.begin());
 
        if (!mPendingInputMessages.empty()
                || !onInputBufferFetched(msg)) {
            mPendingInputMessages.push_back(msg); //实际取出的数据放到这个缓冲消息队列中
        }
    }
 
    return err == -EWOULDBLOCK
            && mSource->feedMoreTSData() == OK;
}

在NuPlayer::Decoder::fetchInputData()函数中,通过调用mSource->dequeueAccessUnit()函数,来跟GenericSource打交道,就把压缩数据填充到buffer中了。

而NuPlayer::Decoder::onInputBufferFetched()函数内部,会通过mCodec->queueInputBuffer()来把数据加到BufferQueue中。

2. 循环逻辑

来看看下面这个图,我的理解是NuPlayerDecoder是包装在MediaCodec外面的一层,而这个input和output都是相对于MediaCodec来说的,NuPlayerDecoder与MediaCodec交互发生在两个port,MediaCodec中也维护着一个BufferQueue,当inputport端口有buffer时,就会调用MediaCodec::onInputBufferAvailable()函数,这个函数就会发送一个CB_INPUT_AVAILABLE msg到NuPlayerDecoder中,通知它在MediaCodec的input port有一个buffer,那么NuPlayerDecoder就会相应的调用NuPlayer::Decoder::handleAnInputBuffer()函数来处理这个事情,怎么处理呢?MediaCodec的作用就是解码,Decoder就是从demux(MediaExtractor)中取数据,交给MediaCodec去处理。

当MediaCodec处理完这些数据后(怎么处理?把H264的码流解析成YUV格式的),数据在内部从input port流到ouput port,这时候就会触发MediaCodec::onOutputBufferAvailable()函数,来告诉NuPlayerDecoder在MediaCodec的ouput port有一个buffer,通过发送一个CB_OUTPUT_AVAILABLE msg,当NuPlayerDecoder接收到这个msg后,调用NuPlayer::Decoder::handleAnOutputBuffer()函数来处理,怎么处理呢?

Decoder的下一步就是Render了,所以下一步就是把数据发送给Renderer。
14. Android MultiMedia框架完全解析 - NuPlayerDecoder与MediaCodec的交互
这是一个循环逻辑,一步一步来看。

2.1 异步消息的建立过程

首先是NuPlayerDecoder是如何与MediaCodec建立消息联系的?

在NuPlayer::Decoder::onConfigure()函数中有如下的代码:

sp<AMessage> reply = new AMessage(kWhatCodecNotify, this); //这是消息类型为kWhatCodecNotify
mCodec->setCallback(reply);

然后在MediaCodec::setCallback函数:

status_t MediaCodec::setCallback(const sp<AMessage> &callback) {
    sp<AMessage> msg = new AMessage(kWhatSetCallback, this);
    msg->setMessage("callback", callback);

    sp<AMessage> response;
    return PostAndAwaitResponse(msg, &response);
}

void MediaCodec::onMessageReceived(const sp<AMessage> &msg) {
    switch (msg->what()) {
    	case kWhatSetCallback:
        {
            sp<AReplyToken> replyID;
            CHECK(msg->senderAwaitsResponse(&replyID));
            ...
            sp<AMessage> callback;
            CHECK(msg->findMessage("callback", &callback));

            mCallback = callback;  //将callback取出来赋值给全局变量mCallback
            ...

            sp<AMessage> response = new AMessage;
            response->postReply(replyID);
            break;
        }
    }
}

这里就将NuPlayerDecoder的reply赋值给MediaCodec的mCallback。
需要回调onInputBufferAvailable给NuPlayerDecoder时我们看一下,此时就是用mCallback去发送消息

void MediaCodec::onInputBufferAvailable() {
    int32_t index;
    while ((index = dequeuePortBuffer(kPortIndexInput)) >= 0) {
        sp<AMessage> msg = mCallback->dup();
        msg->setInt32("callbackID", CB_INPUT_AVAILABLE); 
        msg->setInt32("index", index);
        msg->post();
    }
}

这里就是将MediaCodec的消息发送给当前类Decoder消息队列中,然后在NuPlayer::Decoder::onMessageReceived中处理。

void NuPlayer::Decoder::onMessageReceived(const sp<AMessage> &msg) {
    ALOGV("[%s] onMessage: %s", mComponentName.c_str(), msg->debugString().c_str());

    switch (msg->what()) {
        case kWhatCodecNotify:
        {
            int32_t cbID;
            CHECK(msg->findInt32("callbackID", &cbID));
            ...
            switch (cbID) {
                case MediaCodec::CB_INPUT_AVAILABLE:
                {
                    int32_t index;
                    CHECK(msg->findInt32("index", &index));

                    handleAnInputBuffer(index);
                    break;
                }

2.2 向MediaCodec输入数据的过程

首先是在MediaCodec内部会向NuPlayerDecoder通过MediaCodec::onInputBufferAvailable() 函数向NuPlayerDecoder函数发送一个CB_INPUT_AVAILABLE msg,在NuPlayerDecoder中:

bool NuPlayer::Decoder::handleAnInputBuffer(size_t index) {
    if (isDiscontinuityPending()) {
        return false;
    }
 
    sp<ABuffer> buffer;
    mCodec->getInputBuffer(index, &buffer); //首先从MediaCodec中获取一个可用的输入缓冲
 
    if (buffer == NULL) {
        handleError(UNKNOWN_ERROR);
        return false;
    }
 
    if (index >= mInputBuffers.size()) {
        for (size_t i = mInputBuffers.size(); i <= index; ++i) {
            mInputBuffers.add();
            mMediaBuffers.add();
            mInputBufferIsDequeued.add();
            mMediaBuffers.editItemAt(i) = NULL;
            mInputBufferIsDequeued.editItemAt(i) = false;
        }
    }
    mInputBuffers.editItemAt(index) = buffer;
 
    //CHECK_LT(bufferIx, mInputBuffers.size());
 
    if (mMediaBuffers[index] != NULL) {
        mMediaBuffers[index]->release();
        mMediaBuffers.editItemAt(index) = NULL;
    }
    mInputBufferIsDequeued.editItemAt(index) = true;
 
    if (!mCSDsToSubmit.isEmpty()) {
        sp<AMessage> msg = new AMessage();
        msg->setSize("buffer-ix", index);
 
        sp<ABuffer> buffer = mCSDsToSubmit.itemAt(0);
        ALOGI("[%s] resubmitting CSD", mComponentName.c_str());
        msg->setBuffer("buffer", buffer);
        mCSDsToSubmit.removeAt(0);
        CHECK(onInputBufferFetched(msg));
        return true;
    }
 
    while (!mPendingInputMessages.empty()) {
        sp<AMessage> msg = *mPendingInputMessages.begin();
        if (!onInputBufferFetched(msg)) {//这里是把数据queue到MediaCodec的Input中
//在这个循环中,更重要的一点是对EOS进行处理,如果遇到EOS,这里就会跳出循环。
            break;
        }
        mPendingInputMessages.erase(mPendingInputMessages.begin());
    }
 
    if (!mInputBufferIsDequeued.editItemAt(index)) {
        return true;
    }
 
    mDequeuedInputBuffers.push_back(index);
 
    onRequestInputBuffers();
    return true;
}

真正的循环在onRequestInputBuffers()函数中,这个函数如下所示:

void NuPlayer::DecoderBase::onRequestInputBuffers() {
    if (mRequestInputBuffersPending) {
        return;
    }
 
    // doRequestBuffers() return true if we should request more data
    if (doRequestBuffers()) {
        mRequestInputBuffersPending = true;
 
        sp<AMessage> msg = new AMessage(kWhatRequestInputBuffers, this);
        msg->post(2 * 1000ll);
    }
}

如果还需要获取数据的话,doRequestBuffers()函数就会返回true,然后发送kWhatRequestInputBuffers msg,这个msg的处理函数中仍会去调用onRequestInputBuffers()函数,所以就这么一直循环下去了。所以核心是这个doRequestBuffers()函数。

bool NuPlayer::Decoder::doRequestBuffers() {
    // mRenderer is only NULL if we have a legacy widevine source that
    // is not yet ready. In this case we must not fetch input.
    if (isDiscontinuityPending() || mRenderer == NULL) {
        return false;
    }
    status_t err = OK;
    while (err == OK && !mDequeuedInputBuffers.empty()) {
        size_t bufferIx = *mDequeuedInputBuffers.begin();
        sp<AMessage> msg = new AMessage();
        msg->setSize("buffer-ix", bufferIx);
        err = fetchInputData(msg);
        if (err != OK && err != ERROR_END_OF_STREAM) {
            // if EOS, need to queue EOS buffer
            break;
        }
        mDequeuedInputBuffers.erase(mDequeuedInputBuffers.begin());
 
        if (!mPendingInputMessages.empty()
                || !onInputBufferFetched(msg)) {
            mPendingInputMessages.push_back(msg);
        }
    }
 
    return err == -EWOULDBLOCK
            && mSource->feedMoreTSData() == OK;
}

这里面有2个重要的函数,fetchInputData(msg)函数是从Source里面取数据,onInputBufferFetched()函数里面填充数据,并把数据传给MediaCodec。来看看onInputBufferFetched()函数是如何完成这些操作的:

bool NuPlayer::Decoder::onInputBufferFetched(const sp<AMessage> &msg) {
    size_t bufferIx;
    CHECK(msg->findSize("buffer-ix", &bufferIx));
    CHECK_LT(bufferIx, mInputBuffers.size());
    sp<ABuffer> codecBuffer = mInputBuffers[bufferIx];
 
    sp<ABuffer> buffer;
    bool hasBuffer = msg->findBuffer("buffer", &buffer);
 
    // handle widevine classic source - that fills an arbitrary input buffer
    MediaBuffer *mediaBuffer = NULL;
    if (hasBuffer) {
        mediaBuffer = (MediaBuffer *)(buffer->getMediaBufferBase());
        if (mediaBuffer != NULL) {
            // likely filled another buffer than we requested: adjust buffer index
            size_t ix;
            for (ix = 0; ix < mInputBuffers.size(); ix++) {
                const sp<ABuffer> &buf = mInputBuffers[ix];
                if (buf->data() == mediaBuffer->data()) {
                    // all input buffers are dequeued on start, hence the check
                    if (!mInputBufferIsDequeued[ix]) {
                        ALOGV("[%s] received MediaBuffer for #%zu instead of #%zu",
                                mComponentName.c_str(), ix, bufferIx);
                        mediaBuffer->release();
                        return false;
                    }
 
                    // TRICKY: need buffer for the metadata, so instead, set
                    // codecBuffer to the same (though incorrect) buffer to
                    // avoid a memcpy into the codecBuffer
                    codecBuffer = buffer;
                    codecBuffer->setRange(
                            mediaBuffer->range_offset(),
                            mediaBuffer->range_length());
                    bufferIx = ix;
                    break;
                }
            }
            CHECK(ix < mInputBuffers.size());
        }
    }
 
    if (buffer == NULL /* includes !hasBuffer */) {
        int32_t streamErr = ERROR_END_OF_STREAM;
        CHECK(msg->findInt32("err", &streamErr) || !hasBuffer);
 
        CHECK(streamErr != OK);
 
        // attempt to queue EOS
        status_t err = mCodec->queueInputBuffer(
                bufferIx,
                0,
                0,
                0,
                MediaCodec::BUFFER_FLAG_EOS);
        if (err == OK) {
            mInputBufferIsDequeued.editItemAt(bufferIx) = false;
        } else if (streamErr == ERROR_END_OF_STREAM) {
            streamErr = err;
            // err will not be ERROR_END_OF_STREAM
        }
 
        if (streamErr != ERROR_END_OF_STREAM) {
            ALOGE("Stream error for %s (err=%d), EOS %s queued",
                    mComponentName.c_str(),
                    streamErr,
                    err == OK ? "successfully" : "unsuccessfully");
            handleError(streamErr);
        }
    } else {
        sp<AMessage> extra;
        if (buffer->meta()->findMessage("extra", &extra) && extra != NULL) {
            int64_t resumeAtMediaTimeUs;
            if (extra->findInt64(
                        "resume-at-mediaTimeUs", &resumeAtMediaTimeUs)) {
                ALOGI("[%s] suppressing rendering until %lld us",
                        mComponentName.c_str(), (long long)resumeAtMediaTimeUs);
                mSkipRenderingUntilMediaTimeUs = resumeAtMediaTimeUs;
            }
        }
 
        int64_t timeUs = 0;
        uint32_t flags = 0;
        CHECK(buffer->meta()->findInt64("timeUs", &timeUs));
 
        int32_t eos, csd;
        // we do not expect SYNCFRAME for decoder
        if (buffer->meta()->findInt32("eos", &eos) && eos) {
            flags |= MediaCodec::BUFFER_FLAG_EOS;
        } else if (buffer->meta()->findInt32("csd", &csd) && csd) {
            flags |= MediaCodec::BUFFER_FLAG_CODECCONFIG;
        }
 
        // copy into codec buffer
        if (buffer != codecBuffer) {
            CHECK_LE(buffer->size(), codecBuffer->capacity());
            codecBuffer->setRange(0, buffer->size());
            memcpy(codecBuffer->data(), buffer->data(), buffer->size());
//实际复制数据的地方
        }
 
        status_t err = mCodec->queueInputBuffer( //在这里把buffer交给解码器Codec
                        bufferIx,
                        codecBuffer->offset(),
                        codecBuffer->size(),
                        timeUs,
                        flags);
        if (err != OK) {
            if (mediaBuffer != NULL) {
                mediaBuffer->release();
            }
            ALOGE("Failed to queue input buffer for %s (err=%d)",
                    mComponentName.c_str(), err);
            handleError(err);
        } else {
            mInputBufferIsDequeued.editItemAt(bufferIx) = false;
            if (mediaBuffer != NULL) {
                CHECK(mMediaBuffers[bufferIx] == NULL);
                mMediaBuffers.editItemAt(bufferIx) = mediaBuffer;
            }
        }
    }
    return true;
}

至此,就把数据交给MediaCodec了,解码前的数据就准备好了。

2.3 MediaCodec解码后的数据流向

在MediaCodec的output port有数据时,就会调用MediaCodec::onOutputBufferAvailable()函数,这个函数就是发送CB_OUTPUT_AVAILABLE msg到NuPlayerDecoder中,同样在NuPlayer::Decoder::onMessageReceived函数中进行处理:

case MediaCodec::CB_OUTPUT_AVAILABLE:
    {
        int32_t index;
        size_t offset;
        size_t size;
        int64_t timeUs;
        int32_t flags;

        CHECK(msg->findInt32("index", &index));
        CHECK(msg->findSize("offset", &offset));
        CHECK(msg->findSize("size", &size));
        CHECK(msg->findInt64("timeUs", &timeUs));
        CHECK(msg->findInt32("flags", &flags));

        handleAnOutputBuffer(index, offset, size, timeUs, flags);
        break;
    }

继续看:

bool NuPlayer::Decoder::handleAnOutputBuffer(
        size_t index,
        size_t offset,
        size_t size,
        int64_t timeUs,
        int32_t flags) {
//    CHECK_LT(bufferIx, mOutputBuffers.size());
    sp<ABuffer> buffer;
    mCodec->getOutputBuffer(index, &buffer);
 
    if (index >= mOutputBuffers.size()) {
        for (size_t i = mOutputBuffers.size(); i <= index; ++i) {
            mOutputBuffers.add();
        }
    }
 
    mOutputBuffers.editItemAt(index) = buffer;
 
    buffer->setRange(offset, size);
    buffer->meta()->clear();
    buffer->meta()->setInt64("timeUs", timeUs);
 
    bool eos = flags & MediaCodec::BUFFER_FLAG_EOS;
    // we do not expect CODECCONFIG or SYNCFRAME for decoder
 
    sp<AMessage> reply = new AMessage(kWhatRenderBuffer, this);
    reply->setSize("buffer-ix", index);
    reply->setInt32("generation", mBufferGeneration);
 
    if (eos) {
        ALOGI("[%s] saw output EOS", mIsAudio ? "audio" : "video");
 
        buffer->meta()->setInt32("eos", true);
        reply->setInt32("eos", true);
    } else if (mSkipRenderingUntilMediaTimeUs >= 0) {
        if (timeUs < mSkipRenderingUntilMediaTimeUs) {
            ALOGV("[%s] dropping buffer at time %lld as requested.",
                     mComponentName.c_str(), (long long)timeUs);
 
            reply->post();
            return true;
        }
 
        mSkipRenderingUntilMediaTimeUs = -1;
    }
 
    mNumFramesTotal += !mIsAudio;
 
    // wait until 1st frame comes out to signal resume complete
    notifyResumeCompleteIfNecessary();
 
    if (mRenderer != NULL) {
        // send the buffer to renderer.
        mRenderer->queueBuffer(mIsAudio, buffer, reply);
        if (eos && !isDiscontinuityPending()) {
            mRenderer->queueEOS(mIsAudio, ERROR_END_OF_STREAM);
        }
    }
 
    return true;
}

通过这个mRenderer->queueBuffer()函数,就把数据发送给Renderer了。

Render中根据时间来判断这个Buffer是否需要渲染,是否需要丢帧,然后通过一个notify反馈给NuPlayerDecoder,然后在下面的onRenderBuffer处理函数中,会根据这个notify来判断是否渲染这一帧数据。

同时这里还发送了一个kWhatRenderBuffer msg,这个处理函数如下:

case kWhatRenderBuffer:
    {
        if (!isStaleReply(msg)) {
            onRenderBuffer(msg);
        }
        break;
    }
 
 
void NuPlayer::Decoder::onRenderBuffer(const sp<AMessage> &msg) {
    status_t err;
    int32_t render;
    size_t bufferIx;
    int32_t eos;
    CHECK(msg->findSize("buffer-ix", &bufferIx));
 
    if (!mIsAudio) {
        int64_t timeUs;
        sp<ABuffer> buffer = mOutputBuffers[bufferIx];
        buffer->meta()->findInt64("timeUs", &timeUs);
 
        if (mCCDecoder != NULL && mCCDecoder->isSelected()) {
            mCCDecoder->display(timeUs);
        }
    }
 
    if (msg->findInt32("render", &render) && render) {
        int64_t timestampNs;
        CHECK(msg->findInt64("timestampNs", &timestampNs));
        err = mCodec->renderOutputBufferAndRelease(bufferIx, timestampNs);
    } else {
        mNumOutputFramesDropped += !mIsAudio;
        err = mCodec->releaseOutputBuffer(bufferIx);
    }
    if (err != OK) {
        ALOGE("failed to release output buffer for %s (err=%d)",
                mComponentName.c_str(), err);
        handleError(err);
    }
    if (msg->findInt32("eos", &eos) && eos
            && isDiscontinuityPending()) {
        finishHandleDiscontinuity(true /* flushOnTimeChange */);
    }
}

这个(msg->findInt32(“render”, &render) && render)就是从Renderer中传回来的notify,如果需要渲染的话,这个判断语句就是true,会调用mCodec->renderOutputBufferAndRelease函数渲染这一帧数据,然后释放buffer,把buffer返还给MediaCodec。

如果这个语句判断为false,就会直接调用mCodec->releaseOutputBuffer函数释放buffer,不去渲染,直接把buffer返还给MediaCodec。

至此,Decoder的过程也就分析完毕了,以后有时间的话再分析MediaCodec有OMX的交互。下一步先分析Render。

上一篇:c# – Windows MIDI流媒体和SysEx


下一篇:python – 在没有文件扩展名的情况下返回Django中的文件路径