ffplay.c 源码分析- 队列操作

deep_sadness

发布于 2018-12-18 17:17:14

9550

发布于 2018-12-18 17:17:14

文章被收录于专栏：Flutter入门Flutter入门

前两遍文章，我们分析了视频部分和音频播放。其中包含的队列操作，还是让人迷惑。这边文章，就主要来梳理一下队列操作。

主要是FrameQueue 和 PackQueue

PacketQueue

PacketQueue比较简单，因为AVPackList本身就是一个链表。其实在avformat.h中定义了

typedef struct AVPacketList {
    AVPacket pkt;
    struct AVPacketList *next;
} AVPacketList;

但是这个AVPacketList，需要的serial，所以就自己定义。

typedef struct MyAVPacketList {
    AVPacket pkt;
    struct MyAVPacketList *next;
    //操作数
    int serial;
} MyAVPacketList;

//再次包装的PacketQueue，维持一些锁和统计变量
typedef struct PacketQueue {
    MyAVPacketList *first_pkt, *last_pkt;
    int nb_packets;
    int size;
    int64_t duration;
    int abort_request;
    int serial;
    SDL_mutex *mutex;
    SDL_cond *cond;
} PacketQueue;

使用起来也很简单。

init

/* packet queue handling */
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;
}

start

static void packet_queue_start(PacketQueue *q)
{
    SDL_LockMutex(q->mutex);
    q->abort_request = 0;
    packet_queue_put_private(q, &flush_pkt);
    SDL_UnlockMutex(q->mutex);
}

-flush 方法 flush方法基本就是clear方法

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);
}

destroy 释放的方法，先把队列flush ,同时释放我们的锁资源

static void packet_queue_destroy(PacketQueue *q)
{
    packet_queue_flush(q);
    SDL_DestroyMutex(q->mutex);
    SDL_DestroyCond(q->cond);
}

更改标志位

static void packet_queue_abort(PacketQueue *q)
{
    SDL_LockMutex(q->mutex);

    q->abort_request = 1;

    SDL_CondSignal(q->cond);

    SDL_UnlockMutex(q->mutex);
}

static int packet_queue_put(PacketQueue *q, AVPacket *pkt)
{
    int ret;
    //为了保证线程安全。锁住
    SDL_LockMutex(q->mutex);
    //真正的操作，在packet_queue_put_private方法中
    ret = packet_queue_put_private(q, pkt);
    SDL_UnlockMutex(q->mutex);

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

    return ret;
}


static int packet_queue_put_private(PacketQueue *q, AVPacket *pkt)
{
    //声明一个新的MyAVPacketList
    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;
    
    //放置指针，并记录first_pkt 和 last_pkt
    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;
}
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);
}

/* return < 0 if aborted, 0 if no packet and > 0 if packet.  */
static int packet_queue_get(PacketQueue *q, AVPacket *pkt, int block, int *serial)
{
    MyAVPacketList *pkt1;
    int ret;

    SDL_LockMutex(q->mutex);
    
    //取不到数据时，如果是阻塞的，则会继续等待循环，如果不是阻塞的，就会直接跳出循环
    for (;;) {
        if (q->abort_request) {
            ret = -1;
            break;
        }

        pkt1 = q->first_pkt;
        if (pkt1) {
            q->first_pkt = pkt1->next;
            if (!q->first_pkt)
                q->last_pkt = NULL;
            q->nb_packets--;
            q->size -= pkt1->pkt.size + sizeof(*pkt1);
            q->duration -= pkt1->pkt.duration;
            *pkt = pkt1->pkt;
            if (serial)
                *serial = pkt1->serial;
            av_free(pkt1);
            ret = 1;
            break;
        } else if (!block) {
            ret = 0;
            break;
        } else {
            SDL_CondWait(q->cond, q->mutex);
        }
    }
    SDL_UnlockMutex(q->mutex);
    return ret;
}

小结

可以看到PacketList的每一个操作，都是线程同步的，都用互斥锁给锁住了。其中如果需要阻塞的方法的话，则在数组为空时，去取数组，会阻塞主，等待生产方产生数据。
PacketList的操作也很简单，只用get和put方法就可以取到头部的数据，和放置数据到尾部了。

FrameQueue

FrameQueue 就相对比较复杂。

他和PacketList不同，因为他不是用链表的方式。用的是数组。
它是通过两个角标来控制。一个是rindex负责读。一个是　windex负责写。通过size来判断，当前数组中的剩余。当需要写的时候，先通过windex 判断是否还有可以写的部分，有的话，就会返回一个可写的数据。使用完，需要将windex+1. 当Size满了。就不会返回数据了。当需要读的时候，会通过rindex 返回当前的数据。

//自定义一个Frame结构体。
// 因为有考虑字幕，所以一个正在的Frame包括 AVFrame AVSubtitle AVRational 等参数
typedef struct Frame {
    AVFrame *frame;
    AVSubtitle sub;
    int serial;
    double pts;           /* presentation timestamp for the frame */
    double duration;      /* estimated duration of the frame */
    int64_t pos;          /* byte position of the frame in the input file */
    int width;
    int height;
    int format;
    AVRational sar;
    int uploaded;
    int flip_v;
} Frame;

//定义一个FrameQueue结构体。如上所诉，使用了数组
typedef struct FrameQueue {
    Frame queue[FRAME_QUEUE_SIZE];
    //我觉得最关键的三个变量。来帮助读写的。 
    int rindex;
    int windex;
    int size;
    //最大的size
    int max_size;

    //下面这两个变量有什么用，存在疑问？
    int keep_last;
    int rindex_shown;

    //当前的队列的锁
    SDL_mutex *mutex;
    SDL_cond *cond;
    
    //为什么这个队列还要对应一个PacketQueue？
    PacketQueue *pktq;
} FrameQueue;

init

static int frame_queue_init(FrameQueue *f, PacketQueue *pktq, int max_size, int keep_last)
{
    int i;
    //初始化的时候，都是用memset的方式，将变量重置
    memset(f, 0, sizeof(FrameQueue));
    //创建锁资源
    if (!(f->mutex = SDL_CreateMutex())) {
        av_log(NULL, AV_LOG_FATAL, "SDL_CreateMutex(): %s\n", SDL_GetError());
        return AVERROR(ENOMEM);
    }
    if (!(f->cond = SDL_CreateCond())) {
        av_log(NULL, AV_LOG_FATAL, "SDL_CreateCond(): %s\n", SDL_GetError());
        return AVERROR(ENOMEM);
    }
    //保存对应的packet_queue
    f->pktq = pktq;
    f->max_size = FFMIN(max_size, FRAME_QUEUE_SIZE);
    //判断是否要保留最后一个。ffplay中的音频和视频，都需要保留最后一个。
    f->keep_last = !!keep_last;
    //初始化数组中的frame，因为这些frame最后，都是我们提供来用的，所以要实现初始化好。
    for (i = 0; i < f->max_size; i++)
        if (!(f->queue[i].frame = av_frame_alloc()))
            return AVERROR(ENOMEM);
    return 0;
}

销毁

static void frame_queue_unref_item(Frame *vp)
{
    av_frame_unref(vp->frame);
    avsubtitle_free(&vp->sub);
}

//将内部的frame和锁释放
static void frame_queue_destory(FrameQueue *f)
{
    int i;
    for (i = 0; i < f->max_size; i++) {
        Frame *vp = &f->queue[i];
        frame_queue_unref_item(vp);
        av_frame_free(&vp->frame);
    }
    SDL_DestroyMutex(f->mutex);
    SDL_DestroyCond(f->cond);
}

//单独的这样的方法，也不知道要怎么用？？
static void frame_queue_signal(FrameQueue *f)
{
    SDL_LockMutex(f->mutex);
    SDL_CondSignal(f->cond);
    SDL_UnlockMutex(f->mutex);
}

写入

push push的操作，要先通过frame_queue_peek_writable 得到一个可以写的Frame

static Frame *frame_queue_peek_writable(FrameQueue *f)
{
    /* wait until we have space to put a new frame */
    SDL_LockMutex(f->mutex);
    //等待。直到可以取
    while (f->size >= f->max_size &&
           !f->pktq->abort_request) {
        SDL_CondWait(f->cond, f->mutex);
    }
    SDL_UnlockMutex(f->mutex);
    //可以取到了。这个时候，会接触这个锁？这里的锁，只有当index发生改变的时候，才会锁。  

    if (f->pktq->abort_request)
        return NULL;
    
    //直接返回f->windex
    return &f->queue[f->windex];
}

得到一个可以操作的AVFrame 等我们操作完。之后要调用 frame_queue_push将，角标进行移动。注意这个时候移动的是windex.

static void frame_queue_push(FrameQueue *f)
{
    //windex 的移动没有加锁。
    if (++f->windex == f->max_size)
        f->windex = 0;
    //对size进行加锁了。因为size 会影响取的时候的阻塞。
    SDL_LockMutex(f->mutex);
    f->size++;
    SDL_CondSignal(f->cond);
    SDL_UnlockMutex(f->mutex);
}

读取

先判断是否还有充足的的可读

/* return the number of undisplayed frames in the queue */
static int frame_queue_nb_remaining(FrameQueue *f)
{
    return f->size - f->rindex_shown;
}

除了这个方法，我们之前看到，读取视频的时候，会去计算两个帧之间的差距，会先后调用

            lastvp = frame_queue_peek_last(&is->pictq);
            vp = frame_queue_peek(&is->pictq);

//peek 出当前的。因为f->rindex + f->rindex_shown可能会超过max_size,所以用了取余
static Frame *frame_queue_peek(FrameQueue *f)
{
    return &f->queue[(f->rindex + f->rindex_shown) % f->max_size];
}


//取出最后一个。这个是在keep_last的时候才能用吗？这个表示的是真正的最后一个。因为通常我们会在视频和音频的队列中默认保留一帧的数据
static Frame *frame_queue_peek_last(FrameQueue *f)
{
    return &f->queue[f->rindex];
}

`rindex_shown`

这两个方法的区别就在于rindex_shown。因为视频和音频都会在队列中保留一帧的数据。它会在在第一次调用frame_queue_next时，会将rindex_shown 进行初始化。视频和音频的线程，rindex_shown都会被刷新成1。在视频显示之前，会调用一次。音频播放之前，也会先调用一次。

static void frame_queue_next(FrameQueue *f)
{
    if (f->keep_last && !f->rindex_shown) {
        f->rindex_shown = 1;
        return;
    }
    frame_queue_unref_item(&f->queue[f->rindex]);
    if (++f->rindex == f->max_size)
        f->rindex = 0;
    SDL_LockMutex(f->mutex);
    f->size--;
    SDL_CondSignal(f->cond);
    SDL_UnlockMutex(f->mutex);
}

frame_queue_next除了第一次用于初始化。

其他时候，都是用于抛弃当前帧的情况下，用这个方法，快速进入下一个。
在显示之前也会调用。进入下一个rindex

视频的取

虽然会保留一个数据，但实际上，视频的读取，都是直接使用使用frame_queue_peek_last方法。（没有rindex_show）这里取得的数据，已经是上面rindex-1 的效果了。

音频的取

frame_queue_peek_readable 在音频解码的过程中，会在frame_queue_next方法之前，调用这个方法，确定是否有可读。获取可读的对象。

音频通过这个方法，获取可读.png

对比视频很奇怪？视频是没有锁住等待的过程的。如果没有，就过了。因为视频的处理是在eventloop线程。如果锁住的，就会让整个卡住？

视频获取可读的逻辑，如果没有，就直接跳过了.png

static Frame *frame_queue_peek_readable(FrameQueue *f)
{
    /* wait until we have a readable a new frame */
    //使用f->size - f->rindex_shown ，来判断当前还持有的量
    //第一次调用这个方法时，rindex_shown还为0
    SDL_LockMutex(f->mutex);
    while (f->size - f->rindex_shown <= 0 &&
           !f->pktq->abort_request) {
        SDL_CondWait(f->cond, f->mutex);
    }
    SDL_UnlockMutex(f->mutex);

    if (f->pktq->abort_request)
        return NULL;
  
    //得到的是
    return &f->queue[(f->rindex + f->rindex_shown) % f->max_size];
}

原因是在于，如果没有取到视频的数据，还会继续显示上一次的画面。而音频数据，则需要等待。

frame_queue_last_pos seek 的时候使用

/* return last shown position */
static int64_t frame_queue_last_pos(FrameQueue *f)
{
    Frame *fp = &f->queue[f->rindex];
    if (f->rindex_shown && fp->serial == f->pktq->serial)
        return fp->pos;
    else
        return -1;
}

frame_queue_peek_next 这个方法，用于在显示之前。判断队列中，是否还存在下一个，如果还存在下一个，而且事件已经超过了，就表示当前取到的这个，是不需要的。

在视频流中唯一一次用的地方，判断是否需要抛弃当前帧.png

//取出当前的下一个。
static Frame *frame_queue_peek_next(FrameQueue *f)
{
    return &f->queue[(f->rindex + f->rindex_shown + 1) % f->max_size];
}

总结一下FrameQueue的操作

存入的时候，统一使用。

frame_queue_peek_writable 得到一个可用的frame，对他进行操作
frame_queue_push 将可用的frame返回队列当中，同时偏移wrindex

音频的读取

frame_queue_peek_readable 使用这个方法读取，有可能阻塞
不需要阻塞的话，使用 frame_queue_peek_last
frame_queue_next 进行偏移。

锁的部分

在写入时获取可用的Frame用来写入，会完全锁住。在push时，进行size增加的时候，会锁住，但是windex的增加，是不会锁住的。
读取时同样，在获取可读的数据时，会完全锁住。同样的，在偏移角标的过程中，只锁住了size的变化。

本文参与腾讯云自媒体分享计划，分享自作者个人站点/博客。

原始发表：2018.11.26 ，如有侵权请联系 cloudcommunity@tencent.com 删除

其他

本文分享自作者个人站点/博客前往查看

如有侵权，请联系 cloudcommunity@tencent.com 删除。

本文参与腾讯云自媒体分享计划，欢迎热爱写作的你一起参与！

其他

登录后参与评论

0 条评论

热度