libs/gst/net/gstnetclientclock.c

   1 /* GStreamer
   2  * Copyright (C) 1999,2000 Erik Walthinsen <omega@cse.ogi.edu>
   3  *                    2005 Wim Taymans <wim@fluendo.com>
   4  *                    2005 Andy Wingo <wingo@pobox.com>
   5  * Copyright (C) 2012 Collabora Ltd. <tim.muller@collabora.co.uk>
   6  *
   7  * gstnetclientclock.h: clock that synchronizes itself to a time provider over
   8  * the network
   9  *
  10  * This library is free software; you can redistribute it and/or
  11  * modify it under the terms of the GNU Library General Public
  12  * License as published by the Free Software Foundation; either
  13  * version 2 of the License, or (at your option) any later version.
  14  *
  15  * This library is distributed in the hope that it will be useful,
  16  * but WITHOUT ANY WARRANTY; without even the implied warranty of
  17  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
  18  * Library General Public License for more details.
  19  *
  20  * You should have received a copy of the GNU Library General Public
  21  * License along with this library; if not, write to the
  22  * Free Software Foundation, Inc., 51 Franklin St, Fifth Floor,
  23  * Boston, MA 02110-1301, USA.
  24  */
  25 /**
  26  * SECTION:gstnetclientclock
  27  * @short_description: Special clock that synchronizes to a remote time
  28  *                     provider.
  29  * @see_also: #GstClock, #GstNetTimeProvider, #GstPipeline
  30  *
  31  * This object implements a custom #GstClock that synchronizes its time
  32  * to a remote time provider such as #GstNetTimeProvider.
  33  *
  34  * A new clock is created with gst_net_client_clock_new() which takes the
  35  * address and port of the remote time provider along with a name and
  36  * an initial time.
  37  *
  38  * This clock will poll the time provider and will update its calibration
  39  * parameters based on the local and remote observations.
  40  *
  41  * Various parameters of the clock can be configured with the parent #GstClock
  42  * "timeout", "window-size" and "window-threshold" object properties.
  43  *
  44  * A #GstNetClientClock is typically set on a #GstPipeline with
  45  * gst_pipeline_use_clock().
  46  *
  47  * Last reviewed on 2005-11-23 (0.9.5)
  48  */
  49
  50 #ifdef HAVE_CONFIG_H
  51 #include "config.h"
  52 #endif
  53
  54 #include "gstnettimepacket.h"
  55 #include "gstnetclientclock.h"
  56
  57 #include <gio/gio.h>
  58
  59 GST_DEBUG_CATEGORY_STATIC (ncc_debug);
  60 #define GST_CAT_DEFAULT (ncc_debug)
  61
  62 #define DEFAULT_ADDRESS         "127.0.0.1"
  63 #define DEFAULT_PORT            5637
  64 #define DEFAULT_TIMEOUT         GST_SECOND
  65
  66 enum
  67 {
  68   PROP_0,
  69   PROP_ADDRESS,
  70   PROP_PORT
  71 };
  72
  73 #define GST_NET_CLIENT_CLOCK_GET_PRIVATE(obj)  \
  74   (G_TYPE_INSTANCE_GET_PRIVATE ((obj), GST_TYPE_NET_CLIENT_CLOCK, GstNetClientClockPrivate))
  75
  76 struct _GstNetClientClockPrivate
  77 {
  78   GThread *thread;
  79
  80   GSocket *socket;
  81   GSocketAddress *servaddr;
  82   GCancellable *cancel;
  83
  84   GstClockTime timeout_expiration;
  85
  86   gchar *address;
  87   gint port;
  88 };
  89
  90 #define _do_init \
  91   GST_DEBUG_CATEGORY_INIT (ncc_debug, "netclock", 0, "Network client clock");
  92 #define gst_net_client_clock_parent_class parent_class
  93 G_DEFINE_TYPE_WITH_CODE (GstNetClientClock, gst_net_client_clock,
  94     GST_TYPE_SYSTEM_CLOCK, _do_init);
  95
  96 static void gst_net_client_clock_finalize (GObject * object);
  97 static void gst_net_client_clock_set_property (GObject * object, guint prop_id,
  98     const GValue * value, GParamSpec * pspec);
  99 static void gst_net_client_clock_get_property (GObject * object, guint prop_id,
 100     GValue * value, GParamSpec * pspec);
 101
 102 static void gst_net_client_clock_stop (GstNetClientClock * self);
 103
 104 static void
 105 gst_net_client_clock_class_init (GstNetClientClockClass * klass)
 106 {
 107   GObjectClass *gobject_class;
 108
 109   gobject_class = G_OBJECT_CLASS (klass);
 110
 111   g_type_class_add_private (klass, sizeof (GstNetClientClockPrivate));
 112
 113   gobject_class->finalize = gst_net_client_clock_finalize;
 114   gobject_class->get_property = gst_net_client_clock_get_property;
 115   gobject_class->set_property = gst_net_client_clock_set_property;
 116
 117   g_object_class_install_property (gobject_class, PROP_ADDRESS,
 118       g_param_spec_string ("address", "address",
 119           "The IP address of the machine providing a time server",
 120           DEFAULT_ADDRESS, G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 121   g_object_class_install_property (gobject_class, PROP_PORT,
 122       g_param_spec_int ("port", "port",
 123           "The port on which the remote server is listening", 0, G_MAXUINT16,
 124           DEFAULT_PORT, G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 125 }
 126
 127 static void
 128 gst_net_client_clock_init (GstNetClientClock * self)
 129 {
 130   GstClock *clock = GST_CLOCK_CAST (self);
 131   GstNetClientClockPrivate *priv;
 132
 133   self->priv = priv = GST_NET_CLIENT_CLOCK_GET_PRIVATE (self);
 134
 135   priv->port = DEFAULT_PORT;
 136   priv->address = g_strdup (DEFAULT_ADDRESS);
 137
 138   gst_clock_set_timeout (clock, DEFAULT_TIMEOUT);
 139
 140   priv->thread = NULL;
 141
 142   priv->servaddr = NULL;
 143 }
 144
 145 static void
 146 gst_net_client_clock_finalize (GObject * object)
 147 {
 148   GstNetClientClock *self = GST_NET_CLIENT_CLOCK (object);
 149
 150   if (self->priv->thread) {
 151     gst_net_client_clock_stop (self);
 152   }
 153
 154   g_free (self->priv->address);
 155   self->priv->address = NULL;
 156
 157   if (self->priv->servaddr != NULL) {
 158     g_object_unref (self->priv->servaddr);
 159     self->priv->servaddr = NULL;
 160   }
 161
 162   if (self->priv->socket != NULL) {
 163     g_socket_close (self->priv->socket, NULL);
 164     g_object_unref (self->priv->socket);
 165     self->priv->socket = NULL;
 166   }
 167
 168   G_OBJECT_CLASS (parent_class)->finalize (object);
 169 }
 170
 171 static void
 172 gst_net_client_clock_set_property (GObject * object, guint prop_id,
 173     const GValue * value, GParamSpec * pspec)
 174 {
 175   GstNetClientClock *self = GST_NET_CLIENT_CLOCK (object);
 176
 177   switch (prop_id) {
 178     case PROP_ADDRESS:
 179       g_free (self->priv->address);
 180       self->priv->address = g_value_dup_string (value);
 181       if (self->priv->address == NULL)
 182         self->priv->address = g_strdup (DEFAULT_ADDRESS);
 183       break;
 184     case PROP_PORT:
 185       self->priv->port = g_value_get_int (value);
 186       break;
 187     default:
 188       G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
 189       break;
 190   }
 191 }
 192
 193 static void
 194 gst_net_client_clock_get_property (GObject * object, guint prop_id,
 195     GValue * value, GParamSpec * pspec)
 196 {
 197   GstNetClientClock *self = GST_NET_CLIENT_CLOCK (object);
 198
 199   switch (prop_id) {
 200     case PROP_ADDRESS:
 201       g_value_set_string (value, self->priv->address);
 202       break;
 203     case PROP_PORT:
 204       g_value_set_int (value, self->priv->port);
 205       break;
 206     default:
 207       G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
 208       break;
 209   }
 210 }
 211
 212 static void
 213 gst_net_client_clock_observe_times (GstNetClientClock * self,
 214     GstClockTime local_1, GstClockTime remote, GstClockTime local_2)
 215 {
 216   GstClockTime current_timeout;
 217   GstClockTime local_avg;
 218   gdouble r_squared;
 219   GstClock *clock;
 220
 221   if (local_2 < local_1)
 222     goto bogus_observation;
 223
 224   local_avg = (local_2 + local_1) / 2;
 225
 226   clock = GST_CLOCK_CAST (self);
 227
 228   if (gst_clock_add_observation (GST_CLOCK (self), local_avg, remote,
 229           &r_squared)) {
 230     /* geto formula */
 231     current_timeout = (1e-3 / (1 - MIN (r_squared, 0.99999))) * GST_SECOND;
 232     current_timeout = MIN (current_timeout, gst_clock_get_timeout (clock));
 233   } else {
 234     current_timeout = 0;
 235   }
 236
 237   GST_INFO ("next timeout: %" GST_TIME_FORMAT, GST_TIME_ARGS (current_timeout));
 238   self->priv->timeout_expiration = gst_util_get_timestamp () + current_timeout;
 239
 240   return;
 241
 242 bogus_observation:
 243   {
 244     GST_WARNING_OBJECT (self, "time packet receive time < send time (%"
 245         GST_TIME_FORMAT " < %" GST_TIME_FORMAT ")", GST_TIME_ARGS (local_1),
 246         GST_TIME_ARGS (local_2));
 247     return;
 248   }
 249 }
 250
 251 static gpointer
 252 gst_net_client_clock_thread (gpointer data)
 253 {
 254   GstNetClientClock *self = data;
 255   GstNetTimePacket *packet;
 256   GSocket *socket = self->priv->socket;
 257   GError *err = NULL;
 258   GstClock *clock = data;
 259
 260   GST_INFO_OBJECT (self, "net client clock thread running, socket=%p", socket);
 261
 262   g_socket_set_blocking (socket, TRUE);
 263   g_socket_set_timeout (socket, 0);
 264
 265   while (!g_cancellable_is_cancelled (self->priv->cancel)) {
 266     GstClockTime expiration_time = self->priv->timeout_expiration;
 267     GstClockTime now = gst_util_get_timestamp ();
 268     gint64 socket_timeout;
 269
 270     if (now >= expiration_time || (expiration_time - now) <= GST_MSECOND) {
 271       socket_timeout = 0;
 272     } else {
 273       socket_timeout = (expiration_time - now) / GST_USECOND;
 274     }
 275
 276     GST_TRACE_OBJECT (self, "timeout: %" G_GINT64_FORMAT "us", socket_timeout);
 277
 278     if (!g_socket_condition_timed_wait (socket, G_IO_IN, socket_timeout,
 279             self->priv->cancel, &err)) {
 280       /* cancelled, timeout or error */
 281       if (err->code == G_IO_ERROR_CANCELLED) {
 282         GST_INFO_OBJECT (self, "cancelled");
 283         g_clear_error (&err);
 284         break;
 285       } else if (err->code == G_IO_ERROR_TIMED_OUT) {
 286         /* timed out, let's send another packet */
 287         GST_DEBUG_OBJECT (self, "timed out");
 288
 289         packet = gst_net_time_packet_new (NULL);
 290
 291         packet->local_time = gst_clock_get_internal_time (GST_CLOCK (self));
 292
 293         GST_DEBUG_OBJECT (self,
 294             "sending packet, local time = %" GST_TIME_FORMAT,
 295             GST_TIME_ARGS (packet->local_time));
 296
 297         gst_net_time_packet_send (packet, self->priv->socket,
 298             self->priv->servaddr, NULL);
 299
 300         g_free (packet);
 301
 302         /* reset timeout (but are expecting a response sooner anyway) */
 303         self->priv->timeout_expiration =
 304             gst_util_get_timestamp () + gst_clock_get_timeout (clock);
 305       } else {
 306         GST_DEBUG_OBJECT (self, "socket error: %s", err->message);
 307         g_usleep (G_USEC_PER_SEC / 10); /* throttle */
 308       }
 309       g_clear_error (&err);
 310     } else {
 311       GstClockTime new_local;
 312
 313       /* got packet */
 314
 315       new_local = gst_clock_get_internal_time (GST_CLOCK (self));
 316
 317       packet = gst_net_time_packet_receive (socket, NULL, &err);
 318
 319       if (packet != NULL) {
 320         GST_LOG_OBJECT (self, "got packet back");
 321         GST_LOG_OBJECT (self, "local_1 = %" GST_TIME_FORMAT,
 322             GST_TIME_ARGS (packet->local_time));
 323         GST_LOG_OBJECT (self, "remote = %" GST_TIME_FORMAT,
 324             GST_TIME_ARGS (packet->remote_time));
 325         GST_LOG_OBJECT (self, "local_2 = %" GST_TIME_FORMAT,
 326             GST_TIME_ARGS (new_local));
 327
 328         /* observe_times will reset the timeout */
 329         gst_net_client_clock_observe_times (self, packet->local_time,
 330             packet->remote_time, new_local);
 331
 332         g_free (packet);
 333       } else if (err != NULL) {
 334         GST_WARNING_OBJECT (self, "receive error: %s", err->message);
 335         g_clear_error (&err);
 336       }
 337     }
 338   }
 339
 340   GST_INFO_OBJECT (self, "shutting down net client clock thread");
 341   return NULL;
 342 }
 343
 344 static gboolean
 345 gst_net_client_clock_start (GstNetClientClock * self)
 346 {
 347   GSocketAddress *servaddr;
 348   GSocketAddress *myaddr;
 349   GSocketAddress *anyaddr;
 350   GInetAddress *inetaddr;
 351   GSocket *socket;
 352   GError *error = NULL;
 353   GSocketFamily family;
 354
 355   g_return_val_if_fail (self->priv->address != NULL, FALSE);
 356   g_return_val_if_fail (self->priv->servaddr == NULL, FALSE);
 357
 358   /* create target address */
 359   inetaddr = g_inet_address_new_from_string (self->priv->address);
 360   if (inetaddr == NULL)
 361     goto bad_address;
 362
 363   family = g_inet_address_get_family (inetaddr);
 364
 365   servaddr = g_inet_socket_address_new (inetaddr, self->priv->port);
 366   g_object_unref (inetaddr);
 367
 368   g_assert (servaddr != NULL);
 369
 370   GST_DEBUG_OBJECT (self, "will communicate with %s:%d", self->priv->address,
 371       self->priv->port);
 372
 373   socket = g_socket_new (family, G_SOCKET_TYPE_DATAGRAM,
 374       G_SOCKET_PROTOCOL_UDP, &error);
 375
 376   if (socket == NULL)
 377     goto no_socket;
 378
 379   GST_DEBUG_OBJECT (self, "binding socket");
 380   inetaddr = g_inet_address_new_any (family);
 381   anyaddr = g_inet_socket_address_new (inetaddr, 0);
 382   g_socket_bind (socket, anyaddr, TRUE, &error);
 383   g_object_unref (anyaddr);
 384   g_object_unref (inetaddr);
 385
 386   if (error != NULL)
 387     goto bind_error;
 388
 389   /* check address we're bound to, mostly for debugging purposes */
 390   myaddr = g_socket_get_local_address (socket, &error);
 391
 392   if (myaddr == NULL)
 393     goto getsockname_error;
 394
 395   GST_DEBUG_OBJECT (self, "socket opened on UDP port %hd",
 396       g_inet_socket_address_get_port (G_INET_SOCKET_ADDRESS (myaddr)));
 397
 398   g_object_unref (myaddr);
 399
 400   self->priv->cancel = g_cancellable_new ();
 401   self->priv->socket = socket;
 402   self->priv->servaddr = G_SOCKET_ADDRESS (servaddr);
 403
 404   self->priv->thread = g_thread_try_new ("GstNetClientClock",
 405       gst_net_client_clock_thread, self, &error);
 406
 407   if (error != NULL)
 408     goto no_thread;
 409
 410   return TRUE;
 411
 412   /* ERRORS */
 413 no_socket:
 414   {
 415     GST_ERROR_OBJECT (self, "socket_new() failed: %s", error->message);
 416     g_error_free (error);
 417     return FALSE;
 418   }
 419 bind_error:
 420   {
 421     GST_ERROR_OBJECT (self, "bind failed: %s", error->message);
 422     g_error_free (error);
 423     g_object_unref (socket);
 424     return FALSE;
 425   }
 426 getsockname_error:
 427   {
 428     GST_ERROR_OBJECT (self, "get_local_address() failed: %s", error->message);
 429     g_error_free (error);
 430     g_object_unref (socket);
 431     return FALSE;
 432   }
 433 bad_address:
 434   {
 435     GST_ERROR_OBJECT (self, "inet_address_new_from_string('%s') failed",
 436         self->priv->address);
 437     return FALSE;
 438   }
 439 no_thread:
 440   {
 441     GST_ERROR_OBJECT (self, "could not create thread: %s", error->message);
 442     g_object_unref (self->priv->servaddr);
 443     self->priv->servaddr = NULL;
 444     g_object_unref (self->priv->socket);
 445     self->priv->socket = NULL;
 446     g_error_free (error);
 447     return FALSE;
 448   }
 449 }
 450
 451 static void
 452 gst_net_client_clock_stop (GstNetClientClock * self)
 453 {
 454   if (self->priv->thread == NULL)
 455     return;
 456
 457   GST_INFO_OBJECT (self, "stopping...");
 458   g_cancellable_cancel (self->priv->cancel);
 459
 460   g_thread_join (self->priv->thread);
 461   self->priv->thread = NULL;
 462
 463   g_object_unref (self->priv->cancel);
 464   self->priv->cancel = NULL;
 465
 466   g_object_unref (self->priv->servaddr);
 467   self->priv->servaddr = NULL;
 468
 469   g_object_unref (self->priv->socket);
 470   self->priv->socket = NULL;
 471
 472   GST_INFO_OBJECT (self, "stopped");
 473 }
 474
 475 /**
 476  * gst_net_client_clock_new:
 477  * @name: a name for the clock
 478  * @remote_address: the address of the remote clock provider
 479  * @remote_port: the port of the remote clock provider
 480  * @base_time: initial time of the clock
 481  *
 482  * Create a new #GstNetClientClock that will report the time
 483  * provided by the #GstNetTimeProvider on @remote_address and
 484  * @remote_port.
 485  *
 486  * Returns: a new #GstClock that receives a time from the remote
 487  * clock.
 488  */
 489 GstClock *
 490 gst_net_client_clock_new (const gchar * name, const gchar * remote_address,
 491     gint remote_port, GstClockTime base_time)
 492 {
 493   /* FIXME: gst_net_client_clock_new() should be a thin wrapper for g_object_new() */
 494   GstNetClientClock *ret;
 495   GstClockTime internal;
 496
 497   g_return_val_if_fail (remote_address != NULL, NULL);
 498   g_return_val_if_fail (remote_port > 0, NULL);
 499   g_return_val_if_fail (remote_port <= G_MAXUINT16, NULL);
 500   g_return_val_if_fail (base_time != GST_CLOCK_TIME_NONE, NULL);
 501
 502   ret = g_object_new (GST_TYPE_NET_CLIENT_CLOCK, "address", remote_address,
 503       "port", remote_port, NULL);
 504
 505   /* gst_clock_get_time() values are guaranteed to be increasing. because no one
 506    * has called get_time on this clock yet we are free to adjust to any value
 507    * without worrying about worrying about MAX() issues with the clock's
 508    * internal time.
 509    */
 510
 511   /* update our internal time so get_time() give something around base_time.
 512      assume that the rate is 1 in the beginning. */
 513   internal = gst_clock_get_internal_time (GST_CLOCK (ret));
 514   gst_clock_set_calibration (GST_CLOCK (ret), internal, base_time, 1, 1);
 515
 516   {
 517     GstClockTime now = gst_clock_get_time (GST_CLOCK (ret));
 518
 519     if (GST_CLOCK_DIFF (now, base_time) > 0 ||
 520         GST_CLOCK_DIFF (now, base_time + GST_SECOND) < 0) {
 521       g_warning ("unable to set the base time, expect sync problems!");
 522     }
 523   }
 524
 525   if (!gst_net_client_clock_start (ret))
 526     goto failed_start;
 527
 528   /* all systems go, cap'n */
 529   return (GstClock *) ret;
 530
 531 failed_start:
 532   {
 533     /* already printed a nice error */
 534     gst_object_unref (ret);
 535     return NULL;
 536   }
 537 }