添加新的README.md

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

.classpath

@@ -2,32 +2,17 @@
 <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/JFreeChart1.5.2.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/KNEOptimize.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="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-sources.jar"/>
 	<classpathentry kind="output" path="bin"/>
 </classpath>

.idea/libraries/jctools_core_jdk11_4_0_7.xml

@@ -0,0 +1,11 @@
+<component name="libraryTable">
+  <library name="jctools-core-jdk11-4.0.7">
+    <CLASSES>
+      <root url="jar://$PROJECT_DIR$/lib/jctools-core-jdk11-4.0.7.jar!/" />
+    </CLASSES>
+    <JAVADOC />
+    <SOURCES>
+      <root url="jar://$PROJECT_DIR$/lib/jctools-core-jdk11-4.0.7-sources.jar!/" />
+    </SOURCES>
+  </library>
+</component>

.idea/libraries/jfreechart_1_5_3.xml

@@ -0,0 +1,38 @@
+<component name="libraryTable">
+  <library name="jfreechart-1.5.3">
+    <CLASSES>
+      <root url="jar://$PROJECT_DIR$/lib/jfreechart-1.5.3.jar!/" />
+      <root url="jar://$PROJECT_DIR$/lib/disruptor-4.0.0.jar!/" />
+      <root url="jar://$PROJECT_DIR$/lib/gson-2.1.jar!/" />
+      <root url="jar://$PROJECT_DIR$/lib/ini4j.jar!/" />
+      <root url="jar://$PROJECT_DIR$/lib/javassist.jar!/" />
+      <root url="jar://$PROJECT_DIR$/lib/JavaTUN0.3.jar!/" />
+      <root url="jar://$PROJECT_DIR$/lib/jctools-core-4.0.6.jar!/" />
+      <root url="jar://$PROJECT_DIR$/lib/KNElib1.0.jar!/" />
+      <root url="jar://$PROJECT_DIR$/lib/KNEOptimize.jar!/" />
+      <root url="jar://$PROJECT_DIR$/lib/pcap4j-core-1.8.2.jar!/" />
+      <root url="jar://$PROJECT_DIR$/lib/pcap4j-packetfactory-propertiesbased-1.8.2.jar!/" />
+      <root url="jar://$PROJECT_DIR$/lib/pcap4j-packetfactory-static-1.8.2.jar!/" />
+      <root url="jar://$PROJECT_DIR$/lib/pcap4j-packettest-1.8.2-tests.jar!/" />
+      <root url="jar://$PROJECT_DIR$/lib/pcap4j-sample-1.8.2.jar!/" />
+      <root url="jar://$PROJECT_DIR$/lib/slf4j-api-2.0.9.jar!/" />
+      <root url="jar://$PROJECT_DIR$/lib/slf4j-api-2.0.9-tests.jar!/" />
+      <root url="jar://$PROJECT_DIR$/lib/toml4j-0.7.1.jar!/" />
+      <root url="jar://$PROJECT_DIR$/lib/uuid-creator-6.1.1.jar!/" />
+    </CLASSES>
+    <JAVADOC>
+      <root url="jar://$PROJECT_DIR$/lib/jctools-core-4.0.6-javadoc.jar!/" />
+    </JAVADOC>
+    <SOURCES>
+      <root url="jar://$PROJECT_DIR$/lib/jfreechart-1.5.3-sources.jar!/" />
+      <root url="jar://$PROJECT_DIR$/lib/disruptor-4.0.0-sources.jar!/" />
+      <root url="jar://$PROJECT_DIR$/lib/ini4j.jar!/" />
+      <root url="jar://$PROJECT_DIR$/lib/JavaTUN0.3.jar!/" />
+      <root url="jar://$PROJECT_DIR$/lib/jctools-core-4.0.6-sources.jar!/" />
+      <root url="jar://$PROJECT_DIR$/lib/KNElib1.0.jar!/" />
+      <root url="jar://$PROJECT_DIR$/lib/slf4j-api-2.0.9-sources.jar!/" />
+      <root url="jar://$PROJECT_DIR$/lib/toml4j-0.7.1.jar!/" />
+      <root url="jar://$PROJECT_DIR$/lib/uuid-creator-6.1.1-sources.jar!/" />
+    </SOURCES>
+  </library>
+</component>

.idea/misc.xml

@@ -1,6 +1,6 @@
 <?xml version="1.0" encoding="UTF-8"?>
 <project version="4">
-  <component name="ProjectRootManager" version="2">
+  <component name="ProjectRootManager" version="2" languageLevel="JDK_26" project-jdk-name="jdk-26.0.1" project-jdk-type="JavaSDK">
     <output url="file://$PROJECT_DIR$/classes" />
   </component>
 </project>

.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>

KLALB.iml

@@ -249,5 +249,7 @@
       </library>
     </orderEntry>
     <orderEntry type="jdk" jdkName="jdk-26.0.1" jdkType="JavaSDK" />
+    <orderEntry type="library" name="jfreechart-1.5.3" level="project" />
+    <orderEntry type="library" name="jctools-core-jdk11-4.0.7" level="project" />
   </component>
 </module>

client.cfg

@@ -1,20 +0,0 @@
-//加速线路
-{KLALBRemote}->43.248.189.107:35000
-{KLALBRemote}->cn-bj-bgp-3.openfrp.top:65529
-{KLALBRemote}->180.76.147.250:65529
-{KLALBRemote}->cn-ah-dx-1.natfrp.cloud:65529
-{KLALBRemote}->cn-nn-dx-1.natfrp.cloud:65529
-{KLALBRemote}->cn-wh-dx-1.natfrp.cloud:65529
-{KLALBRemote}->cn-zz-bgp-10.natfrp.cloud:23330
-{KLALBRemote}->cn-zz-bgp-7.natfrp.cloud:33336
-{KLALBRemote}->43.143.109.64:49965
-{KLALBRemote}->frp.104300.xyz:49965
-{KLALBRemote}->us.afrps.cn:49966
-{KLALBRemote}->hk.afrps.cn:49966
-{KLALBRemote}->la.afrps.cn:49966
-{KLALBRemote}->frp.freefrp.net:49965
-{KLALBRemote}->frp1.freefrp.net:49965
-{KLALBRemote}->frp2.freefrp.net:49965
-{KLALBRemote}->frp4.freefrp.net:49965
-
-0.0.0.0:25565->{KLALBVirtual}[171d:a999:e697:4b23:ae52:9f29:4532:e1ee]:25565

klalbconfig-old.json

@@ -1,79 +0,0 @@
-[ 
-  {"Type":"KLALBController" ,
-    "TCPListen" : "0.0.0.0:4565",
-    "UDPListen" : "{UDP}0.0.0.0:4572",
-    "VirtualSocketName" : "KLALB_Stream",
-    "LineTable":["43.248.189.107:65529",
-"cn-bj-bgp-3.openfrp.top:65529",
-"180.76.147.250:65529",
-"cn-ah-dx-1.natfrp.cloud:65529",
-"cn-nn-dx-1.natfrp.cloud:65529",
-"cn-wh-dx-1.natfrp.cloud:65529",
-"cn-zz-bgp-10.natfrp.cloud:23330",
-"cn-zz-bgp-7.natfrp.cloud:33336",
-"43.143.109.64:49965",
-"frp.104300.xyz:49965",
-"us.afrps.cn:49966",
-"hk.afrps.cn:49966",
-"la.afrps.cn:49966",
-"frp.freefrp.net:49965",
-"frp1.freefrp.net:49965",
-"frp2.freefrp.net:49965",
-"frp4.freefrp.net:49965",
-"cn-he-plc-2.openfrp.top:4572",
-"{UDP}cn-he-plc-2.openfrp.top:4572",
-"frp-mom.top:22226",
-"frp-man.top:36555",
-"us-sjc-bgp-2.openfrp.top:36555",
-"us-sjc-bgp-1.openfrp.top:36555",
-"[240e:340:d856:b700:6d02:6a6a:f9fc:865b]:4572"]
-  },
- {"Type":"SocketBridge" ,
-    "Listen" : "0.0.0.0:4572",
-"Bridge":{
-      "KLALB" : "SocketBridge",
-      "RDP":"SocketBridge",
-      "HTTP":"SocketBridge",
-      "HTTPS":"SocketBridge",
-      "DEFAULT":"MinecraftSocketBridge23332"
-    } ,
-    "Connect" : {
-      "KLALB" : "127.0.0.1:4565",
-      "RDP":"127.0.0.1:3389",
-      "HTTP":"127.0.0.1:5212",
-      "HTTPS":"192.168.1.235:443",
-      "DEFAULT":"127.0.0.1:36555"
-    }
-  },
-   {"Type":"SocketBridge" ,
-    "Listen" : "{KLALB_Stream}[::0]:23332",
-"Bridge":"MinecraftSocketBridge23332",
-    "Connect" : "127.0.0.1:36555"
-  },
-  { "Type":"SocketBridge" ,
-    "Listen" : "{KLALB_Stream}[::0]:23333",
-"Bridge":"SocketBridge",
-    "Connect" : "127.0.0.1:5212"
-  },
-
-
-
-
-
- {"Type":"SocketBridge" ,
-    "Listen" : "0.0.0.0:25561",
-"Bridge":{
-      "KLALB" : "SocketBridge",
-      "DEFAULT":"MinecraftSocketBridge23330"
-    } ,
-    "Connect" : {
-      "KLALB" : "127.0.0.1:4565",
-      "DEFAULT":"127.0.0.1:25562"
-    }
-  },
-   {"Type":"SocketBridge" ,
-    "Listen" : "{KLALB_Stream}[::0]:23330",
-"Bridge":"MinecraftSocketBridge23330",
-    "Connect" : "127.0.0.1:25562"
-  }
-]

lines.cfg

@@ -1,29 +0,0 @@
-0.0.0.0:35000(RDP)->192.168.1.233:3389
-0.0.0.0:35000(HTTPS)->192.168.1.233:8444
-0.0.0.0:35000(HTTP)->192.168.1.233:5212
-0.0.0.0:35000(SSH)->192.168.1.236:22
-0.0.0.0:35000(KLALB)->{KLALBRemote}
-0.0.0.0:35000->127.0.0.1:35001   //Winds服(普通线路)
-
-{KLALBVirtual}[::0]:25565->127.0.0.1:35001 //Winds服(加速线路)
-
-//加速线路
-{KLALBRemote}->43.248.189.107:35000
-{KLALBRemote}->cn-bj-bgp-3.openfrp.top:65529
-{KLALBRemote}->180.76.147.250:65529
-{KLALBRemote}->cn-ah-dx-1.natfrp.cloud:65529
-{KLALBRemote}->cn-nn-dx-1.natfrp.cloud:65529
-{KLALBRemote}->cn-wh-dx-1.natfrp.cloud:65529
-{KLALBRemote}->cn-zz-bgp-10.natfrp.cloud:23330
-{KLALBRemote}->cn-zz-bgp-7.natfrp.cloud:33336
-{KLALBRemote}->43.143.109.64:49965
-{KLALBRemote}->frp.104300.xyz:49965
-{KLALBRemote}->us.afrps.cn:49966
-{KLALBRemote}->hk.afrps.cn:49966
-{KLALBRemote}->la.afrps.cn:49966
-{KLALBRemote}->frp.freefrp.net:49965
-{KLALBRemote}->frp1.freefrp.net:49965
-{KLALBRemote}->frp2.freefrp.net:49965
-{KLALBRemote}->frp4.freefrp.net:49965
-
-0.0.0.0:25565->{KLALBVirtual}[171d:a999:e697:4b23:ae52:9f29:4532:e1ee]:25565

linetable - 副本.txt

@@ -1,18 +0,0 @@
-43.248.189.107:65529
-cn-bj-bgp-3.openfrp.top:65529
-180.76.147.250:65529
-cn-ah-dx-1.natfrp.cloud:65529
-cn-nn-dx-1.natfrp.cloud:65529
-cn-wh-dx-1.natfrp.cloud:65529
-cn-zz-bgp-10.natfrp.cloud:23330
-cn-zz-bgp-7.natfrp.cloud:33336
-43.143.109.64:49965
-frp.104300.xyz:49965
-us.afrps.cn:49966
-hk.afrps.cn:49966
-la.afrps.cn:49966
-frp.freefrp.net:49965
-frp1.freefrp.net:49965
-frp2.freefrp.net:49965
-frp4.freefrp.net:49965
-cn-he-plc-2.openfrp.top:4569

server.cfg

@@ -1,8 +0,0 @@
-0.0.0.0:35000(RDP)->192.168.1.233:3389
-0.0.0.0:35000(HTTPS)->192.168.1.233:8444
-0.0.0.0:35000(HTTP)->192.168.1.233:5212
-0.0.0.0:35000(SSH)->192.168.1.236:22
-0.0.0.0:35000(KLALB)->{KLALBRemote}
-0.0.0.0:35000->127.0.0.1:35001   //Winds服(普通线路)
-
-{KLALBVirtual}[::0]:25565->127.0.0.1:35001 //Winds服(加速线路)

src/klalb_en_US.properties

@@ -117,3 +117,11 @@ enablewebapi=Enable Web API
 weblistenaddr=Web API listen address
 invaildweblistenaddr=Invalid Web API listen address
 enabletun=Enable TUN adapter
+congestionmonitor=Congestion control monitor
+rttbaseline=RTT baseline
+rtt=RTT
+rttmax=RTT max
+rttmin=RTT min
+rto=RTO
+cpuload=CPU load
+ramusage=RAM usage

src/klalb_zh_CN.properties

@@ -117,3 +117,11 @@ enablewebapi=启用 Web API
 weblistenaddr=Web API 监听地址:端口
 invaildweblistenaddr=无效的 Web API 监听地址:端口
 enabletun=启用TUN虚拟网卡
+congestionmonitor=拥塞控制监视器
+rttbaseline=往返延迟基线
+rtt=往返延迟
+rttmax=往返延迟上限
+rttmin=往返延迟下限
+rto=重传超时
+cpuload=CPU负载
+ramusage=RAM占用率

src/org/kne/cloud/clock/NTPTimestamps.java

@@ -5,6 +5,9 @@ import java.time.LocalDateTime;
 import java.time.ZoneId;
 import java.time.ZonedDateTime;
 import java.time.format.DateTimeFormatter;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Objects;
 
 import org.kne.math.Long128;
 
@@ -256,4 +259,27 @@ public class NTPTimestamps {
                             .setBit(63);
         }
     }
+    
+    public static Long128 median(List<Long128> values) {
+    	Objects.requireNonNull(values);
+        if ( values.isEmpty()) {
+            throw new IllegalArgumentException("列表不能为空");
+        }
+        
+        // 1. 创建副本并排序
+        List<Long128> sorted = new ArrayList<>(values);
+        sorted.sort(Long128::compareTo);
+        
+        // 2. 计算中位数
+        int size = sorted.size();
+        if (size % 2 == 1) {
+            // 奇数个：取中间值
+            return sorted.get(size / 2);
+        } else {
+            // 偶数个：取中间两个的平均值
+            Long128 left = sorted.get(size / 2 - 1);
+            Long128 right = sorted.get(size / 2);
+            return left.add(right).divide(Long128.valueOf(2));
+        }
+    }
 }

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

@@ -8,6 +8,8 @@ import java.io.DataOutputStream;
 import java.io.EOFException;
 import java.io.IOException;
 import java.net.*;
+import java.nio.ByteBuffer;
+import java.nio.charset.StandardCharsets;
 import java.util.Enumeration;
 import java.util.Iterator;
 import java.util.List;
@@ -148,19 +150,37 @@ public class IPMulticastDiscovery implements Closeable, AutoCloseable {
 		}
 		}).start();
 	}
+	// 1. 接收用的 ThreadLocal 缓冲区
+	private static final ThreadLocal<ByteBuffer> RX_BUFFER = ThreadLocal.withInitial(() -> ByteBuffer.allocate(2048));
+	// 2. 发送用的 ThreadLocal 缓冲区 (广播包一般很小)
+	private static final ThreadLocal<ByteBuffer> TX_BUFFER = ThreadLocal.withInitial(() -> ByteBuffer.allocate(2048));
 
 	private MultiProtocolSocketAddress receive() throws IOException {
-		byte[] b = new byte[65535];
+		ByteBuffer bbf = RX_BUFFER.get();
-		DatagramPacket dp = new DatagramPacket(b, b.length);
+		bbf.clear(); // 重置 position=0, limit=2048
+
+		// 借用底层数组
+		DatagramPacket dp = new DatagramPacket(bbf.array(), bbf.limit());
 		soc.receive(dp);
-		ByteArrayInputStream bis = new ByteArrayInputStream(b, 0, dp.getLength());
+
-		DataInputStream dis = new DataInputStream(bis);
+		// 限制可读范围
-		String val = dis.readUTF();
+		int length = dp.getLength();
-		long h=dis.readLong();
+		bbf.limit(length);
-		long l=dis.readLong();
+
-		UUID uid=new UUID(h, l);
+		// 直接读 UTF 字符串 (不用 DataInputStream)
-		dis.close();
+		// 读取前两个字节的长度，然后创建 String
+		int strLen = bbf.getShort() & 0xFFFF; // 无符号转 int
+		byte[] strBytes = new byte[strLen]; // 这里为了 String 还是得分配一次，但很小
+		bbf.get(strBytes);
+		String val = new String(strBytes, StandardCharsets.UTF_8);
+
+		// 直接读两个 Long 拼成 UUID
+		long h = bbf.getLong();
+		long l = bbf.getLong();
+		UUID uid = new UUID(h, l); // UUID 对象很小，年轻代瞬间回收，不用池化
+
 		MultiProtocolSocketAddress mpsa = new MultiProtocolSocketAddress(val);
+
 		if (mpsa.getInetAddress().isAnyLocalAddress()) {
 			mpsa = new MultiProtocolSocketAddress(mpsa.getProtocol(), dp.getAddress().getHostAddress(), mpsa.getPort());
 		}
@@ -176,15 +196,23 @@ public class IPMulticastDiscovery implements Closeable, AutoCloseable {
 	}
 
 	private void send(MultiProtocolSocketAddress multiProtocolSocketAddress) throws IOException {
-		ByteArrayOutputStream baos = new ByteArrayOutputStream();
+		ByteBuffer bbf = TX_BUFFER.get();
-		DataOutputStream dops = new DataOutputStream(baos);
+		bbf.clear(); // 重置
-		dops.writeUTF(multiProtocolSocketAddress.toString());
+
-		dops.writeLong(selfUUID.getMostSignificantBits());
+		// 写字符串 (长度 + 字节)
-		dops.writeLong(selfUUID.getLeastSignificantBits());
+		byte[] strBytes = multiProtocolSocketAddress.toString().getBytes(StandardCharsets.UTF_8);
-		dops.close();
+		bbf.putShort((short) strBytes.length);
-		byte[] b = baos.toByteArray();
+		bbf.put(strBytes);
-		DatagramPacket dp = new DatagramPacket(b, b.length, group);
+
+		// 写 UUID
+		bbf.putLong(selfUUID.getMostSignificantBits());
+		bbf.putLong(selfUUID.getLeastSignificantBits());
+
+		bbf.flip(); // 准备读取
+
+		DatagramPacket dp = new DatagramPacket(bbf.array(), bbf.limit(), group);
 		soc.send(dp);
+
 		if(debug)
 			System.out.println(ninterface.getDisplayName()+" 发送广播："+ multiProtocolSocketAddress +" "+group+" "+selfUUID);
 

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

@@ -11,14 +11,13 @@ public class BBRCongestionAlgorithm implements CongestionAlgorithm ,DetnetConges
 
 	private Consumer<Long> windowControlConsumer;
 
-	private long MIN_RTTVAR = 5000000L;
 	private volatile long RTTMin = 1000000000L;
-	private volatile long RTTVar = 1000000000L;
-	private volatile long RTTAvg = 1000000000L;
 	private volatile long RTTAvg2 = 1000000000L;
 	private volatile long RTTTotal = 0;
 	private volatile long RTTCount = 0;
-	private volatile long RTO = 1000000000L;
+
+
+	private RTTEstimator rttEstimator=new RTTEstimator();
 
 
 	private volatile long window=MIN_WINDOW;
@@ -54,10 +53,7 @@ public class BBRCongestionAlgorithm implements CongestionAlgorithm ,DetnetConges
 		} else {
 			RTTMin = (RTTMin * 99999 + latencyns) / 100000;
 		}
-		RTTVar = (RTTVar * 3 + Math.abs(RTTAvg - latencyns)) / 4;
+		rttEstimator.putRTT(latencyns);
-		RTTAvg = (RTTAvg * 7 + latencyns) / 8;
-		
-		RTO = RTTAvg + Math.max(MIN_RTTVAR, RTTVar * 4);// RTTVar*4
 
 		RTTTotal += latencyns;
 		RTTCount++;
@@ -93,13 +89,13 @@ public class BBRCongestionAlgorithm implements CongestionAlgorithm ,DetnetConges
 	}
 
 	@Override
-	public void putLoss(long packetSize, long lossns, int losscounter) {
+	public void putLoss(long packetSize, long lossns) {
 		
 	}
 
 	@Override
 	public long getRTO() {
-		return RTO;
+		return rttEstimator.getRTO();
 	}
 
 	@Override
@@ -136,4 +132,29 @@ 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;
+	}
+
+	@Override
+	public double getLowerDelayBound() {
+		return MIN_GAIN;
+	}
+
+	@Override
+	public double getBurstLimit() {
+		return burstLimit;
+	}
+
 }

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

@@ -6,13 +6,13 @@ public interface CongestionAlgorithm {
 	public void reset();
 	public void setWindowControlConsumer(Consumer<Long>windowControlConsumer);
 	public void putAck(long packetSize,long RTTns,boolean ecn);
-	default public void putAck(long packetSize,long RTTns,long OWDup,boolean ecn) {
+	public void putLoss(long packetSize,long lossns);
-		putAck(packetSize,RTTns,ecn);
-	}
-	public void putLoss(long packetSize,long lossns,int losscounter);
 	public void setCurrentWindowUsed(long maxwindow);
 	public void setCurrentBandwidth(long bandwidth);
 	public long getRTO();
 	public String getName();
 
+	public long getBaseRTT();
+
+	public long getCurrentRTT();
 }

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

@@ -1,126 +0,0 @@
-package org.kne.cloud.network.congestion;
-
-import java.util.function.Consumer;
-
-import org.jctools.counters.FixedSizeStripedLongCounter;
-import org.kne.cloud.network.klalb.KLALBUtils;
-
-public class DCTCP2CongestionAlgorithm implements CongestionAlgorithm {
-
-	private static final long BURST_TIME = 1000000L;
-	private static final long BIAS = 2000000L;
-	private static long MIN_WINDOW = 8192;
-
-	private Consumer<Long> windowControlConsumer;
-
-	private long MIN_RTTVAR = 100000000L;
-	private volatile long RTTVar = 1000000000L;
-	private volatile long RTTAvg = 1000000000L;
-	private double ECNAvg2 = 0L;
-	private double alpha=0;
-	private volatile FixedSizeStripedLongCounter ECNSize = KLALBUtils.createCounter();
-	private volatile FixedSizeStripedLongCounter TotalSize = KLALBUtils.createCounter();
-	private volatile long RTO = 500000000L;
-
-
-	private volatile long window=MIN_WINDOW;
-    private volatile long maxwindow = 99999999999L;
-	private volatile long congressWindowSize=MIN_WINDOW;
-
-	private long RTTstartTime = System.nanoTime();
-	
-	@Override
-	public void reset() {
-		window = MIN_WINDOW;
-		if (windowControlConsumer != null) {
-			windowControlConsumer.accept(window);
-		}
-	}
-
-	@Override
-	public void setWindowControlConsumer(Consumer<Long> windowControlConsumer) {
-		this.windowControlConsumer = windowControlConsumer;
-	}
-
-
-	@Override
-	public synchronized void putAck(long packetSize, long latencyns, boolean ecn) {
-		
-		
-		RTTVar = (RTTVar * 3 + Math.abs(RTTAvg - latencyns)) / 4;
-		RTTAvg = (RTTAvg * 7 + latencyns) / 8;
-		
-		RTO = RTTAvg + Math.max(MIN_RTTVAR, RTTVar * 4);// RTTVar*4
-
-		if(ecn) {
-		ECNSize.inc(packetSize);
-		}
-		TotalSize.inc(packetSize);
-		
-		
-		long curr=System.nanoTime();
-		if(curr-RTTstartTime>(RTTAvg+BIAS)) {
-			RTTstartTime=curr;
-			
-			long totalSize=TotalSize.getAndReset();
-			long ecnSize=ECNSize.getAndReset();
-			if(totalSize!=0) {
-			ECNAvg2=ecnSize/(double)totalSize;
-			alpha=(alpha*7+ECNAvg2)/8;
-			}
-			
-			if(ECNAvg2>0.5) {
-					congressWindowSize-=200;
-			}else if(ECNAvg2>0.01){
-					congressWindowSize+=200;
-			}else {
-					congressWindowSize+=4000;
-			}
-			//System.out.println(ECNAvg2);
-			//System.out.println(speed+" "+window);
-			
-					updateWindowSize();
-			
-		}
-		
-		
-	}
-
-	private void updateWindowSize() {
-		if(congressWindowSize<MIN_WINDOW) {
-			congressWindowSize=MIN_WINDOW;
-		}else if(congressWindowSize>maxwindow) {
-			congressWindowSize=maxwindow;
-		}
-		window=Math.max( congressWindowSize,MIN_WINDOW);
-		if (windowControlConsumer != null) {
-			windowControlConsumer.accept(window);
-		}
-	}
-
-	@Override
-	public void putLoss(long packetSize, long lossns, int losscounter) {
-
-	}
-
-	@Override
-	public long getRTO() {
-		return RTO;
-	}
-
-	@Override
-	public void setCurrentWindowUsed(long used) {
-		this.maxwindow=(long) (used*1.5)+MIN_WINDOW;
-	}
-
-	@Override
-	public void setCurrentBandwidth(long bandwidth) {
-		
-	}
-
-	@Override
-	public String getName() {
-		return "DCTCP2";
-	}
-
-}

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

@@ -13,16 +13,13 @@ public class DCTCPCongestionAlgorithm implements CongestionAlgorithm {
 
 	private Consumer<Long> windowControlConsumer;
 
-	private AtomicBoolean firstUpdate = new AtomicBoolean(true);
-	private long MIN_RTTVAR = 100000000L;
 	private volatile long RTTMin = 1000000000L;
-	private volatile long RTTVar = 1000000000L;
-	private volatile long RTTAvg = 1000000000L;
 	private double ECNAvg2 = 0L;
 	private double alpha=0;
 	private volatile AtomicLong ECNSize = new AtomicLong(0);
 	private volatile AtomicLong TotalSize = new AtomicLong(0);
-	private volatile long RTO = 1000000000L;
+
+	private RTTEstimator rttEstimator=new RTTEstimator(100000000L);
 
 
 	private volatile long window=MIN_WINDOW;
@@ -54,14 +51,7 @@ public class DCTCPCongestionAlgorithm implements CongestionAlgorithm {
 		} else {
 			RTTMin = (RTTMin * 99999 + latencyns) / 100000;
 		}
-		if (firstUpdate.compareAndSet(true, false)) {
+		rttEstimator.putRTT(latencyns);
-			RTTAvg = latencyns;
-			RTTVar = latencyns / 2;
-		} else {
-			RTTVar = (RTTVar * 3 + Math.abs(RTTAvg - latencyns)) / 4;
-			RTTAvg = (RTTAvg * 7 + latencyns) / 8;
-		}
-		RTO = RTTAvg + Math.max(MIN_RTTVAR, RTTVar * 4);// RTTVar*4
 
 		if(ecn) {
 		ECNSize.addAndGet(packetSize);
@@ -88,7 +78,7 @@ public class DCTCPCongestionAlgorithm implements CongestionAlgorithm {
 				}
 			}else {
 				if(windowUsed*3L>=congressWindowSize) {
-					congressWindowSize+=4000;
+					congressWindowSize+=4096;
 					updateWindowSize();
 				}
 			}
@@ -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/DetnetCongestionAlgorithm.java

@@ -4,4 +4,7 @@ public interface DetnetCongestionAlgorithm extends CongestionAlgorithm {
 	public void setUpperDelayBound(double upper);
 	public void setLowerDelayBound(double lower);
 	public void setBurstLimit(double limit);
+	public double getUpperDelayBound( );
+	public double getLowerDelayBound();
+	public double getBurstLimit();
 }

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

@@ -1,6 +1,7 @@
 package org.kne.cloud.network.congestion;
 
-import org.jctools.queues.MpscArrayQueue;
+import org.jctools.queues.MessagePassingQueue;
+import org.jctools.queues.varhandle.MpscVarHandleArrayQueue;
 import org.kne.cloud.network.ThreadTool;
 import org.kne.opencl64.Releaser;
 
@@ -82,7 +83,7 @@ class MpscMessageBatcherReleaser<T> extends Releaser<MpscMessageBatcher0<T>> {
 
 class MpscMessageBatcher0<T> implements Runnable, MessageBatcher<T> {
     
-    private final MpscArrayQueue<T> queue;
+    private final MessagePassingQueue<T> queue;
     private final int batchSize;
     private final long maxDelay;
     private volatile Consumer<List<T>> consumer;
@@ -94,7 +95,7 @@ class MpscMessageBatcher0<T> implements Runnable, MessageBatcher<T> {
     public MpscMessageBatcher0(int batchSize, long maxDelay, int queueCapacity) {
         this.batchSize = batchSize;
         this.maxDelay = maxDelay;
-        this.queue = new MpscArrayQueue<>(queueCapacity);
+        this.queue = new MpscVarHandleArrayQueue<>(queueCapacity);
         
         this.batchThread = ThreadTool.makeVDaemonThread("MpscMessageBatcher", this);
         this.batchThread.start();

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

@@ -31,7 +31,7 @@ public class NOCongestionAlgorithm implements CongestionAlgorithm{
 	}
 
 	@Override
-	public void putLoss(long packetSize, long lossns, int losscounter) {
+	public void putLoss(long packetSize, long lossns) {
 		// TODO 自动生成的方法存根
 		
 	}
@@ -63,4 +63,14 @@ public class NOCongestionAlgorithm implements CongestionAlgorithm{
 		return "NO";
 	}
 
+	@Override
+	public long getBaseRTT() {
+		return 0;
+	}
+
+	@Override
+	public long getCurrentRTT() {
+		return 0;
+	}
+
 }

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

@@ -0,0 +1,150 @@
+package org.kne.cloud.network.congestion;
+
+import org.kne.cloud.clock.HighAccuracyClock;
+import org.kne.cloud.network.klalb.KLALBUtils;
+import org.kne.cloud.network.monitor.ArrayListTimestampMonitor;
+import org.kne.cloud.network.monitor.DelaySampler;
+import org.kne.cloud.network.monitor.TimestampMonitor;
+
+import java.util.UUID;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicLong;
+
+/**
+ * 线程安全的 RTT 平滑与 RTO 计算工具。
+ * 基于 RFC 6298 的 Jacobson/Karels 算法。
+ *
+ * 公式:
+ *   RTTAvg = (RTTAvg * 7 + RTTns) / 8
+ *   RTTVar = (RTTVar * 3 + |RTTAvg - RTTns|) / 4
+ *   RTO = RTTAvg + max(MIN_RTTVAR, RTTVar * 4)
+ */
+public class RTTEstimator {
+
+    /** RTTVar 下限（纳秒），防止 RTO 过小。默认 20ms。 */
+    private static final long DEFAULT_MIN_RTTVAR_NS = 20_000_000L;
+
+    private final AtomicLong curr;
+    private final AtomicLong rttAvg;
+    private final AtomicLong rttVar;
+    private final AtomicLong rto;
+    private final AtomicLong basertt;
+    private final AtomicBoolean firstUpdate = new AtomicBoolean(true);  // 首次标记
+    private final long minRttVarNs;
+
+
+    private long timewindowmax=500000000L;
+
+    long stime=System.nanoTime();
+
+    public RTTEstimator() {
+        this(DEFAULT_MIN_RTTVAR_NS);
+    }
+
+    public RTTEstimator(long minRttVarNs) {
+        this.rttAvg = new AtomicLong(1000000000L);
+        this.rttVar = new AtomicLong(1000000000L);
+        this.curr = new AtomicLong(1000000000L);
+        this.rto = new AtomicLong(1000000000L);
+        this.basertt = new AtomicLong(1000000000L);
+        this.minRttVarNs = minRttVarNs;
+    }
+    private final static long TIME_WINDOW=30000000000L;
+    private final TimestampMonitor<UUID> rttMonitor=new ArrayListTimestampMonitor<UUID>(new HighAccuracyClock(),"RTTMonitor", 100, TIME_WINDOW);
+    private final DelaySampler dsp=new DelaySampler();
+
+    public void update() {
+        dsp.update();
+
+        if(dsp.getSnapshotPacketCount()>0&&dsp.getSnapshotTotalDelay()>0) {
+            rttMonitor. recordPacket(KLALBUtils.createGlobalUUID(),0, 0, dsp.getAvgDelayNanos(),dsp.getMinDelayNanos());
+        }
+
+        long time=System.nanoTime();
+        if(time-stime>timewindowmax) {
+            basertt.lazySet( rttMonitor.calculateMinDelay(TIME_WINDOW));
+            stime=time;
+        }
+    }
+    /**
+     * 更新 RTT 估算值（线程安全）。
+     *
+     * @param rttNs 本次 RTT 测量值（纳秒）
+     */
+    /**
+     * 更新 RTT 估算值（线程安全，无锁）。
+     */
+    public void putRTT(long rttNs) {
+        if (rttNs < 0) {
+            throw new IllegalArgumentException("RTT 不能为负数: " + rttNs);
+        }
+        dsp.recordDelay(rttNs);
+        curr.lazySet(rttNs);
+        // 首次更新：CAS 成功则直接初始化，无锁
+        if (firstUpdate.compareAndSet(true, false)) {
+            rttAvg.set(rttNs);
+            rttVar.set(rttNs / 2);
+            return;
+        }
+
+        // 后续更新：CAS 循环保证原子性
+
+        long oldAvg = rttAvg.get();
+        long newAvg = (oldAvg * 7 + rttNs) / 8;
+
+        long delta = Math.abs(oldAvg - rttNs);
+        rttAvg.lazySet(newAvg);
+        long oldVar = rttVar.get();
+            long newVar = (oldVar * 3 + delta) / 4;
+        rttVar.lazySet(newVar);
+
+        rto .lazySet( newAvg + Math.max(minRttVarNs, newVar * 4));
+    }
+
+
+    /**
+     * 获取平滑 RTT 平均值（纳秒）。
+     */
+    public long getRTTAvg() {
+        return rttAvg.getPlain();
+    }
+
+    /**
+     * 获取 RTT 偏差（纳秒）。
+     */
+    public long getRTTVar() {
+        return rttVar.getPlain();
+    }
+
+    /**
+     * 获取 RTO（重传超时时间，纳秒）。
+     * RTO = RTTAvg + max(MIN_RTTVAR, RTTVar * 4)
+     */
+    public long getRTO() {
+        return rto.getPlain();
+    }
+
+    /**
+     * 重置估算器（用于连接重建等场景）。
+     */
+    public void reset() {
+        rttAvg.set(0);
+        rttVar.set(0);
+    }
+
+    @Override
+    public String toString() {
+        return String.format("RttEstimator{avg=%.3fms, var=%.3fms, rto=%.3fms}",
+                rttAvg.get() / 1_000_000.0,
+                rttVar.get() / 1_000_000.0,
+                getRTO() / 1_000_000.0);
+    }
+
+    public long getCurrentRTT() {
+        return curr.get();
+    }
+
+    public long getBaseRTT() {
+        return basertt.get();
+    }
+}

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

@@ -5,7 +5,10 @@ import java.net.SocketException;
 import java.util.Collection;
 import java.util.Iterator;
 import java.util.Map;
+import java.util.Set;
 import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ScheduledFuture;
+import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicLong;
 import java.util.concurrent.locks.Condition;
 import java.util.concurrent.locks.Lock;
@@ -13,13 +16,15 @@ import java.util.function.Consumer;
 
 import org.kne.cloud.network.NetworkPacket;
 import org.kne.cloud.network.ThreadTool;
+import org.kne.cloud.network.klalb.KLALBScheduler;
 import org.kne.cloud.network.klalb.SendItem;
 import org.kne.concurrent.ThreadParker;
 
 public class SendPacketSlidingWindow<K, V extends NetworkPacket> implements Closeable, AutoCloseable {
+	private final ScheduledFuture<?> checkerTask;
 	private ThreadParker tp=new ThreadParker();
 
-	private Map<K, SendItem<V>> sendMap = new ConcurrentHashMap<>(2048);
+	private volatile Map<K, SendItem<V>> sendMap = new ConcurrentHashMap<>(2048);
 	private AtomicLong sendWindowUsed=new AtomicLong(0);
 	//private LongAdder sendWindowUsed = new LongAdder();
 
@@ -56,8 +61,7 @@ public class SendPacketSlidingWindow<K, V extends NetworkPacket> implements Clos
 		this.removeAtResend = removeAtResend;
 		this.algorithm = algorithm;
 		this.windowSize=windowSize;
-		Thread t = ThreadTool.makeVThread("重路由计时器线程", checker);
+		this.checkerTask=KLALBScheduler.getScheduler().scheduleWithFixedDelay(checker,1,1, TimeUnit.MILLISECONDS);
-		t.start();
 	}
 
 	public long getWindowSize() {
@@ -131,16 +135,15 @@ public class SendPacketSlidingWindow<K, V extends NetworkPacket> implements Clos
 		algorithm.setCurrentWindowUsed(newWindow);
 	}
 	public V ack(K sequence) {
-		return ack(sequence,false);
+		return ack(sequence,0,false);
 	}
 
-	public V ack(K sequence,boolean ecn) {
+	public V ack(K sequence,long RTTns,boolean ecn) {
 		SendItem<V> ipv6;
 		if ((ipv6 = sendMap.remove(sequence)) != null) {
 			long newWindow =sendWindowUsed.addAndGet(-ipv6.getPacketLength() );
 			algorithm.setCurrentWindowUsed(newWindow);
-			long RTTC = System.nanoTime() - ipv6.getSendtime();
+			algorithm.putAck(ipv6.getPacketLength(), RTTns, ecn);
-			algorithm.putAck(ipv6.getPacketLength(), RTTC, ecn);
 			tp.unpark();
 			// System.out.println("remove:"+aseq.getSequence()+" "+sendMap.size());
 			Lock lockx = lock;
@@ -162,65 +165,15 @@ public class SendPacketSlidingWindow<K, V extends NetworkPacket> implements Clos
 		}
 	}
 
-	public V ack(K sequence,boolean ecn,long OWDdown) {
+
-		SendItem<V> ipv6;
-		if ((ipv6 = sendMap.remove(sequence)) != null) {
-			long newWindow =sendWindowUsed.addAndGet(-ipv6.getPacketLength() );
-			algorithm.setCurrentWindowUsed(newWindow);
-			long RTTC = System.nanoTime() - ipv6.getSendtime();
-			algorithm.putAck(ipv6.getPacketLength(), RTTC,OWDdown, ecn);
-			tp.unpark();
-			// System.out.println("remove:"+aseq.getSequence()+" "+sendMap.size());
-			Lock lockx = lock;
-			if (lockx != null) {
-				lockx.lock();
-				try {
-					Condition cds = condition;
-					if (cds != null) {
-						cds.signalAll();
-					}
-				} finally {
-					lockx.unlock();
-				}
-			}
-			return ipv6.getPacket();
-		} else {
-			// System.out.println("miss:"+aseq.getSequence()+" "+sendMap.size());
-			return null;
-		}
-	}
 
 	private class ResendChecker implements Runnable {
 
 		@Override
 		public void run() {
-			while (!closed) {
-
-				Collection<SendItem<V>> vals = sendMap.values();
-				for (Iterator<SendItem<V>> iterator = vals.iterator(); iterator.hasNext();) {
-					SendItem<V> sitm = (SendItem<V>) iterator.next();
-					long current = System.nanoTime();
-					if (current - sitm.getSendtime() > algorithm.getRTO()) {
-						if (removeAtResend) {
-							iterator.remove();
-							long newWindow = sendWindowUsed.addAndGet((int) -sitm.getPacketLength() );
-							algorithm.setCurrentWindowUsed(newWindow);
-							tp.unpark();
-						} else {
-							sitm.setSendtime(current);
-						}
-						V pktr = sitm.getPacket();
-						if (resendConsumer != null)
-							resendConsumer.accept(pktr);
-					}
-				}
-
 			try {
-					Thread.sleep(2);
+				if (closed) {
-				} catch (InterruptedException e) {
+					checkerTask.cancel(false);
-					e.printStackTrace();
-				}
-			}
 					// 清理资源
 					if (closingConsumer != null) {
 						for (SendItem<V> item : sendMap.values()) {
@@ -229,6 +182,42 @@ public class SendPacketSlidingWindow<K, V extends NetworkPacket> implements Clos
 					}
 					sendMap.clear();
 					sendWindowUsed.set(0);
+
+				} else {
+
+					Set<Map.Entry<K,SendItem<V>>> vals = sendMap.entrySet();
+					for (Iterator<Map.Entry<K,SendItem<V>>> iterator = vals.iterator(); iterator.hasNext(); ) {
+						Map.Entry<K,SendItem<V>> entry = (Map.Entry<K,SendItem<V>>) iterator.next();
+						SendItem<V> sitm=entry.getValue();
+						long current = System.nanoTime();
+						long ftime=current - sitm.getSendtime();
+						if (ftime> algorithm.getRTO()) {
+
+							V pktr = sitm.getPacket();
+
+							if (removeAtResend) {
+								SendItem<V> sitm2=sendMap.remove(entry.getKey());
+								if(sitm2!=null) {
+									if (resendConsumer != null)
+										resendConsumer.accept(pktr);
+									algorithm.putLoss(sitm2.getPacketLength(),ftime);
+									long newWindow = sendWindowUsed.addAndGet((int) -sitm2.getPacketLength());
+									algorithm.setCurrentWindowUsed(newWindow);
+									tp.unpark();
+								}
+							} else {
+								sitm.setSendtime(current);
+								if (resendConsumer != null)
+									resendConsumer.accept(pktr);
+							}
+						}
+					}
+
+
+				}
+			} catch (Exception e) {
+				e.printStackTrace();
+			}
 		}
 	}
 

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

@@ -2,6 +2,9 @@ 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 {
     
     // 可配置的时延膨胀比上下限 - Vegas2.0核心参数
@@ -11,29 +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 OWDTotal = 0;
-    private volatile long OWDCount = 0;
-    private volatile long OWDAvg2 = 1000000000L;         // 当前平滑OWD
-    private volatile long OWDMin = 1000000000L;          // BaseRTT
     
     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();
 	}
@@ -58,106 +58,13 @@ public class Vegas2CongestionAlgorithm implements CongestionAlgorithm,DetnetCong
     
    
 
-	@Override
-	public void putAck(long packetSize, long RTTns, long OWDup, boolean ecn) {
-		if(RTTns<0) {
-    		throw new IllegalArgumentException("RTT:"+ RTTns+" is negative!");
-    	}
-        // 处理ECN信号 - 将其视为强烈的拥塞信号
-        if (ecn) {
-            // 当收到ECN时，更激进地减少窗口
-        	window2 = Math.max(0, window2 * 3 / 4);
-            if (windowControlConsumer != null) {
-                windowControlConsumer.accept(window2);
-            }
-        }
-        
-        // 更新最小RTT（BaseRTT）
-        if (RTTns <= RTTMin) {
-            RTTMin = RTTns;
-        } else {
-            // 缓慢适应：当网络路径真正变化时，BaseRTT应能缓慢更新
-            // 这里使用极慢的衰减因子，只有在持续观测到更低RTT时才快速更新
-            RTTMin = (RTTMin * 999 + RTTns) / 1000;
-        }
-        
-        // 更新最小OWD（BaseOWD）
-        if (OWDup <= OWDMin) {
-        	OWDMin = OWDup;
-        } else {
-            // 缓慢适应：当网络路径真正变化时，BaseRTT应能缓慢更新
-            // 这里使用极慢的衰减因子，只有在持续观测到更低RTT时才快速更新
-        	OWDMin = (OWDMin * 999 + OWDup) / 1000;
-        }
-        
-        // 更新平滑RTT估计
-        
-            RTTVar = (RTTVar * 3 + Math.abs(RTTAvg - RTTns)) / 4;
-            RTTAvg = (RTTAvg * 7 + RTTns) / 8;
-        
-        
-        RTO = RTTAvg + Math.max(MIN_RTTVAR, RTTVar * 4);
-        
-        RTTTotal += RTTns;
-        RTTCount++;
-        
-        OWDTotal += OWDup;
-        OWDCount++;
-        
-        long curr = System.nanoTime();
-        // 每个RTT周期调整一次窗口（使用当前估计的RTT）
-        if (curr - RTTstartTime > Math.max(BIAS, 2*RTTMin)) {
-        	long RTTCountx=RTTCount;
-            if (RTTCountx > 0) {
-                long currentRTT = RTTTotal / RTTCountx;
-                //RTTAvg2 = (RTTAvg2 * 3 + currentRTT) / 4;  // 平滑当前RTT
-                RTTAvg2=currentRTT;
-                RTTTotal = 0;
-                RTTCount = 0;
-            }
-            
-            long OWDCountx=OWDCount;
-            if (OWDCountx > 0) {
-                long currentOWD = OWDTotal / OWDCountx;
-                OWDAvg2=currentOWD;
-                OWDTotal = 0;
-                OWDCount = 0;
-            }
-            
-            
-            // ================== Vegas2.0核心逻辑 ==================
-            // 1. 计算时延膨胀比
-            double delayRatio = (double) OWDAvg2 / (double) Math.max(BIAS, OWDMin);
-            //System.out.println(delayRatio+" "+RTTAvg2/1000000.0+" "+RTTMin/1000000.0);
-            // 2. 基于比值的窗口调整（取代原来的基于差值的调整）
-            if (delayRatio > delayUpperBound) {
-                // 时延过高：减小窗口，减少幅度与超标程度成正比
-                window2 -= 4096;
-            } else if (delayRatio < delayLowerBound) {
-                // 时延过低：增大窗口，增加幅度与低于目标程度成正比
-                window2 += 4096;
-            } else {
-              
-            }
-            // ===================================================
-            
-            
-            
-            
-            updateWindowSize();
-            RTTstartTime = curr;
-            
-            
-            
-            
-        }
-	}
 
 	@Override
     public  void putAck(long packetSize, long RTTns, boolean ecn) {
     	if(RTTns<0) {
     		throw new IllegalArgumentException("RTT:"+ RTTns+" is negative!");
     	}
+    	this.RTTCurr=RTTns;
         // 处理ECN信号 - 将其视为强烈的拥塞信号
         if (ecn) {
             // 当收到ECN时，更激进地减少窗口
@@ -167,29 +74,20 @@ public class Vegas2CongestionAlgorithm implements CongestionAlgorithm,DetnetCong
             }
         }
 
-        // 更新最小RTT（BaseRTT）
-        if (RTTns <= RTTMin) {
-            RTTMin = RTTns;
-        } else {
-            // 缓慢适应：当网络路径真正变化时，BaseRTT应能缓慢更新
-            // 这里使用极慢的衰减因子，只有在持续观测到更低RTT时才快速更新
-            RTTMin = (RTTMin * 999 + RTTns) / 1000;
-        }
         
         // 更新平滑RTT估计
-        
+       rttEstimator.putRTT(RTTns);
-            RTTVar = (RTTVar * 3 + Math.abs(RTTAvg - RTTns)) / 4;
-            RTTAvg = (RTTAvg * 7 + RTTns) / 8;
-        
-        
-        RTO = RTTAvg + Math.max(MIN_RTTVAR, RTTVar * 4);
         
         RTTTotal += RTTns;
         RTTCount++;
-        
+        if(slock.tryLock())try {
             long curr = System.nanoTime();
             // 每个RTT周期调整一次窗口（使用当前估计的RTT）
             if (curr - RTTstartTime > Math.max(BIAS, 2 * RTTMin)) {
+                rttEstimator.update();
+                RTTMin = rttEstimator.getBaseRTT();
+
+
                 long RTTCountx = RTTCount;
                 if (RTTCountx > 0) {
                     long currentRTT = RTTTotal / RTTCountx;
@@ -206,24 +104,23 @@ public class Vegas2CongestionAlgorithm implements CongestionAlgorithm,DetnetCong
                 // 2. 基于比值的窗口调整（取代原来的基于差值的调整）
                 if (delayRatio > delayUpperBound) {
                     // 时延过高：减小窗口，减少幅度与超标程度成正比
-                window2 -= 4096;
+                    window2 = window2 - 4096;
                 } else if (delayRatio < delayLowerBound) {
                     // 时延过低：增大窗口，增加幅度与低于目标程度成正比
-                window2 += 4096;
+                    window2 = window2 + 4096;
                 } else {
-              
+                    //window2=window2+4096
                 }
                 // ===================================================
 
 
-            
-            
                 updateWindowSize();
                 RTTstartTime = curr;
 
 
-            
+            }
-            
+        }finally{
+            slock.unlock();
         }
     }
     
@@ -245,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
@@ -269,12 +166,14 @@ public class Vegas2CongestionAlgorithm implements CongestionAlgorithm,DetnetCong
     }
     
     public long getBaseRTT() {
-        return RTTMin;
+        return rttEstimator.getBaseRTT();
     }
-    
+    public long getAvgRTT() {
-    public long getCurrentRTT() {
         return RTTAvg2;
     }
+    public long getCurrentRTT() {
+        return RTTCurr;
+    }
 
 	@Override
 	public void setUpperDelayBound(double upper) {
@@ -295,4 +194,19 @@ public class Vegas2CongestionAlgorithm implements CongestionAlgorithm,DetnetCong
 		return "Vegas2";
 	}
 
+	@Override
+	public double getUpperDelayBound() {
+		return delayUpperBound;
+	}
+
+	@Override
+	public double getLowerDelayBound() {
+		return delayLowerBound;
+	}
+
+	@Override
+	public double getBurstLimit() {
+		return burstLimit;
+	}
+
 }

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

@@ -1,5 +1,8 @@
 package org.kne.cloud.network.ipv6;
 
+import java.util.ArrayList;
+import java.util.Iterator;
+import java.util.List;
 import java.util.concurrent.CopyOnWriteArraySet;
 import java.util.function.BiConsumer;
 import java.util.function.Supplier;
@@ -10,7 +13,21 @@ public abstract class AbstractIPv6NetworkLink implements IPv6NetworkLink {
 	private CopyOnWriteArraySet<IPv6LinkStateListener> listeners = new CopyOnWriteArraySet<>();
 	
 	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;
@@ -53,8 +70,54 @@ public abstract class AbstractIPv6NetworkLink implements IPv6NetworkLink {
 	}
 
 	private BiConsumer<IPv6NetworkLink, Supplier<IPv6Packet>> receiveConsumer;
+
+	private Supplier<Long> costSupplier;
 	@Override
 	public void setReceiveConsumer(BiConsumer<IPv6NetworkLink,Supplier< IPv6Packet>> receiveConsumer) {
 		this.receiveConsumer = receiveConsumer;
 	}
+	
+	@Override
+	public boolean isLoopBack() {
+		return loopback;
+	}
+
+	public Supplier<Long> getCostSupplier() {
+		return costSupplier;
+	}
+
+	public void setCostSupplier(Supplier<Long> costSupplier) {
+		this.costSupplier = costSupplier;
+	}
+
+	@Override
+	public List<RouteItem> getRouteItems() {
+		List<RouteItem> rlist = new ArrayList<>();
+		if(isLoopBack()) {
+		for(IPv6AddressGroup group:getAddressGroups()) {
+		rlist.add(new RouteItem(new IPv6AddressGroup(group.getAddress(), 128),
+				group.getAddress(), this, RouteItem.DIRECT, 0, 0, null, "D"));
+		}
+		}
+		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, costSupplier, "D");
+			rlist.add(ri);
+
+			RouteItem ris = new RouteItem(addresses.getLocator(),
+					 addresses.getLocator().getAddress(), this, RouteItem.SRv6_ENDSID, 13, 128,
+					 costSupplier, "D");
+			rlist.add(ris);
+
+		}
+		}
+		return rlist;
+	}
+
+	@Override
+	public boolean isFilter() {
+		return filter;
+	}
 }

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

@@ -61,6 +61,10 @@ public final class IPv6Address implements Comparable<IPv6Address> {
 	                  ((long)(bytes[15] & 0xFF));
 	}
 
+	public IPv6Address(String string) throws UnknownHostException {
+		this((Inet6Address) Inet6Address.getByName(string));
+	}
+
 	/**
      * 从两个long创建IPv6地址
      */

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

@@ -67,7 +67,16 @@ public class IPv6AddressGroup implements Comparable<IPv6AddressGroup>{
 		prefixLength=in.read();
 	}
 	public boolean checkMatch(IPv6Address address2) {
-		return address2.equals(address.maskWith(IPv6Address.createMask(prefixLength)));
+		IPv6Address cmsk= IPv6Address.createMask(prefixLength);
+		return address2.maskWith(cmsk).equals(address.maskWith(cmsk));
+	}
+	/**
+	 * 返回裁剪后的路由表地址（主机位清零）。
+	 * 例如 2001:db8:1::100/64 → 2001:db8:1::/64
+	 */
+	public IPv6AddressGroup toNetworkRoute() {
+	    IPv6Address mask = IPv6Address.createMask(prefixLength);
+	    IPv6Address maskedAddr = address.maskWith(mask);
+	    return new IPv6AddressGroup(maskedAddr, prefixLength);
 	}
-
 }

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/ipv6/IPv6Packet.java

@@ -1378,7 +1378,7 @@ public class IPv6Packet extends NetworkPacket {
     // 重路由计数器
     private AtomicInteger rerouteCounter = new AtomicInteger(0);
 
-    public AtomicInteger getRerouteCounter() {
+    public AtomicInteger getRouteCounter() {
         return rerouteCounter;
     }
 

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

@@ -41,6 +41,7 @@ public class IPv6TUNLoopbackNetworkLink extends AbstractIPv6NetworkLink
 	private Lock slok = new SpinLock();
 
 	public IPv6TUNLoopbackNetworkLink(String name,IPv6AddressGroup hostAddress, int mtu, List<InetAddress> dns) throws IOException {
+		super(true);
 		if (tun != null)
 			throw new IllegalStateException("already open!");
 		try {
@@ -204,13 +205,6 @@ public class IPv6TUNLoopbackNetworkLink extends AbstractIPv6NetworkLink
 			LockSupport.unpark(tr);
 	}
 
-	
-
-	@Override
-	public boolean isLoopBack() {
-		return true;
-	}
-
 	@Override
 	public List<Neighbor> getNeighborsInfo() {
 		List<Neighbor> hs = new ArrayList<>();
@@ -219,10 +213,6 @@ public class IPv6TUNLoopbackNetworkLink extends AbstractIPv6NetworkLink
 		return hs;
 	}
 
-	/*
-	 * @Override public boolean isCongress(IPv6Packet iPv6Packet,double scale) {
-	 * return sendQueue.size() > 1000*scale; }
-	 */
 
 	@Override
 	public String getName() {
@@ -255,27 +245,4 @@ public class IPv6TUNLoopbackNetworkLink extends AbstractIPv6NetworkLink
 		return List.of(hostAddress);
 	}
 
-	@Override
-	public List<RouteItem> getRouteItems() {
-		List<RouteItem> rlist = new ArrayList<>();
-
-		for (IPv6AddressGroup grp : getAddressGroups()) {
-			rlist.add(new RouteItem(new IPv6AddressGroup(grp.getAddress(), 128), grp.getAddress(), this, "Direct", 0, 1,
-					null, "D", true));
-		}
-
-		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, "Direct", 0, 128, null, "D", false);
-			rlist.add(ri);
-
-			RouteItem ris = new RouteItem(addresses.getLocator(), addresses.getLocator().getAddress(), this,
-					"KLALB SRv6", 13, 128, null, "D", false);
-			rlist.add(ris);
-
-		}
-		return rlist;
-	}
-
 }

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

@@ -32,6 +32,7 @@ public class LoopbackIPv6NetworkLink extends AbstractIPv6NetworkLink implements
 		
 		//new Inet6AddressGroup(loopbackAddress, 128) new Inet6AddressGroup((Inet6Address) Inet6Address.getByName("::1"), 128)
 		public LoopbackIPv6NetworkLink(List<IPv6AddressGroup> addressGroupsx,SRv6Router router) {
+			super(true);
 			this.addressGroups .addAll( addressGroupsx);
 		}
 		
@@ -54,10 +55,7 @@ public class LoopbackIPv6NetworkLink extends AbstractIPv6NetworkLink implements
 			}
 		}
 
-		@Override
+		
-		public boolean isLoopBack() {
-			return true;
-		}
 
 		@Override
 		public List<Neighbor> getNeighborsInfo() {
@@ -93,29 +91,6 @@ public class LoopbackIPv6NetworkLink extends AbstractIPv6NetworkLink implements
 			return rerouteConsumer;
 		}
 
-		@Override
-		public List<RouteItem> getRouteItems() {
-			List<RouteItem> rlist = new ArrayList<>();
-			for(IPv6AddressGroup group:addressGroups) {
-			rlist.add(new RouteItem(new IPv6AddressGroup(group.getAddress(), 128),
-					group.getAddress(), this, "Direct", 0, 0, null, "D", true));
-			}
-
-			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, "Direct", 0, 128, null, "D",
-						false);
-				rlist.add(ri);
-
-				RouteItem ris = new RouteItem(addresses.getLocator(),
-						 addresses.getLocator().getAddress(), this, "KLALB SRv6", 13, 128,
-						 null, "D", false);
-				rlist.add(ris);
-
-			}
-			return rlist;
-		}
 
 
 		public SRv6Router getRouter() {

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

@@ -4,6 +4,11 @@ import java.util.Objects;
 import java.util.function.Supplier;
 
 public class RouteItem implements Comparable<RouteItem>,Cloneable{
+	public static final String DIRECT="Direct";
+	public static final String SRv6_ENDSID="SRv6 ENDSID";
+	public static final String SRv6_ENDXSID="SRv6 ENDXSID";
+	public static final String KLALB_SRv6 = "KLALB_SRv6";
+	
 	private IPv6AddressGroup destination;
 	private IPv6Address nexthop;
 	private IPv6NetworkLink destlink;
@@ -12,7 +17,6 @@ public class RouteItem implements Comparable<RouteItem>,Cloneable{
 	private long cost;
 	private String flag;
 	private Supplier<Long> costSupplier;
-	private boolean isLoopback;
 	public String getFlag() {
 		return flag;
 	}
@@ -34,21 +38,20 @@ public class RouteItem implements Comparable<RouteItem>,Cloneable{
 				 && pre == other.pre && Objects.equals(proto, other.proto);
 	}
 	public RouteItem(IPv6AddressGroup destination, IPv6Address nexthop, IPv6NetworkLink destlink, String proto, int pre,
-			long cost,String flag,boolean isLoopback) {
+			long cost,String flag) {
 		super();
-		this.destination = destination;
+		this.destination = destination.toNetworkRoute();
 		this.nexthop = nexthop;
 		this.destlink = destlink;
 		this.proto = proto;
 		this.pre = pre;
 		this.cost = cost;
 		this.flag=flag;
-		this.isLoopback=isLoopback;
 	}
 	public RouteItem(IPv6AddressGroup destination, IPv6Address nexthop, IPv6NetworkLink destlink, String proto, int pre,
-			long cost,Supplier<Long> costSupplier,String flag,boolean isLoopback) {
+			long cost,Supplier<Long> costSupplier,String flag) {
 		super();
-		this.destination = destination;
+		this.destination = destination.toNetworkRoute();
 		this.nexthop = nexthop;
 		this.destlink = destlink;
 		this.proto = proto;
@@ -56,7 +59,6 @@ public class RouteItem implements Comparable<RouteItem>,Cloneable{
 		this.cost = cost;
 		this.costSupplier=costSupplier;
 		this.flag=flag;
-		this.isLoopback=isLoopback;
 	}
 	public IPv6AddressGroup getDestination() {
 		return destination;
@@ -73,9 +75,6 @@ public class RouteItem implements Comparable<RouteItem>,Cloneable{
 	public int getPre() {
 		return pre;
 	}
-	public boolean isLoopback() {
-		return isLoopback;
-	}
 	@Override
 	public String toString() {
 		return getDestination()+"\t"+getProto()+"\t"+getPre()+"\t"+getCost()+"\t"+getFlag()+"\t"+getNexthop()+"\t"+getDestlink().getName();

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

@@ -13,6 +13,10 @@ import org.kne.cloud.network.ipv6.RouteItem;
 
 public abstract class AbstractControlledIPv6NetworkLink extends AbstractIPv6NetworkLink implements ControlledIPv6NetworkLink{
 
+	public AbstractControlledIPv6NetworkLink(boolean loopback) {
+		super(loopback);
+	}
+
 	
 
 }

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

@@ -12,7 +12,7 @@ public class CONST {
 
 	public static final String KLALB_S_RV6 = "KLALB_SRv6";
 	
-	
+	public static final String KLTP_VIRTUAL_SOCKET_NAME="kltp";
 
 	public static final int itemwidth = 1000;
 	public static final int linepanelheight = 41;

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.*;
-import java.util.concurrent.CopyOnWriteArrayList;
-import java.util.concurrent.Executors;
-import java.util.concurrent.ScheduledExecutorService;
-import java.util.concurrent.ThreadFactory;
 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));
@@ -86,7 +80,7 @@ public class KLALBController {
 	private static final boolean debug=false;
 	private static final boolean showpacket = false;
 
-	private static final int PREFIX = 128;// 112
+	private static final int PREFIX = 128;
 	private static final int DISCOVERY_PORT = 4569;
 
 	private static List<IPMulticastDiscovery> ipmd = new ArrayList<>();
@@ -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 Runnable tsk1 = new Runnable() {
-	private TimerTask tsk1 = new TimerTask() {
 
 		@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() {
@@ -634,7 +614,8 @@ public class KLALBController {
 	
 	private void loadSRv6ProtocolStack(IPv6AddressGroup selfx, boolean enableVirtualAdapter) {
 		networkInterfaceManager.getInetAddressesExcept().add(selfx.getAddress().toInet6Address());
-		srv6Router = new SRv6Router(selfx, clock);
+		IPv6AddressGroup group=new IPv6AddressGroup(selfx.getAddress(),32);
+		srv6Router = new SRv6Router(selfx,group, clock);
 		if(configItem!=null) {
 			srv6Router.setPerformanceStrategy(PerformanceStrategy.fromDescription(configItem.getPerformanceStrategy()));
 			srv6Router.setDeviceName(configItem.getDeviceName());
@@ -653,8 +634,7 @@ public class KLALBController {
 		if (enableTUN) {
 			Thread t=new Thread(()->{
 			try {
-				IPv6TUNLoopbackNetworkLink tunlink = new IPv6TUNLoopbackNetworkLink(name.trim(),
+				IPv6TUNLoopbackNetworkLink tunlink = new IPv6TUNLoopbackNetworkLink(name.trim(),group, SRv6Router.MTU, dnsAddresses);
-						new IPv6AddressGroup(srv6Router.getLocator().getAddress(), 32), SRv6Router.MTU, dnsAddresses);
 				tunlink.setMonitor(datatMonitor);
 				// srv6Router.getLinkTabel().add(tunlink);
 				srv6Router.getinLoopback().setFallbackLink(tunlink);
@@ -686,8 +666,8 @@ 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));
+					new MultiProtocolSocketAddress(iproxyname + "datagram","0.0.0.0",123));
 			srv6Router.addSRv6RouterListener(new SRv6RouterListener() {
 
 				@Override
@@ -698,7 +678,7 @@ public class KLALBController {
 					for (Neighbor neighbor : nb) {
 						if (neighbor.getLocator() != null)
 							sp.add(new NTPPeer(
-									new MultiProtocolSocketAddress(iproxyname + "_Datagram",
+									new MultiProtocolSocketAddress(iproxyname + "datagram",
 											neighbor.getLocator().getAddress().toString(), 123),
 									NTPv4Packet.NTP_SYMMETRIC_ACTIVE));
 
@@ -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);
+		KLALBScheduler.getScheduler().scheduleAtFixedRate(tsk1, 5000, 5000,TimeUnit.MILLISECONDS);
-		twk.schedule(tsk2, 5000, 5000);
+		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("线路性能采样线程", ()->{
+		KLALBScheduler.getScheduler().scheduleWithFixedDelay( ()->{
-			while(true) {
 			linkMonitor.update();
+		},1,1,TimeUnit.MILLISECONDS);
+		KLALBScheduler.getScheduler().scheduleWithFixedDelay( ()->{
 		datatMonitor.update();
-			try {
+		},1,1,TimeUnit.MILLISECONDS);
-				Thread.sleep(1);
-			} catch (InterruptedException e) {
-				e.printStackTrace();
-			}
-			}
-		}).start();
 		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) {
@@ -978,8 +952,8 @@ public class KLALBController {
 
 
 	public void registerToProxyTypeAs(String proxyname) {
-		MultiProtocolSocketAddress.getSocketTypeRegister().put(proxyname + "_Stream", streamSocketType);
+		MultiProtocolSocketAddress.getSocketTypeRegister().put(proxyname + "stream", streamSocketType);
-		MultiProtocolSocketAddress.getSocketTypeRegister().put(proxyname + "_Datagram", datagramSocketType);
+		MultiProtocolSocketAddress.getSocketTypeRegister().put(proxyname + "datagram", datagramSocketType);
 		// ProxyProfileEntry.getRegister().put(proxyname, this);
 	}
 

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

@@ -22,8 +22,7 @@ import org.kne.cloud.network.klalb.ui.UIEnv;
 import org.kne.cloud.network.perf.Kperf;
 import org.kne.cloud.network.perf.NodeBenchmark;
 import org.kne.debug.TimeDebugger;
-import org.kne.membandboost.MembandBenchmark;
+import org.kne.membandboost.benchmark.*;;
-import org.kne.membandboost.ObjectPoolBandwidthTest;
 
 public class KLALBMain {
 	public static KLALBStateGUI3 ksg;
@@ -239,10 +238,6 @@ public class KLALBMain {
 		case "membandbenchmark":
 			MembandBenchmark.main(args);
 			break;
-		case "memqueuebenchmark":
-
-			ObjectPoolBandwidthTest.main(args);
-			break;
 		//case "$$SYSTEM:":
 			//System.out.println();
 			//break;

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

@@ -168,9 +168,11 @@ public class KLALBProxySystem {
 				
 				
 				String vsne=kcci.getVirtualSocketName();
-				//if(vsne!=null) {
+				if(vsne!=null) {
 					MultiProtocolSocketAddress.getSocketTypeRegister().put(vsne, klalbController.getStreamSocketType());
-				//}
+				}else{
+					MultiProtocolSocketAddress.getSocketTypeRegister().put(CONST.KLTP_VIRTUAL_SOCKET_NAME, klalbController.getStreamSocketType());
+				}
 				MultiProtocolSocketAddress tcple=kcci.getTCPListen();
 				if(tcple!=null) {
 					

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

@@ -13,13 +13,15 @@ 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;
 import java.util.concurrent.locks.LockSupport;
 import java.util.function.Consumer;
 
-import org.jctools.queues.MpscArrayQueue;
+import org.jctools.queues.varhandle.MpscVarHandleArrayQueue;
 import org.kne.cloud.clock.HighAccuracyClock;
 import org.kne.cloud.clock.NTPTimestamps;
 import org.kne.cloud.clock.ReliabilityBackoffTimeClock;
@@ -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.*;
-import org.kne.cloud.network.ipv6.IPv6Address;
-import org.kne.cloud.network.ipv6.IPv6HopByHopTLV;
-import org.kne.cloud.network.ipv6.IPv6Packet;
 import org.kne.cloud.network.ipv6.IPv6Packet.IPv6HopByHopHeader;
-import org.kne.cloud.network.ipv6.IPv6AddressGroup;
-import org.kne.cloud.network.ipv6.KLALBOAMHopByHopTLV;
-import org.kne.cloud.network.ipv6.KLALBPassportHopByHopTLV;
-import org.kne.cloud.network.ipv6.Neighbor;
-import org.kne.cloud.network.ipv6.PostcardEntry;
-import org.kne.cloud.network.ipv6.RouteItem;
 import org.kne.cloud.network.monitor.CostSupplierFactory;
 import org.kne.cloud.network.monitor.DelayMonitorData;
 import org.kne.cloud.network.monitor.LinkStatus;
@@ -142,15 +135,18 @@ public class KLALBRemoteLink extends AbstractControlledIPv6NetworkLink implement
 	}
 
 	public KLALBRemoteLink(KLALBController controller,KLALBPacketLink kpl, IPv6AddressGroup address) {
+		super(false);
 		this.klalbController=controller;
 		sysclk = klalbController.getClock();
 		this.kplink = kpl;
 		this.addressGroup = address;
 		monitor=new SpeedAndTrafficAndDelayMonitorDataImpl(sysclk);
 		name=kpl.toString();
+		setCostSupplier( CostSupplierFactory.owdSupplier(monitor));
 	}
 
 	private KLALBRemoteLink(KLALBController controller, MultiProtocolSocketAddress mpa, MultiProtocolSocketAddress bindaddr, IPv6AddressGroup address) {
+		super(false);
 		this.klalbController=controller;
 		sysclk = klalbController.getClock();
 		this.socketAddress = mpa;
@@ -162,6 +158,7 @@ public class KLALBRemoteLink extends AbstractControlledIPv6NetworkLink implement
 		} else {
 			name=bindaddr.toString() + "→" + mpa.toString();
 		}
+		setCostSupplier( CostSupplierFactory.owdSupplier( monitor));
 	}
 
 
@@ -187,7 +184,7 @@ public class KLALBRemoteLink extends AbstractControlledIPv6NetworkLink implement
 	
 	private class RemoteSender implements Runnable {
 		
-		private MpscArrayQueue<KLALBPacket> sendQueue = new MpscArrayQueue<KLALBPacket>(8192);
+		private MpscVarHandleArrayQueue<KLALBPacket> sendQueue = new MpscVarHandleArrayQueue<KLALBPacket>(8192);
 		private WatchDogTimer watchdog = new WatchDogTimer(400000000L);
 		
 		private ThreadParker tlock = new ThreadParker();
@@ -281,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) {
@@ -566,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();
-				}
 				
 			}
 			
@@ -751,8 +745,7 @@ public class KLALBRemoteLink extends AbstractControlledIPv6NetworkLink implement
 	
 	protected void startIO() {
 		stateManager=new StateManager();
-		monitorUpdater=new MonitorUpdater();
+		monitorUpdater=KLALBScheduler.getScheduler().scheduleWithFixedDelay(new MonitorUpdater(),1,1, TimeUnit.MILLISECONDS);
-		ThreadTool.makeVDaemonThread("线路性能采样线程", monitorUpdater).start();
 		ThreadTool.makeVDaemonThread("远程线路状态线程", stateManager).start();
 		
 	}
@@ -854,10 +847,6 @@ public class KLALBRemoteLink extends AbstractControlledIPv6NetworkLink implement
 		}
 	}
 
-	@Override
-	public boolean isLoopBack() {
-		return false;
-	}
 	BandwidthDistributer bandwidthDistributer=new BandwidthDistributer<>();
 	@Override
 	public List<Neighbor> getNeighborsInfo() {
@@ -898,6 +887,8 @@ public class KLALBRemoteLink extends AbstractControlledIPv6NetworkLink implement
 						sendWindow.put(val, pack);
 					//}
 				}
+			}else {
+				System.out.println(hop);
 			}
 			
 			sendIPv6PacketToLink(pack, val);
@@ -946,31 +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()) {
-		IPv6AddressGroup adg = addressGroup;
-		/*if (adg != null)
-			rlist.add(new RouteItem(new IPv6AddressGroup(adg.getAddress(), 128), adg.getAddress(), this, "Direct", 0, 0,
-					null, "D", true));*/
-
-		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, "Direct", 0, 128, CostSupplierFactory.expectedDelaySupplier((DelayMonitorData) addresses.getMonitor(), status, algorithm::getRTO), "D", false);
-			rlist.add(ri);
-
-			if (addresses.getLocator() != null) {
-				RouteItem ris = new RouteItem(addresses.getLocator(), addresses.getLocator().getAddress(), this,
-						"KLALB SRv6", 13, 128, CostSupplierFactory.expectedDelaySupplier((DelayMonitorData) addresses.getMonitor(), status, algorithm::getRTO), "D", false);
-				rlist.add(ris);
-			}
-		}
-	}
-		return rlist;
-	}
 
 	@Override
 	public void setCongressCondition(Lock lock, Condition condition) {
@@ -978,14 +944,32 @@ public class KLALBRemoteLink extends AbstractControlledIPv6NetworkLink implement
 			sendWindow.setCongressCondition(lock, condition);
 	}
 
-	public void onPostcardReceive(PostcardEntry postcard) {
+	public void onPostcardReceive(IPv6NetworkLink linkfrom,PostcardEntry postcard) {
 		long deltaxnanos = (long) (postcard.getTimestampDeltaLong() * 1000000000.0 / (1L << 32));
-		if (deltaxnanos > 0) {
+		Long128 curr= sysclk.getCurrentTimeNanos();
 		Long128 psnd = NTPTimestamps.ntp128BitToNanosTimestamp(
-					NTPTimestamps.inferNtp64To128(postcard.getTimestampBase(), sysclk.getCurrentTimeNTP128()));
+				NTPTimestamps.inferNtp64To128(postcard.getTimestampBase(),NTPTimestamps.nanosToNtp128BitTimestamp(curr)));
+		long rtt=curr.subtract(psnd).longValue();
+		if (deltaxnanos >= 0) {
 			monitor.recordUploadDelay(deltaxnanos);
 		}
-		sendWindow.ack(postcard.getPacketUUID(),false,deltaxnanos);
+		if(linkfrom instanceof KLALBRemoteLink ){
+			long rtt2=deltaxnanos+((KLALBRemoteLink) linkfrom).getMonitor().getInDelayMin();
+
+			if(rtt2<0) {
+				if(rtt<0) {
+					rtt=0;
+				}
+				sendWindow.ack(postcard.getPacketUUID(), rtt, false);
+			}else {
+				sendWindow.ack(postcard.getPacketUUID(), rtt2, false);
+			}
+		}else {
+			if(rtt<0) {
+				rtt=0;
+			}
+			sendWindow.ack(postcard.getPacketUUID(), rtt, false);
+		}
 		
 	}
 

src/org/kne/cloud/network/klalb/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;
+    }
+}