support hardware timestamps (#346)

Summary:
Pull Request resolved: #346

Add an option to enable hardware timestamps on ntp responder.
Default is software (same behaviour).
Apart from enabling hardware timestamper (which is only supported on Linux) we also run periodic PHC-SYS measurements

Reviewed By: pmazzini

Differential Revision: D56255701

cmd/ntpresponder/main.go

@@ -31,6 +31,7 @@ import (
 	"github.com/facebook/time/ntp/responder/checker"
 	"github.com/facebook/time/ntp/responder/server"
 	"github.com/facebook/time/ntp/responder/stats"
+	"github.com/facebook/time/timestamp"
 	log "github.com/sirupsen/logrus"
 )
 
@@ -56,6 +57,7 @@ func main() {
 	flag.BoolVar(&s.Config.ShouldAnnounce, "announce", false, "Advertize IPs")
 	flag.DurationVar(&s.Config.ExtraOffset, "extraoffset", 0, "Extra offset to return to clients")
 	flag.BoolVar(&s.Config.ManageLoopback, "manage-loopback", true, "Add/remove IPs. If false, these must be managed elsewhere")
+	flag.StringVar(&s.Config.TimestampType, "timestamptype", timestamp.SWTIMESTAMP, fmt.Sprintf("Timestamp type. Can be: %s, %s", timestamp.HWTIMESTAMP, timestamp.SWTIMESTAMP))
 
 	flag.Parse()
 	s.Config.IPs.SetDefault()

ntp/responder/server/config.go

@@ -21,6 +21,8 @@ import (
 	"net"
 	"strings"
 	"time"
+
+	"github.com/facebook/time/timestamp"
 )
 
 // DefaultServerIPs is a default list of IPs server will bind to if nothing else is specified
@@ -37,7 +39,9 @@ type Config struct {
 	RefID          string
 	ShouldAnnounce bool
 	Stratum        int
+	TimestampType  string
 	Workers        int
+	phcOffset      time.Duration
 }
 
 // MultiIPs is a wrapper allowing to set multiple IPs with flag parser
@@ -76,5 +80,8 @@ func (c *Config) Validate() error {
 	if c.Workers < 1 {
 		return fmt.Errorf("will not start without workers")
 	}
+	if c.TimestampType != timestamp.HWTIMESTAMP && c.TimestampType != timestamp.SWTIMESTAMP {
+		return fmt.Errorf("invalid timestamp type %s", c.TimestampType)
+	}
 	return nil
 }

ntp/responder/server/config_test.go

@@ -21,6 +21,7 @@ import (
 	"net"
 	"testing"
 
+	"github.com/facebook/time/timestamp"
 	"github.com/stretchr/testify/require"
 )
 
@@ -63,8 +64,18 @@ func TestConfigSetDefault(t *testing.T) {
 }
 
 func TestConfigValidate(t *testing.T) {
-	c := Config{Workers: 42}
+	c := Config{Workers: 42, TimestampType: timestamp.SWTIMESTAMP}
 	require.NoError(t, c.Validate())
+
+	// Workers
 	c.Workers = 0
 	require.Error(t, c.Validate())
+	c.Workers = 42
+
+	// Timestamp type
+	c.TimestampType = "bad"
+	require.Error(t, c.Validate())
+	c.TimestampType = timestamp.HWTIMESTAMP
+
+	require.NoError(t, c.Validate())
 }

ntp/responder/server/server.go

@@ -99,6 +99,23 @@ func (s *Server) Start(ctx context.Context, cancelFunc context.CancelFunc) {
 		}
 	}()
 
+	// Run PHC-SYS offset periodically
+	if s.Config.TimestampType == timestamp.HWTIMESTAMP {
+		log.Infof("Starting periodic measurement between phc and sysclock")
+		go func() {
+			for ; ; time.Sleep(time.Second) {
+				offset, err := phcOffset(s.Config.Iface)
+				if err != nil {
+					log.Errorf("[phcoffset] failed to get PHC-SYS offset: %v", err)
+					cancelFunc()
+					return
+				}
+				s.Config.phcOffset = offset
+				log.Debugf("[phcoffset] offset between PHC and SYS: %v", offset)
+			}
+		}()
+	}
+
 	for {
 		select {
 		case <-ctx.Done():
@@ -139,9 +156,9 @@ func (s *Server) startListener(conn *net.UDPConn) {
 		log.Fatalf("Getting event connection FD: %s", err)
 	}
 
-	// Allow reading of kernel timestamps via socket
-	if err = timestamp.EnableSWTimestampsRx(connFd); err != nil {
-		log.Fatalf("enabling timestamp error: %s", err)
+	// Enable RX timestamps
+	if err := timestamp.EnableTimestamps(s.Config.TimestampType, connFd, s.Config.Iface); err != nil {
+		log.Fatal(err)
 	}
 
 	err = unix.SetNonblock(connFd, false)
@@ -162,6 +179,10 @@ func (s *Server) startListener(conn *net.UDPConn) {
 			continue
 		}
 
+		if s.Config.TimestampType == timestamp.HWTIMESTAMP {
+			rxTS = rxTS.Add(s.Config.phcOffset)
+		}
+
 		if err := request.UnmarshalBinary(buf[:bbuf]); err != nil {
 			log.Errorf("failed to parse ntp packet: %s", err)
 			s.Stats.IncReadError()

ntp/responder/server/ip_darwin.go → ntp/responder/server/server_darwin.go

@@ -20,6 +20,7 @@ import (
 	"fmt"
 	"net"
 	"os/exec"
+	"time"
 
 	errors "github.com/pkg/errors"
 )
@@ -75,3 +76,9 @@ func deleteIfaceIP(iface *net.Interface, addr *net.IP) error {
 
 	return nil
 }
+
+// PHCOffset periodically checks for PHC-SYS offset and updates it in the config
+// PHC reading is not supported on Darwin
+func phcOffset(iface string) (time.Duration, error) {
+	return 0, nil
+}

ntp/responder/server/ip_darwin_test.go → ntp/responder/server/server_darwin_test.go

@@ -44,3 +44,9 @@ func TestCheckIPFalse(t *testing.T) {
 	require.NoError(t, err)
 	require.False(t, assigned)
 }
+
+func TestConfigPHCOffset(t *testing.T) {
+	offset, err := tc.PHCOffset()
+	require.Nil(t, err)
+	require.Equal(t, 0, offset)
+}

ntp/responder/server/ip_freebsd.go → ntp/responder/server/server_freebsd.go

@@ -20,6 +20,7 @@ import (
 	"fmt"
 	"net"
 	"os/exec"
+	"time"
 
 	errors "github.com/pkg/errors"
 )
@@ -76,3 +77,9 @@ func deleteIfaceIP(iface *net.Interface, addr *net.IP) error {
 
 	return nil
 }
+
+// PHCOffset periodically checks for PHC-SYS offset and updates it in the config
+// PHC reading is not supported on FreeBSD
+func phcOffset(iface string) (time.Duration, error) {
+	return 0, nil
+}

ntp/responder/server/ip_freebsd_test.go → ntp/responder/server/server_freebsd_test.go

@@ -44,3 +44,9 @@ func TestCheckIPFalse(t *testing.T) {
 	require.NoError(t, err)
 	require.False(t, assigned)
 }
+
+func TestConfigPHCOffset(t *testing.T) {
+	offset, err := tc.PHCOffset()
+	require.Nil(t, err)
+	require.Equal(t, 0, offset)
+}

ntp/responder/server/ip_linux.go → ntp/responder/server/server_linux.go

@@ -18,7 +18,9 @@ package server
 
 import (
 	"net"
+	"time"
 
+	"github.com/facebook/time/phc"
 	"github.com/jsimonetti/rtnetlink/rtnl"
 	errors "github.com/pkg/errors"
 )
@@ -92,3 +94,17 @@ func deleteIfaceIP(iface *net.Interface, addr *net.IP) error {
 
 	return nil
 }
+
+// PHCOffset periodically checks for PHC-SYS offset and updates it in the config
+func phcOffset(iface string) (time.Duration, error) {
+	device, err := phc.DeviceFromIface(iface)
+	if err != nil {
+		return 0, err
+	}
+
+	res, err := phc.TimeAndOffsetFromDevice(device, phc.MethodSyscallClockGettime)
+	if err != nil {
+		return 0, err
+	}
+	return res.Offset, nil
+}

ntp/responder/server/server_test.go

@@ -114,6 +114,7 @@ func TestListener(t *testing.T) {
 			ExpectedListeners: 1,
 			ExpectedWorkers:   0,
 		},
+		Config: Config{Workers: 42, TimestampType: timestamp.SWTIMESTAMP},
 	}
 	conn, err := net.ListenUDP("udp", &net.UDPAddr{IP: net.ParseIP("127.0.0.1"), Port: 0})
 	require.Nil(t, err)
@@ -160,8 +161,9 @@ func TestServer(t *testing.T) {
 			ExpectedListeners: 1,
 			ExpectedWorkers:   int64(workers),
 		},
-		Stats: &stats.JSONStats{},
-		tasks: make(chan task, workers),
+		Stats:  &stats.JSONStats{},
+		tasks:  make(chan task, workers),
+		Config: Config{Workers: workers, TimestampType: timestamp.SWTIMESTAMP},
 	}
 	// create workers
 	for i := 0; i < workers; i++ {

phc/device.go

@@ -139,3 +139,24 @@ func IfacesInfo() ([]IfaceData, error) {
 	}
 	return res, nil
 }
+
+// DeviceFromIface returns a path to a PHC device from a network interface
+func DeviceFromIface(iface string) (string, error) {
+	ifaces, err := IfacesInfo()
+	if err != nil {
+		return "", err
+	}
+	if len(ifaces) == 0 {
+		return "", fmt.Errorf("no network devices found")
+	}
+
+	for _, d := range ifaces {
+		if d.Iface.Name == iface {
+			if d.TSInfo.PHCIndex < 0 {
+				return "", fmt.Errorf("no PHC support for %s", iface)
+			}
+			return fmt.Sprintf("/dev/ptp%d", d.TSInfo.PHCIndex), nil
+		}
+	}
+	return "", fmt.Errorf("%s interface is not found", iface)
+}

phc/device_test.go

@@ -0,0 +1,36 @@
+/*
+Copyright (c) Facebook, Inc. and its affiliates.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package phc
+
+import (
+	"fmt"
+	"testing"
+
+	"github.com/stretchr/testify/require"
+)
+
+func TestDeviceFromIfaceNotSupported(t *testing.T) {
+	dev, err := DeviceFromIface("lo")
+	require.Equal(t, fmt.Errorf("no PHC support for lo"), err)
+	require.Equal(t, "", dev)
+}
+
+func TestDeviceFromIfaceNotFound(t *testing.T) {
+	dev, err := DeviceFromIface("lol-does-not-exist")
+	require.Equal(t, fmt.Errorf("lol-does-not-exist interface is not found"), err)
+	require.Equal(t, "", dev)
+}

timestamp/timestamp_darwin.go

@@ -66,3 +66,16 @@ func EnableSWTimestampsRx(connFd int) error {
 	// Allow reading of SW timestamps via socket
 	return unix.SetsockoptInt(connFd, unix.SOL_SOCKET, timestamping, 1)
 }
+
+// EnableTimestamps enables timestamps on the socket based on requested type
+func EnableTimestamps(ts string, connFd int, _ string) error {
+	switch ts {
+	case SWTIMESTAMP:
+		if err := EnableSWTimestampsRx(connFd); err != nil {
+			return fmt.Errorf("Cannot enable software timestamps: %w", err)
+		}
+	default:
+		return fmt.Errorf("Unrecognized timestamp type: %s", ts)
+	}
+	return nil
+}

timestamp/timestamp_darwin_test.go

@@ -17,6 +17,7 @@ limitations under the License.
 package timestamp
 
 import (
+	"fmt"
 	"net"
 	"testing"
 	"time"
@@ -55,6 +56,32 @@ func TestEnableSWTimestampsRx(t *testing.T) {
 	require.Greater(t, kernelTimestampsEnabled, 0, "Kernel timestamps are not enabled")
 }
 
+func TestEnableTimestamps(t *testing.T) {
+	// listen to incoming udp packets
+	conn, err := net.ListenUDP("udp", &net.UDPAddr{IP: net.ParseIP("127.0.0.1"), Port: 0})
+	require.NoError(t, err)
+	defer conn.Close()
+
+	connFd, err := ConnFd(conn)
+	require.NoError(t, err)
+
+	// SOFTWARE
+	// Allow reading of kernel timestamps via socket
+	err = EnableTimestamps(SWTIMESTAMP, connFd, "lo")
+	require.NoError(t, err)
+
+	// Check that socket option is set
+	kernelTimestampsEnabled, err := unix.GetsockoptInt(connFd, unix.SOL_SOCKET, unix.SO_TIMESTAMP)
+	require.NoError(t, err)
+
+	// To be enabled must be > 0
+	require.Greater(t, kernelTimestampsEnabled, 0, "Kernel timestamps are not enabled")
+
+	// HARDWARE
+	err = EnableTimestamps(HWTIMESTAMP, connFd, "lo")
+	require.Equal(t, fmt.Errorf("Unrecognized timestamp type: %s", HWTIMESTAMP), err)
+}
+
 func TestReadPacketWithRXTimestamp(t *testing.T) {
 	request := []byte{1, 2, 3, 4, 5, 6, 7, 8, 9, 42}
 	// listen to incoming udp packets

timestamp/timestamp_freebsd.go

@@ -66,3 +66,16 @@ func EnableSWTimestampsRx(connFd int) error {
 	// Allow reading of SW timestamps via socket
 	return unix.SetsockoptInt(connFd, unix.SOL_SOCKET, timestamping, 1)
 }
+
+// EnableTimestamps enables timestamps on the socket based on requested type
+func EnableTimestamps(ts string, connFd int, _ string) error {
+	switch ts {
+	case SWTIMESTAMP:
+		if err := EnableSWTimestampsRx(connFd); err != nil {
+			return fmt.Errorf("Cannot enable software timestamps: %w", err)
+		}
+	default:
+		return fmt.Errorf("Unrecognized timestamp type: %s", ts)
+	}
+	return nil
+}