diff --git a/src/org/kne/cloud/network/congestion/SendPacketSlidingWindow.java b/src/org/kne/cloud/network/congestion/SendPacketSlidingWindow.java index 968d4f6..278e40b 100644 --- a/src/org/kne/cloud/network/congestion/SendPacketSlidingWindow.java +++ b/src/org/kne/cloud/network/congestion/SendPacketSlidingWindow.java @@ -265,4 +265,4 @@ public class SendPacketSlidingWindow implements Clos -} +} \ No newline at end of file diff --git a/src/org/kne/cloud/network/klalb/StreamChannelKLALBPacketLink.java b/src/org/kne/cloud/network/klalb/StreamChannelKLALBPacketLink.java index f3ac804..0640dbd 100644 --- a/src/org/kne/cloud/network/klalb/StreamChannelKLALBPacketLink.java +++ b/src/org/kne/cloud/network/klalb/StreamChannelKLALBPacketLink.java @@ -118,46 +118,50 @@ public class StreamChannelKLALBPacketLink extends AbstractKLALBPacketLink implem return true; } - + // 增加两个 ThreadLocal 静态变量,用于复用 4 字节的长度头 + private static final ThreadLocal SIZE_WRITE_BUFFER = ThreadLocal.withInitial(() -> ByteBuffer.allocate(4)); + private static final ThreadLocal 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}); - checkflush(); - - - } - - + // 3. 发送 + writableChannel.write(new ByteBuffer[] {szeWrite, kp}); + checkflush(); + } @Override public ByteBuffer readPacket() throws IOException { - - ByteBuffer szeRead=NetworkPacket.bufferAllocator.allocate(4); - timeoutTimer=System.nanoTime(); - timerenabled=true; - ByteBuffer kp; + // 1. 获取并清空缓存的 4 字节 buffer + ByteBuffer szeRead = SIZE_READ_BUFFER.get(); szeRead.clear(); szeRead.limit(4); - //szeRead.clear(); + + timeoutTimer = System.nanoTime(); + timerenabled = true; + ByteBuffer kp; + try { - KNEChannels.readFully(readableChannel,szeRead); + // 2. 读取长度头 + KNEChannels.readFully(readableChannel, szeRead); szeRead.flip(); - int size=szeRead.getInt(0); - kp=NetworkPacket.bufferAllocator.allocate(size); + int size = szeRead.getInt(0); + + // 3. 分配真正的数据区(这里如果是高频,建议后续也可以池化,不过现在是按需大小分配) + kp = NetworkPacket.bufferAllocator.allocate(size); kp.limit(size); KNEChannels.readFully(readableChannel, kp); kp.flip(); - return kp; - }finally { - timerenabled=false; + return kp; + } finally { + timerenabled = false; } } diff --git a/src/org/kne/cloud/network/ntp/NTPv4Protocol.java b/src/org/kne/cloud/network/ntp/NTPv4Protocol.java index 8ff35a0..f71b991 100644 --- a/src/org/kne/cloud/network/ntp/NTPv4Protocol.java +++ b/src/org/kne/cloud/network/ntp/NTPv4Protocol.java @@ -252,14 +252,23 @@ public class NTPv4Protocol implements Closeable, AutoCloseable { checkLock.unlock(); } } + // 在类成员变量中定义 ThreadLocal 发送缓冲区 + private static final ThreadLocal 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 NTP_BUFFER = ThreadLocal.withInitial(() -> ByteBuffer.allocate(9000)); + private NTPv4Packet receiveNTPPacket(AtomicReference 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; } diff --git a/src/org/kne/cloud/network/srv6/KLALBRoutingProtocol.java b/src/org/kne/cloud/network/srv6/KLALBRoutingProtocol.java index 6ab404e..7a81d2f 100644 --- a/src/org/kne/cloud/network/srv6/KLALBRoutingProtocol.java +++ b/src/org/kne/cloud/network/srv6/KLALBRoutingProtocol.java @@ -9,6 +9,7 @@ import java.net.Inet6Address; import java.net.InetSocketAddress; import java.net.SocketAddress; import java.net.SocketException; +import java.nio.ByteBuffer; import java.nio.channels.Channels; import java.util.ArrayList; import java.util.HashMap; @@ -35,6 +36,7 @@ import org.kne.cloud.network.monitor.MonitorData; import org.kne.cloud.network.monitor.SpeedAndTrafficAndDelayMonitorDataImpl; import org.kne.cloud.network.monitor.SpeedAndTrafficMonitorData; import org.kne.cloud.network.monitor.SpeedAndTrafficMonitorDataImpl; +import org.kne.io.KNEChannels; import org.kne.membandboost.MembandBoost; public class KLALBRoutingProtocol extends Thread{ @@ -211,20 +213,15 @@ public class KLALBRoutingProtocol extends Thread{ while(true) { try { - byte[]ca=MembandBoost.allocateUninitializedByteArray(65535); - DatagramPacket dgp=new DatagramPacket(ca, ca.length); - ds.receive(dgp); - //System.out.println(Arrays.toString( Arrays.copyOf( dgp.getData(),dgp.getLength()))); - ByteArrayInputStream bi=new ByteArrayInputStream(dgp.getData(),0,dgp.getLength()); - KLALBRoutingProtocolPacket kp=KLALBRoutingProtocolPacket.readKLALBPacketFromChannel(Channels.newChannel(bi)); + KLALBRoutingProtocolPacket kp=receivePacket(); switch(kp.getType()) { case KLALBRoutingProtocolPacket.RINFO_REQ: RouterInfo rifr=netmap.get(router.getLocator().getAddress()); if(debug) - System.out.println("get Request from:"+dgp.getSocketAddress()); + System.out.println("get Request from:"+kp.getFrom()); if(rifr!=null) { - writePacket( new RouterInfoPacket( rifr,false,-1), dgp.getSocketAddress()); + writePacket( new RouterInfoPacket( rifr,false,-1), kp.getFrom()); if(debug) System.out.println("response router info:"+rifr); }else { @@ -246,7 +243,7 @@ public class KLALBRoutingProtocol extends Thread{ netmap.put(rif.getLocator().getAddress(), rif); noticeUpdate(); if(rifp.isFlood()) { - floodPacket( dgp.getSocketAddress(), rifp); + floodPacket( kp.getFrom(), rifp); } @@ -259,7 +256,7 @@ public class KLALBRoutingProtocol extends Thread{ break; case KLALBRoutingProtocolPacket.JSON_DATA: JsonDataPacket data=(JsonDataPacket) kp; - receivers.forEach((v)->{v.accept(dgp.getSocketAddress(),data);}); + receivers.forEach((v)->{v.accept(kp.getFrom(),data);}); break; default: break; @@ -276,7 +273,32 @@ public class KLALBRoutingProtocol extends Thread{ } } } - + private static final ThreadLocal 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(); Listln= router.getNeighbors(); @@ -304,20 +326,34 @@ public class KLALBRoutingProtocol extends Thread{ public void sendJsonPacketToAddress(JsonDataPacket pkt,SocketAddress dest) throws IOException { writePacket(pkt,dest); } - - 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); + // 1. 在类成员变量处定义一个 64KB 的 ThreadLocal 缓冲区 +// 注意:这里依然使用 65535,因为你说 LSDB 同步可能需要这么大 + private static final ThreadLocal LSDB_TX_BUFFER = + ThreadLocal.withInitial(() -> ByteBuffer.allocate(65535)); + + private void writePacket(KLALBRoutingProtocolPacket pkt, SocketAddress dest) throws IOException { + // 2. 从 ThreadLocal 获取,并清空状态 + ByteBuffer bbf = LSDB_TX_BUFFER.get(); + bbf.clear(); + + // 3. 使用 KNEChannels 绑定到 ByteBuffer,直接写入 + // 这避免了 ByteArrayOutputStream 内部的动态扩容和拷贝 + KLALBRoutingProtocolPacket.writeKLALBPacketToChannel(KNEChannels.newWritableChannel(bbf), pkt); + + // 4. 翻转,准备读取数据 + bbf.flip(); + + // 5. 直接拿底层的 byte[] 和真实的 limit 发送 + DatagramPacket dgpx = new DatagramPacket(bbf.array(), bbf.limit()); dgpx.setSocketAddress(dest); + sendLock.lock(); try { - ds.send(dgpx); - }catch(BindException e) { + ds.send(dgpx); + } catch (BindException e) { e.printStackTrace(); - System.err.println("Cannot assign:"+dest); - }finally { + System.err.println("Cannot assign:" + dest); + } finally { sendLock.unlock(); } } diff --git a/src/org/kne/cloud/network/srv6/KLALBRoutingProtocolPacket.java b/src/org/kne/cloud/network/srv6/KLALBRoutingProtocolPacket.java index 6986878..d19c4b4 100644 --- a/src/org/kne/cloud/network/srv6/KLALBRoutingProtocolPacket.java +++ b/src/org/kne/cloud/network/srv6/KLALBRoutingProtocolPacket.java @@ -4,6 +4,7 @@ import java.io.DataInputStream; import java.io.DataOutputStream; import java.io.EOFException; import java.io.IOException; +import java.net.SocketAddress; import java.nio.ByteBuffer; import java.nio.channels.Channels; import java.nio.channels.ReadableByteChannel; @@ -133,4 +134,11 @@ public abstract class KLALBRoutingProtocolPacket extends NetworkPacket { public static void writeKLALBPacketToChannel(WritableByteChannel writableByteChannel,KLALBRoutingProtocolPacket klb) throws IOException { klb.writeToChannel(writableByteChannel); } + private transient SocketAddress from; + public void setFrom(SocketAddress from) { + this.from=from; + } + public SocketAddress getFrom(){ + return from; + } } diff --git a/src/org/kne/cloud/network/srv6/SRv6Router.java b/src/org/kne/cloud/network/srv6/SRv6Router.java index cf4216c..4d2eb7a 100644 --- a/src/org/kne/cloud/network/srv6/SRv6Router.java +++ b/src/org/kne/cloud/network/srv6/SRv6Router.java @@ -13,6 +13,7 @@ import java.util.List; import java.util.Map; import java.util.Set; import java.util.UUID; +import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.CopyOnWriteArrayList; import java.util.concurrent.CopyOnWriteArraySet; import java.util.concurrent.TimeUnit; @@ -204,34 +205,19 @@ public class SRv6Router { srListeners.remove(listener); } public void sortRouteTabel() { + List>> routeTabel1x = routeTabel1; - // 构建前缀长度索引的路由表 - List>> routeTabel1x = new ArrayList>>( - IPv6_BITS + 1); - for (int i = 0; i < IPv6_BITS + 1; i++) { - routeTabel1x.add(new HashMap>(1024, 0.5f)); + for (Map> map : routeTabel1x) { + // 使用 compute 直接替换 + map.replaceAll((key, oldList) -> { + if (oldList == null || oldList.isEmpty()) { + return oldList; + } + List newList = new ArrayList<>(oldList); + Collections.sort(newList); + return newList; + }); } - List routeTabel0x=routeTable0; - // 按前缀长度组织路由表 - for (RouteItem item : routeTabel0x) { - IPv6AddressGroup iag = item.getDestination(); - Map> map = routeTabel1x.get(iag.getPrefixLength()); - List newlist = new ArrayList(); - List oldlist = map.putIfAbsent(iag.getAddress(), newlist); - if (oldlist == null) { - oldlist = newlist; - } - oldlist.add(item); - } - for(Map> map:routeTabel1x) { - for(Listv:map.values()) { - Collections.sort(v); - } - } - // System.out.println(routeTabel1x); - routeTabel1 = routeTabel1x; // 更新索引路由表 - - } // 更新路由表 public void updateRouteTabel() { @@ -264,7 +250,7 @@ public class SRv6Router { ArrayList>> routeTabel1x = new ArrayList>>( IPv6_BITS + 1); for (int i = 0; i < IPv6_BITS + 1; i++) { - routeTabel1x.add(new HashMap>(1024, 0.5f)); + routeTabel1x.add(new ConcurrentHashMap>()); } // 按前缀长度组织路由表