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

KLALB 3.6.0开发一半的状态

Browse Source
This commit is contained in:
KNEMC
2026-07-06 17:03:27 +08:00
parent 620afef715
commit c8b183da71
90 changed files with 1963 additions and 3757 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/org/kne/cloud/network/srv6/SRv6Router.java
+160 -101
View File
@@ -11,32 +11,22 @@ import java.util.HashSet;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import java.util.Random;
import java.util.Set;
import java.util.Timer;
import java.util.TimerTask;
import java.util.UUID;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.CopyOnWriteArraySet;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadLocalRandom;
import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;
import java.util.function.BiConsumer;
import java.util.function.Consumer;
import java.util.function.Supplier;
import javax.swing.plaf.synth.SynthOptionPaneUI;
import org.jctools.counters.FixedSizeStripedLongCounter;
import org.kne.cloud.clock.HighAccuracyClock;
import org.kne.cloud.network.ThreadTool;
import org.kne.cloud.network.congestion.ArrayListMessageBatcher;
import org.kne.cloud.network.congestion.MessageBatcher;
import org.kne.cloud.network.ipv6.ControlledIPv6NetworkLink;
import org.kne.cloud.network.ipv6.FlowSession;
import org.kne.cloud.network.ipv6.ICMPv6Packet;
import org.kne.cloud.network.ipv6.ICMPv6PostcardPacket;
import org.kne.cloud.network.ipv6.ICMPv6TimeExceededPacket;
import org.kne.cloud.network.ipv6.IPv6Address;
@@ -46,24 +36,20 @@ import org.kne.cloud.network.ipv6.IPv6NetworkLink;
import org.kne.cloud.network.ipv6.IPv6Packet;
import org.kne.cloud.network.ipv6.IPv6Packet.IPv6HopByHopHeader;
import org.kne.cloud.network.ipv6.IPv6Packet.IPv6SegmentRoutingHeader;
import org.kne.cloud.network.ipv6.IPv6TUNLoopbackNetworkLink;
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.LoopbackIPv6NetworkLink;
import org.kne.cloud.network.ipv6.Neighbor;
import org.kne.cloud.network.ipv6.PacketID;
import org.kne.cloud.network.ipv6.PacketIDGenerator;
import org.kne.cloud.network.ipv6.PadNHopByHopTLV;
import org.kne.cloud.network.ipv6.PostcardEntry;
import org.kne.cloud.network.ipv6.RouteItem;
import org.kne.cloud.network.klalb.KLALBRemoteLine;
import org.kne.cloud.network.klalb.KLALBRemoteLink;
import org.kne.cloud.network.klalb.KLALBUtils;
import org.kne.concurrent.HighPerformanceExecutor;
import org.kne.concurrent.HighPerformanceExecutor2;
import org.kne.concurrent.TimeoutConcurrentHashMap;
import org.pcap4j.packet.IllegalRawDataException;
import com.github.f4b6a3.uuid.UuidCreator;
import com.google.gson.internal.Pair;
// SRv6路由器主类，实现IPv6段路由功能
@@ -132,7 +118,6 @@ public class SRv6Router {
public void accept(IPv6Packet t) {
// HighPerformanceExecutor.defaultExecutor.execute(() -> {
routePacket(null, t, true); // 执行重路由
t.putTimePassport("rerouted"); // 标记重路由时间点
// });
}
};
@@ -268,7 +253,7 @@ public class SRv6Router {
// 预处理和排序路由项
for (Iterator<RouteItem> iterator = routeTabel0x.iterator(); iterator.hasNext();) {
RouteItem routeItem = (RouteItem) iterator.next();
routeItem.preSort(); // 预排序
routeItem.updateCost(); // 预排序
}
Collections.sort(routeTabel0x); // 排序路由表
@@ -319,15 +304,11 @@ public class SRv6Router {
// 插入SRH并路由数据包
public void insertSRHandRoutePacket(IPv6NetworkLink link, IPv6Packet ipp) {
insertHopByHopHeader(ipp);
ipp.putTimePassport("Hop inserted");
insertSRHeader(ipp); // 插入段路由头
ipp.putTimePassport("SRH inserted"); // 标记SRH插入时间点
routePacket(link, ipp); // 路由数据包
ipp.putTimePassport("routed"); // 标记路由时间点
}
private PacketIDGenerator defaultGen = new PacketIDGenerator();
private static final PadNHopByHopTLV pad4 = new PadNHopByHopTLV(4);
// 插入逐跳路由头
@@ -353,16 +334,15 @@ public class SRv6Router {
}
// 如果KLALB路由协议可用，创建段列表
if (klalbRouteProtol != null) {
List<IPv6Address> segs = klalbRouteProtol.createSegmentList(ipp.getDestinationAddress());
IPv6SegmentRoutingHeader segs = klalbRouteProtol.createSegmentList(ipp.getDestinationAddress());
// System.out.println(segs);
if (segs != null && (!segs.isEmpty())) {
if (segs != null && (!segs.getAddresses().isEmpty())) {
// 创建SRH头
IPv6SegmentRoutingHeader irh = new IPv6SegmentRoutingHeader(segs);
ipp.getHeaders().add(irh); // 添加SRH头
ipp.setRoutingHeader(irh);
ipp.setDestinationAddress(segs.get(segs.size() - 1)); // 更新目的地址为最后一个段
ipp.getHeaders().add(segs); // 添加SRH头
ipp.setRoutingHeader(segs);
ipp.setDestinationAddress(segs.getAddresses().get(segs.getAddresses().size() - 1)); // 更新目的地址为最后一个段
return true;
}
@@ -401,8 +381,9 @@ public class SRv6Router {
int hop = iPv6Packet.getHopLimit(); // 获取跳数限制
if (hop > 0) { // 检查TTL值，只有大于0才会被转发
// 搜索匹配的路由表项
List<RouteItem> searchResult = searchTable(iPv6Packet, reroute, dbg);
if (!searchResult.isEmpty()) {
IPv6Address dest=iPv6Packet.getDestinationAddress();
List<RouteItem> searchResult = getTabelByAddress(dest);
if (searchResult!=null&&(!searchResult.isEmpty())) {
// 匹配到路由表，执行负载均衡路由
routingLoadBalance(linkfrom, iPv6Packet, reroute, dbg, searchResult);
return;
@@ -421,15 +402,15 @@ public class SRv6Router {
segmentsLeft--;
}
// 创建修复段列表
List<IPv6Address> repairSegments = klalbRouteProtol
IPv6SegmentRoutingHeader repairSegments = klalbRouteProtol
.createSegmentList(srh.getAddresses().get(segmentsLeft));
if (repairSegments == null || repairSegments.isEmpty()) {
if (repairSegments == null ||repairSegments.getAddresses().isEmpty()) {
System.out.println("FRR保护失败");
return;
}
repairSegments.remove(0); // 移除第一个段（当前节点）
srh.getAddresses().addAll(segmentsLeft + 1, repairSegments); // 添加修复段
segmentsLeft += repairSegments.size(); // 更新剩余段数
repairSegments.getAddresses().remove(0); // 移除第一个段（当前节点）
srh.getAddresses().addAll(segmentsLeft + 1, repairSegments.getAddresses()); // 添加修复段
segmentsLeft += repairSegments.getAddresses().size(); // 更新剩余段数
srh.setSegmentsLeft(segmentsLeft);
iPv6Packet.setDestinationAddress(srh.getAddresses().get(segmentsLeft)); // 更新目的地址
} else {
@@ -437,8 +418,8 @@ public class SRv6Router {
}
System.out.println("FRR修复SRH：" + srh);
// 重新搜索路由表
List<RouteItem> searchResult2 = searchTable(iPv6Packet, reroute, dbg);
if (!searchResult2.isEmpty()) {
List<RouteItem> searchResult2 = getTabelByAddress(dest);
if (searchResult2!=null&&(!searchResult2.isEmpty())) {
routingLoadBalance(linkfrom, iPv6Packet, reroute, dbg, searchResult2);
return;
}
@@ -457,7 +438,7 @@ public class SRv6Router {
}
// 搜索路由表找到匹配的路由项
private List<RouteItem> searchTable(IPv6Packet iPv6Packet, boolean reroute, StringBuilder dbg)
/*private List<RouteItem> searchTable(IPv6Packet iPv6Packet, StringBuilder dbg)
throws IllegalRawDataException, IOException {
if (debug) {
dbg.append("----------------------------------------\n");
@@ -476,17 +457,7 @@ public class SRv6Router {
// 遍历路由表，筛选可用路由
for (RouteItem tri : table) {
if (!tri.getDestlink().isUp()) { // 检查链路状态
if (debug) {
dbg.append(tri + "\n");
dbg.append("linkdown.\n");
}
continue; // 跳过断开连接的链路
}
if (debug) {
dbg.append(tri + "\n");
dbg.append("matched.\n");
}
// 检查ECMP（等价多路径）条件
if (prevr != null) {
if (!tri.ECMPequals(prevr)) {
@@ -497,7 +468,7 @@ public class SRv6Router {
prevr = tri;
}
return routes;
}
}*/
private List<RouteItem> getTabelByAddress(IPv6Address rk6) {
List<RouteItem> table = null;
@@ -519,35 +490,25 @@ public class SRv6Router {
private boolean routingLoadBalance(IPv6NetworkLink linkfrom, IPv6Packet iPv6Packet, boolean reroute,
StringBuilder dbg, List<RouteItem> routes) throws IllegalRawDataException, IOException {
double cscale = 1.0; // 拥塞检测比例因子
if (iPv6Packet.getPayload().getProtocolNumber() == 6) { // TCP协议
cscale = 15.0; // TCP使用更高的拥塞检测阈值
if(routes==null||routes.isEmpty()) {
//System.out.println("Route not found:" + iPv6Packet);
return false;
}
if (iPv6Packet.getPreviousLink() != null) {
List<RouteItem> prev = new ArrayList<>();
for (Iterator<RouteItem> iterator = routes.iterator(); iterator.hasNext();) {
RouteItem routeItem = (RouteItem) iterator.next();
if (routeItem.getDestlink().equals(iPv6Packet.getPreviousLink())) {
prev.add(routeItem);
iterator.remove();
}
}
routes.addAll(prev);
}
for (double i = 1; i < 10; i += 0.1) {
if (loadBalance(linkfrom, iPv6Packet, reroute, dbg, routes, cscale * i)) {
for (double i = 1; i < 100; i += 0.1) {
if (loadBalance(linkfrom, iPv6Packet, reroute, dbg, routes, i)) {
return true;
}
if (iPv6Packet.isEnableECN()) { // 启用ECN
iPv6Packet.markCE(); // 标记拥塞
if (debug) {
dbg.append("ECN enabled.\n");
}
}
}
System.out.println("Send failed:" + iPv6Packet);
//System.out.println("Send failed:" + iPv6Packet);
LosscountAdder.inc(1);
if (debug) {
dbg.append("miss.\n"); // 路由失败
}
@@ -568,7 +529,7 @@ public class SRv6Router {
continue; // 跳过断开链路
}
if (link instanceof ControlledIPv6NetworkLink) {
if (((ControlledIPv6NetworkLink) link).isCongress(iPv6Packet, cscale)) { // 检查拥塞
if (((ControlledIPv6NetworkLink) link).isCongestion(iPv6Packet, cscale)) { // 检查拥塞
if (debug) {
dbg.append(tri + "\n");
dbg.append("congress.\n");
@@ -623,12 +584,6 @@ public class SRv6Router {
if (iPv6Packet.getPayload() instanceof ICMPv6PostcardPacket && ri.getDestlink().isLoopBack()) {
onPostcardReceive(iPv6Packet);
} else {
IPv6NetworkLink oldLink = iPv6Packet.getPreviousLink();
/*
* if(oldLink!=null) {
* System.out.println("重传:"+!oldLink.equals(ri.getDestlink())); }
*/
iPv6Packet.setPreviousLink(ri.getDestlink());
ri.getDestlink().sendPacket(iPv6Packet, ri.getNexthop());
}
}
@@ -642,8 +597,8 @@ public class SRv6Router {
if (ris != null) {
for (RouteItem ri : ris) {
if (ri.getNexthop().equals(sid)) {
if (ri.getDestlink() instanceof KLALBRemoteLine) {
KLALBRemoteLine rem = (KLALBRemoteLine) ri.getDestlink();
if (ri.getDestlink() instanceof KLALBRemoteLink) {
KLALBRemoteLink rem = (KLALBRemoteLink) ri.getDestlink();
rem.onPostcardReceive(postcard);
}
}
@@ -683,9 +638,9 @@ public class SRv6Router {
int sl = srhh.getSegmentsLeft();
// IPv6Address prevSID
// =sl+1>=srhh.getAddresses().size()?iPv6Packet.getSourceAddress():srhh.getAddresses().get(sl+1);
if (linkfrom instanceof KLALBRemoteLine) {
if (linkfrom instanceof KLALBRemoteLink) {
processKLALBOAM(iPv6Packet, linkfrom.getAddressGroups().get(0).getAddress(),
((KLALBRemoteLine) linkfrom).getRemoteVaddr().getAddress(), iPv6HopByHopHeader);
((KLALBRemoteLink) linkfrom).getRemoteVaddr().getAddress(), iPv6HopByHopHeader);
}
}
// System.out.println(iPv6HopByHopHeader);
@@ -698,9 +653,9 @@ public class SRv6Router {
if (iPv6HopByHopHeader != null) {
// IPv6Address prevSID
// =oldSL+1>=srhh.getAddresses().size()?iPv6Packet.getSourceAddress():srhh.getAddresses().get(oldSL+1);
if (linkfrom instanceof KLALBRemoteLine) {
if (linkfrom instanceof KLALBRemoteLink) {
processKLALBOAM(iPv6Packet, linkfrom.getAddressGroups().get(0).getAddress(),
((KLALBRemoteLine) linkfrom).getRemoteVaddr().getAddress(), iPv6HopByHopHeader);
((KLALBRemoteLink) linkfrom).getRemoteVaddr().getAddress(), iPv6HopByHopHeader);
}
}
// 正常SRv6处理：移动到下一个段
@@ -762,8 +717,9 @@ public class SRv6Router {
private void putPostcard(IPv6Address src, IPv6Address dst, IPv6Address srcSID, UUID uuid, long timeStamp,
long delta, int packetLength) {
PostcardEntry pn = new PostcardEntry(srcSID, uuid, timeStamp, (int) (delta >> 4), packetLength);
Pair<IPv6Address, IPv6Address> addressPair = new Pair<IPv6Address, IPv6Address>(src, dst);
/*Pair<IPv6Address, IPv6Address> addressPair = new Pair<IPv6Address, IPv6Address>(src, dst);
MessageBatcher<PostcardEntry> batch = postcardBatch.computeIfAbsent(addressPair, (k) -> {
MessageBatcher<PostcardEntry> mb = new ArrayListMessageBatcher<PostcardEntry>(30, 2000000L);
mb.setConsumer((list) -> {
@@ -786,13 +742,30 @@ public class SRv6Router {
return mb;
});
postcardBatch.refresh(addressPair);
batch.putMessage(pn);
batch.putMessage(pn);*/
enqueuePacketSendTask(() -> {
IPv6Packet icmpv = new IPv6Packet();
icmpv.setSourceAddress(src);
icmpv.setDestinationAddress(dst);
icmpv.setTrafficClass(0);
icmpv.setVersion(6);
icmpv.setFlowLabel(0);
icmpv.setHopLimit(255); // ICMP消息TTL设为最大值
ICMPv6PostcardPacket ipl = new ICMPv6PostcardPacket();
ipl.getPostcards().addAll(List.of(pn));
ipl.setParent(icmpv);
ipl.updateChecksum();
icmpv.setPayload(ipl);
return icmpv;
});
}
private static HighPerformanceExecutor defaultExecutor = new HighPerformanceExecutor(
Runtime.getRuntime().availableProcessors() + 1, Thread.ofPlatform().name("SRv6 Routing").factory());//Runtime.getRuntime().availableProcessors() + 1
private static HighPerformanceExecutor2 defaultExecutor = new HighPerformanceExecutor2(
(int) (Runtime.getRuntime().availableProcessors()), Thread.ofVirtual().name("SRv6 Routing").factory());
//private static HighPerformanceExecutor2 defaultExecutor = new HighPerformanceExecutor2(4, Thread.ofVirtual().name("SRv6 Routing").factory());
public static HighPerformanceExecutor getDefaultExecutor() {
public static HighPerformanceExecutor2 getDefaultExecutor() {
return defaultExecutor;
}
@@ -802,38 +775,52 @@ public class SRv6Router {
// packetQueue.add(object);
defaultExecutor.executeWithCongestionReport((state) -> {
try {
long start=System.nanoTime();
IPv6Packet packet=sendTask.get();
if(state) {
if(packet.isEnableECN()) {
packet.markCE();
//System.out.println("mce");
}
}
insertSRHandRoutePacket(null, packet); // 插入SRH并路由
long time=System.nanoTime()-start;
backplaneCount(time,state);
} catch (Exception e) {
e.printStackTrace();
}
},20);
}
public void runPacketSendTask(Supplier<IPv6Packet> sendTask) {
try {
boolean state=false;
long start=System.nanoTime();
IPv6Packet packet=sendTask.get();
if(state) {
packet.markCE();
}
insertSRHandRoutePacket(null, packet); // 插入SRH并路由
long time=System.nanoTime()-start;
backplaneCount(time,state);
} catch (Exception e) {
e.printStackTrace();
}
}
public void onReceive(IPv6NetworkLink link,IPv6Packet t){
if (unduplicate(t)) {
insertSRHandRoutePacket(link, t); // 插入SRH并路由数据包
t.putTimePassport("routed"); // 标记路由时间点
}
}
public void enqueuePacketReceiveTask(IPv6NetworkLink link, Supplier<IPv6Packet> pack) {
defaultExecutor.executeWithCongestionReport((state) -> {
try {
long start=System.nanoTime();
IPv6Packet packet = pack.get();
if(state) {
if(packet.isEnableECN()) {
packet.markCE();
//System.out.println("mce");
}
}
onReceive(link ,packet);
long time=System.nanoTime()-start;
backplaneCount(time,state);
} catch (Exception e) {
e.printStackTrace();
}
@@ -845,6 +832,67 @@ public class SRv6Router {
return packet;
});
}
//private BandwidthSampler sampler=new BandwidthSampler();
private FixedSizeStripedLongCounter timeAdder=KLALBUtils.createCounter();
private FixedSizeStripedLongCounter countAdder=KLALBUtils.createCounter();
private FixedSizeStripedLongCounter ECNcountAdder=KLALBUtils.createCounter();
private FixedSizeStripedLongCounter LosscountAdder=KLALBUtils.createCounter();
private void backplaneCount(long time,boolean ECN) {
timeAdder.inc(time);
countAdder.inc(1);
if(ECN) {
ECNcountAdder.inc(1);
}
}
private volatile long backplaneTime=0;
private volatile double ECNRate=0;
private volatile double LossRate=0;
private volatile long backplanePPS=0;
private volatile long backplanePPSMax=0;
private volatile long prevTime=System.nanoTime();
private Timer backplaneTimer=new Timer();
private void updateBackplaneCounter() {
long curr=System.nanoTime();
long totalTime=timeAdder.getAndReset();
long totalCount=countAdder.getAndReset();
long ECNCount=ECNcountAdder.getAndReset();
long LossCount=LosscountAdder.getAndReset();
if(totalCount!=0) {
backplaneTime=totalTime/totalCount;
ECNRate=((double)ECNCount)/totalCount;
LossRate=((double)LossCount)/totalCount;
}
backplanePPS=totalCount*1000000000L/(curr-prevTime);
if(backplanePPSMax<backplanePPS) {
backplanePPSMax=backplanePPS;
}else {
backplanePPSMax=(backplanePPSMax*999+backplanePPS)/1000;
}
prevTime=curr;
}
public long getBackplaneTime() {
return backplaneTime;
}
public long getBackplanePPS() {
return backplanePPS;
}
public long getBackplanePPSMax() {
return backplanePPSMax;
}
public double getECNRate() {
return ECNRate;
}
public double getLossRate() {
return LossRate;
}
// 构造函数
public SRv6Router(IPv6AddressGroup hostAddress, HighAccuracyClock clock) {
@@ -856,6 +904,13 @@ public class SRv6Router {
this.inLoopBack = new LoopbackIPv6NetworkLink(
List.of(new IPv6AddressGroup(IPv6Address.LOOPBACK, 128), hostAddress), this);
linkTabel.add(inLoopBack); // 添加到链路表
backplaneTimer.scheduleAtFixedRate(new TimerTask() {
@Override
public void run() {
updateBackplaneCounter();
}
}, 100, 100);
}
/*
@@ -892,7 +947,7 @@ public class SRv6Router {
Object[] nls = links.toArray();
for (int i = 0; i < nls.length; i++) {
IPv6NetworkLink nl = (IPv6NetworkLink) nls[i];
if (!nl.isLoopBack()) { // 排除环回接口
if ((!nl.isLoopBack())&&nl.isUp()) { // 排除环回接口
List<Neighbor> neis = nl.getNeighborsInfo();
for (Iterator<Neighbor> iteratorx = neis.iterator(); iteratorx.hasNext();) {
Neighbor addresses = (Neighbor) iteratorx.next();
@@ -903,7 +958,7 @@ public class SRv6Router {
return result;
}
// 获取所有定位器
// 获取所有SID
public Set<IPv6AddressGroup> getLocators() {
Set<IPv6AddressGroup> result = new HashSet<>();
List<IPv6NetworkLink> links = getLinkTabel();
@@ -915,10 +970,14 @@ public class SRv6Router {
for (Iterator<Neighbor> iteratorx = neis.iterator(); iteratorx.hasNext();) {
Neighbor addresses = (Neighbor) iteratorx.next();
if (addresses.getLocator() != null)
result.add(addresses.getLocator()); // 添加邻居定位器
result.add(addresses.getLocator()); // 添加邻居SID
}
}
}
return result;
}
}