减小体积

修复拥塞算法RTT延迟为0的bug

.classpath

@@ -2,34 +2,19 @@
 <classpath>
 	<classpathentry kind="src" path="src"/>
 	<classpathentry kind="lib" path="lib/gson-2.1.jar"/>
-	<classpathentry kind="lib" path="lib/javassist.jar" sourcepath="C:/Users/ADMINI~1/AppData/Local/Temp/1/.org.sf.feeling.decompiler1686963856070/source/javassist-3.29.2-GA-sources.jar"/>
 	<classpathentry kind="lib" path="lib/ini4j.jar"/>
 	<classpathentry kind="lib" path="lib/toml4j-0.7.1.jar"/>
-	<classpathentry kind="lib" path="lib/pcap4j-core-1.8.2.jar" sourcepath="C:/Users/ADMINI~1/AppData/Local/Temp/1/.org.sf.feeling.decompiler1725520274095/source/pcap4j-core-1.8.2-sources.jar"/>
-	<classpathentry kind="lib" path="lib/pcap4j-packetfactory-propertiesbased-1.8.2.jar"/>
-	<classpathentry kind="lib" path="lib/pcap4j-packetfactory-static-1.8.2.jar"/>
-	<classpathentry kind="lib" path="lib/pcap4j-sample-1.8.2.jar"/>
-	<classpathentry kind="lib" path="lib/slf4j-api-2.0.9.jar"/>
-	<classpathentry kind="lib" path="lib/slf4j-api-2.0.9-javadoc.jar"/>
-	<classpathentry kind="lib" path="lib/slf4j-api-2.0.9-sources.jar"/>
-	<classpathentry kind="lib" path="lib/slf4j-api-2.0.9-tests.jar"/>
-	<classpathentry kind="lib" path="lib/disruptor-4.0.0.jar" sourcepath="lib/disruptor-4.0.0-sources.jar"/>
-	<classpathentry kind="lib" path="lib/disruptor-4.0.0-javadoc.jar"/>
-	<classpathentry kind="lib" path="lib/disruptor-4.0.0-sources.jar"/>
-	<classpathentry kind="lib" path="lib/KNElib1.0.jar"/>
-	<classpathentry kind="lib" path="lib/KNEOptimize.jar"/>
-	<classpathentry kind="lib" path="lib/JavaTUN0.3.jar"/>
-	<classpathentry kind="lib" path="lib/uuid-creator-6.1.1.jar"/>
 	<classpathentry kind="lib" path="lib/jctools-core-4.0.6.jar"/>
-	<classpathentry kind="lib" path="lib/jctools-core-4.0.6-javadoc.jar"/>
 	<classpathentry kind="lib" path="lib/jctools-core-4.0.6-sources.jar"/>
+	<classpathentry kind="lib" path="lib/KNElib1.0.jar"/>
+	<classpathentry kind="lib" path="lib/JavaTUN0.3.jar"/>
+	<classpathentry kind="lib" path="lib/KNEOptimize.jar"/>
 	<classpathentry kind="con" path="org.eclipse.jdt.launching.JRE_CONTAINER/org.eclipse.jdt.internal.debug.ui.launcher.StandardVMType/JavaSE-25">
 		<attributes>
 			<attribute name="module" value="true"/>
 		</attributes>
 	</classpathentry>
 	<classpathentry kind="lib" path="lib/jfreechart-1.5.3.jar"/>
-	<classpathentry kind="lib" path="lib/jfreechart-1.5.3-javadoc.jar"/>
 	<classpathentry kind="lib" path="lib/jfreechart-1.5.3-sources.jar"/>
 	<classpathentry kind="output" path="bin"/>
 </classpath>

src/klalb_en_US.properties

@@ -122,3 +122,4 @@ rttbaseline=RTT baseline
 rtt=RTT
 rttmax=RTT max
 rttmin=RTT min
+rto=RTO

src/klalb_zh_CN.properties

@@ -122,3 +122,4 @@ rttbaseline=往返延迟基线
 rtt=往返延迟
 rttmax=往返延迟上限
 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;
+		rttEstimator.putRTT(latencyns);
-		RTTAvg = (RTTAvg * 7 + latencyns) / 8;
-		
-		RTO = RTTAvg + Math.max(MIN_RTTVAR, RTTVar * 4);// RTTVar*4
 
 		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) {
+	public void putLoss(long packetSize,long lossns);
-		putAck(packetSize,RTTns,ecn);
-	}
-	public void putLoss(long packetSize,long lossns,int losscounter);
 	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)) {
+		rttEstimator.putRTT(latencyns);
-			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
 
 		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();
 
@@ -56,8 +61,7 @@ public class SendPacketSlidingWindow<K, V extends NetworkPacket> implements Clos
 		this.removeAtResend = removeAtResend;
 		this.algorithm = algorithm;
 		this.windowSize=windowSize;
-		Thread t = ThreadTool.makeVThread("重路由计时器线程", checker);
+		this.checkerTask=KLALBScheduler.getScheduler().scheduleWithFixedDelay(checker,1,1, TimeUnit.MILLISECONDS);
-		t.start();
 	}
 
 	public long getWindowSize() {
@@ -131,16 +135,15 @@ public class SendPacketSlidingWindow<K, V extends NetworkPacket> implements Clos
 		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(), RTTns, ecn);
-			algorithm.putAck(ipv6.getPacketLength(), RTTC, ecn);
 			tp.unpark();
 			// System.out.println("remove:"+aseq.getSequence()+" "+sendMap.size());
 			Lock lockx = lock;
@@ -162,65 +165,15 @@ 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);
-						}
-						V pktr = sitm.getPacket();
-						if (resendConsumer != null)
-							resendConsumer.accept(pktr);
-					}
-				}
-
 			try {
-					Thread.sleep(2);
+				if (closed) {
-				} catch (InterruptedException e) {
+					checkerTask.cancel(false);
-					e.printStackTrace();
-				}
-			}
 					// 清理资源
 					if (closingConsumer != null) {
 						for (SendItem<V> item : sendMap.values()) {
@@ -229,6 +182,42 @@ public class SendPacketSlidingWindow<K, V extends NetworkPacket> implements Clos
 					}
 					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);
+							}
+						}
+					}
+
+
+				}
+			} catch (Exception e) {
+				e.printStackTrace();
+			}
 		}
 	}
 

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) {
@@ -73,29 +74,18 @@ public class Vegas2CongestionAlgorithm implements CongestionAlgorithm,DetnetCong
             }
         }
 
-        dsp.recordDelay(RTTns);
         
         // 更新平滑RTT估计
-        
+       rttEstimator.putRTT(RTTns);
-            RTTVar = (RTTVar * 3 + Math.abs(RTTAvg - RTTns)) / 4;
-            RTTAvg = (RTTAvg * 7 + RTTns) / 8;
-        
-        
-        RTO = RTTAvg + Math.max(MIN_RTTVAR, RTTVar * 4);
         
         RTTTotal += RTTns;
         RTTCount++;
-        
+        if(slock.tryLock())try {
             long curr = System.nanoTime();
             // 每个RTT周期调整一次窗口（使用当前估计的RTT）
             if (curr - RTTstartTime > Math.max(BIAS, 2 * RTTMin)) {
-        	dsp.update();
+                rttEstimator.update();
-        	long rmin=dsp.getMinDelayNanos();
+                RTTMin = rttEstimator.getBaseRTT();
-        	if(rmin<=RTTMin) {
-        		RTTMin=rmin;
-        	}else {
-        		RTTMin=Math.min(rmin, RTTMin+1000);
-        	}
 
 
                 long RTTCountx = RTTCount;
@@ -124,14 +114,13 @@ public class Vegas2CongestionAlgorithm implements CongestionAlgorithm,DetnetCong
                 // ===================================================
 
 
-            
-            
                 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/ipv6/AbstractIPv6NetworkLink.java

@@ -15,11 +15,18 @@ public abstract class AbstractIPv6NetworkLink implements IPv6NetworkLink {
 	private SRv6Router srv6Router;
 	private final boolean loopback;
 
+	private final boolean filter;
 
 
 	public AbstractIPv6NetworkLink(boolean loopback) {
 		super();
 		this.loopback = loopback;
+		this.filter = false;
+	}
+	public AbstractIPv6NetworkLink(boolean loopback, boolean filter) {
+		super();
+		this.loopback = loopback;
+        this.filter = filter;
     }
 
 	public SRv6Router getSrv6Router() {
@@ -108,4 +115,9 @@ public abstract class AbstractIPv6NetworkLink implements IPv6NetworkLink {
 		}
 		return rlist;
 	}
+
+	@Override
+	public boolean isFilter() {
+		return filter;
+	}
 }

src/org/kne/cloud/network/ipv6/IPv6NetworkLink.java

@@ -30,4 +30,5 @@ public interface IPv6NetworkLink {
 	public void addIPv6LinkStateListener(IPv6LinkStateListener listener);
 	public void removeIPv6LinkStateListener(IPv6LinkStateListener listener);
 
+	public boolean isFilter();
 }

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,8 +733,10 @@ public class KLALBController {
 	}
 
 	protected void initKLALB(boolean enableVirtualAdapter, IPv6AddressGroup selfg) {
-		KLALBScheduler.getScheduler().scheduleAtFixedRate( ()->{
+		KLALBScheduler.getScheduler().scheduleWithFixedDelay( ()->{
 			linkMonitor.update();
+		},1,1,TimeUnit.MILLISECONDS);
+		KLALBScheduler.getScheduler().scheduleWithFixedDelay( ()->{
 		datatMonitor.update();
 		},1,1,TimeUnit.MILLISECONDS);
 		loadSRv6ProtocolStack(selfg, enableVirtualAdapter);

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.*;
-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.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 curr= sysclk.getCurrentTimeNanos();
 		Long128 psnd = NTPTimestamps.ntp128BitToNanosTimestamp(
-					NTPTimestamps.inferNtp64To128(postcard.getTimestampBase(), sysclk.getCurrentTimeNTP128()));
+				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/KLALBVirtualDatagramSocket.java

@@ -7,7 +7,6 @@ import java.net.SocketException;
 
 import org.kne.cloud.network.VirtualDatagramSocket;
 import org.kne.cloud.network.VirtualDatagramSocketImpl;
-import org.pcap4j.core.Inets;
 
 public class KLALBVirtualDatagramSocket extends VirtualDatagramSocket {
 

src/org/kne/cloud/network/klalb/StreamChannelKLALBPacketLink.java

@@ -118,39 +118,43 @@ public class StreamChannelKLALBPacketLink extends AbstractKLALBPacketLink implem
 		return true;
 	}
 
-	
+	// 增加两个 ThreadLocal 静态变量，用于复用 4 字节的长度头
+	private static final ThreadLocal<ByteBuffer> SIZE_WRITE_BUFFER = ThreadLocal.withInitial(() -> ByteBuffer.allocate(4));
+	private static final ThreadLocal<ByteBuffer> SIZE_READ_BUFFER = ThreadLocal.withInitial(() -> ByteBuffer.allocate(4));
 
 	@Override
 	public void writePacket(ByteBuffer kp) throws IOException {
-		ByteBuffer szeWrite=NetworkPacket.bufferAllocator.allocate(4);
+		// 1. 获取并清空缓存的 4 字节 buffer
-		szeWrite.limit(4);
+		ByteBuffer szeWrite = SIZE_WRITE_BUFFER.get();
-		//szeWrite.clear();
+		szeWrite.clear();
+
+		// 2. 写入长度并翻转
 		szeWrite.putInt(kp.limit());
 		szeWrite.flip();
-		KLALBVirtualSocketChannel obj = null;
+
+		// 3. 发送
 		writableChannel.write(new ByteBuffer[] {szeWrite, kp});
 		checkflush();
-		
-		
 	}
 
-	
-
-
 	@Override
 	public ByteBuffer readPacket() throws IOException {
+		// 1. 获取并清空缓存的 4 字节 buffer
+		ByteBuffer szeRead = SIZE_READ_BUFFER.get();
+		szeRead.clear();
+		szeRead.limit(4);
 
-		ByteBuffer szeRead=NetworkPacket.bufferAllocator.allocate(4);
 		timeoutTimer = System.nanoTime();
 		timerenabled = true;
 		ByteBuffer kp;
-		szeRead.clear();
+
-		szeRead.limit(4);
-		//szeRead.clear();
 		try {
+			// 2. 读取长度头
 			KNEChannels.readFully(readableChannel, szeRead);
 			szeRead.flip();
 			int size = szeRead.getInt(0);
+
+			// 3. 分配真正的数据区（这里如果是高频，建议后续也可以池化，不过现在是按需大小分配）
 			kp = NetworkPacket.bufferAllocator.allocate(size);
 			kp.limit(size);
 			KNEChannels.readFully(readableChannel, kp);

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()) {
@@ -1417,29 +1417,29 @@ public class KLALBStateGUI3 extends XFrame {
     private void updateDashboardData(KLALBController kc) {
         // 更新上行速度
         double outload = kc.getLinkMonitor().getOutSpeed() * 100.0 / kc.getLinkMonitor().getOutSpeedMax();
-        if (Double.isFinite(outload)) {
         upSpeedText.setLabel(KLALBUtils.convertIBUint(kc.getLinkMonitor().getOutSpeed()) + "/s");
+        if (Double.isFinite(outload)) {
             upSpeed.setValue(Math.min(upSpeed.getValue().doubleValue() * 0.98 + outload * 0.02, 101.0));
         }
 
         // 更新下行速度
         double inload = kc.getLinkMonitor().getInSpeed() * 100.0 / kc.getLinkMonitor().getInSpeedMax();
-        if (Double.isFinite(inload)) {
         downSpeedText.setLabel(KLALBUtils.convertIBUint(kc.getLinkMonitor().getInSpeed()) + "/s");
+        if (Double.isFinite(inload)) {
             downSpeed.setValue(Math.min(downSpeed.getValue().doubleValue() * 0.98 + inload * 0.02, 101.0));
         }
 
         // 更新上行包速率
         double outPPSload = kc.getLinkMonitor().getOutPPS() * 100.0 / kc.getLinkMonitor().getOutPPSMax();
-        if (Double.isFinite(outPPSload)) {
         upPPSText.setLabel(KLALBUtils.convertUintDefalut(kc.getLinkMonitor().getOutPPS()) + "PPS");
+        if (Double.isFinite(outPPSload)) {
             upPPS.setValue(Math.min(upPPS.getValue().doubleValue() * 0.98 + outPPSload * 0.02, 101.0));
         }
 
         // 更新下行包速率
         double inPPSload = kc.getLinkMonitor().getInPPS() * 100.0 / kc.getLinkMonitor().getInPPSMax();
-        if (Double.isFinite(inPPSload)) {
         downPPSText.setLabel(KLALBUtils.convertUintDefalut(kc.getLinkMonitor().getInPPS()) + "PPS");
+        if (Double.isFinite(inPPSload)) {
             downPPS.setValue(Math.min(downPPS.getValue().doubleValue() * 0.98 + inPPSload * 0.02, 101.0));
         }
 
@@ -1471,29 +1471,29 @@ public class KLALBStateGUI3 extends XFrame {
        if(router!=null) {
     	   // 更新背板延迟
            double backplaneTimev = router.getBackplaneTime()* 100.0/100000.0;
-           if (Double.isFinite(backplaneTimev)) {
            backplaneDelayText.setLabel(KLALBUtils.convertUintNanoDefalut(router.getBackplaneTime()) + "s");
+           if (Double.isFinite(backplaneTimev)) {
                backplaneDelay.setValue(Math.min(backplaneDelay.getValue().doubleValue() * 0.98 + backplaneTimev * 0.02, 101.0));
            }
 
            // 更新背板转发率
            double backplanePPSv = router.getBackplanePPS() * 100.0 / router.getBackplanePPSMax();
-           if (Double.isFinite(backplanePPSv)) {
            backplanePPSText.setLabel(KLALBUtils.convertUintDefalut(router.getBackplanePPS()) + "PPS");
+           if (Double.isFinite(backplanePPSv)) {
                backplanePPS.setValue(Math.min(backplanePPS.getValue().doubleValue() * 0.98 + backplanePPSv * 0.02, 101.0));
            }
            
            // 更新背板ECN率
            double backplaneECNv = router.getECNRate() * 100.0 ;
-           if (Double.isFinite(backplaneECNv)) {
            backplaneECNRateText.setLabel(String.format("%.1f",backplaneECNv) + "%");
+           if (Double.isFinite(backplaneECNv)) {
                backplaneECNRate.setValue(Math.min(backplaneECNRate.getValue().doubleValue() * 0.98 + backplaneECNv * 0.02, 101.0));
            }
            
         // 更新背板ECN率
            double backplaneLossv = router.getLossRate() * 100.0 ;
-           if (Double.isFinite(backplaneLossv)) {
            backplaneLossRateText.setLabel(String.format("%.1f",backplaneLossv) + "%");
+           if (Double.isFinite(backplaneLossv)) {
                backplaneLossRate.setValue(Math.min(backplaneLossRate.getValue().doubleValue() * 0.98 + backplaneLossv * 0.02, 101.0));
            }
        }

src/org/kne/cloud/network/klalb/ui/KperfGUI.java

@@ -45,7 +45,6 @@ public class KperfGUI extends XFrame{
 	private Kperf kperf;
 	private JTextField targetAddressField;
 	private JButton startButton;
-	private JProgressBar progressBar;
 	private DefaultValueDataset upSpeed;
 	private DefaultValueDataset upDataSpeed;
 	private DialTextAnnotation upSpeedText;
@@ -54,7 +53,7 @@ public class KperfGUI extends XFrame{
 	private DialTextAnnotation downSpeedText;
 	private JTextField sourceAddressField;
 
-	private SpeedAndTrafficAndDelayMonitorDataImpl monitor;
+	private volatile SpeedAndTrafficAndDelayMonitorDataImpl monitor;
 
 	private DefaultValueDataset upPPS;
 
@@ -70,10 +69,6 @@ public class KperfGUI extends XFrame{
 
 	private TimerTask tsk;
 	private class KperfReportsImpl implements KperfReports{
-		@Override
-		public void testProcess(double process) {
-			progressBar.setValue((int) (process*100.0f));
-		}
 		@Override
 		public void testFinish() {
 			if(kperf!=null) {
@@ -521,9 +516,6 @@ public class KperfGUI extends XFrame{
 				
 			
 
-				progressBar = new JProgressBar();
-				progressBar.setPreferredSize(new Dimension(100, 5));
-				getContentPane().add(progressBar, BorderLayout.NORTH);
 				getTitlelabel().setFont(UIEnv.getFont().deriveFont(17.0f));
 				getTitlepanel().setBorder(new LineBorder(Color.LIGHT_GRAY));
 				getTitlelabel().setForeground(Color.WHITE);
@@ -538,26 +530,26 @@ public class KperfGUI extends XFrame{
 					public void run() {
 						if(monitor!=null) {
 						double outload = monitor.getOutSpeed() * 100.0 / monitor.getOutSpeedMax();
-						if (Double.isFinite(outload)) {
 							upSpeedText.setLabel(KLALBUtils.convertIBUint(monitor.getOutSpeed()) + "/s");
+						if (Double.isFinite(outload)) {
 							upSpeed.setValue(Math.min(upSpeed.getValue().doubleValue() * 0.98 + outload * 0.02, 101.0));
 						}
 
 						double inload = monitor.getInSpeed() * 100.0 / monitor.getInSpeedMax();
-						if (Double.isFinite(inload)) {
 							downSpeedText.setLabel(KLALBUtils.convertIBUint(monitor.getInSpeed()) + "/s");
+						if (Double.isFinite(inload)) {
 							downSpeed.setValue(Math.min(downSpeed.getValue().doubleValue() * 0.98 + inload * 0.02, 101.0));
 						}
 
 						double outPPSload = monitor.getOutPPS() * 100.0 / monitor.getOutPPSMax();
-						if (Double.isFinite(outPPSload)) {
 							upPPSText.setLabel(KLALBUtils.convertUintDefalut(monitor.getOutPPS()) + "PPS");
+						if (Double.isFinite(outPPSload)) {
 							upPPS.setValue(Math.min(upPPS.getValue().doubleValue() * 0.98 + outPPSload * 0.02, 101.0));
 						}
 
 						double inPPSload = monitor.getInPPS() * 100.0 / monitor.getInPPSMax();
-						if (Double.isFinite(inPPSload)) {
 							downPPSText.setLabel(KLALBUtils.convertUintDefalut(monitor.getInPPS()) + "PPS");
+						if (Double.isFinite(inPPSload)) {
 							downPPS.setValue(Math.min(downPPS.getValue().doubleValue() * 0.98 + inPPSload * 0.02, 101.0));
 						}
 						}

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) {
+			if(window !=null) {
-				Vegas2CongestionAlgorithm deta=(Vegas2CongestionAlgorithm) window;
+				double baseRTT = window.getBaseRTT();
-				double baseRTT=deta.getBaseRTT();
 				rttbaseline.addOrUpdate(ms, baseRTT / 1000000.0);
-				rtt.addOrUpdate(ms, deta.getCurrentRTT() / 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,13 +236,16 @@ public class KLTPOutputStream extends OutputStream implements KLTPPacketConsumer
 					
 					@Override
 					public void run() {
-						if(sendMap.isClosed())
+						if(sendMap.isClosed()) {
 							tt.cancel(false);
+						}else if(!sendMap.isCongestion()){
 							try {
 								if (olock != null)
 									olock.lock();
 								try {
+									if(!sendMap.isCongestion()) {
 										flush0();
+									}
 								} finally {
 									if (olock != null)
 										olock.unlock();
@@ -255,6 +254,7 @@ public class KLTPOutputStream extends OutputStream implements KLTPPacketConsumer
 								ioe.set(e);
 							}
 						}
+					}  
 				};
 				tt= KLALBScheduler.getScheduler().scheduleWithFixedDelay(r, delaytime, delaytime,TimeUnit.NANOSECONDS);
 				}
@@ -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();
 
@@ -177,7 +177,7 @@ public class SpeedAndTrafficMonitorDataImpl extends MonitorDataImpl implements S
 			inPPSEvaluator.markDirty();
 		}
 		long time=System.nanoTime();
-		if(time-stime>timewindowmax) {
+		if(time-stime>timewindow) {
 			stime=time;
 		long os=uploadBandwidth.calculateBandwidth(timewindowmax);
 		long is=downloadBandwidth.calculateBandwidth(timewindowmax);

src/org/kne/cloud/network/ntp/NTPv4Protocol.java

@@ -252,14 +252,23 @@ public class NTPv4Protocol implements Closeable, AutoCloseable {
 			checkLock.unlock();
 		}
 	}
+	// 在类成员变量中定义 ThreadLocal 发送缓冲区
+	private static final ThreadLocal<ByteBuffer> NTP_TX_BUFFER = ThreadLocal.withInitial(() -> ByteBuffer.allocate(9000));
 
 	private void sendNTPPacket(NTPv4Packet nv4, SocketAddress target) throws IOException {
-		ByteBuffer bbf = ByteBuffer.allocate(65535);
+		// 1. 获取并重置缓冲区
+		ByteBuffer bbf = NTP_TX_BUFFER.get();
+		bbf.clear(); // 重置 position=0, limit=9000
+
+		// 2. 写入数据
 		nv4.writeToChannel(KNEChannels.newWritableChannel(bbf));
-		bbf.flip();
+		bbf.flip(); // 切换为读模式，limit 会变成实际写入的长度
+
+		// 3. 发送
 		DatagramPacket dp = new DatagramPacket(bbf.array(), bbf.limit());
 		dp.setSocketAddress(target);
 		dgs.send(dp);
+
 		if (showPacket)
 			System.out.println("NTPv4 TX:" + target + " " + nv4);
 	}
@@ -379,15 +388,30 @@ public class NTPv4Protocol implements Closeable, AutoCloseable {
 		return sends;
 	}
 
+	// 1. 在类成员变量中定义 ThreadLocal 缓冲区
+	private static final ThreadLocal<ByteBuffer> NTP_BUFFER = ThreadLocal.withInitial(() -> ByteBuffer.allocate(9000));
+
 	private NTPv4Packet receiveNTPPacket(AtomicReference<InetSocketAddress> isa) throws IOException {
-		ByteBuffer bbf = ByteBuffer.allocate(65535);
+		// 2. 从 ThreadLocal 获取，并重置状态 (position=0, limit=9000)
+		ByteBuffer bbf = NTP_BUFFER.get();
+		bbf.clear();
+
+		// 3. 构造 DatagramPacket
 		DatagramPacket dp = new DatagramPacket(bbf.array(), bbf.limit());
 		dgs.receive(dp);
+
 		isa.set((InetSocketAddress) dp.getSocketAddress());
+
+		// 4. 关键：限制 ByteBuffer 的读取范围，防止越界读取历史脏数据
+		bbf.limit(dp.getLength());
+
 		NTPv4Packet nv4 = new NTPv4Packet(context.getClock());
+		// 5. 将 Channel 的读取操作限制在 dp.getLength() 长度内
 		nv4.readFromChannel(KNEChannels.newReadableChannel(bbf), dp.getLength());
+
 		if (showPacket)
 			System.out.println("NTPv4 RX:" + isa.get() + " " + nv4);
+
 		return nv4;
 	}
 

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.*;
-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.monitor.LinkStatus;
 import org.kne.cloud.network.monitor.SpeedAndTrafficAndDelayMonitorDataImpl;
 import org.kne.concurrent.ThreadParker;
@@ -79,25 +77,20 @@ public class Kperf implements Runnable{
 
 	
 	
-	KLALBPacket testPacket=null;
+	private  KLALBPacket testPacket=null;
 	
 	private long starttime=System.nanoTime();
-	long totaltime=1000;
-	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();
-				}
 				
 			}
 			
@@ -126,9 +119,9 @@ public class Kperf implements Runnable{
 		});
 		Thread tc = ThreadTool.makeVDaemonThreadIfSupport("测速控制线程", () -> {
 			final int wtimes=5;
-			totaltime=(20*1000*1+20*1000*4+wtimes*1000*5)*1000000L;
-			starttime=System.nanoTime();
 				System.out.println("测速开始！");
+				if(reports!=null)
+			reports.updateDial(monitor);
 				try {
 					long startTime;
 					
@@ -137,7 +130,6 @@ public class Kperf implements Runnable{
 					 startTime=System.nanoTime();
 					for (int i = 0; i < 20; i++) {
 						Thread.sleep(1000);
-						updateUsedTime();
 					}
 					System.out.printf("上传延迟 最小：%.3fms 平均：%.3fms 抖动：%.3fms\n",monitor.getOutDelay()/1000000.0,monitor.getOutDelay()/1000000.0,monitor.getOutJitter()/1000000.0);
 					System.out.printf("下载延迟 最小：%.3fms 平均：%.3fms 抖动：%.3fms\n",monitor.getInDelay()/1000000.0,monitor.getInDelay()/1000000.0,monitor.getInJitter()/1000000.0);
@@ -153,7 +145,7 @@ public class Kperf implements Runnable{
 					
 					System.out.println("测试2/3：带宽测试");
 					test.setLimitspeed(1000L*1024L*1024L*1024L);
-					testPacket=new TESTNPacket(65535);
+					testPacket=new TESTNPacket(9000);
 					for (int i = 0; i < 60; i++) {
 						Thread.sleep(1000);
 					}
@@ -303,9 +295,6 @@ public class Kperf implements Runnable{
 
 					
 					tlock.unpark();
-					updateUsedTime() ;
-					if(reports!=null)
-					reports.updateDial(monitor);
 					break;
 				case KLALBPacket.TESTN:
 					break;
@@ -330,12 +319,7 @@ public class Kperf implements Runnable{
 			tcc.interrupt();
 		}
 	}
-	private void updateUsedTime() {
+
-		usedtime=System.nanoTime()-starttime;
-		//System.out.println(usedtime+" "+totaltime);
-		if(reports!=null)
-			reports.testProcess(usedtime/(double)totaltime);
-	}
 
 
 
@@ -436,8 +420,7 @@ public class Kperf implements Runnable{
 	
 	
 	public void startPerfing() {
-		monitorUpdater=new MonitorUpdater();
+		monitorUpdater= KLALBScheduler.getScheduler().scheduleWithFixedDelay(new MonitorUpdater(),1,1, TimeUnit.MILLISECONDS);
-		ThreadTool.makeVDaemonThread("线路性能采样线程", monitorUpdater).start();
 		ThreadTool.makeVDaemonThreadIfSupport("测速接收线程" ,this).start();
 	}
 }

src/org/kne/cloud/network/perf/KperfReports.java

@@ -3,7 +3,6 @@ package org.kne.cloud.network.perf;
 import org.kne.cloud.network.monitor.SpeedAndTrafficAndDelayMonitorDataImpl;
 
 public interface KperfReports {
-	public void testProcess(double d);
 	public void testFinish();
 	public void updateDial(SpeedAndTrafficAndDelayMonitorDataImpl mdt);
 	public void uploadSpeedFinish(long uploadSpeed);

src/org/kne/cloud/network/perf/TESTNPacket.java

@@ -1,16 +1,11 @@
 package org.kne.cloud.network.perf;
 
-import java.io.DataInput;
-import java.io.DataInputStream;
-import java.io.DataOutput;
-import java.io.DataOutputStream;
 import java.io.EOFException;
 import java.io.IOException;
 import java.nio.ByteBuffer;
 import java.nio.channels.ReadableByteChannel;
 import java.nio.channels.WritableByteChannel;
 
-import org.kne.cloud.network.ByteBufferAllocator;
 import org.kne.cloud.network.NetworkPacket;
 import org.kne.cloud.network.klalb.KLALBPacket;
 
@@ -18,56 +13,75 @@ public class TESTNPacket extends KLALBPacket {
 
 	private static final int HEADER_LENGTH = 3;
 
-	private ByteBuffer dataBuffer;//=NetworkPacket.databufferpool_65535.borrow();
+	// 核心：一个线程私有的 65535 字节大缓存，生命周期与线程同寿
+	private static final ThreadLocal<ByteBuffer> BENCH_BUFFER =
+			ThreadLocal.withInitial(() -> ByteBuffer.allocate(65535));
+
+	// 不再持有 ByteBuffer，只记录长度
+	private int payloadLength;
+
 	@Override
 	public long getTotalLength() {
-		return HEADER_LENGTH+dataBuffer.limit();
+		return HEADER_LENGTH + payloadLength;
 	}
 
-
+	// 发送端构造
 	public TESTNPacket(int limit) {
 		super(TESTN, HEADER_LENGTH);
-		//dataBuffer.limit(limit);
+		this.payloadLength = limit;
-		dataBuffer=NetworkPacket.bufferAllocator.allocate(limit);
 	}
 
-	
+	// 接收端构造（从已有的 ByteBuffer 解析）
-
+	public TESTNPacket(ByteBuffer bb) {
-	public ByteBuffer getDataBuffer() {
+		super(bb, HEADER_LENGTH);
-		return dataBuffer;
+		this.payloadLength = klalbHeader.getChar(1);
 	}
 
+	public int getPayloadLength() {
+		return payloadLength;
+	}
 
 	@Override
 	public void writeToChannel(WritableByteChannel dto) throws IOException {
-		klalbHeader.putChar(1, (char) dataBuffer.limit());
+		// 1. 写入包头（包含 payloadLength）
+		klalbHeader.putChar(1, (char) payloadLength);
 		super.writeToChannel(dto);
-		//System.out.println(dataBuffer);
+
-		dto.write(dataBuffer.slice(0, dataBuffer.limit()));
+		// 2. 如果有 payload，借用 ThreadLocal 缓冲区并写出去
+		if (payloadLength > 0) {
+			ByteBuffer buf = BENCH_BUFFER.get();
+			buf.clear();
+			buf.limit(payloadLength); // 限制到指定长度
+
+			// 直接写出，里面的数据是上一次残留的或者全 0，对于烤机无所谓
+			dto.write(buf);
+		}
 	}
 
 	@Override
 	public void readFromChannel(ReadableByteChannel din) throws IOException {
+		// 1. 读包头
 		super.readFromChannel(din);
-		int limit=klalbHeader.getChar(1);
+		this.payloadLength = klalbHeader.getChar(1);
-		dataBuffer=NetworkPacket.bufferAllocator.allocate(limit);
-		 while(dataBuffer.hasRemaining()){
-			if(din.read(dataBuffer)==-1) {
-				throw new EOFException();
-			}
-		}
-		 dataBuffer.flip();
-	}
 
-	public TESTNPacket(ByteBuffer bb) {
+		// 2. 如果有 payload，借用缓冲区接收数据
-		super(bb,HEADER_LENGTH);
+		if (payloadLength > 0) {
-	}
+			ByteBuffer buf = BENCH_BUFFER.get();
+			buf.clear();
+			buf.limit(payloadLength);
 
+			// 循环读满（使用你现有的 KNEChannels 工具或手动循环）
+			while (buf.hasRemaining()) {
+				if (din.read(buf) == -1) {
+					throw new EOFException("连接提前断开，未能读满 TESTN 载荷");
+				}
+			}
+			// 数据读完直接丢弃，缓冲区不保存，下一次 clear() 时会自动覆盖
+		}
+	}
 
 	@Override
 	public String toString() {
-		return "TESTN["+dataBuffer.limit()+"]";
+		return "TESTN[" + payloadLength + "]";
 	}
-
-	
 }

src/org/kne/cloud/network/srv6/KLALBRoutingProtocol.java

@@ -9,6 +9,7 @@ import java.net.Inet6Address;
 import java.net.InetSocketAddress;
 import java.net.SocketAddress;
 import java.net.SocketException;
+import java.nio.ByteBuffer;
 import java.nio.channels.Channels;
 import java.util.ArrayList;
 import java.util.HashMap;
@@ -35,6 +36,7 @@ import org.kne.cloud.network.monitor.MonitorData;
 import org.kne.cloud.network.monitor.SpeedAndTrafficAndDelayMonitorDataImpl;
 import org.kne.cloud.network.monitor.SpeedAndTrafficMonitorData;
 import org.kne.cloud.network.monitor.SpeedAndTrafficMonitorDataImpl;
+import org.kne.io.KNEChannels;
 import org.kne.membandboost.MembandBoost;
 
 public class KLALBRoutingProtocol extends Thread{
@@ -211,20 +213,15 @@ public class KLALBRoutingProtocol extends Thread{
 		while(true) {
 
 			try {
-			byte[]ca=MembandBoost.allocateUninitializedByteArray(65535);
+			KLALBRoutingProtocolPacket kp=receivePacket();
-			DatagramPacket dgp=new DatagramPacket(ca, ca.length);
-			ds.receive(dgp);
-			//System.out.println(Arrays.toString( Arrays.copyOf( dgp.getData(),dgp.getLength())));
-			ByteArrayInputStream bi=new ByteArrayInputStream(dgp.getData(),0,dgp.getLength());
-			KLALBRoutingProtocolPacket kp=KLALBRoutingProtocolPacket.readKLALBPacketFromChannel(Channels.newChannel(bi));
 			
 			switch(kp.getType()) {
 			case KLALBRoutingProtocolPacket.RINFO_REQ:
 				RouterInfo rifr=netmap.get(router.getLocator().getAddress());
 				if(debug)
-				System.out.println("get Request from:"+dgp.getSocketAddress());
+				System.out.println("get Request from:"+kp.getFrom());
 				if(rifr!=null) {
-				writePacket( new RouterInfoPacket( rifr,false,-1), dgp.getSocketAddress());
+				writePacket( new RouterInfoPacket( rifr,false,-1), kp.getFrom());
 				if(debug)
 				System.out.println("response router info:"+rifr);
 				}else {
@@ -246,7 +243,7 @@ public class KLALBRoutingProtocol extends Thread{
 				netmap.put(rif.getLocator().getAddress(), rif);
 				noticeUpdate();
 				if(rifp.isFlood()) {
-					floodPacket( dgp.getSocketAddress(), rifp);
+					floodPacket( kp.getFrom(), rifp);
 				
 				}
 				
@@ -259,7 +256,7 @@ public class KLALBRoutingProtocol extends Thread{
 				break;
 			case KLALBRoutingProtocolPacket.JSON_DATA:
 				JsonDataPacket data=(JsonDataPacket) kp;
-				receivers.forEach((v)->{v.accept(dgp.getSocketAddress(),data);});
+				receivers.forEach((v)->{v.accept(kp.getFrom(),data);});
 				break;
 			default:
 				break;
@@ -276,7 +273,32 @@ public class KLALBRoutingProtocol extends Thread{
 			}
 	}
 	}
+	private static final ThreadLocal<ByteBuffer> LSDB_RX_BUFFER =
+			ThreadLocal.withInitial(() -> ByteBuffer.allocate(65535));
 
+	private KLALBRoutingProtocolPacket receivePacket() throws IOException {
+		// 2. 获取并重置缓冲区
+		ByteBuffer bbf = LSDB_RX_BUFFER.get();
+		bbf.clear();
+
+		// 3. 构造 DatagramPacket（注意这里要传底层 array 和 capacity）
+		DatagramPacket dgp = new DatagramPacket(bbf.array(), bbf.capacity());
+
+		// 4. 阻塞接收
+		ds.receive(dgp);
+
+		// 5. 限制缓冲区范围到实际收到的长度
+		bbf.limit(dgp.getLength());
+		bbf.position(0);
+
+		// 6. 直接用 KNEChannels 包装 ByteBuffer，避免 ByteArrayInputStream
+		KLALBRoutingProtocolPacket kp = KLALBRoutingProtocolPacket.readKLALBPacketFromChannel(
+				KNEChannels.newReadableChannel(bbf)
+		);
+		if(kp!=null)
+			kp.setFrom(dgp.getSocketAddress());
+		return kp;
+	}
 	private void floodPacket(SocketAddress except,RouterInfoPacket rifp) throws IOException {
 		long asn=rifp.getASN();
 		List<Neighbor>ln= router.getNeighbors();
@@ -304,13 +326,27 @@ public class KLALBRoutingProtocol extends Thread{
 	public void sendJsonPacketToAddress(JsonDataPacket pkt,SocketAddress dest) throws IOException {
 		writePacket(pkt,dest);
 	}
+	// 1. 在类成员变量处定义一个 64KB 的 ThreadLocal 缓冲区
+// 注意：这里依然使用 65535，因为你说 LSDB 同步可能需要这么大
+	private static final ThreadLocal<ByteBuffer> LSDB_TX_BUFFER =
+			ThreadLocal.withInitial(() -> ByteBuffer.allocate(65535));
 
 	private void writePacket(KLALBRoutingProtocolPacket pkt, SocketAddress dest) throws IOException {
-		ByteArrayOutputStream bos=new ByteArrayOutputStream(65535);
+		// 2. 从 ThreadLocal 获取，并清空状态
-		KLALBRoutingProtocolPacket.writeKLALBPacketToChannel(Channels.newChannel(bos),pkt);
+		ByteBuffer bbf = LSDB_TX_BUFFER.get();
-		byte[]to=bos.toByteArray();
+		bbf.clear();
-		DatagramPacket dgpx=new DatagramPacket(to,to.length);
+
+		// 3. 使用 KNEChannels 绑定到 ByteBuffer，直接写入
+		// 这避免了 ByteArrayOutputStream 内部的动态扩容和拷贝
+		KLALBRoutingProtocolPacket.writeKLALBPacketToChannel(KNEChannels.newWritableChannel(bbf), pkt);
+
+		// 4. 翻转，准备读取数据
+		bbf.flip();
+
+		// 5. 直接拿底层的 byte[] 和真实的 limit 发送
+		DatagramPacket dgpx = new DatagramPacket(bbf.array(), bbf.limit());
 		dgpx.setSocketAddress(dest);
+
 		sendLock.lock();
 		try {
 			ds.send(dgpx);

src/org/kne/cloud/network/srv6/KLALBRoutingProtocolPacket.java

@@ -4,6 +4,7 @@ import java.io.DataInputStream;
 import java.io.DataOutputStream;
 import java.io.EOFException;
 import java.io.IOException;
+import java.net.SocketAddress;
 import java.nio.ByteBuffer;
 import java.nio.channels.Channels;
 import java.nio.channels.ReadableByteChannel;
@@ -133,4 +134,11 @@ public abstract class KLALBRoutingProtocolPacket extends NetworkPacket {
 	public static void writeKLALBPacketToChannel(WritableByteChannel writableByteChannel,KLALBRoutingProtocolPacket klb) throws IOException {
 		klb.writeToChannel(writableByteChannel);
 	}
+	private transient SocketAddress from;
+	public void setFrom(SocketAddress from) {
+		this.from=from;
+	}
+	public SocketAddress getFrom(){
+		return from;
+	}
 }

src/org/kne/cloud/network/srv6/SRHInsertNetworkLink.java

@@ -11,7 +11,7 @@ public class SRHInsertNetworkLink extends AbstractIPv6NetworkLink {
 	private IPv6AddressGroup ipgroup;
 
 	public SRHInsertNetworkLink(SRv6Router sRv6Router,IPv6AddressGroup ipgroup) {
-		super(false);
+		super(false,true);
 		setSRv6Router(sRv6Router);
 		this.ipgroup=ipgroup;
 	}
@@ -28,9 +28,10 @@ public class SRHInsertNetworkLink extends AbstractIPv6NetworkLink {
 
 	@Override
 	public void sendPacket(IPv6Packet pack, IPv6Address inet6Address) throws IOException {
-		getSrv6Router().insertSRHeader( pack);
+		if(getSrv6Router().insertSRHeader( pack)) {
 			getSrv6Router().insertHeaderAndRoutePacket(null, pack);
 		}
+	}
 
 	@Override
 	public List<RouteItem> getRouteItems() {

src/org/kne/cloud/network/srv6/SRv6Router.java

@@ -13,6 +13,7 @@ import java.util.List;
 import java.util.Map;
 import java.util.Set;
 import java.util.UUID;
+import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.CopyOnWriteArrayList;
 import java.util.concurrent.CopyOnWriteArraySet;
 import java.util.concurrent.TimeUnit;
@@ -51,7 +52,6 @@ import org.kne.cloud.network.klalb.ui.PerformanceStrategyItem;
 import org.kne.concurrent.HighPerformanceExecutor2;
 import org.kne.concurrent.TimeoutConcurrentHashMap;
 import org.kne.io.KNEChannels;
-import org.pcap4j.packet.IllegalRawDataException;
 
 import com.google.gson.internal.Pair;
 
@@ -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地址位数
 
@@ -204,34 +204,19 @@ public class SRv6Router {
 		srListeners.remove(listener);
 	}
 	public void sortRouteTabel() {
+		List<Map<IPv6Address, List<RouteItem>>> routeTabel1x = routeTabel1;
 
-		// 构建前缀长度索引的路由表
-		List<Map<IPv6Address, List<RouteItem>>> routeTabel1x = new ArrayList<Map<IPv6Address, List<RouteItem>>>(
-				IPv6_BITS + 1);
-		for (int i = 0; i < IPv6_BITS + 1; i++) {
-			routeTabel1x.add(new HashMap<IPv6Address, List<RouteItem>>(1024, 0.5f));
-		}
-		List<RouteItem> routeTabel0x=routeTable0;
-		// 按前缀长度组织路由表
-		for (RouteItem item : routeTabel0x) {
-			IPv6AddressGroup iag = item.getDestination();
-			Map<IPv6Address, List<RouteItem>> map = routeTabel1x.get(iag.getPrefixLength());
-			List<RouteItem> newlist = new ArrayList<RouteItem>();
-			List<RouteItem> oldlist = map.putIfAbsent(iag.getAddress(), newlist);
-			if (oldlist == null) {
-				oldlist = newlist;
-			}
-			oldlist.add(item);
-		}
 		for (Map<IPv6Address, List<RouteItem>> map : routeTabel1x) {
-			for(List<RouteItem>v:map.values()) {
+			// 使用 compute 直接替换
-				Collections.sort(v);
+			map.replaceAll((key, oldList) -> {
+				if (oldList == null || oldList.isEmpty()) {
+					return oldList;
 				}
+				List<RouteItem> newList = new ArrayList<>(oldList);
+				Collections.sort(newList);
+				return newList;
+			});
 		}
-		// System.out.println(routeTabel1x);
-		routeTabel1 = routeTabel1x; // 更新索引路由表
-		
-		
 	}
 	// 更新路由表
 	public void updateRouteTabel() {
@@ -264,7 +249,7 @@ public class SRv6Router {
 		ArrayList<Map<IPv6Address, List<RouteItem>>> routeTabel1x = new ArrayList<Map<IPv6Address, List<RouteItem>>>(
 				IPv6_BITS + 1);
 		for (int i = 0; i < IPv6_BITS + 1; i++) {
-			routeTabel1x.add(new HashMap<IPv6Address, List<RouteItem>>(1024, 0.5f));
+			routeTabel1x.add(new ConcurrentHashMap<IPv6Address, List<RouteItem>>());
 		}
 
 		// 按前缀长度组织路由表
@@ -497,7 +482,7 @@ public class SRv6Router {
 
 	// 负载均衡路由
 	private boolean routingLoadBalance(IPv6NetworkLink linkfrom, IPv6Packet iPv6Packet,
-			StringBuilder dbg, List<RouteItem> routes) throws IllegalRawDataException, IOException {
+			StringBuilder dbg, List<RouteItem> routes) throws  IOException {
 
 		if(routes==null||routes.isEmpty()) {
 			//System.out.println("Route not found:" + iPv6Packet);
@@ -526,7 +511,7 @@ public class SRv6Router {
 
 	private boolean loadBalance(IPv6NetworkLink linkfrom, IPv6Packet iPv6Packet, StringBuilder dbg,
 			List<RouteItem> routes, double cscale)
-			throws IllegalRawDataException, IOException {
+			throws  IOException {
 		// 遍历可用路由进行负载均衡
 		for (RouteItem tri : routes) {
 			IPv6NetworkLink link = tri.getDestlink();
@@ -560,7 +545,7 @@ public class SRv6Router {
 
 	// 处理数据包转发
 	private void processPacket(IPv6NetworkLink linkfrom, IPv6Packet iPv6Packet, RouteItem ri)
-			throws IllegalRawDataException, IOException {
+			throws  IOException {
 		int hop = iPv6Packet.getHopLimit();
 
 		if (ri.getDestlink().isLoopBack()) { // 环回链路处理
@@ -575,37 +560,40 @@ public class SRv6Router {
 			if (srhh != null) {
 				processSRv6Packet(linkfrom, iPv6Packet, ri, srhh); // SRv6特殊处理
 			} else {
-				sendPacketToRouteItem(iPv6Packet, ri);
+				sendPacketToRouteItem(linkfrom,iPv6Packet, ri);
 			}
 		} else { // 普通链路处理
+			if(ri.getDestlink().isFilter()){
+
+			}else {
 				// 检查重路由计数
 				int count = iPv6Packet.getRouteCounter().getAndIncrement();
-			if (count >= MAX_REROUTE_COUNT) {
+				if (count > MAX_REROUTE_COUNT) {
 					return; // 超过最大重路由次数，丢弃
 				}
 
 				if (count <= 0) {
 					hop--; // 减少TTL（非重路由时）
 				}
-
+			}
 			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 +604,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);
 						}
 					}
 				}
@@ -626,7 +614,7 @@ public class SRv6Router {
 	}
 
 	// 发送ICMP TTL超时消息
-	private void sendTTLExceedPacket(IPv6Packet iPv6Packet) throws IOException, IllegalRawDataException {
+	private void sendTTLExceedPacket(IPv6Packet iPv6Packet) throws IOException {
 		enqueuePacketSendTask(() -> {
 			// 构建IPv6数据包封装ICMP消息
 			IPv6Packet icmpv = new IPv6Packet();
@@ -648,11 +636,11 @@ public class SRv6Router {
 	// 处理SRv6数据包
 	private void processSRv6Packet(IPv6NetworkLink linkfrom, IPv6Packet iPv6Packet, RouteItem ri,
 			IPv6SegmentRoutingHeader srhh)
-			throws IllegalRawDataException, IOException {
+			throws  IOException {
 
 	
 		if (srhh.getSegmentsLeft() <= 0) { // 所有段已处理完毕
-			sendPacketToRouteItem(iPv6Packet, ri);
+			sendPacketToRouteItem(linkfrom,iPv6Packet, ri);
 		} else {
 			if (iPv6Packet.getRouteCounter().get()<=0) {
 				// 重路由不处理
@@ -904,7 +892,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() {

src/org/kne/concurrent/DisruptorExecutor.java

@@ -1,114 +0,0 @@
-package org.kne.concurrent;
-
-import java.util.concurrent.Executor;
-import java.util.concurrent.ExecutorService;
-import java.util.concurrent.Executors;
-import java.util.concurrent.ThreadFactory;
-import java.util.concurrent.atomic.AtomicLong;
-import java.util.concurrent.atomic.AtomicReference;
-
-import com.lmax.disruptor.BlockingWaitStrategy;
-import com.lmax.disruptor.EventFactory;
-import com.lmax.disruptor.EventHandler;
-import com.lmax.disruptor.EventTranslator;
-import com.lmax.disruptor.LiteBlockingWaitStrategy;
-import com.lmax.disruptor.RingBuffer;
-import com.lmax.disruptor.Sequence;
-import com.lmax.disruptor.SequenceBarrier;
-import com.lmax.disruptor.SleepingWaitStrategy;
-import com.lmax.disruptor.YieldingWaitStrategy;
-import com.lmax.disruptor.dsl.Disruptor;
-import com.lmax.disruptor.dsl.ProducerType;
-
-
-public class DisruptorExecutor implements Executor{
-	
-	public static final DisruptorExecutor defaultExecutor=new DisruptorExecutor();
-	private RunnableHandler[]rh;
-	
-	public DisruptorExecutor() {
-		this(Runtime.getRuntime().availableProcessors(),1024);
-	}
-	public DisruptorExecutor(int threadcount,int bufferSize) {
-		this(threadcount,bufferSize,Executors.defaultThreadFactory());
-		//this(threadcount,bufferSize,Thread.ofVirtual().factory());
-	}
-	public DisruptorExecutor(int threadcount,int bufferSize,ThreadFactory th) {
-
-		
-		
-		rh=new RunnableHandler[threadcount];
-		
-		for (int i = 0; i < rh.length; i++) {
-			rh[i]=new RunnableHandler(bufferSize,th);
-		}
-		
-		
-	}
-	
-	private class RunnableHandler implements EventHandler<AtomicReference<Runnable>>{
-
-		private Disruptor<AtomicReference<Runnable>>ring;
-		    public RunnableHandler(int bufferSize,ThreadFactory tf)
-		    {ring=new Disruptor<AtomicReference<Runnable>>(new EventFactory<AtomicReference<Runnable>>() {
-
-			@Override
-			public AtomicReference<Runnable> newInstance() {
-				return new AtomicReference<Runnable>();
-			}
-		}, bufferSize,tf,ProducerType.MULTI, new SleepingWaitStrategy());
-		    	ring.handleEventsWith(this);
-		ring.start();
-		    }
-
-		@Override
-		public void onEvent(AtomicReference<Runnable> arg0, long sequence, boolean onEndOfBatch) throws Exception {
-			//System.out.println(sequence);
-			Runnable r =arg0.get();
-			if(r!=null)
-			try {
-			r.run();
-			}catch(Throwable th) {
-				th.printStackTrace();
-		}
-			 
-		}
-		
-	}
-	/*
-	 * public final class MyHandler implements EventHandler<ValueEvent>
-{
-   
-    public void onEvent(final ValueEvent entry, final long sequence, final boolean onEndOfBatch)
-    {
-        if ((sequence % numberOfConsumers) == ordinal)
-        {
-            // Process the event
-        }
-    }
-}
-	 */
-	private AtomicLong al=new AtomicLong();
-	@Override
-	public void execute(Runnable command) {
-		int v=(int) (al.getAndIncrement()%rh.length);
-		rh[v].ring.publishEvent(new EventTranslator<AtomicReference<Runnable>>() {
-			
-			@Override
-			public void translateTo(AtomicReference<Runnable> arg0, long arg1) {
-				arg0.set(command);
-				//arg0.compareAndSet(null, command);
-			}
-		});
-	}
-	public static void main(String[] args) throws InterruptedException {
-		DisruptorExecutor dre=new DisruptorExecutor();
-		for (int i = 0; i < 20; i++) {
-		dre.execute(()->{
-			System.out.println("run");
-		});
-		System.out.println("exc");
-		}
-		Thread.sleep(1000);
-	}
-}