ffmpeg中av_read_frame 超时设置

朱雀 2023-01-14 13:57 116阅读 0赞

https://trac.ffmpeg.org/  
ffmpeg wiki

ffmpeg中avformat\_open\_input超时设置  
这里有日志可以参考：[日志][Link 1]  
y也可以参考：  
https://blog.csdn.net/qpx0033/article/details/51290535

接着上篇文章，我们来看看av\_read\_frame，这个雷神有一篇文章：  
https://blog.csdn.net/leixiaohua1020/article/details/12678577  
https://www.freesion.com/article/6303637278/  
这个比较有价值  
https://blog.csdn.net/finewind/article/details/39433055

ffmpeg官网  
https://www.ffmpeg.org/doxygen/trunk/avio\_8h.html\#aea3131c92c70177e463538e9a6b4308e

av_read_frame（）-->read_frame_internal(s, pkt)-->
    
    
    
    
    
    int av_read_frame(AVFormatContext *s, AVPacket *pkt)
    { 
    //正常情况下这个数字都为0
        const int genpts = s->flags & AVFMT_FLAG_GENPTS;
        int eof = 0;
        int ret;
        AVStream *st;
    
        if (!genpts) { 
        //如果packet_buffer中还有数据，就直接进入其中去取数据，这个
        //函数单纯的取数据
            ret = s->internal->packet_buffer
                  ? ff_packet_list_get(&s->internal->packet_buffer,
                                            &s->internal->packet_buffer_end, pkt)
                  : read_frame_internal(s, pkt);//一般都是这里，然后goto
            if (ret < 0)
                return ret;
            goto return_packet;
        }
    
        for (;;) { 
            AVPacketList *pktl = s->internal->packet_buffer;
    
            if (pktl) { 
                AVPacket *next_pkt = &pktl->pkt;
    
                if (next_pkt->dts != AV_NOPTS_VALUE) { 
                    int wrap_bits = s->streams[next_pkt->stream_index]->pts_wrap_bits;
                    // last dts seen for this stream. if any of packets following
                    // current one had no dts, we will set this to AV_NOPTS_VALUE.
                    int64_t last_dts = next_pkt->dts;
                    av_assert2(wrap_bits <= 64);
                    while (pktl && next_pkt->pts == AV_NOPTS_VALUE) { 
                        if (pktl->pkt.stream_index == next_pkt->stream_index &&
                            av_compare_mod(next_pkt->dts, pktl->pkt.dts, 2ULL << (wrap_bits - 1)) < 0) { 
                            if (av_compare_mod(pktl->pkt.pts, pktl->pkt.dts, 2ULL << (wrap_bits - 1))) { 
                                // not B-frame
                                next_pkt->pts = pktl->pkt.dts;
                            }
                            if (last_dts != AV_NOPTS_VALUE) { 
                                // Once last dts was set to AV_NOPTS_VALUE, we don't change it.
                                last_dts = pktl->pkt.dts;
                            }
                        }
                        pktl = pktl->next;
                    }
                    if (eof && next_pkt->pts == AV_NOPTS_VALUE && last_dts != AV_NOPTS_VALUE) { 
                        // Fixing the last reference frame had none pts issue (For MXF etc).
                        // We only do this when
                        // 1. eof.
                        // 2. we are not able to resolve a pts value for current packet.
                        // 3. the packets for this stream at the end of the files had valid dts.
                        next_pkt->pts = last_dts + next_pkt->duration;
                    }
                    pktl = s->internal->packet_buffer;
                }
    
                /* read packet from packet buffer, if there is data */
                st = s->streams[next_pkt->stream_index];
                if (!(next_pkt->pts == AV_NOPTS_VALUE && st->discard < AVDISCARD_ALL &&
                      next_pkt->dts != AV_NOPTS_VALUE && !eof)) { 
                    ret = ff_packet_list_get(&s->internal->packet_buffer,
                                                   &s->internal->packet_buffer_end, pkt);
                    goto return_packet;
                }
            }
    //在这里执行下一步
            ret = read_frame_internal(s, pkt);
            if (ret < 0) { 
                if (pktl && ret != AVERROR(EAGAIN)) { 
                    eof = 1;
                    continue;
                } else
                    return ret;
            }
    
            ret = ff_packet_list_put(&s->internal->packet_buffer,
                                     &s->internal->packet_buffer_end,
                                     pkt, 0);
            if (ret < 0) { 
                av_packet_unref(pkt);
                return ret;
            }
        }
    
    return_packet:
    
        st = s->streams[pkt->stream_index];
        if ((s->iformat->flags & AVFMT_GENERIC_INDEX) && pkt->flags & AV_PKT_FLAG_KEY) { 
            ff_reduce_index(s, st->index);
            av_add_index_entry(st, pkt->pos, pkt->dts, 0, 0, AVINDEX_KEYFRAME);
        }
    
        if (is_relative(pkt->dts))
            pkt->dts -= RELATIVE_TS_BASE;
        if (is_relative(pkt->pts))
            pkt->pts -= RELATIVE_TS_BASE;
    
        return ret;
    }
    
    
    static int read_frame_internal(AVFormatContext *s, AVPacket *pkt)
    { 
        int ret, i, got_packet = 0;
        AVDictionary *metadata = NULL;
    //while循环，好多次才能从中获取到一个完整的帧，所以当你设置了超时比如200ms,但是实际超过200就是这个原因。
    //一般两次或者三次就能读完整的一帧
        while (!got_packet && !s->internal->parse_queue) { 
            AVStream *st;
    
            /* read next packet */
            //在这里执行读包
            ret = ff_read_packet(s, pkt);
            if (ret < 0) { 
                if (ret == AVERROR(EAGAIN))
                    return ret;
                /* flush the parsers */
                for (i = 0; i < s->nb_streams; i++) { 
                    st = s->streams[i];
                    if (st->parser && st->need_parsing)
                    //如果出错，将所有包全部发出去
                        parse_packet(s, pkt, st->index, 1);
                }
                /* all remaining packets are now in parse_queue => * really terminate parsing */
                break;
            }
            ret = 0;
            st  = s->streams[pkt->stream_index];
    
            /* update context if required */
            if (st->internal->need_context_update) { 
                if (avcodec_is_open(st->internal->avctx)) { 
                    av_log(s, AV_LOG_DEBUG, "Demuxer context update while decoder is open, closing and trying to re-open\n");
                    avcodec_close(st->internal->avctx);
                    st->info->found_decoder = 0;
                }
    
                /* close parser, because it depends on the codec */
                if (st->parser && st->internal->avctx->codec_id != st->codecpar->codec_id) { 
                    av_parser_close(st->parser);
                    st->parser = NULL;
                }
    
                ret = avcodec_parameters_to_context(st->internal->avctx, st->codecpar);
                if (ret < 0) { 
                    av_packet_unref(pkt);
                    return ret;
                }
    
    #if FF_API_LAVF_AVCTX
    FF_DISABLE_DEPRECATION_WARNINGS
                /* update deprecated public codec context */
                ret = avcodec_parameters_to_context(st->codec, st->codecpar);
                if (ret < 0) { 
                    av_packet_unref(pkt);
                    return ret;
                }
    FF_ENABLE_DEPRECATION_WARNINGS
    #endif
    
                st->internal->need_context_update = 0;
            }
    
            if (pkt->pts != AV_NOPTS_VALUE &&
                pkt->dts != AV_NOPTS_VALUE &&
                pkt->pts < pkt->dts) { 
                av_log(s, AV_LOG_WARNING,
                       "Invalid timestamps stream=%d, pts=%s, dts=%s, size=%d\n",
                       pkt->stream_index,
                       av_ts2str(pkt->pts),
                       av_ts2str(pkt->dts),
                       pkt->size);
            }
            if (s->debug & FF_FDEBUG_TS)
                av_log(s, AV_LOG_DEBUG,
                       "ff_read_packet stream=%d, pts=%s, dts=%s, size=%d, duration=%"PRId64", flags=%d\n",
                       pkt->stream_index,
                       av_ts2str(pkt->pts),
                       av_ts2str(pkt->dts),
                       pkt->size, pkt->duration, pkt->flags);
    
            if (st->need_parsing && !st->parser && !(s->flags & AVFMT_FLAG_NOPARSE)) { 
                st->parser = av_parser_init(st->codecpar->codec_id);
                if (!st->parser) { 
                    av_log(s, AV_LOG_VERBOSE, "parser not found for codec "
                           "%s, packets or times may be invalid.\n",
                           avcodec_get_name(st->codecpar->codec_id));
                    /* no parser available: just output the raw packets */
                    st->need_parsing = AVSTREAM_PARSE_NONE;
                } else if (st->need_parsing == AVSTREAM_PARSE_HEADERS)
                    st->parser->flags |= PARSER_FLAG_COMPLETE_FRAMES;
                else if (st->need_parsing == AVSTREAM_PARSE_FULL_ONCE)
                    st->parser->flags |= PARSER_FLAG_ONCE;
                else if (st->need_parsing == AVSTREAM_PARSE_FULL_RAW)
                    st->parser->flags |= PARSER_FLAG_USE_CODEC_TS;
            }
    
            if (!st->need_parsing || !st->parser) { 
                /* no parsing needed: we just output the packet as is */
                compute_pkt_fields(s, st, NULL, pkt, AV_NOPTS_VALUE, AV_NOPTS_VALUE);
                if ((s->iformat->flags & AVFMT_GENERIC_INDEX) &&
                    (pkt->flags & AV_PKT_FLAG_KEY) && pkt->dts != AV_NOPTS_VALUE) { 
                    ff_reduce_index(s, st->index);
                    av_add_index_entry(st, pkt->pos, pkt->dts,
                                       0, 0, AVINDEX_KEYFRAME);
                }
                got_packet = 1;
            } else if (st->discard < AVDISCARD_ALL) { 
                if ((ret = parse_packet(s, pkt, pkt->stream_index, 0)) < 0)
                    return ret;
                st->codecpar->sample_rate = st->internal->avctx->sample_rate;
                st->codecpar->bit_rate = st->internal->avctx->bit_rate;
                st->codecpar->channels = st->internal->avctx->channels;
                st->codecpar->channel_layout = st->internal->avctx->channel_layout;
                st->codecpar->codec_id = st->internal->avctx->codec_id;
            } else { 
                /* free packet */
                av_packet_unref(pkt);
            }
            if (pkt->flags & AV_PKT_FLAG_KEY)
                st->skip_to_keyframe = 0;
            if (st->skip_to_keyframe) { 
                av_packet_unref(pkt);
                got_packet = 0;
            }
        }
    
        if (!got_packet && s->internal->parse_queue)
            ret = ff_packet_list_get(&s->internal->parse_queue, &s->internal->parse_queue_end, pkt);
    
        if (ret >= 0) { 
            AVStream *st = s->streams[pkt->stream_index];
            int discard_padding = 0;
            if (st->first_discard_sample && pkt->pts != AV_NOPTS_VALUE) { 
                int64_t pts = pkt->pts - (is_relative(pkt->pts) ? RELATIVE_TS_BASE : 0);
                int64_t sample = ts_to_samples(st, pts);
                int duration = ts_to_samples(st, pkt->duration);
                int64_t end_sample = sample + duration;
                if (duration > 0 && end_sample >= st->first_discard_sample &&
                    sample < st->last_discard_sample)
                    discard_padding = FFMIN(end_sample - st->first_discard_sample, duration);
            }
            if (st->start_skip_samples && (pkt->pts == 0 || pkt->pts == RELATIVE_TS_BASE))
                st->skip_samples = st->start_skip_samples;
            if (st->skip_samples || discard_padding) { 
                uint8_t *p = av_packet_new_side_data(pkt, AV_PKT_DATA_SKIP_SAMPLES, 10);
                if (p) { 
                    AV_WL32(p, st->skip_samples);
                    AV_WL32(p + 4, discard_padding);
                    av_log(s, AV_LOG_DEBUG, "demuxer injecting skip %d / discard %d\n", st->skip_samples, discard_padding);
                }
                st->skip_samples = 0;
            }
    
            if (st->inject_global_side_data) { 
                for (i = 0; i < st->nb_side_data; i++) { 
                    AVPacketSideData *src_sd = &st->side_data[i];
                    uint8_t *dst_data;
    
                    if (av_packet_get_side_data(pkt, src_sd->type, NULL))
                        continue;
    
                    dst_data = av_packet_new_side_data(pkt, src_sd->type, src_sd->size);
                    if (!dst_data) { 
                        av_log(s, AV_LOG_WARNING, "Could not inject global side data\n");
                        continue;
                    }
    
                    memcpy(dst_data, src_sd->data, src_sd->size);
                }
                st->inject_global_side_data = 0;
            }
        }
    
        av_opt_get_dict_val(s, "metadata", AV_OPT_SEARCH_CHILDREN, &metadata);
        if (metadata) { 
            s->event_flags |= AVFMT_EVENT_FLAG_METADATA_UPDATED;
            av_dict_copy(&s->metadata, metadata, 0);
            av_dict_free(&metadata);
            av_opt_set_dict_val(s, "metadata", NULL, AV_OPT_SEARCH_CHILDREN);
        }
    
    #if FF_API_LAVF_AVCTX
        update_stream_avctx(s);
    #endif
    
        if (s->debug & FF_FDEBUG_TS)
            av_log(s, AV_LOG_DEBUG,
                   "read_frame_internal stream=%d, pts=%s, dts=%s, "
                   "size=%d, duration=%"PRId64", flags=%d\n",
                   pkt->stream_index,
                   av_ts2str(pkt->pts),
                   av_ts2str(pkt->dts),
                   pkt->size, pkt->duration, pkt->flags);
    
        /* A demuxer might have returned EOF because of an IO error, let's * propagate this back to the user. */
        if (ret == AVERROR_EOF && s->pb && s->pb->error < 0 && s->pb->error != AVERROR(EAGAIN))
            ret = s->pb->error;
    
        return ret;
    }
    
    
    
    int ff_read_packet(AVFormatContext *s, AVPacket *pkt)
    { 
        int ret, i, err;
        AVStream *st;
    
        pkt->data = NULL;
        pkt->size = 0;
        av_init_packet(pkt);
    
        for (;;) { 
            AVPacketList *pktl = s->internal->raw_packet_buffer;
            const AVPacket *pkt1;
    
            if (pktl) { 
                st = s->streams[pktl->pkt.stream_index];
                if (s->internal->raw_packet_buffer_remaining_size <= 0)
                    if ((err = probe_codec(s, st, NULL)) < 0)
                        return err;
                if (st->request_probe <= 0) { 
                    ff_packet_list_get(&s->internal->raw_packet_buffer,
                                       &s->internal->raw_packet_buffer_end, pkt);
                    s->internal->raw_packet_buffer_remaining_size += pkt->size;
                    return 0;
                }
            }
            //在这里执行rtsp_demux的read_packet,这个函数是个static函数
            ret = s->iformat->read_packet(s, pkt);
            if (ret < 0) { 
                av_packet_unref(pkt);
    
                /* Some demuxers return FFERROR_REDO when they consume data and discard it (ignored streams, junk, extradata). We must re-call the demuxer to get the real packet. */
                if (ret == FFERROR_REDO)
                    continue;
                if (!pktl || ret == AVERROR(EAGAIN))
                    return ret;
                for (i = 0; i < s->nb_streams; i++) { 
                    st = s->streams[i];
                    if (st->probe_packets || st->request_probe > 0)
                        if ((err = probe_codec(s, st, NULL)) < 0)
                            return err;
                    av_assert0(st->request_probe <= 0);
                }
                //注意这里的continue
                continue;
            }
    
            err = av_packet_make_refcounted(pkt);
            if (err < 0) { 
                av_packet_unref(pkt);
                return err;
            }
    
            if (pkt->flags & AV_PKT_FLAG_CORRUPT) { 
                av_log(s, AV_LOG_WARNING,
                       "Packet corrupt (stream = %d, dts = %s)",
                       pkt->stream_index, av_ts2str(pkt->dts));
                if (s->flags & AVFMT_FLAG_DISCARD_CORRUPT) { 
                    av_log(s, AV_LOG_WARNING, ", dropping it.\n");
                    av_packet_unref(pkt);
                    continue;
                }
                av_log(s, AV_LOG_WARNING, ".\n");
            }
    
            av_assert0(pkt->stream_index < (unsigned)s->nb_streams &&
                       "Invalid stream index.\n");
    
            st = s->streams[pkt->stream_index];
    
            if (update_wrap_reference(s, st, pkt->stream_index, pkt) && st->pts_wrap_behavior == AV_PTS_WRAP_SUB_OFFSET) { 
                // correct first time stamps to negative values
                if (!is_relative(st->first_dts))
                    st->first_dts = wrap_timestamp(st, st->first_dts);
                if (!is_relative(st->start_time))
                    st->start_time = wrap_timestamp(st, st->start_time);
                if (!is_relative(st->cur_dts))
                    st->cur_dts = wrap_timestamp(st, st->cur_dts);
            }
    
            pkt->dts = wrap_timestamp(st, pkt->dts);
            pkt->pts = wrap_timestamp(st, pkt->pts);
    
            force_codec_ids(s, st);
    
            /* TODO: audio: time filter; video: frame reordering (pts != dts) */
            if (s->use_wallclock_as_timestamps)
                pkt->dts = pkt->pts = av_rescale_q(av_gettime(), AV_TIME_BASE_Q, st->time_base);
    
            if (!pktl && st->request_probe <= 0)
                return ret;
    
            err = ff_packet_list_put(&s->internal->raw_packet_buffer,
                                     &s->internal->raw_packet_buffer_end,
                                     pkt, 0);
            if (err < 0) { 
                av_packet_unref(pkt);
                return err;
            }
            pkt1 = &s->internal->raw_packet_buffer_end->pkt;
            s->internal->raw_packet_buffer_remaining_size -= pkt1->size;
    
            if ((err = probe_codec(s, st, pkt1)) < 0)
                return err;
        }
    }
    
    
    
    AVInputFormat ff_rtsp_demuxer = { 
        .name           = "rtsp",
        .long_name      = NULL_IF_CONFIG_SMALL("RTSP input"),
        .priv_data_size = sizeof(RTSPState),
        .read_probe     = rtsp_probe,
        .read_header    = rtsp_read_header,
        .read_packet    = rtsp_read_packet,
        .read_close     = rtsp_read_close,
        .read_seek      = rtsp_read_seek,
        .flags          = AVFMT_NOFILE,
        .read_play      = rtsp_read_play,
        .read_pause     = rtsp_read_pause,
        .priv_class     = &rtsp_demuxer_class,
    };
    
    
    
    static int rtsp_read_packet(AVFormatContext *s, AVPacket *pkt)
    { 
        RTSPState *rt = s->priv_data;
        int ret;
        RTSPMessageHeader reply1, *reply = &reply1;
        char cmd[1024];
    
    retry:
        if (rt->server_type == RTSP_SERVER_REAL) { 
            int i;
    
            for (i = 0; i < s->nb_streams; i++)
                rt->real_setup[i] = s->streams[i]->discard;
    
            if (!rt->need_subscription) { 
                if (memcmp (rt->real_setup, rt->real_setup_cache,
                            sizeof(enum AVDiscard) * s->nb_streams)) { 
                    snprintf(cmd, sizeof(cmd),
                             "Unsubscribe: %s\r\n",
                             rt->last_subscription);
                    ff_rtsp_send_cmd(s, "SET_PARAMETER", rt->control_uri,
                                     cmd, reply, NULL);
                    if (reply->status_code != RTSP_STATUS_OK)
                        return ff_rtsp_averror(reply->status_code, AVERROR_INVALIDDATA);
                    rt->need_subscription = 1;
                }
            }
    
            if (rt->need_subscription) { 
                int r, rule_nr, first = 1;
    
                memcpy(rt->real_setup_cache, rt->real_setup,
                       sizeof(enum AVDiscard) * s->nb_streams);
                rt->last_subscription[0] = 0;
    
                snprintf(cmd, sizeof(cmd),
                         "Subscribe: ");
                for (i = 0; i < rt->nb_rtsp_streams; i++) { 
                    rule_nr = 0;
                    for (r = 0; r < s->nb_streams; r++) { 
                        if (s->streams[r]->id == i) { 
                            if (s->streams[r]->discard != AVDISCARD_ALL) { 
                                if (!first)
                                    av_strlcat(rt->last_subscription, ",",
                                               sizeof(rt->last_subscription));
                                ff_rdt_subscribe_rule(
                                    rt->last_subscription,
                                    sizeof(rt->last_subscription), i, rule_nr);
                                first = 0;
                            }
                            rule_nr++;
                        }
                    }
                }
                av_strlcatf(cmd, sizeof(cmd), "%s\r\n", rt->last_subscription);
                ff_rtsp_send_cmd(s, "SET_PARAMETER", rt->control_uri,
                                 cmd, reply, NULL);
                if (reply->status_code != RTSP_STATUS_OK)
                    return ff_rtsp_averror(reply->status_code, AVERROR_INVALIDDATA);
                rt->need_subscription = 0;
    
                if (rt->state == RTSP_STATE_STREAMING)
                    rtsp_read_play (s);
            }
        }
    //在这里获取新的packet
        ret = ff_rtsp_fetch_packet(s, pkt);
        if (ret < 0) { 
            if (ret == AVERROR(ETIMEDOUT) && !rt->packets) { 
                if (rt->lower_transport == RTSP_LOWER_TRANSPORT_UDP &&
                    rt->lower_transport_mask & (1 << RTSP_LOWER_TRANSPORT_TCP)) { 
                    RTSPMessageHeader reply1, *reply = &reply1;
                    av_log(s, AV_LOG_WARNING, "UDP timeout, retrying with TCP\n");
                    if (rtsp_read_pause(s) != 0)
                        return -1;
                    // TEARDOWN is required on Real-RTSP, but might make
                    // other servers close the connection.
                    if (rt->server_type == RTSP_SERVER_REAL)
                        ff_rtsp_send_cmd(s, "TEARDOWN", rt->control_uri, NULL,
                                         reply, NULL);
                    rt->session_id[0] = '\0';
                    if (resetup_tcp(s) == 0) { 
                        rt->state = RTSP_STATE_IDLE;
                        rt->need_subscription = 1;
                        if (rtsp_read_play(s) != 0)
                            return -1;
                        goto retry;
                    }
                }
            }
            return ret;
        }
        rt->packets++;
    
        if (!(rt->rtsp_flags & RTSP_FLAG_LISTEN)) { 
            /* send dummy request to keep TCP connection alive */
            if ((av_gettime_relative() - rt->last_cmd_time) / 1000000 >= rt->timeout / 2 ||
                rt->auth_state.stale) { 
                if (rt->server_type == RTSP_SERVER_WMS ||
                    (rt->server_type != RTSP_SERVER_REAL &&
                     rt->get_parameter_supported)) { 
                    ff_rtsp_send_cmd_async(s, "GET_PARAMETER", rt->control_uri, NULL);
                } else { 
                    ff_rtsp_send_cmd_async(s, "OPTIONS", rt->control_uri, NULL);
                }
                /* The stale flag should be reset when creating the auth response in * ff_rtsp_send_cmd_async, but reset it here just in case we never * called the auth code (if we didn't have any credentials set). */
                rt->auth_state.stale = 0;
            }
        }
    
        return 0;
    }
    
    
    int ff_rtsp_fetch_packet(AVFormatContext *s, AVPacket *pkt)
    { 
    //我们在av_open_input的时候不是说s->priv_data是
        RTSPState *rt = s->priv_data;
        int ret, len;
        RTSPStream *rtsp_st, *first_queue_st = NULL;
        int64_t wait_end = 0;
    
        if (rt->nb_byes == rt->nb_rtsp_streams)
            return AVERROR_EOF;
    
        /* get next frames from the same RTP packet */
        if (rt->cur_transport_priv) { 
            if (rt->transport == RTSP_TRANSPORT_RDT) { 
                ret = ff_rdt_parse_packet(rt->cur_transport_priv, pkt, NULL, 0);
            } else if (rt->transport == RTSP_TRANSPORT_RTP) { 
                ret = ff_rtp_parse_packet(rt->cur_transport_priv, pkt, NULL, 0);
            } else if (CONFIG_RTPDEC && rt->ts) { 
                ret = avpriv_mpegts_parse_packet(rt->ts, pkt, rt->recvbuf + rt->recvbuf_pos, rt->recvbuf_len - rt->recvbuf_pos);
                if (ret >= 0) { 
                    rt->recvbuf_pos += ret;
                    ret = rt->recvbuf_pos < rt->recvbuf_len;
                }
            } else
                ret = -1;
            if (ret == 0) { 
                rt->cur_transport_priv = NULL;
                return 0;
            } else if (ret == 1) { 
                return 0;
            } else
                rt->cur_transport_priv = NULL;
        }
    
    redo:
        if (rt->transport == RTSP_TRANSPORT_RTP) { 
            int i;
            int64_t first_queue_time = 0;
            for (i = 0; i < rt->nb_rtsp_streams; i++) { 
                RTPDemuxContext *rtpctx = rt->rtsp_streams[i]->transport_priv;
                int64_t queue_time;
                if (!rtpctx)
                    continue;
                queue_time = ff_rtp_queued_packet_time(rtpctx);
                if (queue_time && (queue_time - first_queue_time < 0 ||
                                   !first_queue_time)) { 
                    first_queue_time = queue_time;
                    first_queue_st   = rt->rtsp_streams[i];
                }
            }
            if (first_queue_time) { 
                wait_end = first_queue_time + s->max_delay;
            } else { 
                wait_end = 0;
                first_queue_st = NULL;
            }
        }
    
        /* read next RTP packet */
        if (!rt->recvbuf) { 
            rt->recvbuf = av_malloc(RECVBUF_SIZE);
            if (!rt->recvbuf)
                return AVERROR(ENOMEM);
        }
    //在这里度packet
        len = read_packet(s, &rtsp_st, first_queue_st, wait_end);
        if (len == AVERROR(EAGAIN) && first_queue_st &&
            rt->transport == RTSP_TRANSPORT_RTP) { 
            av_log(s, AV_LOG_WARNING,
                    "max delay reached. need to consume packet\n");
            rtsp_st = first_queue_st;
            ret = ff_rtp_parse_packet(rtsp_st->transport_priv, pkt, NULL, 0);
            goto end;
        }
        if (len < 0)
            return len;
    
        if (rt->transport == RTSP_TRANSPORT_RDT) { 
            ret = ff_rdt_parse_packet(rtsp_st->transport_priv, pkt, &rt->recvbuf, len);
        } else if (rt->transport == RTSP_TRANSPORT_RTP) { 
            ret = ff_rtp_parse_packet(rtsp_st->transport_priv, pkt, &rt->recvbuf, len);
            if (rtsp_st->feedback) { 
                AVIOContext *pb = NULL;
                if (rt->lower_transport == RTSP_LOWER_TRANSPORT_CUSTOM)
                    pb = s->pb;
                ff_rtp_send_rtcp_feedback(rtsp_st->transport_priv, rtsp_st->rtp_handle, pb);
            }
            if (ret < 0) { 
                /* Either bad packet, or a RTCP packet. Check if the * first_rtcp_ntp_time field was initialized. */
                RTPDemuxContext *rtpctx = rtsp_st->transport_priv;
                if (rtpctx->first_rtcp_ntp_time != AV_NOPTS_VALUE) { 
                    /* first_rtcp_ntp_time has been initialized for this stream, * copy the same value to all other uninitialized streams, * in order to map their timestamp origin to the same ntp time * as this one. */
                    int i;
                    AVStream *st = NULL;
                    if (rtsp_st->stream_index >= 0)
                        st = s->streams[rtsp_st->stream_index];
                    for (i = 0; i < rt->nb_rtsp_streams; i++) { 
                        RTPDemuxContext *rtpctx2 = rt->rtsp_streams[i]->transport_priv;
                        AVStream *st2 = NULL;
                        if (rt->rtsp_streams[i]->stream_index >= 0)
                            st2 = s->streams[rt->rtsp_streams[i]->stream_index];
                        if (rtpctx2 && st && st2 &&
                            rtpctx2->first_rtcp_ntp_time == AV_NOPTS_VALUE) { 
                            rtpctx2->first_rtcp_ntp_time = rtpctx->first_rtcp_ntp_time;
                            rtpctx2->rtcp_ts_offset = av_rescale_q(
                                rtpctx->rtcp_ts_offset, st->time_base,
                                st2->time_base);
                        }
                    }
                    // Make real NTP start time available in AVFormatContext
                    if (s->start_time_realtime == AV_NOPTS_VALUE) { 
                        s->start_time_realtime = av_rescale (rtpctx->first_rtcp_ntp_time - (NTP_OFFSET << 32), 1000000, 1LL << 32);
                        if (rtpctx->st) { 
                            s->start_time_realtime -=
                                av_rescale (rtpctx->rtcp_ts_offset,
                                            (uint64_t) rtpctx->st->time_base.num * 1000000,
                                                       rtpctx->st->time_base.den);
                        }
                    }
                }
                if (ret == -RTCP_BYE) { 
                    rt->nb_byes++;
    
                    av_log(s, AV_LOG_DEBUG, "Received BYE for stream %d (%d/%d)\n",
                           rtsp_st->stream_index, rt->nb_byes, rt->nb_rtsp_streams);
    
                    if (rt->nb_byes == rt->nb_rtsp_streams)
                        return AVERROR_EOF;
                }
            }
        } else if (CONFIG_RTPDEC && rt->ts) { 
            ret = avpriv_mpegts_parse_packet(rt->ts, pkt, rt->recvbuf, len);
            if (ret >= 0) { 
                if (ret < len) { 
                    rt->recvbuf_len = len;
                    rt->recvbuf_pos = ret;
                    rt->cur_transport_priv = rt->ts;
                    return 1;
                } else { 
                    ret = 0;
                }
            }
        } else { 
            return AVERROR_INVALIDDATA;
        }
    end:
        if (ret < 0)
            goto redo;
        if (ret == 1)
            /* more packets may follow, so we save the RTP context */
            rt->cur_transport_priv = rtsp_st->transport_priv;
    
        return ret;
    }
    #endif /* CONFIG_RTPDEC */
    
    
    
    static int read_packet(AVFormatContext *s,
                           RTSPStream **rtsp_st, RTSPStream *first_queue_st,
                           int64_t wait_end)
    { 
        RTSPState *rt = s->priv_data;
        int len;
    
        switch(rt->lower_transport) { 
        default:
    #if CONFIG_RTSP_DEMUXER
        case RTSP_LOWER_TRANSPORT_TCP:
        //如果设置为tcp的话，进入这个函数
            len = ff_rtsp_tcp_read_packet(s, rtsp_st, rt->recvbuf, RECVBUF_SIZE);
            break;
    #endif
        case RTSP_LOWER_TRANSPORT_UDP:
        case RTSP_LOWER_TRANSPORT_UDP_MULTICAST:
            len = udp_read_packet(s, rtsp_st, rt->recvbuf, RECVBUF_SIZE, wait_end);
            if (len > 0 && (*rtsp_st)->transport_priv && rt->transport == RTSP_TRANSPORT_RTP)
                ff_rtp_check_and_send_back_rr((*rtsp_st)->transport_priv, (*rtsp_st)->rtp_handle, NULL, len);
            break;
        case RTSP_LOWER_TRANSPORT_CUSTOM:
            if (first_queue_st && rt->transport == RTSP_TRANSPORT_RTP &&
                wait_end && wait_end < av_gettime_relative())
                len = AVERROR(EAGAIN);
            else
                len = avio_read_partial(s->pb, rt->recvbuf, RECVBUF_SIZE);
            len = pick_stream(s, rtsp_st, rt->recvbuf, len);
            if (len > 0 && (*rtsp_st)->transport_priv && rt->transport == RTSP_TRANSPORT_RTP)
                ff_rtp_check_and_send_back_rr((*rtsp_st)->transport_priv, NULL, s->pb, len);
            break;
        }
    
        if (len == 0)
            return AVERROR_EOF;
    
        return len;
    }
    
    int ff_rtsp_tcp_read_packet(AVFormatContext *s, RTSPStream **prtsp_st,
                                uint8_t *buf, int buf_size)
    { 
        RTSPState *rt = s->priv_data;
        int id, len, i, ret;
        RTSPStream *rtsp_st;
    
        av_log(s, AV_LOG_TRACE, "tcp_read_packet:\n");
    redo:
    //这里是为了寻找rtp包头的位置，一直循环去找
        for (;;) { 
            RTSPMessageHeader reply;
    //https://blog.csdn.net/weixin_35804181/article/details/87383444
    //RTP(tcp) interleaverframe
    //|- Byte[0]:'$'-|- Byte[1]:channel-|-Byte[2:3]:package len-|－RTP common head－｜
    //第二个字节指示通道号，0表示RTP数据，1表示RTCP数据；后面两个字节用来指示插入的有效数据长度
            ret = ff_rtsp_read_reply(s, &reply, NULL, 1, NULL);
            if (ret < 0)
                return ret;
            if (ret == 1) /* received '$' */
                break;
            /* XXX: parse message */
            if (rt->state != RTSP_STATE_STREAMING)
                return 0;
        }
        //在这里获取数据长度，前三个字节是长度
        ret = ffurl_read_complete(rt->rtsp_hd, buf, 3);
        if (ret != 3)
            return -1;
        id  = buf[0];
        len = AV_RB16(buf + 1);
        av_log(s, AV_LOG_TRACE, "id=%d len=%d\n", id, len);
        if (len > buf_size || len < 8)
            goto redo;
        /* get the data */
        //在这里真正读取数据，传入的数据就是RTSPState中的URLContext
        ret = ffurl_read_complete(rt->rtsp_hd, buf, len);
        if (ret != len)
            return -1;
        if (rt->transport == RTSP_TRANSPORT_RDT &&
            ff_rdt_parse_header(buf, len, &id, NULL, NULL, NULL, NULL) < 0)
            return -1;
    
        /* find the matching stream */
        for (i = 0; i < rt->nb_rtsp_streams; i++) { 
            rtsp_st = rt->rtsp_streams[i];
            if (id >= rtsp_st->interleaved_min &&
                id <= rtsp_st->interleaved_max)
                goto found;
        }
        goto redo;
    found:
        *prtsp_st = rtsp_st;
        return len;
    }
    
    
    int ff_rtsp_read_reply(AVFormatContext *s, RTSPMessageHeader *reply,
                           unsigned char **content_ptr,
                           int return_on_interleaved_data, const char *method)
    { 
        RTSPState *rt = s->priv_data;
        char buf[4096], buf1[1024], *q;
        unsigned char ch;
        const char *p;
        int ret, content_length, line_count = 0, request = 0;
        unsigned char *content = NULL;
    
    start:
        line_count = 0;
        request = 0;
        content = NULL;
        memset(reply, 0, sizeof(*reply));
    
        /* parse reply (XXX: use buffers) */
        rt->last_reply[0] = '\0';
        for (;;) { 
            q = buf;
            for (;;) { 
            rtp基于tcp的包头比基于udp的包头多了4个字节： 
    * magic固定为0x24， 
    * channel用来区分音视频等多路流媒体的通道，其中偶数通道为流媒体内容，奇数通道为RTCP 
    * len表示数据包的长度减去开始的4个字节，即len字段之后的数据长度
    * https://blog.csdn.net/qq_23552895/article/details/88226339
                ret = ffurl_read_complete(rt->rtsp_hd, &ch, 1);
                av_log(s, AV_LOG_TRACE, "ret=%d c=%02x [%c]\n", ret, ch, ch);
                //如果没有读到直接返回错误-50532
                if (ret != 1)
                    return AVERROR_EOF;
                    //什么时候返回\n 
                    //rtsp 501 NoITimplementedrtsp 停止发送
                if (ch == '\n')
                    break;
                if (ch == '$' && q == buf) { 
                    if (return_on_interleaved_data) { 
                        return 1;
                    } else
                        ff_rtsp_skip_packet(s);
                } else if (ch != '\r') { 
                    if ((q - buf) < sizeof(buf) - 1)
                        *q++ = ch;
                }
            }
            *q = '\0';
    
            av_log(s, AV_LOG_TRACE, "line='%s'\n", buf);
    
            /* test if last line */
            if (buf[0] == '\0')
                break;
            p = buf;
            if (line_count == 0) { 
                /* get reply code */
                get_word(buf1, sizeof(buf1), &p);
                if (!strncmp(buf1, "RTSP/", 5)) { 
                    get_word(buf1, sizeof(buf1), &p);
                    reply->status_code = atoi(buf1);
                    av_strlcpy(reply->reason, p, sizeof(reply->reason));
                } else { 
                    av_strlcpy(reply->reason, buf1, sizeof(reply->reason)); // method
                    get_word(buf1, sizeof(buf1), &p); // object
                    request = 1;
                }
            } else { 
                ff_rtsp_parse_line(s, reply, p, rt, method);
                av_strlcat(rt->last_reply, p,    sizeof(rt->last_reply));
                av_strlcat(rt->last_reply, "\n", sizeof(rt->last_reply));
            }
            line_count++;
        }
    
        if (rt->session_id[0] == '\0' && reply->session_id[0] != '\0' && !request)
            av_strlcpy(rt->session_id, reply->session_id, sizeof(rt->session_id));
    
        content_length = reply->content_length;
        if (content_length > 0) { 
            /* leave some room for a trailing '\0' (useful for simple parsing) */
            content = av_malloc(content_length + 1);
            if (!content)
                return AVERROR(ENOMEM);
            ffurl_read_complete(rt->rtsp_hd, content, content_length);
            content[content_length] = '\0';
        }
        if (content_ptr)
            *content_ptr = content;
        else
            av_freep(&content);
    
        if (request) { 
            char buf[1024];
            char base64buf[AV_BASE64_SIZE(sizeof(buf))];
            const char* ptr = buf;
    
            if (!strcmp(reply->reason, "OPTIONS")) { 
                snprintf(buf, sizeof(buf), "RTSP/1.0 200 OK\r\n");
                if (reply->seq)
                    av_strlcatf(buf, sizeof(buf), "CSeq: %d\r\n", reply->seq);
                if (reply->session_id[0])
                    av_strlcatf(buf, sizeof(buf), "Session: %s\r\n",
                                                  reply->session_id);
            } else { 
                snprintf(buf, sizeof(buf), "RTSP/1.0 501 Not Implemented\r\n");
            }
            av_strlcat(buf, "\r\n", sizeof(buf));
    
            if (rt->control_transport == RTSP_MODE_TUNNEL) { 
                av_base64_encode(base64buf, sizeof(base64buf), buf, strlen(buf));
                ptr = base64buf;
            }
            ffurl_write(rt->rtsp_hd_out, ptr, strlen(ptr));
    
            rt->last_cmd_time = av_gettime_relative();
            /* Even if the request from the server had data, it is not the data * that the caller wants or expects. The memory could also be leaked * if the actual following reply has content data. */
            if (content_ptr)
                av_freep(content_ptr);
            /* If method is set, this is called from ff_rtsp_send_cmd, * where a reply to exactly this request is awaited. For * callers from within packet receiving, we just want to * return to the caller and go back to receiving packets. */
            if (method)
                goto start;
            return 0;
        }
    
        if (rt->seq != reply->seq) { 
            av_log(s, AV_LOG_WARNING, "CSeq %d expected, %d received.\n",
                rt->seq, reply->seq);
        }
    
        /* EOS */
        if (reply->notice == 2101 /* End-of-Stream Reached */      ||
            reply->notice == 2104 /* Start-of-Stream Reached */    ||
            reply->notice == 2306 /* Continuous Feed Terminated */) { 
            rt->state = RTSP_STATE_IDLE;
        } else if (reply->notice >= 4400 && reply->notice < 5500) { 
            return AVERROR(EIO); /* data or server error */
        } else if (reply->notice == 2401 /* Ticket Expired */ ||
                 (reply->notice >= 5500 && reply->notice < 5600) /* end of term */ )
            return AVERROR(EPERM);
    
        return 0;
    }
    
    
    int ffurl_read_complete(URLContext *h, unsigned char *buf, int size)
    { 
        if (!(h->flags & AVIO_FLAG_READ))
            return AVERROR(EIO);
            //注意这里的 h->prot->url_read是tcp read
        return retry_transfer_wrapper(h, buf, size, size, h->prot->url_read);
    }
    
    
    
    
    
    
    
    
    
    
    
    
    static inline int retry_transfer_wrapper(URLContext *h, uint8_t *buf,
                                             int size, int size_min,
                                             int (*transfer_func)(URLContext *h,
                                                                  uint8_t *buf,
                                                                  int size))
    { 
        int ret, len;
        int fast_retries = 5;
        int64_t wait_since = 0;
    
        len = 0;
        while (len < size_min) { 
        //在这里执行打断回调函数
            if (ff_check_interrupt(&h->interrupt_callback))
                return AVERROR_EXIT;
       //tcp read 回调
            ret = transfer_func(h, buf + len, size - len);
            if (ret == AVERROR(EINTR))
                continue;
                //如果是非阻塞字节返回
            if (h->flags & AVIO_FLAG_NONBLOCK)
                return ret;
            if (ret == AVERROR(EAGAIN)) { 
                ret = 0;
                //循环5次
                if (fast_retries) { 
                    fast_retries--;
                } else { 
                    if (h->rw_timeout) { 
                        if (!wait_since)
                            wait_since = av_gettime_relative();
                        else if (av_gettime_relative() > wait_since + h->rw_timeout)
                            return AVERROR(EIO);
                    }
                    av_usleep(1000);
                }
            } else if (ret == AVERROR_EOF)
                return (len > 0) ? len : AVERROR_EOF;
            else if (ret < 0)
                return ret;
            if (ret) { 
                fast_retries = FFMAX(fast_retries, 2);
                wait_since = 0;
            }
            len += ret;
        }
        return len;
    }
    
    static int tcp_read(URLContext *h, uint8_t *buf, int size)
    { 
        TCPContext *s = h->priv_data;
        int ret;
    //这里设置非阻塞后就不会执行了，直接去recv
    /** * Use non-blocking mode. * If this flag is set, operations on the context will return * AVERROR(EAGAIN) if they can not be performed immediately. * If this flag is not set, operations on the context will never return * AVERROR(EAGAIN). * Note that this flag does not affect the opening/connecting of the * context. Connecting a protocol will always block if necessary (e.g. on * network protocols) but never hang (e.g. on busy devices). * Warning: non-blocking protocols is work-in-progress; this flag may be * silently ignored. */
    //#define AVIO_FLAG_NONBLOCK 8
        if (!(h->flags & AVIO_FLAG_NONBLOCK)) { 
            ret = ff_network_wait_fd_timeout(s->fd, 0, h->rw_timeout, &h->interrupt_callback);
            if (ret)
                return ret;
        }
        成功执行时，返回接收到的字节数。
    
    另一端已关闭则返回0。
    
    失败返回-1，
    
    errno被设为以下的某个值
    
    EAGAIN：套接字已标记为非阻塞，而接收操作被阻塞或者接收超时
    
    EBADF：sock不是有效的描述词
    
    ECONNREFUSE：远程主机阻绝网络连接
    
    EFAULT：内存空间访问出错
    
    EINTR：操作被信号中断
    
    EINVAL：参数无效
    
    ENOMEM：内存不足
    
    ENOTCONN：与面向连接关联的套接字尚未被连接上
    
    ENOTSOCK：sock索引的不是套接字 当返回值是0时，为正常关闭连接
        ret = recv(s->fd, buf, size, 0);
        if (ret == 0)
            return AVERROR_EOF;
        return ret < 0 ? ff_neterrno() : ret;
    }
    
    int ff_network_wait_fd_timeout(int fd, int write, int64_t timeout, AVIOInterruptCB *int_cb)
    { 
        int ret;
        int64_t wait_start = 0;
    
        while (1) { 
            if (ff_check_interrupt(int_cb))
                return AVERROR_EXIT;//负数
            ret = ff_network_wait_fd(fd, write);
            if (ret != AVERROR(EAGAIN))
                return ret;
               //相当于执行若干次循环，timeout就是我们传入的时间stimeout
            if (timeout > 0) { 
                if (!wait_start)
                    wait_start = av_gettime_relative();
                else if (av_gettime_relative() - wait_start > timeout)
                    return AVERROR(ETIMEDOUT);
            }
        }
    }
    
    int ff_network_wait_fd(int fd, int write)
    { 
        int ev = write ? POLLOUT : POLLIN;
        struct pollfd p = {  .fd = fd, .events = ev, .revents = 0 };
        int ret;
        ret = poll(&p, 1, POLLING_TIME);
        return ret < 0 ? ff_neterrno() : p.revents & (ev | POLLERR | POLLHUP) ? 0 : AVERROR(EAGAIN);
    }

[Link 1]: https://blog.csdn.net/weixin_43360707/article/details/116005107