vlc_input buffer管理 & 时钟同步(转)

vlc_input buffer管理 & 时钟同步

一、背景
1、当播放网络视频流时(比如udp视频流)，发送方(编码)和接收方(解码)是并行操作的，如果发送太慢(或因为网络原因出现延迟)的话，接收方将不能及时得到数据，导致解码出错，所以需要对接收buffer进行管理。
2、编码器会把自身的参考时钟(PCR)保存到视频流中，供解码器同步用。对于网络视频流，VLC也通过接收的PCR信息与自身系统时钟比较，以计算网络延迟并相应的调节接收buffer。

二、buffer管理(pts_delay)
1、input_clock

/*****************************************************************************
 * input_clock_Update: manages a clock reference
 *
 *  i_ck_stream: date in stream clock
 *  i_ck_system: date in system clock
 *****************************************************************************/
void input_clock_Update( input_clock_t *cl, vlc_object_t *p_log,
                         bool *pb_late,
                         bool b_can_pace_control, bool b_buffering_allowed,
                         mtime_t i_ck_stream, mtime_t i_ck_system )
{
    bool b_reset_reference = false;

    assert( i_ck_stream > VLC_TS_INVALID && i_ck_system > VLC_TS_INVALID );

    vlc_mutex_lock( &cl->lock );

    if( !cl->b_has_reference )   //失去参考        
    {
       /* */
        b_reset_reference= true;   //重新同步    
    }
   else if( cl->last.i_stream > VLC_TS_INVALID &&
             ( (cl->last.i_stream - i_ck_stream) > CR_MAX_GAP ||
               (cl->last.i_stream - i_ck_stream) < -CR_MAX_GAP ) )  //delay超过60s，视为Stream discontinuity。
    {
        /* Stream discontinuity, for which we haven't received a
         * warning from the stream control facilities (dd-edited
         * stream ?). */    //严重错误，不应该发生。
        msg_Warn( p_log, "clock gap, unexpected stream discontinuity" );
        cl->i_ts_max = VLC_TS_INVALID;

        /* */
        msg_Warn( p_log, "feeding synchro with a new reference point trying to recover from clock gap" );
        b_reset_reference= true;   //重新同步
    }

    /* */
    if( b_reset_reference )
    {
        cl->i_next_drift_update = VLC_TS_INVALID;
        AvgReset( &cl->drift );   //复位平均统计值        

        /* Feed synchro with a new reference point. */
        cl->b_has_reference = true;
        cl->ref = clock_point_Create( i_ck_stream,
                                      __MAX( cl->i_ts_max + CR_MEAN_PTS_GAP, i_ck_system ) );   //建立ref clock point
        cl->b_has_external_clock = false;
    }

    /* Compute the drift between the stream clock and the system clock
     * when we don't control the source pace */
    if( !b_can_pace_control && cl->i_next_drift_update < i_ck_system )
    {
        const mtime_t i_converted = ClockSystemToStream( cl, i_ck_system );

        AvgUpdate( &cl->drift, i_converted - i_ck_stream ); //计算并更新ck_system与ck_stream的偏差(采用累积平均算法减少波动)      
        cl->i_next_drift_update = i_ck_system + CLOCK_FREQ/5; /* FIXME why that */    }    /* Update the extra buffering value */    if( !b_can_pace_control || b_reset_reference )    {        cl->i_buffering_duration = 0;   //Amount of extra buffering expressed in stream clock(复位)        }    else if( b_buffering_allowed )   //input_DecoderGetFifoSize不多，允许es_out缓冲。            {        /* Try to bufferize more than necessary by reading         * CR_BUFFERING_RATE/256 faster until we have CR_BUFFERING_TARGET.         */        const mtime_t i_duration = __MAX( i_ck_stream - cl->last.i_stream, 0 );        cl->i_buffering_duration += ( i_duration * CR_BUFFERING_RATE + 255 ) / 256;   //???        if( cl->i_buffering_duration > CR_BUFFERING_TARGET )            cl->i_buffering_duration = CR_BUFFERING_TARGET;    }    //fprintf( stderr, "input_clock_Update: %d :: %lld ", b_buffering_allowed, cl->i_buffering_duration/1000 );    /* */    cl->last = clock_point_Create( i_ck_stream, i_ck_system );  //记录last clock point      /* It does not take the decoder latency into account but it is not really     * the goal of the clock here */    const mtime_t i_system_expected = ClockStreamToSystem( cl, i_ck_stream + AvgGet( &cl->drift ) );    //给ck_stream加上上面计算的平均差值，转化为对应的system_expected    const mtime_t i_late = ( i_ck_system - cl->i_pts_delay ) - i_system_expected;   //计算是否超出延迟设置(需要增大接收buffer?)    *pb_late = i_late > 0;    if( i_late > 0 )    {        cl->late.pi_value[cl->late.i_index] = i_late;   //记录延迟情况(作为要增大多少buffer的依据)        cl->late.i_index = ( cl->late.i_index + 1 ) % INPUT_CLOCK_LATE_COUNT;    }    vlc_mutex_unlock( &cl->lock );}void input_clock_SetJitter( input_clock_t *cl,                            mtime_t i_pts_delay, int i_cr_average ){    vlc_mutex_lock( &cl->lock );    /* Update late observations */    const mtime_t i_delay_delta = i_pts_delay - cl->i_pts_delay;    mtime_t pi_late[INPUT_CLOCK_LATE_COUNT];    for( int i = 0; i < INPUT_CLOCK_LATE_COUNT; i++ )        pi_late[i] = __MAX( cl->late.pi_value[(cl->late.i_index + 1 + i)%INPUT_CLOCK_LATE_COUNT] - i_delay_delta, 0 );    for( int i = 0; i < INPUT_CLOCK_LATE_COUNT; i++ )        cl->late.pi_value[i] = 0;    cl->late.i_index = 0;    for( int i = 0; i < INPUT_CLOCK_LATE_COUNT; i++ )    {        if( pi_late[i] <= 0 )            continue;        cl->late.pi_value[cl->late.i_index] = pi_late[i];        cl->late.i_index = ( cl->late.i_index + 1 ) % INPUT_CLOCK_LATE_COUNT;    }    /* TODO always save the value, and when rebuffering use the new one if smaller     * TODO when increasing -> force rebuffering     */    if( cl->i_pts_delay < i_pts_delay )        cl->i_pts_delay = i_pts_delay;     //更新pts_delay!!!         /* */    if( i_cr_average < 10 )        i_cr_average = 10;    if( cl->drift.i_divider != i_cr_average )        AvgRescale( &cl->drift, i_cr_average );    vlc_mutex_unlock( &cl->lock );}

2、es_out

    case ES_OUT_SET_PCR:
    case ES_OUT_SET_GROUP_PCR:
    {
.
        /* TODO do not use mdate() but proper stream acquisition date */
        bool b_late;
        input_clock_Update( p_pgrm->p_clock, VLC_OBJECT(p_sys->p_input),
                            &b_late,
                            p_sys->p_input->p->b_can_pace_control || p_sys->b_buffering,
                            EsOutIsExtraBufferingAllowed( out ),
                            i_pcr, mdate() );

        if( !p_sys->p_pgrm )
            return VLC_SUCCESS;

        if( p_sys->b_buffering )
        {
            /* Check buffering state on master clock update */
            EsOutDecodersStopBuffering( out, false );
        }
        else if( p_pgrm == p_sys->p_pgrm )
        {
            if( b_late && ( !p_sys->p_input->p->p_sout ||
                                 !p_sys->p_input->p->b_out_pace_control ) )
            {
                const mtime_t i_pts_delay_base = p_sys->i_pts_delay - p_sys->i_pts_jitter;   //系统当前的pts_delay值(一开始通过用户的"network-caching"初始化，后续根据延迟统计情况动态更新)
                mtime_t i_pts_delay = input_clock_GetJitter( p_pgrm->p_clock );   //获取上面统计的平均延迟                

               /* Avoid dangerously high value */
                const mtime_t i_jitter_max = INT64_C(1000) * var_InheritInteger( p_sys->p_input, "clock-jitter" );
                if( i_pts_delay > __MIN( i_pts_delay_base + i_jitter_max, INPUT_PTS_DELAY_MAX ) )   //pts_delay > 5s (严重延迟)                   
                {  
                     msg_Err( p_sys->p_input,
                             "ES_OUT_SET_(GROUP_)PCR  is called too late (jitter of %d ms ignored)",
                             (int)(i_pts_delay - i_pts_delay_base) / 1000 );
                    i_pts_delay = p_sys->i_pts_delay;   //恢复用户设置的pts_delay                    

                     /* reset clock */
                    for( int i = 0; i < p_sys->i_pgrm; i++ )
                      input_clock_Reset( p_sys->pgrm[i]->p_clock ); //复位时钟，重新同步。                
                }
               else
                {
                    msg_Err( p_sys->p_input,
                             "ES_OUT_SET_(GROUP_)PCR  is called too late (pts_delay increased to %d ms)",
                             (int)(i_pts_delay/1000) );   //延迟在范围内(5s)，增加pts_delay为统计值。

                    /* Force a rebufferization when we are too late */

                    /* It is not really good, as we throw away already buffered data
                     * TODO have a mean to correctly reenter bufferization */
                    es_out_Control( out, ES_OUT_RESET_PCR );
                }

                es_out_SetJitter( out, i_pts_delay_base, i_pts_delay - i_pts_delay_base, p_sys->i_cr_average );   //更新p_sys->i_pts_delay和p_sys->i_pts_jitter            }        }        return VLC_SUCCESS;    }    case ES_OUT_SET_JITTER:    {        mtime_t i_pts_delay  = (mtime_t)va_arg( args, mtime_t );        mtime_t i_pts_jitter = (mtime_t)va_arg( args, mtime_t );        int     i_cr_average = (int)va_arg( args, int );        bool b_change_clock =            i_pts_delay + i_pts_jitter != p_sys->i_pts_delay ||            i_cr_average != p_sys->i_cr_average;        assert( i_pts_jitter >= 0 );        p_sys->i_pts_delay  = i_pts_delay + i_pts_jitter;        p_sys->i_pts_jitter = i_pts_jitter;        p_sys->i_cr_average = i_cr_average;        for( int i = 0; i < p_sys->i_pgrm && b_change_clock; i++ )            input_clock_SetJitter( p_sys->pgrm[i]->p_clock,                                   i_pts_delay + i_pts_jitter, i_cr_average );        return VLC_SUCCESS;    }static void EsOutDecodersStopBuffering( es_out_t *out, bool b_forced )   //汇报buffer进度/等待decoder fifo empty{    es_out_sys_t *p_sys = out->p_sys;    int i_ret;    mtime_t i_stream_start;    mtime_t i_system_start;    mtime_t i_stream_duration;    mtime_t i_system_duration;    if (input_clock_GetState( p_sys->p_pgrm->p_clock,                                  &i_stream_start, &i_system_start,                                  &i_stream_duration, &i_system_duration ))        return;    mtime_t i_preroll_duration = 0;    if( p_sys->i_preroll_end >= 0 )        i_preroll_duration = __MAX( p_sys->i_preroll_end - i_stream_start, 0 );    const mtime_t i_buffering_duration = p_sys->i_pts_delay +                                         i_preroll_duration +                                         p_sys->i_buffering_extra_stream - p_sys->i_buffering_extra_initial;   //计算接收缓冲的长度(时间)    if( i_stream_duration <= i_buffering_duration && !b_forced )   //还buffer完            {double f_level;        if (i_buffering_duration == 0)            f_level = 0;        else            f_level = __MAX( (double)i_stream_duration / i_buffering_duration, 0 );        input_SendEventCache( p_sys->p_input, f_level );   //INPUT_EVENT_CACHE        msg_Dbg( p_sys->p_input, "Buffering %d%%", (int)(100 * f_level) );        return;    }    input_SendEventCache( p_sys->p_input, 1.0 );   //INPUT_EVENT_CACHE    msg_Dbg( p_sys->p_input, "Stream buffering done (%d ms in %d ms)",              (int)(i_stream_duration/1000), (int)(i_system_duration/1000) );    p_sys->b_buffering = false;   //buffer完，标志之，除非EsOutChangePosition或EsOutFrameNext，否则不再buffer。    p_sys->i_preroll_end = -1;    if( p_sys->i_buffering_extra_initial > 0 )    {        /* FIXME wrong ? */        return;    }    const mtime_t i_decoder_buffering_start = mdate();    for( int i = 0; i < p_sys->i_es; i++ )    {        es_out_id_t *p_es = p_sys->es[i];        if( !p_es->p_dec || p_es->fmt.i_cat == SPU_ES )            continue;        input_DecoderWait( p_es->p_dec );   //等待解码器准备好(开始工作)        if( p_es->p_dec_record )            input_DecoderWait( p_es->p_dec_record );    }    msg_Dbg( p_sys->p_input, "Decoder wait done in %d ms",              (int)(mdate() - i_decoder_buffering_start)/1000 );    /* Here is a good place to destroy unused vout with every demuxer */    input_resource_TerminateVout( p_sys->p_input->p->p_resource );    /* */    const mtime_t i_wakeup_delay = 10*1000; /* FIXME CLEANUP thread wake up time*/    const mtime_t i_current_date = p_sys->b_paused ? p_sys->i_pause_date : mdate();    input_clock_ChangeSystemOrigin( p_sys->p_pgrm->p_clock, true,                                    i_current_date + i_wakeup_delay - i_buffering_duration );  // 实际buffer的数据一般要比用户设置的(i_buffering_duration，其实主要是i_pts_delay)要长，这时候要算出多出来的 部分，然后调整本地时钟(ref.i_system)，以补偿buffer过程中引入的缓冲时间，避免解码和显示的时候报错。(建议调试的时候，关掉时钟 同步，容易分析)    for( int i = 0; i < p_sys->i_es; i++ )    {        es_out_id_t *p_es = p_sys->es[i];        if( !p_es->p_dec )            continue;        input_DecoderStopWait( p_es->p_dec );   //buffer前调用input_DecoderStartWait；buffer完后，调用input_DecoderStopWait。        if( p_es->p_dec_record )            input_DecoderStopWait( p_es->p_dec_record );    }}

3、p_sys->i_pts_delay初始值

src/input/input.c

static int InputSourceInit( input_thread_t *p_input,
                            input_source_t *in, const char *psz_mrl,
                            const char *psz_forced_demux, bool b_in_can_fail )
{
.
        if( !p_input->b_preparsing )
        {
            bool b;

            stream_Control( p_stream, STREAM_CAN_CONTROL_PACE,
                            &in->b_can_pace_control );  
            in->b_can_rate_control = in->b_can_pace_control;   //false

            in->b_rescale_ts = true;

            stream_Control( p_stream, STREAM_CAN_PAUSE, &in->b_can_pause );   //false
            var_SetBool( p_input, "can-pause",
                         in->b_can_pause || !in->b_can_pace_control ); /* XXX temporary because of es_out_timeshift*/
            var_SetBool( p_input, "can-rate",
                         !in->b_can_pace_control || in->b_can_rate_control ); /* XXX temporary because of es_out_timeshift*/
            var_SetBool( p_input, "can-rewind",
                         !in->b_rescale_ts && !in->b_can_pace_control );

            stream_Control( p_stream, STREAM_CAN_SEEK, &b );   //false
            var_SetBool( p_input, "can-seek", b );

            in->b_title_demux = false;

            stream_Control( p_stream, STREAM_GET_PTS_DELAY, &i_pts_delay );
        }
....

    if( !p_input->b_preparsing )
    {
        int i_attachment;
        input_attachment_t **attachment;
        if( !demux_Control( in->p_demux, DEMUX_GET_ATTACHMENTS,
                             &attachment, &i_attachment ) )
        {
            vlc_mutex_lock( &p_input->p->p_item->lock );
            AppendAttachment( &p_input->p->i_attachment, &p_input->p->attachment, &p_input->p->attachment_demux,
                              i_attachment, attachment, in->p_demux );
            vlc_mutex_unlock( &p_input->p->p_item->lock );
        }

        /* PTS delay: request from demux first. This is required for
         * access_demux and some special cases like SDP demux. Otherwise,
         * fallback to access */
        if( demux_Control( in->p_demux, DEMUX_GET_PTS_DELAY,
                           &in->i_pts_delay ) )
            in->i_pts_delay = i_pts_delay;   //把i_pts_delay的值赋给in->i_pts_delay
        if( in->i_pts_delay > INPUT_PTS_DELAY_MAX )
            in->i_pts_delay = INPUT_PTS_DELAY_MAX;
        else if( in->i_pts_delay < 0 )
            in->i_pts_delay = 0;
    }

static void UpdatePtsDelay( input_thread_t *p_input )
{
    input_thread_private_t *p_sys = p_input->p;

    /* Get max pts delay from input source */
    mtime_t i_pts_delay = p_sys->input.i_pts_delay;
    for( int i = 0; i < p_sys->i_slave; i++ )
        i_pts_delay = __MAX( i_pts_delay, p_sys->slave[i]->i_pts_delay );

    if( i_pts_delay < 0 )
        i_pts_delay = 0;

    /* Take care of audio/spu delay */
    const mtime_t i_audio_delay = var_GetTime( p_input, "audio-delay" );
    const mtime_t i_spu_delay   = var_GetTime( p_input, "spu-delay" );
    const mtime_t i_extra_delay = __MIN( i_audio_delay, i_spu_delay );
    if( i_extra_delay < 0 )
        i_pts_delay -= i_extra_delay;

    /* Update cr_average depending on the caching */
    const int i_cr_average = var_GetInteger( p_input, "cr-average" ) * i_pts_delay / DEFAULT_PTS_DELAY;   //计算i_pts_delay对应的i_cr_average

    /* */
    es_out_SetDelay( p_input->p->p_es_out_display, AUDIO_ES, i_audio_delay );    es_out_SetDelay( p_input->p->p_es_out_display, SPU_ES, i_spu_delay );    es_out_SetJitter( p_input->p->p_es_out, i_pts_delay, 0, i_cr_average );   //设置p_sys->i_pts_delay初始值(详细见上面的ES_OUT_SET_JITTER分支)}

modules/access/udp.c:

        case ACCESS_GET_PTS_DELAY:
            pi_64 = (int64_t*)va_arg( args, int64_t * );
            *pi_64 = INT64_C(1000)
                   * var_InheritInteger(p_access, "network-caching");   //通过“network-caching”变量获取i_pts_delay初始值

            break;

三、时钟同步(drift)
Data caching is mainly a way to cope with streams which have unreliable 
sending rates (be it network/server latency, jitter, bursts, etc). Now 
if you need heavy caching, that more than likely means you have large 
changes in your receiving rate, so you'll want to increase cr-average to 
smooth out these changes.

This is why the default caching in the udp access module is pretty low (udp 
is mostly used for reliable / low delay streaming), while you have a high 
caching value for http.

I do agree that the problems you raise need to be fixed, although I don't 
think your fixes are appropriate.

The clock synchro algo is there to keep the playback speed synchronised with 
the server, not only to avoid buffer underruns / overruns, but also to keep 
a low delay from the server (depending on the value of caching set by the 
user).

The problem with VLC's clock is that it does its synchro by comparing the 
time of arrival of data with the timestamp set on it by the server, aka the 
PCR (which btw is why the prebuffering introduced lately might not be such 
a good idea).
This method is appropriate for very low delay streaming which doesn't have 
much jitter and bursts in the transmission, but it isn't for anything else 
(and remember, VLC was designed as an intranet streaming client).
If VLC is allowed to cache enough data, then a more sensible way to do this 
synchro is to actually look at how much data we have in our cache (in units 
of time), and if this value is above or below some treshold then start 
adapting the clock. It might still be interesting (or not) to keep using 
the current synchro algo for very low delay streams though.

1、drift计算
请参考前面的"input_clock_Update"

void input_clock_Update( input_clock_t *cl, vlc_object_t *p_log,
                         bool *pb_late,
                         bool b_can_pace_control, bool b_buffering_allowed,
                         mtime_t i_ck_stream, mtime_t i_ck_system )
{
.
    if( !b_can_pace_control && cl->i_next_drift_update < i_ck_system )
    {
        const mtime_t i_converted = ClockSystemToStream( cl, i_ck_system );

        AvgUpdate( &cl->drift, i_converted - i_ck_stream ); //计算并更新平均偏移
        cl->i_next_drift_update = i_ck_system + CLOCK_FREQ/5; /* FIXME why that */
    }

2、drift的使用

int input_clock_ConvertTS( input_clock_t *cl,
                           int *pi_rate, mtime_t *pi_ts0, mtime_t *pi_ts1,
                           mtime_t i_ts_bound )
{
.
    /* */
    const mtime_t i_ts_buffering = cl->i_buffering_duration * cl->i_rate / INPUT_RATE_DEFAULT;
    const mtime_t i_ts_delay = cl->i_pts_delay + ClockGetTsOffset( cl );

    /* */
    if( *pi_ts0 > VLC_TS_INVALID )
    {
        *pi_ts0 = ClockStreamToSystem( cl, *pi_ts0 + AvgGet( &cl->drift ) );   //补偿时钟偏移
        if( *pi_ts0 > cl->i_ts_max )            cl->i_ts_max = *pi_ts0;        *pi_ts0 += i_ts_delay;    }

参考：
http://blog.csdn.net/coroutines/article/details/7472743
https://mailman.videolan.org/pipermail/vlc-devel/2005-November/020526.html