流媒体数据未就绪时如何正确调用av_read_frame()？

问题描述 #

我在C++中使用FFmpeg/libav库处理音频流媒体数据，已实现CustomIOContext和read_packet回调函数向FFmpeg提供流数据，并基于该自定义IO上下文创建了AVFormatContext。

我的需求是循环提取所有可用数据包（用av_read_frame()），当剩余数据不足以生成完整数据包时，等待新数据到达后再继续循环处理。但目前无法确定read_packet中返回什么值才能正确告知FFmpeg等待更多数据：FFmpeg会把无数据的情况误判为文件结束；尝试返回AVERROR(EAGAIN)后，解码器会随机报错，再次调用av_read_frame()时会因状态异常直接返回错误码。

现有代码 #

初始化代码 #

// -------- Set up  --------
size_t ioContextBufferSize = 4096;
avioContextBuffer = avAllocateEmptyBuffer(ioContextBufferSize);
avioContext = avio_alloc_context(avioContextBuffer.ptr,
                         ioContextBufferSize,
                    0, // bWriteable (1=true,0=false)
                        &(opaque),
                        read_packet,
                        0, // Write callback function
                        0); // Seek function not provided

AVFormatContext* inContext = avformat_alloc_context();
inContext->pb = avioContext;
inContext->flags = AVFMT_FLAG_CUSTOM_IO;

avformat_open_input(&inContext, "", nullptr, nullptr);
inputPacket = av_packet_alloc();

读取循环代码 #

// -------- Loop and read data -------
// (the second time we reach this while loop, it never enters the loop :(
//
while ((ret = av_read_frame(inContext, inputPacket)) >= 0) {
  .. do stuff with inputPacket
}

read_packet回调函数 #

// -------- read_packet function -----
// copies data into ffmpeg's internal buffer. What do I return here???
//
static int read_packet(void* opaque, uint8_t* avioContextBuffer, int ioContextBufferSize) {
  OpaqueDataWrapper* streamedData = static_cast<OpaqueDataWrapper*>(opaque);

  size_t bytesToRead = std::min((size_t)ioContextBufferSize, streamedData->remainingBytes());

  if (!bytesToRead) {
    // We need to wait for more data to arrive! What do I return here to tell ffmpeg to wait a while? 
    if (streamedData->streamStillOpen()) {
      return AVERROR(EAGAIN);
    } else {
      return AVERROR_EOF;
    }
  }
  ... otherwise copy the next bit of data
}

解决方案 #

1. 正确处理AVERROR(EAGAIN)的核心逻辑 #

FFmpeg允许read_packet返回AVERROR(EAGAIN)表示当前无数据，但不能让FFmpeg反复进入错误状态。关键是：

• 当av_read_frame()返回AVERROR(EAGAIN)时，不能终止循环，而是要重置数据包、等待新数据后重试。
• read_packet内部如果要返回EAGAIN，必须保证后续有新数据到达后再被调用；更可靠的方式是在read_packet内部阻塞等待数据，直到有数据可读或流关闭。

2. 修改读取循环，兼容EAGAIN #

原来的循环只判断返回值>=0，需要扩展处理EAGAIN：

int ret;
while (true) {
    av_packet_unref(inputPacket); // 重置数据包，避免残留数据干扰
    ret = av_read_frame(inContext, inputPacket);
    
    if (ret >= 0) {
        // 处理数据包
        ... do stuff with inputPacket
    } else if (ret == AVERROR(EAGAIN)) {
        // 无数据，等待新数据到达（根据业务实现等待逻辑，比如条件变量/短休眠）
        std::this_thread::sleep_for(std::chrono::milliseconds(10));
        continue;
    } else if (ret == AVERROR_EOF) {
        // 流已关闭，退出循环
        break;
    } else {
        // 其他错误，按需处理或退出
        av_log(inContext, AV_LOG_ERROR, "Failed to read frame: %s\n", av_err2str(ret));
        break;
    }
}

3. 修复read_packet的阻塞/返回逻辑 #

推荐在read_packet内部阻塞等待数据，避免频繁返回EAGAIN导致FFmpeg状态异常。假设OpaqueDataWrapper支持条件变量等待：

static int read_packet(void* opaque, uint8_t* buf, int buf_size) {
    OpaqueDataWrapper* streamedData = static_cast<OpaqueDataWrapper*>(opaque);
    
    // 阻塞等待，直到有数据或流关闭
    while (streamedData->remainingBytes() == 0 && streamedData->streamStillOpen()) {
        std::unique_lock<std::mutex> lock(streamedData->mtx);
        streamedData->cv.wait(lock, [streamedData](){
            return streamedData->remainingBytes() > 0 || !streamedData->streamStillOpen();
        });
    }
    
    if (!streamedData->streamStillOpen() && streamedData->remainingBytes() == 0) {
        return AVERROR_EOF;
    }
    
    size_t bytesToRead = std::min(static_cast<size_t>(buf_size), streamedData->remainingBytes());
    // 复制数据到FFmpeg缓冲区
    memcpy(buf, streamedData->getDataPtr(), bytesToRead);
    streamedData->consumeData(bytesToRead); // 更新剩余数据
    
    return static_cast<int>(bytesToRead);
}

4. 补充初始化步骤的必要操作 #

调用avformat_open_input后，必须调用avformat_find_stream_info获取流信息，这一步也需要处理可能的EAGAIN：

// 初始化后添加
AVDictionary* options = nullptr;
av_dict_set(&options, "probesize", "4096", 0); // 根据流类型调整探针大小
av_dict_set(&options, "analyzeduration", "1000000", 0);
while (true) {
    ret = avformat_find_stream_info(inContext, &options);
    if (ret >= 0) {
        break;
    } else if (ret == AVERROR(EAGAIN)) {
        std::this_thread::sleep_for(std::chrono::milliseconds(10));
        continue;
    } else {
        av_log(inContext, AV_LOG_ERROR, "Failed to find stream info: %s\n", av_err2str(ret));
        // 处理错误，比如释放资源
        return -1;
    }
}
av_dict_free(&options);

5. 资源清理注意事项 #

• 确保avio_alloc_context的缓冲区由FFmpeg管理（不要自行释放），或设置avio_context->free_buffer = 1让FFmpeg自动释放。
• 退出时依次释放AVPacket、AVFormatContext、AVIOContext（按需）。

内容的提问来源于stack exchange，提问作者CheekyChips