forked from KNEMC/KLALB
434 lines
13 KiB
Java
434 lines
13 KiB
Java
package org.kne.cloud.network.ntp;
|
|
|
|
import java.io.Closeable;
|
|
import java.io.IOException;
|
|
import java.math.BigDecimal;
|
|
import java.util.ArrayList;
|
|
import java.util.Collections;
|
|
import java.util.HashMap;
|
|
import java.util.HashSet;
|
|
import java.util.Iterator;
|
|
import java.util.List;
|
|
import java.util.Map;
|
|
import java.util.Map.Entry;
|
|
import java.util.Set;
|
|
import java.util.Vector;
|
|
import java.util.concurrent.ConcurrentHashMap;
|
|
import java.util.concurrent.atomic.AtomicBoolean;
|
|
|
|
import org.kne.cloud.clock.HighAccuracyClock;
|
|
import org.kne.cloud.clock.NTPTimestamps;
|
|
import org.kne.cloud.network.MultiProtocolSocketAddress;
|
|
import org.kne.cloud.network.ntp.NTPv4Protocol.NTPPeer;
|
|
import org.kne.math.Long128;
|
|
|
|
|
|
public class NTPContext implements Closeable, AutoCloseable {
|
|
private HighAccuracyClock clock;
|
|
private static final boolean debug = false;
|
|
private static final int REQUEST_COUNT = 5;
|
|
|
|
private Long128 systemFrequencyOffset = NTPTimestamps.nanosToNtp128BitTimeInterval(new Long128(5000));
|
|
private Long128 localPrecision = NTPTimestamps.nanosToNtp128BitTimeInterval(new Long128(1000));
|
|
private static final Long128 adjustThreshold0 = Long128.valueOf(1000000000L);
|
|
private static final Long128 adjustThreshold1 = Long128.valueOf(100000000L);
|
|
private AtomicBoolean firstSync=new AtomicBoolean(true);
|
|
|
|
private int minStratum = 16;
|
|
private volatile PeerInfo currentClock = new PeerInfo();
|
|
|
|
private volatile boolean closed = false;
|
|
private Runnable send = new Runnable() {
|
|
@Override
|
|
public void run() {
|
|
while (!closed) {
|
|
try {
|
|
clearPackets();
|
|
Map<NTPv4Protocol, List<NTPPeer>> mlp = new HashMap<>();
|
|
ios.forEach((v) -> {
|
|
mlp.put(v, v.getPeersWillSend());
|
|
});
|
|
for (int i = 0; i < REQUEST_COUNT; i++) {
|
|
Set<Entry<NTPv4Protocol, List<NTPPeer>>> mlps = mlp.entrySet();
|
|
for (Iterator<Entry<NTPv4Protocol, List<NTPPeer>>> iterator = mlps.iterator(); iterator
|
|
.hasNext();) {
|
|
Entry<NTPv4Protocol, List<NTPPeer>> object = iterator.next();
|
|
List<NTPPeer> val = object.getValue();
|
|
for (NTPPeer perr : val) {
|
|
try {
|
|
object.getKey().request(perr);
|
|
} catch (IOException e) {
|
|
if (debug)
|
|
e.printStackTrace();
|
|
}
|
|
}
|
|
|
|
}
|
|
|
|
Thread.sleep(100);
|
|
}
|
|
Thread.sleep(500);
|
|
|
|
if (!mlp.isEmpty())
|
|
mergeAndApply();
|
|
|
|
} catch (InterruptedException e) {
|
|
e.printStackTrace();
|
|
}
|
|
}
|
|
}
|
|
|
|
};
|
|
|
|
public NTPContext(HighAccuracyClock clock) {
|
|
this.clock = clock;
|
|
Thread ts = new Thread(send);
|
|
ts.setName("NTPv4 Send Thread");
|
|
ts.start();
|
|
}
|
|
|
|
public void syncToSystem() {
|
|
clock.syncToClock(new HighAccuracyClock());
|
|
minStratum = 15;
|
|
currentClock.stratum = Math.min(currentClock.stratum, minStratum);
|
|
currentClock.leapIndicator = 0;
|
|
}
|
|
|
|
public int getMinStratum() {
|
|
return minStratum;
|
|
}
|
|
|
|
public void setMinStratum(int minStratum) {
|
|
this.minStratum = minStratum;
|
|
}
|
|
|
|
public int getStratum() {
|
|
return currentClock.stratum;
|
|
}
|
|
|
|
public HighAccuracyClock getClock() {
|
|
return clock;
|
|
}
|
|
|
|
private ConcurrentHashMap<MultiProtocolSocketAddress, List<NTPv4Packet>> recvmap = new ConcurrentHashMap<MultiProtocolSocketAddress, List<NTPv4Packet>>();
|
|
|
|
protected void clearPackets() {
|
|
recvmap.clear();
|
|
}
|
|
|
|
protected void putPacket(NTPv4Packet nv4, MultiProtocolSocketAddress inetSocketAddress) {
|
|
checkIP();
|
|
List<NTPv4Packet> newv = new Vector<NTPv4Packet>();
|
|
List<NTPv4Packet> oldv = recvmap.putIfAbsent(inetSocketAddress, newv);
|
|
if (oldv == null) {
|
|
oldv = newv;
|
|
}
|
|
|
|
synchronized (oldv) {
|
|
oldv.add(nv4);
|
|
while (oldv.size() > 10) {
|
|
oldv.remove(0);
|
|
}
|
|
}
|
|
}
|
|
|
|
private void checkIP() {
|
|
Set<Entry<MultiProtocolSocketAddress, List<NTPv4Packet>>> ens = recvmap.entrySet();
|
|
for (Iterator<Entry<MultiProtocolSocketAddress, List<NTPv4Packet>>> iterator = ens.iterator(); iterator
|
|
.hasNext();) {
|
|
Entry<MultiProtocolSocketAddress, List<NTPv4Packet>> entry = (Entry<MultiProtocolSocketAddress, List<NTPv4Packet>>) iterator
|
|
.next();
|
|
AtomicBoolean ab = new AtomicBoolean(false);
|
|
ios.forEach((x) -> {
|
|
if (!ab.get())
|
|
if (x.findPeer(entry.getKey()) != null) {
|
|
ab.set(true);
|
|
return;
|
|
}
|
|
});
|
|
if (!ab.get()) {
|
|
iterator.remove();
|
|
}
|
|
}
|
|
}
|
|
|
|
private class PeerInfo implements Comparable<PeerInfo> {
|
|
private MultiProtocolSocketAddress address;
|
|
private int leapIndicator = 3;
|
|
private int stratum = 16;
|
|
private int referenceIdentifier;
|
|
private int pollInterval;
|
|
private byte precision;
|
|
private long rootDelay = Integer.MAX_VALUE;
|
|
private long rootDispersion = Integer.MAX_VALUE;
|
|
private Long128 referenceTimestamp = Long128.ZERO;
|
|
private Long128 uploadDelay;
|
|
private Long128 downloadDelay;
|
|
private Long128 rtt;
|
|
|
|
@Override
|
|
public String toString() {
|
|
return "PeerInfo [address=" + address + ", leapIndicator=" + leapIndicator + ", stratum=" + stratum
|
|
+ ", referenceIdentifier=" + referenceIdentifier + ", pollInterval=" + pollInterval + ", precision="
|
|
+ precision + ", rootDelay=" + rootDelay + ", rootDispersion=" + rootDispersion
|
|
+ ", referenceTimestamp=" + referenceTimestamp + ", uploadDelay=" + uploadDelay + ", downloadDelay="
|
|
+ downloadDelay + "]";
|
|
}
|
|
|
|
private long getRootDistance() {
|
|
return rootDelay / 2 + rootDispersion;
|
|
}
|
|
|
|
@Override
|
|
public int compareTo(PeerInfo o) {
|
|
return Long.compare(getRootDistance(), o.getRootDistance());
|
|
}
|
|
|
|
public Long128 getCurrentSelfDispersion128() {
|
|
Long128 vk = (clock.getCurrentTimeNTP128().subtract(referenceTimestamp)).multiply(systemFrequencyOffset)
|
|
.divide(Long128.ONE.shiftLeft(64));
|
|
Long128 vkl;
|
|
if (vk.signum() < 0) {
|
|
vkl = Long128.ZERO;
|
|
} else {
|
|
vkl = vk;
|
|
}
|
|
return vkl;
|
|
}
|
|
|
|
public long getCurrentRootDispersion() {
|
|
Long128 vkl = getCurrentSelfDispersion128().shiftRight(16 + 32);
|
|
long rez = rootDispersion + vkl.longValue();
|
|
if (rez > Integer.MAX_VALUE) {
|
|
rez = Integer.MAX_VALUE;
|
|
}
|
|
return rez;
|
|
}
|
|
|
|
public Long128 getAdj() {
|
|
return NTPTimestamps.ntp128BitToNanosInterval(uploadDelay.subtract(downloadDelay).shiftRight(1));
|
|
}
|
|
}
|
|
|
|
public void mergeAndApply() {
|
|
List<PeerInfo> peerInfo = mergeResponses();
|
|
selectAndApply(peerInfo);
|
|
}
|
|
|
|
private long avgAdj=0;
|
|
private Long128 inte=new Long128(0);
|
|
private void selectAndApply(List<PeerInfo> peerInfo) {
|
|
Collections.sort(peerInfo);
|
|
if(debug)
|
|
System.out.println(peerInfo);
|
|
if (!peerInfo.isEmpty()) {
|
|
PeerInfo pix = peerInfo.get(0);
|
|
currentClock = pix;
|
|
int i;
|
|
int m = Math.min(peerInfo.size(), 3);
|
|
Long128 bi = Long128.ZERO;
|
|
for (i = 0; i < m; i++) {
|
|
PeerInfo pi = peerInfo.get(i);
|
|
Long128 adjt = pi.getAdj();
|
|
bi = bi.add(adjt);
|
|
}
|
|
Long128 delta = bi.divide(Long128.valueOf(i));
|
|
avgAdj=(avgAdj*7+delta.abs().longValue())/8;
|
|
Long128 deltaabs=delta.abs();
|
|
if (deltaabs.compareTo(adjustThreshold0) > 0&&firstSync.compareAndSet(true, false)) {
|
|
Long128 adjustment=delta;
|
|
clock.adjustClock( adjustment);
|
|
if(debug)
|
|
System.out.println("adj:"+delta);
|
|
}else if (deltaabs.compareTo(adjustThreshold1) > 0) {
|
|
Long128 adjustment = Long128.valueOf(new BigDecimal(delta.toBigInteger()).multiply(BigDecimal.valueOf(0.5)).toBigInteger());
|
|
clock.adjustClock( adjustment);
|
|
if(debug)
|
|
System.out.println("adj:"+delta);
|
|
} else {
|
|
Long128 fadj = delta.divide(20).add(inte.divide(200));
|
|
clock.setFrequency(1000000000L+fadj.longValue());
|
|
inte=inte.add(delta);
|
|
if(inte.compareTo(Long128.valueOf(5000000))>0) {
|
|
inte=Long128.valueOf(5000000);
|
|
}else if(inte.compareTo(Long128.valueOf(-5000000))<0) {
|
|
inte=Long128.valueOf(-5000000);
|
|
}
|
|
if(debug)
|
|
System.out.println("inte:"+inte+" fadj:" + fadj);
|
|
|
|
}
|
|
if(debug)
|
|
System.out.println("delta:"+delta);
|
|
}
|
|
}
|
|
|
|
private List<PeerInfo> mergeResponses() {
|
|
List<PeerInfo> peerInfo = new ArrayList<PeerInfo>();
|
|
Set<Entry<MultiProtocolSocketAddress, List<NTPv4Packet>>> ens = recvmap.entrySet();
|
|
for (Iterator<Entry<MultiProtocolSocketAddress, List<NTPv4Packet>>> iterator = ens.iterator(); iterator
|
|
.hasNext();) {
|
|
Entry<MultiProtocolSocketAddress, List<NTPv4Packet>> entry = (Entry<MultiProtocolSocketAddress, List<NTPv4Packet>>) iterator
|
|
.next();
|
|
|
|
List<NTPv4Packet> newv = entry.getValue();
|
|
PeerInfo pi = null;
|
|
for (NTPv4Packet pack : newv) {
|
|
int li = pack.getLeapIndicator();
|
|
if (li == 3) {
|
|
continue;
|
|
}
|
|
int stratum = pack.getStratum();
|
|
if (stratum == 0) {
|
|
stratum = 16;
|
|
}
|
|
int mode = pack.getMode();
|
|
switch (mode) {
|
|
case NTPv4Packet.NTP_SERVER:
|
|
stratum += 1;
|
|
break;
|
|
case NTPv4Packet.NTP_SYMMETRIC_PASSIVE:
|
|
if (stratum < minStratum)
|
|
stratum += 1;
|
|
|
|
break;
|
|
|
|
}
|
|
if (stratum >= 16) {
|
|
continue;
|
|
}
|
|
if (pi == null) {
|
|
pi = new PeerInfo();
|
|
pi.address = entry.getKey();
|
|
}
|
|
pi.leapIndicator = pack.getLeapIndicator();
|
|
pi.stratum = stratum;
|
|
pi.referenceIdentifier = pack.getReferenceIdentifier();
|
|
pi.pollInterval = pack.getPollInterval();
|
|
pi.precision = pack.getPrecision();
|
|
pi.rootDelay = pack.getRootDelay();// (1/65536.0*1000000000)
|
|
pi.rootDispersion = pack.getRootDispersion();
|
|
pi.referenceTimestamp = pack.getReferenceTimestamp128();
|
|
Long128 uploadD = pack.getReceiveTimestamp128().subtract(pack.getOriginateTimestamp128());
|
|
Long128 downloadD = pack.getDestinationTimestamp128().subtract(pack.getTransmitTimestamp128());
|
|
Long128 rtt = uploadD.add(downloadD);
|
|
if (pi.uploadDelay == null) {
|
|
pi.uploadDelay = uploadD;
|
|
} else {
|
|
if (pi.uploadDelay.compareTo(uploadD) > 0) {
|
|
pi.uploadDelay = uploadD;
|
|
}
|
|
}
|
|
if (pi.downloadDelay == null) {
|
|
pi.downloadDelay = downloadD;
|
|
} else {
|
|
if (pi.downloadDelay.compareTo(downloadD) > 0) {
|
|
pi.downloadDelay = downloadD;
|
|
}
|
|
}
|
|
if (pi.rtt == null) {
|
|
pi.rtt = rtt;
|
|
} else {
|
|
if (pi.rtt.compareTo(rtt) > 0) {
|
|
pi.rtt = rtt;
|
|
}
|
|
}
|
|
}
|
|
if (pi != null) {
|
|
|
|
long ndl = pi.rootDelay + (pi.rtt).shiftRight(32 + 16).longValue();
|
|
// System.out.println(pi.rtt+" "+pi.rootDelay+" "+ndl);
|
|
pi.rootDelay = Math.min(ndl, Integer.MAX_VALUE);
|
|
long ndsp = pi.rootDispersion + Math.max(localPrecision.shiftRight(16 + 32).longValue(), 1);
|
|
pi.rootDispersion = Math.min(ndsp, Integer.MAX_VALUE);
|
|
peerInfo.add(pi);
|
|
}
|
|
}
|
|
return peerInfo;
|
|
}
|
|
|
|
@Override
|
|
public String toString() {
|
|
return "NTPContext [clock=" + clock + ", systemFrequencyOffset=" + systemFrequencyOffset + ", minStratum="
|
|
+ minStratum + ", leapIndicator=" + currentClock.leapIndicator + ", stratum=" + currentClock.stratum
|
|
+ ", referenceIdentifier=" + currentClock.referenceIdentifier + ", pollInterval="
|
|
+ currentClock.pollInterval + ", localPrecision=" + localPrecision + ", rootDelay="
|
|
+ currentClock.rootDelay + ", rootDispersion=" + currentClock.rootDispersion + ", referenceTimestamp="
|
|
+ NTPTimestamps.ntp128ToString(currentClock.referenceTimestamp) + "]";
|
|
}
|
|
|
|
public Long128 getSystemFrequencyOffset() {
|
|
return systemFrequencyOffset;
|
|
}
|
|
|
|
public void setSystemFrequencyOffset(Long128 systemFrequencyOffset) {
|
|
this.systemFrequencyOffset = systemFrequencyOffset;
|
|
}
|
|
|
|
public Long128 getLocalPrecision() {
|
|
return localPrecision;
|
|
}
|
|
|
|
public void setLocalPrecision(Long128 localPrecision) {
|
|
this.localPrecision = localPrecision;
|
|
}
|
|
|
|
public int getLeapIndicator() {
|
|
return currentClock.leapIndicator;
|
|
}
|
|
|
|
public int getReferenceIdentifier() {
|
|
return currentClock.referenceIdentifier;
|
|
}
|
|
|
|
public int getPollInterval() {
|
|
return currentClock.pollInterval;
|
|
}
|
|
|
|
public long getRootDelay() {
|
|
return currentClock.rootDelay;
|
|
}
|
|
|
|
public long getRootDispersion() {
|
|
return currentClock.rootDispersion;
|
|
}
|
|
|
|
public Long128 getReferenceTimestamp() {
|
|
return currentClock.referenceTimestamp;
|
|
}
|
|
|
|
public Long128 getReferenceTimestamp64() {
|
|
return NTPTimestamps.ntp128To64(currentClock.referenceTimestamp);
|
|
}
|
|
|
|
public Long128 getCurrentSelfDispersion128() {
|
|
|
|
return currentClock.getCurrentSelfDispersion128();
|
|
}
|
|
|
|
public long getCurrentRootDispersion() {
|
|
|
|
return currentClock.getCurrentRootDispersion();
|
|
}
|
|
|
|
private static Set<NTPv4Protocol> ios = Collections.synchronizedSet(new HashSet<>());
|
|
|
|
public void registerIO(NTPv4Protocol ntPv4Protocol) {
|
|
ios.add(ntPv4Protocol);
|
|
}
|
|
|
|
public void unregisterIO(NTPv4Protocol ntPv4Protocol) {
|
|
ios.remove(ntPv4Protocol);
|
|
}
|
|
|
|
public boolean isClosed() {
|
|
return closed;
|
|
}
|
|
|
|
@Override
|
|
public void close() throws IOException {
|
|
closed = true;
|
|
}
|
|
|
|
}
|