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  *
   6  * gstnetclientclock.h: clock that synchronizes itself to a time provider over
   7  * the network
   8  *
   9  * This library is free software; you can redistribute it and/or
  10  * modify it under the terms of the GNU Library General Public
  11  * License as published by the Free Software Foundation; either
  12  * version 2 of the License, or (at your option) any later version.
  13  *
  14  * This library is distributed in the hope that it will be useful,
  15  * but WITHOUT ANY WARRANTY; without even the implied warranty of
  16  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
  17  * Library General Public License for more details.
  18  *
  19  * You should have received a copy of the GNU Library General Public
  20  * License along with this library; if not, write to the
  21  * Free Software Foundation, Inc., 59 Temple Place - Suite 330,
  22  * Boston, MA 02111-1307, USA.
  23  */
  24 /**
  25  * SECTION:gstnetclientclock
  26  * @short_description: Special clock that synchronizes to a remote time
  27  *                     provider.
  28  * @see_also: #GstClock, #GstNetTimeProvider, #GstPipeline
  29  *
  30  * This object implements a custom #GstClock that synchronizes its time
  31  * to a remote time provider such as #GstNetTimeProvider.
  32  *
  33  * A new clock is created with gst_net_client_clock_new() which takes the
  34  * address and port of the remote time provider along with a name and
  35  * an initial time.
  36  *
  37  * This clock will poll the time provider and will update its calibration
  38  * parameters based on the local and remote observations.
  39  *
  40  * Various parameters of the clock can be configured with the parent #GstClock
  41  * "timeout", "window-size" and "window-threshold" object properties.
  42  *
  43  * A #GstNetClientClock is typically set on a #GstPipeline with
  44  * gst_pipeline_use_clock().
  45  *
  46  * Last reviewed on 2005-11-23 (0.9.5)
  47  */
  48
  49 #ifdef HAVE_CONFIG_H
  50 #include "config.h"
  51 #endif
  52
  53 #include "gstnettimepacket.h"
  54 #include "gstnetclientclock.h"
  55
  56 #ifdef HAVE_UNISTD_H
  57 #include <unistd.h>
  58 #endif
  59
  60 #if defined (_MSC_VER) && _MSC_VER >= 1400
  61 #include <io.h>
  62 #endif
  63
  64 GST_DEBUG_CATEGORY_STATIC (ncc_debug);
  65 #define GST_CAT_DEFAULT (ncc_debug)
  66
  67 #define DEFAULT_ADDRESS         "127.0.0.1"
  68 #define DEFAULT_PORT            5637
  69 #define DEFAULT_TIMEOUT         GST_SECOND
  70
  71 #ifdef G_OS_WIN32
  72 #define getsockname(sock,addr,len) getsockname(sock,addr,(int *)len)
  73 #endif
  74
  75 enum
  76 {
  77   PROP_0,
  78   PROP_ADDRESS,
  79   PROP_PORT,
  80 };
  81
  82 #define GST_NET_CLIENT_CLOCK_GET_PRIVATE(obj)  \
  83   (G_TYPE_INSTANCE_GET_PRIVATE ((obj), GST_TYPE_NET_CLIENT_CLOCK, GstNetClientClockPrivate))
  84
  85 struct _GstNetClientClockPrivate
  86 {
  87   GstPollFD sock;
  88   GstPoll *fdset;
  89 };
  90
  91 #define _do_init \
  92   GST_DEBUG_CATEGORY_INIT (ncc_debug, "netclock", 0, "Network client clock");
  93 #define gst_net_client_clock_parent_class parent_class
  94 G_DEFINE_TYPE_WITH_CODE (GstNetClientClock, gst_net_client_clock,
  95     GST_TYPE_SYSTEM_CLOCK, _do_init);
  96
  97 static void gst_net_client_clock_finalize (GObject * object);
  98 static void gst_net_client_clock_set_property (GObject * object, guint prop_id,
  99     const GValue * value, GParamSpec * pspec);
 100 static void gst_net_client_clock_get_property (GObject * object, guint prop_id,
 101     GValue * value, GParamSpec * pspec);
 102
 103 static void gst_net_client_clock_stop (GstNetClientClock * self);
 104
 105 #ifdef G_OS_WIN32
 106 static int
 107 inet_aton (const char *c, struct in_addr *paddr)
 108 {
 109   /* note that inet_addr is deprecated on unix because
 110    * inet_addr returns -1 (INADDR_NONE) for the valid 255.255.255.255
 111    * address. */
 112   paddr->s_addr = inet_addr (c);
 113   if (paddr->s_addr == INADDR_NONE)
 114     return 0;
 115
 116   return 1;
 117 }
 118 #endif
 119
 120 static void
 121 gst_net_client_clock_class_init (GstNetClientClockClass * klass)
 122 {
 123   GObjectClass *gobject_class;
 124
 125   gobject_class = G_OBJECT_CLASS (klass);
 126
 127   g_type_class_add_private (klass, sizeof (GstNetClientClockPrivate));
 128
 129   gobject_class->finalize = gst_net_client_clock_finalize;
 130   gobject_class->get_property = gst_net_client_clock_get_property;
 131   gobject_class->set_property = gst_net_client_clock_set_property;
 132
 133   g_object_class_install_property (gobject_class, PROP_ADDRESS,
 134       g_param_spec_string ("address", "address",
 135           "The address of the machine providing a time server, "
 136           "as a dotted quad (x.x.x.x)", DEFAULT_ADDRESS,
 137           G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 138   g_object_class_install_property (gobject_class, PROP_PORT,
 139       g_param_spec_int ("port", "port",
 140           "The port on which the remote server is listening", 0, G_MAXUINT16,
 141           DEFAULT_PORT, G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 142 }
 143
 144 static void
 145 gst_net_client_clock_init (GstNetClientClock * self)
 146 {
 147   GstClock *clock = GST_CLOCK_CAST (self);
 148
 149 #ifdef G_OS_WIN32
 150   WSADATA w;
 151   int error = WSAStartup (0x0202, &w);
 152
 153   if (error) {
 154     GST_DEBUG_OBJECT (self, "Error on WSAStartup");
 155   }
 156   if (w.wVersion != 0x0202) {
 157     WSACleanup ();
 158   }
 159 #endif
 160   self->priv = GST_NET_CLIENT_CLOCK_GET_PRIVATE (self);
 161
 162   self->port = DEFAULT_PORT;
 163   self->address = g_strdup (DEFAULT_ADDRESS);
 164
 165   clock->timeout = DEFAULT_TIMEOUT;
 166
 167   self->priv->sock.fd = -1;
 168   self->thread = NULL;
 169
 170   self->servaddr = NULL;
 171 }
 172
 173 static void
 174 gst_net_client_clock_finalize (GObject * object)
 175 {
 176   GstNetClientClock *self = GST_NET_CLIENT_CLOCK (object);
 177
 178   if (self->thread) {
 179     gst_net_client_clock_stop (self);
 180     g_assert (self->thread == NULL);
 181   }
 182
 183   if (self->priv->fdset) {
 184     gst_poll_free (self->priv->fdset);
 185     self->priv->fdset = NULL;
 186   }
 187
 188   g_free (self->address);
 189   self->address = NULL;
 190
 191   g_free (self->servaddr);
 192   self->servaddr = NULL;
 193
 194 #ifdef G_OS_WIN32
 195   WSACleanup ();
 196 #endif
 197
 198   G_OBJECT_CLASS (parent_class)->finalize (object);
 199 }
 200
 201 static void
 202 gst_net_client_clock_set_property (GObject * object, guint prop_id,
 203     const GValue * value, GParamSpec * pspec)
 204 {
 205   GstNetClientClock *self = GST_NET_CLIENT_CLOCK (object);
 206
 207   switch (prop_id) {
 208     case PROP_ADDRESS:
 209       g_free (self->address);
 210       if (g_value_get_string (value) == NULL)
 211         self->address = g_strdup (DEFAULT_ADDRESS);
 212       else
 213         self->address = g_strdup (g_value_get_string (value));
 214       break;
 215     case PROP_PORT:
 216       self->port = g_value_get_int (value);
 217       break;
 218     default:
 219       G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
 220       break;
 221   }
 222 }
 223
 224 static void
 225 gst_net_client_clock_get_property (GObject * object, guint prop_id,
 226     GValue * value, GParamSpec * pspec)
 227 {
 228   GstNetClientClock *self = GST_NET_CLIENT_CLOCK (object);
 229
 230   switch (prop_id) {
 231     case PROP_ADDRESS:
 232       g_value_set_string (value, self->address);
 233       break;
 234     case PROP_PORT:
 235       g_value_set_int (value, self->port);
 236       break;
 237     default:
 238       G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
 239       break;
 240   }
 241 }
 242
 243 static void
 244 gst_net_client_clock_observe_times (GstNetClientClock * self,
 245     GstClockTime local_1, GstClockTime remote, GstClockTime local_2)
 246 {
 247   GstClockTime local_avg;
 248   gdouble r_squared;
 249   GstClock *clock;
 250
 251   if (local_2 < local_1)
 252     goto bogus_observation;
 253
 254   local_avg = (local_2 + local_1) / 2;
 255
 256   clock = GST_CLOCK_CAST (self);
 257
 258   gst_clock_add_observation (GST_CLOCK (self), local_avg, remote, &r_squared);
 259
 260   GST_CLOCK_SLAVE_LOCK (self);
 261   if (clock->filling) {
 262     self->current_timeout = 0;
 263   } else {
 264     /* geto formula */
 265     self->current_timeout =
 266         (1e-3 / (1 - MIN (r_squared, 0.99999))) * GST_SECOND;
 267     self->current_timeout = MIN (self->current_timeout, clock->timeout);
 268   }
 269   GST_CLOCK_SLAVE_UNLOCK (clock);
 270
 271   return;
 272
 273 bogus_observation:
 274   {
 275     GST_WARNING_OBJECT (self, "time packet receive time < send time (%"
 276         GST_TIME_FORMAT " < %" GST_TIME_FORMAT ")", GST_TIME_ARGS (local_1),
 277         GST_TIME_ARGS (local_2));
 278     return;
 279   }
 280 }
 281
 282 static gint
 283 gst_net_client_clock_do_select (GstNetClientClock * self)
 284 {
 285   while (TRUE) {
 286     GstClockTime diff;
 287     gint ret;
 288
 289     GST_LOG_OBJECT (self, "doing select");
 290
 291     diff = gst_clock_get_internal_time (GST_CLOCK (self));
 292     ret = gst_poll_wait (self->priv->fdset, self->current_timeout);
 293     diff = gst_clock_get_internal_time (GST_CLOCK (self)) - diff;
 294
 295     if (diff > self->current_timeout)
 296       self->current_timeout = 0;
 297     else
 298       self->current_timeout -= diff;
 299
 300     GST_LOG_OBJECT (self, "select returned %d", ret);
 301
 302     if (ret < 0 && errno != EBUSY) {
 303       if (errno != EAGAIN && errno != EINTR)
 304         goto select_error;
 305       else
 306         continue;
 307     } else {
 308       return ret;
 309     }
 310
 311     g_assert_not_reached ();
 312
 313     /* log errors and keep going */
 314   select_error:
 315     {
 316       GST_WARNING_OBJECT (self, "select error %d: %s (%d)", ret,
 317           g_strerror (errno), errno);
 318       continue;
 319     }
 320   }
 321
 322   g_assert_not_reached ();
 323   return -1;
 324 }
 325
 326 static gpointer
 327 gst_net_client_clock_thread (gpointer data)
 328 {
 329   GstNetClientClock *self = data;
 330   struct sockaddr_in tmpaddr;
 331   socklen_t len;
 332   GstNetTimePacket *packet;
 333   gint ret;
 334   GstClock *clock = data;
 335
 336   while (TRUE) {
 337     ret = gst_net_client_clock_do_select (self);
 338
 339     if (ret < 0 && errno == EBUSY) {
 340       GST_LOG_OBJECT (self, "stop");
 341       goto stopped;
 342     } else if (ret == 0) {
 343       /* timed out, let's send another packet */
 344       GST_DEBUG_OBJECT (self, "timed out");
 345
 346       packet = gst_net_time_packet_new (NULL);
 347
 348       packet->local_time = gst_clock_get_internal_time (GST_CLOCK (self));
 349
 350       GST_DEBUG_OBJECT (self, "sending packet, local time = %" GST_TIME_FORMAT,
 351           GST_TIME_ARGS (packet->local_time));
 352       gst_net_time_packet_send (packet, self->priv->sock.fd,
 353           (struct sockaddr *) self->servaddr, sizeof (struct sockaddr_in));
 354
 355       g_free (packet);
 356
 357       /* reset timeout */
 358       self->current_timeout = clock->timeout;
 359       continue;
 360     } else if (gst_poll_fd_can_read (self->priv->fdset, &self->priv->sock)) {
 361       /* got data in */
 362       GstClockTime new_local = gst_clock_get_internal_time (GST_CLOCK (self));
 363
 364       len = sizeof (struct sockaddr);
 365       packet = gst_net_time_packet_receive (self->priv->sock.fd,
 366           (struct sockaddr *) &tmpaddr, &len);
 367
 368       if (!packet)
 369         goto receive_error;
 370
 371       GST_LOG_OBJECT (self, "got packet back");
 372       GST_LOG_OBJECT (self, "local_1 = %" GST_TIME_FORMAT,
 373           GST_TIME_ARGS (packet->local_time));
 374       GST_LOG_OBJECT (self, "remote = %" GST_TIME_FORMAT,
 375           GST_TIME_ARGS (packet->remote_time));
 376       GST_LOG_OBJECT (self, "local_2 = %" GST_TIME_FORMAT,
 377           GST_TIME_ARGS (new_local));
 378
 379       /* observe_times will reset the timeout */
 380       gst_net_client_clock_observe_times (self, packet->local_time,
 381           packet->remote_time, new_local);
 382
 383       g_free (packet);
 384       continue;
 385     } else {
 386       GST_WARNING_OBJECT (self, "unhandled select return state?");
 387       continue;
 388     }
 389
 390     g_assert_not_reached ();
 391
 392   stopped:
 393     {
 394       GST_DEBUG_OBJECT (self, "shutting down");
 395       /* socket gets closed in _stop() */
 396       return NULL;
 397     }
 398   receive_error:
 399     {
 400       GST_WARNING_OBJECT (self, "receive error");
 401       continue;
 402     }
 403
 404     g_assert_not_reached ();
 405
 406   }
 407
 408   g_assert_not_reached ();
 409
 410   return NULL;
 411 }
 412
 413 static gboolean
 414 gst_net_client_clock_start (GstNetClientClock * self)
 415 {
 416   struct sockaddr_in servaddr, myaddr;
 417   socklen_t len;
 418   gint ret;
 419   GError *error;
 420
 421   g_return_val_if_fail (self->address != NULL, FALSE);
 422   g_return_val_if_fail (self->servaddr == NULL, FALSE);
 423
 424   if ((ret = socket (AF_INET, SOCK_DGRAM, 0)) < 0)
 425     goto no_socket;
 426
 427   self->priv->sock.fd = ret;
 428
 429   len = sizeof (myaddr);
 430   ret = getsockname (self->priv->sock.fd, (struct sockaddr *) &myaddr, &len);
 431   if (ret < 0)
 432     goto getsockname_error;
 433
 434   memset (&servaddr, 0, sizeof (servaddr));
 435   servaddr.sin_family = AF_INET;        /* host byte order */
 436   servaddr.sin_port = htons (self->port);       /* short, network byte order */
 437
 438   GST_DEBUG_OBJECT (self, "socket opened on UDP port %hd",
 439       ntohs (servaddr.sin_port));
 440
 441   if (!inet_aton (self->address, &servaddr.sin_addr))
 442     goto bad_address;
 443
 444   self->servaddr = g_malloc (sizeof (struct sockaddr_in));
 445   memcpy (self->servaddr, &servaddr, sizeof (servaddr));
 446
 447   GST_DEBUG_OBJECT (self, "will communicate with %s:%d", self->address,
 448       self->port);
 449
 450   gst_poll_add_fd (self->priv->fdset, &self->priv->sock);
 451   gst_poll_fd_ctl_read (self->priv->fdset, &self->priv->sock, TRUE);
 452
 453   self->thread = g_thread_create (gst_net_client_clock_thread, self, TRUE,
 454       &error);
 455   if (!self->thread)
 456     goto no_thread;
 457
 458   return TRUE;
 459
 460   /* ERRORS */
 461 no_socket:
 462   {
 463     GST_ERROR_OBJECT (self, "socket failed %d: %s (%d)", ret,
 464         g_strerror (errno), errno);
 465     return FALSE;
 466   }
 467 getsockname_error:
 468   {
 469     GST_ERROR_OBJECT (self, "getsockname failed %d: %s (%d)", ret,
 470         g_strerror (errno), errno);
 471     close (self->priv->sock.fd);
 472     self->priv->sock.fd = -1;
 473     return FALSE;
 474   }
 475 bad_address:
 476   {
 477     GST_ERROR_OBJECT (self, "inet_aton failed %d: %s (%d)", ret,
 478         g_strerror (errno), errno);
 479     close (self->priv->sock.fd);
 480     self->priv->sock.fd = -1;
 481     return FALSE;
 482   }
 483 no_thread:
 484   {
 485     GST_ERROR_OBJECT (self, "could not create thread: %s", error->message);
 486     gst_poll_remove_fd (self->priv->fdset, &self->priv->sock);
 487     close (self->priv->sock.fd);
 488     self->priv->sock.fd = -1;
 489     g_free (self->servaddr);
 490     self->servaddr = NULL;
 491     g_error_free (error);
 492     return FALSE;
 493   }
 494 }
 495
 496 static void
 497 gst_net_client_clock_stop (GstNetClientClock * self)
 498 {
 499   gst_poll_set_flushing (self->priv->fdset, TRUE);
 500   g_thread_join (self->thread);
 501   self->thread = NULL;
 502
 503   if (self->priv->sock.fd != -1) {
 504     gst_poll_remove_fd (self->priv->fdset, &self->priv->sock);
 505     close (self->priv->sock.fd);
 506     self->priv->sock.fd = -1;
 507   }
 508 }
 509
 510 /**
 511  * gst_net_client_clock_new:
 512  * @name: a name for the clock
 513  * @remote_address: the address of the remote clock provider
 514  * @remote_port: the port of the remote clock provider
 515  * @base_time: initial time of the clock
 516  *
 517  * Create a new #GstNetClientClock that will report the time
 518  * provided by the #GstNetTimeProvider on @remote_address and
 519  * @remote_port.
 520  *
 521  * Returns: a new #GstClock that receives a time from the remote
 522  * clock.
 523  */
 524 GstClock *
 525 gst_net_client_clock_new (gchar * name, const gchar * remote_address,
 526     gint remote_port, GstClockTime base_time)
 527 {
 528   GstNetClientClock *ret;
 529   GstClockTime internal;
 530
 531   g_return_val_if_fail (remote_address != NULL, NULL);
 532   g_return_val_if_fail (remote_port > 0, NULL);
 533   g_return_val_if_fail (remote_port <= G_MAXUINT16, NULL);
 534   g_return_val_if_fail (base_time != GST_CLOCK_TIME_NONE, NULL);
 535
 536   ret = g_object_new (GST_TYPE_NET_CLIENT_CLOCK, "address", remote_address,
 537       "port", remote_port, NULL);
 538
 539   /* gst_clock_get_time() values are guaranteed to be increasing. because no one
 540    * has called get_time on this clock yet we are free to adjust to any value
 541    * without worrying about worrying about MAX() issues with the clock's
 542    * internal time.
 543    */
 544
 545   /* update our internal time so get_time() give something around base_time.
 546      assume that the rate is 1 in the beginning. */
 547   internal = gst_clock_get_internal_time (GST_CLOCK (ret));
 548   gst_clock_set_calibration (GST_CLOCK (ret), internal, base_time, 1, 1);
 549
 550   {
 551     GstClockTime now = gst_clock_get_time (GST_CLOCK (ret));
 552
 553     if (GST_CLOCK_DIFF (now, base_time) > 0 ||
 554         GST_CLOCK_DIFF (now, base_time + GST_SECOND) < 0) {
 555       g_warning ("unable to set the base time, expect sync problems!");
 556     }
 557   }
 558
 559   if ((ret->priv->fdset = gst_poll_new (TRUE)) == NULL)
 560     goto no_fdset;
 561
 562   if (!gst_net_client_clock_start (ret))
 563     goto failed_start;
 564
 565   /* all systems go, cap'n */
 566   return (GstClock *) ret;
 567
 568 no_fdset:
 569   {
 570     GST_ERROR_OBJECT (ret, "could not create an fdset: %s (%d)",
 571         g_strerror (errno), errno);
 572     gst_object_unref (ret);
 573     return NULL;
 574   }
 575 failed_start:
 576   {
 577     /* already printed a nice error */
 578     gst_object_unref (ret);
 579     return NULL;
 580   }
 581 }