PSVAMB-69845: [Dreamforce 2024][ESTT] Buffering and Errors on Live Streams During Extended Streaming-Time Test

nginx-kmp-out-module/src/ngx_kmp_out_track.c

@@ -1210,6 +1210,62 @@ ngx_kmp_out_track_write(ngx_kmp_out_track_t *track, u_char *data,
 }
 
 
+static ngx_int_t
+ngx_kmp_out_track_log_media_info(ngx_kmp_out_track_t *track)
+{
+    u_char              *buf = NULL, *p;
+
+    if(track->extra_data.len > 0)
+    {
+        buf = ngx_pnalloc(track->pool, 2 * track->extra_data.len + 1);
+        if (buf == NULL)
+        {
+            ngx_log_error(NGX_LOG_ERR, &track->log, 0,
+                  "ngx_kmp_out_track_log_media_info:"
+                  "failed to allocate buffer for extra data size %uD",
+                  2 * track->extra_data.len + 1);
+            return NGX_ERROR;
+        }
+
+        p = ngx_hex_dump(buf, track->extra_data.data, track->extra_data.len);
+        *p = 0;
+    }
+
+    ngx_log_error(NGX_LOG_INFO, &track->log, 0,
+        "ngx_kmp_out_track_log_media_info:"
+        " media_type: %uD codec_id:   %uD"
+        " timescale:  %uD bitrate:    %uD"
+        " extra_data: %s",
+        track->media_info.media_type, track->media_info.codec_id,
+        track->media_info.timescale,  track->media_info.bitrate,  buf);
+    switch (track->media_info.media_type)
+    {
+       case KMP_MEDIA_VIDEO:
+           ngx_log_error(NGX_LOG_INFO, &track->log, 0,
+               "ngx_kmp_out_track_log_media_info: video"
+               " width: %uD height: %uD frame_rate: %uD/%uD cea_captions: %s",
+               track->media_info.u.video.width, track->media_info.u.video.height,
+               track->media_info.u.video.frame_rate.num, track->media_info.u.video.frame_rate.denom,
+               track->media_info.u.video.cea_captions ? "true" : "false");
+            break;
+       case KMP_MEDIA_AUDIO:
+           ngx_log_error(NGX_LOG_INFO, &track->log, 0,
+               "ngx_kmp_out_track_log_media_info: audio"
+               " channels: %uD bits_per_sample: %uD sample_rate: %uD channel_layout: %uD",
+               track->media_info.u.audio.channels, track->media_info.u.audio.bits_per_sample,
+               track->media_info.u.audio.sample_rate, track->media_info.u.audio.channel_layout);
+            break;
+    };
+
+    if (buf)
+    {
+        ngx_pfree(track->pool, buf);
+    }
+
+    return NGX_OK;
+}
+
+
 ngx_int_t
 ngx_kmp_out_track_write_media_info(ngx_kmp_out_track_t *track)
 {
@@ -1238,6 +1294,8 @@ ngx_kmp_out_track_write_media_info(ngx_kmp_out_track_t *track)
         return NGX_ERROR;
     }
 
+    ngx_kmp_out_track_log_media_info(track);
+
     return NGX_OK;
 }
 

nginx-rtmp-module/src/ngx_rtmp_codec_module.c

@@ -1456,6 +1456,7 @@ ngx_rtmp_codec_postconfiguration(ngx_conf_t *cf)
     ngx_rtmp_core_main_conf_t          *cmcf;
     ngx_rtmp_handler_pt                *h;
     ngx_rtmp_amf_handler_t             *ch;
+    ngx_uint_t                          nelts;
 
     cmcf = ngx_rtmp_conf_get_module_main_conf(cf, ngx_rtmp_core_module);
 
@@ -1464,13 +1465,23 @@ ngx_rtmp_codec_postconfiguration(ngx_conf_t *cf)
         return NGX_ERROR;
     }
 
+    nelts = cmcf->events[NGX_RTMP_MSG_AUDIO].nelts;
+    for ( ; nelts > 1; nelts--, h--) {
+        *h =  h[-1];
+    }
+
     *h = ngx_rtmp_codec_av;
 
     h = ngx_array_push(&cmcf->events[NGX_RTMP_MSG_VIDEO]);
     if (h == NULL) {
         return NGX_ERROR;
     }
 
+    nelts = cmcf->events[NGX_RTMP_MSG_VIDEO].nelts;
+    for (; nelts > 1; nelts--, h--) {
+        *h =  h[-1];
+    }
+
     *h = ngx_rtmp_codec_av;
 
     h = ngx_array_push(&cmcf->events[NGX_RTMP_DISCONNECT]);

transcoder/KMP/KMP.c

@@ -781,3 +781,67 @@ bool KMP_read_ack(KMP_session_t *context,uint64_t* frame_id)
     }
     return false;
 }
+
+static const char *hex = "0123456789ABCDEF";
+
+int KMP_log_mediainfo(KMP_session_t *context,
+    const char *category, int level,
+    transcode_mediaInfo_t* transcodeMediaInfo) {
+
+    AVCodecParameters* params = transcodeMediaInfo->codecParams;
+
+    if(!params){
+        return -1;
+    }
+
+    char *ex = NULL;
+    if(params->extradata_size > 0 ){
+        ex = av_malloc(2 * params->extradata_size + 1);
+        char *walk = ex;
+        for(int i = 0; i < params->extradata_size; i++) {
+           *walk++ = hex[params->extradata[i] >> 4];
+           *walk++ = hex[params->extradata[i] & 0x0f];
+        }
+        *walk = '\0';
+    }
+
+    if (params->codec_type == AVMEDIA_TYPE_AUDIO) {
+
+        LOGGER(category,level,"[%s] KMP_PACKET_MEDIA_INFO type=audio"
+            " codec_id=%s (0x%x) samplerate=%d bps=%d channels=%d channel_layout=%d"
+            " bitrate=%.3f kbps timescale=%d:%d"
+            " extra_data=%s",
+            context->sessionName,
+            avcodec_get_name(params->codec_id),
+            params->codec_tag,
+            params->sample_rate,
+            params->bits_per_coded_sample,
+            params->channels,
+            params->channel_layout,
+            params->bit_rate / 1024.0, transcodeMediaInfo->timeScale.num, transcodeMediaInfo->timeScale.den,
+            ex);
+    }
+    else if (params->codec_type == AVMEDIA_TYPE_VIDEO) {
+
+        LOGGER(category,level,"[%s] KMP_PACKET_MEDIA_INFO type=video"
+            " codec_id=%s (0x%x) width=%d height=%d frame_rate=%d:%d"
+            " bitrate=%.3f kbps timescale=%d:%d"
+            " cc=%s extra_data=%s",
+            context->sessionName,
+            avcodec_get_name(params->codec_id),
+            params->codec_tag,
+            params->width,
+            params->height,
+            transcodeMediaInfo->frameRate.num, transcodeMediaInfo->frameRate.den,
+            params->bit_rate / 1024.0, transcodeMediaInfo->timeScale.num, transcodeMediaInfo->timeScale.den,
+            transcodeMediaInfo->closed_captions ? "yes" : "no",
+            ex);
+    }
+
+    if(ex) {
+        av_free(ex);
+    }
+
+    return 0;
+}
+

transcoder/KMP/KMP.h

@@ -50,6 +50,8 @@ int KMP_send_ack( KMP_session_t *context,kmp_frame_position_t *cur_pos);
 
 int KMP_close( KMP_session_t *context);
 
+int KMP_log_mediainfo(KMP_session_t *context, const char *category,int level,     transcode_mediaInfo_t* mediaInfo);
+
 
 int KMP_listen( KMP_session_t *context);
 int KMP_accept( KMP_session_t *context, KMP_session_t *client);

transcoder/ackHandler/ackHandlerInternal.cpp

@@ -7,7 +7,7 @@
             try {
                am.addIn(*desc);
              } catch(const std::exception &e) {
-                   LOGGER(LoggingCategory,AV_LOG_ERROR," %s audio map. ack_map_add_input %d %lld failed due to %s",
+                   LOGGER(LoggingCategory,AV_LOG_ERROR," %s map. ack_map_add_input %lld %lld failed due to %s",
                         am.m_name.c_str(),desc->id, desc->pts,e.what());
              }
        }
@@ -20,7 +20,7 @@
                 try {
                    am.addFiltered(*desc);
                  } catch(const std::exception &e) {
-                       LOGGER(LoggingCategory,AV_LOG_ERROR," %s audio map. ack_map_add_filtered %d %lld failed due to %s",
+                       LOGGER(LoggingCategory,AV_LOG_ERROR," %s map. ack_map_add_filtered %lld %lld failed due to %s",
                             am.m_name.c_str(),desc->id, desc->pts,e.what());
                  }
            }
@@ -33,7 +33,7 @@
              try {
                 am.addOut(*desc);
             } catch(const std::exception &e) {
-                  LOGGER(LoggingCategory,AV_LOG_ERROR," %s audio map. ack_map_add_output %d %lld failed due to %s",
+                  LOGGER(LoggingCategory,AV_LOG_ERROR," %s map. ack_map_add_output %lld %lld failed due to %s",
                     am.m_name.c_str(),desc->id, desc->pts,e.what());
             }
         }
@@ -48,7 +48,7 @@
               am.map(ack,*ao);
               return;
             } catch(const std::exception &e) {
-                 LOGGER(LoggingCategory,AV_LOG_ERROR," %s audio map. ack_map_ack %lld failed due to %s",
+                 LOGGER(LoggingCategory,AV_LOG_ERROR," %s map. ack_map_ack %lld failed due to %s",
                         am.m_name.c_str(),ack, e.what());
             }
         }

transcoder/common/json_parser.h

@@ -115,8 +115,8 @@ typedef struct {
 #define JSON_SERIALIZE_ARRAY_END()        JSON_WRITE("]"); js->shouldAddComma=true;
 
 #define JSON_SERIALIZE_STRING(key,value)  ADD_COMMA() JSON_WRITE("\"%s\": \"%s\"",key,value); js->shouldAddComma=true;
-#define JSON_SERIALIZE_INT(key,value)     ADD_COMMA() JSON_WRITE("\"%s\": %d",key,value); js->shouldAddComma=true;
-#define JSON_SERIALIZE_INT64(key,value)   ADD_COMMA() JSON_WRITE("\"%s\": %lld",key,value); js->shouldAddComma=true;
+#define JSON_SERIALIZE_INT(key,value)     ADD_COMMA() JSON_WRITE("\"%s\": %d",key,(int)value); js->shouldAddComma=true;
+#define JSON_SERIALIZE_INT64(key,value)   ADD_COMMA() JSON_WRITE("\"%s\": %ld",key,(int64_t)value); js->shouldAddComma=true;
 #define JSON_SERIALIZE_BOOL(key,value)    ADD_COMMA() JSON_WRITE("\"%s\": %s",key,value ? "true" : "false"); js->shouldAddComma=true;
 #define JSON_SERIALIZE_DOUBLE(key,value)  ADD_COMMA() JSON_WRITE("\"%s\": %.2lf",key,value); js->shouldAddComma=true;
 

transcoder/common/vector.h

@@ -20,7 +20,6 @@ typedef struct  {
 
 void vector_init(vector_t *);
 int vector_total(vector_t *);
-static void vector_resize(vector_t *, int);
 void vector_add(vector_t *, void *);
 void vector_set(vector_t *, int, void *);
 void *vector_get(vector_t *, int);

transcoder/core.h

@@ -5,7 +5,13 @@
 
 #include "libs.h"
 
-#define _S(EXP) if ((EXP)<0) { return EXP;}
+#define _S(EXP) {           \
+    int retVal = (EXP);     \
+    if(retVal < 0) {        \
+        return retVal;      \
+    }                       \
+}
+
 #ifndef u_char
 #define u_char  unsigned char
 #endif

transcoder/debug/file_streamer.c

@@ -57,6 +57,10 @@ void* thread_stream_from_file(void *vargp)
 
     json_get_int64(GetConfig(),"input.hiccupIntervalSec",0,&hiccupIntervalSec);
 
+    int64_t resendMediaInfoIntervalSec;
+    json_get_int64(GetConfig(),"input.resendMediaInfoIntervalSec",0,&resendMediaInfoIntervalSec);
+
+
     AVPacket packet;
     av_init_packet(&packet);
 
@@ -94,7 +98,11 @@ void* thread_stream_from_file(void *vargp)
     int64_t start_time=av_gettime_relative(),
             hiccup_duration =  hiccupDurationSec * 1000 * 1000,
             hiccup_interval = hiccupIntervalSec * 1000 * 1000,
-            next_hiccup = start_time + hiccup_interval;
+            next_hiccup = start_time + hiccup_interval,
+            resendMediaInfoInterval = resendMediaInfoIntervalSec * 1000 * 1000,
+            nextSentMediaInfoTime = av_gettime_relative() + resendMediaInfoInterval;
+
+    uint64_t frame_id_ack = createTime - 1;
 
     samples_stats_t stats;
     sample_stats_init(&stats,standard_timebase);
@@ -173,6 +181,23 @@ void* thread_stream_from_file(void *vargp)
             }
         }
 
+        if(resendMediaInfoIntervalSec > 0
+            && (packet.flags & AV_PKT_FLAG_KEY)
+            && nextSentMediaInfoTime < av_gettime_relative())
+        {
+            LOGGER0(CATEGORY_RECEIVER,AV_LOG_WARNING,"sending mediainfo");
+            if (KMP_send_mediainfo(&kmp,&extra)<0) {
+                LOGGER0(CATEGORY_RECEIVER,AV_LOG_FATAL,"couldn't send mediainfo!");
+                break;
+            }
+
+            if (KMP_read_ack(&kmp,&frame_id_ack)) {
+                LOGGER(CATEGORY_RECEIVER,AV_LOG_DEBUG,"received ack for packet id  %lld",frame_id_ack);
+            }
+
+            nextSentMediaInfoTime = av_gettime_relative() + resendMediaInfoInterval;
+        }
+
         lastDts=packet.dts;
 
         samples_stats_add(&stats,packet.dts,packet.pos,packet.size);
@@ -188,20 +213,21 @@ void* thread_stream_from_file(void *vargp)
                rate)*/
 
 
-        uint64_t frame_id_ack;
+
         if (KMP_send_packet(&kmp,&packet)<0) {
             LOGGER0(CATEGORY_RECEIVER,AV_LOG_FATAL,"couldn't send packet!");
             break;
         }
+        frame_id++;
         if (KMP_read_ack(&kmp,&frame_id_ack)) {
             LOGGER(CATEGORY_RECEIVER,AV_LOG_DEBUG,"received ack for packet id  %lld",frame_id_ack);
         }
 
 
-         LOGGER("SENDER",AV_LOG_DEBUG,"sent packet pts=%s dts=%s  size=%d",
+         LOGGER("SENDER",AV_LOG_DEBUG,"sent packet pts=%s dts=%s  size=%d frame_id=%lld frame_id_ack=%lld",
          ts2str(packet.pts,true),
          ts2str(packet.dts,true),
-         packet.size);
+         packet.size, frame_id, frame_id_ack);
 
 
         av_packet_unref(&packet);

transcoder/receiver_server.c

@@ -107,10 +107,31 @@ int clientLoop(receiver_server_t *server,receiver_server_session_t *session,tran
                 LOGGER(CATEGORY_RECEIVER,AV_LOG_FATAL,"[%s] Invalid mediainfo",session->stream_name);
                 break;
             }
-            LOGGER(CATEGORY_RECEIVER,AV_LOG_INFO,"[%s] received packet  KMP_PACKET_MEDIA_INFO",session->stream_name);
 
-            transcode_session_async_set_mediaInfo(transcode_session, newParams);
-        }
+            KMP_log_mediainfo(&session->kmpClient, CATEGORY_RECEIVER, AV_LOG_INFO, newParams);
+
+            if( (retVal = transcode_session_async_set_mediaInfo(transcode_session, newParams)) < 0) {
+                LOGGER(CATEGORY_RECEIVER,AV_LOG_ERROR,"[%s] transcode_session_async_set_mediaInfo failed",session->stream_name);
+
+                LOGGER(CATEGORY_RECEIVER,AV_LOG_INFO,"[%s] flushing",session->stream_name);
+                _S(transcode_session_async_send_packet(transcode_session, NULL));
+
+                if(!autoAckMode) {
+                    transcode_session_get_ack_frame_id(transcode_session,&current_position);
+                    // TODO: we cheat here, received_frame_id may have not been persisted yet
+                    // so we should ensure all of the frames up to received_frame_id
+                    // inclusively, are consumed by upstream
+                    current_position.frame_id = current_position.transcoded_frame_id = received_frame_id;
+                    LOGGER(CATEGORY_RECEIVER,AV_LOG_INFO,"[%s] sending ack for frame position: %lld, %lld, %lld",
+                        session->stream_name,
+                        current_position.frame_id,
+                        current_position.transcoded_frame_id,
+                        current_position.offset);
+                    _S(KMP_send_ack(&session->kmpClient,&current_position));
+                }
+                break;
+            }
+         }
         if (header.packet_type==KMP_PACKET_FRAME)
         {
             AVPacket* packet=av_packet_alloc();