diff --git a/src/org/kne/cloud/network/klalb/KLALBController.java b/src/org/kne/cloud/network/klalb/KLALBController.java index 7e5d2e1..358cbab 100644 --- a/src/org/kne/cloud/network/klalb/KLALBController.java +++ b/src/org/kne/cloud/network/klalb/KLALBController.java @@ -17,14 +17,8 @@ import java.util.Iterator; import java.util.List; import java.util.Map; import java.util.Set; -import java.util.Timer; -import java.util.TimerTask; import java.util.UUID; -import java.util.concurrent.ConcurrentHashMap; -import java.util.concurrent.CopyOnWriteArrayList; -import java.util.concurrent.Executors; -import java.util.concurrent.ScheduledExecutorService; -import java.util.concurrent.ThreadFactory; +import java.util.concurrent.*; import java.util.concurrent.locks.Lock; import java.util.concurrent.locks.ReadWriteLock; import java.util.concurrent.locks.ReentrantLock; @@ -67,7 +61,7 @@ public class KLALBController { public HighAccuracyClock getClock() { return clock; } - private final long TIME_WINDOW=2000000000L; + private final long TIME_WINDOW=50000000000L; private SpeedAndTrafficMonitorDataImpl linkMonitor = new SpeedAndTrafficMonitorDataImpl( new HashMapTimestampMonitor(HighAccuracyClock.SYSTEM_CLOCK, "up", 100,TIME_WINDOW ), new HashMapTimestampMonitor(HighAccuracyClock.SYSTEM_CLOCK, "down", 100, TIME_WINDOW)); @@ -104,23 +98,9 @@ public class KLALBController { public List getDnsAddresses() { return dnsAddresses; } - private ScheduledExecutorService scheduleTimer=Executors.newScheduledThreadPool(Runtime.getRuntime().availableProcessors(),new ThreadFactory() { - - @Override - public Thread newThread(Runnable r) { - return ThreadTool.makeVDaemonThreadIfSupport("计时器线程", r); - } - }); - - public ScheduledExecutorService getScheduleTimer() { - return scheduleTimer; - } - - private Timer twk = new Timer("网卡检测扫描计时器", true); - - private TimerTask tsk1 = new TimerTask() { + private Runnable tsk1 = new Runnable() { @Override public void run() { @@ -306,7 +286,7 @@ public class KLALBController { return srv6Router.getLocator().getAddress().toInet6Address().equals(inetAddress); } - private TimerTask tsk2 = new TimerTask() { + private Runnable tsk2 = new Runnable() { @Override public void run() { @@ -734,8 +714,8 @@ public class KLALBController { private void loadController() { this.apiServer = new KLALBRoutingProtocolAPIServer(routingProtocol, this); this.apiClient = new KLALBRoutingProtocolAPIClient(routingProtocol); - twk.schedule(tsk1, 5000, 5000); - twk.schedule(tsk2, 5000, 5000); + KLALBScheduler.getScheduler().scheduleAtFixedRate(tsk1, 5000, 5000,TimeUnit.MILLISECONDS); + KLALBScheduler.getScheduler().scheduleAtFixedRate(tsk2, 5000, 5000,TimeUnit.MILLISECONDS); } public KLALBController() { @@ -753,17 +733,10 @@ public class KLALBController { } protected void initKLALB(boolean enableVirtualAdapter, IPv6AddressGroup selfg) { - ThreadTool.makeVDaemonThread("线路性能采样线程", ()->{ - while(true) { + KLALBScheduler.getScheduler().scheduleAtFixedRate( ()->{ linkMonitor.update(); datatMonitor.update(); - try { - Thread.sleep(1); - } catch (InterruptedException e) { - e.printStackTrace(); - } - } - }).start(); + },1,1,TimeUnit.MILLISECONDS); loadSRv6ProtocolStack(selfg, enableVirtualAdapter); loadController(); } @@ -804,7 +777,7 @@ public class KLALBController { List linele = configItem.getExternalEndpoints(); if (linele != null) { getExternalEndpoints().addAll(linele); - } + } List linetoc = configItem.getAutoConnections(); if (linetoc != null) { linetoc.forEach((aline) -> { @@ -856,9 +829,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 +841,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 +852,7 @@ public class KLALBController { } } - }, 10, 10); + }, 1, 1,TimeUnit.MILLISECONDS); } public IPv6Packet createPacketToAddress(IPv6Address addr, int flowlabel, IPv6Packet.IPv6Payload packet) { diff --git a/src/org/kne/cloud/network/klalb/KLALBRemoteLink.java b/src/org/kne/cloud/network/klalb/KLALBRemoteLink.java index e420d85..4c9c9e3 100644 --- a/src/org/kne/cloud/network/klalb/KLALBRemoteLink.java +++ b/src/org/kne/cloud/network/klalb/KLALBRemoteLink.java @@ -13,6 +13,8 @@ import java.util.Iterator; import java.util.List; import java.util.Queue; import java.util.UUID; +import java.util.concurrent.ScheduledFuture; +import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.locks.Condition; import java.util.concurrent.locks.Lock; @@ -570,20 +572,17 @@ public class KLALBRemoteLink extends AbstractControlledIPv6NetworkLink implement private ThreadParker tpk=new ThreadParker(); private StateManager stateManager; - private MonitorUpdater monitorUpdater; + private ScheduledFuture monitorUpdater; private class MonitorUpdater implements Runnable{ @Override public void run() { - while (!closed) { + if(closed) { + monitorUpdater.cancel(false); + }else{ status.updateReliability(); if(status.getState()!=LinkStatus.DOWN) { monitor.update(); } - try { - Thread.sleep(1); - } catch (InterruptedException e) { - e.printStackTrace(); - } } @@ -755,8 +754,7 @@ public class KLALBRemoteLink extends AbstractControlledIPv6NetworkLink implement protected void startIO() { stateManager=new StateManager(); - monitorUpdater=new MonitorUpdater(); - ThreadTool.makeVDaemonThread("线路性能采样线程", monitorUpdater).start(); + monitorUpdater=KLALBScheduler.getScheduler().scheduleAtFixedRate(new MonitorUpdater(),1,1, TimeUnit.MILLISECONDS); ThreadTool.makeVDaemonThread("远程线路状态线程", stateManager).start(); } diff --git a/src/org/kne/cloud/network/klalb/KLALBScheduler.java b/src/org/kne/cloud/network/klalb/KLALBScheduler.java new file mode 100644 index 0000000..94483a3 --- /dev/null +++ b/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; + } +} diff --git a/src/org/kne/cloud/network/klalb/ui/KLALBStateGUI3.java b/src/org/kne/cloud/network/klalb/ui/KLALBStateGUI3.java index 45735cb..028da0e 100644 --- a/src/org/kne/cloud/network/klalb/ui/KLALBStateGUI3.java +++ b/src/org/kne/cloud/network/klalb/ui/KLALBStateGUI3.java @@ -7,8 +7,8 @@ import java.io.IOException; import java.net.*; import java.util.ArrayList; import java.util.List; -import java.util.Timer; -import java.util.TimerTask; +import java.util.concurrent.ScheduledFuture; +import java.util.concurrent.TimeUnit; import java.util.function.Consumer; import javax.imageio.ImageIO; import javax.swing.*; @@ -37,8 +37,7 @@ import org.kne.ui.YScrollPane; public class KLALBStateGUI3 extends XFrame { // ==================== 定时任务相关 ==================== - private TimerTask tsk, tsk2, tsk3, tsk4, tsk5; // 各种定时任务 - private Timer t, t2, t3; // 定时器 + private ScheduledFuture tsk, tsk2, tsk3, tsk4, tsk5; // 各种定时任务 // ==================== 系统托盘相关 ==================== private SystemTray st; // 系统托盘 @@ -77,7 +76,7 @@ public class KLALBStateGUI3 extends XFrame { private JTextField asnFieldSet; // ASN设置框 private JCheckBox enableTUN; // 虚拟网卡启用禁用 private JTextField tunDeviceName; // 虚拟网卡名称设置框 - private JTextField webListenSet; // Web API 监听地址设置框 + private JTextField webListenSet; // Web API 监听地址设置框 private NetworkGraphPanel graph; // 网络图面板 private JTextField textFieldLocate; // 定位地址输入框 private JTextField asnField2; // ASN显示框 @@ -705,10 +704,6 @@ public class KLALBStateGUI3 extends XFrame { * 初始化定时器 */ private void initTimers(KLALBController kc) { - t3 = new Timer("监视器刷新线程", true); - t2 = new Timer("仪表盘刷新线程", true); - t = new Timer("状态刷新线程", true); - // 创建刷新任务 createRefreshTask(kc, ysp); } @@ -924,10 +919,10 @@ public class KLALBStateGUI3 extends XFrame { webApiEnabled=webApic.getCheckBox(); settings.getView().add(webApic); - TextSettingItem webListen = new TextSettingItem(UIEnv.getRsb().getString("weblistenaddr"), - CONST.itemwidth, CONST.settingheight); - webListenSet = webListen.getTextField(); - settings.getView().add(webListen); + TextSettingItem webListen = new TextSettingItem(UIEnv.getRsb().getString("weblistenaddr"), + CONST.itemwidth, CONST.settingheight); + webListenSet = webListen.getTextField(); + settings.getView().add(webListen); //性能设置标题 @@ -1167,19 +1162,19 @@ public class KLALBStateGUI3 extends XFrame { kck.setTUNName(tunNameText); } - // 保存Web API设置 - kck.setWebUI(webApiEnabled.isSelected()); - String webListenText = webListenSet.getText().trim(); - if (webListenText.equals("")) { - kck.setWebListen(new MultiProtocolSocketAddress("http", "0.0.0.0", 4665)); - } else { - try { - kck.setWebListen(new MultiProtocolSocketAddress(webListenText)); - } catch (RuntimeException e) { - e.printStackTrace(); - JOptionPane.showMessageDialog(this, UIEnv.getRsb().getString("invaildweblistenaddr"), - UIEnv.getRsb().getString("warning"), JOptionPane.WARNING_MESSAGE); - return; + // 保存Web API设置 + kck.setWebUI(webApiEnabled.isSelected()); + String webListenText = webListenSet.getText().trim(); + if (webListenText.equals("")) { + kck.setWebListen(new MultiProtocolSocketAddress("http", "0.0.0.0", 4665)); + } else { + try { + kck.setWebListen(new MultiProtocolSocketAddress(webListenText)); + } catch (RuntimeException e) { + e.printStackTrace(); + JOptionPane.showMessageDialog(this, UIEnv.getRsb().getString("invaildweblistenaddr"), + UIEnv.getRsb().getString("warning"), JOptionPane.WARNING_MESSAGE); + return; } } @@ -1322,9 +1317,9 @@ public class KLALBStateGUI3 extends XFrame { private void createRefreshTask(KLALBController kc, YScrollPane ysp) { // 线路状态刷新任务 if (tsk != null) { - tsk.cancel(); + tsk.cancel(false); } - tsk = new TimerTask() { + tsk =KLALBScheduler.getScheduler().scheduleAtFixedRate( new Runnable() { @Override public void run() { if (isVisible()) { @@ -1382,28 +1377,26 @@ public class KLALBStateGUI3 extends XFrame { ysp.updateScrool(); } } - }; - t.scheduleAtFixedRate(tsk, 1000, 1000); + }, 1000, 1000, TimeUnit.MILLISECONDS); // 仪表盘数据刷新任务 if (tsk2 != null) { - tsk2.cancel(); + tsk2.cancel(false); } - tsk2 = new TimerTask() { + tsk2 =KLALBScheduler.getScheduler().scheduleAtFixedRate(new Runnable() { @Override public void run() { if (isVisible()) { updateDashboardData(kc); } } - }; - t2.scheduleAtFixedRate(tsk2, 100, 100); + }, 100, 100,TimeUnit.MILLISECONDS); // 网络图刷新任务 if (tsk5 != null) { - tsk5.cancel(); + tsk5.cancel(false); } - tsk5 = new TimerTask() { + tsk5 = KLALBScheduler.getScheduler().scheduleAtFixedRate(new Runnable() { @Override public void run() { if (isVisible()) { @@ -1415,8 +1408,7 @@ public class KLALBStateGUI3 extends XFrame { graph.revalidate(); } } - }; - t.scheduleAtFixedRate(tsk5, 1000, 1000); + }, 1000, 1000,TimeUnit.MILLISECONDS); } /** @@ -1588,10 +1580,10 @@ public class KLALBStateGUI3 extends XFrame { - // 加载Web API设置 - webApiEnabled.setSelected(kck.isWebUI()); - webListenSet.setText(kck.getWebListen() != null ? kck.getWebListen().toString() - : "http://0.0.0.0:4665"); + // 加载Web API设置 + webApiEnabled.setSelected(kck.isWebUI()); + webListenSet.setText(kck.getWebListen() != null ? kck.getWebListen().toString() + : "http://0.0.0.0:4665"); // 加载TCP监听 MultiProtocolSocketAddress mpat = kck.getTCPListen(); @@ -1764,7 +1756,7 @@ public class KLALBStateGUI3 extends XFrame { public void close() { setVisible(false); if (tsk != null) { - tsk.cancel(); + tsk.cancel(false); } if (st != null) { st.remove(ti); @@ -1779,4 +1771,4 @@ public class KLALBStateGUI3 extends XFrame { KLALBStateGUI3 kcg = new KLALBStateGUI3(new KLALBController()); kcg.setVisible(true); } -} +} diff --git a/src/org/kne/cloud/network/klalb/ui/LineMonitorGUI.java b/src/org/kne/cloud/network/klalb/ui/LineMonitorGUI.java index 494870b..a39360a 100644 --- a/src/org/kne/cloud/network/klalb/ui/LineMonitorGUI.java +++ b/src/org/kne/cloud/network/klalb/ui/LineMonitorGUI.java @@ -4,8 +4,7 @@ import java.awt.Color; import java.awt.Dimension; import java.awt.GridLayout; import java.util.Date; -import java.util.Timer; -import java.util.TimerTask; +import java.util.Scanner; import java.util.UUID; import javax.swing.JPanel; @@ -26,6 +25,7 @@ import org.kne.cloud.network.ipv6.IPv6Address; import org.kne.cloud.network.ipv6.IPv6AddressGroup; import org.kne.cloud.network.ipv6.IPv6Packet; import org.kne.cloud.network.klalb.KLALBRemoteLink; +import org.kne.cloud.network.klalb.KLALBScheduler; import org.kne.cloud.network.monitor.LinkStatus; import org.kne.cloud.network.monitor.TimestampMonitorValue; import org.kne.math.Long128; @@ -38,21 +38,18 @@ import javax.swing.ImageIcon; import javax.swing.JLabel; import java.awt.event.WindowEvent; import java.awt.event.WindowListener; +import java.util.concurrent.ScheduledFuture; +import java.util.concurrent.TimeUnit; import javax.swing.JSlider; public class LineMonitorGUI extends XFrame { - private Timer timer; - private Timer timer1; - private Timer timer2; - private Timer timer3; - private Timer timer4; - private TimerTask tsk3; + private ScheduledFuture tsk3; - private TimerTask tsk4, tsk5,tsk6,tsk7; - private TimerTask d1; - private TimerTask d2; - private TimerTask d3; - private TimerTask d4; + private ScheduledFuture tsk4, tsk5,tsk6,tsk7; + private ScheduledFuture d1; + private ScheduledFuture d2; + private ScheduledFuture d3; + private ScheduledFuture d4; private TimeSeries spdup = new TimeSeries(UIEnv.getRsb().getString("uploadspeed")); private TimeSeries spddown = new TimeSeries(UIEnv.getRsb().getString("downloadspeed")); @@ -386,49 +383,45 @@ public class LineMonitorGUI extends XFrame { @Override public void setVisible(boolean b) { if (b) { - if (timer == null) { - timer = new Timer(); - timer1 = new Timer(); - timer2 = new Timer(); - timer3 = new Timer(); - timer4 = new Timer(); - tsk3 = new TimerTask() { + if (tsk3 == null) { + + tsk3 = KLALBScheduler.getScheduler().scheduleAtFixedRate(new Runnable() { @Override public void run() { updateTraffic(); } - }; - tsk4 = new TimerTask() { + }, 25, 25, TimeUnit.MILLISECONDS); + tsk4 = KLALBScheduler.getScheduler().scheduleAtFixedRate(new Runnable() { @Override public void run() { recordSpeedData(); } - }; + }, 10, 10, TimeUnit.MILLISECONDS);; - tsk5 = new TimerTask() { + tsk5 =KLALBScheduler.getScheduler().scheduleAtFixedRate( new Runnable() { @Override public void run() { recordDelayData(); } - }; - tsk6 = new TimerTask() { + }, 10, 10, TimeUnit.MILLISECONDS); + tsk6 =KLALBScheduler.getScheduler().scheduleAtFixedRate( new Runnable() { @Override public void run() { recordBufferData(); } - }; - tsk7 = new TimerTask() { + }, 10, 10, TimeUnit.MILLISECONDS); + tsk7 = KLALBScheduler.getScheduler().scheduleAtFixedRate( new Runnable() { @Override public void run() { recordCongestionData(); } - }; - d1 = new TimerTask() { + }, 25, 25, TimeUnit.MILLISECONDS); + d1 = KLALBScheduler.getScheduler().scheduleAtFixedRate(new Runnable() { @Override public void run() { @@ -440,8 +433,8 @@ public class LineMonitorGUI extends XFrame { } } } - }; - d2 = new TimerTask() { + } ,25, 25, TimeUnit.MILLISECONDS); + d2 = KLALBScheduler.getScheduler().scheduleAtFixedRate( new Runnable() { @Override public void run() { @@ -454,8 +447,8 @@ public class LineMonitorGUI extends XFrame { } } - }; - d3 = new TimerTask() { + }, 25, 25, TimeUnit.MILLISECONDS); + d3 = KLALBScheduler.getScheduler().scheduleAtFixedRate(new Runnable() { @Override public void run() { @@ -468,8 +461,8 @@ public class LineMonitorGUI extends XFrame { } } - }; - d4 = new TimerTask() { + }, 25, 25, TimeUnit.MILLISECONDS); + d4 =KLALBScheduler.getScheduler().scheduleAtFixedRate( new Runnable() { @Override public void run() { @@ -482,51 +475,32 @@ public class LineMonitorGUI extends XFrame { } } - }; - timer.scheduleAtFixedRate(tsk3, 25, 25); - timer1.scheduleAtFixedRate(tsk4, 10, 10); - timer1.scheduleAtFixedRate(d1, 25, 25); - timer2.scheduleAtFixedRate(tsk5, 10, 10); - timer2.scheduleAtFixedRate(d2, 25, 25); - timer3.scheduleAtFixedRate(tsk6, 10, 10); - timer3.scheduleAtFixedRate(d3, 25, 25); - timer4.scheduleAtFixedRate(tsk7, 25, 25); - timer4.scheduleAtFixedRate(d4, 25, 25); + }, 25, 25, TimeUnit.MILLISECONDS); + + + } } else { if (tsk3 != null) - tsk3.cancel(); + tsk3.cancel(false); if (tsk4 != null) - tsk4.cancel(); + tsk4.cancel(false); if (tsk5 != null) - tsk5.cancel(); + tsk5.cancel(false); if (tsk6 != null) - tsk6.cancel(); + tsk6.cancel(false); if (tsk7 != null) - tsk7.cancel(); + tsk7.cancel(false); if (d1 != null) - d1.cancel(); + d1.cancel(false); if (d2 != null) - d2.cancel(); + d2.cancel(false); if (d3 != null) - d3.cancel(); + d3.cancel(false); if (d4 != null) - d4.cancel(); - if (timer != null) - timer.cancel(); - if (timer1 != null) - timer1.cancel(); - - if (timer2 != null) - timer2.cancel(); - - if (timer3 != null) - timer3.cancel(); - - if (timer4 != null) - timer4.cancel(); - timer = null; + d4.cancel(false); + tsk3=null; } super.setVisible(b); } diff --git a/src/org/kne/cloud/network/kltp/KLTPOutputStream.java b/src/org/kne/cloud/network/kltp/KLTPOutputStream.java index 5bfd38e..e1356db 100644 --- a/src/org/kne/cloud/network/kltp/KLTPOutputStream.java +++ b/src/org/kne/cloud/network/kltp/KLTPOutputStream.java @@ -8,6 +8,8 @@ import java.net.SocketException; import java.net.SocketTimeoutException; import java.nio.ByteBuffer; import java.nio.channels.WritableByteChannel; + +import org.kne.cloud.network.klalb.KLALBScheduler; import org.kne.concurrent.*; import org.kne.cloud.network.NetworkPacket; import org.kne.cloud.network.congestion.CongestionAlgorithm; @@ -254,7 +256,7 @@ public class KLTPOutputStream extends OutputStream implements KLTPPacketConsumer } } }; - tt=controller.getScheduleTimer().scheduleAtFixedRate (r, delaytime, delaytime,TimeUnit.NANOSECONDS); + tt= KLALBScheduler.getScheduler().scheduleWithFixedDelay(r, delaytime, delaytime,TimeUnit.NANOSECONDS); } IOException ioex=ioe.get(); diff --git a/src/org/kne/cloud/network/monitor/ArrayListTimestampMonitor.java b/src/org/kne/cloud/network/monitor/ArrayListTimestampMonitor.java index a3549e0..4950725 100644 --- a/src/org/kne/cloud/network/monitor/ArrayListTimestampMonitor.java +++ b/src/org/kne/cloud/network/monitor/ArrayListTimestampMonitor.java @@ -66,7 +66,12 @@ public class ArrayListTimestampMonitor implements TimestampMonitor { public boolean recordPacket(K key,int count, int length) { return recordPacket(key,clock.getCurrentTimeNanos(),count,length); } - + + + public boolean recordPacket(K key, int count, int length, long delay, long delayMin) { + return recordPacket(key, clock.getCurrentTimeNanos(),count, length, delay,delayMin); + } + public boolean recordPacket(K key,int count, int length, long delay) { return recordPacket(key, clock.getCurrentTimeNanos(),count, length, delay); } @@ -75,17 +80,22 @@ public class ArrayListTimestampMonitor implements TimestampMonitor { return recordPacket(key, timestamp,count, length, Long.MAX_VALUE / 1024); } - + @Override + public boolean recordPacket(K key, Long128 timestamp, int count, int length, long dsndDelay) { + return recordPacket(key, timestamp, count, length, dsndDelay,dsndDelay); + } + + /** * 记录数据包 - 使用插入排序保持列表有序，添加时间戳验证 */ - public boolean recordPacket(K key, Long128 timestamp,int count, int length, long delay) { + public boolean recordPacket(K key, Long128 timestamp,int count, int length, long delay, long dsndDelayMin) { // 验证时间戳的合理性 validateTimestamp(timestamp); // 快速路径：尝试sequence=0 TimestampMonitorKey monitorKey = new TimestampMonitorKey<>(key, 0); - TimestampMonitorValue monitorValue = new TimestampMonitorValue(monitorKey, timestamp,count, length, delay); + TimestampMonitorValue monitorValue = new TimestampMonitorValue(monitorKey, timestamp,count, length, delay,dsndDelayMin); autoCleanup(cleanupThreshold); rwLock.writeLock().lock(); @@ -460,7 +470,49 @@ public class ArrayListTimestampMonitor implements TimestampMonitor { return Long.MAX_VALUE / 1024; } } - + + + + /** + * 从已排序的列表中计算数据量 - 优化版本，只需要startTime + */ + private long calculateMinDelayFromSortedList(List sortedList, Long128 startTime) { + if (sortedList.isEmpty()) { + return 0; + } + + // 使用二分查找找到第一个 >= startTime 的元素位置 + int startIndex = findFirstValidIndexFromSorted(sortedList, startTime); + + if (startIndex >= sortedList.size()) { + return 0; + } + + // 计算从startIndex到末尾的所有数据量（因为数据包不会来自未来） + long sum = Long.MAX_VALUE/1024; + for (int i = startIndex; i < sortedList.size(); i++) { + sum =Math.min(sortedList.get(i).getDelayMin(),sum) ; + } + + return sum; + } + + @Override + public long calculateMinDelay(long timeWindowNs){ + return calculateMinDelay(clock.getCurrentTimeNanos(),timeWindowNs); + } + + @Override + public long calculateMinDelay(Long128 currentTime, long timeWindowNs){ + Long128 startTime = currentTime.subtract(Long128.valueOf(timeWindowNs)); + rwLock.readLock().lock(); + try { + return calculateMinDelayFromSortedList(entries, startTime); + } finally { + rwLock.readLock().unlock(); + } + } + private volatile TimestampMonitorValue recent=null; @Override public TimestampMonitorValue getRecentDelayEntry() { diff --git a/src/org/kne/cloud/network/monitor/HashMapTimestampMonitor.java b/src/org/kne/cloud/network/monitor/HashMapTimestampMonitor.java index b786b03..fe12fa8 100644 --- a/src/org/kne/cloud/network/monitor/HashMapTimestampMonitor.java +++ b/src/org/kne/cloud/network/monitor/HashMapTimestampMonitor.java @@ -75,16 +75,23 @@ private String name; public boolean recordPacket(K key,int count, int length, long delay) { return recordPacket(key,clock.getCurrentTimeNanos(),count,length,delay); } - - + + @Override + public boolean recordPacket(K key, int count, int length, long delay, long delayMin) { + return recordPacket(key,clock.getCurrentTimeNanos(),count,length,delay,delayMin); + } + + public boolean recordPacket(K key, Long128 timestamp,int count, int length, long delay) { + return recordPacket(key,timestamp,count,length,delay,delay); + } /** * 超高性能记录数据包 - 针对路由器优化 */ - public boolean recordPacket(K key, Long128 timestamp,int count, int length, long delay) { + public boolean recordPacket(K key, Long128 timestamp,int count, int length, long delay,long delayMin) { autoCleanup(cleanupThreshold); // 快速路径：尝试sequence=0 TimestampMonitorKey monitorKey = new TimestampMonitorKey<>(key, 0); - TimestampMonitorValue monitorValue = new TimestampMonitorValue(monitorKey,timestamp, count,length,delay); + TimestampMonitorValue monitorValue = new TimestampMonitorValue(monitorKey,timestamp, count,length,delay,delayMin); if (entries.putIfAbsent(monitorKey, monitorValue) == null) { totalDataSize.add(length); @@ -304,7 +311,18 @@ private String name; // TODO 自动生成的方法存根 return 0; } - @Override + + @Override + public long calculateMinDelay(long timeWindowNs) { + return 0; + } + + @Override + public long calculateMinDelay(Long128 currentTime, long timeWindowNs) { + return 0; + } + + @Override public TimestampMonitorValue getRecentDelayEntry() { // TODO 自动生成的方法存根 return null; diff --git a/src/org/kne/cloud/network/monitor/SpeedAndTrafficAndDelayMonitorDataImpl.java b/src/org/kne/cloud/network/monitor/SpeedAndTrafficAndDelayMonitorDataImpl.java index 9230952..9afdb98 100644 --- a/src/org/kne/cloud/network/monitor/SpeedAndTrafficAndDelayMonitorDataImpl.java +++ b/src/org/kne/cloud/network/monitor/SpeedAndTrafficAndDelayMonitorDataImpl.java @@ -19,26 +19,28 @@ public class SpeedAndTrafficAndDelayMonitorDataImpl extends SpeedAndTrafficMonit upSampler.update(); downSampler.update(); if(upSampler.getSnapshotPacketCount()>0&&upSampler.getSnapshotTotalDelay()>0) { - getUploadBandwidth(). recordPacket(KLALBUtils.createGlobalUUID(),0, 0, upSampler.getAvgDelayNanos()); - long odelay=upSampler.getMinDelayNanos(); + getUploadBandwidth(). recordPacket(KLALBUtils.createGlobalUUID(),0, 0, upSampler.getAvgDelayNanos(),upSampler.getMinDelayNanos()); + /*long odelay=upSampler.getMinDelayNanos(); if(outDelayMin==Long.MAX_VALUE||odelay<=outDelayMin) { outDelayMin=odelay; }else{ outDelayMin=Math.min(odelay, outDelayMin+10000); - } + }*/ } if(downSampler.getSnapshotPacketCount()>0&&downSampler.getSnapshotTotalDelay()>0) { - getDownloadBandwidth(). recordPacket(KLALBUtils.createGlobalUUID(),0, 0, downSampler.getAvgDelayNanos()); + getDownloadBandwidth(). recordPacket(KLALBUtils.createGlobalUUID(),0, 0, downSampler.getAvgDelayNanos(),downSampler.getMinDelayNanos()); long idelay=downSampler.getMinDelayNanos(); - if(inDelayMin==Long.MAX_VALUE||idelay<=inDelayMin) { + /*if(inDelayMin==Long.MAX_VALUE||idelay<=inDelayMin) { inDelayMin=idelay; }else{ inDelayMin=Math.min(idelay, inDelayMin+10000); - } + }*/ } long time=System.nanoTime(); if(time-stime>timewindowmax) { + outDelayMin=getUploadBandwidth().calculateMinDelay(super.getTimeWindowMax()); + inDelayMin=getDownloadBandwidth().calculateMinDelay(super.getTimeWindowMax()); stime=time; } } diff --git a/src/org/kne/cloud/network/monitor/SpeedAndTrafficMonitorDataImpl.java b/src/org/kne/cloud/network/monitor/SpeedAndTrafficMonitorDataImpl.java index 033efc6..3b42575 100644 --- a/src/org/kne/cloud/network/monitor/SpeedAndTrafficMonitorDataImpl.java +++ b/src/org/kne/cloud/network/monitor/SpeedAndTrafficMonitorDataImpl.java @@ -11,13 +11,13 @@ public class SpeedAndTrafficMonitorDataImpl extends MonitorDataImpl implements S private TimestampMonitoruploadBandwidth; private TimestampMonitordownloadBandwidth; private long timewindow=1000000000L; - private long timewindowmax=1000000000L; + private long timewindowmax=50000000000L; private BandwidthSampler upSampler=new BandwidthSampler(); private BandwidthSampler downSampler=new BandwidthSampler(); - private final long TIME_WINDOW=2000000000L; + private final long TIME_WINDOW=50000000000L; public SpeedAndTrafficMonitorDataImpl(HighAccuracyClock clock) { super(); this.clock=clock; @@ -184,25 +184,32 @@ public class SpeedAndTrafficMonitorDataImpl extends MonitorDataImpl implements S long op=uploadBandwidth.calculatePacketRate(timewindowmax); long ip=uploadBandwidth.calculatePacketRate(timewindowmax); if(os>outSpeeedMax) { - outSpeeedMax=(outSpeeedMax+os)/2; + outSpeeedMax=os; }else { outSpeeedMax=(outSpeeedMax*999+os)/1000; } if(is>inSpeedMax) { - inSpeedMax=(inSpeedMax+is)/2; + inSpeedMax=is; }else { inSpeedMax=(inSpeedMax*999+is)/1000; } if(op>outPPSMax) { - outPPSMax=(outPPSMax+op)/2; + outPPSMax=op; }else { outPPSMax=(outPPSMax*999+op)/1000; } if(ip>inPPSMax) { - inPPSMax=(inPPSMax+ip)/2; + inPPSMax=ip; }else { inPPSMax=(inPPSMax*999+ip)/1000; } } } + + protected long getTimeWindow() { + return timewindow; + } + protected long getTimeWindowMax() { + return timewindowmax; + } } diff --git a/src/org/kne/cloud/network/monitor/TimestampMonitor.java b/src/org/kne/cloud/network/monitor/TimestampMonitor.java index 37a3a12..c8d0465 100644 --- a/src/org/kne/cloud/network/monitor/TimestampMonitor.java +++ b/src/org/kne/cloud/network/monitor/TimestampMonitor.java @@ -1,7 +1,6 @@ package org.kne.cloud.network.monitor; import org.kne.cloud.clock.HighAccuracyClock; -import org.kne.cloud.network.ipv6.PacketID; import org.kne.math.Long128; /** @@ -36,7 +35,10 @@ public interface TimestampMonitor { * @return 是否记录成功 */ boolean recordPacket(K key,int count, int length); - + + boolean recordPacket(K key, int count,int length, long dsndDelay); + + boolean recordPacket(K key, int count, int length, long avgDelayNanos, long minDelayNanos); /** * 记录数据包（使用指定时间戳） * @param key 数据包标识 @@ -49,8 +51,10 @@ public interface TimestampMonitor { boolean recordPacket(K key, Long128 timestamp, int count,int length, long dsndDelay); - boolean recordPacket(K key, int count,int length, long dsndDelay); - + + + boolean recordPacket(K key, Long128 timestamp, int count,int length, long dsndDelay,long dsndDelayMin); + // ==================== 清理维护方法 ==================== /** @@ -144,9 +148,13 @@ public interface TimestampMonitor { TrafficStats calculateTrafficStats(Long128 currentTime, long timeWindowNs); long getRecentDelay(); - - TimestampMonitorValue getRecentDelayEntry(); + + long calculateMinDelay(long timeWindowNs); + + long calculateMinDelay(Long128 currentTime, long timeWindowNs); + + TimestampMonitorValue getRecentDelayEntry(); // ==================== 全局统计方法 ==================== diff --git a/src/org/kne/cloud/network/monitor/TimestampMonitorValue.java b/src/org/kne/cloud/network/monitor/TimestampMonitorValue.java index c9d06c4..6c2b214 100644 --- a/src/org/kne/cloud/network/monitor/TimestampMonitorValue.java +++ b/src/org/kne/cloud/network/monitor/TimestampMonitorValue.java @@ -8,22 +8,26 @@ public class TimestampMonitorValue implements Comparable private final int length; private final TimestampMonitorKey key; private long delay; + private long delayMin; public TimestampMonitorValue(TimestampMonitorKey key, Long128 timestamp,int count, int length) { - this(key,timestamp,count,length,Long.MAX_VALUE/1024); + this(key,timestamp,count,length,Long.MAX_VALUE/1024,Long.MAX_VALUE/1024); } - public TimestampMonitorValue(TimestampMonitorKey key, Long128 timestamp,int count, int length,long delay) { + public TimestampMonitorValue(TimestampMonitorKey key, Long128 timestamp,int count, int length,long delay,long delayMin) { this.key = key; this.timestamp = timestamp; this.count=count; this.length = length; this.delay=delay; + this.delayMin=delayMin; } @Override public int hashCode() { final int prime = 31; int result = 1; + result = prime * result + count; result = prime * result + (int) (delay ^ (delay >>> 32)); + result = prime * result + (int) (delayMin ^ (delayMin >>> 32)); result = prime * result + length; result = prime * result + ((timestamp == null) ? 0 : timestamp.hashCode()); return result; @@ -38,8 +42,12 @@ public class TimestampMonitorValue implements Comparable if (getClass() != obj.getClass()) return false; TimestampMonitorValue other = (TimestampMonitorValue) obj; + if (count != other.count) + return false; if (delay != other.delay) return false; + if (delayMin != other.delayMin) + return false; if (length != other.length) return false; if (timestamp == null) { @@ -52,8 +60,8 @@ public class TimestampMonitorValue implements Comparable @Override public String toString() { - return "TimestampMonitorValue [timestamp=" + timestamp + ", length=" + length + ", key=" + key + ", delay=" - + delay + "]"; + return "TimestampMonitorValue [timestamp=" + timestamp + ", count=" + count + ", length=" + length + + ", delay=" + delay + ", delayMin=" + delayMin + "]"; } public Long128 getTimestamp() { return timestamp; } @@ -73,6 +81,12 @@ public class TimestampMonitorValue implements Comparable this.delay = delay; } + public long getDelayMin() { + return delayMin; + } + public void setDelayMin(long delayMin) { + this.delayMin = delayMin; + } @Override public int compareTo(TimestampMonitorValue other) { return this.timestamp.compareTo(other.timestamp); diff --git a/src/org/kne/cloud/network/srv6/SRv6Router.java b/src/org/kne/cloud/network/srv6/SRv6Router.java index 860e633..393cd2a 100644 --- a/src/org/kne/cloud/network/srv6/SRv6Router.java +++ b/src/org/kne/cloud/network/srv6/SRv6Router.java @@ -12,11 +12,10 @@ 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.CopyOnWriteArrayList; import java.util.concurrent.CopyOnWriteArraySet; +import java.util.concurrent.TimeUnit; import java.util.concurrent.locks.Condition; import java.util.concurrent.locks.ReentrantLock; import java.util.function.BiConsumer; @@ -45,11 +44,13 @@ import org.kne.cloud.network.ipv6.PadNHopByHopTLV; import org.kne.cloud.network.ipv6.PostcardEntry; import org.kne.cloud.network.ipv6.RouteItem; import org.kne.cloud.network.klalb.KLALBRemoteLink; +import org.kne.cloud.network.klalb.KLALBScheduler; import org.kne.cloud.network.klalb.KLALBUtils; import org.kne.cloud.network.klalb.PerformanceStrategy; import org.kne.cloud.network.klalb.ui.PerformanceStrategyItem; import org.kne.concurrent.HighPerformanceExecutor2; import org.kne.concurrent.TimeoutConcurrentHashMap; +import org.kne.io.KNEChannels; import org.pcap4j.packet.IllegalRawDataException; import com.google.gson.internal.Pair; @@ -849,8 +850,7 @@ public class SRv6Router { 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(); @@ -904,13 +904,13 @@ public class SRv6Router { linkTabel.add(inLoopBack); // 添加到链路表 this.srhInserter=new SRHInsertNetworkLink(this,networkGroup); linkTabel.add(srhInserter); - backplaneTimer.scheduleAtFixedRate(new TimerTask() { + KLALBScheduler.getScheduler().scheduleAtFixedRate(new Runnable() { @Override public void run() { updateBackplaneCounter(); } - }, 100, 100); + }, 100, 100, TimeUnit.MILLISECONDS); } /*