一覧NW-14-3 · DHCPと時刻同期

NTPによる時刻同期

往復の記録から時刻差と遅延を計算し、同期元の状態を確認します。

説明

NTPは時刻を問い合わせるだけでなく、送受信の四つの時刻から通信遅延と時計の差を推定します。サーバーのstratumが小さいことだけでは、その端末が正しく同期しているとは判断できません。

NTP応答の48バイトを組み立てて各欄を復号し、四時刻からoffsetと往復遅延を計算します。往路・復路の非対称による誤差と同期元の状態を確認し、時刻補正によるログへの影響を調べます。

1NTPの64ビット時刻

NTPの時刻は秒の整数32ビットと小数32ビットで構成されます。ここでは同じera内の3900000000秒を基準にします。

import structNTPヘッダーの形式; base=この例の秒の基準3900000000; t1=要求送信時刻base<<32; t2=サーバー受信時刻t1+round(0.060*(2**32)); t3=サーバー送信時刻t1+round(0.070*(2**32)); (hex64ビット値を確認(t1),hex(t2),hex(t3))

t1は端末の要求送信時刻、t2はサーバーの要求受信時刻、t3はサーバーの応答送信時刻です。整数と小数を合わせた値なので、64ビット整数をそのまま秒と読むことはできません。

2応答の48バイトを作る

LI0・version4・mode4、stratum2、poll6、precision-20とし、認証等の拡張を付けない基本ヘッダーを使います。

packet=NTP基本ヘッダーを組み立てるstruct.pack("!BBbbII4sQQQQ",0x24,2,6,-20,0,0,b"REF1",(base-1)<<32,t1,t2,t3); (len長さとバイト列(packet),packet.hex(" "))

48バイトです。poll6は2の6乗で64秒、precision-20は2の-20乗秒を表します。最後の三つの時刻がOriginate、Receive、Transmitに対応し、端末が受け取った瞬間のt4は端末側で記録します。応答だけを保存してもt4まで含まれているわけではありません。

3モードと四時刻のうち三つを復号する

先頭ビットと末尾24バイトを取り出します。

values=Originate・Receive・Transmitstruct.unpack("!QQQ",packet[24:48]); (packet[0]ビット欄と時刻を復号>>6,(packet[0]>>3)&7,packet[0]&7,packet[1],[round(v/(2**32)-base,3) for v in values])

LI0、version4、server mode4、stratum2です。時刻は基準から0・0.060・0.070秒で、要求の送信時刻がOriginateへ返っています。問い合わせへの対応を照合する欄でもあり、単に送信元がUDP123というだけでは正しい応答と決めません。

4四時刻からoffsetとdelayを求める

端末の受信時刻t4は基準から0.050秒です。丸め誤差を避けるため、以下の計算はDecimalでこの時刻条件を使います。

from decimal import Decimal as D十進で時刻差を計算; a,b,c,d=四時刻の条件map(D,["0","0.060","0.070","0.050"]); offset=時計の差の推定((b-a)+(c-d))/2; delay=サーバー処理を除いた往復(d-a)-(c-b); (offset,delay)補正方向と遅延

offsetは+0.040秒、往復遅延は0.040秒です。サーバー時計が端末より40ms進んでいる推定なので、端末は進める方向の補正が必要です。サーバー内の処理時間10msを往復時間50msから差し引いています。

5往路と復路が非対称な場合

時計は本当は一致し、往路10ms、サーバー処理5ms、復路50msとします。

a,b,c,d=非対称な通信時間map(D,["0","0.010","0.015","0.065"]); (((b-a)同じ式で生じる推定誤差+(c-d))/2,(d-a)-(c-b))

計算上はoffset-20ms、遅延60msになります。四時刻の式は片道遅延の非対称を時計差から完全には分離できず、時計が一致していても誤差が生じます。複数の測定や同期元の選択が重要になる理由です。

6stratumだけで同期元を決めない

二つの同期元候補を、stratum・leap indicator・root distance相当の記録で比較します。ここではLI3等を除き、同期可能な候補を残す簡略モデルです。

sources=同期元候補の状態[dict(name="A",stratum=1,li=3,distance_ms=1),dict(name="B",stratum=2,li=0,distance_ms=2)]; usable=未同期と無効な階層を除く[s for s in sources if s["li"]!=3 and 1<=s["stratum"]<=15]; [(s["name"]残った候補,s["stratum"]) for s in usable]

Aはstratum1でもLI3の未同期なので除外され、Bが残ります。実際のNTPは到達性、ばらつき、他候補との整合なども評価します。階層の番号が小さいことだけを精度や正常性の保証としないようにします。

7徐々に補正する場合の時間

40msの差を、時計の進み方を500ppm増やして補正する単純な条件です。元々の周波数誤差は0として計算します。

correction=修正する40msD("0.040"); rate=500ppmを比率へ変換D("500")/D("1000000"); correction/rate必要な秒数

80秒です。時刻を一度に飛ばすstepと、進み方を変えるslewではログの見え方が違います。500ppmはこの計算条件で、すべてのNTP実装の固定値ではありません。

8ログの経過時間を別の時計で確認する

処理中に壁時計が40ms進んだ例で、壁時計と単調時計の差を比べます。

wall_start=補正前の壁時計D("100.000"); wall_end=40ms進めた後の壁時計D("101.040"); mono_start=単調時計の開始値D("10.000"); mono_end=単調時計の終了値D("11.000"); (wall_end二つの経過時間-wall_start,mono_end-mono_start)

壁時計では1.040秒、単調時計では1秒です。イベントの日時と処理の経過時間は用途が違うため、時計補正があった区間のログではその影響を確認します。時刻がそろったことと、経過時間を正しく測れたことは別に確かめます。

まとめ

offsetは四時刻から推定し、非対称な経路では時計差に誤差が含まれます。

同期元の状態と補正方法を確認し、日時と経過時間の用途を分けます。

操作履歴と解説を振り返れます。

Python 3

1
import struct; base=3900000000; t1=base<<32; t2=t1+round(0.060*(2**32)); t3=t1+round(0.070*(2**32)); (hex(t1),hex(t2),hex(t3))
('0xe875470000000000', '0xe87547000f5c28f6', '0xe875470011eb851f')
2
packet=struct.pack("!BBbbII4sQQQQ",0x24,2,6,-20,0,0,b"REF1",(base-1)<<32,t1,t2,t3); (len(packet),packet.hex(" "))
(48, '24 02 06 ec 00 00 00 00 00 00 00 00 52 45 46 31 e8 75 46 ff 00 00 00 00 e8 75 47 00 00 00 00 00 e8 75 47 00 0f 5c 28 f6 e8 75 47 00 11 eb 85 1f')
3
values=struct.unpack("!QQQ",packet[24:48]); (packet[0]>>6,(packet[0]>>3)&7,packet[0]&7,packet[1],[round(v/(2**32)-base,3) for v in values])
(0, 4, 4, 2, [0.0, 0.06, 0.07])
4
from decimal import Decimal as D; a,b,c,d=map(D,["0","0.060","0.070","0.050"]); offset=((b-a)+(c-d))/2; delay=(d-a)-(c-b); (offset,delay)
(Decimal('0.040'), Decimal('0.040'))
5
a,b,c,d=map(D,["0","0.010","0.015","0.065"]); (((b-a)+(c-d))/2,(d-a)-(c-b))
(Decimal('-0.020'), Decimal('0.060'))
6
sources=[dict(name="A",stratum=1,li=3,distance_ms=1),dict(name="B",stratum=2,li=0,distance_ms=2)]; usable=[s for s in sources if s["li"]!=3 and 1<=s["stratum"]<=15]; [(s["name"],s["stratum"]) for s in usable]
[('B', 2)]
7
correction=D("0.040"); rate=D("500")/D("1000000"); correction/rate
Decimal('8E+1')
8
wall_start=D("100.000"); wall_end=D("101.040"); mono_start=D("10.000"); mono_end=D("11.000"); (wall_end-wall_start,mono_end-mono_start)
(Decimal('1.040'), Decimal('1.000'))

演習完了