修复滑动窗口出现负数的bug

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
+rttmin=RTT min
+rto=RTO

src/klalb_zh_CN.properties

@@ -121,4 +121,5 @@ congestionmonitor=拥塞控制监视器
 rttbaseline=往返延迟基线
 rtt=往返延迟
 rttmax=往返延迟上限
-rttmin=往返延迟下限
+rttmin=往返延迟下限
+rto=重传超时

src/org/kne/cloud/network/congestion/BBRCongestionAlgorithm.java

@@ -11,14 +11,13 @@ public class BBRCongestionAlgorithm implements CongestionAlgorithm ,DetnetConges
 
 	private Consumer<Long> 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;

src/org/kne/cloud/network/congestion/CongestionAlgorithm.java

@@ -6,13 +6,13 @@ public interface CongestionAlgorithm {
 	public void reset();
 	public void setWindowControlConsumer(Consumer<Long>windowControlConsumer);
 	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();
 }

src/org/kne/cloud/network/congestion/DCTCP2CongestionAlgorithm.java

@@ -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<Long> 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<Long> 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(congressWindowSize<MIN_WINDOW) {
-			congressWindowSize=MIN_WINDOW;
-		}else if(congressWindowSize>maxwindow) {
-			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";
-	}
-
-}

src/org/kne/cloud/network/congestion/DCTCPCongestionAlgorithm.java

@@ -13,16 +13,13 @@ public class DCTCPCongestionAlgorithm implements CongestionAlgorithm {
 
 	private Consumer<Long> 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();
+	}
+
 
 }

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;
+	}
+
 }

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<UUID> rttMonitor=new ArrayListTimestampMonitor<UUID>(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();
+    }
+}

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<K, V extends NetworkPacket> implements Closeable, AutoCloseable {
+	private final ScheduledFuture<?> checkerTask;
 	private ThreadParker tp=new ThreadParker();
-	
-	private Map<K, SendItem<V>> sendMap = new ConcurrentHashMap<>(2048);
+
+	private volatile Map<K, SendItem<V>> sendMap = new ConcurrentHashMap<>(2048);
 	private AtomicLong sendWindowUsed=new AtomicLong(0);
 	//private LongAdder sendWindowUsed = new LongAdder();
 
@@ -51,13 +56,12 @@ public class SendPacketSlidingWindow<K, V extends NetworkPacket> 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<K, V extends NetworkPacket> 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<K, V extends NetworkPacket> implements Clos
 		waitForAvaliable();
 		put(sequence,packet);
 	}
-	
+
 	public void put(K sequence, V packet)  {
 		SendItem<V> newitem = new SendItem<V>(packet,headerCalibrate);
 		SendItem<V> 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<V> 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<K, V extends NetworkPacket> implements Clos
 		}
 	}
 
-	public V ack(K sequence,boolean ecn,long OWDdown) {
-		SendItem<V> 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<SendItem<V>> vals = sendMap.values();
-				for (Iterator<SendItem<V>> iterator = vals.iterator(); iterator.hasNext();) {
-					SendItem<V> sitm = (SendItem<V>) 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<V> item : sendMap.values()) {
+							closingConsumer.accept(item.getPacket());
 						}
-						V pktr = sitm.getPacket();
-						if (resendConsumer != null)
-							resendConsumer.accept(pktr);
 					}
-				}
+					sendMap.clear();
+					sendWindowUsed.set(0);
+
+				} else {
+
+					Set<Map.Entry<K,SendItem<V>>> vals = sendMap.entrySet();
+					for (Iterator<Map.Entry<K,SendItem<V>>> iterator = vals.iterator(); iterator.hasNext(); ) {
+						Map.Entry<K,SendItem<V>> entry = (Map.Entry<K,SendItem<V>>) iterator.next();
+						SendItem<V> sitm=entry.getValue();
+						long current = System.nanoTime();
+						long ftime=current - sitm.getSendtime();
+						if (ftime> algorithm.getRTO()) {
+
+							V pktr = sitm.getPacket();
+
+							if (removeAtResend) {
+								SendItem<V> 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<V> item : sendMap.values()) {
-					closingConsumer.accept(item.getPacket());
-				}
-			}
-			sendMap.clear();
-			sendWindowUsed.set(0);
 		}
 	}
 

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<Long> 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;

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<UUID>(HighAccuracyClock.SYSTEM_CLOCK, "up", 100,TIME_WINDOW ),
 			new HashMapTimestampMonitor<UUID>(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();

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<RouteItem> getRouteItems() {
-		List<RouteItem> rlist = new ArrayList<>();
-
-	if(isUp()) {
-	
-
-		for (Iterator<Neighbor> 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);
+		}
 		
 	}
 

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()) {

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() {

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;
 				}

src/org/kne/cloud/network/monitor/ArrayListTimestampMonitor.java

@@ -46,7 +46,7 @@ public class ArrayListTimestampMonitor<K> implements TimestampMonitor<K> {
 		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;

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();

src/org/kne/cloud/network/monitor/SpeedAndTrafficMonitorDataImpl.java

@@ -11,7 +11,7 @@ public class SpeedAndTrafficMonitorDataImpl extends MonitorDataImpl implements S
 	private TimestampMonitor<UUID>uploadBandwidth;
 	private TimestampMonitor<UUID>downloadBandwidth;
 	private long timewindow=1000000000L;
-	private long timewindowmax=50000000000L;
+	private long timewindowmax=30000000000L;
 	private BandwidthSampler upSampler=new BandwidthSampler();
 	private BandwidthSampler downSampler=new BandwidthSampler();
 

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();
 	}
 }

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() {