ntp: add support for multiple suspended sockets

With some hardware it takes milliseconds to get the HW TX timestamp.

Rework the code to handle multiple suspended client-only sockets at the
same time in order to allow longer timeouts, which may overlap for
different sources. Instead of waiting for the first read event simply
suspend the socket and create timeout when the HW TX timestamp is
requested.

diff --git a/ntp_io.c b/ntp_io.c
index e5f3418d36508f9dc4cc8c96fb81ba7fdff05f43..81bbc525f08e719844d7598119e3ef57c2d3f881 100644 (file)
--- a/ntp_io.c
+++ b/ntp_io.c
@@ -467,11 +467,6 @@ read_from_socket(int sock_fd, int event, void *anything)
   SCK_Message *messages;
   int i, received, flags = 0;
 
-#ifdef HAVE_LINUX_TIMESTAMPING
-  if (NIO_Linux_ProcessEvent(sock_fd, event))
-    return;
-#endif
-
   if (event == SCH_FILE_EXCEPTION) {
 #ifdef HAVE_LINUX_TIMESTAMPING
     flags |= SCK_FLAG_MSG_ERRQUEUE;

diff --git a/ntp_io_linux.c b/ntp_io_linux.c
index 6fa0184dd6c2a1377afe5671ea763df9c2e99760..cb0768f06323d502bf7b246c2ddf7f4fb9bb63bf 100644 (file)
--- a/ntp_io_linux.c
+++ b/ntp_io_linux.c
@@ -39,6 +39,7 @@
 #include "hwclock.h"
 #include "local.h"
 #include "logging.h"
+#include "memory.h"
 #include "ntp_core.h"
 #include "ntp_io.h"
 #include "ntp_io_linux.h"
@@ -87,16 +88,20 @@ static int permanent_ts_options;
 /* When sending client requests to a close and fast server, it is possible that
    a response will be received before the HW transmit timestamp of the request
    itself.  To avoid processing of the response without the HW timestamp, we
-   monitor events returned by select() and suspend reading of packets from the
-   receive queue for up to 200 microseconds.  As the requests are normally
-   separated by at least about 1 millisecond (1/8th of the minimum poll), it is
-   sufficient to monitor and suspend one socket at a time. */
-static int monitored_socket;
-static int suspended_socket;
-static SCH_TimeoutID resume_timeout_id;
+   suspend reading of packets from the receive queue until a HW transmit
+   timestamp is received from the error queue or a timeout reached. */
 
 #define RESUME_TIMEOUT 200.0e-6
 
+struct HwTsSocket {
+  int sock_fd;
+  int suspended;
+  SCH_TimeoutID timeout_id;
+};
+
+/* Array of (HwTsSocket *) indexed by the file descriptor */
+static ARR_Instance hw_ts_socks;
+
 /* Unbound socket keeping the kernel RX timestamping permanently enabled
    in order to avoid a race condition between receiving a server response
    and the kernel actually starting to timestamp received packets after
@@ -412,8 +417,7 @@ NIO_Linux_Initialise(void)
   /* Kernels before 4.7 ignore timestamping flags set in control messages */
   permanent_ts_options = !SYS_Linux_CheckKernelVersion(4, 7);
 
-  monitored_socket = INVALID_SOCK_FD;
-  suspended_socket = INVALID_SOCK_FD;
+  hw_ts_socks = ARR_CreateInstance(sizeof (struct HwTsSocket *));
   dummy_rxts_socket = INVALID_SOCK_FD;
 }
 
@@ -425,6 +429,10 @@ NIO_Linux_Finalise(void)
   struct Interface *iface;
   unsigned int i;
 
+  for (i = 0; i < ARR_GetSize(hw_ts_socks); i++)
+    Free(*(struct HwTsSocket **)ARR_GetElement(hw_ts_socks, i));
+  ARR_DestroyInstance(hw_ts_socks);
+
   if (dummy_rxts_socket != INVALID_SOCK_FD)
     SCK_CloseSocket(dummy_rxts_socket);
 
@@ -472,69 +480,84 @@ NIO_Linux_SetTimestampSocketOptions(int sock_fd, int client_only, int *events)
 
 /* ================================================== */
 
-static void
-resume_socket(int sock_fd)
+static struct HwTsSocket *
+get_hw_ts_socket(int sock_fd, int new)
 {
-  if (monitored_socket == sock_fd)
-    monitored_socket = INVALID_SOCK_FD;
-
-  if (sock_fd == INVALID_SOCK_FD || sock_fd != suspended_socket)
-    return;
+  struct HwTsSocket *s, **sp;
 
-  suspended_socket = INVALID_SOCK_FD;
+  if (sock_fd < 0)
+    return NULL;
 
-  SCH_SetFileHandlerEvent(sock_fd, SCH_FILE_INPUT, 1);
+  while (sock_fd >= ARR_GetSize(hw_ts_socks)) {
+    if (!new)
+      return NULL;
+    s = NULL;
+    ARR_AppendElement(hw_ts_socks, &s);
+  }
 
-  DEBUG_LOG("Resumed RX processing %s timeout fd=%d",
-            resume_timeout_id ? "before" : "on", sock_fd);
+  sp = ARR_GetElement(hw_ts_socks, sock_fd);
 
-  if (resume_timeout_id) {
-    SCH_RemoveTimeout(resume_timeout_id);
-    resume_timeout_id = 0;
+  if (!*sp && new) {
+    *sp = s = MallocNew(struct HwTsSocket);
+    s->sock_fd = sock_fd;
+    s->suspended = 0;
+    s->timeout_id = 0;
   }
+
+  return *sp;
 }
 
 /* ================================================== */
 
 static void
-resume_timeout(void *arg)
+resume_socket(int sock_fd)
 {
-  resume_timeout_id = 0;
-  resume_socket(suspended_socket);
+  struct HwTsSocket *ts_sock = get_hw_ts_socket(sock_fd, 0);
+
+  if (!ts_sock)
+    return;
+
+  if (ts_sock->suspended) {
+    SCH_SetFileHandlerEvent(ts_sock->sock_fd, SCH_FILE_INPUT, 1);
+
+    DEBUG_LOG("Resumed RX processing %s timeout fd=%d",
+              ts_sock->timeout_id ? "before" : "on", ts_sock->sock_fd);
+  }
+
+  ts_sock->suspended = 0;
+  SCH_RemoveTimeout(ts_sock->timeout_id);
+  ts_sock->timeout_id = 0;
 }
 
 /* ================================================== */
 
 static void
-suspend_socket(int sock_fd)
+resume_timeout(void *arg)
 {
-  resume_socket(suspended_socket);
-
-  suspended_socket = sock_fd;
-
-  SCH_SetFileHandlerEvent(suspended_socket, SCH_FILE_INPUT, 0);
-  resume_timeout_id = SCH_AddTimeoutByDelay(RESUME_TIMEOUT, resume_timeout, NULL);
+  struct HwTsSocket *ts_sock = arg;
 
-  DEBUG_LOG("Suspended RX processing fd=%d", sock_fd);
+  ts_sock->timeout_id = 0;
+  resume_socket(ts_sock->sock_fd);
 }
 
 /* ================================================== */
 
-int
-NIO_Linux_ProcessEvent(int sock_fd, int event)
+static void
+suspend_socket(int sock_fd)
 {
-  if (sock_fd != monitored_socket)
-    return 0;
+  struct HwTsSocket *ts_sock = get_hw_ts_socket(sock_fd, 1);
 
-  if (event == SCH_FILE_INPUT) {
-    suspend_socket(monitored_socket);
-    monitored_socket = INVALID_SOCK_FD;
+  if (!ts_sock)
+    return;
 
-    /* Don't process the message yet */
-    return 1;
-  }
+  /* Remove previous timeout if there is one */
+  SCH_RemoveTimeout(ts_sock->timeout_id);
+
+  ts_sock->suspended = 1;
+  ts_sock->timeout_id = SCH_AddTimeoutByDelay(RESUME_TIMEOUT, resume_timeout, ts_sock);
+  SCH_SetFileHandlerEvent(ts_sock->sock_fd, SCH_FILE_INPUT, 0);
 
-  return 0;
+  DEBUG_LOG("Suspended RX processing fd=%d", ts_sock->sock_fd);
 }
 
 /* ================================================== */
@@ -825,11 +848,11 @@ NIO_Linux_RequestTxTimestamp(SCK_Message *message, int sock_fd)
   if (!ts_flags)
     return;
 
-  /* If a HW transmit timestamp is requested on a client socket, monitor
-     events on the socket in order to avoid processing of a fast response
-     without the HW timestamp of the request */
+  /* If a HW transmit timestamp is requested on a client-only socket,
+     suspend reading from it to avoid processing a response before the
+     HW timestamp of the request is received */
   if (ts_tx_flags & SOF_TIMESTAMPING_TX_HARDWARE && !NIO_IsServerSocket(sock_fd))
-    monitored_socket = sock_fd;
+    suspend_socket(sock_fd);
 
   /* Check if TX timestamping is disabled on this socket */
   if (permanent_ts_options || !NIO_IsServerSocket(sock_fd))

diff --git a/ntp_io_linux.h b/ntp_io_linux.h
index 4d3af133db9708b25163fb0d0b3e2e18cc86e220..b0b52bd58ff0f64bbef2355225b4885daddbfb60 100644 (file)
--- a/ntp_io_linux.h
+++ b/ntp_io_linux.h
@@ -35,8 +35,6 @@ extern void NIO_Linux_Finalise(void);
 
 extern int NIO_Linux_SetTimestampSocketOptions(int sock_fd, int client_only, int *events);
 
-extern int NIO_Linux_ProcessEvent(int sock_fd, int event);
-
 extern int NIO_Linux_ProcessMessage(SCK_Message *message, NTP_Local_Address *local_addr,
                                     NTP_Local_Timestamp *local_ts, int event);