Skip to content

Commit

Permalink
Browse files Browse the repository at this point in the history
tcpclientsrc: Expose connection stats as property
Unfortunately the OS takes care of bad connections for us, so we can't
get the stats in a platform-independent way. Count total bytes received
as well, platform-independently.
  • Loading branch information
vivia committed Feb 28, 2020
1 parent a12f161 commit ea930a4
Show file tree
Hide file tree
Showing 3 changed files with 87 additions and 1 deletion.
83 changes: 82 additions & 1 deletion gst/tcp/gsttcpclientsrc.c
Expand Up @@ -43,6 +43,16 @@
#include "config.h"
#endif

/* macOS and iOS have the .h files but the tcp_info struct is private API */
#if defined(HAVE_NETINET_TCP_H) && defined(HAVE_NETINET_IN_H) && defined(HAVE_SYS_SOCKET_H)
#include <netinet/tcp.h>
#include <netinet/in.h>
#include <sys/socket.h>
#if defined(TCP_INFO)
#define HAVE_SOCKET_METRIC_HEADERS
#endif
#endif

#include <gst/gst-i18n-plugin.h>
#include "gsttcpclientsrc.h"
#include "gsttcp.h"
Expand All @@ -65,7 +75,8 @@ enum
PROP_0,
PROP_HOST,
PROP_PORT,
PROP_TIMEOUT
PROP_TIMEOUT,
PROP_STATS,
};

#define gst_tcp_client_src_parent_class parent_class
Expand All @@ -88,6 +99,7 @@ static void gst_tcp_client_src_set_property (GObject * object, guint prop_id,
const GValue * value, GParamSpec * pspec);
static void gst_tcp_client_src_get_property (GObject * object, guint prop_id,
GValue * value, GParamSpec * pspec);
static GstStructure *gst_tcp_client_src_get_stats (GstTCPClientSrc * src);

static void
gst_tcp_client_src_class_init (GstTCPClientSrcClass * klass)
Expand Down Expand Up @@ -128,6 +140,30 @@ gst_tcp_client_src_class_init (GstTCPClientSrcClass * klass)
G_MAXUINT, TCP_DEFAULT_TIMEOUT,
G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));

/**
* GstTCPClientSrc::stats:
*
* Sends a GstStructure with statistics. We count bytes-received in a
* platform-independent way and the rest via the tcp_info struct, if it's
* available. The OS takes care of the TCP layer for us so we can't know it
* from here.
*
* Struct members:
*
* bytes-received (uint64): Total bytes received (platform-independent)
* reordering (uint): Amount of reordering (linux-specific)
* unacked (uint): Un-acked packets (linux-specific)
* sacked (uint): Selective acked packets (linux-specific)
* lost (uint): Lost packets (linux-specific)
* retrans (uint): Retransmits (linux-specific)
* fackets (uint): Forward acknowledgement (linux-specific)
*
* Since: 1.18
*/
g_object_class_install_property (gobject_class, PROP_STATS,
g_param_spec_boxed ("stats", "Stats", "Retrieve a statistics structure",
GST_TYPE_STRUCTURE, G_PARAM_READABLE | G_PARAM_STATIC_STRINGS));

gst_element_class_add_static_pad_template (gstelement_class, &srctemplate);

gst_element_class_set_static_metadata (gstelement_class,
Expand Down Expand Up @@ -172,6 +208,7 @@ gst_tcp_client_src_finalize (GObject * gobject)
this->socket = NULL;
g_free (this->host);
this->host = NULL;
gst_clear_structure (&this->stats);

G_OBJECT_CLASS (parent_class)->finalize (gobject);
}
Expand Down Expand Up @@ -278,6 +315,7 @@ gst_tcp_client_src_create (GstPushSrc * psrc, GstBuffer ** outbuf)
ret = GST_FLOW_OK;
gst_buffer_unmap (*outbuf, &map);
gst_buffer_resize (*outbuf, 0, rret);
src->bytes_received += read;

GST_LOG_OBJECT (src,
"Returning buffer from _get of size %" G_GSIZE_FORMAT ", ts %"
Expand Down Expand Up @@ -363,6 +401,9 @@ gst_tcp_client_src_get_property (GObject * object, guint prop_id,
case PROP_TIMEOUT:
g_value_set_uint (value, tcpclientsrc->timeout);
break;
case PROP_STATS:
g_value_take_boxed (value, gst_tcp_client_src_get_stats (tcpclientsrc));
break;
default:
G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
break;
Expand All @@ -379,6 +420,9 @@ gst_tcp_client_src_start (GstBaseSrc * bsrc)
GSocketAddress *saddr;
GResolver *resolver;

src->bytes_received = 0;
gst_clear_structure (&src->stats);

/* look up name if we need to */
addr = g_inet_address_new_from_string (src->host);
if (!addr) {
Expand Down Expand Up @@ -480,6 +524,8 @@ gst_tcp_client_src_stop (GstBaseSrc * bsrc)
if (src->socket) {
GST_DEBUG_OBJECT (src, "closing socket");

src->stats = gst_tcp_client_src_get_stats (src);

if (!g_socket_close (src->socket, &err)) {
GST_ERROR_OBJECT (src, "Failed to close socket: %s", err->message);
g_clear_error (&err);
Expand Down Expand Up @@ -517,3 +563,38 @@ gst_tcp_client_src_unlock_stop (GstBaseSrc * bsrc)

return TRUE;
}

static GstStructure *
gst_tcp_client_src_get_stats (GstTCPClientSrc * src)
{
GstStructure *s;

/* we can't get the values post stop so just return the saved ones */
if (src->stats)
return gst_structure_copy (src->stats);

s = gst_structure_new ("GstTCPClientSrcStats",
"bytes-received", G_TYPE_UINT64, src->bytes_received, NULL);

#ifdef HAVE_SOCKET_METRIC_HEADERS
if (src->socket) {
struct tcp_info info;
socklen_t info_len = sizeof info;
int fd;

fd = g_socket_get_fd (src->socket);

if (getsockopt (fd, IPPROTO_TCP, TCP_INFO, &info, &info_len) == 0) {
gst_structure_set (s,
"reordering", G_TYPE_UINT, info.tcpi_reordering,
"unacked", G_TYPE_UINT, info.tcpi_unacked,
"sacked", G_TYPE_UINT, info.tcpi_sacked,
"lost", G_TYPE_UINT, info.tcpi_lost,
"retrans", G_TYPE_UINT, info.tcpi_retrans,
"fackets", G_TYPE_UINT, info.tcpi_fackets, NULL);
}
}
#endif

return s;
}
3 changes: 3 additions & 0 deletions gst/tcp/gsttcpclientsrc.h
Expand Up @@ -60,6 +60,9 @@ struct _GstTCPClientSrc {
/* socket */
GSocket *socket;
GCancellable *cancellable;

guint64 bytes_received;
GstStructure *stats;
};

struct _GstTCPClientSrcClass {
Expand Down
2 changes: 2 additions & 0 deletions meson.build
Expand Up @@ -112,6 +112,8 @@ check_headers = [
['HAVE_EMMINTRIN_H', 'emmintrin.h'],
['HAVE_INTTYPES_H', 'inttypes.h'],
['HAVE_MEMORY_H', 'memory.h'],
['HAVE_NETINET_IN_H', 'netinet/in.h'],
['HAVE_NETINET_TCP_H', 'netinet/tcp.h'],
['HAVE_PROCESS_H', 'process.h'],
['HAVE_SMMINTRIN_H', 'smmintrin.h'],
['HAVE_STDINT_H', 'stdint.h'],
Expand Down

0 comments on commit ea930a4

Please sign in to comment.