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