From 8702a1fa67a281d3c1c4fc66be44fce3b74c0a59 Mon Sep 17 00:00:00 2001 From: Adrian Fiergolski Date: Thu, 4 Nov 2021 12:59:21 +0100 Subject: [PATCH] avtp: cvf: extract AVTP VF payload base class Extract a part which could be common with the AVTP RVF payload plugin to a separate class. Part-of: --- .../gst-docs/symbols/symbol_index.json | 7 +- .../docs/plugins/gst_plugins_cache.json | 100 ++--- .../gst-plugins-bad/ext/avtp/gstavtpcvfpay.c | 337 ++-------------- .../gst-plugins-bad/ext/avtp/gstavtpcvfpay.h | 15 +- .../ext/avtp/gstavtpvfpaybase.c | 364 ++++++++++++++++++ .../ext/avtp/gstavtpvfpaybase.h | 73 ++++ .../gst-plugins-bad/ext/avtp/meson.build | 1 + 7 files changed, 534 insertions(+), 363 deletions(-) create mode 100644 subprojects/gst-plugins-bad/ext/avtp/gstavtpvfpaybase.c create mode 100644 subprojects/gst-plugins-bad/ext/avtp/gstavtpvfpaybase.h diff --git a/subprojects/gst-docs/symbols/symbol_index.json b/subprojects/gst-docs/symbols/symbol_index.json index 5befe93221..1806dfed71 100644 --- a/subprojects/gst-docs/symbols/symbol_index.json +++ b/subprojects/gst-docs/symbols/symbol_index.json @@ -7357,9 +7357,6 @@ "GstAvtpCvfDepay!src", "GstAvtpCvfPay", "GstAvtpCvfPay!sink", - "GstAvtpCvfPay:max-interval-frames", - "GstAvtpCvfPay:measurement-interval", - "GstAvtpCvfPay:mtu", "GstAvtpSink", "GstAvtpSink!sink", "GstAvtpSink:address", @@ -7369,6 +7366,10 @@ "GstAvtpSrc!src", "GstAvtpSrc:address", "GstAvtpSrc:ifname", + "GstAvtpVfPayBase", + "GstAvtpVfPayBase:mtu", + "GstAvtpVfPayBase:measurement-interval", + "GstAvtpVfPayBase:max-interval-frames", "GstBPMDetect", "GstBPMDetect!sink", "GstBPMDetect!src", diff --git a/subprojects/gst-plugins-bad/docs/plugins/gst_plugins_cache.json b/subprojects/gst-plugins-bad/docs/plugins/gst_plugins_cache.json index 92db72ff04..f262c444c9 100644 --- a/subprojects/gst-plugins-bad/docs/plugins/gst_plugins_cache.json +++ b/subprojects/gst-plugins-bad/docs/plugins/gst_plugins_cache.json @@ -3582,6 +3582,7 @@ "description": "Payload-encode compressed video into CVF AVTPDU (IEEE 1722)", "hierarchy": [ "GstAvtpCvfPay", + "GstAvtpVfPayBase", "GstAvtpBasePayload", "GstElement", "GstObject", @@ -3597,50 +3598,6 @@ "presence": "always" } }, - "properties": { - "max-interval-frames": { - "blurb": "Maximum number of network frames to be sent on each Measurement Interval", - "conditionally-available": false, - "construct": false, - "construct-only": false, - "controllable": false, - "default": "1", - "max": "-1", - "min": "1", - "mutable": "null", - "readable": true, - "type": "guint", - "writable": true - }, - "measurement-interval": { - "blurb": "Measurement interval of stream in nanoseconds", - "conditionally-available": false, - "construct": false, - "construct-only": false, - "controllable": false, - "default": "250000", - "max": "18446744073709551615", - "min": "0", - "mutable": "null", - "readable": true, - "type": "guint64", - "writable": true - }, - "mtu": { - "blurb": "Maximum Transit Unit (MTU) of underlying network in bytes", - "conditionally-available": false, - "construct": false, - "construct-only": false, - "controllable": false, - "default": "1500", - "max": "-1", - "min": "0", - "mutable": "null", - "readable": true, - "type": "guint", - "writable": true - } - }, "rank": "none" }, "avtpsink": { @@ -3917,6 +3874,61 @@ "writable": true } } + }, + "GstAvtpVfPayBase": { + "hierarchy": [ + "GstAvtpVfPayBase", + "GstAvtpBasePayload", + "GstElement", + "GstObject", + "GInitiallyUnowned", + "GObject" + ], + "kind": "object", + "properties": { + "max-interval-frames": { + "blurb": "Maximum number of network frames to be sent on each Measurement Interval", + "conditionally-available": false, + "construct": false, + "construct-only": false, + "controllable": false, + "default": "1", + "max": "-1", + "min": "1", + "mutable": "null", + "readable": true, + "type": "guint", + "writable": true + }, + "measurement-interval": { + "blurb": "Measurement interval of stream in nanoseconds", + "conditionally-available": false, + "construct": false, + "construct-only": false, + "controllable": false, + "default": "250000", + "max": "18446744073709551615", + "min": "0", + "mutable": "null", + "readable": true, + "type": "guint64", + "writable": true + }, + "mtu": { + "blurb": "Maximum Transit Unit (MTU) of underlying network in bytes", + "conditionally-available": false, + "construct": false, + "construct-only": false, + "controllable": false, + "default": "1500", + "max": "-1", + "min": "0", + "mutable": "null", + "readable": true, + "type": "guint", + "writable": true + } + } } }, "package": "GStreamer Bad Plug-ins", diff --git a/subprojects/gst-plugins-bad/ext/avtp/gstavtpcvfpay.c b/subprojects/gst-plugins-bad/ext/avtp/gstavtpcvfpay.c index 828b74e434..12f7590bdf 100644 --- a/subprojects/gst-plugins-bad/ext/avtp/gstavtpcvfpay.c +++ b/subprojects/gst-plugins-bad/ext/avtp/gstavtpcvfpay.c @@ -1,6 +1,8 @@ /* * GStreamer AVTP Plugin * Copyright (C) 2019 Intel Corporation + * Copyright (c) 2021, Fastree3D + * Adrian Fiergolski * * This library is free software; you can redistribute it and/or * modify it under the terms of the GNU Lesser General Public @@ -46,31 +48,19 @@ GST_DEBUG_CATEGORY_STATIC (avtpcvfpay_debug); /* prototypes */ -static GstFlowReturn gst_avtp_cvf_pay_chain (GstPad * pad, GstObject * parent, - GstBuffer * buffer); -static gboolean gst_avtp_cvf_pay_sink_event (GstPad * pad, GstObject * parent, - GstEvent * event); - -static void gst_avtp_cvf_set_property (GObject * object, guint prop_id, - const GValue * value, GParamSpec * pspec); -static void gst_avtp_cvf_get_property (GObject * object, guint prop_id, - GValue * value, GParamSpec * pspec); - static GstStateChangeReturn gst_avtp_cvf_change_state (GstElement * element, GstStateChange transition); +static gboolean gst_avtp_cvf_pay_new_caps (GstAvtpVfPayBase * avtpvfpaybase, + GstCaps * caps); +static gboolean gst_avtp_cvf_pay_prepare_avtp_packets (GstAvtpVfPayBase * + avtpvfpaybase, GstBuffer * buffer, GPtrArray * avtp_packets); + enum { PROP_0, - PROP_MTU, - PROP_MEASUREMENT_INTERVAL, - PROP_MAX_INTERVAL_FRAME }; -#define DEFAULT_MTU 1500 -#define DEFAULT_MEASUREMENT_INTERVAL 250000 -#define DEFAULT_MAX_INTERVAL_FRAMES 1 - #define AVTP_CVF_H264_HEADER_SIZE (sizeof(struct avtp_stream_pdu) + sizeof(guint32)) #define FU_A_TYPE 28 #define FU_A_HEADER_SIZE (sizeof(guint16)) @@ -96,17 +86,16 @@ static GstStaticPadTemplate sink_template = GST_STATIC_PAD_TEMPLATE ("sink", /* class initialization */ #define gst_avtp_cvf_pay_parent_class parent_class -G_DEFINE_TYPE (GstAvtpCvfPay, gst_avtp_cvf_pay, GST_TYPE_AVTP_BASE_PAYLOAD); +G_DEFINE_TYPE (GstAvtpCvfPay, gst_avtp_cvf_pay, GST_TYPE_AVTP_VF_PAY_BASE); GST_ELEMENT_REGISTER_DEFINE (avtpcvfpay, "avtpcvfpay", GST_RANK_NONE, GST_TYPE_AVTP_CVF_PAY); static void gst_avtp_cvf_pay_class_init (GstAvtpCvfPayClass * klass) { - GObjectClass *gobject_class = G_OBJECT_CLASS (klass); GstElementClass *gstelement_class = GST_ELEMENT_CLASS (klass); - GstAvtpBasePayloadClass *avtpbasepayload_class = - GST_AVTP_BASE_PAYLOAD_CLASS (klass); + GstAvtpVfPayBaseClass *avtpvfpaybase_class = + GST_AVTP_VF_PAY_BASE_CLASS (klass); gst_element_class_add_static_pad_template (gstelement_class, &sink_template); @@ -116,32 +105,12 @@ gst_avtp_cvf_pay_class_init (GstAvtpCvfPayClass * klass) "Payload-encode compressed video into CVF AVTPDU (IEEE 1722)", "Ederson de Souza "); - gobject_class->set_property = GST_DEBUG_FUNCPTR (gst_avtp_cvf_set_property); - gobject_class->get_property = GST_DEBUG_FUNCPTR (gst_avtp_cvf_get_property); - gstelement_class->change_state = GST_DEBUG_FUNCPTR (gst_avtp_cvf_change_state); - avtpbasepayload_class->chain = GST_DEBUG_FUNCPTR (gst_avtp_cvf_pay_chain); - avtpbasepayload_class->sink_event = - GST_DEBUG_FUNCPTR (gst_avtp_cvf_pay_sink_event); - - g_object_class_install_property (gobject_class, PROP_MTU, - g_param_spec_uint ("mtu", "Maximum Transit Unit", - "Maximum Transit Unit (MTU) of underlying network in bytes", 0, - G_MAXUINT, DEFAULT_MTU, G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS)); - - g_object_class_install_property (gobject_class, PROP_MEASUREMENT_INTERVAL, - g_param_spec_uint64 ("measurement-interval", "Measurement Interval", - "Measurement interval of stream in nanoseconds", 0, - G_MAXUINT64, DEFAULT_MEASUREMENT_INTERVAL, - G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS)); - - g_object_class_install_property (gobject_class, PROP_MAX_INTERVAL_FRAME, - g_param_spec_uint ("max-interval-frames", "Maximum Interval Frames", - "Maximum number of network frames to be sent on each Measurement Interval", - 1, G_MAXUINT, DEFAULT_MAX_INTERVAL_FRAMES, - G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS)); + avtpvfpaybase_class->new_caps = GST_DEBUG_FUNCPTR (gst_avtp_cvf_pay_new_caps); + avtpvfpaybase_class->prepare_avtp_packets = + GST_DEBUG_FUNCPTR (gst_avtp_cvf_pay_prepare_avtp_packets); GST_DEBUG_CATEGORY_INIT (avtpcvfpay_debug, "avtpcvfpay", 0, "debug category for avtpcvfpay element"); @@ -150,60 +119,8 @@ gst_avtp_cvf_pay_class_init (GstAvtpCvfPayClass * klass) static void gst_avtp_cvf_pay_init (GstAvtpCvfPay * avtpcvfpay) { - avtpcvfpay->mtu = DEFAULT_MTU; avtpcvfpay->header = NULL; avtpcvfpay->nal_length_size = 0; - avtpcvfpay->measurement_interval = DEFAULT_MEASUREMENT_INTERVAL; - avtpcvfpay->max_interval_frames = DEFAULT_MAX_INTERVAL_FRAMES; - avtpcvfpay->last_interval_ct = 0; -} - -static void -gst_avtp_cvf_set_property (GObject * object, guint prop_id, - const GValue * value, GParamSpec * pspec) -{ - GstAvtpCvfPay *avtpcvfpay = GST_AVTP_CVF_PAY (object); - - GST_DEBUG_OBJECT (avtpcvfpay, "prop_id: %u", prop_id); - - switch (prop_id) { - case PROP_MTU: - avtpcvfpay->mtu = g_value_get_uint (value); - break; - case PROP_MEASUREMENT_INTERVAL: - avtpcvfpay->measurement_interval = g_value_get_uint64 (value); - break; - case PROP_MAX_INTERVAL_FRAME: - avtpcvfpay->max_interval_frames = g_value_get_uint (value); - break; - default: - G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec); - break; - } -} - -static void -gst_avtp_cvf_get_property (GObject * object, guint prop_id, - GValue * value, GParamSpec * pspec) -{ - GstAvtpCvfPay *avtpcvfpay = GST_AVTP_CVF_PAY (object); - - GST_DEBUG_OBJECT (avtpcvfpay, "prop_id: %u", prop_id); - - switch (prop_id) { - case PROP_MTU: - g_value_set_uint (value, avtpcvfpay->mtu); - break; - case PROP_MEASUREMENT_INTERVAL: - g_value_set_uint64 (value, avtpcvfpay->measurement_interval); - break; - case PROP_MAX_INTERVAL_FRAME: - g_value_set_uint (value, avtpcvfpay->max_interval_frames); - break; - default: - G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec); - break; - } } static GstStateChangeReturn @@ -338,6 +255,7 @@ static GstBuffer * gst_avtpcvpay_fragment_nal (GstAvtpCvfPay * avtpcvfpay, GstBuffer * nal, gsize * offset, gboolean * last_fragment) { + GstAvtpVfPayBase *avtpvfpaybase = GST_AVTP_VF_PAY_BASE (avtpcvfpay); GstBuffer *fragment_header, *fragment; guint8 nal_header, nal_type, nal_nri, fu_indicator, fu_header; gsize available, nal_size, fragment_size, remaining; @@ -346,7 +264,8 @@ gst_avtpcvpay_fragment_nal (GstAvtpCvfPay * avtpcvfpay, GstBuffer * nal, nal_size = gst_buffer_get_size (nal); /* If NAL + header will be smaller than MTU, nothing to fragment */ - if (*offset == 0 && (nal_size + AVTP_CVF_H264_HEADER_SIZE) <= avtpcvfpay->mtu) { + if (*offset == 0 + && (nal_size + AVTP_CVF_H264_HEADER_SIZE) <= avtpvfpaybase->mtu) { *last_fragment = TRUE; *offset = nal_size; GST_DEBUG_OBJECT (avtpcvfpay, @@ -363,7 +282,7 @@ gst_avtpcvpay_fragment_nal (GstAvtpCvfPay * avtpcvfpay, GstBuffer * nal, /* Remaining size is smaller than MTU, so this is the last fragment */ remaining = nal_size - *offset + AVTP_CVF_H264_HEADER_SIZE + FU_A_HEADER_SIZE; - if (remaining <= avtpcvfpay->mtu) { + if (remaining <= avtpvfpaybase->mtu) { *last_fragment = TRUE; } @@ -388,7 +307,7 @@ gst_avtpcvpay_fragment_nal (GstAvtpCvfPay * avtpcvfpay, GstBuffer * nal, gst_buffer_unmap (fragment_header, &map); available = - avtpcvfpay->mtu - AVTP_CVF_H264_HEADER_SIZE - + avtpvfpaybase->mtu - AVTP_CVF_H264_HEADER_SIZE - gst_buffer_get_size (fragment_header); /* NAL unit header is not sent, but spread into FU indicator and header, @@ -411,123 +330,21 @@ gst_avtpcvpay_fragment_nal (GstAvtpCvfPay * avtpcvfpay, GstBuffer * nal, return fragment; } -static void -gst_avtp_cvf_pay_spread_ts (GstAvtpCvfPay * avtpcvfpay, - GPtrArray * avtp_packets) -{ - /* A bit of the idea of what this function do: - * - * After fragmenting the NAL unit, we have a series of AVTPDUs (AVTP Data Units) - * that should be transmitted. They are going to be transmitted according to GstBuffer - * DTS (or PTS in case there's no DTS), but all of them have the same DTS, as they - * came from the same original NAL unit. - * - * However, TSN streams should send their data according to a "measurement interval", - * which is an arbitrary interval defined for the stream. For instance, a class A - * stream has measurement interval of 125us. Also, there's a MaxIntervalFrames - * parameter, that defines how many network frames can be sent on a given measurement - * interval. We also spread MaxIntervalFrames per measurement interval. - * - * To that end, this function will spread the DTS so that fragments follow measurement - * interval and MaxIntervalFrames, adjusting them to end before the actual DTS of the - * original NAL unit. - * - * Roughly, this function does: - * - * DTSn = DTSbase - (measurement_interval/MaxIntervalFrames) * (total - n - 1) - * - * Where: - * DTSn = DTS of nth fragment - * DTSbase = DTS of original NAL unit - * total = # of fragments - * - * Another issue that this function takes care of is avoiding DTSs that overlap between - * two different set of fragments. Assuming DTSlast is the DTS of the last fragment - * generated on previous call to this function, we don't want any DTSn for the current - * call to be smaller than DTSlast + (measurement_interval / MaxIntervalFrames). If - * that's the case, we adjust DTSbase to preserve this difference (so we don't schedule - * packets transmission times that violate stream spec). This will cause the last - * fragment DTS to be bigger than DTSbase - we emit a warning, as this may be a sign - * of a bad pipeline setup or inappropriate stream spec. - * - * Finally, we also avoid underflows - which would occur when DTSbase is zero or small - * enough. In this case, we'll again make last fragment DTS > DTSbase, so we log it. - * - */ - - GstAvtpBasePayload *avtpbasepayload = GST_AVTP_BASE_PAYLOAD (avtpcvfpay); - - gint i, ret; - guint len; - guint64 tx_interval, total_interval; - GstClockTime base_time, base_dts, rt; - GstBuffer *packet; - - base_time = gst_element_get_base_time (GST_ELEMENT (avtpcvfpay)); - base_dts = GST_BUFFER_DTS (g_ptr_array_index (avtp_packets, 0)); - - tx_interval = - avtpcvfpay->measurement_interval / avtpcvfpay->max_interval_frames; - len = avtp_packets->len; - total_interval = tx_interval * (len - 1); - - /* We don't want packets transmission time to overlap, so let's ensure - * packets are scheduled after last interval used */ - if (avtpcvfpay->last_interval_ct != 0) { - GstClockTime dts_ct, dts_rt; - - ret = - gst_segment_to_running_time_full (&avtpbasepayload->segment, - GST_FORMAT_TIME, base_dts, &dts_rt); - if (ret == -1) - dts_rt = -dts_rt; - - dts_ct = base_time + dts_rt; - - if (dts_ct < avtpcvfpay->last_interval_ct + total_interval + tx_interval) { - base_dts += - avtpcvfpay->last_interval_ct + total_interval + tx_interval - dts_ct; - - GST_WARNING_OBJECT (avtpcvfpay, - "Not enough measurements intervals between frames to transmit fragments" - ". Check stream transmission spec."); - } - } - - /* Not enough room to spread tx before DTS (or we would underflow), - * add offset */ - if (total_interval > base_dts) { - base_dts += total_interval - base_dts; - - GST_INFO_OBJECT (avtpcvfpay, - "Not enough measurements intervals to transmit fragments before base " - "DTS. Check pipeline settings. Are we live?"); - } - - for (i = 0; i < len; i++) { - packet = g_ptr_array_index (avtp_packets, i); - GST_BUFFER_DTS (packet) = base_dts - tx_interval * (len - i - 1); - } - - /* Remember last interval used, in clock time */ - ret = - gst_segment_to_running_time_full (&avtpbasepayload->segment, - GST_FORMAT_TIME, GST_BUFFER_DTS (g_ptr_array_index (avtp_packets, - avtp_packets->len - 1)), &rt); - if (ret == -1) - rt = -rt; - avtpcvfpay->last_interval_ct = base_time + rt; -} - static gboolean -gst_avtp_cvf_pay_prepare_avtp_packets (GstAvtpCvfPay * avtpcvfpay, - GPtrArray * nals, GPtrArray * avtp_packets) +gst_avtp_cvf_pay_prepare_avtp_packets (GstAvtpVfPayBase * avtpvfpaybase, + GstBuffer * buffer, GPtrArray * avtp_packets) { - GstAvtpBasePayload *avtpbasepayload = GST_AVTP_BASE_PAYLOAD (avtpcvfpay); + GstAvtpBasePayload *avtpbasepayload = GST_AVTP_BASE_PAYLOAD (avtpvfpaybase); + GstAvtpCvfPay *avtpcvfpay = GST_AVTP_CVF_PAY (avtpvfpaybase); + GPtrArray *nals; GstBuffer *header, *nal; GstMapInfo map; gint i; + /* Get all NALs inside buffer */ + nals = g_ptr_array_new (); + gst_avtp_cvf_pay_extract_nals (avtpcvfpay, buffer, nals); + for (i = 0; i < nals->len; i++) { guint64 avtp_time, h264_time; gboolean last_fragment; @@ -626,71 +443,17 @@ gst_avtp_cvf_pay_prepare_avtp_packets (GstAvtpCvfPay * avtpcvfpay, gst_buffer_unref (nal); } - GST_LOG_OBJECT (avtpcvfpay, "Prepared %u AVTP packets", avtp_packets->len); + g_ptr_array_free (nals, TRUE); - /* Ensure DTS/PTS respect stream transmit spec, so PDUs are transmitted - * according to measurement interval. */ - if (avtp_packets->len > 0) - gst_avtp_cvf_pay_spread_ts (avtpcvfpay, avtp_packets); + GST_LOG_OBJECT (avtpcvfpay, "Prepared %u AVTP packets", avtp_packets->len); return TRUE; } -static GstFlowReturn -gst_avtp_cvf_pay_push_packets (GstAvtpCvfPay * avtpcvfpay, - GPtrArray * avtp_packets) -{ - int i; - GstFlowReturn ret; - GstAvtpBasePayload *avtpbasepayload = GST_AVTP_BASE_PAYLOAD (avtpcvfpay); - - for (i = 0; i < avtp_packets->len; i++) { - GstBuffer *packet; - - packet = g_ptr_array_index (avtp_packets, i); - ret = gst_pad_push (avtpbasepayload->srcpad, packet); - if (ret != GST_FLOW_OK) - return ret; - } - - return GST_FLOW_OK; -} - -static GstFlowReturn -gst_avtp_cvf_pay_chain (GstPad * pad, GstObject * parent, GstBuffer * buffer) -{ - GstAvtpBasePayload *avtpbasepayload = GST_AVTP_BASE_PAYLOAD (parent); - GstAvtpCvfPay *avtpcvfpay = GST_AVTP_CVF_PAY (avtpbasepayload); - GPtrArray *nals, *avtp_packets; - GstFlowReturn ret = GST_FLOW_OK; - - GST_LOG_OBJECT (avtpcvfpay, - "Incoming buffer size: %" G_GSIZE_FORMAT " PTS: %" GST_TIME_FORMAT - " DTS: %" GST_TIME_FORMAT, gst_buffer_get_size (buffer), - GST_TIME_ARGS (GST_BUFFER_PTS (buffer)), - GST_TIME_ARGS (GST_BUFFER_DTS (buffer))); - - /* Get all NALs inside buffer */ - nals = g_ptr_array_new (); - gst_avtp_cvf_pay_extract_nals (avtpcvfpay, buffer, nals); - - /* Prepare a list of avtp_packets to send */ - avtp_packets = g_ptr_array_new (); - gst_avtp_cvf_pay_prepare_avtp_packets (avtpcvfpay, nals, avtp_packets); - - ret = gst_avtp_cvf_pay_push_packets (avtpcvfpay, avtp_packets); - - /* Contents of both ptr_arrays should be unref'd or transferred - * to rightful owner by this point, no need to unref them again */ - g_ptr_array_free (nals, TRUE); - g_ptr_array_free (avtp_packets, TRUE); - - return ret; -} - static gboolean -gst_avtp_cvf_pay_new_caps (GstAvtpCvfPay * avtpcvfpay, GstCaps * caps) +gst_avtp_cvf_pay_new_caps (GstAvtpVfPayBase * avtpvfpaybase, GstCaps * caps) { + GstAvtpCvfPay *avtpcvfpay = GST_AVTP_CVF_PAY (avtpvfpaybase); const GValue *value; GstStructure *str; GstBuffer *buffer; @@ -730,41 +493,3 @@ error: gst_buffer_unmap (buffer, &map); return FALSE; } - -static gboolean -gst_avtp_cvf_pay_sink_event (GstPad * pad, GstObject * parent, GstEvent * event) -{ - GstCaps *caps; - GstAvtpBasePayload *avtpbasepayload = GST_AVTP_BASE_PAYLOAD (parent); - GstAvtpCvfPay *avtpcvfpay = GST_AVTP_CVF_PAY (avtpbasepayload); - gboolean ret; - - GST_DEBUG_OBJECT (avtpcvfpay, "Sink event %s", GST_EVENT_TYPE_NAME (event)); - - switch (GST_EVENT_TYPE (event)) { - case GST_EVENT_CAPS: - gst_event_parse_caps (event, &caps); - ret = gst_avtp_cvf_pay_new_caps (avtpcvfpay, caps); - gst_event_unref (event); - return ret; - case GST_EVENT_FLUSH_STOP: - if (GST_ELEMENT (avtpcvfpay)->current_state == GST_STATE_PLAYING) { - /* After a flush, the sink will reset pipeline base_time, but only - * after it gets the first buffer. So, here, we used the wrong - * base_time to calculate DTS. We'll just notice base_time changed - * when we get the next buffer. So, we'll basically mess with - * timestamps of two frames, which is bad. Known workaround is - * to pause the pipeline before a flushing seek - so that we'll - * be up to date to new pipeline base_time */ - GST_WARNING_OBJECT (avtpcvfpay, - "Flushing seek performed while pipeline is PLAYING, " - "AVTP timestamps will be incorrect!"); - } - break; - default: - break; - } - - return GST_AVTP_BASE_PAYLOAD_CLASS (parent_class)->sink_event (pad, parent, - event); -} diff --git a/subprojects/gst-plugins-bad/ext/avtp/gstavtpcvfpay.h b/subprojects/gst-plugins-bad/ext/avtp/gstavtpcvfpay.h index db23dd98a2..08d4e95107 100644 --- a/subprojects/gst-plugins-bad/ext/avtp/gstavtpcvfpay.h +++ b/subprojects/gst-plugins-bad/ext/avtp/gstavtpcvfpay.h @@ -1,6 +1,8 @@ /* * GStreamer AVTP Plugin * Copyright (C) 2019 Intel Corporation + * Copyright (c) 2021, Fastree3D + * Adrian Fiergolski * * This library is free software; you can redistribute it and/or * modify it under the terms of the GNU Lesser General Public @@ -24,10 +26,9 @@ #include -#include "gstavtpbasepayload.h" +#include "gstavtpvfpaybase.h" G_BEGIN_DECLS - #define GST_TYPE_AVTP_CVF_PAY (gst_avtp_cvf_pay_get_type()) #define GST_AVTP_CVF_PAY(obj) \ (G_TYPE_CHECK_INSTANCE_CAST((obj),GST_TYPE_AVTP_CVF_PAY,GstAvtpCvfPay)) @@ -37,19 +38,14 @@ G_BEGIN_DECLS (G_TYPE_CHECK_INSTANCE_TYPE((obj),GST_TYPE_AVTP_CVF_PAY)) #define GST_IS_AVTP_CVF_PAY_CLASS(klass) \ (G_TYPE_CHECK_CLASS_TYPE((klass),GST_TYPE_AVTP_CVF_PAY)) - typedef struct _GstAvtpCvfPay GstAvtpCvfPay; typedef struct _GstAvtpCvfPayClass GstAvtpCvfPayClass; struct _GstAvtpCvfPay { - GstAvtpBasePayload payload; + GstAvtpVfPayBase vfbase; GstBuffer *header; - guint mtu; - guint64 measurement_interval; - guint max_interval_frames; - guint64 last_interval_ct; /* H.264 specific information */ guint8 nal_length_size; @@ -57,7 +53,7 @@ struct _GstAvtpCvfPay struct _GstAvtpCvfPayClass { - GstAvtpBasePayloadClass parent_class; + GstAvtpVfPayBaseClass parent_class; }; GType gst_avtp_cvf_pay_get_type (void); @@ -65,5 +61,4 @@ GType gst_avtp_cvf_pay_get_type (void); GST_ELEMENT_REGISTER_DECLARE (avtpcvfpay); G_END_DECLS - #endif /* __GST_AVTP_CVF_PAY_H__ */ diff --git a/subprojects/gst-plugins-bad/ext/avtp/gstavtpvfpaybase.c b/subprojects/gst-plugins-bad/ext/avtp/gstavtpvfpaybase.c new file mode 100644 index 0000000000..1a2149156e --- /dev/null +++ b/subprojects/gst-plugins-bad/ext/avtp/gstavtpvfpaybase.c @@ -0,0 +1,364 @@ +/* + * GStreamer AVTP Plugin + * Copyright (C) 2019 Intel Corporation + * Copyright (c) 2021, Fastree3D + * Adrian Fiergolski + * + * This library is free software; you can redistribute it and/or + * modify it under the terms of the GNU Lesser General Public + * License as published by the Free Software Foundation; either + * version 2.1 of the License, or (at your option) any later + * version. + * + * This library is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU + * Lesser General Public License for more details. + * + * You should have received a copy of the GNU Lesser General Public + * License along with this library; if not, write to the + * Free Software Foundation, Inc., 51 Franklin Street, Fifth Floor, + * Boston, MA 02110-1301 USA + */ + +#include "gstavtpvfpaybase.h" + +GST_DEBUG_CATEGORY_STATIC (avtpvfpaybase_debug); +#define GST_CAT_DEFAULT avtpvfpaybase_debug + +/* prototypes */ +static GstFlowReturn gst_avtp_vf_pay_base_chain (GstPad * pad, + GstObject * parent, GstBuffer * buffer); +static gboolean gst_avtp_vf_pay_base_sink_event (GstPad * pad, + GstObject * parent, GstEvent * event); + +static void gst_avtp_rvf_set_property (GObject * object, guint prop_id, + const GValue * value, GParamSpec * pspec); +static void gst_avtp_rvf_get_property (GObject * object, guint prop_id, + GValue * value, GParamSpec * pspec); + +enum +{ + PROP_0, + PROP_MTU, + PROP_MEASUREMENT_INTERVAL, + PROP_MAX_INTERVAL_FRAME +}; + +#define DEFAULT_MTU 1500 +#define DEFAULT_MEASUREMENT_INTERVAL 250000 +#define DEFAULT_MAX_INTERVAL_FRAMES 1 + +#define gst_avtp_vf_pay_base_parent_class parent_class +G_DEFINE_TYPE_EXTENDED (GstAvtpVfPayBase, gst_avtp_vf_pay_base, + GST_TYPE_AVTP_BASE_PAYLOAD, G_TYPE_FLAG_ABSTRACT, { + }); + +/* class initialization */ + +static void +gst_avtp_vf_pay_base_class_init (GstAvtpVfPayBaseClass * klass) +{ + GObjectClass *gobject_class = G_OBJECT_CLASS (klass); + GstAvtpBasePayloadClass *avtpbasepayload_class = + GST_AVTP_BASE_PAYLOAD_CLASS (klass); + + gobject_class->set_property = GST_DEBUG_FUNCPTR (gst_avtp_rvf_set_property); + gobject_class->get_property = GST_DEBUG_FUNCPTR (gst_avtp_rvf_get_property); + + avtpbasepayload_class->chain = GST_DEBUG_FUNCPTR (gst_avtp_vf_pay_base_chain); + avtpbasepayload_class->sink_event = + GST_DEBUG_FUNCPTR (gst_avtp_vf_pay_base_sink_event); + + klass->new_caps = NULL; + klass->prepare_avtp_packets = NULL; + + g_object_class_install_property (gobject_class, PROP_MTU, + g_param_spec_uint ("mtu", "Maximum Transit Unit", + "Maximum Transit Unit (MTU) of underlying network in bytes", 0, + G_MAXUINT, DEFAULT_MTU, G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS)); + + g_object_class_install_property (gobject_class, PROP_MEASUREMENT_INTERVAL, + g_param_spec_uint64 ("measurement-interval", "Measurement Interval", + "Measurement interval of stream in nanoseconds", 0, + G_MAXUINT64, DEFAULT_MEASUREMENT_INTERVAL, + G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS)); + + g_object_class_install_property (gobject_class, PROP_MAX_INTERVAL_FRAME, + g_param_spec_uint ("max-interval-frames", "Maximum Interval Frames", + "Maximum number of network frames to be sent on each Measurement Interval", + 1, G_MAXUINT, DEFAULT_MAX_INTERVAL_FRAMES, + G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS)); + + GST_DEBUG_CATEGORY_INIT (avtpvfpaybase_debug, "avtpvfpaybase", + 0, "debug category for avtpvfpaybase element"); +} + +static void +gst_avtp_vf_pay_base_init (GstAvtpVfPayBase * avtpvfpaybase) +{ + avtpvfpaybase->mtu = DEFAULT_MTU; + avtpvfpaybase->measurement_interval = DEFAULT_MEASUREMENT_INTERVAL; + avtpvfpaybase->max_interval_frames = DEFAULT_MAX_INTERVAL_FRAMES; + avtpvfpaybase->last_interval_ct = 0; +} + +static void +gst_avtp_rvf_set_property (GObject * object, guint prop_id, + const GValue * value, GParamSpec * pspec) +{ + GstAvtpVfPayBase *avtpvfpaybase = GST_AVTP_VF_PAY_BASE (object); + + GST_DEBUG_OBJECT (avtpvfpaybase, "prop_id: %u", prop_id); + + switch (prop_id) { + case PROP_MTU: + avtpvfpaybase->mtu = g_value_get_uint (value); + break; + case PROP_MEASUREMENT_INTERVAL: + avtpvfpaybase->measurement_interval = g_value_get_uint64 (value); + break; + case PROP_MAX_INTERVAL_FRAME: + avtpvfpaybase->max_interval_frames = g_value_get_uint (value); + break; + default: + G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec); + break; + } +} + +static void +gst_avtp_rvf_get_property (GObject * object, guint prop_id, + GValue * value, GParamSpec * pspec) +{ + GstAvtpVfPayBase *avtpvfpaybase = GST_AVTP_VF_PAY_BASE (object); + + GST_DEBUG_OBJECT (avtpvfpaybase, "prop_id: %u", prop_id); + + switch (prop_id) { + case PROP_MTU: + g_value_set_uint (value, avtpvfpaybase->mtu); + break; + case PROP_MEASUREMENT_INTERVAL: + g_value_set_uint64 (value, avtpvfpaybase->measurement_interval); + break; + case PROP_MAX_INTERVAL_FRAME: + g_value_set_uint (value, avtpvfpaybase->max_interval_frames); + break; + default: + G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec); + break; + } +} + +static void +gst_avtp_vf_pay_base_spread_ts (GstAvtpVfPayBase * avtpvfpaybase, + GPtrArray * avtp_packets) +{ + /* A bit of the idea of what this function do: + * + * After fragmenting the buffert, we have a series of AVTPDUs (AVTP Data Units) + * that should be transmitted. They are going to be transmitted according to GstBuffer + * DTS (or PTS in case there's no DTS), but all of them have the same PTS/DTS, as they + * came from the same original buffer. + * + * However, TSN streams should send their data according to a "measurement interval", + * which is an arbitrary interval defined for the stream. For instance, a class A + * stream has measurement interval of 125us. Also, there's a MaxIntervalFrames + * parameter, that defines how many network frames can be sent on a given measurement + * interval. We also spread MaxIntervalFrames per measurement interval. + * + * To that end, this function will spread the DTS/PTS so that fragments follow measurement + * interval and MaxIntervalFrames, adjusting them to end before the actual DTS/PTS of the + * original buffer. + * + * Roughly, this function does: + * + * DTSn = DTSbase - (measurement_interval/MaxIntervalFrames) * (total - n - 1) + * + * Where: + * DTSn = DTS/PTS of nth fragment + * DTSbase = DTS/PTS of original buffer + * total = # of fragments + * + * Another issue that this function takes care of is avoiding DTSs/PTSs that overlap between + * two different set of fragments. Assuming DTSlast/PTSlast is the DTS/PTS of the last fragment + * generated on previous call to this function, we don't want any DTSn for the current + * call to be smaller than DTSlast + (measurement_interval / MaxIntervalFrames). If + * that's the case, we adjust DTSbase to preserve this difference (so we don't schedule + * packets transmission times that violate stream spec). This will cause the last + * fragment DTS to be bigger than DTSbase - we emit a warning, as this may be a sign + * of a bad pipeline setup or inappropriate stream spec. + * + * Finally, we also avoid underflows - which would occur when DTSbase is zero or small + * enough. In this case, we'll again make last fragment DTS > DTSbase, so we log it. + * + */ + + GstAvtpBasePayload *avtpbasepayload = GST_AVTP_BASE_PAYLOAD (avtpvfpaybase); + + gint i, ret; + guint len; + guint64 tx_interval, total_interval; + GstClockTime base_time, base_ts, rt; + GstBuffer *packet; + + base_time = gst_element_get_base_time (GST_ELEMENT (avtpvfpaybase)); + base_ts = GST_BUFFER_DTS_OR_PTS (g_ptr_array_index (avtp_packets, 0)); + + tx_interval = + avtpvfpaybase->measurement_interval / avtpvfpaybase->max_interval_frames; + len = avtp_packets->len; + total_interval = tx_interval * (len - 1); + + /* We don't want packets transmission time to overlap, so let's ensure + * packets are scheduled after last interval used */ + if (avtpvfpaybase->last_interval_ct != 0) { + GstClockTime ts_ct, ts_rt; + + ret = + gst_segment_to_running_time_full (&avtpbasepayload->segment, + GST_FORMAT_TIME, base_ts, &ts_rt); + if (ret == -1) + ts_rt = -ts_rt; + + ts_ct = base_time + ts_rt; + + if (ts_ct < avtpvfpaybase->last_interval_ct + total_interval + tx_interval) { + base_ts += + avtpvfpaybase->last_interval_ct + total_interval + tx_interval - + ts_ct; + + GST_WARNING_OBJECT (avtpvfpaybase, + "Not enough measurements intervals between frames to transmit fragments" + ". Check stream transmission spec."); + } + } + + /* Not enough room to spread tx before TS (or we would underflow), + * add offset */ + if (total_interval > base_ts) { + base_ts += total_interval - base_ts; + + GST_INFO_OBJECT (avtpvfpaybase, + "Not enough measurements intervals to transmit fragments before base " + "DTS/PTS. Check pipeline settings. Are we live?"); + } + + for (i = 0; i < len; i++) { + GstClockTime *packet_ts; + packet = g_ptr_array_index (avtp_packets, i); + packet_ts = + GST_BUFFER_DTS_IS_VALID (packet) ? &GST_BUFFER_DTS (packet) : + &GST_BUFFER_PTS (packet); + *packet_ts = base_ts - tx_interval * (len - i - 1); + } + + /* Remember last interval used, in clock time */ + ret = + gst_segment_to_running_time_full (&avtpbasepayload->segment, + GST_FORMAT_TIME, GST_BUFFER_DTS_OR_PTS (g_ptr_array_index (avtp_packets, + avtp_packets->len - 1)), &rt); + if (ret == -1) + rt = -rt; + avtpvfpaybase->last_interval_ct = base_time + rt; +} + +static GstFlowReturn +gst_avtp_vf_pay_base_push_packets (GstAvtpVfPayBase * avtpvfpaybase, + GPtrArray * avtp_packets) +{ + int i; + GstFlowReturn ret; + GstAvtpBasePayload *avtpbasepayload = GST_AVTP_BASE_PAYLOAD (avtpvfpaybase); + + for (i = 0; i < avtp_packets->len; i++) { + GstBuffer *packet; + + packet = g_ptr_array_index (avtp_packets, i); + ret = gst_pad_push (avtpbasepayload->srcpad, packet); + if (ret != GST_FLOW_OK) + return ret; + } + + return GST_FLOW_OK; +} + +static GstFlowReturn +gst_avtp_vf_pay_base_chain (GstPad * pad, GstObject * parent, + GstBuffer * buffer) +{ + GstAvtpBasePayload *avtpbasepayload = GST_AVTP_BASE_PAYLOAD (parent); + GstAvtpVfPayBase *avtpvfpaybase = GST_AVTP_VF_PAY_BASE (avtpbasepayload); + GPtrArray *avtp_packets; + GstFlowReturn ret = GST_FLOW_OK; + + GST_LOG_OBJECT (avtpvfpaybase, + "Incoming buffer size: %" G_GSIZE_FORMAT " PTS: %" GST_TIME_FORMAT + " DTS: %" GST_TIME_FORMAT, gst_buffer_get_size (buffer), + GST_TIME_ARGS (GST_BUFFER_PTS (buffer)), + GST_TIME_ARGS (GST_BUFFER_DTS (buffer))); + + /* Prepare a list of avtp_packets to send */ + avtp_packets = g_ptr_array_new (); + + g_assert (GST_AVTP_VF_PAY_BASE_GET_CLASS (avtpvfpaybase)->prepare_avtp_packets + != NULL); + GST_AVTP_VF_PAY_BASE_GET_CLASS (avtpvfpaybase)->prepare_avtp_packets + (avtpvfpaybase, buffer, avtp_packets); + + if (avtp_packets->len > 0) + gst_avtp_vf_pay_base_spread_ts (avtpvfpaybase, avtp_packets); + + ret = gst_avtp_vf_pay_base_push_packets (avtpvfpaybase, avtp_packets); + + /* Contents of both ptr_arrays should be unref'd or transferred + * to rightful owner by this point, no need to unref them again */ + g_ptr_array_free (avtp_packets, TRUE); + + return ret; +} + +static gboolean +gst_avtp_vf_pay_base_sink_event (GstPad * pad, GstObject * parent, + GstEvent * event) +{ + GstCaps *caps; + GstAvtpBasePayload *avtpbasepayload = GST_AVTP_BASE_PAYLOAD (parent); + GstAvtpVfPayBase *avtpvfpaybase = GST_AVTP_VF_PAY_BASE (avtpbasepayload); + gboolean ret; + + GST_DEBUG_OBJECT (avtpvfpaybase, "Sink event %s", + GST_EVENT_TYPE_NAME (event)); + + switch (GST_EVENT_TYPE (event)) { + case GST_EVENT_CAPS: + gst_event_parse_caps (event, &caps); + g_assert (GST_AVTP_VF_PAY_BASE_GET_CLASS (avtpvfpaybase)->new_caps != + NULL); + ret = + GST_AVTP_VF_PAY_BASE_GET_CLASS (avtpvfpaybase)->new_caps + (avtpvfpaybase, caps); + gst_event_unref (event); + return ret; + case GST_EVENT_FLUSH_STOP: + if (GST_ELEMENT (avtpvfpaybase)->current_state == GST_STATE_PLAYING) { + /* After a flush, the sink will reset pipeline base_time, but only + * after it gets the first buffer. So, here, we used the wrong + * base_time to calculate DTS. We'll just notice base_time changed + * when we get the next buffer. So, we'll basically mess with + * timestamps of two frames, which is bad. Known workaround is + * to pause the pipeline before a flushing seek - so that we'll + * be up to date to new pipeline base_time */ + GST_WARNING_OBJECT (avtpvfpaybase, + "Flushing seek performed while pipeline is PLAYING, " + "AVTP timestamps will be incorrect!"); + } + break; + default: + break; + } + + return GST_AVTP_BASE_PAYLOAD_CLASS (parent_class)->sink_event (pad, parent, + event); +} diff --git a/subprojects/gst-plugins-bad/ext/avtp/gstavtpvfpaybase.h b/subprojects/gst-plugins-bad/ext/avtp/gstavtpvfpaybase.h new file mode 100644 index 0000000000..e7eb99bd0f --- /dev/null +++ b/subprojects/gst-plugins-bad/ext/avtp/gstavtpvfpaybase.h @@ -0,0 +1,73 @@ +/* + * GStreamer AVTP Plugin + * Copyright (C) 2019 Intel Corporation + * Copyright (c) 2021, Fastree3D + * Adrian Fiergolski + * + * This library is free software; you can redistribute it and/or + * modify it under the terms of the GNU Lesser General Public + * License as published by the Free Software Foundation; either + * version 2.1 of the License, or (at your option) any later + * version. + * + * This library is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU + * Lesser General Public License for more details. + * + * You should have received a copy of the GNU Lesser General Public + * License along with this library; if not, write to the + * Free Software Foundation, Inc., 51 Franklin Street, Fifth Floor, + * Boston, MA 02110-1301 USA + */ + +#ifndef __GST_AVTP_PAY_BASE_H__ +#define __GST_AVTP_PAY_BASE_H__ + +#include + +#include "gstavtpbasepayload.h" + +G_BEGIN_DECLS +#define GST_TYPE_AVTP_VF_PAY_BASE (gst_avtp_vf_pay_base_get_type()) +#define GST_AVTP_VF_PAY_BASE(obj) \ + (G_TYPE_CHECK_INSTANCE_CAST((obj),GST_TYPE_AVTP_VF_PAY_BASE,GstAvtpVfPayBase)) +#define GST_AVTP_VF_PAY_BASE_CLASS(klass) \ + (G_TYPE_CHECK_CLASS_CAST((klass),GST_TYPE_AVTP_VF_PAY_BASE,GstAvtpVfPayBaseClass)) +#define GST_IS_AVTP_VF_PAY_BASE(obj) \ + (G_TYPE_CHECK_INSTANCE_TYPE((obj),GST_TYPE_AVTP_VF_PAY_BASE)) +#define GST_IS_AVTP_VF_PAY_BASE_CLASS(klass) \ + (G_TYPE_CHECK_CLASS_TYPE((klass),GST_TYPE_AVTP_VF_PAY_BASE)) +#define GST_AVTP_VF_PAY_BASE_GET_CLASS(obj) \ + (G_TYPE_INSTANCE_GET_CLASS ((obj), GST_AVTP_VF_PAY_BASE, GstAvtpVfPayBaseClass)) +typedef struct _GstAvtpVfPayBase GstAvtpVfPayBase; +typedef struct _GstAvtpVfPayBaseClass GstAvtpVfPayBaseClass; + +typedef gboolean (*GstAvtpVfPayNewCapsFunction) (GstAvtpVfPayBase * + avtpvfpaybase, GstCaps * caps); +typedef gboolean (*GstAvtpVfPayPrepareAvtpPacketsFunction) (GstAvtpVfPayBase * + avtprvfpaybase, GstBuffer * buffer, GPtrArray * avtp_packets); + +struct _GstAvtpVfPayBase +{ + GstAvtpBasePayload payload; + + guint mtu; + guint64 measurement_interval; + guint max_interval_frames; + guint64 last_interval_ct; +}; + +struct _GstAvtpVfPayBaseClass +{ + GstAvtpBasePayloadClass parent_class; + + /* Pure virtual function. */ + GstAvtpVfPayNewCapsFunction new_caps; + GstAvtpVfPayPrepareAvtpPacketsFunction prepare_avtp_packets; +}; + +GType gst_avtp_vf_pay_base_get_type (void); + +G_END_DECLS +#endif /* __GST_AVTP_PAY_BASE_H__ */ diff --git a/subprojects/gst-plugins-bad/ext/avtp/meson.build b/subprojects/gst-plugins-bad/ext/avtp/meson.build index 2ede892718..05e1be72c7 100644 --- a/subprojects/gst-plugins-bad/ext/avtp/meson.build +++ b/subprojects/gst-plugins-bad/ext/avtp/meson.build @@ -4,6 +4,7 @@ avtp_sources = [ 'gstavtpaafpay.c', 'gstavtpcvfdepay.c', 'gstavtpcvfpay.c', + 'gstavtpvfpaybase.c', 'gstavtpbasedepayload.c', 'gstavtpbasepayload.c', 'gstavtpsink.c',