From d56b83f6621a76090ee5511db4dcb27f8832b8b5 Mon Sep 17 00:00:00 2001 From: Andy Wingo Date: Thu, 17 Nov 2005 17:55:17 +0000 Subject: [PATCH] check/: Add a most minimal test for the net client clock. More to come later. Original commit message from CVS: 2005-11-17 Andy Wingo * check/Makefile.am (check_PROGRAMS): * check/net/gstnetclientclock.c: Add a most minimal test for the net client clock. More to come later. * gst/net/gstnet.h: * gst/net/Makefile.am: Add netclientclock. * gst/net/gstnetclientclock.h: * gst/net/gstnetclientclock.c: New files, implement an untested GstClock that takes its time from a network time provider. Implements the algorithm in network-clock.scm. * tests/network-clock.scm (*window-size*): Rename from *queue-length*. * tests/network-clock.scm (network-time): * tests/network-clock-utils.scm (q-push): Update callers. --- ChangeLog | 19 + check/Makefile.am | 4 + check/net/gstnetclientclock.c | 137 ++++++ check/net/gstnettimeprovider.c | 2 +- gst/net/Makefile.am | 2 + gst/net/gstnet.h | 1 + gst/net/gstnetclientclock.c | 592 ++++++++++++++++++++++++++ gst/net/gstnetclientclock.h | 99 +++++ libs/gst/net/Makefile.am | 2 + libs/gst/net/gstnet.h | 1 + libs/gst/net/gstnetclientclock.c | 592 ++++++++++++++++++++++++++ libs/gst/net/gstnetclientclock.h | 99 +++++ tests/check/Makefile.am | 4 + tests/check/libs/gstnetclientclock.c | 137 ++++++ tests/check/libs/gstnettimeprovider.c | 2 +- tests/misc/network-clock-utils.scm | 2 +- tests/misc/network-clock.scm | 4 +- tests/network-clock-utils.scm | 2 +- tests/network-clock.scm | 4 +- 19 files changed, 1697 insertions(+), 8 deletions(-) create mode 100644 check/net/gstnetclientclock.c create mode 100644 gst/net/gstnetclientclock.c create mode 100644 gst/net/gstnetclientclock.h create mode 100644 libs/gst/net/gstnetclientclock.c create mode 100644 libs/gst/net/gstnetclientclock.h create mode 100644 tests/check/libs/gstnetclientclock.c diff --git a/ChangeLog b/ChangeLog index f868a90ae6..25e05e12a6 100644 --- a/ChangeLog +++ b/ChangeLog @@ -1,3 +1,22 @@ +2005-11-17 Andy Wingo + + * check/Makefile.am (check_PROGRAMS): + * check/net/gstnetclientclock.c: Add a most minimal test for the + net client clock. More to come later. + + * gst/net/gstnet.h: + * gst/net/Makefile.am: Add netclientclock. + + * gst/net/gstnetclientclock.h: + * gst/net/gstnetclientclock.c: New files, implement an untested + GstClock that takes its time from a network time provider. + Implements the algorithm in network-clock.scm. + + * tests/network-clock.scm (*window-size*): Rename from + *queue-length*. + * tests/network-clock.scm (network-time): + * tests/network-clock-utils.scm (q-push): Update callers. + 2005-11-17 Wim Taymans * gst/gstbin.c: (gst_bin_provide_clock_func), diff --git a/check/Makefile.am b/check/Makefile.am index 9cc9d2de83..3dca798c4c 100644 --- a/check/Makefile.am +++ b/check/Makefile.am @@ -57,6 +57,7 @@ check_PROGRAMS = \ states/sinks \ gst-libs/controller \ gst-libs/gdp \ + net/gstnetclientclock \ net/gstnettimeprovider TESTS = $(check_PROGRAMS) @@ -79,6 +80,9 @@ gst_libs_controller_LDADD = \ $(top_builddir)/libs/gst/controller/libgstcontroller-@GST_MAJORMINOR@.la \ $(LDADD) +net_gstnetclientclock_LDADD = \ + $(top_builddir)/gst/net/libgstnet-tempname-@GST_MAJORMINOR@.la \ + $(LDADD) net_gstnettimeprovider_LDADD = \ $(top_builddir)/gst/net/libgstnet-tempname-@GST_MAJORMINOR@.la \ $(LDADD) diff --git a/check/net/gstnetclientclock.c b/check/net/gstnetclientclock.c new file mode 100644 index 0000000000..005d02f388 --- /dev/null +++ b/check/net/gstnetclientclock.c @@ -0,0 +1,137 @@ +/* GStreamer + * Copyright (C) 2005 Andy Wingo + * + * gstnetclientclock.c: Unit test for the network client clock + * + * 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., 59 Temple Place - Suite 330, + * Boston, MA 02111-1307, USA. + */ + +#include +#include + +#include + +GST_START_TEST (test_instantiation) +{ + GstClock *client, *local; + + local = gst_system_clock_obtain (); + client = gst_net_client_clock_new (NULL, "127.0.0.1", 1234, GST_SECOND); + fail_unless (local != NULL, "failed to get system clock"); + fail_unless (client != NULL, "failed to get network client clock"); + + /* one for gstreamer, one for us */ + ASSERT_OBJECT_REFCOUNT (local, "system clock", 2); + ASSERT_OBJECT_REFCOUNT (client, "network client clock", 1); + + gst_object_unref (client); + + ASSERT_OBJECT_REFCOUNT (local, "net time provider", 2); + + gst_object_unref (local); +} + +GST_END_TEST; + +#if 0 +GST_START_TEST (test_functioning) +{ + GstNetTimeProvider *ntp; + GstNetTimePacket *packet; + GstClock *clock; + GstClockTime local; + struct sockaddr_in servaddr; + gint port = -1, sockfd, ret; + socklen_t len; + + clock = gst_system_clock_obtain (); + fail_unless (clock != NULL, "failed to get system clock"); + ntp = gst_net_time_provider_new (clock, "127.0.0.1", -1); + fail_unless (ntp != NULL, "failed to create net time provider"); + + g_object_get (ntp, "port", &port, NULL); + fail_unless (port > 0); + + sockfd = socket (AF_INET, SOCK_DGRAM, 0); + fail_if (sockfd < 0, "socket failed"); + + memset (&servaddr, 0, sizeof (servaddr)); + servaddr.sin_family = AF_INET; + servaddr.sin_port = htons (port); + inet_aton ("127.0.0.1", &servaddr.sin_addr); + + packet = gst_net_time_packet_new (NULL); + fail_unless (packet != NULL, "failed to create packet"); + + packet->local_time = local = gst_clock_get_time (clock); + + len = sizeof (servaddr); + ret = gst_net_time_packet_send (packet, sockfd, + (struct sockaddr *) &servaddr, len); + + fail_unless (ret == GST_NET_TIME_PACKET_SIZE, "failed to send packet"); + + g_free (packet); + + packet = gst_net_time_packet_receive (sockfd, (struct sockaddr *) &servaddr, + &len); + + fail_unless (packet != NULL, "failed to receive packet"); + fail_unless (packet->local_time == local, "local time is not the same"); + fail_unless (packet->remote_time > local, "remote time not after local time"); + fail_unless (packet->remote_time < gst_clock_get_time (clock), + "remote time in the future"); + + g_free (packet); + + close (sockfd); + + gst_object_unref (ntp); + gst_object_unref (clock); +} + +GST_END_TEST; +#endif + +Suite * +gst_net_client_clock_suite (void) +{ + Suite *s = suite_create ("GstNetClientClock"); + TCase *tc_chain = tcase_create ("generic tests"); + + suite_add_tcase (s, tc_chain); + tcase_add_test (tc_chain, test_instantiation); + /* tcase_add_test (tc_chain, test_functioning); */ + + return s; +} + +int +main (int argc, char **argv) +{ + int nf; + + Suite *s = gst_net_client_clock_suite (); + SRunner *sr = srunner_create (s); + + gst_check_init (&argc, &argv); + + srunner_run_all (sr, CK_NORMAL); + nf = srunner_ntests_failed (sr); + srunner_free (sr); + + return nf; +} diff --git a/check/net/gstnettimeprovider.c b/check/net/gstnettimeprovider.c index da1e93a078..e005008d22 100644 --- a/check/net/gstnettimeprovider.c +++ b/check/net/gstnettimeprovider.c @@ -74,7 +74,7 @@ GST_START_TEST (test_functioning) memset (&servaddr, 0, sizeof (servaddr)); servaddr.sin_family = AF_INET; - servaddr.sin_port = port; + servaddr.sin_port = htons (port); inet_aton ("127.0.0.1", &servaddr.sin_addr); packet = gst_net_time_packet_new (NULL); diff --git a/gst/net/Makefile.am b/gst/net/Makefile.am index 2f09ab8523..62f1ebffe8 100644 --- a/gst/net/Makefile.am +++ b/gst/net/Makefile.am @@ -5,10 +5,12 @@ lib_LTLIBRARIES = libgstnet-tempname-@GST_MAJORMINOR@.la libgstnet_tempname_@GST_MAJORMINOR@_includedir = $(includedir)/gstreamer-@GST_MAJORMINOR@/gst/net libgstnet_tempname_@GST_MAJORMINOR@_include_HEADERS = \ gstnet.h \ + gstnetclientclock.h \ gstnettimepacket.h \ gstnettimeprovider.h libgstnet_tempname_@GST_MAJORMINOR@_la_SOURCES = \ + gstnetclientclock.c \ gstnettimepacket.c \ gstnettimeprovider.c diff --git a/gst/net/gstnet.h b/gst/net/gstnet.h index 7d28228bac..078cad59d1 100644 --- a/gst/net/gstnet.h +++ b/gst/net/gstnet.h @@ -22,6 +22,7 @@ #define __GST_NET_H__ +#include #include #include diff --git a/gst/net/gstnetclientclock.c b/gst/net/gstnetclientclock.c new file mode 100644 index 0000000000..aa4c7d33e2 --- /dev/null +++ b/gst/net/gstnetclientclock.c @@ -0,0 +1,592 @@ +/* GStreamer + * Copyright (C) 1999,2000 Erik Walthinsen + * 2005 Wim Taymans + * 2005 Andy Wingo + * + * 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., 59 Temple Place - Suite 330, + * Boston, MA 02111-1307, USA. + */ + +#ifdef HAVE_CONFIG_H +#include "config.h" +#endif + +#include + +#include "gstnettimepacket.h" +#include "gstnetclientclock.h" + +GST_DEBUG_CATEGORY (ncc_debug); +#define GST_CAT_DEFAULT (ncc_debug) + +/* the select call is also performed on the control sockets, that way + * we can send special commands to unblock or restart the select call */ +#define CONTROL_RESTART 'R' /* restart the select call */ +#define CONTROL_STOP 'S' /* stop the select call */ +#define CONTROL_SOCKETS(self) self->control_sock +#define WRITE_SOCKET(self) self->control_sock[1] +#define READ_SOCKET(self) self->control_sock[0] + +#define SEND_COMMAND(self, command) \ +G_STMT_START { \ + unsigned char c; c = command; \ + write (WRITE_SOCKET(self), &c, 1); \ +} G_STMT_END + +#define READ_COMMAND(self, command, res) \ +G_STMT_START { \ + res = read(READ_SOCKET(self), &command, 1); \ +} G_STMT_END + +#define DEFAULT_ADDRESS "127.0.0.1" +#define DEFAULT_PORT 5637 +#define DEFAULT_WINDOW_SIZE 32 +#define DEFAULT_TIMEOUT GST_SECOND + +enum +{ + PROP_0, + PROP_ADDRESS, + PROP_PORT, + PROP_WINDOW_SIZE, + PROP_TIMEOUT +}; + +#define _do_init(type) \ + GST_DEBUG_CATEGORY_INIT (ncc_debug, "netclock", 0, "Network client clock"); + +GST_BOILERPLATE_FULL (GstNetClientClock, gst_net_client_clock, + GstSystemClock, 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_stop (GstNetClientClock * self); + +static void +gst_net_client_clock_base_init (gpointer g_class) +{ + /* nop */ +} + +static void +gst_net_client_clock_class_init (GstNetClientClockClass * klass) +{ + GObjectClass *gobject_class; + + gobject_class = (GObjectClass *) klass; + + 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; + + g_object_class_install_property (gobject_class, PROP_ADDRESS, + g_param_spec_string ("address", "address", + "The address of the machine providing a time server, " + "as a dotted quad (x.x.x.x)", DEFAULT_ADDRESS, G_PARAM_READWRITE)); + g_object_class_install_property (G_OBJECT_CLASS (klass), PROP_PORT, + g_param_spec_int ("port", "port", + "The port on which the remote server is listenind", 0, 32768, + DEFAULT_PORT, G_PARAM_READWRITE)); + g_object_class_install_property (G_OBJECT_CLASS (klass), PROP_WINDOW_SIZE, + g_param_spec_int ("window-size", "Window size", + "The size of the window used to calculate rate and offset", 2, 1024, + DEFAULT_WINDOW_SIZE, G_PARAM_READWRITE)); + g_object_class_install_property (G_OBJECT_CLASS (klass), PROP_TIMEOUT, + g_param_spec_uint64 ("timeout", "Timeout", + "The amount of time, in nanoseconds, to wait for replies", 0, + G_MAXUINT64, DEFAULT_TIMEOUT, G_PARAM_READWRITE)); +} + +static void +gst_net_client_clock_init (GstNetClientClock * self, + GstNetClientClockClass * g_class) +{ + self->port = DEFAULT_PORT; + self->address = g_strdup (DEFAULT_ADDRESS); + self->window_size = DEFAULT_WINDOW_SIZE; + self->timeout = DEFAULT_TIMEOUT; + + self->sock = -1; + self->thread = NULL; + + self->filling = TRUE; + self->time_index = 0; + self->local_times = g_new0 (GstClockTime, self->window_size); + self->remote_times = g_new0 (GstClockTime, self->window_size); + + self->servaddr = NULL; + + READ_SOCKET (self) = -1; + WRITE_SOCKET (self) = -1; +} + +static void +gst_net_client_clock_finalize (GObject * object) +{ + GstNetClientClock *self = GST_NET_CLIENT_CLOCK (object); + + if (self->thread) { + gst_net_client_clock_stop (self); + g_assert (self->thread == NULL); + } + + if (READ_SOCKET (self) != -1) { + close (READ_SOCKET (self)); + close (WRITE_SOCKET (self)); + READ_SOCKET (self) = -1; + WRITE_SOCKET (self) = -1; + } + + g_free (self->address); + self->address = NULL; + + g_free (self->servaddr); + self->servaddr = 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: + g_free (self->address); + if (g_value_get_string (value) == NULL) + self->address = g_strdup (DEFAULT_ADDRESS); + else + self->address = g_strdup (g_value_get_string (value)); + break; + case PROP_PORT: + self->port = g_value_get_int (value); + break; + case PROP_WINDOW_SIZE: + self->window_size = g_value_get_int (value); + break; + case PROP_TIMEOUT: + self->timeout = 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: + g_value_set_string (value, self->address); + break; + case PROP_PORT: + g_value_set_int (value, self->port); + break; + case PROP_WINDOW_SIZE: + g_value_set_int (value, self->window_size); + break; + case PROP_TIMEOUT: + g_value_set_uint64 (value, self->timeout); + break; + default: + G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec); + break; + } +} + +/* http://mathworld.wolfram.com/LeastSquaresFitting.html */ +static void +do_linear_regression (GstClockTime * x, GstClockTime * y, gint n, gdouble * m, + gdouble * b, gdouble * r_squared) +{ + gdouble sxx, syy, sxy, xbar, ybar; + gint i; + + /* we lose precision with the doubles, but there's no other way except to use + ~80 bit arithmetic */ + for (i = 0; i < n; i++) { + xbar += x[i]; + ybar += y[i]; + } + xbar /= n; + ybar /= n; + + for (i = 0; i < n; i++) { + sxx += ((double) x[i]) * x[i]; + syy += ((double) y[i]) * y[i]; + sxy += ((double) x[i]) * y[i]; + } + sxx -= ((double) n) * xbar * xbar; + syy -= ((double) n) * ybar * ybar; + sxy -= ((double) n) * xbar * ybar; + + *m = sxy / sxx; + *b = ybar - xbar * *m; + *r_squared = (sxy * sxy) / (sxx * syy); +} + +static void +gst_net_client_clock_observe_times (GstNetClientClock * self, + GstClockTime local_1, GstClockTime remote, GstClockTime local_2) +{ + GstClockTime local_avg; + gdouble m, b, r_squared; + + if (local_2 < local_1) + goto bogus_observation; + + local_avg = (local_2 - local_1) / 2; + + self->local_times[self->time_index] = local_avg; + self->remote_times[self->time_index] = remote; + + self->time_index++; + if (self->time_index == self->window_size) { + self->filling = FALSE; + self->time_index = 0; + } + + if (!self->filling || self->time_index >= 4) { + /* need to allow tuning of the "4" parameter -- means that we need 4 samples + * before beginning to adjust the clock */ + do_linear_regression (self->local_times, self->remote_times, + self->filling ? self->time_index : self->window_size, &m, &b, + &r_squared); + + GST_LOG_OBJECT (self, "adjusting clock to m=%g, b=%g (rsquared=%g)", m, b, + r_squared); + + gst_clock_set_rate_offset (GST_CLOCK (self), m, b); + } + + if (self->filling) { + self->current_timeout = 0; + } else { + /* geto formula */ + self->current_timeout = + (1e-3 / (1 - MIN (r_squared, 0.99999))) * GST_SECOND; + self->current_timeout = MIN (self->current_timeout, self->timeout); + } + + return; + +bogus_observation: + { + GST_WARNING_OBJECT (self, "time packet receive time < send time (%", + GST_TIME_FORMAT, " < %" GST_TIME_FORMAT ")", GST_TIME_ARGS (local_1), + GST_TIME_ARGS (local_2)); + return; + } +} + +static gint +gst_net_client_clock_do_select (GstNetClientClock * self, fd_set * readfds) +{ + guint max_sock; + gint ret; + + while (TRUE) { + FD_ZERO (readfds); + FD_SET (self->sock, readfds); + FD_SET (READ_SOCKET (self), readfds); + max_sock = MAX (self->sock, READ_SOCKET (self)); + + GST_LOG_OBJECT (self, "doing select"); + { + GstClockTime diff; + GTimeVal tv; + + diff = gst_clock_get_internal_time (GST_CLOCK (self)); + GST_TIME_TO_TIMEVAL (self->current_timeout, tv); + + ret = select (max_sock + 1, readfds, NULL, NULL, (struct timeval *) &tv); + + diff = gst_clock_get_internal_time (GST_CLOCK (self)) - diff; + + if (diff > self->current_timeout) + self->current_timeout = 0; + else + self->current_timeout -= diff; + } + GST_LOG_OBJECT (self, "select returned %d", ret); + + if (ret < 0) { + if (errno != EAGAIN && errno != EINTR) + goto select_error; + else + continue; + } else { + return ret; + } + + g_assert_not_reached (); + + /* log errors and keep going */ + select_error: + { + GST_WARNING_OBJECT (self, "select error %d: %s (%d)", ret, + g_strerror (errno), errno); + continue; + } + } + + g_assert_not_reached (); + return -1; +} + +static gpointer +gst_net_client_clock_thread (gpointer data) +{ + GstNetClientClock *self = data; + struct sockaddr_in tmpaddr; + socklen_t len; + fd_set read_fds; + GstNetTimePacket *packet; + gint ret; + + while (TRUE) { + ret = gst_net_client_clock_do_select (self, &read_fds); + + if (FD_ISSET (READ_SOCKET (self), &read_fds)) { + /* got control message */ + while (TRUE) { + gchar command; + int res; + + READ_COMMAND (self, command, res); + if (res < 0) { + GST_LOG_OBJECT (self, "no more commands"); + break; + } + + switch (command) { + case CONTROL_STOP: + /* break out of the select loop */ + GST_LOG_OBJECT (self, "stop"); + goto stopped; + default: + GST_WARNING_OBJECT (self, "unknown message: '%c'", command); + g_warning ("netclientclock: unknown control message received"); + continue; + } + + g_assert_not_reached (); + } + + continue; + } else if (ret == 0) { + /* timed out, let's send another packet */ + packet = gst_net_time_packet_new (NULL); + + packet->local_time = gst_clock_get_internal_time (GST_CLOCK (self)); + + gst_net_time_packet_send (packet, self->sock, + (struct sockaddr *) self->servaddr, sizeof (struct sockaddr_in)); + + g_free (packet); + + /* reset timeout */ + self->current_timeout = self->timeout; + } else if (FD_ISSET (READ_SOCKET (self), &read_fds)) { + /* got data in */ + GstClockTime new_local = gst_clock_get_internal_time (GST_CLOCK (self)); + + len = sizeof (struct sockaddr); + packet = gst_net_time_packet_receive (self->sock, + (struct sockaddr *) &tmpaddr, &len); + + if (!packet) + goto receive_error; + + /* observe_times will reset the timeout */ + gst_net_client_clock_observe_times (self, packet->local_time, + packet->remote_time, new_local); + + g_free (packet); + continue; + } else { + GST_WARNING_OBJECT (self, "unhandled select return state?"); + continue; + } + + g_assert_not_reached (); + + stopped: + { + GST_DEBUG_OBJECT (self, "shutting down"); + /* close socket */ + return NULL; + } + receive_error: + { + GST_WARNING_OBJECT (self, "receive error"); + continue; + } + + g_assert_not_reached (); + + } + + g_assert_not_reached (); + + return NULL; +} + +static gboolean +gst_net_client_clock_start (GstNetClientClock * self) +{ + struct sockaddr_in servaddr, myaddr; + socklen_t len; + gint ret; + GError *error; + + g_return_val_if_fail (self->address != NULL, FALSE); + g_return_val_if_fail (self->servaddr == NULL, FALSE); + + if ((ret = socket (AF_INET, SOCK_DGRAM, 0)) < 0) + goto no_socket; + + self->sock = ret; + + len = sizeof (myaddr); + ret = getsockname (self->sock, (struct sockaddr *) &myaddr, &len); + if (ret < 0) + goto getsockname_error; + + GST_DEBUG_OBJECT (self, "socket opened on UDP port %hd", + ntohs (servaddr.sin_port)); + + memset (&servaddr, 0, sizeof (servaddr)); + servaddr.sin_family = AF_INET; /* host byte order */ + servaddr.sin_port = htons (self->port); /* short, network byte order */ + if (!inet_aton (self->address, &servaddr.sin_addr)) + goto bad_address; + + self->servaddr = g_malloc (sizeof (struct sockaddr_in)); + memcpy (self->servaddr, &servaddr, sizeof (servaddr)); + + GST_DEBUG_OBJECT (self, "will communicate with %s:%d", self->address, + self->port); + + self->thread = g_thread_create (gst_net_client_clock_thread, self, TRUE, + &error); + if (!self->thread) + goto no_thread; + + return TRUE; + + /* ERRORS */ +no_socket: + { + GST_ERROR_OBJECT (self, "socket failed %d: %s (%d)", ret, + g_strerror (errno), errno); + return FALSE; + } +getsockname_error: + { + GST_ERROR_OBJECT (self, "getsockname failed %d: %s (%d)", ret, + g_strerror (errno), errno); + close (self->sock); + self->sock = -1; + return FALSE; + } +bad_address: + { + GST_ERROR_OBJECT (self, "inet_aton failed %d: %s (%d)", ret, + g_strerror (errno), errno); + close (self->sock); + self->sock = -1; + return FALSE; + } +no_thread: + { + GST_ERROR_OBJECT (self, "could not create thread: %s", error->message); + close (self->sock); + self->sock = -1; + g_free (self->servaddr); + self->servaddr = NULL; + g_error_free (error); + return FALSE; + } +} + +static void +gst_net_client_clock_stop (GstNetClientClock * self) +{ + SEND_COMMAND (self, CONTROL_STOP); + g_thread_join (self->thread); + self->thread = NULL; + + if (self->sock != -1) { + close (self->sock); + self->sock = -1; + } +} + +GstClock * +gst_net_client_clock_new (gchar * name, const gchar * remote_address, + gint remote_port, GstClockTime base_time) +{ + GstNetClientClock *ret; + gint iret; + + 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 < 32768, 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, NULL); + + GST_DEBUG_OBJECT (ret, "creating socket pair"); + if ((iret = socketpair (PF_UNIX, SOCK_STREAM, 0, CONTROL_SOCKETS (ret))) < 0) + goto no_socket_pair; + + fcntl (READ_SOCKET (ret), F_SETFL, O_NONBLOCK); + fcntl (WRITE_SOCKET (ret), F_SETFL, O_NONBLOCK); + + if (!gst_net_client_clock_start (ret)) + goto failed_start; + + /* all systems go, cap'n */ + return (GstClock *) ret; + +no_socket_pair: + { + GST_ERROR_OBJECT (ret, "no socket pair %d: %s (%d)", iret, + g_strerror (errno), errno); + gst_object_unref (ret); + return NULL; + } +failed_start: + { + /* already printed a nice error */ + gst_object_unref (ret); + return NULL; + } +} diff --git a/gst/net/gstnetclientclock.h b/gst/net/gstnetclientclock.h new file mode 100644 index 0000000000..0997888a41 --- /dev/null +++ b/gst/net/gstnetclientclock.h @@ -0,0 +1,99 @@ +/* GStreamer + * Copyright (C) 1999,2000 Erik Walthinsen + * 2005 Wim Taymans + * 2005 Andy Wingo + * + * 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., 59 Temple Place - Suite 330, + * Boston, MA 02111-1307, USA. + */ + + +#ifndef __GST_NET_CLIENT_CLOCK_H__ +#define __GST_NET_CLIENT_CLOCK_H__ + +#include +#include + +G_BEGIN_DECLS + +#include +#include +#include +#include +#include +#include +#include + +#include + +#define GST_TYPE_NET_CLIENT_CLOCK \ + (gst_net_client_clock_get_type()) +#define GST_NET_CLIENT_CLOCK(obj) \ + (G_TYPE_CHECK_INSTANCE_CAST((obj),GST_TYPE_NET_CLIENT_CLOCK,GstNetClientClock)) +#define GST_NET_CLIENT_CLOCK_CLASS(klass) \ + (G_TYPE_CHECK_CLASS_CAST((klass),GST_TYPE_NET_CLIENT_CLOCK,GstNetClientClockClass)) +#define GST_IS_NET_CLIENT_CLOCK(obj) \ + (G_TYPE_CHECK_INSTANCE_TYPE((obj),GST_TYPE_NET_CLIENT_CLOCK)) +#define GST_IS_NET_CLIENT_CLOCK_CLASS(obj) \ + (G_TYPE_CHECK_CLASS_TYPE((klass),GST_TYPE_NET_CLIENT_CLOCK)) + +typedef struct _GstNetClientClock GstNetClientClock; +typedef struct _GstNetClientClockClass GstNetClientClockClass; + +struct _GstNetClientClock { + GstSystemClock clock; + + /*< protected >*/ + gchar *address; + gint port; + + gint window_size; + GstClockTime timeout; + + /*< private >*/ + int sock; + int control_sock[2]; + + GstClockTime current_timeout; + + gboolean filling; + gint time_index; + GstClockTime *local_times; + GstClockTime *remote_times; + + struct sockaddr_id *servaddr; + + GThread *thread; + + gpointer _gst_reserved[GST_PADDING]; +}; + +struct _GstNetClientClockClass { + GstSystemClockClass parent_class; + + /*< private >*/ + gpointer _gst_reserved[GST_PADDING]; +}; + +GType gst_net_client_clock_get_type (void); +GstClock* gst_net_client_clock_new (gchar *name, const gchar *remote_address, + gint remote_port, GstClockTime base_time); + +G_END_DECLS + +#endif /* __GST_NET_CLIENT_CLOCK_H__ */ diff --git a/libs/gst/net/Makefile.am b/libs/gst/net/Makefile.am index 2f09ab8523..62f1ebffe8 100644 --- a/libs/gst/net/Makefile.am +++ b/libs/gst/net/Makefile.am @@ -5,10 +5,12 @@ lib_LTLIBRARIES = libgstnet-tempname-@GST_MAJORMINOR@.la libgstnet_tempname_@GST_MAJORMINOR@_includedir = $(includedir)/gstreamer-@GST_MAJORMINOR@/gst/net libgstnet_tempname_@GST_MAJORMINOR@_include_HEADERS = \ gstnet.h \ + gstnetclientclock.h \ gstnettimepacket.h \ gstnettimeprovider.h libgstnet_tempname_@GST_MAJORMINOR@_la_SOURCES = \ + gstnetclientclock.c \ gstnettimepacket.c \ gstnettimeprovider.c diff --git a/libs/gst/net/gstnet.h b/libs/gst/net/gstnet.h index 7d28228bac..078cad59d1 100644 --- a/libs/gst/net/gstnet.h +++ b/libs/gst/net/gstnet.h @@ -22,6 +22,7 @@ #define __GST_NET_H__ +#include #include #include diff --git a/libs/gst/net/gstnetclientclock.c b/libs/gst/net/gstnetclientclock.c new file mode 100644 index 0000000000..aa4c7d33e2 --- /dev/null +++ b/libs/gst/net/gstnetclientclock.c @@ -0,0 +1,592 @@ +/* GStreamer + * Copyright (C) 1999,2000 Erik Walthinsen + * 2005 Wim Taymans + * 2005 Andy Wingo + * + * 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., 59 Temple Place - Suite 330, + * Boston, MA 02111-1307, USA. + */ + +#ifdef HAVE_CONFIG_H +#include "config.h" +#endif + +#include + +#include "gstnettimepacket.h" +#include "gstnetclientclock.h" + +GST_DEBUG_CATEGORY (ncc_debug); +#define GST_CAT_DEFAULT (ncc_debug) + +/* the select call is also performed on the control sockets, that way + * we can send special commands to unblock or restart the select call */ +#define CONTROL_RESTART 'R' /* restart the select call */ +#define CONTROL_STOP 'S' /* stop the select call */ +#define CONTROL_SOCKETS(self) self->control_sock +#define WRITE_SOCKET(self) self->control_sock[1] +#define READ_SOCKET(self) self->control_sock[0] + +#define SEND_COMMAND(self, command) \ +G_STMT_START { \ + unsigned char c; c = command; \ + write (WRITE_SOCKET(self), &c, 1); \ +} G_STMT_END + +#define READ_COMMAND(self, command, res) \ +G_STMT_START { \ + res = read(READ_SOCKET(self), &command, 1); \ +} G_STMT_END + +#define DEFAULT_ADDRESS "127.0.0.1" +#define DEFAULT_PORT 5637 +#define DEFAULT_WINDOW_SIZE 32 +#define DEFAULT_TIMEOUT GST_SECOND + +enum +{ + PROP_0, + PROP_ADDRESS, + PROP_PORT, + PROP_WINDOW_SIZE, + PROP_TIMEOUT +}; + +#define _do_init(type) \ + GST_DEBUG_CATEGORY_INIT (ncc_debug, "netclock", 0, "Network client clock"); + +GST_BOILERPLATE_FULL (GstNetClientClock, gst_net_client_clock, + GstSystemClock, 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_stop (GstNetClientClock * self); + +static void +gst_net_client_clock_base_init (gpointer g_class) +{ + /* nop */ +} + +static void +gst_net_client_clock_class_init (GstNetClientClockClass * klass) +{ + GObjectClass *gobject_class; + + gobject_class = (GObjectClass *) klass; + + 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; + + g_object_class_install_property (gobject_class, PROP_ADDRESS, + g_param_spec_string ("address", "address", + "The address of the machine providing a time server, " + "as a dotted quad (x.x.x.x)", DEFAULT_ADDRESS, G_PARAM_READWRITE)); + g_object_class_install_property (G_OBJECT_CLASS (klass), PROP_PORT, + g_param_spec_int ("port", "port", + "The port on which the remote server is listenind", 0, 32768, + DEFAULT_PORT, G_PARAM_READWRITE)); + g_object_class_install_property (G_OBJECT_CLASS (klass), PROP_WINDOW_SIZE, + g_param_spec_int ("window-size", "Window size", + "The size of the window used to calculate rate and offset", 2, 1024, + DEFAULT_WINDOW_SIZE, G_PARAM_READWRITE)); + g_object_class_install_property (G_OBJECT_CLASS (klass), PROP_TIMEOUT, + g_param_spec_uint64 ("timeout", "Timeout", + "The amount of time, in nanoseconds, to wait for replies", 0, + G_MAXUINT64, DEFAULT_TIMEOUT, G_PARAM_READWRITE)); +} + +static void +gst_net_client_clock_init (GstNetClientClock * self, + GstNetClientClockClass * g_class) +{ + self->port = DEFAULT_PORT; + self->address = g_strdup (DEFAULT_ADDRESS); + self->window_size = DEFAULT_WINDOW_SIZE; + self->timeout = DEFAULT_TIMEOUT; + + self->sock = -1; + self->thread = NULL; + + self->filling = TRUE; + self->time_index = 0; + self->local_times = g_new0 (GstClockTime, self->window_size); + self->remote_times = g_new0 (GstClockTime, self->window_size); + + self->servaddr = NULL; + + READ_SOCKET (self) = -1; + WRITE_SOCKET (self) = -1; +} + +static void +gst_net_client_clock_finalize (GObject * object) +{ + GstNetClientClock *self = GST_NET_CLIENT_CLOCK (object); + + if (self->thread) { + gst_net_client_clock_stop (self); + g_assert (self->thread == NULL); + } + + if (READ_SOCKET (self) != -1) { + close (READ_SOCKET (self)); + close (WRITE_SOCKET (self)); + READ_SOCKET (self) = -1; + WRITE_SOCKET (self) = -1; + } + + g_free (self->address); + self->address = NULL; + + g_free (self->servaddr); + self->servaddr = 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: + g_free (self->address); + if (g_value_get_string (value) == NULL) + self->address = g_strdup (DEFAULT_ADDRESS); + else + self->address = g_strdup (g_value_get_string (value)); + break; + case PROP_PORT: + self->port = g_value_get_int (value); + break; + case PROP_WINDOW_SIZE: + self->window_size = g_value_get_int (value); + break; + case PROP_TIMEOUT: + self->timeout = 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: + g_value_set_string (value, self->address); + break; + case PROP_PORT: + g_value_set_int (value, self->port); + break; + case PROP_WINDOW_SIZE: + g_value_set_int (value, self->window_size); + break; + case PROP_TIMEOUT: + g_value_set_uint64 (value, self->timeout); + break; + default: + G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec); + break; + } +} + +/* http://mathworld.wolfram.com/LeastSquaresFitting.html */ +static void +do_linear_regression (GstClockTime * x, GstClockTime * y, gint n, gdouble * m, + gdouble * b, gdouble * r_squared) +{ + gdouble sxx, syy, sxy, xbar, ybar; + gint i; + + /* we lose precision with the doubles, but there's no other way except to use + ~80 bit arithmetic */ + for (i = 0; i < n; i++) { + xbar += x[i]; + ybar += y[i]; + } + xbar /= n; + ybar /= n; + + for (i = 0; i < n; i++) { + sxx += ((double) x[i]) * x[i]; + syy += ((double) y[i]) * y[i]; + sxy += ((double) x[i]) * y[i]; + } + sxx -= ((double) n) * xbar * xbar; + syy -= ((double) n) * ybar * ybar; + sxy -= ((double) n) * xbar * ybar; + + *m = sxy / sxx; + *b = ybar - xbar * *m; + *r_squared = (sxy * sxy) / (sxx * syy); +} + +static void +gst_net_client_clock_observe_times (GstNetClientClock * self, + GstClockTime local_1, GstClockTime remote, GstClockTime local_2) +{ + GstClockTime local_avg; + gdouble m, b, r_squared; + + if (local_2 < local_1) + goto bogus_observation; + + local_avg = (local_2 - local_1) / 2; + + self->local_times[self->time_index] = local_avg; + self->remote_times[self->time_index] = remote; + + self->time_index++; + if (self->time_index == self->window_size) { + self->filling = FALSE; + self->time_index = 0; + } + + if (!self->filling || self->time_index >= 4) { + /* need to allow tuning of the "4" parameter -- means that we need 4 samples + * before beginning to adjust the clock */ + do_linear_regression (self->local_times, self->remote_times, + self->filling ? self->time_index : self->window_size, &m, &b, + &r_squared); + + GST_LOG_OBJECT (self, "adjusting clock to m=%g, b=%g (rsquared=%g)", m, b, + r_squared); + + gst_clock_set_rate_offset (GST_CLOCK (self), m, b); + } + + if (self->filling) { + self->current_timeout = 0; + } else { + /* geto formula */ + self->current_timeout = + (1e-3 / (1 - MIN (r_squared, 0.99999))) * GST_SECOND; + self->current_timeout = MIN (self->current_timeout, self->timeout); + } + + return; + +bogus_observation: + { + GST_WARNING_OBJECT (self, "time packet receive time < send time (%", + GST_TIME_FORMAT, " < %" GST_TIME_FORMAT ")", GST_TIME_ARGS (local_1), + GST_TIME_ARGS (local_2)); + return; + } +} + +static gint +gst_net_client_clock_do_select (GstNetClientClock * self, fd_set * readfds) +{ + guint max_sock; + gint ret; + + while (TRUE) { + FD_ZERO (readfds); + FD_SET (self->sock, readfds); + FD_SET (READ_SOCKET (self), readfds); + max_sock = MAX (self->sock, READ_SOCKET (self)); + + GST_LOG_OBJECT (self, "doing select"); + { + GstClockTime diff; + GTimeVal tv; + + diff = gst_clock_get_internal_time (GST_CLOCK (self)); + GST_TIME_TO_TIMEVAL (self->current_timeout, tv); + + ret = select (max_sock + 1, readfds, NULL, NULL, (struct timeval *) &tv); + + diff = gst_clock_get_internal_time (GST_CLOCK (self)) - diff; + + if (diff > self->current_timeout) + self->current_timeout = 0; + else + self->current_timeout -= diff; + } + GST_LOG_OBJECT (self, "select returned %d", ret); + + if (ret < 0) { + if (errno != EAGAIN && errno != EINTR) + goto select_error; + else + continue; + } else { + return ret; + } + + g_assert_not_reached (); + + /* log errors and keep going */ + select_error: + { + GST_WARNING_OBJECT (self, "select error %d: %s (%d)", ret, + g_strerror (errno), errno); + continue; + } + } + + g_assert_not_reached (); + return -1; +} + +static gpointer +gst_net_client_clock_thread (gpointer data) +{ + GstNetClientClock *self = data; + struct sockaddr_in tmpaddr; + socklen_t len; + fd_set read_fds; + GstNetTimePacket *packet; + gint ret; + + while (TRUE) { + ret = gst_net_client_clock_do_select (self, &read_fds); + + if (FD_ISSET (READ_SOCKET (self), &read_fds)) { + /* got control message */ + while (TRUE) { + gchar command; + int res; + + READ_COMMAND (self, command, res); + if (res < 0) { + GST_LOG_OBJECT (self, "no more commands"); + break; + } + + switch (command) { + case CONTROL_STOP: + /* break out of the select loop */ + GST_LOG_OBJECT (self, "stop"); + goto stopped; + default: + GST_WARNING_OBJECT (self, "unknown message: '%c'", command); + g_warning ("netclientclock: unknown control message received"); + continue; + } + + g_assert_not_reached (); + } + + continue; + } else if (ret == 0) { + /* timed out, let's send another packet */ + packet = gst_net_time_packet_new (NULL); + + packet->local_time = gst_clock_get_internal_time (GST_CLOCK (self)); + + gst_net_time_packet_send (packet, self->sock, + (struct sockaddr *) self->servaddr, sizeof (struct sockaddr_in)); + + g_free (packet); + + /* reset timeout */ + self->current_timeout = self->timeout; + } else if (FD_ISSET (READ_SOCKET (self), &read_fds)) { + /* got data in */ + GstClockTime new_local = gst_clock_get_internal_time (GST_CLOCK (self)); + + len = sizeof (struct sockaddr); + packet = gst_net_time_packet_receive (self->sock, + (struct sockaddr *) &tmpaddr, &len); + + if (!packet) + goto receive_error; + + /* observe_times will reset the timeout */ + gst_net_client_clock_observe_times (self, packet->local_time, + packet->remote_time, new_local); + + g_free (packet); + continue; + } else { + GST_WARNING_OBJECT (self, "unhandled select return state?"); + continue; + } + + g_assert_not_reached (); + + stopped: + { + GST_DEBUG_OBJECT (self, "shutting down"); + /* close socket */ + return NULL; + } + receive_error: + { + GST_WARNING_OBJECT (self, "receive error"); + continue; + } + + g_assert_not_reached (); + + } + + g_assert_not_reached (); + + return NULL; +} + +static gboolean +gst_net_client_clock_start (GstNetClientClock * self) +{ + struct sockaddr_in servaddr, myaddr; + socklen_t len; + gint ret; + GError *error; + + g_return_val_if_fail (self->address != NULL, FALSE); + g_return_val_if_fail (self->servaddr == NULL, FALSE); + + if ((ret = socket (AF_INET, SOCK_DGRAM, 0)) < 0) + goto no_socket; + + self->sock = ret; + + len = sizeof (myaddr); + ret = getsockname (self->sock, (struct sockaddr *) &myaddr, &len); + if (ret < 0) + goto getsockname_error; + + GST_DEBUG_OBJECT (self, "socket opened on UDP port %hd", + ntohs (servaddr.sin_port)); + + memset (&servaddr, 0, sizeof (servaddr)); + servaddr.sin_family = AF_INET; /* host byte order */ + servaddr.sin_port = htons (self->port); /* short, network byte order */ + if (!inet_aton (self->address, &servaddr.sin_addr)) + goto bad_address; + + self->servaddr = g_malloc (sizeof (struct sockaddr_in)); + memcpy (self->servaddr, &servaddr, sizeof (servaddr)); + + GST_DEBUG_OBJECT (self, "will communicate with %s:%d", self->address, + self->port); + + self->thread = g_thread_create (gst_net_client_clock_thread, self, TRUE, + &error); + if (!self->thread) + goto no_thread; + + return TRUE; + + /* ERRORS */ +no_socket: + { + GST_ERROR_OBJECT (self, "socket failed %d: %s (%d)", ret, + g_strerror (errno), errno); + return FALSE; + } +getsockname_error: + { + GST_ERROR_OBJECT (self, "getsockname failed %d: %s (%d)", ret, + g_strerror (errno), errno); + close (self->sock); + self->sock = -1; + return FALSE; + } +bad_address: + { + GST_ERROR_OBJECT (self, "inet_aton failed %d: %s (%d)", ret, + g_strerror (errno), errno); + close (self->sock); + self->sock = -1; + return FALSE; + } +no_thread: + { + GST_ERROR_OBJECT (self, "could not create thread: %s", error->message); + close (self->sock); + self->sock = -1; + g_free (self->servaddr); + self->servaddr = NULL; + g_error_free (error); + return FALSE; + } +} + +static void +gst_net_client_clock_stop (GstNetClientClock * self) +{ + SEND_COMMAND (self, CONTROL_STOP); + g_thread_join (self->thread); + self->thread = NULL; + + if (self->sock != -1) { + close (self->sock); + self->sock = -1; + } +} + +GstClock * +gst_net_client_clock_new (gchar * name, const gchar * remote_address, + gint remote_port, GstClockTime base_time) +{ + GstNetClientClock *ret; + gint iret; + + 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 < 32768, 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, NULL); + + GST_DEBUG_OBJECT (ret, "creating socket pair"); + if ((iret = socketpair (PF_UNIX, SOCK_STREAM, 0, CONTROL_SOCKETS (ret))) < 0) + goto no_socket_pair; + + fcntl (READ_SOCKET (ret), F_SETFL, O_NONBLOCK); + fcntl (WRITE_SOCKET (ret), F_SETFL, O_NONBLOCK); + + if (!gst_net_client_clock_start (ret)) + goto failed_start; + + /* all systems go, cap'n */ + return (GstClock *) ret; + +no_socket_pair: + { + GST_ERROR_OBJECT (ret, "no socket pair %d: %s (%d)", iret, + g_strerror (errno), errno); + gst_object_unref (ret); + return NULL; + } +failed_start: + { + /* already printed a nice error */ + gst_object_unref (ret); + return NULL; + } +} diff --git a/libs/gst/net/gstnetclientclock.h b/libs/gst/net/gstnetclientclock.h new file mode 100644 index 0000000000..0997888a41 --- /dev/null +++ b/libs/gst/net/gstnetclientclock.h @@ -0,0 +1,99 @@ +/* GStreamer + * Copyright (C) 1999,2000 Erik Walthinsen + * 2005 Wim Taymans + * 2005 Andy Wingo + * + * 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., 59 Temple Place - Suite 330, + * Boston, MA 02111-1307, USA. + */ + + +#ifndef __GST_NET_CLIENT_CLOCK_H__ +#define __GST_NET_CLIENT_CLOCK_H__ + +#include +#include + +G_BEGIN_DECLS + +#include +#include +#include +#include +#include +#include +#include + +#include + +#define GST_TYPE_NET_CLIENT_CLOCK \ + (gst_net_client_clock_get_type()) +#define GST_NET_CLIENT_CLOCK(obj) \ + (G_TYPE_CHECK_INSTANCE_CAST((obj),GST_TYPE_NET_CLIENT_CLOCK,GstNetClientClock)) +#define GST_NET_CLIENT_CLOCK_CLASS(klass) \ + (G_TYPE_CHECK_CLASS_CAST((klass),GST_TYPE_NET_CLIENT_CLOCK,GstNetClientClockClass)) +#define GST_IS_NET_CLIENT_CLOCK(obj) \ + (G_TYPE_CHECK_INSTANCE_TYPE((obj),GST_TYPE_NET_CLIENT_CLOCK)) +#define GST_IS_NET_CLIENT_CLOCK_CLASS(obj) \ + (G_TYPE_CHECK_CLASS_TYPE((klass),GST_TYPE_NET_CLIENT_CLOCK)) + +typedef struct _GstNetClientClock GstNetClientClock; +typedef struct _GstNetClientClockClass GstNetClientClockClass; + +struct _GstNetClientClock { + GstSystemClock clock; + + /*< protected >*/ + gchar *address; + gint port; + + gint window_size; + GstClockTime timeout; + + /*< private >*/ + int sock; + int control_sock[2]; + + GstClockTime current_timeout; + + gboolean filling; + gint time_index; + GstClockTime *local_times; + GstClockTime *remote_times; + + struct sockaddr_id *servaddr; + + GThread *thread; + + gpointer _gst_reserved[GST_PADDING]; +}; + +struct _GstNetClientClockClass { + GstSystemClockClass parent_class; + + /*< private >*/ + gpointer _gst_reserved[GST_PADDING]; +}; + +GType gst_net_client_clock_get_type (void); +GstClock* gst_net_client_clock_new (gchar *name, const gchar *remote_address, + gint remote_port, GstClockTime base_time); + +G_END_DECLS + +#endif /* __GST_NET_CLIENT_CLOCK_H__ */ diff --git a/tests/check/Makefile.am b/tests/check/Makefile.am index 9cc9d2de83..3dca798c4c 100644 --- a/tests/check/Makefile.am +++ b/tests/check/Makefile.am @@ -57,6 +57,7 @@ check_PROGRAMS = \ states/sinks \ gst-libs/controller \ gst-libs/gdp \ + net/gstnetclientclock \ net/gstnettimeprovider TESTS = $(check_PROGRAMS) @@ -79,6 +80,9 @@ gst_libs_controller_LDADD = \ $(top_builddir)/libs/gst/controller/libgstcontroller-@GST_MAJORMINOR@.la \ $(LDADD) +net_gstnetclientclock_LDADD = \ + $(top_builddir)/gst/net/libgstnet-tempname-@GST_MAJORMINOR@.la \ + $(LDADD) net_gstnettimeprovider_LDADD = \ $(top_builddir)/gst/net/libgstnet-tempname-@GST_MAJORMINOR@.la \ $(LDADD) diff --git a/tests/check/libs/gstnetclientclock.c b/tests/check/libs/gstnetclientclock.c new file mode 100644 index 0000000000..005d02f388 --- /dev/null +++ b/tests/check/libs/gstnetclientclock.c @@ -0,0 +1,137 @@ +/* GStreamer + * Copyright (C) 2005 Andy Wingo + * + * gstnetclientclock.c: Unit test for the network client clock + * + * 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., 59 Temple Place - Suite 330, + * Boston, MA 02111-1307, USA. + */ + +#include +#include + +#include + +GST_START_TEST (test_instantiation) +{ + GstClock *client, *local; + + local = gst_system_clock_obtain (); + client = gst_net_client_clock_new (NULL, "127.0.0.1", 1234, GST_SECOND); + fail_unless (local != NULL, "failed to get system clock"); + fail_unless (client != NULL, "failed to get network client clock"); + + /* one for gstreamer, one for us */ + ASSERT_OBJECT_REFCOUNT (local, "system clock", 2); + ASSERT_OBJECT_REFCOUNT (client, "network client clock", 1); + + gst_object_unref (client); + + ASSERT_OBJECT_REFCOUNT (local, "net time provider", 2); + + gst_object_unref (local); +} + +GST_END_TEST; + +#if 0 +GST_START_TEST (test_functioning) +{ + GstNetTimeProvider *ntp; + GstNetTimePacket *packet; + GstClock *clock; + GstClockTime local; + struct sockaddr_in servaddr; + gint port = -1, sockfd, ret; + socklen_t len; + + clock = gst_system_clock_obtain (); + fail_unless (clock != NULL, "failed to get system clock"); + ntp = gst_net_time_provider_new (clock, "127.0.0.1", -1); + fail_unless (ntp != NULL, "failed to create net time provider"); + + g_object_get (ntp, "port", &port, NULL); + fail_unless (port > 0); + + sockfd = socket (AF_INET, SOCK_DGRAM, 0); + fail_if (sockfd < 0, "socket failed"); + + memset (&servaddr, 0, sizeof (servaddr)); + servaddr.sin_family = AF_INET; + servaddr.sin_port = htons (port); + inet_aton ("127.0.0.1", &servaddr.sin_addr); + + packet = gst_net_time_packet_new (NULL); + fail_unless (packet != NULL, "failed to create packet"); + + packet->local_time = local = gst_clock_get_time (clock); + + len = sizeof (servaddr); + ret = gst_net_time_packet_send (packet, sockfd, + (struct sockaddr *) &servaddr, len); + + fail_unless (ret == GST_NET_TIME_PACKET_SIZE, "failed to send packet"); + + g_free (packet); + + packet = gst_net_time_packet_receive (sockfd, (struct sockaddr *) &servaddr, + &len); + + fail_unless (packet != NULL, "failed to receive packet"); + fail_unless (packet->local_time == local, "local time is not the same"); + fail_unless (packet->remote_time > local, "remote time not after local time"); + fail_unless (packet->remote_time < gst_clock_get_time (clock), + "remote time in the future"); + + g_free (packet); + + close (sockfd); + + gst_object_unref (ntp); + gst_object_unref (clock); +} + +GST_END_TEST; +#endif + +Suite * +gst_net_client_clock_suite (void) +{ + Suite *s = suite_create ("GstNetClientClock"); + TCase *tc_chain = tcase_create ("generic tests"); + + suite_add_tcase (s, tc_chain); + tcase_add_test (tc_chain, test_instantiation); + /* tcase_add_test (tc_chain, test_functioning); */ + + return s; +} + +int +main (int argc, char **argv) +{ + int nf; + + Suite *s = gst_net_client_clock_suite (); + SRunner *sr = srunner_create (s); + + gst_check_init (&argc, &argv); + + srunner_run_all (sr, CK_NORMAL); + nf = srunner_ntests_failed (sr); + srunner_free (sr); + + return nf; +} diff --git a/tests/check/libs/gstnettimeprovider.c b/tests/check/libs/gstnettimeprovider.c index da1e93a078..e005008d22 100644 --- a/tests/check/libs/gstnettimeprovider.c +++ b/tests/check/libs/gstnettimeprovider.c @@ -74,7 +74,7 @@ GST_START_TEST (test_functioning) memset (&servaddr, 0, sizeof (servaddr)); servaddr.sin_family = AF_INET; - servaddr.sin_port = port; + servaddr.sin_port = htons (port); inet_aton ("127.0.0.1", &servaddr.sin_addr); packet = gst_net_time_packet_new (NULL); diff --git a/tests/misc/network-clock-utils.scm b/tests/misc/network-clock-utils.scm index e5fdae2010..174b5f7ddf 100644 --- a/tests/misc/network-clock-utils.scm +++ b/tests/misc/network-clock-utils.scm @@ -164,7 +164,7 @@ (if (null? (q-tail q)) (make-q tail) (let ((l (append! (q-head q) tail))) - (if (> (length (q-head q)) *queue-length*) + (if (> (length (q-head q)) *window-size*) (make-q (cdr (q-head q))) q))))) diff --git a/tests/misc/network-clock.scm b/tests/misc/network-clock.scm index 513977343c..10878c6abe 100755 --- a/tests/misc/network-clock.scm +++ b/tests/misc/network-clock.scm @@ -125,7 +125,7 @@ exec guile --debug -l $0 -e main -- "$@" (lambda (m b r-squared) (define (next-time) (max - (if (< (length (q-head x)) *queue-length*) + (if (< (length (q-head x)) *window-size*) 0 (/ 1 (- 1 (min r-squared 0.99999)) 1000)) 0.10)) @@ -188,7 +188,7 @@ exec guile --debug -l $0 -e main -- "$@" (define-parameter *packet-loss* 0.01) (define-parameter *send-jitter* 0.1) (define-parameter *recv-jitter* 0.1) -(define-parameter *queue-length* 32) +(define-parameter *window-size* 32) (define-parameter *local-rate* 1.0) (define-parameter *remote-rate* 1.1) (define-parameter *total-time* 5.0) diff --git a/tests/network-clock-utils.scm b/tests/network-clock-utils.scm index e5fdae2010..174b5f7ddf 100644 --- a/tests/network-clock-utils.scm +++ b/tests/network-clock-utils.scm @@ -164,7 +164,7 @@ (if (null? (q-tail q)) (make-q tail) (let ((l (append! (q-head q) tail))) - (if (> (length (q-head q)) *queue-length*) + (if (> (length (q-head q)) *window-size*) (make-q (cdr (q-head q))) q))))) diff --git a/tests/network-clock.scm b/tests/network-clock.scm index 513977343c..10878c6abe 100755 --- a/tests/network-clock.scm +++ b/tests/network-clock.scm @@ -125,7 +125,7 @@ exec guile --debug -l $0 -e main -- "$@" (lambda (m b r-squared) (define (next-time) (max - (if (< (length (q-head x)) *queue-length*) + (if (< (length (q-head x)) *window-size*) 0 (/ 1 (- 1 (min r-squared 0.99999)) 1000)) 0.10)) @@ -188,7 +188,7 @@ exec guile --debug -l $0 -e main -- "$@" (define-parameter *packet-loss* 0.01) (define-parameter *send-jitter* 0.1) (define-parameter *recv-jitter* 0.1) -(define-parameter *queue-length* 32) +(define-parameter *window-size* 32) (define-parameter *local-rate* 1.0) (define-parameter *remote-rate* 1.1) (define-parameter *total-time* 5.0)