减小体积

修复拥塞算法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>

.idea/vcs.xml

@@ -2,5 +2,6 @@
 <project version="4">
   <component name="VcsDirectoryMappings">
     <mapping directory="" vcs="Git" />
+    <mapping directory="$PROJECT_DIR$/dashboard" vcs="Git" />
   </component>
 </project>

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/AsynchronousDatagramChannel.java

@@ -0,0 +1,277 @@
+package org.kne.cloud.network;
+
+import java.io.IOException;
+import java.net.DatagramSocket;
+import java.net.InetAddress;
+import java.net.InetSocketAddress;
+import java.net.NetworkInterface;
+import java.net.SocketAddress;
+import java.net.SocketOption;
+import java.nio.ByteBuffer;
+import java.nio.channels.ClosedChannelException;
+import java.nio.channels.DatagramChannel;
+import java.nio.channels.MembershipKey;
+import java.nio.channels.SelectableChannel;
+import java.nio.channels.SelectionKey;
+import java.nio.channels.Selector;
+import java.nio.channels.spi.SelectorProvider;
+import java.util.Set;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+/**
+ * 自制的 UDP AIO 通道，内部使用线程池模拟异步收包。
+ * 用户只需设置回调，无需自己管理线程。
+ * 
+ * 完整委托了 DatagramChannel 的所有公共方法。
+ */
+public class AsynchronousDatagramChannel implements AutoCloseable {
+
+    private final DatagramChannel delegate;
+    private final ExecutorService workerPool;
+    private final AtomicBoolean running = new AtomicBoolean(true);
+    private volatile ReceiveHandler handler;
+    private final int bufferSize;
+	private int threadCount;
+
+    // ==================== 回调接口 ====================
+
+    @FunctionalInterface
+    public interface ReceiveHandler {
+        void onReceive(SocketAddress from, ByteBuffer data);
+    }
+
+    // ==================== 构造方法 ====================
+
+    public AsynchronousDatagramChannel() throws IOException {
+        this(65535, Runtime.getRuntime().availableProcessors());
+    }
+
+    public AsynchronousDatagramChannel(int bufferSize, int threadCount) throws IOException {
+        this.bufferSize = bufferSize;
+        this.delegate = DatagramChannel.open();
+        delegate.configureBlocking(true);
+        this.threadCount=threadCount;
+        this.workerPool = Executors.newFixedThreadPool(threadCount, r -> {
+            Thread t = new Thread(r, "AsyncDatagramReader");
+            t.setDaemon(true);
+            return t;
+        });
+
+    }
+
+    public void startWorkers() {
+
+        for (int i = 0; i < threadCount; i++) {
+            workerPool.submit(this::readerLoop);
+        }
+    }
+    
+    // ==================== 核心方法 ====================
+
+    public void setReceiveHandler(ReceiveHandler handler) {
+        this.handler = handler;
+    }
+
+    private void readerLoop() {
+        ByteBuffer buffer = ByteBuffer.allocateDirect(bufferSize);
+        while (running.get()) {
+            try {
+                buffer.clear();
+                SocketAddress from = delegate.receive(buffer);
+                if (from == null) {
+                    continue;
+                }
+                buffer.flip();
+                ReceiveHandler localHandler = handler;
+                if (localHandler != null) {
+                    localHandler.onReceive(from, buffer);
+                }
+            } catch (IOException e) {
+                if (running.get()) {
+                    e.printStackTrace();
+                }
+                break;
+            }
+        }
+    }
+
+    public int send(ByteBuffer data, SocketAddress target) throws IOException {
+        return delegate.send(data, target);
+    }
+
+    // ==================== DatagramChannel 委托方法 ====================
+
+    // --- 连接相关 ---
+
+    public DatagramChannel connect(SocketAddress remote) throws IOException {
+        return delegate.connect(remote);
+    }
+
+    public DatagramChannel disconnect() throws IOException {
+        return delegate.disconnect();
+    }
+
+    public boolean isConnected() {
+        return delegate.isConnected();
+    }
+
+    // --- 绑定相关 ---
+
+    public DatagramChannel bind(SocketAddress local) throws IOException {
+        return delegate.bind(local);
+    }
+
+    public SocketAddress getLocalAddress() throws IOException {
+        return delegate.getLocalAddress();
+    }
+
+    public SocketAddress getRemoteAddress() throws IOException {
+        return delegate.getRemoteAddress();
+    }
+
+    // --- 读写相关 ---
+
+    public int read(ByteBuffer dst) throws IOException {
+        return delegate.read(dst);
+    }
+
+    public long read(ByteBuffer[] dsts) throws IOException {
+        return delegate.read(dsts);
+    }
+
+    public long read(ByteBuffer[] dsts, int offset, int length) throws IOException {
+        return delegate.read(dsts, offset, length);
+    }
+
+    public int write(ByteBuffer src) throws IOException {
+        return delegate.write(src);
+    }
+
+    public long write(ByteBuffer[] srcs) throws IOException {
+        return delegate.write(srcs);
+    }
+
+    public long write(ByteBuffer[] srcs, int offset, int length) throws IOException {
+        return delegate.write(srcs, offset, length);
+    }
+
+    // --- 接收相关 ---
+
+    public SocketAddress receive(ByteBuffer dst) throws IOException {
+        return delegate.receive(dst);
+    }
+
+    // --- 多播 (Multicast) ---
+
+    public MembershipKey join(InetAddress group, NetworkInterface interf) throws IOException {
+        return delegate.join(group, interf);
+    }
+
+    public MembershipKey join(InetAddress group, NetworkInterface interf, InetAddress source) throws IOException {
+        return delegate.join(group, interf, source);
+    }
+
+    // --- 选项相关 ---
+
+    public <T> DatagramChannel setOption(SocketOption<T> name, T value) throws IOException {
+        return delegate.setOption(name, value);
+    }
+
+    public <T> T getOption(SocketOption<T> name) throws IOException {
+        return delegate.getOption(name);
+    }
+
+    public Set<SocketOption<?>> supportedOptions() {
+        return delegate.supportedOptions();
+    }
+
+    // --- 阻塞模式 ---
+
+
+    public boolean isBlocking() {
+        return delegate.isBlocking();
+    }
+
+    // --- 选择器相关 ---
+
+    public SelectableChannel configureBlocking(boolean block) throws IOException {
+        return delegate.configureBlocking(block);
+    }
+
+    public SelectorProvider provider() {
+        return delegate.provider();
+    }
+
+    public int validOps() {
+        return delegate.validOps();
+    }
+
+    public SelectionKey keyFor(Selector sel) {
+        return delegate.keyFor(sel);
+    }
+
+    public SelectionKey register(Selector sel, int ops) throws ClosedChannelException {
+        return delegate.register(sel, ops);
+    }
+
+    public SelectionKey register(Selector sel, int ops, Object att) throws ClosedChannelException {
+        return delegate.register(sel, ops, att);
+    }
+
+    // --- 状态查询 ---
+
+    public boolean isOpen() {
+        return delegate.isOpen() && running.get();
+    }
+
+    public DatagramSocket socket() {
+        return delegate.socket();
+    }
+
+    // ==================== 关闭方法 ====================
+
+    @Override
+    public void close() throws Exception {
+        running.set(false);
+        workerPool.shutdownNow();
+        delegate.close();
+    }
+
+    // ==================== 静态工厂方法 ====================
+
+    public static AsynchronousDatagramChannel open() throws IOException {
+        return new AsynchronousDatagramChannel();
+    }
+
+    public static AsynchronousDatagramChannel open(int bufferSize) throws IOException {
+        return new AsynchronousDatagramChannel(bufferSize, Runtime.getRuntime().availableProcessors());
+    }
+
+    public static AsynchronousDatagramChannel open(int bufferSize, int threadCount) throws IOException {
+        return new AsynchronousDatagramChannel(bufferSize, threadCount);
+    }
+
+    // ==================== 测试示例 ====================
+
+    public static void main(String[] args) throws Exception {
+        int port = 4567;
+        AsynchronousDatagramChannel channel = AsynchronousDatagramChannel.open();
+        channel.bind(new InetSocketAddress(port));
+
+        channel.setReceiveHandler((from, data) -> {
+            byte[] bytes = new byte[data.remaining()];
+            data.get(bytes);
+            String msg = new String(bytes);
+            System.out.println("Received from " + from + ": " + msg);
+        });
+
+        System.out.println("Echo server running on port " + port);
+        System.out.println("Press Enter to stop...");
+        System.in.read();
+
+        channel.close();
+        System.out.println("Server stopped.");
+    }
+}

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() {
@@ -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(), 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) {
@@ -73,65 +74,53 @@ public class Vegas2CongestionAlgorithm implements CongestionAlgorithm,DetnetCong
             }
         }
 
-        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);
-        	}
+        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;
 
 
-        	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;
-            
-            
-            
-            
+        }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,12 +15,19 @@ 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() {
 		return srv6Router;
@@ -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

@@ -17,14 +17,8 @@ import java.util.Iterator;
 import java.util.List;
 import java.util.Map;
 import java.util.Set;
-import java.util.Timer;
-import java.util.TimerTask;
 import java.util.UUID;
-import java.util.concurrent.ConcurrentHashMap;
-import java.util.concurrent.CopyOnWriteArrayList;
-import java.util.concurrent.Executors;
-import java.util.concurrent.ScheduledExecutorService;
-import java.util.concurrent.ThreadFactory;
+import java.util.concurrent.*;
 import java.util.concurrent.locks.Lock;
 import java.util.concurrent.locks.ReadWriteLock;
 import java.util.concurrent.locks.ReentrantLock;
@@ -67,7 +61,7 @@ public class KLALBController {
 	public HighAccuracyClock getClock() {
 		return clock;
 	}
-	private final long TIME_WINDOW=2000000000L;
+	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));
@@ -104,23 +98,9 @@ public class KLALBController {
 	public List<InetAddress> getDnsAddresses() {
 		return dnsAddresses;
 	}
-	private ScheduledExecutorService scheduleTimer=Executors.newScheduledThreadPool(Runtime.getRuntime().availableProcessors(),new ThreadFactory() {
-		
-		@Override
-		public Thread newThread(Runnable r) {
-			return ThreadTool.makeVDaemonThreadIfSupport("计时器线程", r);
-		}
-	});
-
-	public ScheduledExecutorService getScheduleTimer() {
-		return scheduleTimer;
-	}
-
-	private Timer twk = new Timer("网卡检测扫描计时器", true);
 
 
-
-	private TimerTask tsk1 = new TimerTask() {
+	private Runnable tsk1 = new Runnable() {
 
 		@Override
 		public void run() {
@@ -306,7 +286,7 @@ public class KLALBController {
 		return srv6Router.getLocator().getAddress().toInet6Address().equals(inetAddress);
 	}
 
-	private TimerTask tsk2 = new TimerTask() {
+	private Runnable tsk2 = new Runnable() {
 
 		@Override
 		public void run() {
@@ -686,7 +666,7 @@ public class KLALBController {
 		context.syncToSystem();
 		// System.out.println(context);
 		try {
-			NTPv4Protocol nvc = new NTPv4Protocol(context,
+			/*NTPv4Protocol nvc = new NTPv4Protocol(context,
 					new MultiProtocolSocketAddress(iproxyname + "datagram","0.0.0.0",123));
 			srv6Router.addSRv6RouterListener(new SRv6RouterListener() {
 
@@ -718,7 +698,7 @@ public class KLALBController {
 						return true;
 					});
 				}
-			});
+			});*/
 			nvc2 = new NTPv4Protocol(context, new MultiProtocolSocketAddress("udp","0.0.0.0",0));
 
 			System.out.println("  Loaded: NTP Module");
@@ -734,8 +714,8 @@ public class KLALBController {
 	private void loadController() {
 		this.apiServer = new KLALBRoutingProtocolAPIServer(routingProtocol, this);
 		this.apiClient = new KLALBRoutingProtocolAPIClient(routingProtocol);
-		twk.schedule(tsk1, 5000, 5000);
-		twk.schedule(tsk2, 5000, 5000);
+		KLALBScheduler.getScheduler().scheduleAtFixedRate(tsk1, 5000, 5000,TimeUnit.MILLISECONDS);
+		KLALBScheduler.getScheduler().scheduleAtFixedRate(tsk2, 5000, 5000,TimeUnit.MILLISECONDS);
 	}
 
 	public KLALBController() {
@@ -753,17 +733,12 @@ public class KLALBController {
 	}
 
 	protected void initKLALB(boolean enableVirtualAdapter, IPv6AddressGroup selfg) {
-		ThreadTool.makeVDaemonThread("线路性能采样线程", ()->{
-			while(true) {
+		KLALBScheduler.getScheduler().scheduleWithFixedDelay( ()->{
 			linkMonitor.update();
-			datatMonitor.update();
-			try {
-				Thread.sleep(1);
-			} catch (InterruptedException e) {
-				e.printStackTrace();
-			}
-			}
-		}).start();
+		},1,1,TimeUnit.MILLISECONDS);
+		KLALBScheduler.getScheduler().scheduleWithFixedDelay( ()->{
+		datatMonitor.update();
+		},1,1,TimeUnit.MILLISECONDS);
 		loadSRv6ProtocolStack(selfg, enableVirtualAdapter);
 		loadController();
 	}
@@ -856,9 +831,8 @@ public class KLALBController {
 		return srv6Router;
 	}
 
-	private Timer trtr = new Timer("路由表刷新计时器", true);
 	{
-		trtr.scheduleAtFixedRate(new TimerTask() {
+		KLALBScheduler.getScheduler().scheduleAtFixedRate(new Runnable() {
 
 			@Override
 			public void run() {
@@ -869,9 +843,9 @@ public class KLALBController {
 					srv6Router.updateRouteTabel();
 				}
 			}
-		}, 100, 100);
+		}, 100, 100, TimeUnit.MILLISECONDS);
 
-		trtr.scheduleAtFixedRate(new TimerTask() {
+		KLALBScheduler.getScheduler().scheduleAtFixedRate(new Runnable() {
 
 			@Override
 			public void run() {
@@ -880,7 +854,7 @@ public class KLALBController {
 				}
 
 			}
-		}, 10, 10);
+		}, 1, 1,TimeUnit.MILLISECONDS);
 	}
 
 	public IPv6Packet createPacketToAddress(IPv6Address addr, int flowlabel, IPv6Packet.IPv6Payload packet) {

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

@@ -13,6 +13,8 @@ import java.util.Iterator;
 import java.util.List;
 import java.util.Queue;
 import java.util.UUID;
+import java.util.concurrent.ScheduledFuture;
+import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.locks.Condition;
 import java.util.concurrent.locks.Lock;
@@ -31,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;
@@ -285,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) {
@@ -570,20 +563,17 @@ public class KLALBRemoteLink extends AbstractControlledIPv6NetworkLink implement
 	private ThreadParker tpk=new ThreadParker();
 
 	private StateManager stateManager;
-	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();
 				if(status.getState()!=LinkStatus.DOWN) {
 					monitor.update();
 				}
-				try {
-					Thread.sleep(1);
-				} catch (InterruptedException e) {
-					e.printStackTrace();
-				}
 				
 			}
 			
@@ -755,8 +745,7 @@ public class KLALBRemoteLink extends AbstractControlledIPv6NetworkLink implement
 	
 	protected void startIO() {
 		stateManager=new StateManager();
-		monitorUpdater=new MonitorUpdater();
-		ThreadTool.makeVDaemonThread("线路性能采样线程", monitorUpdater).start();
+		monitorUpdater=KLALBScheduler.getScheduler().scheduleWithFixedDelay(new MonitorUpdater(),1,1, TimeUnit.MILLISECONDS);
 		ThreadTool.makeVDaemonThread("远程线路状态线程", stateManager).start();
 		
 	}
@@ -898,6 +887,8 @@ public class KLALBRemoteLink extends AbstractControlledIPv6NetworkLink implement
 						sendWindow.put(val, pack);
 					//}
 				}
+			}else {
+				System.out.println(hop);
 			}
 			
 			sendIPv6PacketToLink(pack, val);
@@ -946,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) {
@@ -975,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/KLALBScheduler.java

@@ -0,0 +1,19 @@
+package org.kne.cloud.network.klalb;
+
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+
+public class KLALBScheduler {
+    private static final ScheduledExecutorService scheduler;
+    static {
+        scheduler= Executors.newScheduledThreadPool( Runtime.getRuntime().availableProcessors(),
+                (r) -> {
+                    Thread t = new Thread(r, "KLALB-Scheduler");
+                    t.setDaemon(true);  // 守护线程，不阻止 JVM 退出
+                    return t;
+                });
+    }
+    public static ScheduledExecutorService getScheduler(){
+        return  scheduler;
+    }
+}

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,46 +118,50 @@ 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);
-		szeWrite.limit(4);
-		//szeWrite.clear();
+		// 1. 获取并清空缓存的 4 字节 buffer
+		ByteBuffer szeWrite = SIZE_WRITE_BUFFER.get();
+		szeWrite.clear();
+
+		// 2. 写入长度并翻转
 		szeWrite.putInt(kp.limit());
 		szeWrite.flip();
-		KLALBVirtualSocketChannel obj = null;
-		writableChannel.write(new ByteBuffer[] {szeWrite,kp});
+
+		// 3. 发送
+		writableChannel.write(new ByteBuffer[] {szeWrite, kp});
 		checkflush();
-		
-		
 	}
 
-	
-
-
 	@Override
 	public ByteBuffer readPacket() throws IOException {
-
-		ByteBuffer szeRead=NetworkPacket.bufferAllocator.allocate(4);
-		timeoutTimer=System.nanoTime();
-		timerenabled=true;
-		ByteBuffer kp;
+		// 1. 获取并清空缓存的 4 字节 buffer
+		ByteBuffer szeRead = SIZE_READ_BUFFER.get();
 		szeRead.clear();
 		szeRead.limit(4);
-		//szeRead.clear();
+
+		timeoutTimer = System.nanoTime();
+		timerenabled = true;
+		ByteBuffer kp;
+
 		try {
-			KNEChannels.readFully(readableChannel,szeRead);
+			// 2. 读取长度头
+			KNEChannels.readFully(readableChannel, szeRead);
 			szeRead.flip();
-			int size=szeRead.getInt(0);
-			kp=NetworkPacket.bufferAllocator.allocate(size);
+			int size = szeRead.getInt(0);
+
+			// 3. 分配真正的数据区（这里如果是高频，建议后续也可以池化，不过现在是按需大小分配）
+			kp = NetworkPacket.bufferAllocator.allocate(size);
 			kp.limit(size);
 			KNEChannels.readFully(readableChannel, kp);
 			kp.flip();
-		return kp;
-		}finally {
-			timerenabled=false;
+			return kp;
+		} finally {
+			timerenabled = false;
 		}
 	}
 	

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

@@ -7,8 +7,8 @@ import java.io.IOException;
 import java.net.*;
 import java.util.ArrayList;
 import java.util.List;
-import java.util.Timer;
-import java.util.TimerTask;
+import java.util.concurrent.ScheduledFuture;
+import java.util.concurrent.TimeUnit;
 import java.util.function.Consumer;
 import javax.imageio.ImageIO;
 import javax.swing.*;
@@ -37,8 +37,7 @@ import org.kne.ui.YScrollPane;
 public class KLALBStateGUI3 extends XFrame {
 
     // ==================== 定时任务相关 ====================
-    private TimerTask tsk, tsk2, tsk3, tsk4, tsk5; // 各种定时任务
-    private Timer t, t2, t3; // 定时器
+    private ScheduledFuture<?> tsk, tsk2, tsk3, tsk4, tsk5; // 各种定时任务
 
     // ==================== 系统托盘相关 ====================
     private SystemTray st;  // 系统托盘
@@ -705,10 +704,6 @@ public class KLALBStateGUI3 extends XFrame {
      * 初始化定时器
      */
     private void initTimers(KLALBController kc) {
-        t3 = new Timer("监视器刷新线程", true);
-        t2 = new Timer("仪表盘刷新线程", true);
-        t = new Timer("状态刷新线程", true);
-        
         // 创建刷新任务
         createRefreshTask(kc, ysp);
     }
@@ -1322,9 +1317,9 @@ public class KLALBStateGUI3 extends XFrame {
     private void createRefreshTask(KLALBController kc, YScrollPane ysp) {
         // 线路状态刷新任务
         if (tsk != null) {
-            tsk.cancel();
+            tsk.cancel(false);
         }
-        tsk = new TimerTask() {
+        tsk =KLALBScheduler.getScheduler().scheduleWithFixedDelay( new Runnable() {
             @Override
             public void run() {
                 if (isVisible()) {
@@ -1382,28 +1377,26 @@ public class KLALBStateGUI3 extends XFrame {
                     ysp.updateScrool();
                 }
             }
-        };
-        t.scheduleAtFixedRate(tsk, 1000, 1000);
+        }, 1000, 1000, TimeUnit.MILLISECONDS);
 
         // 仪表盘数据刷新任务
         if (tsk2 != null) {
-            tsk2.cancel();
+            tsk2.cancel(false);
         }
-        tsk2 = new TimerTask() {
+        tsk2 =KLALBScheduler.getScheduler().scheduleWithFixedDelay(new Runnable() {
             @Override
             public void run() {
                 if (isVisible()) {
                     updateDashboardData(kc);
                 }
             }
-        };
-        t2.scheduleAtFixedRate(tsk2, 100, 100);
+        }, 100, 100,TimeUnit.MILLISECONDS);
 
         // 网络图刷新任务
         if (tsk5 != null) {
-            tsk5.cancel();
+            tsk5.cancel(false);
         }
-        tsk5 = new TimerTask() {
+        tsk5 = KLALBScheduler.getScheduler().scheduleWithFixedDelay(new Runnable() {
             @Override
             public void run() {
                 if (isVisible()) {
@@ -1415,8 +1408,7 @@ public class KLALBStateGUI3 extends XFrame {
                     graph.revalidate();
                 }
             }
-        };
-        t.scheduleAtFixedRate(tsk5, 1000, 1000);
+        }, 1000, 1000,TimeUnit.MILLISECONDS);
     }
 
     /**
@@ -1425,29 +1417,29 @@ public class KLALBStateGUI3 extends XFrame {
     private void updateDashboardData(KLALBController kc) {
         // 更新上行速度
         double outload = kc.getLinkMonitor().getOutSpeed() * 100.0 / kc.getLinkMonitor().getOutSpeedMax();
+        upSpeedText.setLabel(KLALBUtils.convertIBUint(kc.getLinkMonitor().getOutSpeed()) + "/s");
         if (Double.isFinite(outload)) {
-            upSpeedText.setLabel(KLALBUtils.convertIBUint(kc.getLinkMonitor().getOutSpeed()) + "/s");
             upSpeed.setValue(Math.min(upSpeed.getValue().doubleValue() * 0.98 + outload * 0.02, 101.0));
         }
 
         // 更新下行速度
         double inload = kc.getLinkMonitor().getInSpeed() * 100.0 / kc.getLinkMonitor().getInSpeedMax();
+        downSpeedText.setLabel(KLALBUtils.convertIBUint(kc.getLinkMonitor().getInSpeed()) + "/s");
         if (Double.isFinite(inload)) {
-            downSpeedText.setLabel(KLALBUtils.convertIBUint(kc.getLinkMonitor().getInSpeed()) + "/s");
             downSpeed.setValue(Math.min(downSpeed.getValue().doubleValue() * 0.98 + inload * 0.02, 101.0));
         }
 
         // 更新上行包速率
         double outPPSload = kc.getLinkMonitor().getOutPPS() * 100.0 / kc.getLinkMonitor().getOutPPSMax();
+        upPPSText.setLabel(KLALBUtils.convertUintDefalut(kc.getLinkMonitor().getOutPPS()) + "PPS");
         if (Double.isFinite(outPPSload)) {
-            upPPSText.setLabel(KLALBUtils.convertUintDefalut(kc.getLinkMonitor().getOutPPS()) + "PPS");
             upPPS.setValue(Math.min(upPPS.getValue().doubleValue() * 0.98 + outPPSload * 0.02, 101.0));
         }
 
         // 更新下行包速率
         double inPPSload = kc.getLinkMonitor().getInPPS() * 100.0 / kc.getLinkMonitor().getInPPSMax();
+        downPPSText.setLabel(KLALBUtils.convertUintDefalut(kc.getLinkMonitor().getInPPS()) + "PPS");
         if (Double.isFinite(inPPSload)) {
-            downPPSText.setLabel(KLALBUtils.convertUintDefalut(kc.getLinkMonitor().getInPPS()) + "PPS");
             downPPS.setValue(Math.min(downPPS.getValue().doubleValue() * 0.98 + inPPSload * 0.02, 101.0));
         }
 
@@ -1479,29 +1471,29 @@ public class KLALBStateGUI3 extends XFrame {
        if(router!=null) {
     	   // 更新背板延迟
            double backplaneTimev = router.getBackplaneTime()* 100.0/100000.0;
+           backplaneDelayText.setLabel(KLALBUtils.convertUintNanoDefalut(router.getBackplaneTime()) + "s");
            if (Double.isFinite(backplaneTimev)) {
-               backplaneDelayText.setLabel(KLALBUtils.convertUintNanoDefalut(router.getBackplaneTime()) + "s");
                backplaneDelay.setValue(Math.min(backplaneDelay.getValue().doubleValue() * 0.98 + backplaneTimev * 0.02, 101.0));
            }
 
            // 更新背板转发率
            double backplanePPSv = router.getBackplanePPS() * 100.0 / router.getBackplanePPSMax();
+           backplanePPSText.setLabel(KLALBUtils.convertUintDefalut(router.getBackplanePPS()) + "PPS");
            if (Double.isFinite(backplanePPSv)) {
-               backplanePPSText.setLabel(KLALBUtils.convertUintDefalut(router.getBackplanePPS()) + "PPS");
                backplanePPS.setValue(Math.min(backplanePPS.getValue().doubleValue() * 0.98 + backplanePPSv * 0.02, 101.0));
            }
            
            // 更新背板ECN率
            double backplaneECNv = router.getECNRate() * 100.0 ;
+           backplaneECNRateText.setLabel(String.format("%.1f",backplaneECNv) + "%");
            if (Double.isFinite(backplaneECNv)) {
-               backplaneECNRateText.setLabel(String.format("%.1f",backplaneECNv) + "%");
                backplaneECNRate.setValue(Math.min(backplaneECNRate.getValue().doubleValue() * 0.98 + backplaneECNv * 0.02, 101.0));
            }
            
         // 更新背板ECN率
            double backplaneLossv = router.getLossRate() * 100.0 ;
+           backplaneLossRateText.setLabel(String.format("%.1f",backplaneLossv) + "%");
            if (Double.isFinite(backplaneLossv)) {
-               backplaneLossRateText.setLabel(String.format("%.1f",backplaneLossv) + "%");
                backplaneLossRate.setValue(Math.min(backplaneLossRate.getValue().doubleValue() * 0.98 + backplaneLossv * 0.02, 101.0));
            }
        }
@@ -1764,7 +1756,7 @@ public class KLALBStateGUI3 extends XFrame {
     public void close() {
         setVisible(false);
         if (tsk != null) {
-            tsk.cancel();
+            tsk.cancel(false);
         }
         if (st != null) {
             st.remove(ti);

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

@@ -4,8 +4,7 @@ import java.awt.Color;
 import java.awt.Dimension;
 import java.awt.GridLayout;
 import java.util.Date;
-import java.util.Timer;
-import java.util.TimerTask;
+import java.util.Scanner;
 import java.util.UUID;
 
 import javax.swing.JPanel;
@@ -26,6 +25,7 @@ import org.kne.cloud.network.ipv6.IPv6Address;
 import org.kne.cloud.network.ipv6.IPv6AddressGroup;
 import org.kne.cloud.network.ipv6.IPv6Packet;
 import org.kne.cloud.network.klalb.KLALBRemoteLink;
+import org.kne.cloud.network.klalb.KLALBScheduler;
 import org.kne.cloud.network.monitor.LinkStatus;
 import org.kne.cloud.network.monitor.TimestampMonitorValue;
 import org.kne.math.Long128;
@@ -38,21 +38,18 @@ import javax.swing.ImageIcon;
 import javax.swing.JLabel;
 import java.awt.event.WindowEvent;
 import java.awt.event.WindowListener;
+import java.util.concurrent.ScheduledFuture;
+import java.util.concurrent.TimeUnit;
 import javax.swing.JSlider;
 
 public class LineMonitorGUI extends XFrame {
-	private Timer timer;
-	private Timer timer1;
-	private Timer timer2;
-	private Timer timer3;
-	private Timer timer4;
-	private TimerTask tsk3;
+	private ScheduledFuture<?> tsk3;
 
-	private TimerTask tsk4, tsk5,tsk6,tsk7;
-	private TimerTask d1;
-	private TimerTask d2;
-	private TimerTask d3;
-	private TimerTask d4;
+	private ScheduledFuture<?> tsk4, tsk5,tsk6,tsk7;
+	private ScheduledFuture<?> d1;
+	private ScheduledFuture<?> d2;
+	private ScheduledFuture<?> d3;
+	private ScheduledFuture<?> d4;
 
 	private TimeSeries spdup = new TimeSeries(UIEnv.getRsb().getString("uploadspeed"));
 	private TimeSeries spddown = new TimeSeries(UIEnv.getRsb().getString("downloadspeed"));
@@ -72,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;
@@ -173,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);
@@ -180,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));
@@ -371,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);
+				}
 			}
 			}
 		}
@@ -386,49 +390,45 @@ public class LineMonitorGUI extends XFrame {
 	@Override
 	public void setVisible(boolean b) {
 		if (b) {
-			if (timer == null) {
-				timer = new Timer();
-				timer1 = new Timer();
-				timer2 = new Timer();
-				timer3 = new Timer();
-				timer4 = new Timer();
-				tsk3 = new TimerTask() {
+			if (tsk3 == null) {
+
+				tsk3 = KLALBScheduler.getScheduler().scheduleWithFixedDelay(new Runnable() {
 
 					@Override
 					public void run() {
 						updateTraffic();
 					}
-				};
-				tsk4 = new TimerTask() {
+				}, 25, 25, TimeUnit.MILLISECONDS);
+				tsk4 = KLALBScheduler.getScheduler().scheduleWithFixedDelay(new Runnable() {
 
 					@Override
 					public void run() {
 						recordSpeedData();
 					}
-				};
+				}, 10, 10, TimeUnit.MILLISECONDS);;
 
-				tsk5 = new TimerTask() {
+				tsk5 =KLALBScheduler.getScheduler().scheduleWithFixedDelay( new Runnable() {
 
 					@Override
 					public void run() {
 						recordDelayData();
 					}
-				};
-				tsk6 = new TimerTask() {
+				}, 10, 10, TimeUnit.MILLISECONDS);
+				tsk6 =KLALBScheduler.getScheduler().scheduleWithFixedDelay( new Runnable() {
 
 					@Override
 					public void run() {
 						recordBufferData();
 					}
-				};
-				tsk7 = new TimerTask() {
+				}, 10, 10, TimeUnit.MILLISECONDS);
+				tsk7 =	KLALBScheduler.getScheduler().scheduleWithFixedDelay( new Runnable() {
 
 					@Override
 					public void run() {
 						recordCongestionData();
 					}
-				};
-				d1 = new TimerTask() {
+				}, 25, 25, TimeUnit.MILLISECONDS);
+				d1 = KLALBScheduler.getScheduler().scheduleWithFixedDelay(new Runnable() {
 
 					@Override
 					public void run() {
@@ -440,8 +440,8 @@ public class LineMonitorGUI extends XFrame {
 							}
 						}
 					}
-				};
-				d2 = new TimerTask() {
+				} ,25, 25, TimeUnit.MILLISECONDS);
+				d2 =	KLALBScheduler.getScheduler().scheduleWithFixedDelay( new Runnable() {
 
 					@Override
 					public void run() {
@@ -454,8 +454,8 @@ public class LineMonitorGUI extends XFrame {
 
 						}
 					}
-				};
-				d3 = new TimerTask() {
+				}, 25, 25, TimeUnit.MILLISECONDS);
+				d3 = 	KLALBScheduler.getScheduler().scheduleWithFixedDelay(new Runnable() {
 
 					@Override
 					public void run() {
@@ -468,8 +468,8 @@ public class LineMonitorGUI extends XFrame {
 
 						}
 					}
-				};
-				d4 = new TimerTask() {
+				}, 25, 25, TimeUnit.MILLISECONDS);
+				d4 =KLALBScheduler.getScheduler().scheduleWithFixedDelay( new Runnable() {
 
 					@Override
 					public void run() {
@@ -482,51 +482,32 @@ public class LineMonitorGUI extends XFrame {
 
 						}
 					}
-				};
-				timer.scheduleAtFixedRate(tsk3, 25, 25);
-				timer1.scheduleAtFixedRate(tsk4, 10, 10);
-				timer1.scheduleAtFixedRate(d1, 25, 25);
-				timer2.scheduleAtFixedRate(tsk5, 10, 10);
-				timer2.scheduleAtFixedRate(d2, 25, 25);
-				timer3.scheduleAtFixedRate(tsk6, 10, 10);
-				timer3.scheduleAtFixedRate(d3, 25, 25);
-				timer4.scheduleAtFixedRate(tsk7, 25, 25);
-				timer4.scheduleAtFixedRate(d4, 25, 25);
+				}, 25, 25, TimeUnit.MILLISECONDS);
+
+
+
 			}
 
 		} else {
 			if (tsk3 != null)
-				tsk3.cancel();
+				tsk3.cancel(false);
 			if (tsk4 != null)
-				tsk4.cancel();
+				tsk4.cancel(false);
 			if (tsk5 != null)
-				tsk5.cancel();
+				tsk5.cancel(false);
 			if (tsk6 != null)
-				tsk6.cancel();
+				tsk6.cancel(false);
 			if (tsk7 != null)
-				tsk7.cancel();
+				tsk7.cancel(false);
 			if (d1 != null)
-				d1.cancel();
+				d1.cancel(false);
 			if (d2 != null)
-				d2.cancel();
+				d2.cancel(false);
 			if (d3 != null)
-				d3.cancel();
+				d3.cancel(false);
 			if (d4 != null)
-				d4.cancel();
-			if (timer != null)
-				timer.cancel();
-			if (timer1 != null)
-				timer1.cancel();
-
-			if (timer2 != null)
-				timer2.cancel();
-
-			if (timer3 != null)
-				timer3.cancel();
-
-			if (timer4 != null)
-				timer4.cancel();
-			timer = null;
+				d4.cancel(false);
+			tsk3=null;
 		}
 		super.setVisible(b);
 	}

src/org/kne/cloud/network/kltp/KLTPOutputStream.java

@@ -3,20 +3,18 @@ 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;
 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;
@@ -238,23 +236,27 @@ 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);
 						}
 					}  
 				};
-				tt=controller.getScheduleTimer().scheduleAtFixedRate (r, delaytime, delaytime,TimeUnit.NANOSECONDS);
+				tt= KLALBScheduler.getScheduler().scheduleWithFixedDelay(r, delaytime, delaytime,TimeUnit.NANOSECONDS);
 				}
 				
 				IOException ioex=ioe.get();
@@ -271,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);
@@ -353,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;
@@ -67,6 +67,11 @@ public class ArrayListTimestampMonitor<K> implements TimestampMonitor<K> {
 		 return recordPacket(key,clock.getCurrentTimeNanos(),count,length);
 	 }
 
+
+	public boolean recordPacket(K key, int count, int length, long delay, long delayMin) {
+		return recordPacket(key, clock.getCurrentTimeNanos(),count, length, delay,delayMin);
+	}
+
 	public boolean recordPacket(K key,int count, int length, long delay) {
 		return recordPacket(key, clock.getCurrentTimeNanos(),count, length, delay);
 	}
@@ -75,17 +80,22 @@ public class ArrayListTimestampMonitor<K> implements TimestampMonitor<K> {
 		return recordPacket(key, timestamp,count, length, Long.MAX_VALUE / 1024);
 	}
 
+	@Override
+	public boolean recordPacket(K key, Long128 timestamp, int count, int length, long dsndDelay) {
+		return recordPacket(key, timestamp, count, length, dsndDelay,dsndDelay);
+	}
+
 
 	/**
 	 * 记录数据包 - 使用插入排序保持列表有序，添加时间戳验证
 	 */
-	public boolean recordPacket(K key, Long128 timestamp,int count, int length, long delay) {
+	public boolean recordPacket(K key, Long128 timestamp,int count, int length, long delay, long dsndDelayMin) {
 		// 验证时间戳的合理性
 		validateTimestamp(timestamp);
 
 		// 快速路径：尝试sequence=0
 		TimestampMonitorKey<K> monitorKey = new TimestampMonitorKey<>(key, 0);
-		TimestampMonitorValue monitorValue = new TimestampMonitorValue(monitorKey, timestamp,count, length, delay);
+		TimestampMonitorValue monitorValue = new TimestampMonitorValue(monitorKey, timestamp,count, length, delay,dsndDelayMin);
 		autoCleanup(cleanupThreshold);
 
 		rwLock.writeLock().lock();
@@ -461,6 +471,48 @@ public class ArrayListTimestampMonitor<K> implements TimestampMonitor<K> {
 		}
 	}
 
+
+
+	/**
+	 * 从已排序的列表中计算数据量 - 优化版本，只需要startTime
+	 */
+	private long calculateMinDelayFromSortedList(List<TimestampMonitorValue> sortedList, Long128 startTime) {
+		if (sortedList.isEmpty()) {
+			return 0;
+		}
+
+		// 使用二分查找找到第一个 >= startTime 的元素位置
+		int startIndex = findFirstValidIndexFromSorted(sortedList, startTime);
+
+		if (startIndex >= sortedList.size()) {
+			return 0;
+		}
+
+		// 计算从startIndex到末尾的所有数据量（因为数据包不会来自未来）
+		long sum = Long.MAX_VALUE/1024;
+		for (int i = startIndex; i < sortedList.size(); i++) {
+			sum =Math.min(sortedList.get(i).getDelayMin(),sum) ;
+		}
+
+		return sum;
+	}
+
+	@Override
+	public long calculateMinDelay(long timeWindowNs){
+		return calculateMinDelay(clock.getCurrentTimeNanos(),timeWindowNs);
+	}
+
+	@Override
+	public long calculateMinDelay(Long128 currentTime, long timeWindowNs){ 
+		Long128 startTime = currentTime.subtract(Long128.valueOf(timeWindowNs));
+		rwLock.readLock().lock();
+		try {
+			return calculateMinDelayFromSortedList(entries, startTime);
+		} finally {
+			rwLock.readLock().unlock();
+		}
+	}
+
 	private volatile TimestampMonitorValue recent=null;
 	@Override
 	public TimestampMonitorValue getRecentDelayEntry() {

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

@@ -76,15 +76,22 @@ private String name;
 			return recordPacket(key,clock.getCurrentTimeNanos(),count,length,delay);
 		}
 
+    @Override
+    public boolean recordPacket(K key, int count, int length, long delay, long delayMin) {
+        return recordPacket(key,clock.getCurrentTimeNanos(),count,length,delay,delayMin);
+    }
 
+    public boolean recordPacket(K key, Long128 timestamp,int count, int length, long delay) {
+         return recordPacket(key,timestamp,count,length,delay,delay);
+    }
     /**
      * 超高性能记录数据包 - 针对路由器优化
      */
-    public boolean recordPacket(K key, Long128 timestamp,int count, int length, long delay) {
+    public boolean recordPacket(K key, Long128 timestamp,int count, int length, long delay,long delayMin) {
     	autoCleanup(cleanupThreshold);
         // 快速路径：尝试sequence=0
         TimestampMonitorKey<K> monitorKey = new TimestampMonitorKey<>(key, 0);
-        TimestampMonitorValue monitorValue = new TimestampMonitorValue(monitorKey,timestamp, count,length,delay);
+        TimestampMonitorValue monitorValue = new TimestampMonitorValue(monitorKey,timestamp, count,length,delay,delayMin);
         
         if (entries.putIfAbsent(monitorKey, monitorValue) == null) {
             totalDataSize.add(length);
@@ -304,7 +311,18 @@ private String name;
 			// TODO 自动生成的方法存根
 			return 0;
 		}
-		@Override
+
+    @Override
+    public long calculateMinDelay(long timeWindowNs) {
+        return 0;
+    }
+
+    @Override
+    public long calculateMinDelay(Long128 currentTime, long timeWindowNs) {
+        return 0;
+    }
+
+    @Override
 		public TimestampMonitorValue getRecentDelayEntry() {
 			// TODO 自动生成的方法存根
 			return null;

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

@@ -19,26 +19,16 @@ public class SpeedAndTrafficAndDelayMonitorDataImpl extends SpeedAndTrafficMonit
 		upSampler.update();
 		downSampler.update();
 		if(upSampler.getSnapshotPacketCount()>0&&upSampler.getSnapshotTotalDelay()>0) {
-		getUploadBandwidth(). recordPacket(KLALBUtils.createGlobalUUID(),0, 0, upSampler.getAvgDelayNanos());
-		long odelay=upSampler.getMinDelayNanos();
-		if(outDelayMin==Long.MAX_VALUE||odelay<=outDelayMin) {
-			outDelayMin=odelay;
-		}else{
-			outDelayMin=Math.min(odelay, outDelayMin+10000);
-		}
+		getUploadBandwidth(). recordPacket(KLALBUtils.createGlobalUUID(),0, 0, upSampler.getAvgDelayNanos(),upSampler.getMinDelayNanos());
 		}
 		if(downSampler.getSnapshotPacketCount()>0&&downSampler.getSnapshotTotalDelay()>0) {
-		getDownloadBandwidth(). recordPacket(KLALBUtils.createGlobalUUID(),0, 0, downSampler.getAvgDelayNanos());
-		long idelay=downSampler.getMinDelayNanos();
-		if(inDelayMin==Long.MAX_VALUE||idelay<=inDelayMin) {
-			inDelayMin=idelay;
-		}else{
-			inDelayMin=Math.min(idelay, inDelayMin+10000);
-		}
+		getDownloadBandwidth(). recordPacket(KLALBUtils.createGlobalUUID(),0, 0, downSampler.getAvgDelayNanos(),downSampler.getMinDelayNanos());
 		}
 		
 		long time=System.nanoTime();
 		if(time-stime>timewindowmax) {
+			outDelayMin=getUploadBandwidth().calculateMinDelay(super.getTimeWindowMax());
+			inDelayMin=getDownloadBandwidth().calculateMinDelay(super.getTimeWindowMax());
 			stime=time;
 		}
 	}

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

@@ -11,13 +11,13 @@ public class SpeedAndTrafficMonitorDataImpl extends MonitorDataImpl implements S
 	private TimestampMonitor<UUID>uploadBandwidth;
 	private TimestampMonitor<UUID>downloadBandwidth;
 	private long timewindow=1000000000L;
-	private long timewindowmax=1000000000L;
+	private long timewindowmax=30000000000L;
 	private BandwidthSampler upSampler=new BandwidthSampler();
 	private BandwidthSampler downSampler=new BandwidthSampler();
 
 
 
-	private final long TIME_WINDOW=2000000000L;
+	private final long TIME_WINDOW=50000000000L;
 	public SpeedAndTrafficMonitorDataImpl(HighAccuracyClock clock) {
 		super();
 		this.clock=clock;
@@ -177,32 +177,39 @@ 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);
 		long op=uploadBandwidth.calculatePacketRate(timewindowmax);
 		long ip=uploadBandwidth.calculatePacketRate(timewindowmax);
 		if(os>outSpeeedMax) {
-			outSpeeedMax=(outSpeeedMax+os)/2;
+			outSpeeedMax=os;
 		}else {
 			outSpeeedMax=(outSpeeedMax*999+os)/1000;
 		}
 		if(is>inSpeedMax) {
-			inSpeedMax=(inSpeedMax+is)/2;
+			inSpeedMax=is;
 		}else {
 			inSpeedMax=(inSpeedMax*999+is)/1000;
 		}
 		if(op>outPPSMax) {
-			outPPSMax=(outPPSMax+op)/2;
+			outPPSMax=op;
 		}else {
 			outPPSMax=(outPPSMax*999+op)/1000;
 		}
 		if(ip>inPPSMax) {
-			inPPSMax=(inPPSMax+ip)/2;
+			inPPSMax=ip;
 		}else {
 			inPPSMax=(inPPSMax*999+ip)/1000;
 		}
 		}
 	}
+
+	protected long getTimeWindow() {
+		return timewindow;
+	}
+	protected long getTimeWindowMax() {
+		return timewindowmax;
+	}
 }

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

@@ -1,7 +1,6 @@
 package org.kne.cloud.network.monitor;
 
 import org.kne.cloud.clock.HighAccuracyClock;
-import org.kne.cloud.network.ipv6.PacketID;
 import org.kne.math.Long128;
 
 /**
@@ -37,6 +36,9 @@ public interface TimestampMonitor<K> {
      */
     boolean recordPacket(K key,int count, int length);
 
+    boolean recordPacket(K key, int count,int length, long dsndDelay);
+
+    boolean recordPacket(K key, int count, int length, long avgDelayNanos, long minDelayNanos);
     /**
      * 记录数据包（使用指定时间戳）
      * @param key 数据包标识
@@ -49,7 +51,9 @@ public interface TimestampMonitor<K> {
 
     boolean recordPacket(K key, Long128 timestamp, int count,int length, long dsndDelay);
 
-    boolean recordPacket(K key, int count,int length, long dsndDelay);
+
+
+    boolean recordPacket(K key, Long128 timestamp, int count,int length, long dsndDelay,long dsndDelayMin);
 
     // ==================== 清理维护方法 ====================
     
@@ -146,7 +150,11 @@ public interface TimestampMonitor<K> {
     long getRecentDelay();
 
 
-	TimestampMonitorValue getRecentDelayEntry();
+    long calculateMinDelay(long timeWindowNs);
+
+    long calculateMinDelay(Long128 currentTime, long timeWindowNs);
+
+    TimestampMonitorValue getRecentDelayEntry();
     
     
     // ==================== 全局统计方法 ====================

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

@@ -8,22 +8,26 @@ public  class TimestampMonitorValue implements Comparable<TimestampMonitorValue>
         private final int length;
         private final TimestampMonitorKey<?> key;
         private long delay;
+	    private long delayMin;
         public TimestampMonitorValue(TimestampMonitorKey<?> key, Long128 timestamp,int count, int length) {
-        	this(key,timestamp,count,length,Long.MAX_VALUE/1024);
+        	this(key,timestamp,count,length,Long.MAX_VALUE/1024,Long.MAX_VALUE/1024);
         }
-        public TimestampMonitorValue(TimestampMonitorKey<?> key, Long128 timestamp,int count, int length,long delay) {
+        public TimestampMonitorValue(TimestampMonitorKey<?> key, Long128 timestamp,int count, int length,long delay,long delayMin) {
             this.key = key;
             this.timestamp = timestamp;
             this.count=count;
             this.length = length;
             this.delay=delay;
+			this.delayMin=delayMin;
         }
         
         @Override
 		public int hashCode() {
 			final int prime = 31;
 			int result = 1;
+			result = prime * result + count;
 			result = prime * result + (int) (delay ^ (delay >>> 32));
+			result = prime * result + (int) (delayMin ^ (delayMin >>> 32));
 			result = prime * result + length;
 			result = prime * result + ((timestamp == null) ? 0 : timestamp.hashCode());
 			return result;
@@ -38,8 +42,12 @@ public  class TimestampMonitorValue implements Comparable<TimestampMonitorValue>
 			if (getClass() != obj.getClass())
 				return false;
 			TimestampMonitorValue other = (TimestampMonitorValue) obj;
+			if (count != other.count)
+				return false;
 			if (delay != other.delay)
 				return false;
+			if (delayMin != other.delayMin)
+				return false;
 			if (length != other.length)
 				return false;
 			if (timestamp == null) {
@@ -52,8 +60,8 @@ public  class TimestampMonitorValue implements Comparable<TimestampMonitorValue>
 
         @Override
 		public String toString() {
-			return "TimestampMonitorValue [timestamp=" + timestamp + ", length=" + length + ", key=" + key + ", delay="
-					+ delay + "]";
+			return "TimestampMonitorValue [timestamp=" + timestamp + ", count=" + count + ", length=" + length
+					+ ", delay=" + delay + ", delayMin=" + delayMin + "]";
 		}
 
         public Long128 getTimestamp() { return timestamp; }
@@ -73,6 +81,12 @@ public  class TimestampMonitorValue implements Comparable<TimestampMonitorValue>
 			this.delay = delay;
 		}
 
+		public long getDelayMin() {
+			return delayMin;
+		}
+		public void setDelayMin(long delayMin) {
+			this.delayMin = delayMin;
+		}
 		@Override
         public int compareTo(TimestampMonitorValue other) {
             return this.timestamp.compareTo(other.timestamp);

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.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;
@@ -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();
-		ThreadTool.makeVDaemonThread("线路性能采样线程", monitorUpdater).start();
+		monitorUpdater= KLALBScheduler.getScheduler().scheduleWithFixedDelay(new MonitorUpdater(),1,1, TimeUnit.MILLISECONDS);
 		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,73 +1,87 @@
 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;
 
 public class TESTNPacket extends KLALBPacket {
 
-	private static final int HEADER_LENGTH=3;
+	private static final int HEADER_LENGTH = 3;
+
+	// 核心：一个线程私有的 65535 字节大缓存，生命周期与线程同寿
+	private static final ThreadLocal<ByteBuffer> BENCH_BUFFER =
+			ThreadLocal.withInitial(() -> ByteBuffer.allocate(65535));
+
+	// 不再持有 ByteBuffer，只记录长度
+	private int payloadLength;
 
-	private ByteBuffer dataBuffer;//=NetworkPacket.databufferpool_65535.borrow();
 	@Override
 	public long getTotalLength() {
-		return HEADER_LENGTH+dataBuffer.limit();
+		return HEADER_LENGTH + payloadLength;
 	}
 
-
+	// 发送端构造
 	public TESTNPacket(int limit) {
-		super(TESTN,HEADER_LENGTH);
-		//dataBuffer.limit(limit);
-		dataBuffer=NetworkPacket.bufferAllocator.allocate(limit);
+		super(TESTN, HEADER_LENGTH);
+		this.payloadLength = limit;
 	}
 
-	
-
-	public ByteBuffer getDataBuffer() {
-		return dataBuffer;
+	// 接收端构造（从已有的 ByteBuffer 解析）
+	public TESTNPacket(ByteBuffer bb) {
+		super(bb, HEADER_LENGTH);
+		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);
-		dataBuffer=NetworkPacket.bufferAllocator.allocate(limit);
-		 while(dataBuffer.hasRemaining()){
-			if(din.read(dataBuffer)==-1) {
-				throw new EOFException();
+		this.payloadLength = klalbHeader.getChar(1);
+
+		// 2. 如果有 payload，借用缓冲区接收数据
+		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() 时会自动覆盖
 		}
-		 dataBuffer.flip();
 	}
 
-	public TESTNPacket(ByteBuffer bb) {
-		super(bb,HEADER_LENGTH);
-	}
-
-
 	@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);
-			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));
+			KLALBRoutingProtocolPacket kp=receivePacket();
 			
 			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,20 +326,34 @@ 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);
-		KLALBRoutingProtocolPacket.writeKLALBPacketToChannel(Channels.newChannel(bos),pkt);
-		byte[]to=bos.toByteArray();
-		DatagramPacket dgpx=new DatagramPacket(to,to.length);
+	private void writePacket(KLALBRoutingProtocolPacket pkt, SocketAddress dest) throws IOException {
+		// 2. 从 ThreadLocal 获取，并清空状态
+		ByteBuffer bbf = LSDB_TX_BUFFER.get();
+		bbf.clear();
+
+		// 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);
-		}catch(BindException e) {
+			ds.send(dgpx);
+		} catch (BindException e) {
 			e.printStackTrace();
-			System.err.println("Cannot assign:"+dest);
-		}finally {
+			System.err.println("Cannot assign:" + dest);
+		} finally {
 			sendLock.unlock();
 		}
 	}

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,8 +28,9 @@ public class SRHInsertNetworkLink extends AbstractIPv6NetworkLink {
 
 	@Override
 	public void sendPacket(IPv6Packet pack, IPv6Address inet6Address) throws IOException {
-		getSrv6Router().insertSRHeader( pack);
-		getSrv6Router().insertHeaderAndRoutePacket(null, pack);
+		if(getSrv6Router().insertSRHeader( pack)) {
+			getSrv6Router().insertHeaderAndRoutePacket(null, pack);
+		}
 	}
 
 	@Override

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

@@ -12,11 +12,11 @@ import java.util.Iterator;
 import java.util.List;
 import java.util.Map;
 import java.util.Set;
-import java.util.Timer;
-import java.util.TimerTask;
 import java.util.UUID;
+import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.CopyOnWriteArrayList;
 import java.util.concurrent.CopyOnWriteArraySet;
+import java.util.concurrent.TimeUnit;
 import java.util.concurrent.locks.Condition;
 import java.util.concurrent.locks.ReentrantLock;
 import java.util.function.BiConsumer;
@@ -45,12 +45,13 @@ import org.kne.cloud.network.ipv6.PadNHopByHopTLV;
 import org.kne.cloud.network.ipv6.PostcardEntry;
 import org.kne.cloud.network.ipv6.RouteItem;
 import org.kne.cloud.network.klalb.KLALBRemoteLink;
+import org.kne.cloud.network.klalb.KLALBScheduler;
 import org.kne.cloud.network.klalb.KLALBUtils;
 import org.kne.cloud.network.klalb.PerformanceStrategy;
 import org.kne.cloud.network.klalb.ui.PerformanceStrategyItem;
 import org.kne.concurrent.HighPerformanceExecutor2;
 import org.kne.concurrent.TimeoutConcurrentHashMap;
-import org.pcap4j.packet.IllegalRawDataException;
+import org.kne.io.KNEChannels;
 
 import com.google.gson.internal.Pair;
 
@@ -60,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地址位数
 
@@ -203,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));
+		for (Map<IPv6Address, List<RouteItem>> map : routeTabel1x) {
+			// 使用 compute 直接替换
+			map.replaceAll((key, oldList) -> {
+				if (oldList == null || oldList.isEmpty()) {
+					return oldList;
+				}
+				List<RouteItem> newList = new ArrayList<>(oldList);
+				Collections.sort(newList);
+				return newList;
+			});
 		}
-		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()) {
-				Collections.sort(v);
-			}
-		}
-		// System.out.println(routeTabel1x);
-		routeTabel1 = routeTabel1x; // 更新索引路由表
-		
-		
 	}
 	// 更新路由表
 	public void updateRouteTabel() {
@@ -263,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>>());
 		}
 
 		// 按前缀长度组织路由表
@@ -496,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);
@@ -525,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();
@@ -559,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()) { // 环回链路处理
@@ -574,37 +560,40 @@ public class SRv6Router {
 			if (srhh != null) {
 				processSRv6Packet(linkfrom, iPv6Packet, ri, srhh); // SRv6特殊处理
 			} else {
-				sendPacketToRouteItem(iPv6Packet, ri);
+				sendPacketToRouteItem(linkfrom,iPv6Packet, ri);
 			}
 		} else { // 普通链路处理
-			// 检查重路由计数
-			int count=iPv6Packet.getRouteCounter().getAndIncrement();
-			if (count >= MAX_REROUTE_COUNT) {
-				return; // 超过最大重路由次数，丢弃
-			}
+			if(ri.getDestlink().isFilter()){
 
-			if (count<=0) {
-				hop--; // 减少TTL（非重路由时）
-			}
+			}else {
+				// 检查重路由计数
+				int count = iPv6Packet.getRouteCounter().getAndIncrement();
+				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();
@@ -615,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);
 						}
 					}
 				}
@@ -625,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();
@@ -647,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) {
 				// 重路由不处理
@@ -850,7 +839,6 @@ public class SRv6Router {
 	private volatile long backplanePPSMax=0;
 	private volatile long prevTime=System.nanoTime();
 
-	private Timer backplaneTimer=new Timer();
 	private void updateBackplaneCounter() {
 		long curr=System.nanoTime();
 		long totalTime=timeAdder.getAndReset();
@@ -904,13 +892,13 @@ public class SRv6Router {
 		linkTabel.add(inLoopBack); // 添加到链路表
 		this.srhInserter=new SRHInsertNetworkLink(this,networkGroup);
 		linkTabel.add(srhInserter);
-		backplaneTimer.scheduleAtFixedRate(new TimerTask() {
+		KLALBScheduler.getScheduler().scheduleWithFixedDelay(new Runnable() {
 			
 			@Override
 			public void run() {
 		    	   updateBackplaneCounter();
 			}
-		}, 100, 100);
+		}, 100, 100, TimeUnit.MILLISECONDS);
 	}
 
 	/*

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