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., 59 Temple Place - Suite 330,
  23  * Boston, MA 02111-1307, 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 typedef struct
 252 {
 253   GSource source;
 254   GstNetClientClock *clock;
 255   gboolean *p_timeout;
 256 } GstNetClientClockTimeoutSource;
 257
 258 static gboolean
 259 gst_net_client_clock_timeout_source_prepare (GSource * s, gint * p_timeout)
 260 {
 261   GstNetClientClockTimeoutSource *source = (GstNetClientClockTimeoutSource *) s;
 262   GstClockTime expiration_time = source->clock->priv->timeout_expiration;
 263   GstClockTime now = gst_util_get_timestamp ();
 264
 265   if (now >= expiration_time || (expiration_time - now) <= GST_MSECOND) {
 266     *p_timeout = 0;
 267     return TRUE;
 268   }
 269
 270   *p_timeout = (expiration_time - now) / GST_MSECOND;
 271   GST_TRACE_OBJECT (source->clock, "time out in %d ms please", *p_timeout);
 272   return FALSE;
 273 }
 274
 275 static gboolean
 276 gst_net_client_clock_timeout_source_check (GSource * s)
 277 {
 278   GstNetClientClockTimeoutSource *source = (GstNetClientClockTimeoutSource *) s;
 279
 280   return (gst_util_get_timestamp () >= source->clock->priv->timeout_expiration);
 281 }
 282
 283 static gboolean
 284 gst_net_client_clock_timeout_source_dispatch (GSource * s, GSourceFunc cb,
 285     gpointer data)
 286 {
 287   GstNetClientClockTimeoutSource *source = (GstNetClientClockTimeoutSource *) s;
 288
 289   GST_TRACE_OBJECT (source->clock, "timed out");
 290   *source->p_timeout = TRUE;
 291   return TRUE;
 292 }
 293
 294 static gboolean
 295 gst_net_client_clock_socket_cb (GSocket * socket, GIOCondition condition,
 296     gpointer user_data)
 297 {
 298   GIOCondition *p_cond = user_data;
 299
 300   GST_TRACE ("socket %p I/O condition: 0x%02x", socket, condition);
 301   *p_cond = condition;
 302   return TRUE;
 303 }
 304
 305 static gpointer
 306 gst_net_client_clock_thread (gpointer data)
 307 {
 308   GstNetClientClock *self = data;
 309   GstNetTimePacket *packet;
 310   GMainContext *ctx;
 311   GSourceFuncs funcs = { NULL, };
 312   GSource *source;
 313   GIOCondition cond;
 314   gboolean timeout;
 315   GSocket *socket = self->priv->socket;
 316   GError *err = NULL;
 317   GstClock *clock = data;
 318
 319   GST_INFO_OBJECT (self, "net client clock thread running, socket=%p", socket);
 320
 321   g_socket_set_blocking (socket, TRUE);
 322   g_socket_set_timeout (socket, 0);
 323
 324   ctx = g_main_context_new ();
 325
 326   source = g_socket_create_source (socket, G_IO_IN, self->priv->cancel);
 327   g_source_set_name (source, "GStreamer net client clock thread socket");
 328   g_source_set_callback (source, (GSourceFunc) gst_net_client_clock_socket_cb,
 329       &cond, NULL);
 330   g_source_attach (source, ctx);
 331   g_source_unref (source);
 332
 333   /* GSocket only support second granularity for timeouts, so roll our own
 334    * timeout source (so we don't have to create a new source whenever the
 335    * timeout changes, as we would have to do with the default timeout source) */
 336   funcs.prepare = gst_net_client_clock_timeout_source_prepare;
 337   funcs.check = gst_net_client_clock_timeout_source_check;
 338   funcs.dispatch = gst_net_client_clock_timeout_source_dispatch;
 339   funcs.finalize = NULL;
 340   source = g_source_new (&funcs, sizeof (GstNetClientClockTimeoutSource));
 341   ((GstNetClientClockTimeoutSource *) source)->clock = self;
 342   ((GstNetClientClockTimeoutSource *) source)->p_timeout = &timeout;
 343   g_source_set_name (source, "GStreamer net client clock timeout");
 344   g_source_attach (source, ctx);
 345   g_source_unref (source);
 346
 347   while (!g_cancellable_is_cancelled (self->priv->cancel)) {
 348     cond = 0;
 349     timeout = FALSE;
 350     g_main_context_iteration (ctx, TRUE);
 351
 352     if (g_cancellable_is_cancelled (self->priv->cancel))
 353       break;
 354
 355     if (timeout) {
 356       /* timed out, let's send another packet */
 357       GST_DEBUG_OBJECT (self, "timed out");
 358
 359       packet = gst_net_time_packet_new (NULL);
 360
 361       packet->local_time = gst_clock_get_internal_time (GST_CLOCK (self));
 362
 363       GST_DEBUG_OBJECT (self, "sending packet, local time = %" GST_TIME_FORMAT,
 364           GST_TIME_ARGS (packet->local_time));
 365
 366       gst_net_time_packet_send (packet, self->priv->socket,
 367           self->priv->servaddr, NULL);
 368
 369       g_free (packet);
 370
 371       /* reset timeout (but are expecting a response sooner anyway) */
 372       self->priv->timeout_expiration =
 373           gst_util_get_timestamp () + gst_clock_get_timeout (clock);
 374       continue;
 375     }
 376
 377     /* got data to read? */
 378     if ((cond & G_IO_IN)) {
 379       GstClockTime new_local;
 380
 381       new_local = gst_clock_get_internal_time (GST_CLOCK (self));
 382
 383       packet = gst_net_time_packet_receive (socket, NULL, &err);
 384
 385       if (err != NULL) {
 386         GST_WARNING_OBJECT (self, "receive error: %s", err->message);
 387         g_error_free (err);
 388         err = NULL;
 389         continue;
 390       }
 391
 392       GST_LOG_OBJECT (self, "got packet back");
 393       GST_LOG_OBJECT (self, "local_1 = %" GST_TIME_FORMAT,
 394           GST_TIME_ARGS (packet->local_time));
 395       GST_LOG_OBJECT (self, "remote = %" GST_TIME_FORMAT,
 396           GST_TIME_ARGS (packet->remote_time));
 397       GST_LOG_OBJECT (self, "local_2 = %" GST_TIME_FORMAT,
 398           GST_TIME_ARGS (new_local));
 399
 400       /* observe_times will reset the timeout */
 401       gst_net_client_clock_observe_times (self, packet->local_time,
 402           packet->remote_time, new_local);
 403
 404       g_free (packet);
 405       continue;
 406     }
 407
 408     if ((cond & (G_IO_ERR | G_IO_HUP))) {
 409       GST_DEBUG_OBJECT (self, "socket error?! %s", g_strerror (errno));
 410       g_usleep (G_USEC_PER_SEC / 10);
 411       continue;
 412     }
 413   }
 414
 415   GST_INFO_OBJECT (self, "shutting down net client clock thread");
 416   g_main_context_unref (ctx);
 417   return NULL;
 418 }
 419
 420 static gboolean
 421 gst_net_client_clock_start (GstNetClientClock * self)
 422 {
 423   GSocketAddress *servaddr;
 424   GSocketAddress *myaddr;
 425   GInetAddress *inetaddr;
 426   GSocket *socket;
 427   GError *error = NULL;
 428
 429   g_return_val_if_fail (self->priv->address != NULL, FALSE);
 430   g_return_val_if_fail (self->priv->servaddr == NULL, FALSE);
 431
 432   socket = g_socket_new (G_SOCKET_FAMILY_IPV4, G_SOCKET_TYPE_DATAGRAM,
 433       G_SOCKET_PROTOCOL_UDP, &error);
 434
 435   if (socket == NULL)
 436     goto no_socket;
 437
 438   /* check address we're bound to, mostly for debugging purposes */
 439   myaddr = g_socket_get_local_address (socket, &error);
 440
 441   if (myaddr == NULL)
 442     goto getsockname_error;
 443
 444   GST_DEBUG_OBJECT (self, "socket opened on UDP port %hd",
 445       g_inet_socket_address_get_port (G_INET_SOCKET_ADDRESS (myaddr)));
 446
 447   g_object_unref (myaddr);
 448
 449   /* create target address */
 450   inetaddr = g_inet_address_new_from_string (self->priv->address);
 451
 452   if (inetaddr == NULL)
 453     goto bad_address;
 454
 455   servaddr = g_inet_socket_address_new (inetaddr, self->priv->port);
 456   g_object_unref (inetaddr);
 457
 458   g_assert (servaddr != NULL);
 459
 460   GST_DEBUG_OBJECT (self, "will communicate with %s:%d", self->priv->address,
 461       self->priv->port);
 462
 463   self->priv->cancel = g_cancellable_new ();
 464   self->priv->socket = socket;
 465   self->priv->servaddr = G_SOCKET_ADDRESS (servaddr);
 466
 467   self->priv->thread = g_thread_try_new ("GstNetClientClock",
 468       gst_net_client_clock_thread, self, &error);
 469
 470   if (error != NULL)
 471     goto no_thread;
 472
 473   return TRUE;
 474
 475   /* ERRORS */
 476 no_socket:
 477   {
 478     GST_ERROR_OBJECT (self, "socket_new() failed: %s", error->message);
 479     g_error_free (error);
 480     return FALSE;
 481   }
 482 getsockname_error:
 483   {
 484     GST_ERROR_OBJECT (self, "get_local_address() failed: %s", error->message);
 485     g_error_free (error);
 486     g_object_unref (socket);
 487     return FALSE;
 488   }
 489 bad_address:
 490   {
 491     GST_ERROR_OBJECT (self, "inet_address_new_from_string('%s') failed",
 492         self->priv->address);
 493     g_object_unref (socket);
 494     return FALSE;
 495   }
 496 no_thread:
 497   {
 498     GST_ERROR_OBJECT (self, "could not create thread: %s", error->message);
 499     g_object_unref (self->priv->servaddr);
 500     self->priv->servaddr = NULL;
 501     g_object_unref (self->priv->socket);
 502     self->priv->socket = NULL;
 503     g_error_free (error);
 504     return FALSE;
 505   }
 506 }
 507
 508 static void
 509 gst_net_client_clock_stop (GstNetClientClock * self)
 510 {
 511   if (self->priv->thread == NULL)
 512     return;
 513
 514   GST_INFO_OBJECT (self, "stopping...");
 515   g_cancellable_cancel (self->priv->cancel);
 516
 517   g_thread_join (self->priv->thread);
 518   self->priv->thread = NULL;
 519
 520   g_object_unref (self->priv->cancel);
 521   self->priv->cancel = NULL;
 522
 523   g_object_unref (self->priv->servaddr);
 524   self->priv->servaddr = NULL;
 525
 526   g_object_unref (self->priv->socket);
 527   self->priv->socket = NULL;
 528
 529   GST_INFO_OBJECT (self, "stopped");
 530 }
 531
 532 /**
 533  * gst_net_client_clock_new:
 534  * @name: a name for the clock
 535  * @remote_address: the address of the remote clock provider
 536  * @remote_port: the port of the remote clock provider
 537  * @base_time: initial time of the clock
 538  *
 539  * Create a new #GstNetClientClock that will report the time
 540  * provided by the #GstNetTimeProvider on @remote_address and
 541  * @remote_port.
 542  *
 543  * Returns: a new #GstClock that receives a time from the remote
 544  * clock.
 545  */
 546 GstClock *
 547 gst_net_client_clock_new (gchar * name, const gchar * remote_address,
 548     gint remote_port, GstClockTime base_time)
 549 {
 550   /* FIXME: gst_net_client_clock_new() should be a thin wrapper for g_object_new() */
 551   GstNetClientClock *ret;
 552   GstClockTime internal;
 553
 554   g_return_val_if_fail (remote_address != NULL, NULL);
 555   g_return_val_if_fail (remote_port > 0, NULL);
 556   g_return_val_if_fail (remote_port <= G_MAXUINT16, NULL);
 557   g_return_val_if_fail (base_time != GST_CLOCK_TIME_NONE, NULL);
 558
 559   ret = g_object_new (GST_TYPE_NET_CLIENT_CLOCK, "address", remote_address,
 560       "port", remote_port, NULL);
 561
 562   /* gst_clock_get_time() values are guaranteed to be increasing. because no one
 563    * has called get_time on this clock yet we are free to adjust to any value
 564    * without worrying about worrying about MAX() issues with the clock's
 565    * internal time.
 566    */
 567
 568   /* update our internal time so get_time() give something around base_time.
 569      assume that the rate is 1 in the beginning. */
 570   internal = gst_clock_get_internal_time (GST_CLOCK (ret));
 571   gst_clock_set_calibration (GST_CLOCK (ret), internal, base_time, 1, 1);
 572
 573   {
 574     GstClockTime now = gst_clock_get_time (GST_CLOCK (ret));
 575
 576     if (GST_CLOCK_DIFF (now, base_time) > 0 ||
 577         GST_CLOCK_DIFF (now, base_time + GST_SECOND) < 0) {
 578       g_warning ("unable to set the base time, expect sync problems!");
 579     }
 580   }
 581
 582   if (!gst_net_client_clock_start (ret))
 583     goto failed_start;
 584
 585   /* all systems go, cap'n */
 586   return (GstClock *) ret;
 587
 588 failed_start:
 589   {
 590     /* already printed a nice error */
 591     gst_object_unref (ret);
 592     return NULL;
 593   }
 594 }