KNEMC/KLALB
1
2
Fork 1
Code Issues Pull Requests Actions Packages Projects Releases 1 Wiki Activity

内存优化,减轻GC压力

Browse Source
This commit is contained in:
KNEMC
2026-10-29 13:40:01 +08:00
parent b898e4c5be
commit c0e73bce55
6 changed files with 134 additions and 76 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/org/kne/cloud/network/klalb/StreamChannelKLALBPacketLink.java
+26 -22
View File
@@ -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;
} finally {
timerenabled = false;
}
}
src/org/kne/cloud/network/ntp/NTPv4Protocol.java
+27 -3
View File
@@ -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/srv6/KLALBRoutingProtocol.java
+54 -18
View File
@@ -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) {
} 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
+8
View File
@@ -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/SRv6Router.java
+12 -26
View File
@@ -13,6 +13,7 @@ import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.UUID;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.CopyOnWriteArraySet;
import java.util.concurrent.TimeUnit;
@@ -204,34 +205,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> 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;
List<RouteItem> newList = new ArrayList<>(oldList);
Collections.sort(newList);
return 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() {
@@ -264,7 +250,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>>());
}
// 按前缀长度组织路由表