深入剖析ffplay.c之数据队列（1）

大家好，欢迎来到IT知识分享网。

一、ffplay核心功能

ffplay是一个基于 FFmpeg 库的媒体播放器，其主要功能是播放各种格式的音视频文件。以下是ffplay的主要代码流程和核心函数的功能解释：

1. 初始化解析命令行参数，设置各种播放选项，如音频、视频、字幕的开关，显示窗口的大小和位置，同步类型等。注册所有的编解码器、解复用器和协议。初始化 SDL 库，创建窗口和渲染器（如果显示未禁用）。创建用于存储音视频包和帧的队列，以及相关的锁和条件变量。初始化时钟，用于同步音频、视频和外部时钟。
2. 打开输入流根据指定的文件名或输入格式打开输入流。查找并打开音频、视频和字幕流，配置相应的解码器。启动读取线程，从输入流中读取数据包并放入相应的队列。
3. 解码和播放启动音频、视频和字幕解码线程，从队列中获取数据包进行解码。解码后的音频帧通过音频转换和重采样后，放入音频帧队列，等待音频回调函数播放。解码后的视频帧进行格式转换后，放入视频帧队列，等待视频刷新函数显示。字幕帧解码后直接放入字幕帧队列，等待视频刷新函数显示。
4. 事件循环不断监听用户输入事件，如键盘、鼠标事件。根据用户事件执行相应操作，如暂停、播放、切换音视频流、调整音量、全屏切换、seek 等。定期调用视频刷新函数，根据同步策略显示视频帧和音频波形（如果需要）。
5. 清理资源当播放结束或用户退出时，释放所有分配的资源，包括窗口、渲染器、解码器上下文、队列、纹理等。退出 SDL 库和 FFmpeg 相关的网络和全局资源。

核心函数功能解释

1. packet_queue_*系列函数packet_queue_init：初始化数据包队列，创建互斥锁和条件变量。packet_queue_put：将数据包放入队列，若队列已满则等待。packet_queue_get：从队列中获取数据包，若队列为空则等待或根据参数决定是否阻塞。packet_queue_flush：清空队列，释放所有数据包。packet_queue_destroy：销毁队列，释放锁和条件变量。

2. decoder_*系列函数decoder_init：初始化解码器结构体，设置相关参数。decoder_decode_frame：从队列中获取数据包并解码，处理解码结果，如返回解码后的帧、处理结束标志等。decoder_destroy：释放解码器相关资源，如解码器上下文和未处理的数据包。

3. frame_queue_*系列函数frame_queue_init：初始化帧队列，创建互斥锁和条件变量，分配帧内存。frame_queue_peek：获取队列中当前可读的帧，但不移动读指针。frame_queue_peek_next：获取队列中下一个可读的帧，但不移动读指针。frame_queue_peek_last：获取队列中最后一个可读的帧，但不移动读指针。frame_queue_peek_writable：获取可写入的帧位置，若队列已满则等待。frame_queue_peek_readable：获取可读的帧位置，若队列中无可读帧则等待。frame_queue_push：将帧写入队列，移动写指针并发送信号。frame_queue_next：移动读指针到下一个帧，释放当前帧资源并发送信号。frame_queue_nb_remaining：返回队列中未显示的帧数。frame_queue_last_pos：返回队列中最后显示帧的位置。frame_queue_destory：销毁帧队列，释放所有帧资源、锁和条件变量。

4. video_*系列函数video_open：打开视频窗口，设置窗口标题、大小和位置，根据视频帧大小调整显示窗口大小。video_display：显示当前视频帧或音频波形（根据显示模式），包括清空渲染器、上传纹理、设置显示区域和渲染。video_image_display：处理视频图像的显示，包括字幕的渲染和视频帧的上传与显示。video_audio_display：处理音频波形的显示，计算音频数据的显示位置和高度，绘制波形图。

5. audio_*系列函数audio_thread：音频解码线程函数，从音频队列中获取数据包解码，进行格式转换和重采样，将处理后的音频帧放入音频帧队列。audio_decode_frame：解码一个音频帧，进行同步处理、格式转换（如果需要），更新音频时钟。sdl_audio_callback：SDL 音频回调函数，用于将解码后的音频数据填充到音频设备缓冲区，实现音频播放，同时更新音频时钟和同步外部时钟。

6. stream_*系列函数stream_open：打开输入流，初始化视频状态结构体，创建各种队列和线程，启动读取线程。stream_component_open：打开特定类型的流（音频、视频、字幕），配置解码器、打开音频设备、启动解码线程。stream_close：关闭输入流，释放所有相关资源，包括解码器、队列、纹理、窗口等。

7. read_thread函数从输入流中读取数据包，根据数据包类型放入相应的队列（音频、视频、字幕）。处理暂停、seek、循环播放等操作，根据用户指定的播放范围和条件控制数据包的读取和处理。

8. event_loop函数处理用户输入事件，如键盘、鼠标事件，根据事件类型执行相应操作，如暂停、播放、切换流、调整音量、seek 等。定期调用视频刷新函数，根据同步策略更新视频显示。

二、队列相关的剖析

static int packet_queue_put_private(PacketQueue *q, AVPacket *pkt) { MyAVPacketList *pkt1; if (q->abort_request) return -1; pkt1 = av_malloc(sizeof(MyAVPacketList)); if (!pkt1) return -1; pkt1->pkt = *pkt; pkt1->next = NULL; if (pkt == &flush_pkt) q->serial++; pkt1->serial = q->serial; if (!q->last_pkt) q->first_pkt = pkt1; else q->last_pkt->next = pkt1; q->last_pkt = pkt1; q->nb_packets++; q->size += pkt1->pkt.size + sizeof(*pkt1); q->duration += pkt1->pkt.duration; /* XXX: should duplicate packet data in DV case */ SDL_CondSignal(q->cond); return 0; }

packet_queue_put_private函数用于将一个AVPacket放入PacketQueue队列中，是ffplay中处理数据包队列的关键函数之一，主要用于将音频、视频或字幕数据包放入相应的队列，以供后续解码和播放使用。

函数流程

1. 首先检查队列的abort_request标志，如果为真，表示队列已被请求终止，直接返回 -1。
2. 分配一个MyAVPacketList结构体的内存空间，用于存储数据包及其相关信息。如果分配失败，返回 -1。
3. 将传入的AVPacket赋值给新分配的MyAVPacketList结构体的pkt成员，并设置next指针为NULL。
4. 如果放入的数据包是flush_pkt，表示需要刷新队列，将队列的serial（序列号）加 1。然后设置pkt1的serial为队列的当前serial。
5. 如果队列当前为空（即last_pkt为NULL），则将pkt1设置为队列的第一个数据包（first_pkt）；否则，将pkt1添加到队列末尾，即当前最后一个数据包（last_pkt）的后面，并更新last_pkt指针。
6. 更新队列的统计信息，包括数据包数量（nb_packets）、总大小（size）和总时长（duration）。
7. 发送信号（SDL_CondSignal）给等待在该队列条件变量（cond）上的线程，表示队列中有新数据可用。
8. 最后，函数成功将数据包放入队列，返回 0。

注意事项

• 在处理flush_pkt时，通过增加serial来标记队列状态的变化，以便在后续的处理中能够识别出需要特殊处理的情况，如刷新解码器状态等。
• 对队列的操作都在加锁（通过SDL_LockMutex和SDL_UnlockMutex包围的代码块中）的情况下进行，以确保线程安全，防止多个线程同时访问和修改队列导致的数据不一致问题。
• 函数中对于DV（Digital Video）情况的处理只是简单地添加了注释，实际应用中可能需要根据具体需求进一步实现数据包数据的复制逻辑，以满足某些特殊格式或场景的要求。

static int packet_queue_put(PacketQueue *q, AVPacket *pkt) { int ret; SDL_LockMutex(q->mutex); ret = packet_queue_put_private(q, pkt); SDL_UnlockMutex(q->mutex); if (pkt != &flush_pkt && ret < 0) av_packet_unref(pkt); return ret; }

packet_queue_put函数用于将一个AVPacket放入PacketQueue队列中，同时处理了线程安全和错误情况，是ffplay中向数据包队列添加数据的重要接口。

1. 函数声明与变量定义

static int packet_queue_put(PacketQueue *q, AVPacket *pkt) { int ret; 

• 函数接受一个PacketQueue结构体指针q和一个AVPacket结构体指针pkt作为参数。
• 定义一个整型变量ret，用于存储函数调用的返回值。

2. 加锁队列

SDL_LockMutex(q->mutex); 

• 使用SDL_LockMutex函数对队列q的互斥锁进行加锁，确保在多线程环境下对队列的操作是线程安全的。这一步防止其他线程同时访问和修改队列，避免数据竞争和不一致性。

3. 调用私有函数放入数据包

ret = packet_queue_put_private(q, pkt); 

• 调用packet_queue_put_private函数将数据包pkt放入队列q中。这个私有函数负责实际的数据包入队操作，包括分配内存、更新队列指针和统计信息等。ret将接收packet_queue_put_private函数的返回值，以判断数据包是否成功放入队列。

4. 解锁队列

SDL_UnlockMutex(q->mutex); 

• 使用SDL_UnlockMutex函数对队列q的互斥锁进行解锁，释放对队列的独占访问权限，允许其他线程访问队列（如果它们在等待锁的话）。

5. 处理错误情况（非 flush_pkt）

if (pkt!= &flush_pkt && ret < 0) av_packet_unref(pkt); 

• 检查放入的数据包是否不是flush_pkt（刷新包）并且ret小于 0，表示数据包放入队列失败（可能是队列已被终止或内存分配失败等原因）。
• 如果满足条件，调用av_packet_unref函数减少数据包pkt的引用计数，并释放其相关资源（如果引用计数为 0）。这是为了避免在数据包入队失败时造成内存泄漏，确保资源的正确管理。

6. 返回结果

return ret; 

• 返回ret的值，该值表示数据包放入队列的操作结果。如果ret为 0，表示数据包成功放入队列；如果ret为 -1，表示操作失败（如队列已被终止）。其他负值可能根据具体情况表示不同的错误类型（如果packet_queue_put_private函数有更多的错误返回值定义）。

static int packet_queue_put_nullpacket(PacketQueue *q, int stream_index) { AVPacket pkt1, *pkt = &pkt1; av_init_packet(pkt); pkt->data = NULL; pkt->size = 0; pkt->stream_index = stream_index; return packet_queue_put(q, pkt); }

packet_queue_put_nullpacket函数用于向PacketQueue队列中放入一个空数据包（null packet），通常用于表示流的结束或在特定情况下作为占位符，其主要目的是在处理媒体流时，能够在合适的时机向解码器或其他处理模块发送一个特殊的标记，以指示流的某种状态变化或结束条件。

函数流程

1. 创建并初始化空数据包

AVPacket pkt1, *pkt = &pkt1; av_init_packet(pkt); pkt->data = NULL; pkt->size = 0; pkt->stream_index = stream_index; 

• 定义一个AVPacket结构体变量pkt1，并将其地址赋值给pkt指针。
• 使用av_init_packet函数初始化pkt所指向的AVPacket结构体，为其分配必要的资源（如引用计数等）。
• 将pkt的data指针设置为NULL，表示数据包没有实际的数据内容。
• 将pkt的size设置为 0，明确数据包的大小为 0 字节。
• 设置pkt的stream_index为传入的stream_index参数，指定该空数据包所属的流索引，以便在后续处理中能够正确识别该数据包与哪个流相关。

2. 将空数据包放入队列

return packet_queue_put(q, pkt); 

• 调用packet_queue_put函数将初始化后的空数据包pkt放入指定的PacketQueue队列q中。packet_queue_put函数会处理数据包入队的各种操作，包括加锁队列、调用packet_queue_put_private函数进行实际的入队操作、解锁队列以及处理可能的错误情况（如队列已被终止或内存分配失败等）。最终，packet_queue_put函数返回一个整数值，表示数据包入队操作的结果，packet_queue_put_nullpacket函数直接将该结果返回给调用者。

注意事项

• 空数据包在ffplay中的具体用途可能因上下文而异。例如，在处理音频或视频流结束时，可能会发送一个空数据包来通知解码器或播放线程停止处理该流。在某些情况下，也可能用于在流中插入特殊的占位符，以满足特定的同步或处理逻辑需求。
• 由于空数据包没有实际的数据内容，其主要作用是传递与流相关的控制信息或状态标记，因此在处理空数据包时，解码器或其他处理模块需要能够正确识别并处理这种特殊情况，而不是尝试对其进行解码或播放操作。

static int packet_queue_init(PacketQueue *q) { memset(q, 0, sizeof(PacketQueue)); q->mutex = SDL_CreateMutex(); if (!q->mutex) { av_log(NULL, AV_LOG_FATAL, "SDL_CreateMutex(): %s\n", SDL_GetError()); return AVERROR(ENOMEM); } q->cond = SDL_CreateCond(); if (!q->cond) { av_log(NULL, AV_LOG_FATAL, "SDL_CreateCond(): %s\n", SDL_GetError()); return AVERROR(ENOMEM); } q->abort_request = 1; return 0; } 

packet_queue_init函数用于初始化一个PacketQueue结构体，主要用于创建互斥锁和条件变量，以实现对数据包队列的线程安全操作，并初始化队列的相关状态变量。

函数流程

1. 初始化队列结构体

memset(q, 0, sizeof(PacketQueue)); 

• 使用memset函数将PacketQueue结构体q的所有成员初始化为 0。这一步确保了结构体中的所有指针和计数器等成员都处于初始状态，避免了可能的垃圾数据或未初始化状态导致的问题。

2. 创建互斥锁

q->mutex = SDL_CreateMutex(); if (!q->mutex) { av_log(NULL, AV_LOG_FATAL, "SDL_CreateMutex(): %s\n", SDL_GetError()); return AVERROR(ENOMEM); } 

• 调用SDL_CreateMutex函数创建一个互斥锁，并将其赋值给PacketQueue结构体的mutex成员。互斥锁用于保护对队列的并发访问，确保在同一时刻只有一个线程能够访问和修改队列。
• 如果SDL_CreateMutex函数返回NULL，表示互斥锁创建失败。在这种情况下，通过av_log函数记录错误信息，包括SDL_GetError函数返回的详细错误描述，并返回AVERROR(ENOMEM)（表示内存分配错误，因为创建互斥锁失败可能是由于系统资源不足导致无法分配内存给互斥锁结构）。

3. 创建条件变量

q->cond = SDL_CreateCond(); if (!q->cond) { av_log(NULL, AV_LOG_FATAL, "SDL_CreateCond(): %s\n", SDL_GetError()); return AVERROR(ENOMEM); } 

• 调用SDL_CreateCond函数创建一个条件变量，并将其赋值给PacketQueue结构体的cond成员。条件变量用于线程之间的同步，允许线程在特定条件下等待或被唤醒。
• 同样，如果SDL_CreateCond函数返回NULL，表示条件变量创建失败。记录错误信息并返回AVERROR(ENOMEM)，原因与创建互斥锁失败类似，可能是系统资源不足导致无法创建条件变量结构。

4. 设置初始状态

q->abort_request = 1; 

• 将PacketQueue结构体的abort_request成员初始化为 1，表示队列的初始状态为已请求终止。这个标志通常用于在需要停止队列操作时（如播放结束或发生错误时），通知其他操作队列的线程停止工作。

5. 返回成功

return 0; 

• 如果所有的初始化操作（结构体初始化、互斥锁创建、条件变量创建）都成功完成，函数返回 0，表示队列初始化成功。

注意事项

• 初始化PacketQueue结构体是使用数据包队列的前提条件，必须在向队列中添加或获取数据包之前成功调用此函数。
• 互斥锁和条件变量的正确创建和初始化对于多线程环境下队列的正常运行至关重要。如果创建失败，可能导致程序在后续操作中出现崩溃或不可预测的行为，因为多线程访问队列时将无法保证线程安全。
• abort_request标志的初始值为 1，意味着在初始化后，如果没有其他操作改变这个标志，队列将被视为已请求终止状态。在实际使用中，需要根据具体情况正确设置和检查这个标志，以确保队列操作的正确性和线程的安全退出。例如，在启动读取线程或其他操作队列的线程时，可能需要先将abort_request设置为 0，以允许队列正常工作，直到需要停止队列操作时再将其设置为 1。

static void packet_queue_flush(PacketQueue *q) { MyAVPacketList *pkt, *pkt1; SDL_LockMutex(q->mutex); for (pkt = q->first_pkt; pkt; pkt = pkt1) { pkt1 = pkt->next; av_packet_unref(&pkt->pkt); av_freep(&pkt); } q->last_pkt = NULL; q->first_pkt = NULL; q->nb_packets = 0; q->size = 0; q->duration = 0; SDL_UnlockMutex(q->mutex); }

packet_queue_flush函数用于清空PacketQueue队列中的所有数据包，释放相关资源，通常在需要重新填充队列、切换播放内容或停止播放等情况下使用，确保队列处于初始状态，避免数据残留或资源泄漏。

函数流程

1. 加锁队列

SDL_LockMutex(q->mutex); 

• 使用SDL_LockMutex函数对队列q的互斥锁进行加锁，确保在清空队列操作过程中，不会有其他线程同时访问和修改队列，保证线程安全。

2. 遍历并释放数据包

for (pkt = q->first_pkt; pkt; pkt = pkt1) { pkt1 = pkt->next; av_packet_unref(&pkt->pkt); av_freep(&pkt); } 

• 从队列的第一个数据包（q->first_pkt）开始，通过循环遍历整个链表结构的数据包队列。
• 在每次循环中，首先保存当前数据包的下一个数据包指针（pkt1 = pkt->next），以便在释放当前数据包后能够继续遍历下一个数据包。
• 调用av_packet_unref函数减少当前数据包pkt->pkt的引用计数，并释放其相关资源（如果引用计数为 0）。这一步确保了数据包内部的数据缓冲区等资源得到正确释放，避免内存泄漏。
• 调用av_freep函数释放当前数据包节点（pkt）所占用的内存空间，将其从链表中移除。

3. 重置队列状态

q->last_pkt = NULL; q->first_pkt = NULL; q->nb_packets = 0; q->size = 0; q->duration = 0; 

• 将队列的last_pkt和first_pkt指针都设置为NULL，表示队列为空。
• 将数据包数量（nb_packets）、总大小（size）和总时长（duration）都重置为 0，以反映队列的清空状态。

4. 解锁队列

SDL_UnlockMutex(q->mutex); 

• 使用SDL_UnlockMutex函数对队列q的互斥锁进行解锁，释放对队列的独占访问权限，允许其他线程访问队列（如果它们在等待锁的话）。

注意事项

• 清空队列操作会释放所有数据包占用的内存资源，因此在调用此函数之前，确保没有其他线程正在使用队列中的数据包，否则可能导致数据丢失或程序崩溃。
• 函数中对每个数据包先调用av_packet_unref再调用av_freep的顺序是必要的，因为av_packet_unref负责正确释放数据包内部的数据缓冲区等资源，而av_freep则释放数据包节点本身所占用的内存。如果顺序颠倒，可能会导致内存泄漏或其他资源管理问题。
• 在多线程环境下，packet_queue_flush函数应该与其他操作队列的函数（如packet_queue_put、packet_queue_get等）正确同步，以避免竞态条件和数据不一致性。例如，在调用packet_queue_flush之前，应该确保其他线程已经停止向队列中添加数据包，并且在清空队列后，其他线程应该能够正确检测到队列状态的变化并做出相应的处理。