AndroidNuPlayer要点详解.doc

Android NuPlayer要点详解1、AHandler机制首先介绍NuPlayer中无处不在的AHandler机制 frameworks/av/include/media/stagefright/foundation/ frameworks/av/media/libstagefright/foundation/ AHandler是Android native层实现的一个异步消息机制，在这个机制中所有的处理都是异步的，将变量封装到一个消息AMessage结构体中，然后放到队列中去，后台专门有一个线程会从这个队列中取出消息然后执行，执行函数就是onMessageReceived。Ahandler机制包括以下几个类AMessage消息类，用于构造消息，通过post方法投递出去给ALooperstatus_t AMessage:post(int64_t delayUs) sp looper = mLmote(); if (looper = NULL) ALOGW(failed to post message as target looper for handler %d is gone., mTarget); return -ENOENT; looper-post(this, delayUs); return OK;void AMessage:deliver() sp handler = mHmote(); if (handler = NULL) ALOGW(failed to deliver message as target handler %d is gone., mTarget); return; handler-deliverMessage(this); /see AHandlerdeliverMessage，前面通过looper post最后就是调用这里的deliever送到handler手里AHandler 消息处理类，一般当做父类，继承该类的子类需要实现onMessageReceived方法void AHandler:deliverMessage(const sp &msg) onMessageReceived(msg); mMessageCounter+;.ALooper 与Ahander一一对应，负责存储消息并分发Ahandler的消息，与AMessage一对多关系/ posts a message on this looper with the given timeoutvoid ALooper:post(const sp &msg, int64_t delayUs) Mutex:Autolock autoLock(mLock); int64_t whenUs; if (delayUs 0) whenUs = GetNowUs() + delayUs; else whenUs = GetNowUs(); List:iterator it = mEventQueue.begin(); while (it != mEventQueue.end() & (*it).mWhenUs run( mName.empty() ? ALooper : mName.c_str(), priority); if (err != OK) mThread.clear(); return err;bool ALooper:loop() Event event; Mutex:Autolock autoLock(mLock); if (mThread = NULL & !mRunningLocally) return false; if (mEventQueue.empty() mQueueChangedCondition.wait(mLock); return true; int64_t whenUs = (*mEventQueue.begin().mWhenUs; int64_t nowUs = GetNowUs(); if (whenUs nowUs) int64_t delayUs = whenUs - nowUs; mQueueChangedCondition.waitRelative(mLock, delayUs * 1000ll); return true; event = *mEventQueue.begin(); mEventQueue.erase(mEventQueue.begin(); event.mMessage-deliver();/see AHandler.deliverMessage. return true;LooperThread 此线程调用ALooper的loop方法来分发消息 virtual status_t readyToRun() mThreadId = androidGetThreadId(); return Thread:readyToRun(); virtual bool threadLoop() return mLooper-loop();ALooperRoaster 与Handler是一对多的关系， 管理Looper和Handler一一对应关系，负责释放stale handlerALooper:handler_id ALooperRoster:registerHandler( const sp looper, const sp &handler) Mutex:Autolock autoLock(mLock); if (handler-id() != 0) CHECK(!A handler must only be registered once.); return INVALID_OPERATION; HandlerInfo info; info.mLooper = looper; info.mHandler = handler; ALooper:handler_id handlerID = mNextHandlerID+;/一对一 mHandlers.add(handlerID, info);/一对多 handler-setID(handlerID, looper); return handlerID;void ALooperRoster:unregisterHandler(ALooper:handler_id handlerID) Mutex:Autolock autoLock(mLock); ssize_t index = mHandlers.indexOfKey(handlerID); if (index 0) return; const HandlerInfo &info = mHandlers.valueAt(index); sp handler = info.mHmote(); if (handler != NULL) handler-setID(0, NULL); mHandlers.removeItemsAt(index);void ALooperRoster:unregisterStaleHandlers() Vectorsp activeLoopers; Mutex:Autolock autoLock(mLock); for (size_t i = mHandlers.size(); i 0;) i-; const HandlerInfo &info = mHandlers.valueAt(i); sp looper = info.mLmote(); if (looper = NULL) ALOGV(Unregistering stale handler %d, mHandlers.keyAt(i); mHandlers.removeItemsAt(i); else / At this point looper might be the only sp keeping / the object alive. To prevent it from going out of scope / and having ALooper call this method again recursively / and then deadlocking because of the Autolock above, add / it to a Vector which will go out of scope after the lock / has been released. activeLoopers.add(looper); 异步消息机制的创建sp mLooper = new ALooper; /创建一个Alooper实例sp mHandler = new AHandlerReflector /创建一个Ahandler实例mLooper-setName(“xxxxx”); /设置looper名字mLooper-start(false, true, PRIORITY_XXX); /根据参数创建并启动 looper threadmLooper-regiserHandler(mHandler); /register handler 会调用AHandler的setID方法将looper设置到Handler里去Post消息sp msg = new AMessage(kWhatSayGoodbye, mHandler); /在AMessage的构造方法里获取Ahandler对应的Looper并保存msg-post(); / 调用looper的post方法Message Post的调用过程Message:postALooper:postmEventQueue.insertmQueueChangedCondition.signal() /如果之前没有event，通知looper threadALooper:loop()if (mEventQueue.empty() /如果消息队列为空，则等待mQueueChangedCondition.wait(mLock);return true;event = *mEventQueue.begin();event.mMessage-deliver();AHandler:deliverMessageAHandlerReflector: onMessageReceived具体的实现NuPlayer下面就进入我们的正题，NuPlayer frameworks/av/media/libmediaplayerservice/nuplayer/NuPlayerDriver NuPlayerDriver是对NuPlayer的封装，继承MediaPlayerInterface接口。通过NuPlayer来实现播放的功能。看这部分代码的方法就是先看NuPlayerDriver里面干了啥，转头就去找NuPlayer里面的实现，一般都要再去NuPlayer的onMessageReceive中看消息的响应，最后回到NuPlayerDriver的各种notify中看流程的周转，下面附上一张播放器状态机流转图NuPlayerDriver:NuPlayerDriver(pid_t pid) : mState(STATE_IDLE), /对应播放器状态机的初始化状态 mIsAsyncPrepare(false), mAsyncResult(UNKNOWN_ERROR), mSetSurfaceInProgress(false), mDurationUs(-1), mPositionUs(-1), mSeekInProgress(false), mLooper(new ALooper), mPlayerFlags(0), mAtEOS(false), mLooping(false), mAutoLoop(false) ALOGV(NuPlayerDriver(%p), this); /和前面所述的异步消息创建机制相符mLooper-setName(NuPlayerDriver Looper); mLooper-start( false, /* runOnCallingThread */ true, /* canCallJava */ PRIORITY_AUDIO);/mPlayer即NuPlayer，继承于AHandler mPlayer = AVNuFactory:get()-createNuPlayer(pid); mLooper-registerHandler(mPlayer); mPlayer-setDriver(this);NuPlayerDriver:NuPlayerDriver() ALOGV(NuPlayerDriver(%p), this); mLooper-stop(); /整个NuPlayerDriver就是一个大ALooperAVNuFactory 负责关键组件的create，通过它能看到： 1.每一个NuPlayer对应一个进程 2.数据流从Source-Decoder-Renderer，中间由AMessages驱动sp AVNuFactory:createNuPlayer(pid_t pid) return new NuPlayer(pid);sp AVNuFactory:createPassThruDecoder( const sp ¬ify, const sp &source, const sp &renderer) return new NuPlayer:DecoderPassThrough(notify, source, renderer);sp AVNuFactory:createDecoder( const sp ¬ify, const sp &source, pid_t pid, const sp &renderer) return new NuPlayer:Decoder(notify, source, pid, renderer);sp AVNuFactory:createRenderer( const sp &sink, const sp ¬ify, uint32_t flags) return new NuPlayer:Renderer(sink, notify, flags);下面分别分析Source, Decoder, RendererSource以setDataSource为切入点status_t NuPlayerDriver:setDataSource(const sp &source) ALOGV(setDataSource(%p) stream source, this); Mutex:Autolock autoLock(mLock); if (mState != STATE_IDLE) return INVALID_OPERATION; mState = STATE_SET_DATASOURCE_PENDING; mPlayer-setDataSourceAsync(source);/因为driver只是NuPlayer的封装，所以还是要去调用NuPlayer完成实际动作 while (mState = STATE_SET_DATASOURCE_PENDING) mCondition.wait(mLock); return mAsyncResult;void NuPlayer:setDataSourceAsync(const sp &source) sp msg = new AMessage(kWhatSetDataSource, this); sp notify = new AMessage(kWhatSourceNotify, this); msg-setObject(source, new StreamingSource(notify, source); msg-post();/到了NuPlayer中，也不是直接进行操作，而是先发个消息，验证前面所说的一切都由AMessage驱动void NuPlayer:onMessageReceived(const sp &msg) switch (msg-what() case kWhatSetDataSource:/实际的处理在这里 ALOGV(kWhatSetDataSource); CHECK(mSource = NULL); status_t err = OK; sp obj; CHECK(msg-findObject(source, &obj); if (obj != NULL) Mutex:Autolock autoLock(mSourceLock); mSource = static_cast(obj.get();/赋值给mSource else err = UNKNOWN_ERROR; CHECK(mDriver != NULL); sp driver = mDmote(); if (driver != NULL) driver-notifySetDataSourceCompleted(err);/通知driver设置完毕 break; .void NuPlayerDriver:notifySetDataSourceCompleted(status_t err) Mutex:Autolock autoLock(mLock); CHECK_EQ(mState, STATE_SET_DATASOURCE_PENDING); mAsyncResult = err; mState = (err = OK) ? STATE_UNPREPARED : STATE_IDLE;/回到driver中，流转播放器状态进入下一阶段 mCondition.broadcast();下面就来看看具体有哪些source，它们都继承自NuPlayer:Source（NuPlayerSource.h & NuPlayerSource.cpp） 1.HTTP-进一步判断是以下的哪一种：HTTPLiveSource,RTSPSource,GenericSource 2.File-GenericSource 3.StreamSource-StreamingSource 4.DataSource-GenericSourceGenericSourcenuplayer/GenericSource.h & GenericSource.cpp几个水位static int64_t kLowWaterMarkUs = 2000000ll; / 2secsstatic int64_t kHighWaterMarkUs = 5000000ll; / 5secsstatic int64_t kHighWaterMarkRebufferUs = 15000000ll; / 15secs,这一个是新增加的水位static const ssize_t kLowWaterMarkBytes = 40000;static const ssize_t kHighWaterMarkBytes = 200000;status_t NuPlayer:GenericSource:initFromDataSource() init extractor;get track info and metadatavoid NuPlayer:GenericSource:prepareAsync() if (mLooper = NULL) mLooper = new ALooper; mLooper-setName(generic); mLooper-start(); mLooper-registerHandler(this); sp msg = new AMessage(kWhatPrepareAsync, this); msg-post();status_t NuPlayer:GenericSource:feedMoreTSData() return OK;LiveSessionlibstagefright/httplive/LiveSession.h & cpp/ static/ Bandwidth Switch Mark Defaultsconst int64_t LiveSession:kUpSwitchMarkUs = 15000000ll;const int64_t LiveSession:kDownSwitchMarkUs = 20000000ll;const int64_t LiveSession:kUpSwitchMarginUs = 5000000ll;const int64_t LiveSession:kResumeThresholdUs = 100000ll;/ Buffer Prepare/Ready/Underflow Marksconst int64_t LiveSession:kReadyMarkUs = 5000000ll;const int64_t LiveSession:kPrepareMarkUs = 1500000ll;const int64_t LiveSession:kUnderflowMarkUs = 1000000ll;与Fetcher,Bandwidth Estimater(和ExoPlayer一样是滑动窗口平均),switching,Buffering相关的操作都在这里HTTPLiveSourcenuplayer目录下enum Flags / Dont log any URLs.不在log中记录URL kFlagIncognito = 1, ;NuPlayer:HTTPLiveSource:HTTPLiveSource( if (headers) /也搞了一个header机制 mExtraHeaders = *headers; ssize_t index = mExtraHeaders.indexOfKey(String8(x-hide-urls-from-log); if (index = 0) mFlags |= kFlagIncognito; mExtraHeaders.removeItemsAt(index); void NuPlayer:HTTPLiveSource:prepareAsync() if (mLiveLooper = NULL) mLiveLooper = new ALooper;/一如既往的ALooper mLiveLooper-setName(http live); mLiveLooper-start(); mLiveLooper-registerHandler(this); sp notify = new AMessage(kWhatSessionNotify, this); mLiveSession = new LiveSession( notify, (mFlags & kFlagIncognito) ? LiveSession:kFlagIncognito : 0, mHTTPService); mLiveLooper-registerHandler(mLiveSession); mLiveSession-connectAsync(/HTTPLiveSource包含LiveSession，很多实际的工作都由LiveSession完成 mURL.c_str(), mExtraHeaders.isEmpty() ? NULL : &mExtraHeaders);ATSParser frameworks/av/media/libstagefright/mpeg2ts/ATSParser.cpp 就是一个TS Parser，虽然也叫Axx，但是没有消息机制在里面StreamingSource nuplayer目录void NuPlayer:StreamingSource:prepareAsync() if (mLooper = NULL) mLooper = new ALooper; mLooper-setName(streaming); mLooper-start();/何其相似 mLooper-registerHandler(this); notifyVideoSizeChanged(); notifyFlagsChanged(0); notifyPrepared();StreamingSource中的数据由onReadBuffer驱动，最后的EOS,Discontiunity等都交给ATSParser去处理，ATSParser又最终交给AnotherPacketSource去做真正的处理实际上，这里提到的三个Source最后都会用到AnotherPacketSourcevoid NuPlayer:StreamingSource:onReadBuffer() for (int32_t i = 0; i kNumListenerQueuePackets; +i) char buffer188; sp extra; ssize_t n = mStreamListener-read(buffer, sizeof(buffer), &extra);/实际用NuPlayerStreamListener完成工作 if (n = 0) ALOGI(input data EOS reached.); mTSParser-signalEOS(ERROR_END_OF_STREAM);/EOS了 setError(ERROR_END_OF_STREAM); break; else if (n = INFO_DISCONTINUITY) int32_t type = ATSParser:DISCONTINUITY_TIME; int32_t mask; if (extra != NULL & extra-findInt32( IStreamListener:kKeyDiscontinuityMask, &mask) if (mask = 0) ALOGE(Client specified an illegal discontinuity type.); setError(ERROR_UNSUPPORTED); break; type = mask; mTSParser-signalDiscontinuity( (ATSParser:DiscontinuityType)type, extra); else if (n setInt64(IStreamListener:kKeyMediaTimeUs, mediaTimeUs); mTSParser-signalDiscontinuity( (type & 1) = 0) ? ATSParser:DISCONTINUITY_TIME : ATSParser:DISCONTINUITY_FORMATCHANGE, extra); else status_t err = mTSParser-feedTSPacket(buffer, sizeof(buffer); if (err != OK) ALOGE(TS Parser returned error %d, err); mTSParser-signalEOS(err); setError(err); break; AnotherPacketSourceframeworks/av/media/libstagefright/mpeg2ts 可以类比ExoPlayer中的chunk source，同时负责buffer管理，EOSDiscontinuity的处理等等前面三个Source最后都会落到AnotherPacketSourcebool AnotherPacketSource:hasBufferAvailable(status_t *finalResult) Mutex:Autolock autoLock(mLock); *finalResult = OK; if (!mEnabled) return false; if (!mBuffers.empty() /一个ABuffer List，其实就是一个环形缓冲 return true; *finalResult = mEOSResult; return false;void AnotherPacketSource:queueDiscontinuity( ATSParser:DiscontinuityType type, const sp &extra, bool discard) Mutex:Autolock autoLock(mLock); if (discard) / Leave only discontinuities in the queue. mEOSResult = OK; mLastQueuedTimeUs = 0; mLatestEnqueuedMeta = NULL; if (type = ATSParser:DISCONTINUITY_NONE) return; mDiscontinuitySegments.push_back(DiscontinuitySegment(); sp buffer = n