mirrors/gstreamer
mirrors/gstreamer
1
1
Fork 0
mirror of https://gitlab.freedesktop.org/gstreamer/gstreamer.git synced 2024-10-06 02:32:23 +00:00
Code Issues 1 Projects Releases Wiki Activity
gstreamer/libs/gst/net/gstnetclientclock.c
Sebastian Dröge 558c0b97fc netclientclock: Make the clock a wrapper clock around an internal clock 
The internal clock is only used for slaving against the remote clock, while
the user-facing GstClock can be additionally slaved to another clock if
desired. By default, if no master clock is set, this has exactly the same
behaviour as before. If a master clock is set (which was not allowed before),
the user-facing clock is reporting the remote clock as internal time and
slaves this to the master clock.

This also removes the weirdness that the internal time of the netclientclock
was always the system clock time, and not the remote clock time.

https://bugzilla.gnome.org/show_bug.cgi?id=750574
2015-06-09 10:03:49 +02:00

1098 lines
37 KiB
C
Raw Blame History

/* GStreamer
* Copyright (C) 1999,2000 Erik Walthinsen <omega@cse.ogi.edu>
* 2005 Wim Taymans <wim@fluendo.com>
* 2005 Andy Wingo <wingo@pobox.com>
* Copyright (C) 2012 Collabora Ltd. <tim.muller@collabora.co.uk>
* Copyright (C) 2015 Sebastian Dröge <sebastian@centricular.com>
*
* gstnetclientclock.h: clock that synchronizes itself to a time provider over
* the network
*
* This library is free software; you can redistribute it and/or
* modify it under the terms of the GNU Library General Public
* License as published by the Free Software Foundation; either
* version 2 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
* Library General Public License for more details.
*
* You should have received a copy of the GNU Library General Public
* License along with this library; if not, write to the
* Free Software Foundation, Inc., 51 Franklin St, Fifth Floor,
* Boston, MA 02110-1301, USA.
*/
/**
* SECTION:gstnetclientclock
* @short_description: Special clock that synchronizes to a remote time
* provider.
* @see_also: #GstClock, #GstNetTimeProvider, #GstPipeline
*
* #GstNetClientClock implements a custom #GstClock that synchronizes its time
* to a remote time provider such as #GstNetTimeProvider. #GstNtpClock
* implements a #GstClock that synchronizes its time to a remote NTPv4 server.
*
* A new clock is created with gst_net_client_clock_new() or
* gst_ntp_clock_new(), which takes the address and port of the remote time
* provider along with a name and an initial time.
*
* This clock will poll the time provider and will update its calibration
* parameters based on the local and remote observations.
*
* The "round-trip" property limits the maximum round trip packets can take.
*
* Various parameters of the clock can be configured with the parent #GstClock
* "timeout", "window-size" and "window-threshold" object properties.
*
* A #GstNetClientClock and #GstNtpClock is typically set on a #GstPipeline with
* gst_pipeline_use_clock().
*
* If you set a #GstBus on the clock via the "bus" object property, it will
* send @GST_MESSAGE_ELEMENT messages with an attached #GstStructure containing
* statistics about clock accuracy and network traffic.
*/
#ifdef HAVE_CONFIG_H
#include "config.h"
#endif
#include "gstnettimepacket.h"
#include "gstntppacket.h"
#include "gstnetclientclock.h"
#include <gio/gio.h>
#include <string.h>
GST_DEBUG_CATEGORY_STATIC (ncc_debug);
#define GST_CAT_DEFAULT (ncc_debug)
#define DEFAULT_ADDRESS "127.0.0.1"
#define DEFAULT_PORT 5637
#define DEFAULT_TIMEOUT GST_SECOND
#define DEFAULT_ROUNDTRIP_LIMIT GST_SECOND
/* Minimum timeout will be immediately (ie, as fast as one RTT), but no
* more often than 1/20th second (arbitrarily, to spread observations a little) */
#define DEFAULT_MINIMUM_UPDATE_INTERVAL (GST_SECOND / 20)
#define DEFAULT_BASE_TIME 0
/* Maximum number of clock updates we can skip before updating */
#define MAX_SKIPPED_UPDATES 5
#define MEDIAN_PRE_FILTERING_WINDOW 9
enum
{
PROP_0,
PROP_ADDRESS,
PROP_PORT,
PROP_ROUNDTRIP_LIMIT,
PROP_MINIMUM_UPDATE_INTERVAL,
PROP_BUS,
PROP_BASE_TIME,
PROP_INTERNAL_CLOCK
};
#define GST_NET_CLIENT_CLOCK_GET_PRIVATE(obj) \
(G_TYPE_INSTANCE_GET_PRIVATE ((obj), GST_TYPE_NET_CLIENT_CLOCK, GstNetClientClockPrivate))
struct _GstNetClientClockPrivate
{
GstClock *internal_clock;
GThread *thread;
GSocket *socket;
GSocketAddress *servaddr;
GCancellable *cancel;
gboolean made_cancel_fd;
GstClockTime timeout_expiration;
GstClockTime roundtrip_limit;
GstClockTime rtt_avg;
GstClockTime minimum_update_interval;
GstClockTime last_remote_poll_interval;
guint skipped_updates;
GstClockTime last_rtts[MEDIAN_PRE_FILTERING_WINDOW];
gint last_rtts_missing;
GstClockTime base_time;
gchar *address;
gint port;
GstBus *bus;
gboolean is_ntp;
};
#define _do_init \
GST_DEBUG_CATEGORY_INIT (ncc_debug, "netclock", 0, "Network client clock");
#define gst_net_client_clock_parent_class parent_class
G_DEFINE_TYPE_WITH_CODE (GstNetClientClock, gst_net_client_clock,
GST_TYPE_SYSTEM_CLOCK, _do_init);
static void gst_net_client_clock_finalize (GObject * object);
static void gst_net_client_clock_set_property (GObject * object, guint prop_id,
const GValue * value, GParamSpec * pspec);
static void gst_net_client_clock_get_property (GObject * object, guint prop_id,
GValue * value, GParamSpec * pspec);
static void gst_net_client_clock_constructed (GObject * object);
static gboolean gst_net_client_clock_start (GstNetClientClock * self);
static void gst_net_client_clock_stop (GstNetClientClock * self);
static GstClockTime gst_net_client_clock_get_internal_time (GstClock * clock);
static void
gst_net_client_clock_class_init (GstNetClientClockClass * klass)
{
GObjectClass *gobject_class;
GstClockClass *clock_class;
gobject_class = G_OBJECT_CLASS (klass);
clock_class = GST_CLOCK_CLASS (klass);
g_type_class_add_private (klass, sizeof (GstNetClientClockPrivate));
gobject_class->finalize = gst_net_client_clock_finalize;
gobject_class->get_property = gst_net_client_clock_get_property;
gobject_class->set_property = gst_net_client_clock_set_property;
gobject_class->constructed = gst_net_client_clock_constructed;
g_object_class_install_property (gobject_class, PROP_ADDRESS,
g_param_spec_string ("address", "address",
"The IP address of the machine providing a time server",
DEFAULT_ADDRESS,
G_PARAM_READWRITE | G_PARAM_CONSTRUCT | G_PARAM_STATIC_STRINGS));
g_object_class_install_property (gobject_class, PROP_PORT,
g_param_spec_int ("port", "port",
"The port on which the remote server is listening", 0, G_MAXUINT16,
DEFAULT_PORT,
G_PARAM_READWRITE | G_PARAM_CONSTRUCT | G_PARAM_STATIC_STRINGS));
g_object_class_install_property (gobject_class, PROP_BUS,
g_param_spec_object ("bus", "bus",
"A GstBus on which to send clock status information", GST_TYPE_BUS,
G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
/**
* GstNetClientClock::round-trip-limit:
*
* Maximum allowed round-trip for packets. If this property is set to a nonzero
* value, all packets with a round-trip interval larger than this limit will be
* ignored. This is useful for networks with severe and fluctuating transport
* delays. Filtering out these packets increases stability of the synchronization.
* On the other hand, the lower the limit, the higher the amount of filtered
* packets. Empirical tests are typically necessary to estimate a good value
* for the limit.
* If the property is set to zero, the limit is disabled.
*
* Since: 1.4
*/
g_object_class_install_property (gobject_class, PROP_ROUNDTRIP_LIMIT,
g_param_spec_uint64 ("round-trip-limit", "round-trip limit",
"Maximum tolerable round-trip interval for packets, in nanoseconds "
"(0 = no limit)", 0, G_MAXUINT64, DEFAULT_ROUNDTRIP_LIMIT,
G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
g_object_class_install_property (gobject_class, PROP_MINIMUM_UPDATE_INTERVAL,
g_param_spec_uint64 ("minimum-update-interval", "minimum update interval",
"Minimum polling interval for packets, in nanoseconds"
"(0 = no limit)", 0, G_MAXUINT64, DEFAULT_MINIMUM_UPDATE_INTERVAL,
G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
g_object_class_install_property (gobject_class, PROP_BASE_TIME,
g_param_spec_uint64 ("base-time", "Base Time",
"Initial time that is reported before synchronization", 0,
G_MAXUINT64, DEFAULT_BASE_TIME,
G_PARAM_READWRITE | G_PARAM_CONSTRUCT_ONLY | G_PARAM_STATIC_STRINGS));
g_object_class_install_property (gobject_class, PROP_INTERNAL_CLOCK,
g_param_spec_object ("internal-clock", "Internal Clock",
"Internal clock that directly slaved to the remote clock",
GST_TYPE_CLOCK, G_PARAM_READABLE | G_PARAM_STATIC_STRINGS));
clock_class->get_internal_time = gst_net_client_clock_get_internal_time;
}
static void
gst_net_client_clock_init (GstNetClientClock * self)
{
GstNetClientClockPrivate *priv;
self->priv = priv = GST_NET_CLIENT_CLOCK_GET_PRIVATE (self);
GST_OBJECT_FLAG_SET (self, GST_CLOCK_FLAG_CAN_SET_MASTER);
priv->internal_clock = g_object_new (GST_TYPE_SYSTEM_CLOCK, NULL);
priv->port = DEFAULT_PORT;
priv->address = g_strdup (DEFAULT_ADDRESS);
gst_clock_set_timeout (priv->internal_clock, DEFAULT_TIMEOUT);
priv->thread = NULL;
priv->servaddr = NULL;
priv->rtt_avg = GST_CLOCK_TIME_NONE;
priv->roundtrip_limit = DEFAULT_ROUNDTRIP_LIMIT;
priv->minimum_update_interval = DEFAULT_MINIMUM_UPDATE_INTERVAL;
priv->last_remote_poll_interval = GST_CLOCK_TIME_NONE;
priv->skipped_updates = 0;
priv->last_rtts_missing = MEDIAN_PRE_FILTERING_WINDOW;
priv->base_time = DEFAULT_BASE_TIME;
}
static void
gst_net_client_clock_finalize (GObject * object)
{
GstNetClientClock *self = GST_NET_CLIENT_CLOCK (object);
if (self->priv->thread) {
gst_net_client_clock_stop (self);
}
g_free (self->priv->address);
self->priv->address = NULL;
if (self->priv->servaddr != NULL) {
g_object_unref (self->priv->servaddr);
self->priv->servaddr = NULL;
}
if (self->priv->socket != NULL) {
g_socket_close (self->priv->socket, NULL);
g_object_unref (self->priv->socket);
self->priv->socket = NULL;
}
if (self->priv->bus != NULL) {
gst_object_unref (self->priv->bus);
self->priv->bus = NULL;
}
if (self->priv->internal_clock != NULL) {
gst_object_unref (self->priv->internal_clock);
self->priv->internal_clock = NULL;
}
G_OBJECT_CLASS (parent_class)->finalize (object);
}
static void
gst_net_client_clock_set_property (GObject * object, guint prop_id,
const GValue * value, GParamSpec * pspec)
{
GstNetClientClock *self = GST_NET_CLIENT_CLOCK (object);
switch (prop_id) {
case PROP_ADDRESS:
GST_OBJECT_LOCK (self);
g_free (self->priv->address);
self->priv->address = g_value_dup_string (value);
if (self->priv->address == NULL)
self->priv->address = g_strdup (DEFAULT_ADDRESS);
GST_OBJECT_UNLOCK (self);
break;
case PROP_PORT:
GST_OBJECT_LOCK (self);
self->priv->port = g_value_get_int (value);
GST_OBJECT_UNLOCK (self);
break;
case PROP_ROUNDTRIP_LIMIT:
GST_OBJECT_LOCK (self);
self->priv->roundtrip_limit = g_value_get_uint64 (value);
GST_OBJECT_UNLOCK (self);
break;
case PROP_MINIMUM_UPDATE_INTERVAL:
GST_OBJECT_LOCK (self);
self->priv->minimum_update_interval = g_value_get_uint64 (value);
GST_OBJECT_UNLOCK (self);
break;
case PROP_BUS:
GST_OBJECT_LOCK (self);
if (self->priv->bus)
gst_object_unref (self->priv->bus);
self->priv->bus = g_value_dup_object (value);
GST_OBJECT_UNLOCK (self);
break;
case PROP_BASE_TIME:
self->priv->base_time = g_value_get_uint64 (value);
break;
default:
G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
break;
}
}
static void
gst_net_client_clock_get_property (GObject * object, guint prop_id,
GValue * value, GParamSpec * pspec)
{
GstNetClientClock *self = GST_NET_CLIENT_CLOCK (object);
switch (prop_id) {
case PROP_ADDRESS:
GST_OBJECT_LOCK (self);
g_value_set_string (value, self->priv->address);
GST_OBJECT_UNLOCK (self);
break;
case PROP_PORT:
g_value_set_int (value, self->priv->port);
break;
case PROP_ROUNDTRIP_LIMIT:
GST_OBJECT_LOCK (self);
g_value_set_uint64 (value, self->priv->roundtrip_limit);
GST_OBJECT_UNLOCK (self);
break;
case PROP_MINIMUM_UPDATE_INTERVAL:
GST_OBJECT_LOCK (self);
g_value_set_uint64 (value, self->priv->minimum_update_interval);
GST_OBJECT_UNLOCK (self);
break;
case PROP_BUS:
GST_OBJECT_LOCK (self);
g_value_set_object (value, self->priv->bus);
GST_OBJECT_UNLOCK (self);
break;
case PROP_BASE_TIME:
g_value_set_uint64 (value, self->priv->base_time);
break;
case PROP_INTERNAL_CLOCK:
g_value_set_object (value, self->priv->internal_clock);
break;
default:
G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
break;
}
}
static void
gst_net_client_clock_constructed (GObject * object)
{
GstNetClientClock *self = GST_NET_CLIENT_CLOCK (object);
GstClock *internal_clock = self->priv->internal_clock;
GstClockTime internal;
G_OBJECT_CLASS (parent_class)->constructed (object);
/* gst_clock_get_time() values are guaranteed to be increasing. because no one
* has called get_time on this clock yet we are free to adjust to any value
* without worrying about worrying about MAX() issues with the clock's
* internal time.
*/
/* update our internal time so get_time() give something around base_time.
assume that the rate is 1 in the beginning. */
internal = gst_clock_get_internal_time (internal_clock);
gst_clock_set_calibration (internal_clock, internal,
self->priv->base_time, 1, 1);
{
GstClockTime now = gst_clock_get_time (internal_clock);
if (GST_CLOCK_DIFF (now, self->priv->base_time) > 0 ||
GST_CLOCK_DIFF (now, self->priv->base_time + GST_SECOND) < 0) {
g_warning ("unable to set the base time, expect sync problems!");
}
}
if (!gst_net_client_clock_start (self)) {
g_warning ("failed to start clock '%s'", GST_OBJECT_NAME (self));
}
/* all systems go, cap'n */
}
static GstClockTime
gst_net_client_clock_get_internal_time (GstClock * clock)
{
GstNetClientClock *self = GST_NET_CLIENT_CLOCK (clock);
return gst_clock_get_time (self->priv->internal_clock);
}
static gint
compare_clock_time (const GstClockTime * a, const GstClockTime * b)
{
if (*a < *b)
return -1;
else if (*a > *b)
return 1;
return 0;
}
static void
gst_net_client_clock_observe_times (GstNetClientClock * self,
GstClockTime local_1, GstClockTime remote_1, GstClockTime remote_2,
GstClockTime local_2)
{
GstNetClientClockPrivate *priv = self->priv;
GstClock *internal_clock = priv->internal_clock;
GstClockTime current_timeout = 0;
GstClockTime local_avg, remote_avg;
gdouble r_squared;
GstClock *clock;
GstClockTime rtt, rtt_limit, min_update_interval;
GstBus *bus = NULL;
/* Use for discont tracking */
GstClockTime time_before = 0;
GstClockTime min_guess = 0;
GstClockTimeDiff time_discont = 0;
gboolean synched, now_synched;
GstClockTime internal_time, external_time, rate_num, rate_den;
GstClockTime orig_internal_time, orig_external_time, orig_rate_num,
orig_rate_den;
GstClockTime max_discont;
GstClockTime last_rtts[MEDIAN_PRE_FILTERING_WINDOW];
GstClockTime median;
gint i;
GST_OBJECT_LOCK (self);
rtt_limit = self->priv->roundtrip_limit;
/* If the server told us a poll interval and it's bigger than the
* one configured via the property, use the server's */
if (self->priv->last_remote_poll_interval != GST_CLOCK_TIME_NONE &&
self->priv->last_remote_poll_interval >
self->priv->minimum_update_interval)
min_update_interval = self->priv->last_remote_poll_interval;
else
min_update_interval = self->priv->minimum_update_interval;
if (self->priv->bus)
bus = gst_object_ref (self->priv->bus);
GST_OBJECT_UNLOCK (self);
if (local_2 < local_1) {
GST_LOG_OBJECT (self, "Dropping observation: receive time %" GST_TIME_FORMAT
" < send time %" GST_TIME_FORMAT, GST_TIME_ARGS (local_1),
GST_TIME_ARGS (local_2));
goto bogus_observation;
}
if (remote_2 < remote_1) {
GST_LOG_OBJECT (self,
"Dropping observation: remote receive time %" GST_TIME_FORMAT
" < send time %" GST_TIME_FORMAT, GST_TIME_ARGS (remote_1),
GST_TIME_ARGS (remote_2));
goto bogus_observation;
}
/* The round trip time is (assuming symmetric path delays)
* delta = (local_2 - local_1) - (remote_2 - remote_1)
*/
rtt = GST_CLOCK_DIFF (local_1, local_2) - GST_CLOCK_DIFF (remote_1, remote_2);
if ((rtt_limit > 0) && (rtt > rtt_limit)) {
GST_LOG_OBJECT (self,
"Dropping observation: RTT %" GST_TIME_FORMAT " > limit %"
GST_TIME_FORMAT, GST_TIME_ARGS (rtt), GST_TIME_ARGS (rtt_limit));
goto bogus_observation;
}
for (i = 1; i < MEDIAN_PRE_FILTERING_WINDOW; i++)
self->priv->last_rtts[i - 1] = self->priv->last_rtts[i];
self->priv->last_rtts[i - 1] = rtt;
if (self->priv->last_rtts_missing) {
self->priv->last_rtts_missing--;
} else {
memcpy (&last_rtts, &self->priv->last_rtts, sizeof (last_rtts));
g_qsort_with_data (&last_rtts,
MEDIAN_PRE_FILTERING_WINDOW, sizeof (GstClockTime),
(GCompareDataFunc) compare_clock_time, NULL);
median = last_rtts[MEDIAN_PRE_FILTERING_WINDOW / 2];
/* FIXME: We might want to use something else here, like only allowing
* things in the interquartile range, or also filtering away delays that
* are too small compared to the median. This here worked well enough
* in tests so far.
*/
if (rtt > 2 * median) {
GST_LOG_OBJECT (self,
"Dropping observation, long RTT %" GST_TIME_FORMAT " > 2 * median %"
GST_TIME_FORMAT, GST_TIME_ARGS (rtt), GST_TIME_ARGS (median));
goto bogus_observation;
}
}
/* Track an average round trip time, for a bit of smoothing */
/* Always update before discarding a sample, so genuine changes in
* the network get picked up, eventually */
if (priv->rtt_avg == GST_CLOCK_TIME_NONE)
priv->rtt_avg = rtt;
else if (rtt < priv->rtt_avg) /* Shorter RTTs carry more weight than longer */
priv->rtt_avg = (3 * priv->rtt_avg + rtt) / 4;
else
priv->rtt_avg = (15 * priv->rtt_avg + rtt) / 16;
if (rtt > 2 * priv->rtt_avg) {
GST_LOG_OBJECT (self,
"Dropping observation, long RTT %" GST_TIME_FORMAT " > 2 * avg %"
GST_TIME_FORMAT, GST_TIME_ARGS (rtt), GST_TIME_ARGS (priv->rtt_avg));
goto bogus_observation;
}
/* The difference between the local and remote clock (again assuming
* symmetric path delays):
*
* local_1 + delta / 2 - remote_1 = theta
* or
* local_2 - delta / 2 - remote_2 = theta
*
* which gives after some simple algebraic transformations:
*
* (remote_1 - local_1) + (remote_2 - local_2)
* theta = -------------------------------------------
* 2
*
*
* Thus remote time at local_avg is equal to:
*
* local_avg + theta =
*
* local_1 + local_2 (remote_1 - local_1) + (remote_2 - local_2)
* ----------------- + -------------------------------------------
* 2 2
*
* =
*
* remote_1 + remote_2
* ------------------- = remote_avg
* 2
*
* We use this for our clock estimation, i.e. local_avg at remote clock
* being the same as remote_avg.
*/
local_avg = (local_2 + local_1) / 2;
remote_avg = (remote_2 + remote_1) / 2;
GST_LOG_OBJECT (self,
"local1 %" G_GUINT64_FORMAT " remote1 %" G_GUINT64_FORMAT " remote2 %"
G_GUINT64_FORMAT " remoteavg %" G_GUINT64_FORMAT " localavg %"
G_GUINT64_FORMAT " local2 %" G_GUINT64_FORMAT, local_1, remote_1,
remote_2, remote_avg, local_avg, local_2);
clock = GST_CLOCK_CAST (self);
/* Store what the clock produced as 'now' before this update */
gst_clock_get_calibration (internal_clock, &orig_internal_time,
&orig_external_time, &orig_rate_num, &orig_rate_den);
internal_time = orig_internal_time;
external_time = orig_external_time;
rate_num = orig_rate_num;
rate_den = orig_rate_den;
min_guess =
gst_clock_adjust_with_calibration (internal_clock, local_1,
internal_time, external_time, rate_num, rate_den);
time_before =
gst_clock_adjust_with_calibration (internal_clock, local_2,
internal_time, external_time, rate_num, rate_den);
/* Maximum discontinuity, when we're synched with the master. Could make this a property,
* but this value seems to work fine */
max_discont = priv->rtt_avg / 4;
/* If the remote observation was within a max_discont window around our min/max estimates, we're synched */
synched =
(GST_CLOCK_DIFF (remote_avg, min_guess) < (GstClockTimeDiff) (max_discont)
&& GST_CLOCK_DIFF (time_before,
remote_avg) < (GstClockTimeDiff) (max_discont));
if (gst_clock_add_observation_unapplied (internal_clock,
local_avg, remote_avg, &r_squared, &internal_time, &external_time,
&rate_num, &rate_den)) {
/* Now compare the difference (discont) in the clock
* after this observation */
time_discont = GST_CLOCK_DIFF (time_before,
gst_clock_adjust_with_calibration (internal_clock, local_2,
internal_time, external_time, rate_num, rate_den));
/* If we were in sync with the remote clock, clamp the allowed
* discontinuity to within quarter of one RTT. In sync means our send/receive estimates
* of remote time correctly windowed the actual remote time observation */
if (synched && ABS (time_discont) > max_discont) {
GstClockTimeDiff offset;
GST_DEBUG_OBJECT (clock,
"Too large a discont, clamping to 1/4 average RTT = %"
GST_TIME_FORMAT, GST_TIME_ARGS (max_discont));
if (time_discont > 0) { /* Too large a forward step - add a -ve offset */
offset = max_discont - time_discont;
if (-offset > external_time)
external_time = 0;
else
external_time += offset;
} else { /* Too large a backward step - add a +ve offset */
offset = -(max_discont + time_discont);
external_time += offset;
}
time_discont += offset;
}
/* Check if the new clock params would have made our observation within range */
now_synched =
(GST_CLOCK_DIFF (remote_avg,
gst_clock_adjust_with_calibration (internal_clock,
local_1, internal_time, external_time, rate_num,
rate_den)) < (GstClockTimeDiff) (max_discont))
&&
(GST_CLOCK_DIFF (gst_clock_adjust_with_calibration
(internal_clock, local_2, internal_time, external_time,
rate_num, rate_den),
remote_avg) < (GstClockTimeDiff) (max_discont));
/* Only update the clock if we had synch or just gained it */
if (synched || now_synched || priv->skipped_updates > MAX_SKIPPED_UPDATES) {
gst_clock_set_calibration (internal_clock, internal_time,
external_time, rate_num, rate_den);
/* ghetto formula - shorter timeout for bad correlations */
current_timeout = (1e-3 / (1 - MIN (r_squared, 0.99999))) * GST_SECOND;
current_timeout =
MIN (current_timeout, gst_clock_get_timeout (internal_clock));
priv->skipped_updates = 0;
} else {
/* Restore original calibration vars for the report, we're not changing the clock */
internal_time = orig_internal_time;
external_time = orig_external_time;
rate_num = orig_rate_num;
rate_den = orig_rate_den;
time_discont = 0;
priv->skipped_updates++;
}
}
/* Limit the polling to at most one per minimum_update_interval */
if (rtt < min_update_interval)
current_timeout = MAX (min_update_interval - rtt, current_timeout);
if (bus) {
GstStructure *s;
GstMessage *msg;
/* Output a stats message, whether we updated the clock or not */
s = gst_structure_new ("gst-netclock-statistics",
"synchronised", G_TYPE_BOOLEAN, synched,
"rtt", G_TYPE_UINT64, rtt,
"rtt-average", G_TYPE_UINT64, priv->rtt_avg,
"local", G_TYPE_UINT64, local_avg,
"remote", G_TYPE_UINT64, remote_avg,
"discontinuity", G_TYPE_INT64, time_discont,
"remote-min-estimate", G_TYPE_UINT64, min_guess,
"remote-max-estimate", G_TYPE_UINT64, time_before,
"remote-min-error", G_TYPE_INT64, GST_CLOCK_DIFF (remote_avg,
min_guess), "remote-max-error", G_TYPE_INT64,
GST_CLOCK_DIFF (remote_avg, time_before), "request-send", G_TYPE_UINT64,
local_1, "request-receive", G_TYPE_UINT64, local_2, "r-squared",
G_TYPE_DOUBLE, r_squared, "timeout", G_TYPE_UINT64, current_timeout,
"internal-time", G_TYPE_UINT64, internal_time, "external-time",
G_TYPE_UINT64, external_time, "rate-num", G_TYPE_UINT64, rate_num,
"rate-den", G_TYPE_UINT64, rate_den, "rate", G_TYPE_DOUBLE,
(gdouble) (rate_num) / rate_den, "local-clock-offset", G_TYPE_INT64,
GST_CLOCK_DIFF (internal_time, external_time), NULL);
msg = gst_message_new_element (GST_OBJECT (self), s);
gst_bus_post (bus, msg);
}
GST_INFO ("next timeout: %" GST_TIME_FORMAT, GST_TIME_ARGS (current_timeout));
self->priv->timeout_expiration = gst_util_get_timestamp () + current_timeout;
if (bus)
gst_object_unref (bus);
return;
bogus_observation:
if (bus)
gst_object_unref (bus);
/* Schedule a new packet again soon */
self->priv->timeout_expiration = gst_util_get_timestamp () + (GST_SECOND / 4);
return;
}
static gpointer
gst_net_client_clock_thread (gpointer data)
{
GstNetClientClock *self = data;
GstClock *internal_clock = self->priv->internal_clock;
GSocket *socket = self->priv->socket;
GError *err = NULL;
GST_INFO_OBJECT (self, "net client clock thread running, socket=%p", socket);
g_socket_set_blocking (socket, TRUE);
g_socket_set_timeout (socket, 0);
while (!g_cancellable_is_cancelled (self->priv->cancel)) {
GstClockTime expiration_time = self->priv->timeout_expiration;
GstClockTime now = gst_util_get_timestamp ();
gint64 socket_timeout;
if (now >= expiration_time || (expiration_time - now) <= GST_MSECOND) {
socket_timeout = 0;
} else {
socket_timeout = (expiration_time - now) / GST_USECOND;
}
GST_TRACE_OBJECT (self, "timeout: %" G_GINT64_FORMAT "us", socket_timeout);
if (!g_socket_condition_timed_wait (socket, G_IO_IN, socket_timeout,
self->priv->cancel, &err)) {
/* cancelled, timeout or error */
if (err->code == G_IO_ERROR_CANCELLED) {
GST_INFO_OBJECT (self, "cancelled");
g_clear_error (&err);
break;
} else if (err->code == G_IO_ERROR_TIMED_OUT) {
/* timed out, let's send another packet */
GST_DEBUG_OBJECT (self, "timed out");
if (self->priv->is_ntp) {
GstNtpPacket *packet;
packet = gst_ntp_packet_new (NULL, NULL);
packet->transmit_time = gst_clock_get_internal_time (internal_clock);
GST_DEBUG_OBJECT (self,
"sending packet, local time = %" GST_TIME_FORMAT,
GST_TIME_ARGS (packet->transmit_time));
gst_ntp_packet_send (packet, self->priv->socket,
self->priv->servaddr, NULL);
g_free (packet);
} else {
GstNetTimePacket *packet;
packet = gst_net_time_packet_new (NULL);
packet->local_time = gst_clock_get_internal_time (internal_clock);
GST_DEBUG_OBJECT (self,
"sending packet, local time = %" GST_TIME_FORMAT,
GST_TIME_ARGS (packet->local_time));
gst_net_time_packet_send (packet, self->priv->socket,
self->priv->servaddr, NULL);
g_free (packet);
}
/* reset timeout (but are expecting a response sooner anyway) */
self->priv->timeout_expiration =
gst_util_get_timestamp () + gst_clock_get_timeout (internal_clock);
} else {
GST_DEBUG_OBJECT (self, "socket error: %s", err->message);
g_usleep (G_USEC_PER_SEC / 10); /* throttle */
}
g_clear_error (&err);
} else {
GstClockTime new_local;
/* got packet */
new_local = gst_clock_get_internal_time (internal_clock);
if (self->priv->is_ntp) {
GstNtpPacket *packet;
packet = gst_ntp_packet_receive (socket, NULL, &err);
if (packet != NULL) {
GST_LOG_OBJECT (self, "got packet back");
GST_LOG_OBJECT (self, "local_1 = %" GST_TIME_FORMAT,
GST_TIME_ARGS (packet->origin_time));
GST_LOG_OBJECT (self, "remote_1 = %" GST_TIME_FORMAT,
GST_TIME_ARGS (packet->receive_time));
GST_LOG_OBJECT (self, "remote_2 = %" GST_TIME_FORMAT,
GST_TIME_ARGS (packet->transmit_time));
GST_LOG_OBJECT (self, "local_2 = %" GST_TIME_FORMAT,
GST_TIME_ARGS (new_local));
GST_LOG_OBJECT (self, "poll_interval = %" GST_TIME_FORMAT,
GST_TIME_ARGS (packet->poll_interval));
/* Remember the last poll interval we ever got from the server */
if (packet->poll_interval != GST_CLOCK_TIME_NONE)
self->priv->last_remote_poll_interval = packet->poll_interval;
/* observe_times will reset the timeout */
gst_net_client_clock_observe_times (self, packet->origin_time,
packet->receive_time, packet->transmit_time, new_local);
g_free (packet);
} else if (err != NULL) {
if (g_error_matches (err, GST_NTP_ERROR, GST_NTP_ERROR_WRONG_VERSION)
|| g_error_matches (err, GST_NTP_ERROR, GST_NTP_ERROR_KOD_DENY)) {
GST_ERROR_OBJECT (self, "fatal receive error: %s", err->message);
break;
} else if (g_error_matches (err, GST_NTP_ERROR,
GST_NTP_ERROR_KOD_RATE)) {
GST_WARNING_OBJECT (self, "need to limit rate");
/* If the server did not tell us a poll interval before, double
* our minimum poll interval. Otherwise we assume that the server
* already told us something sensible and that this error here
* was just a spurious error */
if (self->priv->last_remote_poll_interval == GST_CLOCK_TIME_NONE)
self->priv->minimum_update_interval *= 2;
/* And wait a bit before we send the next packet instead of
* sending it immediately */
self->priv->timeout_expiration =
gst_util_get_timestamp () +
gst_clock_get_timeout (internal_clock);
} else {
GST_WARNING_OBJECT (self, "receive error: %s", err->message);
}
g_clear_error (&err);
}
} else {
GstNetTimePacket *packet;
packet = gst_net_time_packet_receive (socket, NULL, &err);
if (packet != NULL) {
GST_LOG_OBJECT (self, "got packet back");
GST_LOG_OBJECT (self, "local_1 = %" GST_TIME_FORMAT,
GST_TIME_ARGS (packet->local_time));
GST_LOG_OBJECT (self, "remote = %" GST_TIME_FORMAT,
GST_TIME_ARGS (packet->remote_time));
GST_LOG_OBJECT (self, "local_2 = %" GST_TIME_FORMAT,
GST_TIME_ARGS (new_local));
/* observe_times will reset the timeout */
gst_net_client_clock_observe_times (self, packet->local_time,
packet->remote_time, packet->remote_time, new_local);
g_free (packet);
} else if (err != NULL) {
GST_WARNING_OBJECT (self, "receive error: %s", err->message);
g_clear_error (&err);
}
}
}
}
GST_INFO_OBJECT (self, "shutting down net client clock thread");
return NULL;
}
static gboolean
gst_net_client_clock_start (GstNetClientClock * self)
{
GSocketAddress *servaddr;
GSocketAddress *myaddr;
GSocketAddress *anyaddr;
GInetAddress *inetaddr;
GSocket *socket;
GError *error = NULL;
GSocketFamily family;
GPollFD dummy_pollfd;
g_return_val_if_fail (self->priv->address != NULL, FALSE);
g_return_val_if_fail (self->priv->servaddr == NULL, FALSE);
/* create target address */
inetaddr = g_inet_address_new_from_string (self->priv->address);
if (inetaddr == NULL)
goto bad_address;
family = g_inet_address_get_family (inetaddr);
servaddr = g_inet_socket_address_new (inetaddr, self->priv->port);
g_object_unref (inetaddr);
g_assert (servaddr != NULL);
GST_DEBUG_OBJECT (self, "will communicate with %s:%d", self->priv->address,
self->priv->port);
socket = g_socket_new (family, G_SOCKET_TYPE_DATAGRAM,
G_SOCKET_PROTOCOL_UDP, &error);
if (socket == NULL)
goto no_socket;
GST_DEBUG_OBJECT (self, "binding socket");
inetaddr = g_inet_address_new_any (family);
anyaddr = g_inet_socket_address_new (inetaddr, 0);
g_socket_bind (socket, anyaddr, TRUE, &error);
g_object_unref (anyaddr);
g_object_unref (inetaddr);
if (error != NULL)
goto bind_error;
/* check address we're bound to, mostly for debugging purposes */
myaddr = g_socket_get_local_address (socket, &error);
if (myaddr == NULL)
goto getsockname_error;
GST_DEBUG_OBJECT (self, "socket opened on UDP port %d",
g_inet_socket_address_get_port (G_INET_SOCKET_ADDRESS (myaddr)));
g_object_unref (myaddr);
self->priv->cancel = g_cancellable_new ();
self->priv->made_cancel_fd =
g_cancellable_make_pollfd (self->priv->cancel, &dummy_pollfd);
self->priv->socket = socket;
self->priv->servaddr = G_SOCKET_ADDRESS (servaddr);
self->priv->thread = g_thread_try_new ("GstNetClientClock",
gst_net_client_clock_thread, self, &error);
if (error != NULL)
goto no_thread;
return TRUE;
/* ERRORS */
no_socket:
{
GST_ERROR_OBJECT (self, "socket_new() failed: %s", error->message);
g_error_free (error);
return FALSE;
}
bind_error:
{
GST_ERROR_OBJECT (self, "bind failed: %s", error->message);
g_error_free (error);
g_object_unref (socket);
return FALSE;
}
getsockname_error:
{
GST_ERROR_OBJECT (self, "get_local_address() failed: %s", error->message);
g_error_free (error);
g_object_unref (socket);
return FALSE;
}
bad_address:
{
GST_ERROR_OBJECT (self, "inet_address_new_from_string('%s') failed",
self->priv->address);
return FALSE;
}
no_thread:
{
GST_ERROR_OBJECT (self, "could not create thread: %s", error->message);
g_object_unref (self->priv->servaddr);
self->priv->servaddr = NULL;
g_object_unref (self->priv->socket);
self->priv->socket = NULL;
g_error_free (error);
return FALSE;
}
}
static void
gst_net_client_clock_stop (GstNetClientClock * self)
{
if (self->priv->thread == NULL)
return;
GST_INFO_OBJECT (self, "stopping...");
g_cancellable_cancel (self->priv->cancel);
g_thread_join (self->priv->thread);
self->priv->thread = NULL;
if (self->priv->made_cancel_fd)
g_cancellable_release_fd (self->priv->cancel);
g_object_unref (self->priv->cancel);
self->priv->cancel = NULL;
g_object_unref (self->priv->servaddr);
self->priv->servaddr = NULL;
g_object_unref (self->priv->socket);
self->priv->socket = NULL;
GST_INFO_OBJECT (self, "stopped");
}
/**
* gst_net_client_clock_new:
* @name: a name for the clock
* @remote_address: the address of the remote clock provider
* @remote_port: the port of the remote clock provider
* @base_time: initial time of the clock
*
* Create a new #GstNetClientClock that will report the time
* provided by the #GstNetTimeProvider on @remote_address and
* @remote_port.
*
* Returns: a new #GstClock that receives a time from the remote
* clock.
*/
GstClock *
gst_net_client_clock_new (const gchar * name, const gchar * remote_address,
gint remote_port, GstClockTime base_time)
{
GstNetClientClock *ret;
g_return_val_if_fail (remote_address != NULL, NULL);
g_return_val_if_fail (remote_port > 0, NULL);
g_return_val_if_fail (remote_port <= G_MAXUINT16, NULL);
g_return_val_if_fail (base_time != GST_CLOCK_TIME_NONE, NULL);
ret = g_object_new (GST_TYPE_NET_CLIENT_CLOCK, "address", remote_address,
"port", remote_port, "base-time", base_time, NULL);
return (GstClock *) ret;
}
G_DEFINE_TYPE (GstNtpClock, gst_ntp_clock, GST_TYPE_NET_CLIENT_CLOCK);
static void
gst_ntp_clock_class_init (GstNtpClockClass * klass)
{
}
static void
gst_ntp_clock_init (GstNtpClock * self)
{
GST_NET_CLIENT_CLOCK (self)->priv->is_ntp = TRUE;
}
/**
* gst_ntp_clock_new:
* @name: a name for the clock
* @remote_address: the address of the remote clock provider
* @remote_port: the port of the remote clock provider
* @base_time: initial time of the clock
*
* Create a new #GstNtpClock that will report the time provided by
* the NTPv4 server on @remote_address and @remote_port.
*
* Returns: a new #GstClock that receives a time from the remote
* clock.
*
* Since: 1.6
*/
GstClock *
gst_ntp_clock_new (const gchar * name, const gchar * remote_address,
gint remote_port, GstClockTime base_time)
{
GstNetClientClock *ret;
g_return_val_if_fail (remote_address != NULL, NULL);
g_return_val_if_fail (remote_port > 0, NULL);
g_return_val_if_fail (remote_port <= G_MAXUINT16, NULL);
g_return_val_if_fail (base_time != GST_CLOCK_TIME_NONE, NULL);
ret = g_object_new (GST_TYPE_NTP_CLOCK, "address", remote_address,
"port", remote_port, "base-time", base_time, NULL);
return (GstClock *) ret;
}
Reference in a new issue View git blame Copy permalink