Histreamer框架分析

Histreamer是OpenHarmony的轻量级的媒体引擎，提供播放、录制等场景的媒体数据流水线处理。

Histreamer是OpenHarmony的轻量级的媒体引擎，提供播放、录制等场景的媒体数据流水线处理。

Histreamer使用pipeline管理音频数据，一条pipeline由若干个filter组成。Filter的功能通过plugins插件来实现，支持静态和动态注册插件的方式。

1. 框架分析

1.1 Plugin的注册和管理

本节首先介绍什么是插件，插件是如何进行注册和管理的，然后介绍新增一个插件需要做什么事情。

1.1.1 PluginLoader

PluginLoader 主要是用于动态插件的加载和注册，它使用 dlopen() 打开动态库，并通过 dlsym() 从动态库中提取注册和反注册方法。注册插件方法的名字为 “register_“ + name ，反注册插件方法的名字为 “unregister_” + name。

一个loader实例对应一个动态注册的插件。

1.1.2 PluginManager

PluginManager 是对外提供的插件管理类，它是一个单例实现。典型使用如下：

Plugin::PluginManager::Instance().ListPlugins(pluginType)

数据结构：

PluginManager把插件的注册封装到 pluginRegister_类型中。Filter可以通过 CreaterxxxPlugin() 方法创建对应的插件。

创建插件：

创建插件就是从提前注册好的 pluginRegister_ 中找到注册器信息，然后调用它们自己的creator() 方法。

template<typename T, typename U>
std::shared_ptr<T> CreatePlugin(const std::string& name, PluginType pluginType)
{
    auto regInfo = pluginRegister_->GetPluginRegInfo(pluginType, name);
    if (!regInfo) {
        return {};
    }
    auto plugin = ReinterpretPointerCast<U>(regInfo->creator(name));
    if (!plugin) {
        return {};
    }
    return std::shared_ptr<T>(
        new T(regInfo->packageDef->pkgVersion, regInfo->info->apiVersion, plugin));
}

假设我们要创建一个FileSource的插件，那么这里会调用到FileSource注册时提供的creator方法：

std::shared_ptr<SourcePlugin> FileSourcePluginCreator(const std::string& name)
{
	return std::make_shared<FileSourcePlugin>(name);
}

其实就是创建一个对象。

1.1.3 PluginRegister

PluginRegister 依赖内部类 RegisterImpl 内部类来完成实际的注册管理工作。实际是在私有成员 registerData_ 中保存所有的注册信息。

管理所有注册信息的数据结构 RegisterData ：

register_data

1.1.4 插件注册流程

以FileSource为例说明插件注册的流程。

定义插件

PLUGIN_DEFINITION(FileSource, LicenseType::APACHE_V2, FileSourceRegister, [] {});

宏展开后得到：

extern "C" __attribute__((dllexport))OHOS::Media::Plugin::Status
register_FileSource(const std::shared_ptr<OHOS::Media::Plugin::PackageRegister> &pkgReg) {
    pkgReg->AddPackage({(((((1)) & 0xFFFF) << 16) | (((0)) & 0xFFFF)), "FileSource", LicenseType::APACHE_V2});
    std::shared_ptr<OHOS::Media::Plugin::Register> pluginReg = pkgReg;
    return FileSourceRegister(pluginReg);
}
extern "C" __attribute__((dllexport))void unregister_FileSource() { 
	[] {}(); 
}

可以看到，是定义了一个注册函数 register_FileSource() 和一个反注册函数 unregister_FileSource() 。

插件注册流程

程序流程图：

插件注册

核心流程说明如下：

1）PluginManager初始化时调用PluginRegister的注册注册插件方法 RegisterPlugins()；

2）RegisterPlugins() 调用 RegisterStaticPlugins() 注册静态插件，调用 RegisterDynamicPlugins() 注册动态插件；

3) RegisterPluginStatic() 会依次调用各个静态插件的注册函数：

void RegisterPluginStatic(std::shared_ptr<OHOS::Media::Plugin::Register> reg)
{
    REGISTER_STATIC(FileSource, reg);
    REGISTER_STATIC(FileFdSource, reg);
    REGISTER_STATIC(WavDemuxer, reg);
    REGISTER_STATIC(FFmpegDemuxer, reg);
    REGISTER_STATIC(FFmpegMuxers, reg);
    REGISTER_STATIC(FFmpegAudioDecoders, reg);
    REGISTER_STATIC(FFmpegAudioEncoders, reg);
	...
}

4）每个插件的注册函数又会反过来调用注册器提供的 AddPlugin() 接口

5）AddPlugin() 会检查参数合法性以及版本信息，然后在 UpdateRegisterTableAndRegisterNames() 中真正完成一个插件的注册工作。

6）UpdateRegisterTableAndRegisterNames() 主要是根据PluginDefBase得到PluginRegInfo，并把插件添加到私有的registerData数据结构中进行统一管理。

小结：

如何定义一个静态插件：

插件在自己的类中使用宏 PLUGIN_DEFINITION() 来定义插件，它其实是向外导出了一个注册方法和反注册方法；

PLUGIN_DEFINITION(FileSource, LicenseType::APACHE_V2, FileSourceRegister, [] {});

在 all_plugin_static.h 的声明：

PLUGIN_REGISTER_STATIC_DECLARE(FileSource);

在RegisterPluginStatic()中注册插件：

void RegisterPluginStatic(std::shared_ptr<OHOS::Media::Plugin::Register> reg) {}{
	...
    REGISTER_STATIC(FileSource, reg);
	...
}

典型的插件注册函数要完成两件事情：定义PluginDefBase数据结构，调用 reg->AddPlugin() 注册插件。

1.2 Pipeline、Filter和Port

上一节介绍了什么是插件，本节介绍pipeline、filter和port。

Pipeline是数据流的管道，是整体，pipeline由一个个filter过滤器组成。

filter会对音频数据进行处理，主要有source/sink、decoder/encoder、demuxer/muxer这几种filter，它们分别负责数据的输入/输出、解码/编码、解封装/封装。

port是filter的端口，分为inPort和outPort，一个filter可以有一个或多个inPorts或outPorts。

1.2.1 数据结构

类图：

class_diagram

数据结构的说明：

一个播放器创建一个pipeline实例，播放器需实现 EvenReceiver和FilterCallback这两个接口，pipeline和filter可以把状态报给播放器。
pipeline由若干个filter组成。Pipeline派生自EventReceiver接口，在FilterBase初始化时，Pipeline把自己传给FilterBase的eventReceiver成员变量，后续Filter就可以通过eventReceiver->OnEvent()把状态回传给pipeline。
filter有两组Ports：InPorts和OutPorts，这里不是两个而是两组，原因是比如demuxer这样的filter可以有多个输出的。Port属于某一个Filter。

1.2.2 pipeline-状态机

pipeline管理着所有filter的状态，因此，pipeline的启动停止都是由filter完成的，pipeline保存着一个FilterState类型的私有成员变量state_，所以pipeline自己的状态，其实就是它所管理的filter的状态。

pipeline的状态机如下：

state

可以看到，在prepare的时候，pipeline需要收到所有的filter的ready事件，才会变成ready状态，才可以start。

具体代码如下：

void PipelineCore::OnEvent(const Event& event)
{
    if (event.type != EventType::EVENT_READY) {
        NotifyEvent(event);
        return;
    }
    readyEventCnt_++;
    // 所有的filter均ready，才是真的ready。
    if (readyEventCnt_ == filters_.size()) {
        state_ = FilterState::READY;
        NotifyEvent(event);
        readyEventCnt_ = 0;
    }
}

1.2.3 pipeline-添加和连接filter

PipelineCore::AddFilters()方法向pipeline添加一个filter。添加filter只是简单地将其添加到PipelineCore的成员变量 filters_ 中。

PipelineCore::LinkFilter() 将filter连接起来，实现下图这样的效果：

filter_and_ports

1.2.4 FilterFactory-注册和创建filter

所有向系统注册的filter生成器，保存在FilterFactory的generator()这个map中。

FilterFactory是Filter的工厂类，但是另外封装了AutoRegisterFilter() 这个帮助类给开发者使用。

register_filter

举个例子说明注册插件的流程：

注册插件

向系统注册一个demuxer filter，只需要在它自己的类中定义一个静态的生成器对象：

1 2	// demuxer_filter.cpp static AutoRegisterFilter<DemuxerFilter> g_registerFilterHelper("builtin.player.demuxer");

AutoRegisterFilter() 这个模板类在构造函数中就会调用FilterFactory的方法向系统注册一个Filter：

1
2
3

explicit AutoRegisterFilter(const std::string& name) {
    FilterFactory::Instance().RegisterFilter<T>(name);
}

RegisterFilter() 方法其实是向系统中注册了一个默认的生成器：

void RegisterFilter(const std::string& name) {
	// 默认的生成器就是new一个对象
	RegisterGenerator(name, [](const std::string& aliaName) { return std::make_shared<T>(aliaName); });
}

补充说明：系统也支持filter定义定义自己的生成器：

1
2
3

AutoRegisterFilter(const std::string& name, const InstanceGenerator& generator) {
	FilterFactory::Instance().RegisterGenerator(name, generator);
}

而RegisterGenerator() 方法就是把生成器对象添加到FilterFactory的generators这个map中进行统一管理：

void FilterFactory::RegisterGenerator(const std::string& name, const InstanceGenerator& generator) {
    auto result = generators.emplace(name, generator);
    if (!result.second) {
        result.first->second = generator;
    }
}

创建插件

可以通过FilterFactory的工厂方法创建插件：

1 2	// hiplayer_impl.cpp cdemuxer_ = FilterFactory::Instance().CreateFilterWithType<DemuxerFilter>("builtin.player.demuxer", "demuxer");

它的实现就是根据插件的名字，从生成器map中找到生成器，然后调用生成器函数创造一个对应类型的filter对象。

1.2.5 Filter的工作机制

filterbase

Init()

给eventReceiver_和callback_赋值，都是传进来的pipeline，用于回调。

InitPorts() 初始化：对inPorts_和outPorts_进行初始化，它们的第一个成员都是FilterBase自己。对routeMap_进行初始化，这里保存的是in到out的路由，第一个成员是自己指向自己。

GetInPort()，GetOutPort()，GetNextFilters()，GetPreFilters()，FindPort()

这些都是根据filter和port得到的帮助函数，理解这个数据结构即可：

filter_and_ports

OnEvent

OnEvent(const Event& event) 用于port给filter和pipeline传递信息

OnEvent(const Plugin::PluginEvent &event) 用于plugin给filter和pipeline传递信息

plugin_cb

Prepare()

Start()，Pause()，Stop()除了修改state外，都没做什么实际的事情。

在Prepare()的时候会调用Inport的Active()方法，把port激活，filter也就开始数据的搬运工作了。

ErrorCode FilterBase::Prepare() {
    state_ = FilterState::PREPARING;
    // Filter默认InPort按Push方式获取数据
    // 只有 Demuxer 的 Prepare() 需要覆写此实现， Demuxer的 InPort可能按 Pull 或 Push 方式获取数据
    WorkMode mode;
    return GetInPort(PORT_NAME_DEFAULT)->Activate({WorkMode::PUSH}, mode);
}

ConfigPluginWithMeta(plugin，meta)

对meta参数进行合法性检查之后，调用plugin的Setparameter()方法给plugin设置参数

主要是codec和sink相关的filter需要配置一些格式信息。

比如如下日志：

CodecFilterBase INFO (codec_filter_base.cpp, 238) : Func(Configure) receive upstream meta Meta{mime:(string)audio/mpeg, track_id:(uint32_t)0, bits_per_coded_sample:(uint32_t)0, duration:(int64_t)262512437500, bit_rate:(int64_t)128000, channels:(uint32_t)2, channel_layout:(AudioChannelLayout)STEREO, sample_rate:(uint32_t)44100, sample_fmt:(AudioSampleFormat)F32P, sample_per_frame:(uint32_t)1152, output_channels:(uint32_t)2, output_channel_layout:(AudioChannelLayout)STEREO, ad_mpeg_ver:(uint32_t)1, ad_mpeg_layer:(uint32_t)3}

UpdateAndInitPluginByInfo()

创建plugin并调用plugin的init()

template <typename T>
bool FilterBase::UpdateAndInitPluginByInfo(std::shared_ptr<T>& plugin, std::shared_ptr<Plugin::PluginInfo>& pluginInfo,
	const std::shared_ptr<Plugin::PluginInfo>& selectedPluginInfo,
	const std::function<std::shared_ptr<T>(const std::string&)>& pluginCreator)
{
    ...
    plugin = pluginCreator(selectedPluginInfo->name);
    if (plugin == nullptr) {
    	MEDIA_LOG_E("cannot create plugin " PUBLIC_LOG_S, selectedPluginInfo->name.c_str());
    	return false;
    }
    auto err = TranslatePluginStatus(plugin->Init());
    ...
    return true;
}

1.3 Filter和plugin的工作方式

1.3.1 Soure

以MediaSourceFilter和HttpSourcePlugin为例分析Source的流程。这里主要分析数据的流程，关于状态和配置太细节了，就不分析了。

类图

HttpSourcePlugin依赖DownloaderMonitor来完成网络数据的请求。它的类图如下：

downloader

其中，Downloader会创建一个Task，不断调用HttpDownloadLoop从网络请求数据保存到本地的buffer。这里是数据的生产者。

plugin下载网络数据

request_data

1
2
3

// 最关键的一句，网络数据被保存到buffer中

buffer_->WriteBuffer(data, len)

MediaSourceFilter循环读取数据

Push模式

如果seek模式是unseekable，那么使用push模式：

std::vector<WorkMode> MediaSourceFilter::GetWorkModes()
{
    if (seekable_ == Seekable::SEEKABLE) {
    	return {WorkMode::PUSH, WorkMode::PULL};
    } else {
    	return {WorkMode::PUSH};
    }
}

在push模式下，表明这是一个live stream，那么会创建一个DataReader数据读取线程，循环读取数据送往demuxer节点。

MediaSourceFilter创建的DataReader Task，不断调用ReadLoop()方法，从plugin不断读数据，然后调用 outPorts_[0]->PushData() 把数据写到它的输出端口。-

Pull模式

如果是pull模式，则不会创建线程。而是等待demuxer从source filter来pull数据。

ErrorCode MediaSourceFilter::PullData(const std::string& outPort, uint64_t offset, size_t size, AVBufferPtr& data)
{
    ...
    plugin_->Read(data, readSize);
    ...
}

至此，数据从source节点送到了demuxer节点。

1.3.2 Demuxer

Demuxer在Prepare的时候，会从它的输入节点获取工作模式，来激活自己的PULL或者PUSH模式。

ErrorCode DemuxerFilter::Prepare()
{
    dataPacker_->Start();
    pluginState_ = DemuxerState::DEMUXER_STATE_NULL;
    task_->RegisterHandler([this] { DemuxerLoop(); });
    Pipeline::WorkMode mode;
    // 从Source filter获取WorkMode
    GetInPort(PORT_NAME_DEFAULT)->Activate({Pipeline::WorkMode::PULL, Pipeline::WorkMode::PUSH}, mode);
    if (mode == Pipeline::WorkMode::PULL) {
        dataPacker_->Flush();
        ActivatePullMode();
    } else {
    	ActivatePushMode();
    }
    state_ = FilterState::PREPARING;
    return ErrorCode::SUCCESS;
}

push mode和pull mode的区别：

ActivatePullMode()时，提供的checkRange()接口在数据不足时，主动从inPort pull数据：

checkRange_ = [this](uint64_t offset, uint32_t size) {
    uint64_t curOffset = offset;
    // 如果数据够，则什么也不需要做
    if (dataPacker_->IsDataAvailable(offset, size, curOffset)) {
    	return true;
    }

    AVBufferPtr bufferPtr = std::make_shared<AVBuffer>();
    bufferPtr->AllocMemory(pluginAllocator_, size);
    // 如果数据不够，则手动向上一个节点pull数据
    auto ret = inPorts_.front()->PullData(curOffset, size, bufferPtr);
    if (ret == ErrorCode::SUCCESS) {
        // 拿到的数据传给dataPacker_
        dataPacker_->PushData(std::move(bufferPtr), curOffset);
        return true;
    } 
    ...
    return false;
}

ActivatePushMode() 时，则不会在checkRange_时手动pull数据，而是等待上一个节点调用DemuxerFilter的PushData() 给我们写数据：

ErrorCode DemuxerFilter::PushData(const std::string& inPort, const AVBufferPtr& buffer, int64_t offset)
{
    if (buffer->flag & BUFFER_FLAG_EOS) {
    	dataPacker_->SetEos();
    } else {
    	// 被上一个节点调用，给dataPacker_写数据
    	dataPacker_->PushData(std::move(buffer), offset);
    }
    return ErrorCode::SUCCESS;
}

类图：

demuxer_class

DemuxerFilter通过plugin_->SetDataSource() 把DataSource传给FFmpegDemuxerPlugin使用；
插件调用DataSource的ReadAt()接口读取原始数据；
DataSourceImpl的ReadAt()接口会使用DemuxerFilter的checkRange_()、peek Range_()和getRange_()这几个接口读取数据，它们从DataPacker来获取数据的。

数据流程：

demuxer

这里展示了利用FFmpeg来做为DemuxerPlugin解析数据的流程。

DemuxerFilter总是会创建DemuxerLoop线程，对数据进行解封装，DemuxerLoop()不断调用ReadFrame()方法向FFmpegDemuxerPlugin插件读取解封装之后的数据帧；
FFmpegDemuxerPlugin调用av_read_frame() 这个三方库的接口读取数据帧，读取动作触发三方库使用初始化传进的AVReadPacket()读数据接口获取解封装前的数据，对数据进行demux处理之后，数据通过av_read_frame()返回给插件。
DemuxerFilter读取frame之后，调用HandleFrame()方法，通过 stream.port->PushData() 把数据发给下一个节点。

至此，数据在demuxer节点完成了解封装，并到了下一个节点。

1.3.3 Decoder

解码器节点是何时创建的？

在Prepare阶段，demuxer filter在获取到mediaInfo后，会根据mediaInfo调用PrepareStreams() 来准备数据流。然后根据mediaInfo的track的数量生成outPorts，接着回调给HiplayerImpl PORT_ADDED 事件。播放器在收到新的PORT通知时，根据portDesc的信息，创建对应数量的sink节点。如果数据不是pcm数据流，则还创建decoder filter节点。

port_added

类图：

codec_class

解码流程：

decoder

解码器有三个线程：

audioDecAsyncDecodeFrame
audioDecAsyncPush
audioDecAsyncHandleFrame

这三个线程里面异步解码，最后调用 oPort->PushData() 到下一个filter。

1.3.4 Sink

sink节点不复杂，主要是有一个MediaSync同步机制，这里暂不分析。

2. Code flow

2.1 创建播放器

1
2
3

auto engineFactory = std::unique_ptr<OHOS::Media::IEngineFactory>(CreateEngineFactory());

auto player = engineFactory->CreatePlayerEngine();

创建播放器

创建创建播放器主要做两个事情：

HiPlayerImpl() 实例得到player
调用player的Init()

HiPlayerImpl的构造函数：

在初始化task_私有成员，创建 callbackThread 回调线程，它的线程处理函数是：HiPlayerCallbackLooper::LoopOnce()
创建FilterFactory的实例并进行初始化：

FilterFactory::Instance().Init();

依次创建 syncManager_、audioSource_、demuxer_、audioSink_、pipeline_这些必要的私有成员。

player->Init()

pipeline->Init()；
给pipeline添加filter，这里AddFilter的时候会掉Filter进行初始化；
连接filter。

由此可见：对于一个player而言，必要的filter是 AudioSource 和 Demuxer。

2.2 SetSource

player->SetSource(url);

set_source

2.3 Prepare

prepare1 prepare2