video-pc/libs/AVPlayer2/read_thread.cpp

#include "read_thread.h"

#include "playercontroller.h"

#include <QLoggingCategory>
#include <QString>

Q_LOGGING_CATEGORY(playerControllerReadThread, "player.controller.ReadThread")

extern int infinite_buffer;
extern int64_t start_time;
static int64_t duration = AV_NOPTS_VALUE;

ReadThread::ReadThread(VideoState* pState)
    : m_pPlayData(pState)
{
    qCDebug(playerControllerReadThread) << "[ReadThread] constructed, pState:" << (void*) pState;
}

ReadThread::~ReadThread()
{
    qCDebug(playerControllerReadThread) << "[ReadThread] destructed.";
    stop();
    join();
}

void ReadThread::set_video_state(VideoState* pState)
{
    assert(pState);
    m_pPlayData = pState;
}

int ReadThread::loop_read()
{
    int ret = -1;
    VideoState* is = m_pPlayData;
    AVPacket* pkt = nullptr;
    int pkt_in_play_range = 0;
    int64_t stream_start_time = 0;
    int64_t pkt_ts = 0;
    
    // 重连计数器和配置
    static int reconnect_count = 0;
    static const int max_fast_reconnect_attempts = 3;  // 快速重连次数
    static const int max_normal_reconnect_attempts = 10; // 正常重连次数
    static bool infinite_reconnect_mode = false;       // 无限重连模式
    
    if (!is)
        return ret;

    pkt = av_packet_alloc();
    if (!pkt) {
        av_log(nullptr, AV_LOG_FATAL, "Could not allocate packet.\n");
        ret = AVERROR(ENOMEM);
        return ret;
    }
    qCDebug(playerControllerReadThread) << "[ReadThread] loop_read start, file:"
                                 << (is && is->ic ? is->ic->url : "null");

    is->read_thread_exit = 0;

    for (;;) {
        if (isExit()) {
            qCDebug(playerControllerReadThread) << "[ReadThread] m_exit set, break.";
            break;
        }

        if (is->abort_request) {
            qCDebug(playerControllerReadThread) << "[ReadThread] abort_request set, break.";
            break;
        }

        if (is->paused != is->last_paused) {
            is->last_paused = is->paused;
            if (is->paused)
                is->read_pause_return = av_read_pause(is->ic);
            else
                av_read_play(is->ic);
        }

        if (is->seek_req) {
            qCDebug(playerControllerReadThread)
                << "[ReadThread] seek_req, seek_pos:" << is->seek_pos
                << ", seek_rel:" << is->seek_rel << ", seek_flags:" << is->seek_flags;
            int64_t seek_target = is->seek_pos;
            int64_t seek_min = is->seek_rel > 0 ? seek_target - is->seek_rel + 2 : INT64_MIN;
            int64_t seek_max = is->seek_rel < 0 ? seek_target - is->seek_rel - 2 : INT64_MAX;

            ret = avformat_seek_file(is->ic, -1, seek_min, seek_target, seek_max, is->seek_flags);
            if (ret < 0) {
                av_log(nullptr, AV_LOG_ERROR, "%s: error while seeking\n", is->ic->url);
                qCWarning(playerControllerReadThread)
                    << "[ReadThread] avformat_seek_file failed, ret:" << ret;
            } else {
                qCDebug(playerControllerReadThread)
                    << "[ReadThread] avformat_seek_file success, flush queues.";
                if (is->audio_stream >= 0)
                    packet_queue_flush(&is->audioq);
                if (is->subtitle_stream >= 0)
                    packet_queue_flush(&is->subtitleq);
                if (is->video_stream >= 0)
                    packet_queue_flush(&is->videoq);
                if (is->seek_flags & AVSEEK_FLAG_BYTE) {
                    set_clock(&is->extclk, NAN, 0);
                } else {
                    set_clock(&is->extclk, seek_target / (double) AV_TIME_BASE, 0);
                }
            }
            is->seek_req = 0;
            is->queue_attachments_req = 1;
            is->eof = 0;
            if (is->paused)
                step_to_next_frame(is);
        }

        if (is->queue_attachments_req) {
            qCDebug(playerControllerReadThread)
                << "[ReadThread] queue_attachments_req, video_st:" << (void*) is->video_st;
            if (is->video_st && is->video_st->disposition & AV_DISPOSITION_ATTACHED_PIC) {
                ret = av_packet_ref(pkt, &is->video_st->attached_pic);
                if (ret < 0)
                    break;
                packet_queue_put(&is->videoq, pkt);
                packet_queue_put_nullpacket(&is->videoq, pkt, is->video_stream);
            }
            is->queue_attachments_req = 0;
        }

        /* if the queue are full, no need to read more */
        if (infinite_buffer < 1
            && (is->audioq.size + is->videoq.size + is->subtitleq.size > MAX_QUEUE_SIZE
                || (stream_has_enough_packets(is->audio_st, is->audio_stream, &is->audioq)
                    && stream_has_enough_packets(is->video_st, is->video_stream, &is->videoq)
                    && stream_has_enough_packets(is->subtitle_st,
                                                 is->subtitle_stream,
                                                 &is->subtitleq)))) {
            qCDebug(playerControllerReadThread)
                << "[ReadThread] queues full, waiting... audioq:" << is->audioq.size
                << ", videoq:" << is->videoq.size << ", subtitleq:" << is->subtitleq.size;
            std::unique_lock<std::mutex> lock(m_mutex);
            m_cv.wait_for(lock, std::chrono::milliseconds(10), [this, is] { return isExit() || is->abort_request; });
            continue;
        }

        if (!is->paused
            && (!is->audio_st
                || (is->auddec.finished == is->audioq.serial
                    && frame_queue_nb_remaining(&is->sampq) == 0))
            && (!is->video_st
                || (is->viddec.finished == is->videoq.serial
                    && frame_queue_nb_remaining(&is->pictq) == 0))) {
            if (is->loop) {
                stream_seek(is, 0, 0, 0);
            } else {
                // autoexit
                break;
            }
        }

        ret = av_read_frame(is->ic, pkt);
        if (ret < 0) {
            char buf[256] = {0};
            av_strerror(ret, buf, 256);

            qCWarning(playerControllerReadThread)
                << "[ReadThread] av_read_frame failed, ret:" << ret << ", eof:" << is->eof
                << ", pb_error:" << (is->ic->pb ? is->ic->pb->error : 0)
                << "error:" << QString::fromUtf8(buf);

            // 统一处理EOF和I/O错误 - 无感重连策略
            bool shouldAttemptReconnect = false;
            bool isEOFError = (ret == AVERROR_EOF || avio_feof(is->ic->pb)) && !is->eof;
            bool isIOError = is->ic->pb && is->ic->pb->error;
            
            // 增强错误检测：检查更多可能的网络错误
            bool isNetworkError = (ret == AVERROR(EAGAIN) || ret == AVERROR(ETIMEDOUT) || 
                                 ret == AVERROR(ECONNRESET) || ret == AVERROR(EPIPE) ||
                                 ret == AVERROR(ECONNABORTED) || ret == -5); // -5 是常见的I/O错误
            
            if (isEOFError || isIOError || isNetworkError) {
                if (isEOFError) {
                    qCDebug(playerControllerReadThread) << "[ReadThread] EOF reached for stream:" << is->filename;
                }
                if (isIOError) {
                    qCWarning(playerControllerReadThread) << "[ReadThread] IO error detected, error code:" << is->ic->pb->error;
                }
                if (isNetworkError) {
                    qCWarning(playerControllerReadThread) << "[ReadThread] Network error detected, return code:" << ret;
                }
                
                // 检查是否为实时流（RTMP、RTSP等）
                bool isRealTimeStream = false;
                if (is->filename) {
                    QString filename = QString::fromUtf8(is->filename);
                    isRealTimeStream = filename.startsWith("rtmp://") || 
                                     filename.startsWith("rtsp://") || 
                                     filename.startsWith("rtp://") ||
                                     filename.startsWith("udp://") ||
                                     filename.startsWith("tcp://") ||
                                     filename.startsWith("http://") ||
                                     filename.startsWith("https://");
                }
                
                if (isRealTimeStream) {
                    reconnect_count++;
                    QString errorType = isEOFError ? "EOF" : (isIOError ? "I/O error" : "Network error");
                    
                    // 分层重连策略
                    bool shouldContinueReconnect = false;
                    int wait_time_ms = 0;
                    
                    if (reconnect_count <= max_fast_reconnect_attempts) {
                        // 快速重连阶段：短间隔，快速恢复
                        wait_time_ms = 500 * reconnect_count; // 0.5s, 1s, 1.5s
                        shouldContinueReconnect = true;
                        qCWarning(playerControllerReadThread) << "[ReadThread] Fast reconnection" << errorType 
                                                             << "attempt" << reconnect_count << "/" << max_fast_reconnect_attempts;
                    } else if (reconnect_count <= max_normal_reconnect_attempts) {
                        // 正常重连阶段：较长间隔
                        wait_time_ms = 2000 + (reconnect_count - max_fast_reconnect_attempts) * 1000; // 2s, 3s, 4s...
                        shouldContinueReconnect = true;
                        qCWarning(playerControllerReadThread) << "[ReadThread] Normal reconnection" << errorType 
                                                             << "attempt" << reconnect_count << "/" << max_normal_reconnect_attempts;
                    } else {
                        // 无限重连模式：长间隔，但永不放弃
                        infinite_reconnect_mode = true;
                        wait_time_ms = 10000; // 固定10秒间隔
                        shouldContinueReconnect = true;
                        qCWarning(playerControllerReadThread) << "[ReadThread] Infinite reconnection mode" << errorType 
                                                             << "attempt" << reconnect_count << "(never give up)";
                    }
                    
                    if (shouldContinueReconnect) {
                         // 清空当前的错误状态
                         if (is->ic && is->ic->pb) {
                             is->ic->pb->error = 0;
                             is->ic->pb->eof_reached = 0;
                         }
                         
                         // 无感重连：不设置EOF状态，保持播放器运行
                         // 这样即使网络中断，播放器界面也不会退出
                         qCDebug(playerControllerReadThread) << "[ReadThread] Seamless reconnection: waiting" << wait_time_ms << "ms...";
                         
                         std::unique_lock<std::mutex> lock(m_mutex);
                         m_cv.wait_for(lock, std::chrono::milliseconds(wait_time_ms), [this, is] { return isExit() || is->abort_request; });
                         
                         if (!isExit() && !is->abort_request) {
                             qCDebug(playerControllerReadThread) << "[ReadThread] Attempting seamless reconnection...";
                             continue; // 继续循环，永不放弃，真正的无感重连
                         }
                     }
                } else {
                    // 非实时流：传统处理方式
                    qCDebug(playerControllerReadThread) << "[ReadThread] File stream error, send null packets.";
                    
                    // 发送null packets并设置EOF
                    if (isEOFError) {
                        if (is->video_stream >= 0)
                            packet_queue_put_nullpacket(&is->videoq, pkt, is->video_stream);
                        if (is->audio_stream >= 0)
                            packet_queue_put_nullpacket(&is->audioq, pkt, is->audio_stream);
                        if (is->subtitle_stream >= 0)
                            packet_queue_put_nullpacket(&is->subtitleq, pkt, is->subtitle_stream);
                        is->eof = 1;
                    }
                    
                    // 对于I/O错误，直接退出循环
                    if (isIOError) {
                        break;
                    }
                }
            }
            // 错误处理已经统一到上面的逻辑中
            std::unique_lock<std::mutex> lock(m_mutex);
            m_cv.wait_for(lock, std::chrono::milliseconds(10), [this, is] { return isExit() || is->abort_request; });
            continue;
        } else {
            is->eof = 0;
            
            // 成功读取数据时重置重连计数器和状态
            if (reconnect_count > 0) {
                if (infinite_reconnect_mode) {
                    qCInfo(playerControllerReadThread) << "[ReadThread] Stream recovered from infinite reconnection mode after" 
                                                       << reconnect_count << "attempts";
                    infinite_reconnect_mode = false;
                } else {
                    qCDebug(playerControllerReadThread) << "[ReadThread] Stream recovered, resetting reconnection counter";
                }
                reconnect_count = 0;
            }
            
            qCDebug(playerControllerReadThread)
                << "[ReadThread] av_read_frame success, stream_index:" << pkt->stream_index
                << ", pts:" << pkt->pts;
        }

        /* check if packet is in play range specified by user, then queue, otherwise discard */
        stream_start_time = is->ic->streams[pkt->stream_index]->start_time;
        pkt_ts = pkt->pts == AV_NOPTS_VALUE ? pkt->dts : pkt->pts;
        pkt_in_play_range = duration == AV_NOPTS_VALUE
                            || (pkt_ts
                                - (stream_start_time != AV_NOPTS_VALUE ? stream_start_time : 0))
                                           * av_q2d(is->ic->streams[pkt->stream_index]->time_base)
                                       - (double) (start_time != AV_NOPTS_VALUE ? start_time : 0)
                                             / 1000000
                                   <= ((double) duration / 1000000);
        if (pkt->stream_index == is->audio_stream && pkt_in_play_range) {
            qCDebug(playerControllerReadThread) << "[ReadThread] put audio packet, pts:" << pkt->pts;
            packet_queue_put(&is->audioq, pkt);
        } else if (pkt->stream_index == is->video_stream && pkt_in_play_range
                   && !(is->video_st->disposition & AV_DISPOSITION_ATTACHED_PIC)) {
            qCDebug(playerControllerReadThread) << "[ReadThread] put video packet, pts:" << pkt->pts;
            packet_queue_put(&is->videoq, pkt);
        } else if (pkt->stream_index == is->subtitle_stream && pkt_in_play_range) {
            qCDebug(playerControllerReadThread) << "[ReadThread] put subtitle packet, pts:" << pkt->pts;
            packet_queue_put(&is->subtitleq, pkt);
        } else {
            qCDebug(playerControllerReadThread)
                << "[ReadThread] drop packet, stream_index:" << pkt->stream_index;
            av_packet_unref(pkt);
        }
    }
    qCWarning(playerControllerReadThread)
        << "[ReadThread] loop_read about to exit, eof:" << is->eof
        << ", audioq.size:" << is->audioq.size << ", videoq.size:" << is->videoq.size
        << ", subtitleq.size:" << is->subtitleq.size << ", abort_request:" << is->abort_request
        << ", m_exit:" << (m_exit ? m_exit->load() : -1);
    qCDebug(playerControllerReadThread) << "[ReadThread] loop_read exit, set read_thread_exit = -1";
    is->read_thread_exit = -1;

    if (!is->ic)
        avformat_close_input(&is->ic);
    av_packet_free(&pkt);
    return 0;
}

void ReadThread::run()
{
    qCDebug(playerControllerReadThread) << "[ReadThread] run start";
    int ret = loop_read();
    qCDebug(playerControllerReadThread) << "[ReadThread] run end, loop_read ret:" << ret;
    // 可加日志输出
}