History

avdevice/decklink: move AVPacketQueue into decklink_common and rename it to DecklinkPacketQueue

Move the AVPacketQueue functionality that is currently only used for the decklink decode module into decklink_common, so it can be shared by the decklink encoder (i.e. for VANC insertion when we receive data packets separate from video). The threadsafe queue used within the decklink module was named "AVPacketQueue" which implies that it is part of the public API, which it is not. Rename the functions and the name of the queue struct to make clear it is used exclusively by decklink, per Marton Balint's suggestion.

Devin Heitmueller committed 3 years ago

Commit 4fd729a1 parent 4373dad3 files · +133 −125Browse files

Unified · Split

3 files
  1. libavdevice/decklink_common.cpp +115 −0
  2. libavdevice/decklink_common.h +10 −3
  3. libavdevice/decklink_dec.cpp +8 −122
libavdevice/decklink_common.cpp+115−0
@@ -390,6 +390,121 @@ int ff_decklink_set_format(AVFormatContext *avctx, decklink_direction_t directio · 389 unmodified lines
390 return ff_decklink_set_format(avctx, 0, 0, 0, 0, AV_FIELD_UNKNOWN, direction);390 return ff_decklink_set_format(avctx, 0, 0, 0, 0, AV_FIELD_UNKNOWN, direction);
391}391}
392 392
393void ff_decklink_packet_queue_init(AVFormatContext *avctx, DecklinkPacketQueue *q)
394{
395 struct decklink_cctx *ctx = (struct decklink_cctx *)avctx->priv_data;
396 memset(q, 0, sizeof(DecklinkPacketQueue));
397 pthread_mutex_init(&q->mutex, NULL);
398 pthread_cond_init(&q->cond, NULL);
399 q->avctx = avctx;
400 q->max_q_size = ctx->queue_size;
401}
402
403void ff_decklink_packet_queue_flush(DecklinkPacketQueue *q)
404{
405 PacketListEntry *pkt, *pkt1;
406
407 pthread_mutex_lock(&q->mutex);
408 for (pkt = q->pkt_list.head; pkt != NULL; pkt = pkt1) {
409 pkt1 = pkt->next;
410 av_packet_unref(&pkt->pkt);
411 av_freep(&pkt);
412 }
413 q->pkt_list.head = NULL;
414 q->pkt_list.tail = NULL;
415 q->nb_packets = 0;
416 q->size = 0;
417 pthread_mutex_unlock(&q->mutex);
418}
419
420void ff_decklink_packet_queue_end(DecklinkPacketQueue *q)
421{
422 ff_decklink_packet_queue_flush(q);
423 pthread_mutex_destroy(&q->mutex);
424 pthread_cond_destroy(&q->cond);
425}
426
427unsigned long long ff_decklink_packet_queue_size(DecklinkPacketQueue *q)
428{
429 unsigned long long size;
430 pthread_mutex_lock(&q->mutex);
431 size = q->size;
432 pthread_mutex_unlock(&q->mutex);
433 return size;
434}
435
436int ff_decklink_packet_queue_put(DecklinkPacketQueue *q, AVPacket *pkt)
437{
438 PacketListEntry *pkt1;
439
440 // Drop Packet if queue size is > maximum queue size
441 if (ff_decklink_packet_queue_size(q) > (uint64_t)q->max_q_size) {
442 av_packet_unref(pkt);
443 av_log(q->avctx, AV_LOG_WARNING, "Decklink input buffer overrun!\n");
444 return -1;
445 }
446 /* ensure the packet is reference counted */
447 if (av_packet_make_refcounted(pkt) < 0) {
448 av_packet_unref(pkt);
449 return -1;
450 }
451
452 pkt1 = (PacketListEntry *)av_malloc(sizeof(*pkt1));
453 if (!pkt1) {
454 av_packet_unref(pkt);
455 return -1;
456 }
457 av_packet_move_ref(&pkt1->pkt, pkt);
458 pkt1->next = NULL;
459
460 pthread_mutex_lock(&q->mutex);
461
462 if (!q->pkt_list.tail) {
463 q->pkt_list.head = pkt1;
464 } else {
465 q->pkt_list.tail->next = pkt1;
466 }
467
468 q->pkt_list.tail = pkt1;
469 q->nb_packets++;
470 q->size += pkt1->pkt.size + sizeof(*pkt1);
471
472 pthread_cond_signal(&q->cond);
473
474 pthread_mutex_unlock(&q->mutex);
475 return 0;
476}
477
478int ff_decklink_packet_queue_get(DecklinkPacketQueue *q, AVPacket *pkt, int block)
479{
480 int ret;
481
482 pthread_mutex_lock(&q->mutex);
483
484 for (;; ) {
485 PacketListEntry *pkt1 = q->pkt_list.head;
486 if (pkt1) {
487 q->pkt_list.head = pkt1->next;
488 if (!q->pkt_list.head) {
489 q->pkt_list.tail = NULL;
490 }
491 q->nb_packets--;
492 q->size -= pkt1->pkt.size + sizeof(*pkt1);
493 *pkt = pkt1->pkt;
494 av_free(pkt1);
495 ret = 1;
496 break;
497 } else if (!block) {
498 ret = 0;
499 break;
500 } else {
501 pthread_cond_wait(&q->cond, &q->mutex);
502 }
503 }
504 pthread_mutex_unlock(&q->mutex);
505 return ret;
506}
507
393int ff_decklink_list_devices(AVFormatContext *avctx,508int ff_decklink_list_devices(AVFormatContext *avctx,
394 struct AVDeviceInfoList *device_list,509 struct AVDeviceInfoList *device_list,
395 int show_inputs, int show_outputs)510 int show_inputs, int show_outputs)
libavdevice/decklink_common.h+10−3
@@ -78,7 +78,7 @@ static char *dup_cfstring_to_utf8(CFStringRef w) · 77 unmodified lines
78class decklink_output_callback;78class decklink_output_callback;
79class decklink_input_callback;79class decklink_input_callback;
80 80
81typedef struct AVPacketQueue {81typedef struct DecklinkPacketQueue {
82 PacketList pkt_list;82 PacketList pkt_list;
83 int nb_packets;83 int nb_packets;
84 unsigned long long size;84 unsigned long long size;
@@ -87,7 +87,7 @@ typedef struct AVPacketQueue { · 2 unmodified lines
87 pthread_cond_t cond;87 pthread_cond_t cond;
88 AVFormatContext *avctx;88 AVFormatContext *avctx;
89 int64_t max_q_size;89 int64_t max_q_size;
90} AVPacketQueue;90} DecklinkPacketQueue;
91 91
92struct decklink_ctx {92struct decklink_ctx {
93 /* DeckLink SDK interfaces */93 /* DeckLink SDK interfaces */
@@ -111,7 +111,7 @@ struct decklink_ctx { · 17 unmodified lines
111 int supports_vanc;111 int supports_vanc;
112 112
113 /* Capture buffer queue */113 /* Capture buffer queue */
114 AVPacketQueue queue;114 DecklinkPacketQueue queue;
115 115
116 AVCCFifo *cc_fifo; ///< closed captions116 AVCCFifo *cc_fifo; ///< closed captions
117 117
@@ -235,4 +235,11 @@ int ff_decklink_list_formats(AVFormatContext *avctx, decklink_direction_t direct · 117 unmodified lines
235void ff_decklink_cleanup(AVFormatContext *avctx);235void ff_decklink_cleanup(AVFormatContext *avctx);
236int ff_decklink_init_device(AVFormatContext *avctx, const char* name);236int ff_decklink_init_device(AVFormatContext *avctx, const char* name);
237 237
238void ff_decklink_packet_queue_init(AVFormatContext *avctx, DecklinkPacketQueue *q);
239void ff_decklink_packet_queue_flush(DecklinkPacketQueue *q);
240void ff_decklink_packet_queue_end(DecklinkPacketQueue *q);
241unsigned long long ff_decklink_packet_queue_size(DecklinkPacketQueue *q);
242int ff_decklink_packet_queue_put(DecklinkPacketQueue *q, AVPacket *pkt);
243int ff_decklink_packet_queue_get(DecklinkPacketQueue *q, AVPacket *pkt, int block);
244
238#endif /* AVDEVICE_DECKLINK_COMMON_H */245#endif /* AVDEVICE_DECKLINK_COMMON_H */
libavdevice/decklink_dec.cpp+8−122
@@ -471,120 +471,6 @@ skip_packet: · 470 unmodified lines
471 return tgt;471 return tgt;
472}472}
473 473
474static void avpacket_queue_init(AVFormatContext *avctx, AVPacketQueue *q)
475{
476 struct decklink_cctx *ctx = (struct decklink_cctx *)avctx->priv_data;
477 memset(q, 0, sizeof(AVPacketQueue));
478 pthread_mutex_init(&q->mutex, NULL);
479 pthread_cond_init(&q->cond, NULL);
480 q->avctx = avctx;
481 q->max_q_size = ctx->queue_size;
482}
483
484static void avpacket_queue_flush(AVPacketQueue *q)
485{
486 PacketListEntry *pkt, *pkt1;
487
488 pthread_mutex_lock(&q->mutex);
489 for (pkt = q->pkt_list.head; pkt != NULL; pkt = pkt1) {
490 pkt1 = pkt->next;
491 av_packet_unref(&pkt->pkt);
492 av_freep(&pkt);
493 }
494 q->pkt_list.head = NULL;
495 q->pkt_list.tail = NULL;
496 q->nb_packets = 0;
497 q->size = 0;
498 pthread_mutex_unlock(&q->mutex);
499}
500
501static void avpacket_queue_end(AVPacketQueue *q)
502{
503 avpacket_queue_flush(q);
504 pthread_mutex_destroy(&q->mutex);
505 pthread_cond_destroy(&q->cond);
506}
507
508static unsigned long long avpacket_queue_size(AVPacketQueue *q)
509{
510 unsigned long long size;
511 pthread_mutex_lock(&q->mutex);
512 size = q->size;
513 pthread_mutex_unlock(&q->mutex);
514 return size;
515}
516
517static int avpacket_queue_put(AVPacketQueue *q, AVPacket *pkt)
518{
519 PacketListEntry *pkt1;
520
521 // Drop Packet if queue size is > maximum queue size
522 if (avpacket_queue_size(q) > (uint64_t)q->max_q_size) {
523 av_packet_unref(pkt);
524 av_log(q->avctx, AV_LOG_WARNING, "Decklink input buffer overrun!\n");
525 return -1;
526 }
527 /* ensure the packet is reference counted */
528 if (av_packet_make_refcounted(pkt) < 0) {
529 av_packet_unref(pkt);
530 return -1;
531 }
532
533 pkt1 = (PacketListEntry *)av_malloc(sizeof(*pkt1));
534 if (!pkt1) {
535 av_packet_unref(pkt);
536 return -1;
537 }
538 av_packet_move_ref(&pkt1->pkt, pkt);
539 pkt1->next = NULL;
540
541 pthread_mutex_lock(&q->mutex);
542
543 if (!q->pkt_list.tail) {
544 q->pkt_list.head = pkt1;
545 } else {
546 q->pkt_list.tail->next = pkt1;
547 }
548
549 q->pkt_list.tail = pkt1;
550 q->nb_packets++;
551 q->size += pkt1->pkt.size + sizeof(*pkt1);
552
553 pthread_cond_signal(&q->cond);
554
555 pthread_mutex_unlock(&q->mutex);
556 return 0;
557}
558
559static int avpacket_queue_get(AVPacketQueue *q, AVPacket *pkt, int block)
560{
561 int ret;
562
563 pthread_mutex_lock(&q->mutex);
564
565 for (;; ) {
566 PacketListEntry *pkt1 = q->pkt_list.head;
567 if (pkt1) {
568 q->pkt_list.head = pkt1->next;
569 if (!q->pkt_list.head) {
570 q->pkt_list.tail = NULL;
571 }
572 q->nb_packets--;
573 q->size -= pkt1->pkt.size + sizeof(*pkt1);
574 *pkt = pkt1->pkt;
575 av_free(pkt1);
576 ret = 1;
577 break;
578 } else if (!block) {
579 ret = 0;
580 break;
581 } else {
582 pthread_cond_wait(&q->cond, &q->mutex);
583 }
584 }
585 pthread_mutex_unlock(&q->mutex);
586 return ret;
587}
588 474
589static void handle_klv(AVFormatContext *avctx, decklink_ctx *ctx, IDeckLinkVideoInputFrame *videoFrame, int64_t pts)475static void handle_klv(AVFormatContext *avctx, decklink_ctx *ctx, IDeckLinkVideoInputFrame *videoFrame, int64_t pts)
590{476{
@@ -682,7 +568,7 @@ static void handle_klv(AVFormatContext *avctx, decklink_ctx *ctx, IDeckLinkVideo · 91 unmodified lines
682 klv_packet.data = klv.data();568 klv_packet.data = klv.data();
683 klv_packet.size = klv.size();569 klv_packet.size = klv.size();
684 570
685 if (avpacket_queue_put(&ctx->queue, &klv_packet) < 0) {571 if (ff_decklink_packet_queue_put(&ctx->queue, &klv_packet) < 0) {
686 ++ctx->dropped;572 ++ctx->dropped;
687 }573 }
688 }574 }
@@ -874,7 +760,7 @@ HRESULT decklink_input_callback::VideoInputFrameArrived( · 185 unmodified lines
874 if (videoFrame) {760 if (videoFrame) {
875 AVPacket pkt = { 0 };761 AVPacket pkt = { 0 };
876 if (ctx->frameCount % 25 == 0) {762 if (ctx->frameCount % 25 == 0) {
877 unsigned long long qsize = avpacket_queue_size(&ctx->queue);763 unsigned long long qsize = ff_decklink_packet_queue_size(&ctx->queue);
878 av_log(avctx, AV_LOG_DEBUG,764 av_log(avctx, AV_LOG_DEBUG,
879 "Frame received (#%lu) - Valid (%liB) - QSize %fMB\n",765 "Frame received (#%lu) - Valid (%liB) - QSize %fMB\n",
880 ctx->frameCount,766 ctx->frameCount,
@@ -1038,7 +924,7 @@ HRESULT decklink_input_callback::VideoInputFrameArrived( · 157 unmodified lines
1038 txt_pkt.stream_index = ctx->teletext_st->index;924 txt_pkt.stream_index = ctx->teletext_st->index;
1039 txt_pkt.data = txt_buf0;925 txt_pkt.data = txt_buf0;
1040 txt_pkt.size = txt_buf - txt_buf0;926 txt_pkt.size = txt_buf - txt_buf0;
1041 if (avpacket_queue_put(&ctx->queue, &txt_pkt) < 0) {927 if (ff_decklink_packet_queue_put(&ctx->queue, &txt_pkt) < 0) {
1042 ++ctx->dropped;928 ++ctx->dropped;
1043 }929 }
1044 }930 }
@@ -1049,7 +935,7 @@ HRESULT decklink_input_callback::VideoInputFrameArrived( · 4 unmodified lines
1049 if (pkt.buf)935 if (pkt.buf)
1050 videoFrame->AddRef();936 videoFrame->AddRef();
1051 937
1052 if (avpacket_queue_put(&ctx->queue, &pkt) < 0) {938 if (ff_decklink_packet_queue_put(&ctx->queue, &pkt) < 0) {
1053 ++ctx->dropped;939 ++ctx->dropped;
1054 }940 }
1055 }941 }
@@ -1071,7 +957,7 @@ HRESULT decklink_input_callback::VideoInputFrameArrived( · 15 unmodified lines
1071 pkt.stream_index = ctx->audio_st->index;957 pkt.stream_index = ctx->audio_st->index;
1072 pkt.data = (uint8_t *)audioFrameBytes;958 pkt.data = (uint8_t *)audioFrameBytes;
1073 959
1074 if (avpacket_queue_put(&ctx->queue, &pkt) < 0) {960 if (ff_decklink_packet_queue_put(&ctx->queue, &pkt) < 0) {
1075 ++ctx->dropped;961 ++ctx->dropped;
1076 }962 }
1077 }963 }
@@ -1153,7 +1039,7 @@ av_cold int ff_decklink_read_close(AVFormatContext *avctx) · 75 unmodified lines
1153 }1039 }
1154 1040
1155 ff_decklink_cleanup(avctx);1041 ff_decklink_cleanup(avctx);
1156 avpacket_queue_end(&ctx->queue);1042 ff_decklink_packet_queue_end(&ctx->queue);
1157 1043
1158 av_freep(&cctx->ctx);1044 av_freep(&cctx->ctx);
1159 1045
@@ -1411,7 +1297,7 @@ av_cold int ff_decklink_read_header(AVFormatContext *avctx) · 251 unmodified lines
1411 goto error;1297 goto error;
1412 }1298 }
1413 1299
1414 avpacket_queue_init (avctx, &ctx->queue);1300 ff_decklink_packet_queue_init(avctx, &ctx->queue);
1415 1301
1416 if (ctx->dli->StartStreams() != S_OK) {1302 if (ctx->dli->StartStreams() != S_OK) {
1417 av_log(avctx, AV_LOG_ERROR, "Cannot start input stream\n");1303 av_log(avctx, AV_LOG_ERROR, "Cannot start input stream\n");
@@ -1431,7 +1317,7 @@ int ff_decklink_read_packet(AVFormatContext *avctx, AVPacket *pkt) · 13 unmodified lines
1431 struct decklink_cctx *cctx = (struct decklink_cctx *)avctx->priv_data;1317 struct decklink_cctx *cctx = (struct decklink_cctx *)avctx->priv_data;
1432 struct decklink_ctx *ctx = (struct decklink_ctx *)cctx->ctx;1318 struct decklink_ctx *ctx = (struct decklink_ctx *)cctx->ctx;
1433 1319
1434 avpacket_queue_get(&ctx->queue, pkt, 1);1320 ff_decklink_packet_queue_get(&ctx->queue, pkt, 1);
1435 1321
1436 if (ctx->tc_format && !(av_dict_get(ctx->video_st->metadata, "timecode", NULL, 0))) {1322 if (ctx->tc_format && !(av_dict_get(ctx->video_st->metadata, "timecode", NULL, 0))) {
1437 size_t size;1323 size_t size;

Keyboard shortcuts

General

Show keyboard shortcuts
?
Focus the current search
/

Issues and pull requests

Open filters
f

Issues, pull requests and Inbox

Select next item
j↓
Select previous item
k↑
Open selected item
Enter
Dismiss filters or clear selection
Esc

Repository code

Go to file
t