libs/gst/net/gstptpclock.c

   1 /* GStreamer
   2  * Copyright (C) 2015 Sebastian Dröge <sebastian@centricular.com>
   3  *
   4  *
   5  * This library is free software; you can redistribute it and/or
   6  * modify it under the terms of the GNU Library General Public
   7  * License as published by the Free Software Foundation; either
   8  * version 2 of the License, or (at your option) any later version.
   9  *
  10  * This library is distributed in the hope that it will be useful,
  11  * but WITHOUT ANY WARRANTY; without even the implied warranty of
  12  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
  13  * Library General Public License for more details.
  14  *
  15  * You should have received a copy of the GNU Library General Public
  16  * License along with this library; if not, write to the
  17  * Free Software Foundation, Inc., 51 Franklin St, Fifth Floor,
  18  * Boston, MA 02110-1301, USA.
  19  */
  20 /**
  21  * SECTION:gstptpclock
  22  * @short_description: Special clock that synchronizes to a remote time
  23  *                     provider via PTP (IEEE1588:2008).
  24  * @see_also: #GstClock, #GstNetClientClock, #GstPipeline
  25  *
  26  * GstPtpClock implements a PTP (IEEE1588:2008) ordinary clock in slave-only
  27  * mode, that allows a GStreamer pipeline to synchronize to a PTP network
  28  * clock in some specific domain.
  29  *
  30  * The PTP subsystem can be initialized with gst_ptp_init(), which then starts
  31  * a helper process to do the actual communication via the PTP ports. This is
  32  * required as PTP listens on ports < 1024 and thus requires special
  33  * privileges. Once this helper process is started, the main process will
  34  * synchronize to all PTP domains that are detected on the selected
  35  * interfaces.
  36  *
  37  * gst_ptp_clock_new() then allows to create a GstClock that provides the PTP
  38  * time from a master clock inside a specific PTP domain. This clock will only
  39  * return valid timestamps once the timestamps in the PTP domain are known. To
  40  * check this, the GstPtpClock::internal-clock property and the related
  41  * notify::clock signal can be used. Once the internal clock is not NULL, the
  42  * PTP domain's time is known. Alternatively you can wait for this with
  43  * gst_ptp_clock_wait_ready().
  44  *
  45  *
  46  * To gather statistics about the PTP clock synchronization,
  47  * gst_ptp_statistics_callback_add() can be used. This gives the application
  48  * the possibility to collect all kinds of statistics from the clock
  49  * synchronization.
  50  *
  51  * Since: 1.6
  52  *
  53  */
  54 #ifdef HAVE_CONFIG_H
  55 #include "config.h"
  56 #endif
  57
  58 #include "gstptpclock.h"
  59
  60 #ifdef HAVE_PTP
  61
  62 #include "gstptp_private.h"
  63
  64 #include <sys/wait.h>
  65 #include <sys/types.h>
  66 #include <unistd.h>
  67
  68 #include <gst/base/base.h>
  69
  70 GST_DEBUG_CATEGORY_STATIC (ptp_debug);
  71 #define GST_CAT_DEFAULT (ptp_debug)
  72
  73 /* IEEE 1588 7.7.3.1 */
  74 #define PTP_ANNOUNCE_RECEIPT_TIMEOUT 4
  75
  76 /* Use a running average for calculating the mean path delay instead
  77  * of just using the last measurement. Enabling this helps in unreliable
  78  * networks, like wifi, with often changing delays
  79  *
  80  * Undef for following IEEE1588-2008 by the letter
  81  */
  82 #define USE_RUNNING_AVERAGE_DELAY 1
  83
  84 /* Filter out any measurements that are above a certain threshold compared to
  85  * previous measurements. Enabling this helps filtering out outliers that
  86  * happen fairly often in unreliable networks, like wifi.
  87  *
  88  * Undef for following IEEE1588-2008 by the letter
  89  */
  90 #define USE_MEASUREMENT_FILTERING 1
  91
  92 /* Select the first clock from which we capture a SYNC message as the master
  93  * clock of the domain until we are ready to run the best master clock
  94  * algorithm. This allows faster syncing but might mean a change of the master
  95  * clock in the beginning. As all clocks in a domain are supposed to use the
  96  * same time, this shouldn't be much of a problem.
  97  *
  98  * Undef for following IEEE1588-2008 by the letter
  99  */
 100 #define USE_OPPORTUNISTIC_CLOCK_SELECTION 1
 101
 102 /* How many updates should be skipped at maximum when using USE_MEASUREMENT_FILTERING */
 103 #define MAX_SKIPPED_UPDATES 5
 104
 105 typedef enum
 106 {
 107   PTP_MESSAGE_TYPE_SYNC = 0x0,
 108   PTP_MESSAGE_TYPE_DELAY_REQ = 0x1,
 109   PTP_MESSAGE_TYPE_PDELAY_REQ = 0x2,
 110   PTP_MESSAGE_TYPE_PDELAY_RESP = 0x3,
 111   PTP_MESSAGE_TYPE_FOLLOW_UP = 0x8,
 112   PTP_MESSAGE_TYPE_DELAY_RESP = 0x9,
 113   PTP_MESSAGE_TYPE_PDELAY_RESP_FOLLOW_UP = 0xA,
 114   PTP_MESSAGE_TYPE_ANNOUNCE = 0xB,
 115   PTP_MESSAGE_TYPE_SIGNALING = 0xC,
 116   PTP_MESSAGE_TYPE_MANAGEMENT = 0xD
 117 } PtpMessageType;
 118
 119 typedef struct
 120 {
 121   guint64 seconds_field;        /* 48 bits valid */
 122   guint32 nanoseconds_field;
 123 } PtpTimestamp;
 124
 125 #define PTP_TIMESTAMP_TO_GST_CLOCK_TIME(ptp) (ptp.seconds_field * GST_SECOND + ptp.nanoseconds_field)
 126 #define GST_CLOCK_TIME_TO_PTP_TIMESTAMP_SECONDS(gst) (((GstClockTime) gst) / GST_SECOND)
 127 #define GST_CLOCK_TIME_TO_PTP_TIMESTAMP_NANOSECONDS(gst) (((GstClockTime) gst) % GST_SECOND)
 128
 129 typedef struct
 130 {
 131   guint64 clock_identity;
 132   guint16 port_number;
 133 } PtpClockIdentity;
 134
 135 static gint
 136 compare_clock_identity (const PtpClockIdentity * a, const PtpClockIdentity * b)
 137 {
 138   if (a->clock_identity < b->clock_identity)
 139     return -1;
 140   else if (a->clock_identity > b->clock_identity)
 141     return 1;
 142
 143   if (a->port_number < b->port_number)
 144     return -1;
 145   else if (a->port_number > b->port_number)
 146     return 1;
 147
 148   return 0;
 149 }
 150
 151 typedef struct
 152 {
 153   guint8 clock_class;
 154   guint8 clock_accuracy;
 155   guint16 offset_scaled_log_variance;
 156 } PtpClockQuality;
 157
 158 typedef struct
 159 {
 160   guint8 transport_specific;
 161   PtpMessageType message_type;
 162   /* guint8 reserved; */
 163   guint8 version_ptp;
 164   guint16 message_length;
 165   guint8 domain_number;
 166   /* guint8 reserved; */
 167   guint16 flag_field;
 168   gint64 correction_field;      /* 48.16 fixed point nanoseconds */
 169   /* guint32 reserved; */
 170   PtpClockIdentity source_port_identity;
 171   guint16 sequence_id;
 172   guint8 control_field;
 173   gint8 log_message_interval;
 174
 175   union
 176   {
 177     struct
 178     {
 179       PtpTimestamp origin_timestamp;
 180       gint16 current_utc_offset;
 181       /* guint8 reserved; */
 182       guint8 grandmaster_priority_1;
 183       PtpClockQuality grandmaster_clock_quality;
 184       guint8 grandmaster_priority_2;
 185       guint64 grandmaster_identity;
 186       guint16 steps_removed;
 187       guint8 time_source;
 188     } announce;
 189
 190     struct
 191     {
 192       PtpTimestamp origin_timestamp;
 193     } sync;
 194
 195     struct
 196     {
 197       PtpTimestamp precise_origin_timestamp;
 198     } follow_up;
 199
 200     struct
 201     {
 202       PtpTimestamp origin_timestamp;
 203     } delay_req;
 204
 205     struct
 206     {
 207       PtpTimestamp receive_timestamp;
 208       PtpClockIdentity requesting_port_identity;
 209     } delay_resp;
 210
 211   } message_specific;
 212 } PtpMessage;
 213
 214 static GMutex ptp_lock;
 215 static GCond ptp_cond;
 216 static gboolean initted = FALSE;
 217 static gboolean supported = TRUE;
 218 static GPid ptp_helper_pid;
 219 static GThread *ptp_helper_thread;
 220 static GMainContext *main_context;
 221 static GMainLoop *main_loop;
 222 static GIOChannel *stdin_channel, *stdout_channel;
 223 static GRand *delay_req_rand;
 224 static PtpClockIdentity ptp_clock_id = { GST_PTP_CLOCK_ID_NONE, 0 };
 225
 226 typedef struct
 227 {
 228   GstClockTime receive_time;
 229
 230   PtpClockIdentity master_clock_identity;
 231
 232   guint8 grandmaster_priority_1;
 233   PtpClockQuality grandmaster_clock_quality;
 234   guint8 grandmaster_priority_2;
 235   guint64 grandmaster_identity;
 236   guint16 steps_removed;
 237   guint8 time_source;
 238
 239   guint16 sequence_id;
 240 } PtpAnnounceMessage;
 241
 242 typedef struct
 243 {
 244   PtpClockIdentity master_clock_identity;
 245
 246   GstClockTime announce_interval;       /* last interval we received */
 247   GQueue announce_messages;
 248 } PtpAnnounceSender;
 249
 250 typedef struct
 251 {
 252   guint domain;
 253   PtpClockIdentity master_clock_identity;
 254
 255   guint16 sync_seqnum;
 256   GstClockTime sync_recv_time_local;    /* t2 */
 257   GstClockTime sync_send_time_remote;   /* t1, might be -1 if FOLLOW_UP pending */
 258   GstClockTime follow_up_recv_time_local;
 259
 260   GSource *timeout_source;
 261   guint16 delay_req_seqnum;
 262   GstClockTime delay_req_send_time_local;       /* t3, -1 if we wait for FOLLOW_UP */
 263   GstClockTime delay_req_recv_time_remote;      /* t4, -1 if we wait */
 264   GstClockTime delay_resp_recv_time_local;
 265
 266   gint64 correction_field_sync; /* sum of the correction fields of SYNC/FOLLOW_UP */
 267   gint64 correction_field_delay;        /* sum of the correction fields of DELAY_RESP */
 268 } PtpPendingSync;
 269
 270 static void
 271 ptp_pending_sync_free (PtpPendingSync * sync)
 272 {
 273   if (sync->timeout_source)
 274     g_source_destroy (sync->timeout_source);
 275   g_free (sync);
 276 }
 277
 278 typedef struct
 279 {
 280   guint domain;
 281
 282   GstClockTime last_ptp_time;
 283   GstClockTime last_local_time;
 284   gint skipped_updates;
 285
 286   /* Used for selecting the master/grandmaster */
 287   GList *announce_senders;
 288
 289   /* Last selected master clock */
 290   gboolean have_master_clock;
 291   PtpClockIdentity master_clock_identity;
 292   guint64 grandmaster_identity;
 293
 294   /* Last SYNC or FOLLOW_UP timestamp we received */
 295   GstClockTime last_ptp_sync_time;
 296   GstClockTime sync_interval;
 297
 298   GstClockTime mean_path_delay;
 299   GstClockTime last_delay_req, min_delay_req_interval;
 300   guint16 last_delay_req_seqnum;
 301
 302   GQueue pending_syncs;
 303
 304   GstClock *domain_clock;
 305 } PtpDomainData;
 306
 307 static GList *domain_data;
 308 static GMutex domain_clocks_lock;
 309 static GList *domain_clocks;
 310
 311 /* Protected by PTP lock */
 312 static void emit_ptp_statistics (guint8 domain, const GstStructure * stats);
 313 static GHookList domain_stats_hooks;
 314 static gint domain_stats_n_hooks;
 315 static gboolean domain_stats_hooks_initted = FALSE;
 316
 317 /* Converts log2 seconds to GstClockTime */
 318 static GstClockTime
 319 log2_to_clock_time (gint l)
 320 {
 321   if (l < 0)
 322     return GST_SECOND >> (-l);
 323   else
 324     return GST_SECOND << l;
 325 }
 326
 327 static void
 328 dump_ptp_message (PtpMessage * msg)
 329 {
 330   GST_TRACE ("PTP message:");
 331   GST_TRACE ("\ttransport_specific: %u", msg->transport_specific);
 332   GST_TRACE ("\tmessage_type: 0x%01x", msg->message_type);
 333   GST_TRACE ("\tversion_ptp: %u", msg->version_ptp);
 334   GST_TRACE ("\tmessage_length: %u", msg->message_length);
 335   GST_TRACE ("\tdomain_number: %u", msg->domain_number);
 336   GST_TRACE ("\tflag_field: 0x%04x", msg->flag_field);
 337   GST_TRACE ("\tcorrection_field: %" G_GINT64_FORMAT ".%03u",
 338       (msg->correction_field / 65536),
 339       (guint) ((msg->correction_field & 0xffff) * 1000) / 65536);
 340   GST_TRACE ("\tsource_port_identity: 0x%016" G_GINT64_MODIFIER "x %u",
 341       msg->source_port_identity.clock_identity,
 342       msg->source_port_identity.port_number);
 343   GST_TRACE ("\tsequence_id: %u", msg->sequence_id);
 344   GST_TRACE ("\tcontrol_field: 0x%02x", msg->control_field);
 345   GST_TRACE ("\tmessage_interval: %" GST_TIME_FORMAT,
 346       GST_TIME_ARGS (log2_to_clock_time (msg->log_message_interval)));
 347
 348   switch (msg->message_type) {
 349     case PTP_MESSAGE_TYPE_ANNOUNCE:
 350       GST_TRACE ("\tANNOUNCE:");
 351       GST_TRACE ("\t\torigin_timestamp: %" G_GUINT64_FORMAT ".%09u",
 352           msg->message_specific.announce.origin_timestamp.seconds_field,
 353           msg->message_specific.announce.origin_timestamp.nanoseconds_field);
 354       GST_TRACE ("\t\tcurrent_utc_offset: %d",
 355           msg->message_specific.announce.current_utc_offset);
 356       GST_TRACE ("\t\tgrandmaster_priority_1: %u",
 357           msg->message_specific.announce.grandmaster_priority_1);
 358       GST_TRACE ("\t\tgrandmaster_clock_quality: 0x%02x 0x%02x %u",
 359           msg->message_specific.announce.grandmaster_clock_quality.clock_class,
 360           msg->message_specific.announce.
 361           grandmaster_clock_quality.clock_accuracy,
 362           msg->message_specific.announce.
 363           grandmaster_clock_quality.offset_scaled_log_variance);
 364       GST_TRACE ("\t\tgrandmaster_priority_2: %u",
 365           msg->message_specific.announce.grandmaster_priority_2);
 366       GST_TRACE ("\t\tgrandmaster_identity: 0x%016" G_GINT64_MODIFIER "x",
 367           msg->message_specific.announce.grandmaster_identity);
 368       GST_TRACE ("\t\tsteps_removed: %u",
 369           msg->message_specific.announce.steps_removed);
 370       GST_TRACE ("\t\ttime_source: 0x%02x",
 371           msg->message_specific.announce.time_source);
 372       break;
 373     case PTP_MESSAGE_TYPE_SYNC:
 374       GST_TRACE ("\tSYNC:");
 375       GST_TRACE ("\t\torigin_timestamp: %" G_GUINT64_FORMAT ".%09u",
 376           msg->message_specific.sync.origin_timestamp.seconds_field,
 377           msg->message_specific.sync.origin_timestamp.nanoseconds_field);
 378       break;
 379     case PTP_MESSAGE_TYPE_FOLLOW_UP:
 380       GST_TRACE ("\tFOLLOW_UP:");
 381       GST_TRACE ("\t\tprecise_origin_timestamp: %" G_GUINT64_FORMAT ".%09u",
 382           msg->message_specific.follow_up.
 383           precise_origin_timestamp.seconds_field,
 384           msg->message_specific.follow_up.
 385           precise_origin_timestamp.nanoseconds_field);
 386       break;
 387     case PTP_MESSAGE_TYPE_DELAY_REQ:
 388       GST_TRACE ("\tDELAY_REQ:");
 389       GST_TRACE ("\t\torigin_timestamp: %" G_GUINT64_FORMAT ".%09u",
 390           msg->message_specific.delay_req.origin_timestamp.seconds_field,
 391           msg->message_specific.delay_req.origin_timestamp.nanoseconds_field);
 392       break;
 393     case PTP_MESSAGE_TYPE_DELAY_RESP:
 394       GST_TRACE ("\tDELAY_RESP:");
 395       GST_TRACE ("\t\treceive_timestamp: %" G_GUINT64_FORMAT ".%09u",
 396           msg->message_specific.delay_resp.receive_timestamp.seconds_field,
 397           msg->message_specific.delay_resp.receive_timestamp.nanoseconds_field);
 398       GST_TRACE ("\t\trequesting_port_identity: 0x%016" G_GINT64_MODIFIER
 399           "x %u",
 400           msg->message_specific.delay_resp.
 401           requesting_port_identity.clock_identity,
 402           msg->message_specific.delay_resp.
 403           requesting_port_identity.port_number);
 404       break;
 405     default:
 406       break;
 407   }
 408   GST_TRACE (" ");
 409 }
 410
 411 /* IEEE 1588-2008 5.3.3 */
 412 static gboolean
 413 parse_ptp_timestamp (PtpTimestamp * timestamp, GstByteReader * reader)
 414 {
 415   g_return_val_if_fail (gst_byte_reader_get_remaining (reader) >= 10, FALSE);
 416
 417   timestamp->seconds_field =
 418       (((guint64) gst_byte_reader_get_uint32_be_unchecked (reader)) << 16) |
 419       gst_byte_reader_get_uint16_be_unchecked (reader);
 420   timestamp->nanoseconds_field =
 421       gst_byte_reader_get_uint32_be_unchecked (reader);
 422
 423   if (timestamp->nanoseconds_field >= 1000000000)
 424     return FALSE;
 425
 426   return TRUE;
 427 }
 428
 429 /* IEEE 1588-2008 13.3 */
 430 static gboolean
 431 parse_ptp_message_header (PtpMessage * msg, GstByteReader * reader)
 432 {
 433   guint8 b;
 434
 435   g_return_val_if_fail (gst_byte_reader_get_remaining (reader) >= 34, FALSE);
 436
 437   b = gst_byte_reader_get_uint8_unchecked (reader);
 438   msg->transport_specific = b >> 4;
 439   msg->message_type = b & 0x0f;
 440
 441   b = gst_byte_reader_get_uint8_unchecked (reader);
 442   msg->version_ptp = b & 0x0f;
 443   if (msg->version_ptp != 2) {
 444     GST_WARNING ("Unsupported PTP message version (%u != 2)", msg->version_ptp);
 445     return FALSE;
 446   }
 447
 448   msg->message_length = gst_byte_reader_get_uint16_be_unchecked (reader);
 449   if (gst_byte_reader_get_remaining (reader) + 4 < msg->message_length) {
 450     GST_WARNING ("Not enough data (%u < %u)",
 451         gst_byte_reader_get_remaining (reader) + 4, msg->message_length);
 452     return FALSE;
 453   }
 454
 455   msg->domain_number = gst_byte_reader_get_uint8_unchecked (reader);
 456   gst_byte_reader_skip_unchecked (reader, 1);
 457
 458   msg->flag_field = gst_byte_reader_get_uint16_be_unchecked (reader);
 459   msg->correction_field = gst_byte_reader_get_uint64_be_unchecked (reader);
 460   gst_byte_reader_skip_unchecked (reader, 4);
 461
 462   msg->source_port_identity.clock_identity =
 463       gst_byte_reader_get_uint64_be_unchecked (reader);
 464   msg->source_port_identity.port_number =
 465       gst_byte_reader_get_uint16_be_unchecked (reader);
 466
 467   msg->sequence_id = gst_byte_reader_get_uint16_be_unchecked (reader);
 468   msg->control_field = gst_byte_reader_get_uint8_unchecked (reader);
 469   msg->log_message_interval = gst_byte_reader_get_uint8_unchecked (reader);
 470
 471   return TRUE;
 472 }
 473
 474 /* IEEE 1588-2008 13.5 */
 475 static gboolean
 476 parse_ptp_message_announce (PtpMessage * msg, GstByteReader * reader)
 477 {
 478   g_return_val_if_fail (msg->message_type == PTP_MESSAGE_TYPE_ANNOUNCE, FALSE);
 479
 480   if (gst_byte_reader_get_remaining (reader) < 20)
 481     return FALSE;
 482
 483   if (!parse_ptp_timestamp (&msg->message_specific.announce.origin_timestamp,
 484           reader))
 485     return FALSE;
 486
 487   msg->message_specific.announce.current_utc_offset =
 488       gst_byte_reader_get_uint16_be_unchecked (reader);
 489   gst_byte_reader_skip_unchecked (reader, 1);
 490
 491   msg->message_specific.announce.grandmaster_priority_1 =
 492       gst_byte_reader_get_uint8_unchecked (reader);
 493   msg->message_specific.announce.grandmaster_clock_quality.clock_class =
 494       gst_byte_reader_get_uint8_unchecked (reader);
 495   msg->message_specific.announce.grandmaster_clock_quality.clock_accuracy =
 496       gst_byte_reader_get_uint8_unchecked (reader);
 497   msg->message_specific.announce.
 498       grandmaster_clock_quality.offset_scaled_log_variance =
 499       gst_byte_reader_get_uint16_be_unchecked (reader);
 500   msg->message_specific.announce.grandmaster_priority_2 =
 501       gst_byte_reader_get_uint8_unchecked (reader);
 502   msg->message_specific.announce.grandmaster_identity =
 503       gst_byte_reader_get_uint64_be_unchecked (reader);
 504   msg->message_specific.announce.steps_removed =
 505       gst_byte_reader_get_uint16_be_unchecked (reader);
 506   msg->message_specific.announce.time_source =
 507       gst_byte_reader_get_uint8_unchecked (reader);
 508
 509   return TRUE;
 510 }
 511
 512 /* IEEE 1588-2008 13.6 */
 513 static gboolean
 514 parse_ptp_message_sync (PtpMessage * msg, GstByteReader * reader)
 515 {
 516   g_return_val_if_fail (msg->message_type == PTP_MESSAGE_TYPE_SYNC, FALSE);
 517
 518   if (gst_byte_reader_get_remaining (reader) < 10)
 519     return FALSE;
 520
 521   if (!parse_ptp_timestamp (&msg->message_specific.sync.origin_timestamp,
 522           reader))
 523     return FALSE;
 524
 525   return TRUE;
 526 }
 527
 528 /* IEEE 1588-2008 13.6 */
 529 static gboolean
 530 parse_ptp_message_delay_req (PtpMessage * msg, GstByteReader * reader)
 531 {
 532   g_return_val_if_fail (msg->message_type == PTP_MESSAGE_TYPE_DELAY_REQ, FALSE);
 533
 534   if (gst_byte_reader_get_remaining (reader) < 10)
 535     return FALSE;
 536
 537   if (!parse_ptp_timestamp (&msg->message_specific.delay_req.origin_timestamp,
 538           reader))
 539     return FALSE;
 540
 541   return TRUE;
 542 }
 543
 544 /* IEEE 1588-2008 13.7 */
 545 static gboolean
 546 parse_ptp_message_follow_up (PtpMessage * msg, GstByteReader * reader)
 547 {
 548   g_return_val_if_fail (msg->message_type == PTP_MESSAGE_TYPE_FOLLOW_UP, FALSE);
 549
 550   if (gst_byte_reader_get_remaining (reader) < 10)
 551     return FALSE;
 552
 553   if (!parse_ptp_timestamp (&msg->message_specific.
 554           follow_up.precise_origin_timestamp, reader))
 555     return FALSE;
 556
 557   return TRUE;
 558 }
 559
 560 /* IEEE 1588-2008 13.8 */
 561 static gboolean
 562 parse_ptp_message_delay_resp (PtpMessage * msg, GstByteReader * reader)
 563 {
 564   g_return_val_if_fail (msg->message_type == PTP_MESSAGE_TYPE_DELAY_RESP,
 565       FALSE);
 566
 567   if (gst_byte_reader_get_remaining (reader) < 20)
 568     return FALSE;
 569
 570   if (!parse_ptp_timestamp (&msg->message_specific.delay_resp.receive_timestamp,
 571           reader))
 572     return FALSE;
 573
 574   msg->message_specific.delay_resp.requesting_port_identity.clock_identity =
 575       gst_byte_reader_get_uint64_be_unchecked (reader);
 576   msg->message_specific.delay_resp.requesting_port_identity.port_number =
 577       gst_byte_reader_get_uint16_be_unchecked (reader);
 578
 579   return TRUE;
 580 }
 581
 582 static gboolean
 583 parse_ptp_message (PtpMessage * msg, const guint8 * data, gsize size)
 584 {
 585   GstByteReader reader;
 586   gboolean ret = FALSE;
 587
 588   gst_byte_reader_init (&reader, data, size);
 589
 590   if (!parse_ptp_message_header (msg, &reader)) {
 591     GST_WARNING ("Failed to parse PTP message header");
 592     return FALSE;
 593   }
 594
 595   switch (msg->message_type) {
 596     case PTP_MESSAGE_TYPE_SYNC:
 597       ret = parse_ptp_message_sync (msg, &reader);
 598       break;
 599     case PTP_MESSAGE_TYPE_FOLLOW_UP:
 600       ret = parse_ptp_message_follow_up (msg, &reader);
 601       break;
 602     case PTP_MESSAGE_TYPE_DELAY_REQ:
 603       ret = parse_ptp_message_delay_req (msg, &reader);
 604       break;
 605     case PTP_MESSAGE_TYPE_DELAY_RESP:
 606       ret = parse_ptp_message_delay_resp (msg, &reader);
 607       break;
 608     case PTP_MESSAGE_TYPE_ANNOUNCE:
 609       ret = parse_ptp_message_announce (msg, &reader);
 610       break;
 611     default:
 612       /* ignore for now */
 613       break;
 614   }
 615
 616   return ret;
 617 }
 618
 619 static gint
 620 compare_announce_message (const PtpAnnounceMessage * a,
 621     const PtpAnnounceMessage * b)
 622 {
 623   /* IEEE 1588 Figure 27 */
 624   if (a->grandmaster_identity == b->grandmaster_identity) {
 625     if (a->steps_removed + 1 < b->steps_removed)
 626       return -1;
 627     else if (a->steps_removed > b->steps_removed + 1)
 628       return 1;
 629
 630     /* Error cases are filtered out earlier */
 631     if (a->steps_removed < b->steps_removed)
 632       return -1;
 633     else if (a->steps_removed > b->steps_removed)
 634       return 1;
 635
 636     /* Error cases are filtered out earlier */
 637     if (a->master_clock_identity.clock_identity <
 638         b->master_clock_identity.clock_identity)
 639       return -1;
 640     else if (a->master_clock_identity.clock_identity >
 641         b->master_clock_identity.clock_identity)
 642       return 1;
 643
 644     /* Error cases are filtered out earlier */
 645     if (a->master_clock_identity.port_number <
 646         b->master_clock_identity.port_number)
 647       return -1;
 648     else if (a->master_clock_identity.port_number >
 649         b->master_clock_identity.port_number)
 650       return 1;
 651     else
 652       g_assert_not_reached ();
 653
 654     return 0;
 655   }
 656
 657   if (a->grandmaster_priority_1 < b->grandmaster_priority_1)
 658     return -1;
 659   else if (a->grandmaster_priority_1 > b->grandmaster_priority_1)
 660     return 1;
 661
 662   if (a->grandmaster_clock_quality.clock_class <
 663       b->grandmaster_clock_quality.clock_class)
 664     return -1;
 665   else if (a->grandmaster_clock_quality.clock_class >
 666       b->grandmaster_clock_quality.clock_class)
 667     return 1;
 668
 669   if (a->grandmaster_clock_quality.clock_accuracy <
 670       b->grandmaster_clock_quality.clock_accuracy)
 671     return -1;
 672   else if (a->grandmaster_clock_quality.clock_accuracy >
 673       b->grandmaster_clock_quality.clock_accuracy)
 674     return 1;
 675
 676   if (a->grandmaster_clock_quality.offset_scaled_log_variance <
 677       b->grandmaster_clock_quality.offset_scaled_log_variance)
 678     return -1;
 679   else if (a->grandmaster_clock_quality.offset_scaled_log_variance >
 680       b->grandmaster_clock_quality.offset_scaled_log_variance)
 681     return 1;
 682
 683   if (a->grandmaster_priority_2 < b->grandmaster_priority_2)
 684     return -1;
 685   else if (a->grandmaster_priority_2 > b->grandmaster_priority_2)
 686     return 1;
 687
 688   if (a->grandmaster_identity < b->grandmaster_identity)
 689     return -1;
 690   else if (a->grandmaster_identity > b->grandmaster_identity)
 691     return 1;
 692   else
 693     g_assert_not_reached ();
 694
 695   return 0;
 696 }
 697
 698 static void
 699 select_best_master_clock (PtpDomainData * domain, GstClockTime now)
 700 {
 701   GList *qualified_messages = NULL;
 702   GList *l, *m;
 703   PtpAnnounceMessage *best = NULL;
 704
 705   /* IEEE 1588 9.3.2.5 */
 706   for (l = domain->announce_senders; l; l = l->next) {
 707     PtpAnnounceSender *sender = l->data;
 708     GstClockTime window = 4 * sender->announce_interval;
 709     gint count = 0;
 710
 711     for (m = sender->announce_messages.head; m; m = m->next) {
 712       PtpAnnounceMessage *msg = m->data;
 713
 714       if (now - msg->receive_time <= window)
 715         count++;
 716     }
 717
 718     /* Only include the newest message of announce senders that had at least 2
 719      * announce messages in the last 4 announce intervals. Which also means
 720      * that we wait at least 4 announce intervals before we select a master
 721      * clock. Until then we just report based on the newest SYNC we received
 722      */
 723     if (count >= 2) {
 724       qualified_messages =
 725           g_list_prepend (qualified_messages,
 726           g_queue_peek_tail (&sender->announce_messages));
 727     }
 728   }
 729
 730   if (!qualified_messages) {
 731     GST_DEBUG
 732         ("No qualified announce messages for domain %u, can't select a master clock",
 733         domain->domain);
 734     domain->have_master_clock = FALSE;
 735     return;
 736   }
 737
 738   for (l = qualified_messages; l; l = l->next) {
 739     PtpAnnounceMessage *msg = l->data;
 740
 741     if (!best || compare_announce_message (msg, best) < 0)
 742       best = msg;
 743   }
 744
 745   if (domain->have_master_clock
 746       && compare_clock_identity (&domain->master_clock_identity,
 747           &best->master_clock_identity) == 0) {
 748     GST_DEBUG ("Master clock in domain %u did not change", domain->domain);
 749   } else {
 750     GST_DEBUG ("Selected master clock for domain %u: 0x%016" G_GINT64_MODIFIER
 751         "x %u with grandmaster clock 0x%016" G_GINT64_MODIFIER "x",
 752         domain->domain, best->master_clock_identity.clock_identity,
 753         best->master_clock_identity.port_number, best->grandmaster_identity);
 754
 755     domain->have_master_clock = TRUE;
 756     domain->grandmaster_identity = best->grandmaster_identity;
 757
 758     /* Opportunistic master clock selection likely gave us the same master
 759      * clock before, no need to reset all statistics */
 760     if (compare_clock_identity (&domain->master_clock_identity,
 761             &best->master_clock_identity) != 0) {
 762       memcpy (&domain->master_clock_identity, &best->master_clock_identity,
 763           sizeof (PtpClockIdentity));
 764       domain->mean_path_delay = 0;
 765       domain->last_delay_req = 0;
 766       domain->min_delay_req_interval = 0;
 767       domain->sync_interval = 0;
 768       domain->last_ptp_sync_time = 0;
 769       domain->skipped_updates = 0;
 770       g_queue_foreach (&domain->pending_syncs, (GFunc) ptp_pending_sync_free,
 771           NULL);
 772       g_queue_clear (&domain->pending_syncs);
 773     }
 774
 775     if (g_atomic_int_get (&domain_stats_n_hooks)) {
 776       GstStructure *stats =
 777           gst_structure_new (GST_PTP_STATISTICS_BEST_MASTER_CLOCK_SELECTED,
 778           "domain", G_TYPE_UINT, domain->domain,
 779           "master-clock-id", G_TYPE_UINT64,
 780           domain->master_clock_identity.clock_identity,
 781           "master-clock-port", G_TYPE_UINT,
 782           domain->master_clock_identity.port_number,
 783           "grandmaster-clock-id", G_TYPE_UINT64, domain->grandmaster_identity,
 784           NULL);
 785       emit_ptp_statistics (domain->domain, stats);
 786       gst_structure_free (stats);
 787     }
 788   }
 789 }
 790
 791 static void
 792 handle_announce_message (PtpMessage * msg, GstClockTime receive_time)
 793 {
 794   GList *l;
 795   PtpDomainData *domain = NULL;
 796   PtpAnnounceSender *sender = NULL;
 797   PtpAnnounceMessage *announce;
 798
 799   /* IEEE1588 9.3.2.2 e)
 800    * Don't consider messages with the alternate master flag set
 801    */
 802   if ((msg->flag_field & 0x0100))
 803     return;
 804
 805   /* IEEE 1588 9.3.2.5 d)
 806    * Don't consider announce messages with steps_removed>=255
 807    */
 808   if (msg->message_specific.announce.steps_removed >= 255)
 809     return;
 810
 811   for (l = domain_data; l; l = l->next) {
 812     PtpDomainData *tmp = l->data;
 813
 814     if (tmp->domain == msg->domain_number) {
 815       domain = tmp;
 816       break;
 817     }
 818   }
 819
 820   if (!domain) {
 821     gchar *clock_name;
 822
 823     domain = g_new0 (PtpDomainData, 1);
 824     domain->domain = msg->domain_number;
 825     clock_name = g_strdup_printf ("ptp-clock-%u", domain->domain);
 826     domain->domain_clock =
 827         g_object_new (GST_TYPE_SYSTEM_CLOCK, "name", clock_name, NULL);
 828     g_free (clock_name);
 829     g_queue_init (&domain->pending_syncs);
 830     domain_data = g_list_prepend (domain_data, domain);
 831
 832     g_mutex_lock (&domain_clocks_lock);
 833     domain_clocks = g_list_prepend (domain_clocks, domain);
 834     g_mutex_unlock (&domain_clocks_lock);
 835
 836     if (g_atomic_int_get (&domain_stats_n_hooks)) {
 837       GstStructure *stats =
 838           gst_structure_new (GST_PTP_STATISTICS_NEW_DOMAIN_FOUND, "domain",
 839           G_TYPE_UINT, domain->domain, "clock", GST_TYPE_CLOCK,
 840           domain->domain_clock, NULL);
 841       emit_ptp_statistics (domain->domain, stats);
 842       gst_structure_free (stats);
 843     }
 844   }
 845
 846   for (l = domain->announce_senders; l; l = l->next) {
 847     PtpAnnounceSender *tmp = l->data;
 848
 849     if (compare_clock_identity (&tmp->master_clock_identity,
 850             &msg->source_port_identity) == 0) {
 851       sender = tmp;
 852       break;
 853     }
 854   }
 855
 856   if (!sender) {
 857     sender = g_new0 (PtpAnnounceSender, 1);
 858
 859     memcpy (&sender->master_clock_identity, &msg->source_port_identity,
 860         sizeof (PtpClockIdentity));
 861     g_queue_init (&sender->announce_messages);
 862     domain->announce_senders =
 863         g_list_prepend (domain->announce_senders, sender);
 864   }
 865
 866   for (l = sender->announce_messages.head; l; l = l->next) {
 867     PtpAnnounceMessage *tmp = l->data;
 868
 869     /* IEEE 1588 9.3.2.5 c)
 870      * Don't consider identical messages, i.e. duplicates
 871      */
 872     if (tmp->sequence_id == msg->sequence_id)
 873       return;
 874   }
 875
 876   sender->announce_interval = log2_to_clock_time (msg->log_message_interval);
 877
 878   announce = g_new0 (PtpAnnounceMessage, 1);
 879   announce->receive_time = receive_time;
 880   announce->sequence_id = msg->sequence_id;
 881   memcpy (&announce->master_clock_identity, &msg->source_port_identity,
 882       sizeof (PtpClockIdentity));
 883   announce->grandmaster_identity =
 884       msg->message_specific.announce.grandmaster_identity;
 885   announce->grandmaster_priority_1 =
 886       msg->message_specific.announce.grandmaster_priority_1;
 887   announce->grandmaster_clock_quality.clock_class =
 888       msg->message_specific.announce.grandmaster_clock_quality.clock_class;
 889   announce->grandmaster_clock_quality.clock_accuracy =
 890       msg->message_specific.announce.grandmaster_clock_quality.clock_accuracy;
 891   announce->grandmaster_clock_quality.offset_scaled_log_variance =
 892       msg->message_specific.announce.
 893       grandmaster_clock_quality.offset_scaled_log_variance;
 894   announce->grandmaster_priority_2 =
 895       msg->message_specific.announce.grandmaster_priority_2;
 896   announce->steps_removed = msg->message_specific.announce.steps_removed;
 897   announce->time_source = msg->message_specific.announce.time_source;
 898   g_queue_push_tail (&sender->announce_messages, announce);
 899
 900   select_best_master_clock (domain, receive_time);
 901 }
 902
 903 static gboolean
 904 send_delay_req_timeout (PtpPendingSync * sync)
 905 {
 906   StdIOHeader header = { 0, };
 907   guint8 delay_req[44];
 908   GstByteWriter writer;
 909   GIOStatus status;
 910   gsize written;
 911   GError *err = NULL;
 912
 913   header.type = TYPE_EVENT;
 914   header.size = 44;
 915
 916   gst_byte_writer_init_with_data (&writer, delay_req, 44, FALSE);
 917   gst_byte_writer_put_uint8_unchecked (&writer, PTP_MESSAGE_TYPE_DELAY_REQ);
 918   gst_byte_writer_put_uint8_unchecked (&writer, 2);
 919   gst_byte_writer_put_uint16_be_unchecked (&writer, 44);
 920   gst_byte_writer_put_uint8_unchecked (&writer, sync->domain);
 921   gst_byte_writer_put_uint8_unchecked (&writer, 0);
 922   gst_byte_writer_put_uint16_be_unchecked (&writer, 0);
 923   gst_byte_writer_put_uint64_be_unchecked (&writer, 0);
 924   gst_byte_writer_put_uint32_be_unchecked (&writer, 0);
 925   gst_byte_writer_put_uint64_be_unchecked (&writer,
 926       ptp_clock_id.clock_identity);
 927   gst_byte_writer_put_uint16_be_unchecked (&writer, ptp_clock_id.port_number);
 928   gst_byte_writer_put_uint16_be_unchecked (&writer, sync->delay_req_seqnum);
 929   gst_byte_writer_put_uint8_unchecked (&writer, 0x01);
 930   gst_byte_writer_put_uint8_unchecked (&writer, 0x7f);
 931   gst_byte_writer_put_uint64_be_unchecked (&writer, 0);
 932   gst_byte_writer_put_uint16_be_unchecked (&writer, 0);
 933
 934   status =
 935       g_io_channel_write_chars (stdout_channel, (gchar *) & header,
 936       sizeof (header), &written, &err);
 937   if (status == G_IO_STATUS_ERROR) {
 938     g_warning ("Failed to write to stdout: %s", err->message);
 939     return G_SOURCE_REMOVE;
 940   } else if (status == G_IO_STATUS_EOF) {
 941     g_message ("EOF on stdout");
 942     g_main_loop_quit (main_loop);
 943     return G_SOURCE_REMOVE;
 944   } else if (status != G_IO_STATUS_NORMAL) {
 945     g_warning ("Unexpected stdout write status: %d", status);
 946     g_main_loop_quit (main_loop);
 947     return G_SOURCE_REMOVE;
 948   } else if (written != sizeof (header)) {
 949     g_warning ("Unexpected write size: %" G_GSIZE_FORMAT, written);
 950     g_main_loop_quit (main_loop);
 951     return G_SOURCE_REMOVE;
 952   }
 953
 954   sync->delay_req_send_time_local = gst_util_get_timestamp ();
 955
 956   status =
 957       g_io_channel_write_chars (stdout_channel,
 958       (const gchar *) delay_req, 44, &written, &err);
 959   if (status == G_IO_STATUS_ERROR) {
 960     g_warning ("Failed to write to stdout: %s", err->message);
 961     g_main_loop_quit (main_loop);
 962     return G_SOURCE_REMOVE;
 963   } else if (status == G_IO_STATUS_EOF) {
 964     g_message ("EOF on stdout");
 965     g_main_loop_quit (main_loop);
 966     return G_SOURCE_REMOVE;
 967   } else if (status != G_IO_STATUS_NORMAL) {
 968     g_warning ("Unexpected stdout write status: %d", status);
 969     g_main_loop_quit (main_loop);
 970     return G_SOURCE_REMOVE;
 971   } else if (written != 44) {
 972     g_warning ("Unexpected write size: %" G_GSIZE_FORMAT, written);
 973     g_main_loop_quit (main_loop);
 974     return G_SOURCE_REMOVE;
 975   }
 976
 977   return G_SOURCE_REMOVE;
 978 }
 979
 980 static gboolean
 981 send_delay_req (PtpDomainData * domain, PtpPendingSync * sync)
 982 {
 983   GstClockTime now = gst_util_get_timestamp ();
 984   guint timeout;
 985   GSource *timeout_source;
 986
 987   if (domain->last_delay_req != 0
 988       && domain->last_delay_req + domain->min_delay_req_interval > now)
 989     return FALSE;
 990
 991   domain->last_delay_req = now;
 992   sync->delay_req_seqnum = domain->last_delay_req_seqnum++;
 993
 994   /* IEEE 1588 9.5.11.2 */
 995   if (domain->last_delay_req == 0 || domain->min_delay_req_interval == 0)
 996     timeout = 0;
 997   else
 998     timeout =
 999         g_rand_int_range (delay_req_rand, 0,
1000         (domain->min_delay_req_interval * 2) / GST_MSECOND);
1001
1002   sync->timeout_source = timeout_source = g_timeout_source_new (timeout);
1003   g_source_set_priority (timeout_source, G_PRIORITY_DEFAULT);
1004   g_source_set_callback (timeout_source, (GSourceFunc) send_delay_req_timeout,
1005       sync, NULL);
1006   g_source_attach (timeout_source, main_context);
1007
1008   return TRUE;
1009 }
1010
1011 /* Filtering of outliers for RTT and time calculations inspired
1012  * by the code from gstnetclientclock.c
1013  */
1014 static void
1015 update_ptp_time (PtpDomainData * domain, PtpPendingSync * sync)
1016 {
1017   GstClockTime internal_time, external_time, rate_num, rate_den;
1018   GstClockTime corrected_ptp_time, corrected_local_time;
1019   gdouble r_squared = 0.0;
1020   gboolean synced;
1021   GstClockTimeDiff discont = 0;
1022   GstClockTime estimated_ptp_time = GST_CLOCK_TIME_NONE;
1023
1024 #ifdef USE_MEASUREMENT_FILTERING
1025   GstClockTime orig_internal_time, orig_external_time, orig_rate_num,
1026       orig_rate_den;
1027   GstClockTime new_estimated_ptp_time;
1028   GstClockTime max_discont, estimated_ptp_time_min, estimated_ptp_time_max;
1029   gboolean now_synced;
1030
1031   /* We check this here and when updating the mean path delay, because
1032    * we can get here without a delay response too */
1033   if (sync->follow_up_recv_time_local != GST_CLOCK_TIME_NONE
1034       && sync->follow_up_recv_time_local >
1035       sync->sync_recv_time_local + 2 * domain->mean_path_delay) {
1036     GST_WARNING ("Sync-follow-up delay for domain %u too big: %" GST_TIME_FORMAT
1037         " > 2 * %" GST_TIME_FORMAT, domain->domain,
1038         GST_TIME_ARGS (sync->follow_up_recv_time_local),
1039         GST_TIME_ARGS (domain->mean_path_delay));
1040     goto out;
1041   }
1042 #endif
1043
1044   /* IEEE 1588 11.2 */
1045   corrected_ptp_time =
1046       sync->sync_send_time_remote +
1047       (sync->correction_field_sync + 32768) / 65536;
1048   corrected_local_time = sync->sync_recv_time_local - domain->mean_path_delay;
1049
1050   /* Set an initial local-remote relation */
1051   if (domain->last_ptp_time == 0)
1052     gst_clock_set_calibration (domain->domain_clock, corrected_local_time,
1053         corrected_ptp_time, 1, 1);
1054
1055 #ifdef USE_MEASUREMENT_FILTERING
1056   /* Check if the corrected PTP time is +/- 3/4 RTT around what we would
1057    * estimate with our present knowledge about the clock
1058    */
1059   /* Store what the clock produced as 'now' before this update */
1060   gst_clock_get_calibration (GST_CLOCK_CAST (domain->domain_clock),
1061       &orig_internal_time, &orig_external_time, &orig_rate_num, &orig_rate_den);
1062   internal_time = orig_internal_time;
1063   external_time = orig_external_time;
1064   rate_num = orig_rate_num;
1065   rate_den = orig_rate_den;
1066
1067   /* 3/4 RTT window around the estimation */
1068   max_discont = domain->mean_path_delay * 3 / 2;
1069
1070   /* Check if the estimated sync time is inside our window */
1071   estimated_ptp_time_min = corrected_local_time - max_discont;
1072   estimated_ptp_time_min =
1073       gst_clock_adjust_with_calibration (GST_CLOCK_CAST (domain->domain_clock),
1074       estimated_ptp_time_min, internal_time, external_time, rate_num, rate_den);
1075   estimated_ptp_time_max = corrected_local_time + max_discont;
1076   estimated_ptp_time_max =
1077       gst_clock_adjust_with_calibration (GST_CLOCK_CAST (domain->domain_clock),
1078       estimated_ptp_time_max, internal_time, external_time, rate_num, rate_den);
1079
1080   synced = (estimated_ptp_time_min < corrected_ptp_time
1081       && corrected_ptp_time < estimated_ptp_time_max);
1082
1083   GST_DEBUG ("Adding observation for domain %u: %" GST_TIME_FORMAT " - %"
1084       GST_TIME_FORMAT, domain->domain,
1085       GST_TIME_ARGS (corrected_ptp_time), GST_TIME_ARGS (corrected_local_time));
1086
1087   GST_DEBUG ("Synced %d: %" GST_TIME_FORMAT " < %" GST_TIME_FORMAT " < %"
1088       GST_TIME_FORMAT, synced, GST_TIME_ARGS (estimated_ptp_time_min),
1089       GST_TIME_ARGS (corrected_ptp_time),
1090       GST_TIME_ARGS (estimated_ptp_time_max));
1091
1092   if (gst_clock_add_observation_unapplied (domain->domain_clock,
1093           corrected_local_time, corrected_ptp_time, &r_squared,
1094           &internal_time, &external_time, &rate_num, &rate_den)) {
1095     GST_DEBUG ("Regression gave r_squared: %f", r_squared);
1096
1097     /* Old estimated PTP time based on receive time and path delay */
1098     estimated_ptp_time = corrected_local_time;
1099     estimated_ptp_time =
1100         gst_clock_adjust_with_calibration (GST_CLOCK_CAST
1101         (domain->domain_clock), estimated_ptp_time, orig_internal_time,
1102         orig_external_time, orig_rate_num, orig_rate_den);
1103
1104     /* New estimated PTP time based on receive time and path delay */
1105     new_estimated_ptp_time = corrected_local_time;
1106     new_estimated_ptp_time =
1107         gst_clock_adjust_with_calibration (GST_CLOCK_CAST
1108         (domain->domain_clock), new_estimated_ptp_time, internal_time,
1109         external_time, rate_num, rate_den);
1110
1111     discont = GST_CLOCK_DIFF (estimated_ptp_time, new_estimated_ptp_time);
1112     if (synced && ABS (discont) > max_discont) {
1113       GstClockTimeDiff offset;
1114       GST_DEBUG ("Too large a discont %s%" GST_TIME_FORMAT
1115           ", clamping to 1/4 average RTT = %" GST_TIME_FORMAT,
1116           (discont < 0 ? "-" : ""), GST_TIME_ARGS (ABS (discont)),
1117           GST_TIME_ARGS (max_discont));
1118       if (discont > 0) {        /* Too large a forward step - add a -ve offset */
1119         offset = max_discont - discont;
1120         if (-offset > external_time)
1121           external_time = 0;
1122         else
1123           external_time += offset;
1124       } else {                  /* Too large a backward step - add a +ve offset */
1125         offset = -(max_discont + discont);
1126         external_time += offset;
1127       }
1128
1129       discont += offset;
1130     } else {
1131       GST_DEBUG ("Discont %s%" GST_TIME_FORMAT " (max: %" GST_TIME_FORMAT ")",
1132           (discont < 0 ? "-" : ""), GST_TIME_ARGS (ABS (discont)),
1133           GST_TIME_ARGS (max_discont));
1134     }
1135
1136     /* Check if the estimated sync time is now (still) inside our window */
1137     estimated_ptp_time_min = corrected_local_time - max_discont;
1138     estimated_ptp_time_min =
1139         gst_clock_adjust_with_calibration (GST_CLOCK_CAST
1140         (domain->domain_clock), estimated_ptp_time_min, internal_time,
1141         external_time, rate_num, rate_den);
1142     estimated_ptp_time_max = corrected_local_time + max_discont;
1143     estimated_ptp_time_max =
1144         gst_clock_adjust_with_calibration (GST_CLOCK_CAST
1145         (domain->domain_clock), estimated_ptp_time_max, internal_time,
1146         external_time, rate_num, rate_den);
1147
1148     now_synced = (estimated_ptp_time_min < corrected_ptp_time
1149         && corrected_ptp_time < estimated_ptp_time_max);
1150
1151     GST_DEBUG ("Now synced %d: %" GST_TIME_FORMAT " < %" GST_TIME_FORMAT " < %"
1152         GST_TIME_FORMAT, now_synced, GST_TIME_ARGS (estimated_ptp_time_min),
1153         GST_TIME_ARGS (corrected_ptp_time),
1154         GST_TIME_ARGS (estimated_ptp_time_max));
1155
1156     if (synced || now_synced || domain->skipped_updates > MAX_SKIPPED_UPDATES) {
1157       gst_clock_set_calibration (GST_CLOCK_CAST (domain->domain_clock),
1158           internal_time, external_time, rate_num, rate_den);
1159       domain->skipped_updates = 0;
1160
1161       domain->last_ptp_time = corrected_ptp_time;
1162       domain->last_local_time = corrected_local_time;
1163     } else {
1164       domain->skipped_updates++;
1165     }
1166   } else {
1167     domain->last_ptp_time = corrected_ptp_time;
1168     domain->last_local_time = corrected_local_time;
1169   }
1170
1171 #else
1172   GST_DEBUG ("Adding observation for domain %u: %" GST_TIME_FORMAT " - %"
1173       GST_TIME_FORMAT, domain->domain,
1174       GST_TIME_ARGS (corrected_ptp_time), GST_TIME_ARGS (corrected_local_time));
1175
1176   gst_clock_get_calibration (GST_CLOCK_CAST (domain->domain_clock),
1177       &internal_time, &external_time, &rate_num, &rate_den);
1178
1179   estimated_ptp_time = corrected_local_time;
1180   estimated_ptp_time =
1181       gst_clock_adjust_with_calibration (GST_CLOCK_CAST
1182       (domain->domain_clock), estimated_ptp_time, internal_time,
1183       external_time, rate_num, rate_den);
1184
1185   gst_clock_add_observation (domain->domain_clock,
1186       corrected_local_time, corrected_ptp_time, &r_squared);
1187
1188   gst_clock_get_calibration (GST_CLOCK_CAST (domain->domain_clock),
1189       &internal_time, &external_time, &rate_num, &rate_den);
1190
1191   synced = TRUE;
1192   domain->last_ptp_time = corrected_ptp_time;
1193   domain->last_local_time = corrected_local_time;
1194 #endif
1195
1196 #ifdef USE_MEASUREMENT_FILTERING
1197 out:
1198 #endif
1199   if (g_atomic_int_get (&domain_stats_n_hooks)) {
1200     GstStructure *stats = gst_structure_new (GST_PTP_STATISTICS_TIME_UPDATED,
1201         "domain", G_TYPE_UINT, domain->domain,
1202         "mean-path-delay-avg", GST_TYPE_CLOCK_TIME, domain->mean_path_delay,
1203         "local-time", GST_TYPE_CLOCK_TIME, corrected_local_time,
1204         "ptp-time", GST_TYPE_CLOCK_TIME, corrected_ptp_time,
1205         "estimated-ptp-time", GST_TYPE_CLOCK_TIME, estimated_ptp_time,
1206         "discontinuity", G_TYPE_INT64, discont,
1207         "synced", G_TYPE_BOOLEAN, synced,
1208         "r-squared", G_TYPE_DOUBLE, r_squared,
1209         "internal-time", GST_TYPE_CLOCK_TIME, internal_time,
1210         "external-time", GST_TYPE_CLOCK_TIME, external_time,
1211         "rate-num", G_TYPE_UINT64, rate_num,
1212         "rate-den", G_TYPE_UINT64, rate_den,
1213         "rate", G_TYPE_DOUBLE, (gdouble) (rate_num) / rate_den,
1214         NULL);
1215     emit_ptp_statistics (domain->domain, stats);
1216     gst_structure_free (stats);
1217   }
1218
1219 }
1220
1221 static gboolean
1222 update_mean_path_delay (PtpDomainData * domain, PtpPendingSync * sync)
1223 {
1224   GstClockTime mean_path_delay, delay_req_delay;
1225   gboolean ret;
1226
1227   /* IEEE 1588 11.3 */
1228   mean_path_delay =
1229       (sync->delay_req_recv_time_remote - sync->sync_send_time_remote +
1230       sync->sync_recv_time_local - sync->delay_req_send_time_local -
1231       (sync->correction_field_sync + sync->correction_field_delay +
1232           32768) / 65536) / 2;
1233
1234 #ifdef USE_RUNNING_AVERAGE_DELAY
1235   /* Track an average round trip time, for a bit of smoothing */
1236   /* Always update before discarding a sample, so genuine changes in
1237    * the network get picked up, eventually */
1238   if (domain->mean_path_delay == 0)
1239     domain->mean_path_delay = mean_path_delay;
1240   else if (mean_path_delay < domain->mean_path_delay)   /* Shorter RTTs carry more weight than longer */
1241     domain->mean_path_delay =
1242         (3 * domain->mean_path_delay + mean_path_delay) / 4;
1243   else
1244     domain->mean_path_delay =
1245         (15 * domain->mean_path_delay + mean_path_delay) / 16;
1246 #else
1247   domain->mean_path_delay = mean_path_delay;
1248 #endif
1249
1250 #ifdef USE_MEASUREMENT_FILTERING
1251   if (sync->follow_up_recv_time_local != GST_CLOCK_TIME_NONE &&
1252       domain->mean_path_delay != 0
1253       && sync->follow_up_recv_time_local >
1254       sync->sync_recv_time_local + 2 * domain->mean_path_delay) {
1255     GST_WARNING ("Sync-follow-up delay for domain %u too big: %" GST_TIME_FORMAT
1256         " > 2 * %" GST_TIME_FORMAT, domain->domain,
1257         GST_TIME_ARGS (sync->follow_up_recv_time_local),
1258         GST_TIME_ARGS (domain->mean_path_delay));
1259     ret = FALSE;
1260     goto out;
1261   }
1262
1263   if (mean_path_delay > 2 * domain->mean_path_delay) {
1264     GST_WARNING ("Mean path delay for domain %u too big: %" GST_TIME_FORMAT
1265         " > 2 * %" GST_TIME_FORMAT, domain->domain,
1266         GST_TIME_ARGS (mean_path_delay),
1267         GST_TIME_ARGS (domain->mean_path_delay));
1268     ret = FALSE;
1269     goto out;
1270   }
1271 #endif
1272
1273   delay_req_delay =
1274       sync->delay_resp_recv_time_local - sync->delay_req_send_time_local;
1275
1276 #ifdef USE_MEASUREMENT_FILTERING
1277   /* delay_req_delay is a RTT, so 2 times the path delay */
1278   if (delay_req_delay > 4 * domain->mean_path_delay) {
1279     GST_WARNING ("Delay-request-response delay for domain %u too big: %"
1280         GST_TIME_FORMAT " > 4 * %" GST_TIME_FORMAT, domain->domain,
1281         GST_TIME_ARGS (delay_req_delay),
1282         GST_TIME_ARGS (domain->mean_path_delay));
1283     ret = FALSE;
1284     goto out;
1285   }
1286 #endif
1287
1288   ret = TRUE;
1289
1290   GST_DEBUG ("Got mean path delay for domain %u: %" GST_TIME_FORMAT " (new: %"
1291       GST_TIME_FORMAT ")", domain->domain,
1292       GST_TIME_ARGS (domain->mean_path_delay), GST_TIME_ARGS (mean_path_delay));
1293   GST_DEBUG ("Delay request delay for domain %u: %" GST_TIME_FORMAT,
1294       domain->domain, GST_TIME_ARGS (delay_req_delay));
1295
1296 #ifdef USE_MEASUREMENT_FILTERING
1297 out:
1298 #endif
1299   if (g_atomic_int_get (&domain_stats_n_hooks)) {
1300     GstStructure *stats =
1301         gst_structure_new (GST_PTP_STATISTICS_PATH_DELAY_MEASURED,
1302         "domain", G_TYPE_UINT, domain->domain,
1303         "mean-path-delay-avg", GST_TYPE_CLOCK_TIME, domain->mean_path_delay,
1304         "mean-path-delay", GST_TYPE_CLOCK_TIME, mean_path_delay,
1305         "delay-request-delay", GST_TYPE_CLOCK_TIME, delay_req_delay, NULL);
1306     emit_ptp_statistics (domain->domain, stats);
1307     gst_structure_free (stats);
1308   }
1309
1310   return ret;
1311 }
1312
1313 static void
1314 handle_sync_message (PtpMessage * msg, GstClockTime receive_time)
1315 {
1316   GList *l;
1317   PtpDomainData *domain = NULL;
1318   PtpPendingSync *sync = NULL;
1319
1320   /* Don't consider messages with the alternate master flag set */
1321   if ((msg->flag_field & 0x0100))
1322     return;
1323
1324   for (l = domain_data; l; l = l->next) {
1325     PtpDomainData *tmp = l->data;
1326
1327     if (msg->domain_number == tmp->domain) {
1328       domain = tmp;
1329       break;
1330     }
1331   }
1332
1333   if (!domain) {
1334     gchar *clock_name;
1335     domain = g_new0 (PtpDomainData, 1);
1336     domain->domain = msg->domain_number;
1337     clock_name = g_strdup_printf ("ptp-clock-%u", domain->domain);
1338     domain->domain_clock =
1339         g_object_new (GST_TYPE_SYSTEM_CLOCK, "name", clock_name, NULL);
1340     g_free (clock_name);
1341     g_queue_init (&domain->pending_syncs);
1342     domain_data = g_list_prepend (domain_data, domain);
1343
1344     g_mutex_lock (&domain_clocks_lock);
1345     domain_clocks = g_list_prepend (domain_clocks, domain);
1346     g_mutex_unlock (&domain_clocks_lock);
1347   }
1348
1349   /* If we have a master clock, ignore this message if it's not coming from there */
1350   if (domain->have_master_clock
1351       && compare_clock_identity (&domain->master_clock_identity,
1352           &msg->source_port_identity) != 0)
1353     return;
1354
1355 #ifdef USE_OPPORTUNISTIC_CLOCK_SELECTION
1356   /* Opportunistic selection of master clock */
1357   if (!domain->have_master_clock)
1358     memcpy (&domain->master_clock_identity, &msg->source_port_identity,
1359         sizeof (PtpClockIdentity));
1360 #else
1361   if (!domain->have_master_clock)
1362     return;
1363 #endif
1364
1365   domain->sync_interval = log2_to_clock_time (msg->log_message_interval);
1366
1367   /* Check if duplicated */
1368   for (l = domain->pending_syncs.head; l; l = l->next) {
1369     PtpPendingSync *tmp = l->data;
1370
1371     if (tmp->sync_seqnum == msg->sequence_id)
1372       return;
1373   }
1374
1375   if (msg->message_specific.sync.origin_timestamp.seconds_field >
1376       GST_CLOCK_TIME_NONE / GST_SECOND) {
1377     GST_FIXME ("Unsupported sync message seconds field value: %"
1378         G_GUINT64_FORMAT " > %" G_GUINT64_FORMAT,
1379         msg->message_specific.sync.origin_timestamp.seconds_field,
1380         GST_CLOCK_TIME_NONE / GST_SECOND);
1381     return;
1382   }
1383
1384   sync = g_new0 (PtpPendingSync, 1);
1385   sync->domain = domain->domain;
1386   sync->sync_seqnum = msg->sequence_id;
1387   sync->sync_recv_time_local = receive_time;
1388   sync->sync_send_time_remote = GST_CLOCK_TIME_NONE;
1389   sync->follow_up_recv_time_local = GST_CLOCK_TIME_NONE;
1390   sync->delay_req_send_time_local = GST_CLOCK_TIME_NONE;
1391   sync->delay_req_recv_time_remote = GST_CLOCK_TIME_NONE;
1392   sync->delay_resp_recv_time_local = GST_CLOCK_TIME_NONE;
1393
1394   /* 0.5 correction factor for division later */
1395   sync->correction_field_sync = msg->correction_field;
1396
1397   if ((msg->flag_field & 0x0200)) {
1398     /* Wait for FOLLOW_UP */
1399   } else {
1400     sync->sync_send_time_remote =
1401         PTP_TIMESTAMP_TO_GST_CLOCK_TIME (msg->message_specific.
1402         sync.origin_timestamp);
1403
1404     if (domain->last_ptp_sync_time != 0
1405         && domain->last_ptp_sync_time >= sync->sync_send_time_remote) {
1406       GST_WARNING ("Backwards PTP times in domain %u: %" GST_TIME_FORMAT " >= %"
1407           GST_TIME_FORMAT, domain->domain,
1408           GST_TIME_ARGS (domain->last_ptp_sync_time),
1409           GST_TIME_ARGS (sync->sync_send_time_remote));
1410       ptp_pending_sync_free (sync);
1411       sync = NULL;
1412       return;
1413     }
1414     domain->last_ptp_sync_time = sync->sync_send_time_remote;
1415
1416     if (send_delay_req (domain, sync)) {
1417       /* Sent delay request */
1418     } else {
1419       update_ptp_time (domain, sync);
1420       ptp_pending_sync_free (sync);
1421       sync = NULL;
1422     }
1423   }
1424
1425   if (sync)
1426     g_queue_push_tail (&domain->pending_syncs, sync);
1427 }
1428
1429 static void
1430 handle_follow_up_message (PtpMessage * msg, GstClockTime receive_time)
1431 {
1432   GList *l;
1433   PtpDomainData *domain = NULL;
1434   PtpPendingSync *sync = NULL;
1435
1436   /* Don't consider messages with the alternate master flag set */
1437   if ((msg->flag_field & 0x0100))
1438     return;
1439
1440   for (l = domain_data; l; l = l->next) {
1441     PtpDomainData *tmp = l->data;
1442
1443     if (msg->domain_number == tmp->domain) {
1444       domain = tmp;
1445       break;
1446     }
1447   }
1448
1449   if (!domain)
1450     return;
1451
1452   /* If we have a master clock, ignore this message if it's not coming from there */
1453   if (domain->have_master_clock
1454       && compare_clock_identity (&domain->master_clock_identity,
1455           &msg->source_port_identity) != 0)
1456     return;
1457
1458   /* Check if we know about this one */
1459   for (l = domain->pending_syncs.head; l; l = l->next) {
1460     PtpPendingSync *tmp = l->data;
1461
1462     if (tmp->sync_seqnum == msg->sequence_id) {
1463       sync = tmp;
1464       break;
1465     }
1466   }
1467
1468   if (!sync)
1469     return;
1470
1471   /* Got a FOLLOW_UP for this already */
1472   if (sync->sync_send_time_remote != GST_CLOCK_TIME_NONE)
1473     return;
1474
1475   if (sync->sync_recv_time_local >= receive_time) {
1476     GST_ERROR ("Got bogus follow up in domain %u: %" GST_TIME_FORMAT " > %"
1477         GST_TIME_FORMAT, domain->domain,
1478         GST_TIME_ARGS (sync->sync_recv_time_local),
1479         GST_TIME_ARGS (receive_time));
1480     g_queue_remove (&domain->pending_syncs, sync);
1481     ptp_pending_sync_free (sync);
1482     return;
1483   }
1484
1485   sync->correction_field_sync += msg->correction_field;
1486   sync->sync_send_time_remote =
1487       PTP_TIMESTAMP_TO_GST_CLOCK_TIME (msg->message_specific.
1488       follow_up.precise_origin_timestamp);
1489   sync->follow_up_recv_time_local = receive_time;
1490
1491   if (domain->last_ptp_sync_time >= sync->sync_send_time_remote) {
1492     GST_WARNING ("Backwards PTP times in domain %u: %" GST_TIME_FORMAT " >= %"
1493         GST_TIME_FORMAT, domain->domain,
1494         GST_TIME_ARGS (domain->last_ptp_sync_time),
1495         GST_TIME_ARGS (sync->sync_send_time_remote));
1496     g_queue_remove (&domain->pending_syncs, sync);
1497     ptp_pending_sync_free (sync);
1498     sync = NULL;
1499     return;
1500   }
1501   domain->last_ptp_sync_time = sync->sync_send_time_remote;
1502
1503   if (send_delay_req (domain, sync)) {
1504     /* Sent delay request */
1505   } else {
1506     update_ptp_time (domain, sync);
1507     g_queue_remove (&domain->pending_syncs, sync);
1508     ptp_pending_sync_free (sync);
1509     sync = NULL;
1510   }
1511 }
1512
1513 static void
1514 handle_delay_resp_message (PtpMessage * msg, GstClockTime receive_time)
1515 {
1516   GList *l;
1517   PtpDomainData *domain = NULL;
1518   PtpPendingSync *sync = NULL;
1519
1520   /* Don't consider messages with the alternate master flag set */
1521   if ((msg->flag_field & 0x0100))
1522     return;
1523
1524   for (l = domain_data; l; l = l->next) {
1525     PtpDomainData *tmp = l->data;
1526
1527     if (msg->domain_number == tmp->domain) {
1528       domain = tmp;
1529       break;
1530     }
1531   }
1532
1533   if (!domain)
1534     return;
1535
1536   /* If we have a master clock, ignore this message if it's not coming from there */
1537   if (domain->have_master_clock
1538       && compare_clock_identity (&domain->master_clock_identity,
1539           &msg->source_port_identity) != 0)
1540     return;
1541
1542   /* Not for us */
1543   if (msg->message_specific.delay_resp.
1544       requesting_port_identity.clock_identity != ptp_clock_id.clock_identity
1545       || msg->message_specific.delay_resp.
1546       requesting_port_identity.port_number != ptp_clock_id.port_number)
1547     return;
1548
1549   domain->min_delay_req_interval =
1550       log2_to_clock_time (msg->log_message_interval);
1551
1552   /* Check if we know about this one */
1553   for (l = domain->pending_syncs.head; l; l = l->next) {
1554     PtpPendingSync *tmp = l->data;
1555
1556     if (tmp->delay_req_seqnum == msg->sequence_id) {
1557       sync = tmp;
1558       break;
1559     }
1560   }
1561
1562   if (!sync)
1563     return;
1564
1565   /* Got a DELAY_RESP for this already */
1566   if (sync->delay_req_recv_time_remote != GST_CLOCK_TIME_NONE)
1567     return;
1568
1569   if (sync->delay_req_send_time_local > receive_time) {
1570     GST_ERROR ("Got bogus delay response in domain %u: %" GST_TIME_FORMAT " > %"
1571         GST_TIME_FORMAT, domain->domain,
1572         GST_TIME_ARGS (sync->delay_req_send_time_local),
1573         GST_TIME_ARGS (receive_time));
1574     g_queue_remove (&domain->pending_syncs, sync);
1575     ptp_pending_sync_free (sync);
1576     return;
1577   }
1578
1579   sync->correction_field_delay = msg->correction_field;
1580
1581   sync->delay_req_recv_time_remote =
1582       PTP_TIMESTAMP_TO_GST_CLOCK_TIME (msg->message_specific.
1583       delay_resp.receive_timestamp);
1584   sync->delay_resp_recv_time_local = receive_time;
1585
1586   if (domain->mean_path_delay != 0
1587       && sync->sync_send_time_remote > sync->delay_req_recv_time_remote) {
1588     GST_WARNING ("Sync send time after delay req receive time for domain %u: %"
1589         GST_TIME_FORMAT " > %" GST_TIME_FORMAT, domain->domain,
1590         GST_TIME_ARGS (sync->sync_send_time_remote),
1591         GST_TIME_ARGS (sync->delay_req_recv_time_remote));
1592     g_queue_remove (&domain->pending_syncs, sync);
1593     ptp_pending_sync_free (sync);
1594     return;
1595   }
1596
1597   if (update_mean_path_delay (domain, sync))
1598     update_ptp_time (domain, sync);
1599   g_queue_remove (&domain->pending_syncs, sync);
1600   ptp_pending_sync_free (sync);
1601 }
1602
1603 static void
1604 handle_ptp_message (PtpMessage * msg, GstClockTime receive_time)
1605 {
1606   /* Ignore our own messages */
1607   if (msg->source_port_identity.clock_identity == ptp_clock_id.clock_identity &&
1608       msg->source_port_identity.port_number == ptp_clock_id.port_number)
1609     return;
1610
1611   switch (msg->message_type) {
1612     case PTP_MESSAGE_TYPE_ANNOUNCE:
1613       handle_announce_message (msg, receive_time);
1614       break;
1615     case PTP_MESSAGE_TYPE_SYNC:
1616       handle_sync_message (msg, receive_time);
1617       break;
1618     case PTP_MESSAGE_TYPE_FOLLOW_UP:
1619       handle_follow_up_message (msg, receive_time);
1620       break;
1621     case PTP_MESSAGE_TYPE_DELAY_RESP:
1622       handle_delay_resp_message (msg, receive_time);
1623       break;
1624     default:
1625       break;
1626   }
1627 }
1628
1629 static gboolean
1630 have_stdin_data_cb (GIOChannel * channel, GIOCondition condition,
1631     gpointer user_data)
1632 {
1633   GIOStatus status;
1634   StdIOHeader header;
1635   gchar buffer[8192];
1636   GError *err = NULL;
1637   gsize read;
1638
1639   if ((condition & G_IO_STATUS_EOF)) {
1640     GST_ERROR ("Got EOF on stdin");
1641     g_main_loop_quit (main_loop);
1642     return G_SOURCE_REMOVE;
1643   }
1644
1645   status =
1646       g_io_channel_read_chars (channel, (gchar *) & header, sizeof (header),
1647       &read, &err);
1648   if (status == G_IO_STATUS_ERROR) {
1649     GST_ERROR ("Failed to read from stdin: %s", err->message);
1650     g_main_loop_quit (main_loop);
1651     return G_SOURCE_REMOVE;
1652   } else if (status == G_IO_STATUS_EOF) {
1653     GST_ERROR ("Got EOF on stdin");
1654     g_main_loop_quit (main_loop);
1655     return G_SOURCE_REMOVE;
1656   } else if (status != G_IO_STATUS_NORMAL) {
1657     GST_ERROR ("Unexpected stdin read status: %d", status);
1658     g_main_loop_quit (main_loop);
1659     return G_SOURCE_REMOVE;
1660   } else if (read != sizeof (header)) {
1661     GST_ERROR ("Unexpected read size: %" G_GSIZE_FORMAT, read);
1662     g_main_loop_quit (main_loop);
1663     return G_SOURCE_REMOVE;
1664   } else if (header.size > 8192) {
1665     GST_ERROR ("Unexpected size: %u", header.size);
1666     g_main_loop_quit (main_loop);
1667     return G_SOURCE_REMOVE;
1668   }
1669
1670   status = g_io_channel_read_chars (channel, buffer, header.size, &read, &err);
1671   if (status == G_IO_STATUS_ERROR) {
1672     GST_ERROR ("Failed to read from stdin: %s", err->message);
1673     g_main_loop_quit (main_loop);
1674     return G_SOURCE_REMOVE;
1675   } else if (status == G_IO_STATUS_EOF) {
1676     GST_ERROR ("EOF on stdin");
1677     g_main_loop_quit (main_loop);
1678     return G_SOURCE_REMOVE;
1679   } else if (status != G_IO_STATUS_NORMAL) {
1680     GST_ERROR ("Unexpected stdin read status: %d", status);
1681     g_main_loop_quit (main_loop);
1682     return G_SOURCE_REMOVE;
1683   } else if (read != header.size) {
1684     GST_ERROR ("Unexpected read size: %" G_GSIZE_FORMAT, read);
1685     g_main_loop_quit (main_loop);
1686     return G_SOURCE_REMOVE;
1687   }
1688
1689   switch (header.type) {
1690     case TYPE_EVENT:
1691     case TYPE_GENERAL:{
1692       GstClockTime receive_time = gst_util_get_timestamp ();
1693       PtpMessage msg;
1694
1695       if (parse_ptp_message (&msg, (const guint8 *) buffer, header.size)) {
1696         dump_ptp_message (&msg);
1697         handle_ptp_message (&msg, receive_time);
1698       }
1699       break;
1700     }
1701     default:
1702     case TYPE_CLOCK_ID:{
1703       if (header.size != 8) {
1704         GST_ERROR ("Unexpected clock id size (%u != 8)", header.size);
1705         g_main_loop_quit (main_loop);
1706         return G_SOURCE_REMOVE;
1707       }
1708       g_mutex_lock (&ptp_lock);
1709       ptp_clock_id.clock_identity = GST_READ_UINT64_BE (buffer);
1710       ptp_clock_id.port_number = getpid ();
1711       GST_DEBUG ("Got clock id 0x%016" G_GINT64_MODIFIER "x %u",
1712           ptp_clock_id.clock_identity, ptp_clock_id.port_number);
1713       g_cond_signal (&ptp_cond);
1714       g_mutex_unlock (&ptp_lock);
1715       break;
1716     }
1717   }
1718
1719   return G_SOURCE_CONTINUE;
1720 }
1721
1722 /* Cleanup all announce messages and announce message senders
1723  * that are timed out by now, and clean up all pending syncs
1724  * that are missing their FOLLOW_UP or DELAY_RESP */
1725 static gboolean
1726 cleanup_cb (gpointer data)
1727 {
1728   GstClockTime now = gst_util_get_timestamp ();
1729   GList *l, *m, *n;
1730
1731   for (l = domain_data; l; l = l->next) {
1732     PtpDomainData *domain = l->data;
1733
1734     for (n = domain->announce_senders; n;) {
1735       PtpAnnounceSender *sender = n->data;
1736       gboolean timed_out = TRUE;
1737
1738       /* Keep only 5 messages per sender around */
1739       while (g_queue_get_length (&sender->announce_messages) > 5) {
1740         PtpAnnounceMessage *msg = g_queue_pop_head (&sender->announce_messages);
1741         g_free (msg);
1742       }
1743
1744       for (m = sender->announce_messages.head; m; m = m->next) {
1745         PtpAnnounceMessage *msg = m->data;
1746
1747         if (msg->receive_time +
1748             sender->announce_interval * PTP_ANNOUNCE_RECEIPT_TIMEOUT > now) {
1749           timed_out = FALSE;
1750           break;
1751         }
1752       }
1753
1754       if (timed_out) {
1755         GST_DEBUG ("Announce sender 0x%016" G_GINT64_MODIFIER "x %u timed out",
1756             sender->master_clock_identity.clock_identity,
1757             sender->master_clock_identity.port_number);
1758         g_queue_foreach (&sender->announce_messages, (GFunc) g_free, NULL);
1759         g_queue_clear (&sender->announce_messages);
1760       }
1761
1762       if (g_queue_get_length (&sender->announce_messages) == 0) {
1763         GList *tmp = n->next;
1764
1765         if (compare_clock_identity (&sender->master_clock_identity,
1766                 &domain->master_clock_identity) == 0)
1767           GST_WARNING ("currently selected master clock timed out");
1768         g_free (sender);
1769         domain->announce_senders =
1770             g_list_delete_link (domain->announce_senders, n);
1771         n = tmp;
1772       } else {
1773         n = n->next;
1774       }
1775     }
1776     select_best_master_clock (domain, now);
1777
1778     /* Clean up any pending syncs */
1779     for (n = domain->pending_syncs.head; n;) {
1780       PtpPendingSync *sync = n->data;
1781       gboolean timed_out = FALSE;
1782
1783       /* Time out pending syncs after 4 sync intervals or 10 seconds,
1784        * and pending delay reqs after 4 delay req intervals or 10 seconds
1785        */
1786       if (sync->delay_req_send_time_local != GST_CLOCK_TIME_NONE &&
1787           ((domain->min_delay_req_interval != 0
1788                   && sync->delay_req_send_time_local +
1789                   4 * domain->min_delay_req_interval < now)
1790               || (sync->delay_req_send_time_local + 10 * GST_SECOND < now))) {
1791         timed_out = TRUE;
1792       } else if ((domain->sync_interval != 0
1793               && sync->sync_recv_time_local + 4 * domain->sync_interval < now)
1794           || (sync->sync_recv_time_local + 10 * GST_SECOND < now)) {
1795         timed_out = TRUE;
1796       }
1797
1798       if (timed_out) {
1799         GList *tmp = n->next;
1800         ptp_pending_sync_free (sync);
1801         g_queue_delete_link (&domain->pending_syncs, n);
1802         n = tmp;
1803       } else {
1804         n = n->next;
1805       }
1806     }
1807   }
1808
1809   return G_SOURCE_CONTINUE;
1810 }
1811
1812 static gpointer
1813 ptp_helper_main (gpointer data)
1814 {
1815   GSource *cleanup_source;
1816
1817   GST_DEBUG ("Starting PTP helper loop");
1818
1819   /* Check all 5 seconds, if we have to cleanup ANNOUNCE or pending syncs message */
1820   cleanup_source = g_timeout_source_new_seconds (5);
1821   g_source_set_priority (cleanup_source, G_PRIORITY_DEFAULT);
1822   g_source_set_callback (cleanup_source, (GSourceFunc) cleanup_cb, NULL, NULL);
1823   g_source_attach (cleanup_source, main_context);
1824   g_source_unref (cleanup_source);
1825
1826   g_main_loop_run (main_loop);
1827   GST_DEBUG ("Stopped PTP helper loop");
1828
1829   g_mutex_lock (&ptp_lock);
1830   ptp_clock_id.clock_identity = GST_PTP_CLOCK_ID_NONE;
1831   ptp_clock_id.port_number = 0;
1832   initted = FALSE;
1833   g_cond_signal (&ptp_cond);
1834   g_mutex_unlock (&ptp_lock);
1835
1836   return NULL;
1837 }
1838
1839 /**
1840  * gst_ptp_is_supported:
1841  *
1842  * Check if PTP clocks are generally supported on this system, and if previous
1843  * initializations did not fail.
1844  *
1845  * Returns: %TRUE if PTP clocks are generally supported on this system, and
1846  * previous initializations did not fail.
1847  *
1848  * Since: 1.6
1849  */
1850 gboolean
1851 gst_ptp_is_supported (void)
1852 {
1853   return supported;
1854 }
1855
1856 /**
1857  * gst_ptp_is_initialized:
1858  *
1859  * Check if the GStreamer PTP clock subsystem is initialized.
1860  *
1861  * Returns: %TRUE if the GStreamer PTP clock subsystem is intialized.
1862  *
1863  * Since: 1.6
1864  */
1865 gboolean
1866 gst_ptp_is_initialized (void)
1867 {
1868   return initted;
1869 }
1870
1871 /**
1872  * gst_ptp_init:
1873  * @clock_id: PTP clock id of this process' clock or %GST_PTP_CLOCK_ID_NONE
1874  * @interfaces: (transfer none) (array zero-terminated=1): network interfaces to run the clock on
1875  *
1876  * Initialize the GStreamer PTP subsystem and create a PTP ordinary clock in
1877  * slave-only mode for all domains on the given @interfaces with the
1878  * given @clock_id.
1879  *
1880  * If @clock_id is %GST_PTP_CLOCK_ID_NONE, a clock id is automatically
1881  * generated from the MAC address of the first network interface.
1882  *
1883  *
1884  * This function is automatically called by gst_ptp_clock_new() with default
1885  * parameters if it wasn't called before.
1886  *
1887  * Returns: %TRUE if the GStreamer PTP clock subsystem could be initialized.
1888  *
1889  * Since: 1.6
1890  */
1891 gboolean
1892 gst_ptp_init (guint64 clock_id, gchar ** interfaces)
1893 {
1894   gboolean ret;
1895   const gchar *env;
1896   gchar **argv = NULL;
1897   gint argc, argc_c;
1898   gint fd_r, fd_w;
1899   GError *err = NULL;
1900   GSource *stdin_source;
1901
1902   GST_DEBUG_CATEGORY_INIT (ptp_debug, "ptp", 0, "PTP clock");
1903
1904   g_mutex_lock (&ptp_lock);
1905   if (!supported) {
1906     GST_ERROR ("PTP not supported");
1907     ret = FALSE;
1908     goto done;
1909   }
1910
1911   if (initted) {
1912     GST_DEBUG ("PTP already initialized");
1913     ret = TRUE;
1914     goto done;
1915   }
1916
1917   if (ptp_helper_pid) {
1918     GST_DEBUG ("PTP currently initializing");
1919     goto wait;
1920   }
1921
1922   if (!domain_stats_hooks_initted) {
1923     g_hook_list_init (&domain_stats_hooks, sizeof (GHook));
1924     domain_stats_hooks_initted = TRUE;
1925   }
1926
1927   argc = 1;
1928   if (clock_id != GST_PTP_CLOCK_ID_NONE)
1929     argc += 2;
1930   if (interfaces != NULL)
1931     argc += 2 * g_strv_length (interfaces);
1932
1933   argv = g_new0 (gchar *, argc + 2);
1934   argc_c = 0;
1935
1936   env = g_getenv ("GST_PTP_HELPER_1_0");
1937   if (env == NULL)
1938     env = g_getenv ("GST_PTP_HELPER");
1939   if (env != NULL && *env != '\0') {
1940     GST_LOG ("Trying GST_PTP_HELPER env var: %s", env);
1941     argv[argc_c++] = g_strdup (env);
1942   } else {
1943     argv[argc_c++] = g_strdup (GST_PTP_HELPER_INSTALLED);
1944   }
1945
1946   if (clock_id != GST_PTP_CLOCK_ID_NONE) {
1947     argv[argc_c++] = g_strdup ("-c");
1948     argv[argc_c++] = g_strdup_printf ("0x%016" G_GINT64_MODIFIER "x", clock_id);
1949   }
1950
1951   if (interfaces != NULL) {
1952     gchar **ptr = interfaces;
1953
1954     while (*ptr) {
1955       argv[argc_c++] = g_strdup ("-i");
1956       argv[argc_c++] = g_strdup (*ptr);
1957       ptr++;
1958     }
1959   }
1960
1961   main_context = g_main_context_new ();
1962   main_loop = g_main_loop_new (main_context, FALSE);
1963
1964   ptp_helper_thread =
1965       g_thread_try_new ("ptp-helper-thread", ptp_helper_main, NULL, &err);
1966   if (!ptp_helper_thread) {
1967     GST_ERROR ("Failed to start PTP helper thread: %s", err->message);
1968     g_clear_error (&err);
1969     ret = FALSE;
1970     goto done;
1971   }
1972
1973   if (!g_spawn_async_with_pipes (NULL, argv, NULL, 0, NULL, NULL,
1974           &ptp_helper_pid, &fd_w, &fd_r, NULL, &err)) {
1975     GST_ERROR ("Failed to start ptp helper process: %s", err->message);
1976     g_clear_error (&err);
1977     ret = FALSE;
1978     supported = FALSE;
1979     goto done;
1980   }
1981
1982   stdin_channel = g_io_channel_unix_new (fd_r);
1983   g_io_channel_set_encoding (stdin_channel, NULL, NULL);
1984   g_io_channel_set_buffered (stdin_channel, FALSE);
1985   g_io_channel_set_close_on_unref (stdin_channel, TRUE);
1986   stdin_source =
1987       g_io_create_watch (stdin_channel, G_IO_IN | G_IO_PRI | G_IO_HUP);
1988   g_source_set_priority (stdin_source, G_PRIORITY_DEFAULT);
1989   g_source_set_callback (stdin_source, (GSourceFunc) have_stdin_data_cb, NULL,
1990       NULL);
1991   g_source_attach (stdin_source, main_context);
1992   g_source_unref (stdin_source);
1993
1994   /* Create stdout channel */
1995   stdout_channel = g_io_channel_unix_new (fd_w);
1996   g_io_channel_set_encoding (stdout_channel, NULL, NULL);
1997   g_io_channel_set_close_on_unref (stdout_channel, TRUE);
1998   g_io_channel_set_buffered (stdout_channel, FALSE);
1999
2000   delay_req_rand = g_rand_new ();
2001
2002   initted = TRUE;
2003
2004 wait:
2005   GST_DEBUG ("Waiting for PTP to be initialized");
2006
2007   while (ptp_clock_id.clock_identity == GST_PTP_CLOCK_ID_NONE && initted)
2008     g_cond_wait (&ptp_cond, &ptp_lock);
2009
2010   ret = initted;
2011   if (ret) {
2012     GST_DEBUG ("Initialized and got clock id 0x%016" G_GINT64_MODIFIER "x %u",
2013         ptp_clock_id.clock_identity, ptp_clock_id.port_number);
2014   } else {
2015     GST_ERROR ("Failed to initialize");
2016     supported = FALSE;
2017   }
2018
2019 done:
2020   g_strfreev (argv);
2021
2022   if (!ret) {
2023     if (ptp_helper_pid) {
2024       kill (ptp_helper_pid, SIGKILL);
2025       waitpid (ptp_helper_pid, NULL, 0);
2026       g_spawn_close_pid (ptp_helper_pid);
2027     }
2028     ptp_helper_pid = 0;
2029
2030     if (stdin_channel)
2031       g_io_channel_unref (stdin_channel);
2032     stdin_channel = NULL;
2033     if (stdout_channel)
2034       g_io_channel_unref (stdout_channel);
2035     stdout_channel = NULL;
2036
2037     if (main_loop && ptp_helper_thread) {
2038       g_main_loop_quit (main_loop);
2039       g_thread_join (ptp_helper_thread);
2040     }
2041     ptp_helper_thread = NULL;
2042     if (main_loop)
2043       g_main_loop_unref (main_loop);
2044     main_loop = NULL;
2045     if (main_context)
2046       g_main_context_unref (main_context);
2047     main_context = NULL;
2048
2049     if (delay_req_rand)
2050       g_rand_free (delay_req_rand);
2051     delay_req_rand = NULL;
2052   }
2053
2054   g_mutex_unlock (&ptp_lock);
2055
2056   return ret;
2057 }
2058
2059 /**
2060  * gst_ptp_deinit:
2061  *
2062  * Deinitialize the GStreamer PTP subsystem and stop the PTP clock. If there
2063  * are any remaining GstPtpClock instances, they won't be further synchronized
2064  * to the PTP network clock.
2065  *
2066  * Since: 1.6
2067  */
2068 void
2069 gst_ptp_deinit (void)
2070 {
2071   GList *l, *m;
2072
2073   g_mutex_lock (&ptp_lock);
2074
2075   if (ptp_helper_pid) {
2076     kill (ptp_helper_pid, SIGKILL);
2077     waitpid (ptp_helper_pid, NULL, 0);
2078     g_spawn_close_pid (ptp_helper_pid);
2079   }
2080   ptp_helper_pid = 0;
2081
2082   if (stdin_channel)
2083     g_io_channel_unref (stdin_channel);
2084   stdin_channel = NULL;
2085   if (stdout_channel)
2086     g_io_channel_unref (stdout_channel);
2087   stdout_channel = NULL;
2088
2089   if (main_loop && ptp_helper_thread) {
2090     GThread *tmp = ptp_helper_thread;
2091     ptp_helper_thread = NULL;
2092     g_mutex_unlock (&ptp_lock);
2093     g_main_loop_quit (main_loop);
2094     g_thread_join (tmp);
2095     g_mutex_lock (&ptp_lock);
2096   }
2097   if (main_loop)
2098     g_main_loop_unref (main_loop);
2099   main_loop = NULL;
2100   if (main_context)
2101     g_main_context_unref (main_context);
2102   main_context = NULL;
2103
2104   if (delay_req_rand)
2105     g_rand_free (delay_req_rand);
2106   delay_req_rand = NULL;
2107
2108   for (l = domain_data; l; l = l->next) {
2109     PtpDomainData *domain = l->data;
2110
2111     for (m = domain->announce_senders; m; m = m->next) {
2112       PtpAnnounceSender *sender = m->data;
2113
2114       g_queue_foreach (&sender->announce_messages, (GFunc) g_free, NULL);
2115       g_queue_clear (&sender->announce_messages);
2116       g_free (sender);
2117     }
2118     g_list_free (domain->announce_senders);
2119
2120     g_queue_foreach (&domain->pending_syncs, (GFunc) ptp_pending_sync_free,
2121         NULL);
2122     g_queue_clear (&domain->pending_syncs);
2123     gst_object_unref (domain->domain_clock);
2124     g_free (domain);
2125   }
2126   g_list_free (domain_data);
2127   domain_data = NULL;
2128   g_list_foreach (domain_clocks, (GFunc) g_free, NULL);
2129   g_list_free (domain_clocks);
2130   domain_clocks = NULL;
2131
2132   ptp_clock_id.clock_identity = GST_PTP_CLOCK_ID_NONE;
2133   ptp_clock_id.port_number = 0;
2134
2135   initted = FALSE;
2136
2137   g_mutex_unlock (&ptp_lock);
2138 }
2139
2140 #define DEFAULT_DOMAIN 0
2141
2142 enum
2143 {
2144   PROP_0,
2145   PROP_DOMAIN,
2146   PROP_INTERNAL_CLOCK
2147 };
2148
2149 #define GST_PTP_CLOCK_GET_PRIVATE(obj)  \
2150   (G_TYPE_INSTANCE_GET_PRIVATE ((obj), GST_TYPE_PTP_CLOCK, GstPtpClockPrivate))
2151
2152 struct _GstPtpClockPrivate
2153 {
2154   guint domain;
2155   GstClock *domain_clock;
2156   gulong domain_stats_id;
2157 };
2158
2159 #define gst_ptp_clock_parent_class parent_class
2160 G_DEFINE_TYPE (GstPtpClock, gst_ptp_clock, GST_TYPE_SYSTEM_CLOCK);
2161
2162 static void gst_ptp_clock_set_property (GObject * object, guint prop_id,
2163     const GValue * value, GParamSpec * pspec);
2164 static void gst_ptp_clock_get_property (GObject * object, guint prop_id,
2165     GValue * value, GParamSpec * pspec);
2166 static void gst_ptp_clock_finalize (GObject * object);
2167
2168 static GstClockTime gst_ptp_clock_get_internal_time (GstClock * clock);
2169
2170 static void
2171 gst_ptp_clock_class_init (GstPtpClockClass * klass)
2172 {
2173   GObjectClass *gobject_class;
2174   GstClockClass *clock_class;
2175
2176   gobject_class = G_OBJECT_CLASS (klass);
2177   clock_class = GST_CLOCK_CLASS (klass);
2178
2179   g_type_class_add_private (klass, sizeof (GstPtpClockPrivate));
2180
2181   gobject_class->finalize = gst_ptp_clock_finalize;
2182   gobject_class->get_property = gst_ptp_clock_get_property;
2183   gobject_class->set_property = gst_ptp_clock_set_property;
2184
2185   g_object_class_install_property (gobject_class, PROP_DOMAIN,
2186       g_param_spec_uint ("domain", "Domain",
2187           "The PTP domain", 0, G_MAXUINT8,
2188           DEFAULT_DOMAIN,
2189           G_PARAM_READWRITE | G_PARAM_CONSTRUCT_ONLY | G_PARAM_STATIC_STRINGS));
2190
2191   g_object_class_install_property (gobject_class, PROP_INTERNAL_CLOCK,
2192       g_param_spec_object ("internal-clock", "Internal Clock",
2193           "Internal clock", GST_TYPE_CLOCK,
2194           G_PARAM_READABLE | G_PARAM_STATIC_STRINGS));
2195
2196   clock_class->get_internal_time = gst_ptp_clock_get_internal_time;
2197 }
2198
2199 static void
2200 gst_ptp_clock_init (GstPtpClock * self)
2201 {
2202   GstPtpClockPrivate *priv;
2203
2204   self->priv = priv = GST_PTP_CLOCK_GET_PRIVATE (self);
2205
2206   GST_OBJECT_FLAG_SET (self, GST_CLOCK_FLAG_CAN_SET_MASTER);
2207   GST_OBJECT_FLAG_SET (self, GST_CLOCK_FLAG_NEEDS_STARTUP_SYNC);
2208
2209   priv->domain = DEFAULT_DOMAIN;
2210 }
2211
2212 static gboolean
2213 gst_ptp_clock_ensure_domain_clock (GstPtpClock * self)
2214 {
2215   gboolean got_clock = TRUE;
2216
2217   if (G_UNLIKELY (!self->priv->domain_clock)) {
2218     g_mutex_lock (&domain_clocks_lock);
2219     if (!self->priv->domain_clock) {
2220       GList *l;
2221
2222       got_clock = FALSE;
2223
2224       for (l = domain_clocks; l; l = l->next) {
2225         PtpDomainData *clock_data = l->data;
2226
2227         if (clock_data->domain == self->priv->domain
2228             && clock_data->last_ptp_time != 0) {
2229           self->priv->domain_clock = clock_data->domain_clock;
2230           got_clock = TRUE;
2231           break;
2232         }
2233       }
2234     }
2235     g_mutex_unlock (&domain_clocks_lock);
2236     if (got_clock) {
2237       g_object_notify (G_OBJECT (self), "internal-clock");
2238       gst_clock_set_synced (GST_CLOCK (self), TRUE);
2239     }
2240   }
2241
2242   return got_clock;
2243 }
2244
2245 static gboolean
2246 gst_ptp_clock_stats_callback (guint8 domain, const GstStructure * stats,
2247     gpointer user_data)
2248 {
2249   GstPtpClock *self = user_data;
2250
2251   if (domain != self->priv->domain
2252       || !gst_structure_has_name (stats, GST_PTP_STATISTICS_TIME_UPDATED))
2253     return TRUE;
2254
2255   /* Let's set our internal clock */
2256   if (!gst_ptp_clock_ensure_domain_clock (self))
2257     return TRUE;
2258
2259   self->priv->domain_stats_id = 0;
2260
2261   return FALSE;
2262 }
2263
2264 static void
2265 gst_ptp_clock_set_property (GObject * object, guint prop_id,
2266     const GValue * value, GParamSpec * pspec)
2267 {
2268   GstPtpClock *self = GST_PTP_CLOCK (object);
2269
2270   switch (prop_id) {
2271     case PROP_DOMAIN:
2272       self->priv->domain = g_value_get_uint (value);
2273       gst_ptp_clock_ensure_domain_clock (self);
2274       if (!self->priv->domain_clock)
2275         self->priv->domain_stats_id =
2276             gst_ptp_statistics_callback_add (gst_ptp_clock_stats_callback, self,
2277             NULL);
2278       break;
2279     default:
2280       G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
2281       break;
2282   }
2283 }
2284
2285 static void
2286 gst_ptp_clock_get_property (GObject * object, guint prop_id,
2287     GValue * value, GParamSpec * pspec)
2288 {
2289   GstPtpClock *self = GST_PTP_CLOCK (object);
2290
2291   switch (prop_id) {
2292     case PROP_DOMAIN:
2293       g_value_set_uint (value, self->priv->domain);
2294       break;
2295     case PROP_INTERNAL_CLOCK:
2296       gst_ptp_clock_ensure_domain_clock (self);
2297       g_value_set_object (value, self->priv->domain_clock);
2298       break;
2299     default:
2300       G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
2301       break;
2302   }
2303 }
2304
2305 static void
2306 gst_ptp_clock_finalize (GObject * object)
2307 {
2308   GstPtpClock *self = GST_PTP_CLOCK (object);
2309
2310   if (self->priv->domain_stats_id)
2311     gst_ptp_statistics_callback_remove (self->priv->domain_stats_id);
2312
2313   G_OBJECT_CLASS (gst_ptp_clock_parent_class)->finalize (object);
2314 }
2315
2316 static GstClockTime
2317 gst_ptp_clock_get_internal_time (GstClock * clock)
2318 {
2319   GstPtpClock *self = GST_PTP_CLOCK (clock);
2320
2321   gst_ptp_clock_ensure_domain_clock (self);
2322
2323   if (!self->priv->domain_clock) {
2324     GST_ERROR_OBJECT (self, "Domain %u has no clock yet and is not synced",
2325         self->priv->domain);
2326     return GST_CLOCK_TIME_NONE;
2327   }
2328
2329   return gst_clock_get_time (self->priv->domain_clock);
2330 }
2331
2332 /**
2333  * gst_ptp_clock_new:
2334  * @name: Name of the clock
2335  * @domain: PTP domain
2336  *
2337  * Creates a new PTP clock instance that exports the PTP time of the master
2338  * clock in @domain. This clock can be slaved to other clocks as needed.
2339  *
2340  * If gst_ptp_init() was not called before, this will call gst_ptp_init() with
2341  * default parameters.
2342  *
2343  *
2344  * This clock only returns valid timestamps after it received the first
2345  * times from the PTP master clock on the network. Once this happens the
2346  * GstPtpClock::internal-clock property will become non-NULL. You can connect
2347  * to the notify::internal-clock signal to get notified about this, or
2348  * alternatively use gst_ptp_clock_wait_ready() to wait for this to happen.
2349  *
2350  * Since: 1.6
2351  */
2352 GstClock *
2353 gst_ptp_clock_new (const gchar * name, guint domain)
2354 {
2355   g_return_val_if_fail (name != NULL, NULL);
2356   g_return_val_if_fail (domain <= G_MAXUINT8, NULL);
2357
2358   if (!initted && !gst_ptp_init (GST_PTP_CLOCK_ID_NONE, NULL)) {
2359     GST_ERROR ("Failed to initialize PTP");
2360     return NULL;
2361   }
2362
2363   return g_object_new (GST_TYPE_PTP_CLOCK, "name", name, "domain", domain,
2364       NULL);
2365 }
2366
2367 typedef struct
2368 {
2369   guint8 domain;
2370   const GstStructure *stats;
2371 } DomainStatsMarshalData;
2372
2373 static void
2374 domain_stats_marshaller (GHook * hook, DomainStatsMarshalData * data)
2375 {
2376   GstPtpStatisticsCallback callback = (GstPtpStatisticsCallback) hook->func;
2377
2378   if (!callback (data->domain, data->stats, hook->data))
2379     g_hook_destroy (&domain_stats_hooks, hook->hook_id);
2380 }
2381
2382 static void
2383 emit_ptp_statistics (guint8 domain, const GstStructure * stats)
2384 {
2385   DomainStatsMarshalData data = { domain, stats };
2386
2387   g_mutex_lock (&ptp_lock);
2388   g_hook_list_marshal (&domain_stats_hooks, TRUE,
2389       (GHookMarshaller) domain_stats_marshaller, &data);
2390   g_mutex_unlock (&ptp_lock);
2391 }
2392
2393 /**
2394  * gst_ptp_statistics_callback_add:
2395  * @callback: GstPtpStatisticsCallback to call
2396  * @user_data: Data to pass to the callback
2397  * @destroy_data: GDestroyNotify to destroy the data
2398  *
2399  * Installs a new statistics callback for gathering PTP statistics. See
2400  * GstPtpStatisticsCallback for a list of statistics that are provided.
2401  *
2402  * Returns: Id for the callback that can be passed to
2403  * gst_ptp_statistics_callback_remove()
2404  *
2405  * Since: 1.6
2406  */
2407 gulong
2408 gst_ptp_statistics_callback_add (GstPtpStatisticsCallback callback,
2409     gpointer user_data, GDestroyNotify destroy_data)
2410 {
2411   GHook *hook;
2412
2413   g_mutex_lock (&ptp_lock);
2414
2415   if (!domain_stats_hooks_initted) {
2416     g_hook_list_init (&domain_stats_hooks, sizeof (GHook));
2417     domain_stats_hooks_initted = TRUE;
2418   }
2419
2420   hook = g_hook_alloc (&domain_stats_hooks);
2421   hook->func = callback;
2422   hook->data = user_data;
2423   hook->destroy = destroy_data;
2424   g_hook_prepend (&domain_stats_hooks, hook);
2425   g_atomic_int_add (&domain_stats_n_hooks, 1);
2426
2427   g_mutex_unlock (&ptp_lock);
2428
2429   return hook->hook_id;
2430 }
2431
2432 /**
2433  * gst_ptp_statistics_callback_remove:
2434  * @id: Callback id to remove
2435  *
2436  * Removes a PTP statistics callback that was previously added with
2437  * gst_ptp_statistics_callback_add().
2438  *
2439  * Since: 1.6
2440  */
2441 void
2442 gst_ptp_statistics_callback_remove (gulong id)
2443 {
2444   g_mutex_lock (&ptp_lock);
2445   if (g_hook_destroy (&domain_stats_hooks, id))
2446     g_atomic_int_add (&domain_stats_n_hooks, -1);
2447   g_mutex_unlock (&ptp_lock);
2448 }
2449
2450 #else /* HAVE_PTP */
2451
2452 GType
2453 gst_ptp_clock_get_type (void)
2454 {
2455   return G_TYPE_INVALID;
2456 }
2457
2458 gboolean
2459 gst_ptp_is_supported (void)
2460 {
2461   return FALSE;
2462 }
2463
2464 gboolean
2465 gst_ptp_is_initialized (void)
2466 {
2467   return FALSE;
2468 }
2469
2470 gboolean
2471 gst_ptp_init (guint64 clock_id, gchar ** interfaces)
2472 {
2473   return FALSE;
2474 }
2475
2476 void
2477 gst_ptp_deinit (void)
2478 {
2479 }
2480
2481 GstClock *
2482 gst_ptp_clock_new (const gchar * name, guint domain)
2483 {
2484   return NULL;
2485 }
2486
2487 gboolean
2488 gst_ptp_clock_wait_ready (GstPtpClock * self, GstClockTime timeout)
2489 {
2490   return FALSE;
2491 }
2492
2493 gulong
2494 gst_ptp_statistics_callback_add (GstPtpStatisticsCallback callback,
2495     gpointer user_data, GDestroyNotify destroy_data)
2496 {
2497   return 0;
2498 }
2499
2500 void
2501 gst_ptp_statistics_callback_remove (gulong id)
2502 {
2503   return;
2504 }
2505
2506 #endif