生产者与消费者简介

全面接触生产者/消费者问题是在操作系统原理中，并发性原理讨论的问题生产者/消费者问题。最近的工作偏向音频，接着上一篇文章，用生产者，消费者模型来理解Android音频。

维基百科给出的描述

In computing, the producer–consumer problem[1][2] (also known as the bounded-buffer problem) is a classic example of a multi-process synchronization problem. The problem describes two processes, the producer and the consumer, who share a common, fixed-size buffer used as a queue. The producer's job is to generate data, put it into the buffer, and start again. At the same time, the consumer is consuming the data (i.e., removing it from the buffer), one piece at a time. The problem is to make sure that the producer won't try to add data into the buffer if it's full and that the consumer won't try to remove data from an empty buffer.

生产者的主要作用是生成一定量的数据放到缓冲区中，然后重复此过程。与此同时，消费者也在缓冲区消耗这些数据。该问题的关键就是要保证生产者不会在缓冲区满时加入数据，消费者也不会在缓冲区中空时消耗数据。

Android中生产者与消费者

生产者与消费者模式在Android普遍存在这里以Android audio为代表分析。

上篇media player 提到了minibuffersize 在audio 是一个十分重要的阈值。生产者写完数据，等待消费者消费，当消费者消费到数据size到了minibuffersize是，唤醒生产者生产填充数据，在消费者进行将来的消费前把数据消费完，进入等待。填充，等待，消费，唤醒，填充，等待，消费，唤醒。这是生产数据快，消费者速度慢的情况下。当两者速度相当时，一边生产，一边消费。当生产慢时就进入了underrun模式。

直接主要述说Framework层

talk is cheap ，shou me the code.

生产者

从Java上层（AudioTrack.java)根据Android框架进行分析，使用简单的流程图

public int write(@NonNull short[] audioData, int offsetInShorts, int sizeInShorts) {
    return write(audioData, offsetInShorts, sizeInShorts, WRITE_BLOCKING);
}

WRITE_BLOCKING 和 WRITE_NON_BLOCKING 解析：
  /**
    * The write mode indicating the write operation will block until all data has been written。
    */

   /**
    * The write mode indicating the write operation will return immediately after
    * queuing as much audio data for playback as possible without blocking。
    */

public int write(@NonNull byte[] audioData, int offsetInBytes, int sizeInBytes,
        @WriteMode int writeMode) {

    if (mState == STATE_UNINITIALIZED || mAudioFormat == AudioFormat.ENCODING_PCM_FLOAT) {
        return ERROR_INVALID_OPERATION;
    }
    // 生产数据只有两种情况，一种是blocking，另外一种是non blocking
    if ((writeMode != WRITE_BLOCKING) && (writeMode != WRITE_NON_BLOCKING)) {
        return ERROR_BAD_VALUE;
    }
...
    return ret;
}

/* 这个函数主要检查，然后进入JNI函数*/

static jint android_media_AudioTrack_write_native_bytes(JNIEnv *env,  jobject thiz,
        jbyteArray javaBytes, jint byteOffset, jint sizeInBytes,
        jint javaAudioFormat, jboolean isWriteBlocking) {
   ...
    jint written = writeToTrack(lpTrack, javaAudioFormat, bytes.get(), byteOffset,
            sizeInBytes, isWriteBlocking == JNI_TRUE /* blocking */);

    return written;
}

template <typename T>
static jint writeToTrack(const sp& track, jint audioFormat, const T *data,
                         jint offsetInSamples, jint sizeInSamples, bool blocking) {
    // give the data to the native AudioTrack object (the data starts at the offset)
    ssize_t written = 0;
    // regular write() or copy the data to the AudioTrack's shared memory?
    size_t sizeInBytes = sizeInSamples * sizeof(T);
/* 是否使用Sharememory  处理,就是生产者与消费者 在这里是通过ShareMemeory来构建*/
    if (track->sharedBuffer() == 0) {
        written = track->write(data + offsetInSamples, sizeInBytes, blocking);
       ... (经过一下处理，将数据填入ShareMemeory）
        memcpy(track->sharedBuffer()->pointer(), data + offsetInSamples, sizeInBytes);
        written = sizeInBytes;
    }

..

}

下面将Track在上面填充数据到buffer

ssize_t AudioTrack::write(const void* buffer, size_t userSize, bool blocking)
{
    ...
    Buffer audioBuffer;
    // 进入主要生产者填充，直到userSize填满满
    while (userSize >= mFrameSize) {
        audioBuffer.frameCount = userSize / mFrameSize;

        status_t err = obtainBuffer(&audioBuffer,
                blocking ? &ClientProxy::kForever : &ClientProxy::kNonBlocking);
//填充数据空间的大小，然后进行数据拷贝，变量状态更新,接着release。  

        releaseBuffer(&audioBuffer);

}

上述主要调用obtainbuffer 和releasebuffer。

status_t AudioTrack::obtainBuffer(Buffer* audioBuffer, const struct timespec *requested,
        struct timespec *elapsed, size_t *nonContig)
{
 ...
        {   // start of lock scope,通过互斥量 来解决生产/消费问题）
            AutoMutex lock(mLock);
            newSequence = mSequence;
            // did previous obtainBuffer() fail due to media server death or voluntary invalidation?（是否上一次obtainBuffer失败是由于media server挂掉或者自动放弃治疗？）
          ...
最终进入代理相关接口
        // FIXME starts the requested timeout and elapsed over from scratch
        status = proxy->obtainBuffer(&buffer, requested, elapsed);
    } while (((status == DEAD_OBJECT) || (status == NOT_ENOUGH_DATA)) && (tryCounter-- > 0));

}

obtainBuffer比较长，主要做的内容：

判断IAudioTrack异常，状态进行对应的处理
通过proxy去获取buffer
如果发现track挂掉了，并且次数还没用完则再次循环。
封装数据，返回状态。

接下来就是proxy obtainbuffer (frameworks/av/media/libmedia/AudioTrackShared.cpp)同样代理对应的obtainbuffer也比较长

status_t ClientProxy::obtainBuffer(Buffer* buffer, const struct timespec *requested,
        struct timespec *elapsed)
{
    ...
    // check for shared memory corruption （检查下share memory是否发生内存脏块，如果有问题，就直接go end）
    if (mIsShutdown) {
        status = NO_INIT;
        goto end;
    }
    for (;;) {
        int32_t flags = android_atomic_and(~CBLK_INTERRUPT, &cblk->mFlags);
      ...
        // write to rear, read from front 计算出当前已经填充的量，就是尚未被消耗的帧
        ssize_t filled = rear - front;
        // pipe should not be overfull
        if (!(0 <= filled && (size_t) filled <= mFrameCount)) {
            if (mIsOut) {
                ALOGE("Shared memory control block is corrupt (filled=%zd, mFrameCount=%zu); "
                        "shutting down", filled, mFrameCount);
                mIsShutdown = true;
                status = NO_INIT;
                goto end;
            }
      ...
        ssize_t adjustableSize = (ssize_t) getBufferSizeInFrames();
        ssize_t avail =  (mIsOut) ? adjustableSize - filled : filled;
      ....
//正常唤醒、出错、超时等都会继续在跑循环。
            switch (error) {
            case 0:            // normal wakeup by server, or by binderDied()
            case EWOULDBLOCK:  // benign race condition with server
            case EINTR:        // wait was interrupted by signal or other spurious wakeup
            case ETIMEDOUT:    // time-out expired
                // FIXME these error/non-0 status are being dropped
                break;
            default:
                status = error;
                ALOGE("%s unexpected error %s", __func__, strerror(status));
                goto end;
            }

这里只是简单说明。这两个函数都超过500行，比较大，需要仔细，搞明白逻辑。

消费者

在这里消费者是AudioFlinger threadLooper.中MixerThreadLoop。这里不考虑Fast ，重采样，效果链，这样分析比较简单。

大致过程：

对应的track被添加到active list，可以检索到了，会去判断当前的track的状态，以及是否上次循环已经有mixed过了，来决定它的填充帧数的要求；

判断是否ready了，第一次的填充需要将app申请缓冲区都填满，之后就会将mFillingUpStatus设置为filled，那之后这边就直接pass了

判断暂停、停止、终止等状态

如果以上都pass的话，那就会开始进行audiomixer的设置，包括音量，缓冲区，格式等一系列可能需要在mixer中完成的参数告诉它，因为可能牵扯到downmix、reformat、resampler等等。并且使能audiomixer ，更新下mMixerStatus和mMixerStatusIgnoringFastTracks 如果有效果的话，则对缓冲区进行相关清理等工作。

在Loop 中比较重要的两个函数prepareTracks_l和进行分析。

 AudioFlinger::MixerThread::prepareTracks_l(Vector< sp > *tracksToRemove)
{
..
    mixer_state mixerStatus = MIXER_IDLE;
    // 看有多少个track需要被处理的
    size_t count = mActiveTracks.size();
    size_t mixedTracks = 0;
    size_t tracksWithEffect = 0;
针对tracklist ,x

    mMixerBufferValid = false;  // 有使用mixerbuffer的tracks被找到并且有数据会被置为true
    mEffectBufferValid = false; // 有使用mixerbuffer的tracks被找到并且有数据会被置为true

    for (size_t i=0 ; i<count ; i++) {
        const sp t = mActiveTracks[i].promote();
        if (t == 0) {
            continue;
        }

        Track* const track = t.get();

        {   // local variable scope to avoid goto warning

        audio_track_cblk_t* cblk = track->cblk();

        // 对于第一次被添加进来的track，我们会等待它把buffer都填充满才来处理的。
        int name = track->name();
        // 为了防止用户在结束的时候没有通过调用stop函数来终止播放，而是依赖underrun来结束，添加了变量
        // mMixerStatus来记录上一次的track的mix的情况。
        // mMixerStatus == MIXER_TRACKS_READY意味着上一次的循环，track已经被mixed过了。
        // 也侧面反映了，上一次还是有数据的，如果上一次没有被mixed过的话，那我们就直接采用最小的来帮忙
        // 将剩下的数据排空
        size_t desiredFrames;
        const uint32_t sampleRate = track->mAudioTrackServerProxy->getSampleRate();
        AudioPlaybackRate playbackRate = track->mAudioTrackServerProxy->getPlaybackRate();
        // 确保有足够的数据来填充满hal层的一个buffer
        // 计算hal层buffer映射到当前速率，采样率下是多少帧
        desiredFrames = sourceFramesNeededWithTimestretch(
                sampleRate, mNormalFrameCount, mSampleRate, playbackRate.mSpeed);
        // 数据已经被AudioMixer获取，但是尚未release，所以需要加上这个部分，因为此时client还没呈现
        // 出来变化
        desiredFrames += mAudioMixer->getUnreleasedFrames(track->name());

        uint32_t minFrames = 1;
        // minFrames需要AudioMixer在上一轮的操作中已经被enable了，并且track的需要满足下面条件
        // 才会被设置为desiredFrames，为什么pausing的时候，也不给机会呢，这个时候缓冲区的数据可能
        // 没办法完全满足desiredFrames，但是我们依然要制造volume ramp，所以只能这么做了。
        if ((track->sharedBuffer() == 0) && !track->isStopped() && !track->isPausing() &&
                (mMixerStatusIgnoringFastTracks == MIXER_TRACKS_READY)) {
            minFrames = desiredFrames;
        }

        size_t framesReady = track->framesReady();
        // 对于pause的场景，是会直接戛然而止，而不会将app写进去尚未消费的数据用掉
        // 对于stop的场景， 是会将数据都用光的。
        if ((framesReady >= minFrames) && track->isReady() &&
                !track->isPaused() && !track->isTerminated())
        {
            mixedTracks++;

            chain.clear();
            if (track->mainBuffer() != mSinkBuffer &&
                    track->mainBuffer() != mMixerBuffer) {
                if (mEffectBufferEnabled) {
                    mEffectBufferValid = true; // Later can set directly.
                }
                // 能进入到这边，说明这个track关联到了一个效果链中。
                chain = getEffectChain_l(track->sessionId());
                if (chain != 0) {
                    tracksWithEffect++;
                } else {
                }
            }

            int param = AudioMixer::VOLUME;
            if (track->mFillingUpStatus == Track::FS_FILLED) {
                // no ramp for the first volume setting
                // 对于第一次的音量设置，不采用ramp volume，一旦进来就被设置为FS_ACTIVE了。
                track->mFillingUpStatus = Track::FS_ACTIVE;
                // RESUMING只有pausing或者PAUSED才可能触发。
                // 可以想到的场景就是，start了之后数据尚未满足ready，然后就被pause，然后再被start。
                if (track->mState == TrackBase::RESUMING) {
                    track->mState = TrackBase::ACTIVE;
                    param = AudioMixer::RAP_MVOLUME;
                }
          ....(音量，效果链等.).

            // 如果上一次没有ready，这次只要有一个track已经ready，则为ready
            if (mMixerStatusIgnoringFastTracks != MIXER_TRACKS_READY ||
                    mixerStatus != MIXER_TRACKS_ENABLED) {
                mixerStatus = MIXER_TRACKS_READY;
            }
        } else {
            if (framesReady < desiredFrames && !track->isStopped() && !track->isPaused()) {
                track->mAudioTrackServerProxy->tallyUnderrunFrames(desiredFrames);
            }
            // clear effect chain input buffer if an active track underruns to avoid sending
            // previous audio buffer again to effects
            chain = getEffectChain_l(track->sessionId());
            if (chain != 0) {
                chain->clearInputBuffer();
            }
            // 如果track是使用共享内存，或者已经终止、停止、暂停，则会进入下面逻辑。 
            if ((track->sharedBuffer() != 0) || track->isTerminated() ||
                    track->isStopped() || track->isPaused()) {
                // 会在继续填充latency的时间长度的空白数据，这个有优化空间
                size_t audioHALFrames = (latency_l() * mSampleRate) / 1000;
                size_t framesWritten = mBytesWritten / mFrameSize;
                if (mStandby || track->presentationComplete(framesWritten, audioHALFrames)) {
                    // 如果track是stopped状态，则重置track。
                    if (track->isStopped()) {
                        track->reset();
                    }
                    // 加入到移除队列
                    tracksToRemove->add(track);
                }
            } else {
                // 如果不是共享内存，并且当前状态不是终止、停止、暂停，就说明肯定就是数据给的不够。
                // 这个时候会给它50次的机会进行填充，如果还不满足就移掉active list了。
                if (--(track->mRetryCount) <= 0) {
                    tracksToRemove->add(track);
                    // 告诉app的track由于underrun被disable，如果数据够，会自动调用start()方法
                    android_atomic_or(CBLK_DISABLED, &cblk->mFlags);
                // 只要有一个active track尚未ready，那就标志mixer states为not ready
                // 但是这边有个例外的，上一次已经设置为enable，并且有两个track，一个ready，另外
                // 一个not ready，则这次的mixer states会被设置为READY
                } else if (mMixerStatusIgnoringFastTracks == MIXER_TRACKS_READY ||
                                mixerStatus != MIXER_TRACKS_READY) {
                    mixerStatus = MIXER_TRACKS_ENABLED;
                }
            }
          ...

code 太长，逻辑太多就比较复杂。

针对整个过程用一张时序图来总结。

ITPub博客

生产者与消费者之Android audio

最新文章