From 156d74bbe3925aaf894f7e1432d6e71afce5b39a Mon Sep 17 00:00:00 2001 From: KNEMC <2014539504@qq.com> Date: Wed, 7 Oct 2026 12:34:23 +0800 Subject: [PATCH] =?UTF-8?q?=E4=BF=AE=E5=A4=8D=E6=BB=91=E5=8A=A8=E7=AA=97?= =?UTF-8?q?=E5=8F=A3=E5=87=BA=E7=8E=B0=E8=B4=9F=E6=95=B0=E7=9A=84bug?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- src/klalb_en_US.properties | 17 +- src/klalb_zh_CN.properties | 3 +- .../congestion/BBRCongestionAlgorithm.java | 26 +-- .../congestion/CongestionAlgorithm.java | 10 +- .../congestion/DCTCP2CongestionAlgorithm.java | 126 --------------- .../congestion/DCTCPCongestionAlgorithm.java | 31 ++-- .../congestion/NOCongestionAlgorithm.java | 12 +- .../network/congestion/RTTEstimator.java | 150 ++++++++++++++++++ .../congestion/SendPacketSlidingWindow.java | 131 +++++++-------- .../congestion/Vegas2CongestionAlgorithm.java | 115 ++++++-------- .../cloud/network/klalb/KLALBController.java | 8 +- .../cloud/network/klalb/KLALBRemoteLink.java | 69 ++++---- .../network/klalb/ui/KLALBStateGUI3.java | 6 +- .../network/klalb/ui/LineMonitorGUI.java | 39 +++-- .../cloud/network/kltp/KLTPOutputStream.java | 32 ++-- .../monitor/ArrayListTimestampMonitor.java | 2 +- ...peedAndTrafficAndDelayMonitorDataImpl.java | 12 -- .../SpeedAndTrafficMonitorDataImpl.java | 2 +- src/org/kne/cloud/network/perf/Kperf.java | 22 +-- .../kne/cloud/network/srv6/SRv6Router.java | 18 +-- 20 files changed, 415 insertions(+), 416 deletions(-) delete mode 100644 src/org/kne/cloud/network/congestion/DCTCP2CongestionAlgorithm.java create mode 100644 src/org/kne/cloud/network/congestion/RTTEstimator.java diff --git a/src/klalb_en_US.properties b/src/klalb_en_US.properties index 029dd46..6d5082a 100644 --- a/src/klalb_en_US.properties +++ b/src/klalb_en_US.properties @@ -4,9 +4,9 @@ language=Language basicsettings=Basic settings ipv6addr=IPv6 address devicename=Device name -devicedescription=Device description +devicedescription=Device description dnsserver=DNS server -extraroutes=Extra routes +extraroutes=Extra routes asnumber=AS number tcplistening=TCP listening udplistening=UDP listening @@ -111,14 +111,15 @@ performancesettings=Performance settings performancestrategy=Performance strategy singlecore=Single-Core - Cache Affinity First (Best energy/performance ratio, for low-power & cloud) multifill=Multi-Core - Fill Cores Sequentially (Recommended for general-purpose physical servers) -multiscatter=Multi-Core - Spread Load Evenly (Optimized for multi-socket NUMA architectures) -webapisettings=Web API settings -enablewebapi=Enable Web API -weblistenaddr=Web API listen address +multiscatter=Multi-Core - Spread Load Evenly (Optimized for multi-socket NUMA architectures) +webapisettings=Web API settings +enablewebapi=Enable Web API +weblistenaddr=Web API listen address invaildweblistenaddr=Invalid Web API listen address -enabletun=Enable TUN adapter +enabletun=Enable TUN adapter congestionmonitor=Congestion control monitor rttbaseline=RTT baseline rtt=RTT rttmax=RTT max -rttmin=RTT min \ No newline at end of file +rttmin=RTT min +rto=RTO \ No newline at end of file diff --git a/src/klalb_zh_CN.properties b/src/klalb_zh_CN.properties index 2b815e1..eb0093f 100644 --- a/src/klalb_zh_CN.properties +++ b/src/klalb_zh_CN.properties @@ -121,4 +121,5 @@ congestionmonitor=拥塞控制监视器 rttbaseline=往返延迟基线 rtt=往返延迟 rttmax=往返延迟上限 -rttmin=往返延迟下限 \ No newline at end of file +rttmin=往返延迟下限 +rto=重传超时 \ No newline at end of file diff --git a/src/org/kne/cloud/network/congestion/BBRCongestionAlgorithm.java b/src/org/kne/cloud/network/congestion/BBRCongestionAlgorithm.java index 5a95437..78a42fe 100644 --- a/src/org/kne/cloud/network/congestion/BBRCongestionAlgorithm.java +++ b/src/org/kne/cloud/network/congestion/BBRCongestionAlgorithm.java @@ -11,14 +11,13 @@ public class BBRCongestionAlgorithm implements CongestionAlgorithm ,DetnetConges private Consumer windowControlConsumer; - private long MIN_RTTVAR = 5000000L; private volatile long RTTMin = 1000000000L; - private volatile long RTTVar = 1000000000L; - private volatile long RTTAvg = 1000000000L; private volatile long RTTAvg2 = 1000000000L; private volatile long RTTTotal = 0; private volatile long RTTCount = 0; - private volatile long RTO = 1000000000L; + + + private RTTEstimator rttEstimator=new RTTEstimator(); private volatile long window=MIN_WINDOW; @@ -54,10 +53,7 @@ public class BBRCongestionAlgorithm implements CongestionAlgorithm ,DetnetConges } else { RTTMin = (RTTMin * 99999 + latencyns) / 100000; } - RTTVar = (RTTVar * 3 + Math.abs(RTTAvg - latencyns)) / 4; - RTTAvg = (RTTAvg * 7 + latencyns) / 8; - - RTO = RTTAvg + Math.max(MIN_RTTVAR, RTTVar * 4);// RTTVar*4 + rttEstimator.putRTT(latencyns); RTTTotal += latencyns; RTTCount++; @@ -93,13 +89,13 @@ public class BBRCongestionAlgorithm implements CongestionAlgorithm ,DetnetConges } @Override - public void putLoss(long packetSize, long lossns, int losscounter) { + public void putLoss(long packetSize, long lossns) { } @Override public long getRTO() { - return RTO; + return rttEstimator.getRTO(); } @Override @@ -136,6 +132,16 @@ public class BBRCongestionAlgorithm implements CongestionAlgorithm ,DetnetConges return "BBR"; } + @Override + public long getBaseRTT() { + return rttEstimator.getBaseRTT(); + } + + @Override + public long getCurrentRTT() { + return rttEstimator.getCurrentRTT(); + } + @Override public double getUpperDelayBound() { return MAX_GAIN; diff --git a/src/org/kne/cloud/network/congestion/CongestionAlgorithm.java b/src/org/kne/cloud/network/congestion/CongestionAlgorithm.java index ec41593..ade6e45 100644 --- a/src/org/kne/cloud/network/congestion/CongestionAlgorithm.java +++ b/src/org/kne/cloud/network/congestion/CongestionAlgorithm.java @@ -6,13 +6,13 @@ public interface CongestionAlgorithm { public void reset(); public void setWindowControlConsumer(ConsumerwindowControlConsumer); public void putAck(long packetSize,long RTTns,boolean ecn); - default public void putAck(long packetSize,long RTTns,long OWDup,boolean ecn) { - putAck(packetSize,RTTns,ecn); - } - public void putLoss(long packetSize,long lossns,int losscounter); + public void putLoss(long packetSize,long lossns); public void setCurrentWindowUsed(long maxwindow); public void setCurrentBandwidth(long bandwidth); public long getRTO(); public String getName(); - + + public long getBaseRTT(); + + public long getCurrentRTT(); } diff --git a/src/org/kne/cloud/network/congestion/DCTCP2CongestionAlgorithm.java b/src/org/kne/cloud/network/congestion/DCTCP2CongestionAlgorithm.java deleted file mode 100644 index 611b9e5..0000000 --- a/src/org/kne/cloud/network/congestion/DCTCP2CongestionAlgorithm.java +++ /dev/null @@ -1,126 +0,0 @@ -package org.kne.cloud.network.congestion; - -import java.util.function.Consumer; - -import org.jctools.counters.FixedSizeStripedLongCounter; -import org.kne.cloud.network.klalb.KLALBUtils; - -public class DCTCP2CongestionAlgorithm implements CongestionAlgorithm { - - private static final long BURST_TIME = 1000000L; - private static final long BIAS = 2000000L; - private static long MIN_WINDOW = 8192; - - private Consumer windowControlConsumer; - - private long MIN_RTTVAR = 100000000L; - private volatile long RTTVar = 1000000000L; - private volatile long RTTAvg = 1000000000L; - private double ECNAvg2 = 0L; - private double alpha=0; - private volatile FixedSizeStripedLongCounter ECNSize = KLALBUtils.createCounter(); - private volatile FixedSizeStripedLongCounter TotalSize = KLALBUtils.createCounter(); - private volatile long RTO = 500000000L; - - - private volatile long window=MIN_WINDOW; - private volatile long maxwindow = 99999999999L; - private volatile long congressWindowSize=MIN_WINDOW; - - private long RTTstartTime = System.nanoTime(); - - @Override - public void reset() { - window = MIN_WINDOW; - if (windowControlConsumer != null) { - windowControlConsumer.accept(window); - } - } - - @Override - public void setWindowControlConsumer(Consumer windowControlConsumer) { - this.windowControlConsumer = windowControlConsumer; - } - - - @Override - public synchronized void putAck(long packetSize, long latencyns, boolean ecn) { - - - RTTVar = (RTTVar * 3 + Math.abs(RTTAvg - latencyns)) / 4; - RTTAvg = (RTTAvg * 7 + latencyns) / 8; - - RTO = RTTAvg + Math.max(MIN_RTTVAR, RTTVar * 4);// RTTVar*4 - - if(ecn) { - ECNSize.inc(packetSize); - } - TotalSize.inc(packetSize); - - - long curr=System.nanoTime(); - if(curr-RTTstartTime>(RTTAvg+BIAS)) { - RTTstartTime=curr; - - long totalSize=TotalSize.getAndReset(); - long ecnSize=ECNSize.getAndReset(); - if(totalSize!=0) { - ECNAvg2=ecnSize/(double)totalSize; - alpha=(alpha*7+ECNAvg2)/8; - } - - if(ECNAvg2>0.5) { - congressWindowSize-=200; - }else if(ECNAvg2>0.01){ - congressWindowSize+=200; - }else { - congressWindowSize+=4000; - } - //System.out.println(ECNAvg2); - //System.out.println(speed+" "+window); - - updateWindowSize(); - - } - - - } - - private void updateWindowSize() { - if(congressWindowSizemaxwindow) { - congressWindowSize=maxwindow; - } - window=Math.max( congressWindowSize,MIN_WINDOW); - if (windowControlConsumer != null) { - windowControlConsumer.accept(window); - } - } - - @Override - public void putLoss(long packetSize, long lossns, int losscounter) { - - } - - @Override - public long getRTO() { - return RTO; - } - - @Override - public void setCurrentWindowUsed(long used) { - this.maxwindow=(long) (used*1.5)+MIN_WINDOW; - } - - @Override - public void setCurrentBandwidth(long bandwidth) { - - } - - @Override - public String getName() { - return "DCTCP2"; - } - -} diff --git a/src/org/kne/cloud/network/congestion/DCTCPCongestionAlgorithm.java b/src/org/kne/cloud/network/congestion/DCTCPCongestionAlgorithm.java index b7ee01e..bb46c7d 100644 --- a/src/org/kne/cloud/network/congestion/DCTCPCongestionAlgorithm.java +++ b/src/org/kne/cloud/network/congestion/DCTCPCongestionAlgorithm.java @@ -13,16 +13,13 @@ public class DCTCPCongestionAlgorithm implements CongestionAlgorithm { private Consumer windowControlConsumer; - private AtomicBoolean firstUpdate = new AtomicBoolean(true); - private long MIN_RTTVAR = 100000000L; private volatile long RTTMin = 1000000000L; - private volatile long RTTVar = 1000000000L; - private volatile long RTTAvg = 1000000000L; private double ECNAvg2 = 0L; private double alpha=0; private volatile AtomicLong ECNSize = new AtomicLong(0); private volatile AtomicLong TotalSize = new AtomicLong(0); - private volatile long RTO = 1000000000L; + + private RTTEstimator rttEstimator=new RTTEstimator(100000000L); private volatile long window=MIN_WINDOW; @@ -54,14 +51,7 @@ public class DCTCPCongestionAlgorithm implements CongestionAlgorithm { } else { RTTMin = (RTTMin * 99999 + latencyns) / 100000; } - if (firstUpdate.compareAndSet(true, false)) { - RTTAvg = latencyns; - RTTVar = latencyns / 2; - } else { - RTTVar = (RTTVar * 3 + Math.abs(RTTAvg - latencyns)) / 4; - RTTAvg = (RTTAvg * 7 + latencyns) / 8; - } - RTO = RTTAvg + Math.max(MIN_RTTVAR, RTTVar * 4);// RTTVar*4 + rttEstimator.putRTT(latencyns); if(ecn) { ECNSize.addAndGet(packetSize); @@ -109,13 +99,13 @@ public class DCTCPCongestionAlgorithm implements CongestionAlgorithm { } @Override - public void putLoss(long packetSize, long lossns, int losscounter) { + public void putLoss(long packetSize, long lossns) { } @Override public long getRTO() { - return RTO; + return rttEstimator.getRTO(); } @Override @@ -133,6 +123,15 @@ public class DCTCPCongestionAlgorithm implements CongestionAlgorithm { return "DCTCP"; } - + @Override + public long getBaseRTT() { + return rttEstimator.getBaseRTT(); + } + + @Override + public long getCurrentRTT() { + return rttEstimator.getCurrentRTT(); + } + } diff --git a/src/org/kne/cloud/network/congestion/NOCongestionAlgorithm.java b/src/org/kne/cloud/network/congestion/NOCongestionAlgorithm.java index a736748..002d81f 100644 --- a/src/org/kne/cloud/network/congestion/NOCongestionAlgorithm.java +++ b/src/org/kne/cloud/network/congestion/NOCongestionAlgorithm.java @@ -31,7 +31,7 @@ public class NOCongestionAlgorithm implements CongestionAlgorithm{ } @Override - public void putLoss(long packetSize, long lossns, int losscounter) { + public void putLoss(long packetSize, long lossns) { // TODO 自动生成的方法存根 } @@ -63,4 +63,14 @@ public class NOCongestionAlgorithm implements CongestionAlgorithm{ return "NO"; } + @Override + public long getBaseRTT() { + return 0; + } + + @Override + public long getCurrentRTT() { + return 0; + } + } diff --git a/src/org/kne/cloud/network/congestion/RTTEstimator.java b/src/org/kne/cloud/network/congestion/RTTEstimator.java new file mode 100644 index 0000000..da6f068 --- /dev/null +++ b/src/org/kne/cloud/network/congestion/RTTEstimator.java @@ -0,0 +1,150 @@ +package org.kne.cloud.network.congestion; + +import org.kne.cloud.clock.HighAccuracyClock; +import org.kne.cloud.network.klalb.KLALBUtils; +import org.kne.cloud.network.monitor.ArrayListTimestampMonitor; +import org.kne.cloud.network.monitor.DelaySampler; +import org.kne.cloud.network.monitor.TimestampMonitor; + +import java.util.UUID; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicLong; + +/** + * 线程安全的 RTT 平滑与 RTO 计算工具。 + * 基于 RFC 6298 的 Jacobson/Karels 算法。 + * + * 公式: + * RTTAvg = (RTTAvg * 7 + RTTns) / 8 + * RTTVar = (RTTVar * 3 + |RTTAvg - RTTns|) / 4 + * RTO = RTTAvg + max(MIN_RTTVAR, RTTVar * 4) + */ +public class RTTEstimator { + + /** RTTVar 下限（纳秒），防止 RTO 过小。默认 20ms。 */ + private static final long DEFAULT_MIN_RTTVAR_NS = 20_000_000L; + + private final AtomicLong curr; + private final AtomicLong rttAvg; + private final AtomicLong rttVar; + private final AtomicLong rto; + private final AtomicLong basertt; + private final AtomicBoolean firstUpdate = new AtomicBoolean(true); // 首次标记 + private final long minRttVarNs; + + + private long timewindowmax=500000000L; + + long stime=System.nanoTime(); + + public RTTEstimator() { + this(DEFAULT_MIN_RTTVAR_NS); + } + + public RTTEstimator(long minRttVarNs) { + this.rttAvg = new AtomicLong(1000000000L); + this.rttVar = new AtomicLong(1000000000L); + this.curr = new AtomicLong(1000000000L); + this.rto = new AtomicLong(1000000000L); + this.basertt = new AtomicLong(1000000000L); + this.minRttVarNs = minRttVarNs; + } + private final static long TIME_WINDOW=30000000000L; + private final TimestampMonitor rttMonitor=new ArrayListTimestampMonitor(new HighAccuracyClock(),"RTTMonitor", 100, TIME_WINDOW); + private final DelaySampler dsp=new DelaySampler(); + + public void update() { + dsp.update(); + + if(dsp.getSnapshotPacketCount()>0&&dsp.getSnapshotTotalDelay()>0) { + rttMonitor. recordPacket(KLALBUtils.createGlobalUUID(),0, 0, dsp.getAvgDelayNanos(),dsp.getMinDelayNanos()); + } + + long time=System.nanoTime(); + if(time-stime>timewindowmax) { + basertt.lazySet( rttMonitor.calculateMinDelay(TIME_WINDOW)); + stime=time; + } + } + /** + * 更新 RTT 估算值（线程安全）。 + * + * @param rttNs 本次 RTT 测量值（纳秒） + */ + /** + * 更新 RTT 估算值（线程安全，无锁）。 + */ + public void putRTT(long rttNs) { + if (rttNs < 0) { + throw new IllegalArgumentException("RTT 不能为负数: " + rttNs); + } + dsp.recordDelay(rttNs); + curr.lazySet(rttNs); + // 首次更新：CAS 成功则直接初始化，无锁 + if (firstUpdate.compareAndSet(true, false)) { + rttAvg.set(rttNs); + rttVar.set(rttNs / 2); + return; + } + + // 后续更新：CAS 循环保证原子性 + + long oldAvg = rttAvg.get(); + long newAvg = (oldAvg * 7 + rttNs) / 8; + + long delta = Math.abs(oldAvg - rttNs); + rttAvg.lazySet(newAvg); + long oldVar = rttVar.get(); + long newVar = (oldVar * 3 + delta) / 4; + rttVar.lazySet(newVar); + + rto .lazySet( newAvg + Math.max(minRttVarNs, newVar * 4)); + } + + + /** + * 获取平滑 RTT 平均值（纳秒）。 + */ + public long getRTTAvg() { + return rttAvg.getPlain(); + } + + /** + * 获取 RTT 偏差（纳秒）。 + */ + public long getRTTVar() { + return rttVar.getPlain(); + } + + /** + * 获取 RTO（重传超时时间，纳秒）。 + * RTO = RTTAvg + max(MIN_RTTVAR, RTTVar * 4) + */ + public long getRTO() { + return rto.getPlain(); + } + + /** + * 重置估算器（用于连接重建等场景）。 + */ + public void reset() { + rttAvg.set(0); + rttVar.set(0); + } + + @Override + public String toString() { + return String.format("RttEstimator{avg=%.3fms, var=%.3fms, rto=%.3fms}", + rttAvg.get() / 1_000_000.0, + rttVar.get() / 1_000_000.0, + getRTO() / 1_000_000.0); + } + + public long getCurrentRTT() { + return curr.get(); + } + + public long getBaseRTT() { + return basertt.get(); + } +} \ No newline at end of file diff --git a/src/org/kne/cloud/network/congestion/SendPacketSlidingWindow.java b/src/org/kne/cloud/network/congestion/SendPacketSlidingWindow.java index 5260374..968d4f6 100644 --- a/src/org/kne/cloud/network/congestion/SendPacketSlidingWindow.java +++ b/src/org/kne/cloud/network/congestion/SendPacketSlidingWindow.java @@ -5,7 +5,10 @@ import java.net.SocketException; import java.util.Collection; import java.util.Iterator; import java.util.Map; +import java.util.Set; import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.ScheduledFuture; +import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicLong; import java.util.concurrent.locks.Condition; import java.util.concurrent.locks.Lock; @@ -13,13 +16,15 @@ import java.util.function.Consumer; import org.kne.cloud.network.NetworkPacket; import org.kne.cloud.network.ThreadTool; +import org.kne.cloud.network.klalb.KLALBScheduler; import org.kne.cloud.network.klalb.SendItem; import org.kne.concurrent.ThreadParker; public class SendPacketSlidingWindow implements Closeable, AutoCloseable { + private final ScheduledFuture checkerTask; private ThreadParker tp=new ThreadParker(); - - private Map> sendMap = new ConcurrentHashMap<>(2048); + + private volatile Map> sendMap = new ConcurrentHashMap<>(2048); private AtomicLong sendWindowUsed=new AtomicLong(0); //private LongAdder sendWindowUsed = new LongAdder(); @@ -51,13 +56,12 @@ public class SendPacketSlidingWindow implements Clos } public SendPacketSlidingWindow(CongestionAlgorithm algorithm, long windowSize, int headerCalibrate, - boolean removeAtResend) { + boolean removeAtResend) { this.headerCalibrate = headerCalibrate; this.removeAtResend = removeAtResend; this.algorithm = algorithm; this.windowSize=windowSize; - Thread t = ThreadTool.makeVThread("重路由计时器线程", checker); - t.start(); + this.checkerTask=KLALBScheduler.getScheduler().scheduleWithFixedDelay(checker,1,1, TimeUnit.MILLISECONDS); } public long getWindowSize() { @@ -106,7 +110,7 @@ public class SendPacketSlidingWindow implements Clos boolean congress = sendWindowUsed.get() > windowSize ; return congress; } - + public void waitForAvaliable() throws SocketException { while(isCongestion()) { if(closed) { @@ -119,28 +123,27 @@ public class SendPacketSlidingWindow implements Clos waitForAvaliable(); put(sequence,packet); } - + public void put(K sequence, V packet) { SendItem newitem = new SendItem(packet,headerCalibrate); SendItem old = sendMap.put(sequence, newitem); long dx=0; - if (old != null) { + if (old != null) { dx=-old.getPacketLength() ; } long newWindow =sendWindowUsed.addAndGet(newitem.getPacketLength() +dx); algorithm.setCurrentWindowUsed(newWindow); } public V ack(K sequence) { - return ack(sequence,false); + return ack(sequence,0,false); } - - public V ack(K sequence,boolean ecn) { + + public V ack(K sequence,long RTTns,boolean ecn) { SendItem ipv6; if ((ipv6 = sendMap.remove(sequence)) != null) { long newWindow =sendWindowUsed.addAndGet(-ipv6.getPacketLength() ); algorithm.setCurrentWindowUsed(newWindow); - long RTTC = System.nanoTime() - ipv6.getSendtime(); - algorithm.putAck(ipv6.getPacketLength(), RTTC, ecn); + algorithm.putAck(ipv6.getPacketLength(), RTTns, ecn); tp.unpark(); // System.out.println("remove:"+aseq.getSequence()+" "+sendMap.size()); Lock lockx = lock; @@ -162,73 +165,59 @@ public class SendPacketSlidingWindow implements Clos } } - public V ack(K sequence,boolean ecn,long OWDdown) { - SendItem ipv6; - if ((ipv6 = sendMap.remove(sequence)) != null) { - long newWindow =sendWindowUsed.addAndGet(-ipv6.getPacketLength() ); - algorithm.setCurrentWindowUsed(newWindow); - long RTTC = System.nanoTime() - ipv6.getSendtime(); - algorithm.putAck(ipv6.getPacketLength(), RTTC,OWDdown, ecn); - tp.unpark(); - // System.out.println("remove:"+aseq.getSequence()+" "+sendMap.size()); - Lock lockx = lock; - if (lockx != null) { - lockx.lock(); - try { - Condition cds = condition; - if (cds != null) { - cds.signalAll(); - } - } finally { - lockx.unlock(); - } - } - return ipv6.getPacket(); - } else { - // System.out.println("miss:"+aseq.getSequence()+" "+sendMap.size()); - return null; - } - } - + + private class ResendChecker implements Runnable { @Override public void run() { - while (!closed) { - - Collection> vals = sendMap.values(); - for (Iterator> iterator = vals.iterator(); iterator.hasNext();) { - SendItem sitm = (SendItem) iterator.next(); - long current = System.nanoTime(); - if (current - sitm.getSendtime() > algorithm.getRTO()) { - if (removeAtResend) { - iterator.remove(); - long newWindow = sendWindowUsed.addAndGet((int) -sitm.getPacketLength() ); - algorithm.setCurrentWindowUsed(newWindow); - tp.unpark(); - } else { - sitm.setSendtime(current); + try { + if (closed) { + checkerTask.cancel(false); + // 清理资源 + if (closingConsumer != null) { + for (SendItem item : sendMap.values()) { + closingConsumer.accept(item.getPacket()); } - V pktr = sitm.getPacket(); - if (resendConsumer != null) - resendConsumer.accept(pktr); } - } + sendMap.clear(); + sendWindowUsed.set(0); + + } else { + + Set>> vals = sendMap.entrySet(); + for (Iterator>> iterator = vals.iterator(); iterator.hasNext(); ) { + Map.Entry> entry = (Map.Entry>) iterator.next(); + SendItem sitm=entry.getValue(); + long current = System.nanoTime(); + long ftime=current - sitm.getSendtime(); + if (ftime> algorithm.getRTO()) { + + V pktr = sitm.getPacket(); + + if (removeAtResend) { + SendItem sitm2=sendMap.remove(entry.getKey()); + if(sitm2!=null) { + if (resendConsumer != null) + resendConsumer.accept(pktr); + algorithm.putLoss(sitm2.getPacketLength(),ftime); + long newWindow = sendWindowUsed.addAndGet((int) -sitm2.getPacketLength()); + algorithm.setCurrentWindowUsed(newWindow); + tp.unpark(); + } + } else { + sitm.setSendtime(current); + if (resendConsumer != null) + resendConsumer.accept(pktr); + } + } + } + - try { - Thread.sleep(2); - } catch (InterruptedException e) { - e.printStackTrace(); } + } catch (Exception e) { + e.printStackTrace(); } - // 清理资源 - if (closingConsumer != null) { - for (SendItem item : sendMap.values()) { - closingConsumer.accept(item.getPacket()); - } - } - sendMap.clear(); - sendWindowUsed.set(0); } } diff --git a/src/org/kne/cloud/network/congestion/Vegas2CongestionAlgorithm.java b/src/org/kne/cloud/network/congestion/Vegas2CongestionAlgorithm.java index 87641e8..ca2caed 100644 --- a/src/org/kne/cloud/network/congestion/Vegas2CongestionAlgorithm.java +++ b/src/org/kne/cloud/network/congestion/Vegas2CongestionAlgorithm.java @@ -3,6 +3,7 @@ package org.kne.cloud.network.congestion; import java.util.function.Consumer; import org.kne.cloud.network.monitor.DelaySampler; +import org.kne.concurrent.SpinLock; public class Vegas2CongestionAlgorithm implements CongestionAlgorithm,DetnetCongestionAlgorithm { @@ -13,25 +14,26 @@ public class Vegas2CongestionAlgorithm implements CongestionAlgorithm,DetnetCong private static final long BURST_TIME = 2000000L; private static final long BIAS = 1000000L; - private static long MIN_WINDOW = 16384; + private static long MIN_WINDOW = 8192; private Consumer windowControlConsumer; - private long MIN_RTTVAR = 30000000L; private volatile long RTTCurr = 1000000000L; // RTT private volatile long RTTMin = 1000000000L; // BaseRTT - private volatile long RTTVar = 1000000000L; - private volatile long RTTAvg = 1000000000L; private volatile long RTTAvg2 = 1000000000L; // 当前平滑RTT private volatile long RTTTotal = 0; private volatile long RTTCount = 0; - private volatile long RTO = 1000000000L; + + private RTTEstimator rttEstimator=new RTTEstimator(); private volatile long window = MIN_WINDOW; private volatile long window2 = MIN_WINDOW; private volatile long maxwindow = 99999999999L; + + private final SpinLock slock=new SpinLock(); + public Vegas2CongestionAlgorithm() { reset(); } @@ -56,7 +58,6 @@ public class Vegas2CongestionAlgorithm implements CongestionAlgorithm,DetnetCong - private DelaySampler dsp=new DelaySampler(); @Override public void putAck(long packetSize, long RTTns, boolean ecn) { @@ -72,66 +73,54 @@ public class Vegas2CongestionAlgorithm implements CongestionAlgorithm,DetnetCong windowControlConsumer.accept(window2); } } - - dsp.recordDelay(RTTns); + // 更新平滑RTT估计 - - RTTVar = (RTTVar * 3 + Math.abs(RTTAvg - RTTns)) / 4; - RTTAvg = (RTTAvg * 7 + RTTns) / 8; - - - RTO = RTTAvg + Math.max(MIN_RTTVAR, RTTVar * 4); + rttEstimator.putRTT(RTTns); RTTTotal += RTTns; RTTCount++; - - long curr = System.nanoTime(); - // 每个RTT周期调整一次窗口（使用当前估计的RTT） - if (curr - RTTstartTime > Math.max(BIAS, 2*RTTMin)) { - dsp.update(); - long rmin=dsp.getMinDelayNanos(); - if(rmin<=RTTMin) { - RTTMin=rmin; - }else { - RTTMin=Math.min(rmin, RTTMin+1000); - } - - - long RTTCountx=RTTCount; - if (RTTCountx > 0) { - long currentRTT = RTTTotal / RTTCountx; - //RTTAvg2 = (RTTAvg2 * 3 + currentRTT) / 4; // 平滑当前RTT - RTTAvg2=currentRTT; - RTTTotal = 0; - RTTCount = 0; + if(slock.tryLock())try { + long curr = System.nanoTime(); + // 每个RTT周期调整一次窗口（使用当前估计的RTT） + if (curr - RTTstartTime > Math.max(BIAS, 2 * RTTMin)) { + rttEstimator.update(); + RTTMin = rttEstimator.getBaseRTT(); + + + long RTTCountx = RTTCount; + if (RTTCountx > 0) { + long currentRTT = RTTTotal / RTTCountx; + //RTTAvg2 = (RTTAvg2 * 3 + currentRTT) / 4; // 平滑当前RTT + RTTAvg2 = currentRTT; + RTTTotal = 0; + RTTCount = 0; + } + + // ================== Vegas2.0核心逻辑 ================== + // 1. 计算时延膨胀比 + double delayRatio = (double) RTTAvg2 / (double) Math.max(BIAS, RTTMin); + //System.out.println(delayRatio+" "+RTTAvg2/1000000.0+" "+RTTMin/1000000.0); + // 2. 基于比值的窗口调整（取代原来的基于差值的调整） + if (delayRatio > delayUpperBound) { + // 时延过高：减小窗口，减少幅度与超标程度成正比 + window2 = window2 - 4096; + } else if (delayRatio < delayLowerBound) { + // 时延过低：增大窗口，增加幅度与低于目标程度成正比 + window2 = window2 + 4096; + } else { + //window2=window2+4096 + } + // =================================================== + + + updateWindowSize(); + RTTstartTime = curr; + + } - - // ================== Vegas2.0核心逻辑 ================== - // 1. 计算时延膨胀比 - double delayRatio = (double) RTTAvg2 / (double) Math.max(BIAS, RTTMin); - //System.out.println(delayRatio+" "+RTTAvg2/1000000.0+" "+RTTMin/1000000.0); - // 2. 基于比值的窗口调整（取代原来的基于差值的调整） - if (delayRatio > delayUpperBound) { - // 时延过高：减小窗口，减少幅度与超标程度成正比 - window2 =window2-4096; - } else if (delayRatio < delayLowerBound) { - // 时延过低：增大窗口，增加幅度与低于目标程度成正比 - window2 =window2+4096; - } else { - //window2=window2+4096 - } - // =================================================== - - - - - updateWindowSize(); - RTTstartTime = curr; - - - - + }finally{ + slock.unlock(); } } @@ -153,12 +142,12 @@ public class Vegas2CongestionAlgorithm implements CongestionAlgorithm,DetnetCong } @Override - public void putLoss(long packetSize, long lossns, int losscounter) { + public void putLoss(long packetSize, long lossns) { } @Override public long getRTO() { - return RTO; + return rttEstimator.getRTO(); } @Override @@ -177,7 +166,7 @@ public class Vegas2CongestionAlgorithm implements CongestionAlgorithm,DetnetCong } public long getBaseRTT() { - return RTTMin; + return rttEstimator.getBaseRTT(); } public long getAvgRTT() { return RTTAvg2; diff --git a/src/org/kne/cloud/network/klalb/KLALBController.java b/src/org/kne/cloud/network/klalb/KLALBController.java index 358cbab..de289e2 100644 --- a/src/org/kne/cloud/network/klalb/KLALBController.java +++ b/src/org/kne/cloud/network/klalb/KLALBController.java @@ -61,7 +61,7 @@ public class KLALBController { public HighAccuracyClock getClock() { return clock; } - private final long TIME_WINDOW=50000000000L; + private final long TIME_WINDOW=30000000000L; private SpeedAndTrafficMonitorDataImpl linkMonitor = new SpeedAndTrafficMonitorDataImpl( new HashMapTimestampMonitor(HighAccuracyClock.SYSTEM_CLOCK, "up", 100,TIME_WINDOW ), new HashMapTimestampMonitor(HighAccuracyClock.SYSTEM_CLOCK, "down", 100, TIME_WINDOW)); @@ -733,9 +733,11 @@ public class KLALBController { } protected void initKLALB(boolean enableVirtualAdapter, IPv6AddressGroup selfg) { - KLALBScheduler.getScheduler().scheduleAtFixedRate( ()->{ + KLALBScheduler.getScheduler().scheduleWithFixedDelay( ()->{ linkMonitor.update(); - datatMonitor.update(); + },1,1,TimeUnit.MILLISECONDS); + KLALBScheduler.getScheduler().scheduleWithFixedDelay( ()->{ + datatMonitor.update(); },1,1,TimeUnit.MILLISECONDS); loadSRv6ProtocolStack(selfg, enableVirtualAdapter); loadController(); diff --git a/src/org/kne/cloud/network/klalb/KLALBRemoteLink.java b/src/org/kne/cloud/network/klalb/KLALBRemoteLink.java index 4c9c9e3..fc37180 100644 --- a/src/org/kne/cloud/network/klalb/KLALBRemoteLink.java +++ b/src/org/kne/cloud/network/klalb/KLALBRemoteLink.java @@ -33,17 +33,8 @@ import org.kne.cloud.network.congestion.CongestionAlgorithm; import org.kne.cloud.network.congestion.CongestionAlgorithms; import org.kne.cloud.network.congestion.DetnetCongestionAlgorithm; import org.kne.cloud.network.congestion.SendPacketSlidingWindow; -import org.kne.cloud.network.ipv6.ControlledIPv6NetworkLink; -import org.kne.cloud.network.ipv6.IPv6Address; -import org.kne.cloud.network.ipv6.IPv6HopByHopTLV; -import org.kne.cloud.network.ipv6.IPv6Packet; +import org.kne.cloud.network.ipv6.*; import org.kne.cloud.network.ipv6.IPv6Packet.IPv6HopByHopHeader; -import org.kne.cloud.network.ipv6.IPv6AddressGroup; -import org.kne.cloud.network.ipv6.KLALBOAMHopByHopTLV; -import org.kne.cloud.network.ipv6.KLALBPassportHopByHopTLV; -import org.kne.cloud.network.ipv6.Neighbor; -import org.kne.cloud.network.ipv6.PostcardEntry; -import org.kne.cloud.network.ipv6.RouteItem; import org.kne.cloud.network.monitor.CostSupplierFactory; import org.kne.cloud.network.monitor.DelayMonitorData; import org.kne.cloud.network.monitor.LinkStatus; @@ -287,12 +278,12 @@ public class KLALBRemoteLink extends AbstractControlledIPv6NetworkLink implement sendLock.lock(); try { kplink.flush(); - watchdog.feed(); }finally { sendLock.unlock(); } + watchdog.feed(); } - tlock.parkNanos(10000000L); + tlock.parkNanos(1000000L); } } catch (StreamCorruptedException sce) { @@ -754,7 +745,7 @@ public class KLALBRemoteLink extends AbstractControlledIPv6NetworkLink implement protected void startIO() { stateManager=new StateManager(); - monitorUpdater=KLALBScheduler.getScheduler().scheduleAtFixedRate(new MonitorUpdater(),1,1, TimeUnit.MILLISECONDS); + monitorUpdater=KLALBScheduler.getScheduler().scheduleWithFixedDelay(new MonitorUpdater(),1,1, TimeUnit.MILLISECONDS); ThreadTool.makeVDaemonThread("远程线路状态线程", stateManager).start(); } @@ -896,6 +887,8 @@ public class KLALBRemoteLink extends AbstractControlledIPv6NetworkLink implement sendWindow.put(val, pack); //} } + }else { + System.out.println(hop); } sendIPv6PacketToLink(pack, val); @@ -944,28 +937,6 @@ public class KLALBRemoteLink extends AbstractControlledIPv6NetworkLink implement return (!isClosed()) && (status.getState() == LinkStatus.UP) ; } - /*@Override - public List getRouteItems() { - List rlist = new ArrayList<>(); - - if(isUp()) { - - - for (Iterator iteratorx = getNeighborsInfo().iterator(); iteratorx.hasNext(); ) { - Neighbor addresses = (Neighbor) iteratorx.next(); - RouteItem ri = new RouteItem(new IPv6AddressGroup(addresses.getAddress().getAddress(), 128), - addresses.getAddress().getAddress(), this, RouteItem.SRv6_ENDXSID, 0, 128, CostSupplierFactory.expectedDelaySupplier((DelayMonitorData) addresses.getMonitor(), status, algorithm::getRTO), "D"); - rlist.add(ri); - - if (addresses.getLocator() != null) { - RouteItem ris = new RouteItem(addresses.getLocator(), addresses.getLocator().getAddress(), this, - RouteItem.SRv6_ENDSID, 13, 128, CostSupplierFactory.expectedDelaySupplier((DelayMonitorData) addresses.getMonitor(), status, algorithm::getRTO), "D"); - rlist.add(ris); - } - } - } - return rlist; - }*/ @Override public void setCongressCondition(Lock lock, Condition condition) { @@ -973,14 +944,32 @@ public class KLALBRemoteLink extends AbstractControlledIPv6NetworkLink implement sendWindow.setCongressCondition(lock, condition); } - public void onPostcardReceive(PostcardEntry postcard) { + public void onPostcardReceive(IPv6NetworkLink linkfrom,PostcardEntry postcard) { long deltaxnanos = (long) (postcard.getTimestampDeltaLong() * 1000000000.0 / (1L << 32)); - if (deltaxnanos > 0) { - Long128 psnd = NTPTimestamps.ntp128BitToNanosTimestamp( - NTPTimestamps.inferNtp64To128(postcard.getTimestampBase(), sysclk.getCurrentTimeNTP128())); + Long128 curr= sysclk.getCurrentTimeNanos(); + Long128 psnd = NTPTimestamps.ntp128BitToNanosTimestamp( + NTPTimestamps.inferNtp64To128(postcard.getTimestampBase(),NTPTimestamps.nanosToNtp128BitTimestamp(curr))); + long rtt=curr.subtract(psnd).longValue(); + if (deltaxnanos >= 0) { monitor.recordUploadDelay(deltaxnanos); } - sendWindow.ack(postcard.getPacketUUID(),false,deltaxnanos); + if(linkfrom instanceof KLALBRemoteLink ){ + long rtt2=deltaxnanos+((KLALBRemoteLink) linkfrom).getMonitor().getInDelayMin(); + + if(rtt2<0) { + if(rtt<0) { + rtt=0; + } + sendWindow.ack(postcard.getPacketUUID(), rtt, false); + }else { + sendWindow.ack(postcard.getPacketUUID(), rtt2, false); + } + }else { + if(rtt<0) { + rtt=0; + } + sendWindow.ack(postcard.getPacketUUID(), rtt, false); + } } diff --git a/src/org/kne/cloud/network/klalb/ui/KLALBStateGUI3.java b/src/org/kne/cloud/network/klalb/ui/KLALBStateGUI3.java index 028da0e..cb108fe 100644 --- a/src/org/kne/cloud/network/klalb/ui/KLALBStateGUI3.java +++ b/src/org/kne/cloud/network/klalb/ui/KLALBStateGUI3.java @@ -1319,7 +1319,7 @@ public class KLALBStateGUI3 extends XFrame { if (tsk != null) { tsk.cancel(false); } - tsk =KLALBScheduler.getScheduler().scheduleAtFixedRate( new Runnable() { + tsk =KLALBScheduler.getScheduler().scheduleWithFixedDelay( new Runnable() { @Override public void run() { if (isVisible()) { @@ -1383,7 +1383,7 @@ public class KLALBStateGUI3 extends XFrame { if (tsk2 != null) { tsk2.cancel(false); } - tsk2 =KLALBScheduler.getScheduler().scheduleAtFixedRate(new Runnable() { + tsk2 =KLALBScheduler.getScheduler().scheduleWithFixedDelay(new Runnable() { @Override public void run() { if (isVisible()) { @@ -1396,7 +1396,7 @@ public class KLALBStateGUI3 extends XFrame { if (tsk5 != null) { tsk5.cancel(false); } - tsk5 = KLALBScheduler.getScheduler().scheduleAtFixedRate(new Runnable() { + tsk5 = KLALBScheduler.getScheduler().scheduleWithFixedDelay(new Runnable() { @Override public void run() { if (isVisible()) { diff --git a/src/org/kne/cloud/network/klalb/ui/LineMonitorGUI.java b/src/org/kne/cloud/network/klalb/ui/LineMonitorGUI.java index a39360a..614185a 100644 --- a/src/org/kne/cloud/network/klalb/ui/LineMonitorGUI.java +++ b/src/org/kne/cloud/network/klalb/ui/LineMonitorGUI.java @@ -69,6 +69,7 @@ public class LineMonitorGUI extends XFrame { private TimeSeries rtt = new TimeSeries(UIEnv.getRsb().getString("rtt")); private TimeSeries rttmin=new TimeSeries(UIEnv.getRsb().getString("rttmin")); private TimeSeries rttmax=new TimeSeries(UIEnv.getRsb().getString("rttmax")); + private TimeSeries rto=new TimeSeries(UIEnv.getRsb().getString("rto")); private KLALBRemoteLink tr; private JFreeChart jfc; @@ -170,6 +171,7 @@ public class LineMonitorGUI extends XFrame { tbsc1.addSeries(rtt); tbsc1.addSeries(rttmax); tbsc1.addSeries(rttmin); + tbsc1.addSeries(rto); jbfcv = ChartFactory.createTimeSeriesChart(UIEnv.getRsb().getString("congestionmonitor"), UIEnv.getRsb().getString("time") + "(s)", UIEnv.getRsb().getString("delay") + "(ms)", tbsc1); jbfcv.getXYPlot().setBackgroundPaint(Color.BLACK); @@ -177,6 +179,7 @@ public class LineMonitorGUI extends XFrame { jbfcv.getXYPlot().getRenderer().setSeriesPaint(1, Color.YELLOW); jbfcv.getXYPlot().getRenderer().setSeriesPaint(2, Color.RED); jbfcv.getXYPlot().getRenderer().setSeriesPaint(3, Color.GREEN); + jbfcv.getXYPlot().getRenderer().setSeriesPaint(4, Color.ORANGE); changeFont(jbfcv); sbpdpv = new JLabel(); sbpdpv.setBorder(new LineBorder(Color.DARK_GRAY)); @@ -368,14 +371,18 @@ public class LineMonitorGUI extends XFrame { new Date(tr.getMonitor().getClock().getCurrentTimeMillis())); synchronized (jbfcv) { CongestionAlgorithm window=tr.getWindow().getAlgorithm(); - if(window !=null&&window instanceof DetnetCongestionAlgorithm) { - Vegas2CongestionAlgorithm deta=(Vegas2CongestionAlgorithm) window; - double baseRTT=deta.getBaseRTT(); + if(window !=null) { + double baseRTT = window.getBaseRTT(); rttbaseline.addOrUpdate(ms, baseRTT / 1000000.0); - rtt.addOrUpdate(ms, deta.getCurrentRTT() / 1000000.0); - rttmax.addOrUpdate(ms, baseRTT*deta.getUpperDelayBound() / 1000000.0); - rttmin.addOrUpdate(ms, baseRTT*deta.getLowerDelayBound() / 1000000.0); - + rtt.addOrUpdate(ms, window.getCurrentRTT() / 1000000.0); + rto.addOrUpdate(ms, window.getRTO() / 1000000.0); + if (window instanceof DetnetCongestionAlgorithm) { + + + Vegas2CongestionAlgorithm deta = (Vegas2CongestionAlgorithm) window; + rttmax.addOrUpdate(ms, baseRTT * deta.getUpperDelayBound() / 1000000.0); + rttmin.addOrUpdate(ms, baseRTT * deta.getLowerDelayBound() / 1000000.0); + } } } } @@ -385,14 +392,14 @@ public class LineMonitorGUI extends XFrame { if (b) { if (tsk3 == null) { - tsk3 = KLALBScheduler.getScheduler().scheduleAtFixedRate(new Runnable() { + tsk3 = KLALBScheduler.getScheduler().scheduleWithFixedDelay(new Runnable() { @Override public void run() { updateTraffic(); } }, 25, 25, TimeUnit.MILLISECONDS); - tsk4 = KLALBScheduler.getScheduler().scheduleAtFixedRate(new Runnable() { + tsk4 = KLALBScheduler.getScheduler().scheduleWithFixedDelay(new Runnable() { @Override public void run() { @@ -400,28 +407,28 @@ public class LineMonitorGUI extends XFrame { } }, 10, 10, TimeUnit.MILLISECONDS);; - tsk5 =KLALBScheduler.getScheduler().scheduleAtFixedRate( new Runnable() { + tsk5 =KLALBScheduler.getScheduler().scheduleWithFixedDelay( new Runnable() { @Override public void run() { recordDelayData(); } }, 10, 10, TimeUnit.MILLISECONDS); - tsk6 =KLALBScheduler.getScheduler().scheduleAtFixedRate( new Runnable() { + tsk6 =KLALBScheduler.getScheduler().scheduleWithFixedDelay( new Runnable() { @Override public void run() { recordBufferData(); } }, 10, 10, TimeUnit.MILLISECONDS); - tsk7 = KLALBScheduler.getScheduler().scheduleAtFixedRate( new Runnable() { + tsk7 = KLALBScheduler.getScheduler().scheduleWithFixedDelay( new Runnable() { @Override public void run() { recordCongestionData(); } }, 25, 25, TimeUnit.MILLISECONDS); - d1 = KLALBScheduler.getScheduler().scheduleAtFixedRate(new Runnable() { + d1 = KLALBScheduler.getScheduler().scheduleWithFixedDelay(new Runnable() { @Override public void run() { @@ -434,7 +441,7 @@ public class LineMonitorGUI extends XFrame { } } } ,25, 25, TimeUnit.MILLISECONDS); - d2 = KLALBScheduler.getScheduler().scheduleAtFixedRate( new Runnable() { + d2 = KLALBScheduler.getScheduler().scheduleWithFixedDelay( new Runnable() { @Override public void run() { @@ -448,7 +455,7 @@ public class LineMonitorGUI extends XFrame { } } }, 25, 25, TimeUnit.MILLISECONDS); - d3 = KLALBScheduler.getScheduler().scheduleAtFixedRate(new Runnable() { + d3 = KLALBScheduler.getScheduler().scheduleWithFixedDelay(new Runnable() { @Override public void run() { @@ -462,7 +469,7 @@ public class LineMonitorGUI extends XFrame { } } }, 25, 25, TimeUnit.MILLISECONDS); - d4 =KLALBScheduler.getScheduler().scheduleAtFixedRate( new Runnable() { + d4 =KLALBScheduler.getScheduler().scheduleWithFixedDelay( new Runnable() { @Override public void run() { diff --git a/src/org/kne/cloud/network/kltp/KLTPOutputStream.java b/src/org/kne/cloud/network/kltp/KLTPOutputStream.java index e1356db..683fd63 100644 --- a/src/org/kne/cloud/network/kltp/KLTPOutputStream.java +++ b/src/org/kne/cloud/network/kltp/KLTPOutputStream.java @@ -3,7 +3,6 @@ package org.kne.cloud.network.kltp; import java.io.IOException; import java.io.OutputStream; import java.net.BindException; -import java.net.Inet6Address; import java.net.SocketException; import java.net.SocketTimeoutException; import java.nio.ByteBuffer; @@ -11,14 +10,11 @@ import java.nio.channels.WritableByteChannel; import org.kne.cloud.network.klalb.KLALBScheduler; import org.kne.concurrent.*; -import org.kne.cloud.network.NetworkPacket; import org.kne.cloud.network.congestion.CongestionAlgorithm; -import org.kne.cloud.network.congestion.DCTCP2CongestionAlgorithm; import org.kne.cloud.network.congestion.DCTCPCongestionAlgorithm; import org.kne.cloud.network.congestion.SendPacketSlidingWindow; import org.kne.cloud.network.ipv6.IPv6Address; import org.kne.cloud.network.ipv6.IPv6Packet; -import org.kne.cloud.network.klalb.DATATPacket; import org.kne.cloud.network.klalb.KLALBController; import java.util.UUID; @@ -240,19 +236,23 @@ public class KLTPOutputStream extends OutputStream implements KLTPPacketConsumer @Override public void run() { - if(sendMap.isClosed()) + if(sendMap.isClosed()) { tt.cancel(false); - try { - if(olock!=null) - olock.lock(); + }else if(!sendMap.isCongestion()){ try { - flush0(); - }finally { - if(olock!=null) - olock.unlock(); + if (olock != null) + olock.lock(); + try { + if(!sendMap.isCongestion()) { + flush0(); + } + } finally { + if (olock != null) + olock.unlock(); + } + } catch (IOException e) { + ioe.set(e); } - } catch (IOException e) { - ioe.set(e); } } }; @@ -273,7 +273,7 @@ public class KLTPOutputStream extends OutputStream implements KLTPPacketConsumer sendMap.waitForAvaliable(); - controller.getIpv6Router().runPacketSendTask(()->{ + controller.getIpv6Router().enqueuePacketSendTask(()->{ pack.getKLTPData().flip(); pack.incSendCounter(); sendMap.put(pack.getSequence(),pack); @@ -355,7 +355,7 @@ public class KLTPOutputStream extends OutputStream implements KLTPPacketConsumer KLTPPacket kltp=(KLTPPacket) u.getPayload(); switch(kltp.getType()) { case KLTPPacket.KLTP_TYPE_ACK: - sendMap.ack(kltp.getSequence(), kltp.isCE()); + sendMap.ack(kltp.getSequence(),0, kltp.isCE()); break; } diff --git a/src/org/kne/cloud/network/monitor/ArrayListTimestampMonitor.java b/src/org/kne/cloud/network/monitor/ArrayListTimestampMonitor.java index 4950725..9fddd7a 100644 --- a/src/org/kne/cloud/network/monitor/ArrayListTimestampMonitor.java +++ b/src/org/kne/cloud/network/monitor/ArrayListTimestampMonitor.java @@ -46,7 +46,7 @@ public class ArrayListTimestampMonitor implements TimestampMonitor { return clock; } - protected ArrayListTimestampMonitor(HighAccuracyClock clock, String name, int maxRetry, long cleanupThreshold) { + public ArrayListTimestampMonitor(HighAccuracyClock clock, String name, int maxRetry, long cleanupThreshold) { super(); this.clock = clock; this.name = name; diff --git a/src/org/kne/cloud/network/monitor/SpeedAndTrafficAndDelayMonitorDataImpl.java b/src/org/kne/cloud/network/monitor/SpeedAndTrafficAndDelayMonitorDataImpl.java index 9afdb98..de631cf 100644 --- a/src/org/kne/cloud/network/monitor/SpeedAndTrafficAndDelayMonitorDataImpl.java +++ b/src/org/kne/cloud/network/monitor/SpeedAndTrafficAndDelayMonitorDataImpl.java @@ -20,21 +20,9 @@ public class SpeedAndTrafficAndDelayMonitorDataImpl extends SpeedAndTrafficMonit downSampler.update(); if(upSampler.getSnapshotPacketCount()>0&&upSampler.getSnapshotTotalDelay()>0) { getUploadBandwidth(). recordPacket(KLALBUtils.createGlobalUUID(),0, 0, upSampler.getAvgDelayNanos(),upSampler.getMinDelayNanos()); - /*long odelay=upSampler.getMinDelayNanos(); - if(outDelayMin==Long.MAX_VALUE||odelay<=outDelayMin) { - outDelayMin=odelay; - }else{ - outDelayMin=Math.min(odelay, outDelayMin+10000); - }*/ } if(downSampler.getSnapshotPacketCount()>0&&downSampler.getSnapshotTotalDelay()>0) { getDownloadBandwidth(). recordPacket(KLALBUtils.createGlobalUUID(),0, 0, downSampler.getAvgDelayNanos(),downSampler.getMinDelayNanos()); - long idelay=downSampler.getMinDelayNanos(); - /*if(inDelayMin==Long.MAX_VALUE||idelay<=inDelayMin) { - inDelayMin=idelay; - }else{ - inDelayMin=Math.min(idelay, inDelayMin+10000); - }*/ } long time=System.nanoTime(); diff --git a/src/org/kne/cloud/network/monitor/SpeedAndTrafficMonitorDataImpl.java b/src/org/kne/cloud/network/monitor/SpeedAndTrafficMonitorDataImpl.java index 3b42575..35781d4 100644 --- a/src/org/kne/cloud/network/monitor/SpeedAndTrafficMonitorDataImpl.java +++ b/src/org/kne/cloud/network/monitor/SpeedAndTrafficMonitorDataImpl.java @@ -11,7 +11,7 @@ public class SpeedAndTrafficMonitorDataImpl extends MonitorDataImpl implements S private TimestampMonitoruploadBandwidth; private TimestampMonitordownloadBandwidth; private long timewindow=1000000000L; - private long timewindowmax=50000000000L; + private long timewindowmax=30000000000L; private BandwidthSampler upSampler=new BandwidthSampler(); private BandwidthSampler downSampler=new BandwidthSampler(); diff --git a/src/org/kne/cloud/network/perf/Kperf.java b/src/org/kne/cloud/network/perf/Kperf.java index a70ecfb..c4a46ef 100644 --- a/src/org/kne/cloud/network/perf/Kperf.java +++ b/src/org/kne/cloud/network/perf/Kperf.java @@ -4,17 +4,15 @@ import java.io.IOException; import java.nio.channels.UnresolvedAddressException; import java.util.Objects; import java.util.concurrent.ConcurrentLinkedQueue; +import java.util.concurrent.ScheduledFuture; +import java.util.concurrent.TimeUnit; import org.kne.cloud.clock.AdjustedNanoClock; import org.kne.cloud.clock.HighAccuracyClock; import org.kne.cloud.network.MultiProtocolSocketAddress; import org.kne.cloud.network.SpeedLimiter; import org.kne.cloud.network.ThreadTool; -import org.kne.cloud.network.klalb.KLALBPacket; -import org.kne.cloud.network.klalb.KLALBPacketLink; -import org.kne.cloud.network.klalb.KLALBUtils; -import org.kne.cloud.network.klalb.PINGPacket; -import org.kne.cloud.network.klalb.PONGPacket; +import org.kne.cloud.network.klalb.*; import org.kne.cloud.network.monitor.LinkStatus; import org.kne.cloud.network.monitor.SpeedAndTrafficAndDelayMonitorDataImpl; import org.kne.concurrent.ThreadParker; @@ -86,18 +84,15 @@ public class Kperf implements Runnable{ long usedtime=0; - private MonitorUpdater monitorUpdater; + private ScheduledFuture monitorUpdater; private class MonitorUpdater implements Runnable{ @Override public void run() { - while (!closed) { + if (closed) { + monitorUpdater.cancel(false); + }else{ status.updateReliability(); monitor.update(); - try { - Thread.sleep(1); - } catch (InterruptedException e) { - e.printStackTrace(); - } } @@ -436,8 +431,7 @@ public class Kperf implements Runnable{ public void startPerfing() { - monitorUpdater=new MonitorUpdater(); - ThreadTool.makeVDaemonThread("线路性能采样线程", monitorUpdater).start(); + monitorUpdater= KLALBScheduler.getScheduler().scheduleWithFixedDelay(new MonitorUpdater(),1,1, TimeUnit.MILLISECONDS); ThreadTool.makeVDaemonThreadIfSupport("测速接收线程" ,this).start(); } } diff --git a/src/org/kne/cloud/network/srv6/SRv6Router.java b/src/org/kne/cloud/network/srv6/SRv6Router.java index 393cd2a..90b1955 100644 --- a/src/org/kne/cloud/network/srv6/SRv6Router.java +++ b/src/org/kne/cloud/network/srv6/SRv6Router.java @@ -61,7 +61,7 @@ public class SRv6Router { // public static final int MTU = 16384; public static final int MTU = 16384; // 最大传输单元 - public static final int MAX_REROUTE_COUNT = 2;// 最大重路由次数 + public static final int MAX_REROUTE_COUNT = 1;// 最大重路由次数 private static final int IPv6_BITS = 128; // IPv6地址位数 @@ -575,7 +575,7 @@ public class SRv6Router { if (srhh != null) { processSRv6Packet(linkfrom, iPv6Packet, ri, srhh); // SRv6特殊处理 } else { - sendPacketToRouteItem(iPv6Packet, ri); + sendPacketToRouteItem(linkfrom,iPv6Packet, ri); } } else { // 普通链路处理 // 检查重路由计数 @@ -590,22 +590,22 @@ public class SRv6Router { if (hop > 0) { // TTL有效 iPv6Packet.setHopLimit(hop); // 更新TTL - sendPacketToRouteItem(iPv6Packet, ri); // 发送数据包 + sendPacketToRouteItem(linkfrom,iPv6Packet, ri); // 发送数据包 } else { sendTTLExceedPacket(iPv6Packet); // TTL超时 } } } - private void sendPacketToRouteItem(IPv6Packet iPv6Packet, RouteItem ri) throws IOException { + private void sendPacketToRouteItem(IPv6NetworkLink linkfrom,IPv6Packet iPv6Packet, RouteItem ri) throws IOException { if (iPv6Packet.getPayload() instanceof ICMPv6PostcardPacket && ri.getDestlink().isLoopBack()) { - onPostcardReceive(iPv6Packet); + onPostcardReceive(linkfrom,iPv6Packet); } else { ri.getDestlink().sendPacket(iPv6Packet, ri.getNexthop()); } } - private void onPostcardReceive(IPv6Packet postcards) { + private void onPostcardReceive(IPv6NetworkLink linkfrom,IPv6Packet postcards) { ICMPv6PostcardPacket postcardsi = (ICMPv6PostcardPacket) postcards.getPayload(); for (PostcardEntry postcard : postcardsi.getPostcards()) { IPv6Address sid = postcard.getSID(); @@ -616,7 +616,7 @@ public class SRv6Router { if (ri.getNexthop().equals(sid)) { if (ri.getDestlink() instanceof KLALBRemoteLink) { KLALBRemoteLink rem = (KLALBRemoteLink) ri.getDestlink(); - rem.onPostcardReceive(postcard); + rem.onPostcardReceive(linkfrom,postcard); } } } @@ -652,7 +652,7 @@ public class SRv6Router { if (srhh.getSegmentsLeft() <= 0) { // 所有段已处理完毕 - sendPacketToRouteItem(iPv6Packet, ri); + sendPacketToRouteItem(linkfrom,iPv6Packet, ri); } else { if (iPv6Packet.getRouteCounter().get()<=0) { // 重路由不处理 @@ -904,7 +904,7 @@ public class SRv6Router { linkTabel.add(inLoopBack); // 添加到链路表 this.srhInserter=new SRHInsertNetworkLink(this,networkGroup); linkTabel.add(srhInserter); - KLALBScheduler.getScheduler().scheduleAtFixedRate(new Runnable() { + KLALBScheduler.getScheduler().scheduleWithFixedDelay(new Runnable() { @Override public void run() {