diff --git a/.classpath b/.classpath index 70a394f..bef6664 100644 --- a/.classpath +++ b/.classpath @@ -4,8 +4,6 @@ - - diff --git a/.idea/libraries/jctools_core_jdk11_4_0_7.xml b/.idea/libraries/jctools_core_jdk11_4_0_7.xml new file mode 100644 index 0000000..f48a35f --- /dev/null +++ b/.idea/libraries/jctools_core_jdk11_4_0_7.xml @@ -0,0 +1,11 @@ + + + + + + + + + + + \ No newline at end of file diff --git a/KLALB.iml b/KLALB.iml index 493b3cc..112977b 100644 --- a/KLALB.iml +++ b/KLALB.iml @@ -250,5 +250,6 @@ + \ No newline at end of file diff --git a/lib/KNElib1.0.jar b/lib/KNElib1.0.jar index 2d38b26..ddab9ad 100644 Binary files a/lib/KNElib1.0.jar and b/lib/KNElib1.0.jar differ diff --git a/lib/jctools-core-4.0.6-sources.jar b/lib/jctools-core-4.0.6-sources.jar deleted file mode 100644 index 6c98e07..0000000 Binary files a/lib/jctools-core-4.0.6-sources.jar and /dev/null differ diff --git a/lib/jctools-core-4.0.6.jar b/lib/jctools-core-4.0.6.jar deleted file mode 100644 index 94a8307..0000000 Binary files a/lib/jctools-core-4.0.6.jar and /dev/null differ diff --git a/src/klalb_en_US.properties b/src/klalb_en_US.properties index 6d5082a..17719d0 100644 --- a/src/klalb_en_US.properties +++ b/src/klalb_en_US.properties @@ -122,4 +122,6 @@ rttbaseline=RTT baseline rtt=RTT rttmax=RTT max rttmin=RTT min -rto=RTO \ No newline at end of file +rto=RTO +cpuload=CPU load +ramusage=RAM usage \ No newline at end of file diff --git a/src/klalb_zh_CN.properties b/src/klalb_zh_CN.properties index eb0093f..e3667ba 100644 --- a/src/klalb_zh_CN.properties +++ b/src/klalb_zh_CN.properties @@ -122,4 +122,6 @@ rttbaseline=往返延迟基线 rtt=往返延迟 rttmax=往返延迟上限 rttmin=往返延迟下限 -rto=重传超时 \ No newline at end of file +rto=重传超时 +cpuload=CPU负载 +ramusage=RAM占用率 \ No newline at end of file diff --git a/src/org/kne/cloud/network/IPMulticastDiscovery.java b/src/org/kne/cloud/network/IPMulticastDiscovery.java index e6aff23..06ed44b 100644 --- a/src/org/kne/cloud/network/IPMulticastDiscovery.java +++ b/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,21 +150,39 @@ public class IPMulticastDiscovery implements Closeable, AutoCloseable { } }).start(); } + // 1. 接收用的 ThreadLocal 缓冲区 + private static final ThreadLocal RX_BUFFER = ThreadLocal.withInitial(() -> ByteBuffer.allocate(2048)); + // 2. 发送用的 ThreadLocal 缓冲区 (广播包一般很小) + private static final ThreadLocal TX_BUFFER = ThreadLocal.withInitial(() -> ByteBuffer.allocate(2048)); private MultiProtocolSocketAddress receive() throws IOException { - byte[] b = new byte[65535]; - DatagramPacket dp = new DatagramPacket(b, b.length); + ByteBuffer bbf = RX_BUFFER.get(); + 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(); - long h=dis.readLong(); - long l=dis.readLong(); - UUID uid=new UUID(h, l); - dis.close(); + + // 限制可读范围 + int length = dp.getLength(); + bbf.limit(length); + + // 直接读 UTF 字符串 (不用 DataInputStream) + // 读取前两个字节的长度，然后创建 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()); + + if (mpsa.getInetAddress().isAnyLocalAddress()) { + mpsa = new MultiProtocolSocketAddress(mpsa.getProtocol(), dp.getAddress().getHostAddress(), mpsa.getPort()); } if(debug) { if(uid.equals(selfUUID)) { @@ -176,15 +196,23 @@ public class IPMulticastDiscovery implements Closeable, AutoCloseable { } private void send(MultiProtocolSocketAddress multiProtocolSocketAddress) throws IOException { - ByteArrayOutputStream baos = new ByteArrayOutputStream(); - DataOutputStream dops = new DataOutputStream(baos); - dops.writeUTF(multiProtocolSocketAddress.toString()); - dops.writeLong(selfUUID.getMostSignificantBits()); - dops.writeLong(selfUUID.getLeastSignificantBits()); - dops.close(); - byte[] b = baos.toByteArray(); - DatagramPacket dp = new DatagramPacket(b, b.length, group); + ByteBuffer bbf = TX_BUFFER.get(); + bbf.clear(); // 重置 + + // 写字符串 (长度 + 字节) + byte[] strBytes = multiProtocolSocketAddress.toString().getBytes(StandardCharsets.UTF_8); + bbf.putShort((short) strBytes.length); + bbf.put(strBytes); + + // 写 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); diff --git a/src/org/kne/cloud/network/congestion/MpscMessageBatcher.java b/src/org/kne/cloud/network/congestion/MpscMessageBatcher.java index cd21127..e0e76f6 100644 --- a/src/org/kne/cloud/network/congestion/MpscMessageBatcher.java +++ b/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 extends Releaser> { class MpscMessageBatcher0 implements Runnable, MessageBatcher { - private final MpscArrayQueue queue; + private final MessagePassingQueue queue; private final int batchSize; private final long maxDelay; private volatile Consumer> consumer; @@ -94,7 +95,7 @@ class MpscMessageBatcher0 implements Runnable, MessageBatcher { 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(); diff --git a/src/org/kne/cloud/network/klalb/KLALBRemoteLink.java b/src/org/kne/cloud/network/klalb/KLALBRemoteLink.java index fc37180..2f440fd 100644 --- a/src/org/kne/cloud/network/klalb/KLALBRemoteLink.java +++ b/src/org/kne/cloud/network/klalb/KLALBRemoteLink.java @@ -21,7 +21,7 @@ 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; @@ -184,7 +184,7 @@ public class KLALBRemoteLink extends AbstractControlledIPv6NetworkLink implement private class RemoteSender implements Runnable { - private MpscArrayQueue sendQueue = new MpscArrayQueue(8192); + private MpscVarHandleArrayQueue sendQueue = new MpscVarHandleArrayQueue(8192); private WatchDogTimer watchdog = new WatchDogTimer(400000000L); private ThreadParker tlock = new ThreadParker(); diff --git a/src/org/kne/cloud/network/klalb/KLALBUtils.java b/src/org/kne/cloud/network/klalb/KLALBUtils.java index 7322633..fd5a839 100644 --- a/src/org/kne/cloud/network/klalb/KLALBUtils.java +++ b/src/org/kne/cloud/network/klalb/KLALBUtils.java @@ -13,8 +13,6 @@ import java.util.UUID; import java.util.regex.Matcher; import java.util.regex.Pattern; -import org.jctools.counters.CountersFactory; -import org.jctools.counters.FixedSizeStripedLongCounter; import org.kne.cloud.network.MultiProtocolSocketAddress; import org.kne.cloud.network.ipv6.IPv6Address; @@ -434,7 +432,5 @@ public static String parseAbbrIPv6(String IPv6Str) { } - public static FixedSizeStripedLongCounter createCounter() { - return CountersFactory.createFixedSizeStripedCounter(Runtime.getRuntime().availableProcessors()); - } + } diff --git a/src/org/kne/cloud/network/klalb/ui/KLALBStateGUI3.java b/src/org/kne/cloud/network/klalb/ui/KLALBStateGUI3.java index 2bcc09d..e0c0f17 100644 --- a/src/org/kne/cloud/network/klalb/ui/KLALBStateGUI3.java +++ b/src/org/kne/cloud/network/klalb/ui/KLALBStateGUI3.java @@ -4,6 +4,7 @@ import java.awt.*; import java.awt.datatransfer.StringSelection; import java.awt.event.*; import java.io.IOException; +import java.lang.management.ManagementFactory; import java.net.*; import java.util.ArrayList; import java.util.List; @@ -16,6 +17,7 @@ import javax.swing.border.LineBorder; import javax.swing.event.ChangeEvent; import javax.swing.event.ChangeListener; +import com.sun.management.OperatingSystemMXBean; import org.jfree.chart.plot.dial.*; import org.jfree.data.general.DefaultValueDataset; import org.kne.cloud.clock.HighAccuracyClock; @@ -56,7 +58,8 @@ public class KLALBStateGUI3 extends XFrame { private DefaultValueDataset backplanePPS; private DefaultValueDataset backplaneECNRate; private DefaultValueDataset backplaneLossRate; - + private DefaultValueDataset CPUload; + private DefaultValueDataset RAMusage; // ==================== 仪表盘文本标注 ==================== private DialTextAnnotation upSpeedText; // 上行速度文本标注 private DialTextAnnotation downSpeedText; // 下行速度文本标注 @@ -66,6 +69,8 @@ public class KLALBStateGUI3 extends XFrame { private DialTextAnnotation backplanePPSText; private DialTextAnnotation backplaneECNRateText; private DialTextAnnotation backplaneLossRateText; + private DialTextAnnotation CPUloadText; + private DialTextAnnotation RAMusageText; // ==================== UI组件 ==================== private YScrollPane ysp, yspc; // 可垂直滚动的面板 @@ -433,6 +438,22 @@ public class KLALBStateGUI3 extends XFrame { backplaneLossRateText = lossRate1.getTextAnnotation(); dashs.add(lossRate1); + + // CPU + CPUload = new DefaultValueDataset(0); + DialPanel cload=new DialPanel(CONST.dialwidth, UIEnv.getRsb().getString("cpuload"), CPUload, null, Color.BLUE, new Color(0, 0, 127)); + cload.addGradientRanges(UIEnv.RANGE_1); + CPUloadText = cload.getTextAnnotation(); + + dashs.add(cload); + + // RAM + RAMusage = new DefaultValueDataset(0); + DialPanel ruse=new DialPanel(CONST.dialwidth, UIEnv.getRsb().getString("ramusage"), RAMusage, null, Color.BLUE, new Color(0, 0, 127)); + ruse.addGradientRanges(UIEnv.RANGE_1); + RAMusageText = ruse.getTextAnnotation(); + + dashs.add(ruse); } @@ -1392,6 +1413,7 @@ public class KLALBStateGUI3 extends XFrame { } }, 100, 100,TimeUnit.MILLISECONDS); + // 网络图刷新任务 if (tsk5 != null) { tsk5.cancel(false); @@ -1497,7 +1519,28 @@ public class KLALBStateGUI3 extends XFrame { backplaneLossRate.setValue(Math.min(backplaneLossRate.getValue().doubleValue() * 0.98 + backplaneLossv * 0.02, 101.0)); } } - + + // 更新CPU + OperatingSystemMXBean osBean = (OperatingSystemMXBean) ManagementFactory.getOperatingSystemMXBean(); + double sys=osBean.getCpuLoad(); + double proc=osBean.getProcessCpuLoad(); + double load = proc/(1-sys+proc)* 100.0 ; + if (Double.isFinite(load)) { + CPUload.setValue(Math.min(CPUload.getValue().doubleValue() * 0.98 + load * 0.02, 101.0)); + } + CPUloadText.setLabel(String.format("%.1f",load) + "%"); + + + // 更新RAM + long max=Runtime.getRuntime().maxMemory(); + long free=Runtime.getRuntime().freeMemory(); + double ramUsage = (double) (max - free) /max* 100.0 ; + if (Double.isFinite(ramUsage)) { + RAMusage.setValue(Math.min(RAMusage.getValue().doubleValue() * 0.98 + ramUsage * 0.02, 101.0)); + } + RAMusageText.setLabel(KLALBUtils.convertIBUint(max- free)+"/"+KLALBUtils.convertIBUint(max ) ); + + // 更新在线设备数 String contt = Long.toString(kc.getIpv6Router().getKlalbRouteProtol().getDevicesFound()); if (!contt.equals(devicesOnline.getText())) { diff --git a/src/org/kne/cloud/network/klalb/ui/KperfGUI.java b/src/org/kne/cloud/network/klalb/ui/KperfGUI.java index 33afc00..fa08b7c 100644 --- a/src/org/kne/cloud/network/klalb/ui/KperfGUI.java +++ b/src/org/kne/cloud/network/klalb/ui/KperfGUI.java @@ -10,6 +10,7 @@ import org.jfree.chart.plot.dial.DialTextAnnotation; import org.jfree.data.general.DefaultValueDataset; import org.kne.cloud.network.MultiProtocolSocketAddress; import org.kne.cloud.network.klalb.CONST; +import org.kne.cloud.network.klalb.KLALBScheduler; import org.kne.cloud.network.klalb.KLALBUtils; import org.kne.cloud.network.monitor.SpeedAndTrafficAndDelayMonitorDataImpl; import org.kne.cloud.network.perf.Kperf; @@ -35,11 +36,12 @@ import java.awt.event.ActionEvent; import javax.swing.JLabel; import javax.imageio.ImageIO; import java.awt.Font; +import java.util.concurrent.ScheduledFuture; +import java.util.concurrent.TimeUnit; import javax.swing.SwingConstants; import javax.swing.ImageIcon; public class KperfGUI extends XFrame{ - private static Timer tmr=new Timer(); private KperfReports reports=new KperfReportsImpl(); private Kperf kperf; @@ -67,7 +69,7 @@ public class KperfGUI extends XFrame{ private DialTextAnnotation downPPSText; - private TimerTask tsk; + private ScheduledFuture tsk; private class KperfReportsImpl implements KperfReports{ @Override public void testFinish() { @@ -211,254 +213,12 @@ public class KperfGUI extends XFrame{ panel_6.add(downPPSLabel, BorderLayout.EAST); - - /* - { - upSpeed = new DefaultValueDataset(0); - upDataSpeed = new DefaultValueDataset(0); - DialPlot dpup = new DialPlot(); - dpup.setDataset(1, upSpeed); - dpup.setDataset(0, upDataSpeed); - StandardDialFrame sdfs = new StandardDialFrame(); - sdfs.setVisible(false); - dpup.setDialFrame(sdfs); - StandardDialScale sds = new StandardDialScale(0, 100, -120, -300, 10, 5); - sds.setTickRadius(0.9); - sds.setTickLabelsVisible(false); - dpup.addScale(0, sds); - StandardDialRange sdrr = new StandardDialRange(0, 70, Color.GREEN); - sdrr.setInnerRadius(0.92); - sdrr.setOuterRadius(0.93); - dpup.addLayer(sdrr); - - StandardDialRange sdrr2 = new StandardDialRange(70, 90, Color.YELLOW); - sdrr2.setInnerRadius(0.92); - sdrr2.setOuterRadius(0.93); - dpup.addLayer(sdrr2); - - StandardDialRange sdrr3 = new StandardDialRange(90, 100, Color.RED); - sdrr3.setInnerRadius(0.92); - sdrr3.setOuterRadius(0.93); - dpup.addLayer(sdrr3); - - DialPointer.Pointer dpd = new DialPointer.Pointer(); - dpd.setRadius(0.8); - dpd.setFillPaint(new Color(127, 0, 0, 0)); - dpd.setOutlinePaint(new Color(127, 0, 0)); - dpd.setDatasetIndex(0); - //dpup.addLayer(dpd); - - DialPointer.Pointer dp = new DialPointer.Pointer(); - dp.setRadius(0.8); - dp.setFillPaint(Color.RED); - dp.setOutlinePaint(Color.RED); - dp.setDatasetIndex(1); - dpup.addLayer(dp); - - upSpeedText = new DialTextAnnotation("0%"); - upSpeedText.setFont(UIEnv.getFont().deriveFont(20.0f)); - dpup.addLayer(upSpeedText); - - JFreeChart jfup = new JFreeChart(dpup); - jfup.setBorderPaint(Color.WHITE); - jfup.setTitle(UIEnv.getRsb().getString("uploadspeed")); - jfup.getTitle().setFont(UIEnv.getFont().deriveFont(22.0f)); - jfup.getTitle().setPosition(RectangleEdge.BOTTOM); - dpup.setView(0, 0, 1, 1); - ChartPanel cpup = new ChartPanel(jfup); - cpup.setPreferredSize(dashSize); - cpup.setSize(dashSize); - cpup.setMinimumSize(dashSize); - cpup.setBackground(Color.WHITE); - cpup.setBorder(new LineBorder(Color.LIGHT_GRAY)); - panel_3.add(cpup, BorderLayout.WEST); - - downSpeed = new DefaultValueDataset(0); - downDataSpeed = new DefaultValueDataset(0); - DialPlot dpup1 = new DialPlot(); - dpup1.setDataset(1, downSpeed); - dpup1.setDataset(0, downDataSpeed); - StandardDialFrame sdfs1 = new StandardDialFrame(); - sdfs1.setVisible(false); - dpup1.setDialFrame(sdfs1); - StandardDialScale sds1 = new StandardDialScale(0, 100, -120, -300, 10, 5); - sds1.setTickRadius(0.9); - sds1.setTickLabelsVisible(false); - dpup1.addScale(0, sds1); - - StandardDialRange sdrr1 = new StandardDialRange(0, 70, Color.GREEN); - sdrr1.setInnerRadius(0.92); - sdrr1.setOuterRadius(0.93); - dpup1.addLayer(sdrr1); - - StandardDialRange sdrr21 = new StandardDialRange(70, 90, Color.YELLOW); - sdrr21.setInnerRadius(0.92); - sdrr21.setOuterRadius(0.93); - dpup1.addLayer(sdrr21); - - StandardDialRange sdrr31 = new StandardDialRange(90, 100, Color.RED); - sdrr31.setInnerRadius(0.92); - sdrr31.setOuterRadius(0.93); - dpup1.addLayer(sdrr31); - - DialPointer.Pointer dp11 = new DialPointer.Pointer(); - dp11.setRadius(0.8); - dp11.setFillPaint(new Color(0, 127, 0, 0)); - dp11.setOutlinePaint(new Color(0, 127, 0)); - dp11.setDatasetIndex(0); - //dpup1.addLayer(dp11); - - DialPointer.Pointer dp1 = new DialPointer.Pointer(); - dp1.setRadius(0.8); - dp1.setFillPaint(Color.GREEN); - dp1.setOutlinePaint(Color.GREEN); - dp1.setDatasetIndex(1); - dpup1.addLayer(dp1); - - downSpeedText = new DialTextAnnotation("0%"); - downSpeedText.setFont(UIEnv.getFont().deriveFont(20.0f)); - dpup1.addLayer(downSpeedText); - - JFreeChart jfup1 = new JFreeChart(dpup1); - jfup1.setBorderPaint(Color.WHITE); - jfup1.setTitle(UIEnv.getRsb().getString("downloadspeed")); - jfup1.getTitle().setFont(UIEnv.getFont().deriveFont(22.0f)); - jfup1.getTitle().setPosition(RectangleEdge.BOTTOM); - dpup1.setView(0, 0, 1, 1); - ChartPanel cpup1 = new ChartPanel(jfup1); - cpup1.setPreferredSize(dashSize); - cpup1.setSize(dashSize); - cpup1.setMinimumSize(dashSize); - cpup1.setBackground(Color.WHITE); - cpup1.setBorder(new LineBorder(Color.LIGHT_GRAY)); - panel_4.add(cpup1, BorderLayout.WEST); - - } - { - upPPS = new DefaultValueDataset(0); - upDataPPS = new DefaultValueDataset(0); - DialPlot dpup = new DialPlot(); - dpup.setDataset(1, upPPS); - dpup.setDataset(0, upDataPPS); - StandardDialFrame sdfs = new StandardDialFrame(); - sdfs.setVisible(false); - dpup.setDialFrame(sdfs); - StandardDialScale sds = new StandardDialScale(0, 100, -120, -300, 10, 5); - sds.setTickRadius(0.9); - sds.setTickLabelsVisible(false); - dpup.addScale(0, sds); - - StandardDialRange sdrr = new StandardDialRange(0, 70, Color.GREEN); - sdrr.setInnerRadius(0.92); - sdrr.setOuterRadius(0.93); - dpup.addLayer(sdrr); - - StandardDialRange sdrr2 = new StandardDialRange(70, 90, Color.YELLOW); - sdrr2.setInnerRadius(0.92); - sdrr2.setOuterRadius(0.93); - dpup.addLayer(sdrr2); - - StandardDialRange sdrr3 = new StandardDialRange(90, 100, Color.RED); - sdrr3.setInnerRadius(0.92); - sdrr3.setOuterRadius(0.93); - dpup.addLayer(sdrr3); - - DialPointer.Pointer dp1 = new DialPointer.Pointer(); - dp1.setRadius(0.8); - dp1.setFillPaint(new Color(127, 0, 0, 0)); - dp1.setOutlinePaint(new Color(127, 0, 0)); - dp1.setDatasetIndex(0); - //dpup.addLayer(dp1); - - DialPointer.Pointer dp = new DialPointer.Pointer(); - dp.setRadius(0.8); - dp.setFillPaint(Color.RED); - dp.setOutlinePaint(Color.RED); - dp.setDatasetIndex(1); - dpup.addLayer(dp); - - upPPSText = new DialTextAnnotation("0PPS"); - upPPSText.setFont(UIEnv.getFont().deriveFont(20.0f)); - dpup.addLayer(upPPSText); - - JFreeChart jfup = new JFreeChart(dpup); - jfup.setBorderPaint(Color.WHITE); - jfup.setTitle(UIEnv.getRsb().getString("uploadpps")); - jfup.getTitle().setFont(UIEnv.getFont().deriveFont(22.0f)); - jfup.getTitle().setPosition(RectangleEdge.BOTTOM); - dpup.setView(0, 0, 1, 1); - ChartPanel cpup = new ChartPanel(jfup); - cpup.setPreferredSize(dashSize); - cpup.setSize(cpup.getPreferredSize()); - cpup.setBackground(Color.WHITE); - cpup.setBorder(new LineBorder(Color.LIGHT_GRAY)); - panel_5.add(cpup, BorderLayout.WEST); - - downPPS = new DefaultValueDataset(0); - downDataPPS = new DefaultValueDataset(0); - DialPlot dpup1 = new DialPlot(); - dpup1.setDataset(1, downPPS); - dpup1.setDataset(0, downDataPPS); - StandardDialFrame sdfs1 = new StandardDialFrame(); - sdfs1.setVisible(false); - dpup1.setDialFrame(sdfs1); - StandardDialScale sds1 = new StandardDialScale(0, 100, -120, -300, 10, 5); - sds1.setTickRadius(0.9); - sds1.setTickLabelsVisible(false); - dpup1.addScale(0, sds1); - - StandardDialRange sdrr1 = new StandardDialRange(0, 70, Color.GREEN); - sdrr1.setInnerRadius(0.92); - sdrr1.setOuterRadius(0.93); - dpup1.addLayer(sdrr1); - - StandardDialRange sdrr21 = new StandardDialRange(70, 90, Color.YELLOW); - sdrr21.setInnerRadius(0.92); - sdrr21.setOuterRadius(0.93); - dpup1.addLayer(sdrr21); - - StandardDialRange sdrr31 = new StandardDialRange(90, 100, Color.RED); - sdrr31.setInnerRadius(0.92); - sdrr31.setOuterRadius(0.93); - dpup1.addLayer(sdrr31); - - DialPointer.Pointer dp111 = new DialPointer.Pointer(); - dp111.setRadius(0.8); - dp111.setFillPaint(new Color(0, 127, 0, 0)); - dp111.setOutlinePaint(new Color(0, 127, 0)); - dp111.setDatasetIndex(0); - ///dpup1.addLayer(dp111); - DialPointer.Pointer dp11 = new DialPointer.Pointer(); - dp11.setRadius(0.8); - dp11.setFillPaint(Color.GREEN); - dp11.setOutlinePaint(Color.GREEN); - dp11.setDatasetIndex(1); - dpup1.addLayer(dp11); - - downPPSText = new DialTextAnnotation("0PPS"); - downPPSText.setFont(UIEnv.getFont().deriveFont(20.0f)); - dpup1.addLayer(downPPSText); - - JFreeChart jfup1 = new JFreeChart(dpup1); - jfup1.setBorderPaint(Color.WHITE); - jfup1.setTitle(UIEnv.getRsb().getString("downloadpps")); - jfup1.getTitle().setFont(UIEnv.getFont().deriveFont(22.0f)); - jfup1.getTitle().setPosition(RectangleEdge.BOTTOM); - dpup1.setView(0, 0, 1, 1); - ChartPanel cpup1 = new ChartPanel(jfup1); - cpup1.setPreferredSize(dashSize); - cpup1.setSize(cpup1.getPreferredSize()); - cpup1.setBackground(Color.WHITE); - cpup1.setBorder(new LineBorder(Color.LIGHT_GRAY)); - panel_6.add(cpup1, BorderLayout.WEST); - }*/ - // 上行速度仪表盘 upSpeed = new DefaultValueDataset(0); upDataSpeed = new DefaultValueDataset(0); - DialPanel dialup=new DialPanel(CONST.dialwidth, UIEnv.getRsb().getString("uploadspeed"), upSpeed, upDataSpeed, Color.RED, new Color(127, 0, 0)); + DialPanel dialup=new DialPanel(CONST.dialwidth, UIEnv.getRsb().getString("uploadspeed"), upSpeed, null, Color.RED, new Color(127, 0, 0)); dialup.addGradientRanges(UIEnv.RANGE_1); upSpeedText =dialup.getTextAnnotation(); @@ -469,7 +229,7 @@ public class KperfGUI extends XFrame{ // 下行速度仪表盘 downSpeed = new DefaultValueDataset(0); downDataSpeed = new DefaultValueDataset(0); - DialPanel dialdown=new DialPanel(CONST.dialwidth, UIEnv.getRsb().getString("downloadspeed"), downSpeed, downDataSpeed, Color.GREEN, new Color(0, 127, 0)); + DialPanel dialdown=new DialPanel(CONST.dialwidth, UIEnv.getRsb().getString("downloadspeed"), downSpeed, null, Color.GREEN, new Color(0, 127, 0)); dialdown.addGradientRanges(UIEnv.RANGE_1); downSpeedText =dialdown.getTextAnnotation(); @@ -479,7 +239,7 @@ public class KperfGUI extends XFrame{ // 上行包速率仪表盘 upPPS = new DefaultValueDataset(0); upDataPPS = new DefaultValueDataset(0); - DialPanel dialupPPS=new DialPanel(CONST.dialwidth, UIEnv.getRsb().getString("uploadpps"), upPPS, upDataPPS, Color.RED, new Color(127, 0, 0)); + DialPanel dialupPPS=new DialPanel(CONST.dialwidth, UIEnv.getRsb().getString("uploadpps"), upPPS, null, Color.RED, new Color(127, 0, 0)); dialupPPS.addGradientRanges(UIEnv.RANGE_1); upPPSText = dialupPPS.getTextAnnotation(); @@ -488,7 +248,7 @@ public class KperfGUI extends XFrame{ // 下行包速率仪表盘 downPPS = new DefaultValueDataset(0); downDataPPS = new DefaultValueDataset(0); - DialPanel dialdownPPS=new DialPanel(CONST.dialwidth, UIEnv.getRsb().getString("downloadpps"), downPPS, downDataPPS, Color.GREEN, new Color(0, 127, 0)); + DialPanel dialdownPPS=new DialPanel(CONST.dialwidth, UIEnv.getRsb().getString("downloadpps"), downPPS, null, Color.GREEN, new Color(0, 127, 0)); dialdownPPS.addGradientRanges(UIEnv.RANGE_1); downPPSText = dialdownPPS.getTextAnnotation(); @@ -524,38 +284,37 @@ public class KperfGUI extends XFrame{ setSize(CONST.windowwidth,CONST.windowheight ); setResizable(false); setLocationRelativeTo(comp); - tsk=new TimerTask() { - - @Override - public void run() { - if(monitor!=null) { - double outload = monitor.getOutSpeed() * 100.0 / monitor.getOutSpeedMax(); - upSpeedText.setLabel(KLALBUtils.convertIBUint(monitor.getOutSpeed()) + "/s"); - if (Double.isFinite(outload)) { - upSpeed.setValue(Math.min(upSpeed.getValue().doubleValue() * 0.98 + outload * 0.02, 101.0)); - } + tsk=KLALBScheduler.getScheduler().scheduleWithFixedDelay(new TimerTask() { - double inload = monitor.getInSpeed() * 100.0 / monitor.getInSpeedMax(); - downSpeedText.setLabel(KLALBUtils.convertIBUint(monitor.getInSpeed()) + "/s"); - if (Double.isFinite(inload)) { - downSpeed.setValue(Math.min(downSpeed.getValue().doubleValue() * 0.98 + inload * 0.02, 101.0)); - } - - double outPPSload = monitor.getOutPPS() * 100.0 / monitor.getOutPPSMax(); - upPPSText.setLabel(KLALBUtils.convertUintDefalut(monitor.getOutPPS()) + "PPS"); - if (Double.isFinite(outPPSload)) { - upPPS.setValue(Math.min(upPPS.getValue().doubleValue() * 0.98 + outPPSload * 0.02, 101.0)); - } - - double inPPSload = monitor.getInPPS() * 100.0 / monitor.getInPPSMax(); - downPPSText.setLabel(KLALBUtils.convertUintDefalut(monitor.getInPPS()) + "PPS"); - if (Double.isFinite(inPPSload)) { - downPPS.setValue(Math.min(downPPS.getValue().doubleValue() * 0.98 + inPPSload * 0.02, 101.0)); - } - } + @Override + public void run() { + if(monitor!=null) { + double outload = monitor.getOutSpeed() * 100.0 / monitor.getOutSpeedMax(); + upSpeedText.setLabel(KLALBUtils.convertIBUint(monitor.getOutSpeed()) + "/s"); + if (Double.isFinite(outload)) { + upSpeed.setValue(Math.min(upSpeed.getValue().doubleValue() * 0.98 + outload * 0.02, 101.0)); } - }; - tmr.scheduleAtFixedRate(tsk, 200, 100); + + double inload = monitor.getInSpeed() * 100.0 / monitor.getInSpeedMax(); + downSpeedText.setLabel(KLALBUtils.convertIBUint(monitor.getInSpeed()) + "/s"); + if (Double.isFinite(inload)) { + downSpeed.setValue(Math.min(downSpeed.getValue().doubleValue() * 0.98 + inload * 0.02, 101.0)); + } + + double outPPSload = monitor.getOutPPS() * 100.0 / monitor.getOutPPSMax(); + upPPSText.setLabel(KLALBUtils.convertUintDefalut(monitor.getOutPPS()) + "PPS"); + if (Double.isFinite(outPPSload)) { + upPPS.setValue(Math.min(upPPS.getValue().doubleValue() * 0.98 + outPPSload * 0.02, 101.0)); + } + + double inPPSload = monitor.getInPPS() * 100.0 / monitor.getInPPSMax(); + downPPSText.setLabel(KLALBUtils.convertUintDefalut(monitor.getInPPS()) + "PPS"); + if (Double.isFinite(inPPSload)) { + downPPS.setValue(Math.min(downPPS.getValue().doubleValue() * 0.98 + inPPSload * 0.02, 101.0)); + } + } + } + }, 200, 100, TimeUnit.MILLISECONDS); addWindowListener(new WindowListener() { @Override @@ -585,7 +344,7 @@ public class KperfGUI extends XFrame{ @Override public void windowClosing(WindowEvent e) { if(tsk!=null) - tsk.cancel(); + tsk.cancel(false); } @Override diff --git a/src/org/kne/cloud/network/perf/Kperf.java b/src/org/kne/cloud/network/perf/Kperf.java index c15d917..aaf588f 100644 --- a/src/org/kne/cloud/network/perf/Kperf.java +++ b/src/org/kne/cloud/network/perf/Kperf.java @@ -9,6 +9,7 @@ import java.util.concurrent.TimeUnit; import org.kne.cloud.clock.AdjustedNanoClock; import org.kne.cloud.clock.HighAccuracyClock; +import org.kne.cloud.clock.NTPTimestamps; import org.kne.cloud.network.MultiProtocolSocketAddress; import org.kne.cloud.network.SpeedLimiter; import org.kne.cloud.network.ThreadTool; @@ -16,6 +17,7 @@ import org.kne.cloud.network.klalb.*; import org.kne.cloud.network.monitor.LinkStatus; import org.kne.cloud.network.monitor.SpeedAndTrafficAndDelayMonitorDataImpl; import org.kne.concurrent.ThreadParker; +import org.kne.math.Long128; public class Kperf implements Runnable{ @@ -36,8 +38,8 @@ public class Kperf implements Runnable{ private volatile KLALBPacketLink link; private LinkStatus status=new LinkStatus(); - private SpeedAndTrafficAndDelayMonitorDataImpl monitor=new SpeedAndTrafficAndDelayMonitorDataImpl(HighAccuracyClock.SYSTEM_CLOCK); - + private HighAccuracyClock sysclk =HighAccuracyClock.SYSTEM_CLOCK; + private SpeedAndTrafficAndDelayMonitorDataImpl monitor=new SpeedAndTrafficAndDelayMonitorDataImpl(sysclk); private volatile boolean closed = false; private volatile ThreadParker tlock=new ThreadParker(); @@ -200,7 +202,9 @@ public class Kperf implements Runnable{ while ((!link.isClosed()) && (!closed)) { if (checkPingTimeSleep()) { - writePacketToKPL(new PINGPacket(System.nanoTime(),0, 0)); + PINGPacket ping = new PINGPacket(sysclk.getCurrentTimeNTP64().longValue(), monitor.getOutSpeed(), + monitor.getInSpeed()); + writePacketToKPL(ping); }else { KLALBPacket kpip = IsendDequeList.poll(); @@ -243,7 +247,17 @@ public class Kperf implements Runnable{ } switch (kpp.getType()) { case KLALBPacket.PING: - sendIPacket(new PONGPacket(((PINGPacket) kpp).getTime() ,kpp.getRcvtime(), System.nanoTime(),0, 0)); + long pct = System.nanoTime() - kpp.getRcvtime(); + Long128 curr = sysclk.getCurrentTimeNanos(); + Long128 rcv = curr.subtract(pct); + + Long128 local1281 = NTPTimestamps.nanosToNtp128BitTimestamp(curr); + PONGPacket pong = new PONGPacket(((PINGPacket) kpp).getTime(), + NTPTimestamps.ntp128To64(NTPTimestamps.nanosToNtp128BitTimestamp(rcv)).longValue(), + NTPTimestamps.ntp128To64(NTPTimestamps.nanosToNtp128BitTimestamp(curr)).longValue(), + monitor.getUploadBandwidth().calculateBandwidth(500000000L), + monitor.getDownloadBandwidth().calculateBandwidth(500000000L)); + sendIPacket(pong); break; case KLALBPacket.PONG: if (fst) { @@ -260,37 +274,43 @@ public class Kperf implements Runnable{ tc.start(); fst = false; } - + PONGPacket png = (PONGPacket) kpp; - long ul = png.getRcvtime() - png.getTimepingsnd()-(png.getTimepongsnd()-png.getTimepingrcv()); + Long128 local = sysclk.getCurrentTimeNanos(); + Long128 local128 = NTPTimestamps.nanosToNtp128BitTimestamp(local); - long DsndDelayFactor0 = png.getTimepingrcv() - png.getTimepingsnd(); - long DrcvDelayFactor0 = png.getRcvtime() - png.getTimepongsnd(); - long DsndDelayFactor; - long DrcvDelayFactor; + Long128 psnd = NTPTimestamps.ntp128BitToNanosTimestamp( + NTPTimestamps.inferNtp64To128(png.getTimepingsnd(), local128)); + // System.out.println("psnd"+NTPTimestamps.nanosSince1970ToString( + // NTPTimestamps.ntp128BitToNanosTimestamp( NTPTimestamps.inferNtp64To128( + // psnd,local128)))); + Long128 prcv = NTPTimestamps.ntp128BitToNanosTimestamp( + NTPTimestamps.inferNtp64To128(png.getTimepingrcv(), local128)); + // System.out.println("prcv"+NTPTimestamps.nanosSince1970ToString( + // NTPTimestamps.ntp128BitToNanosTimestamp( NTPTimestamps.inferNtp64To128( + // prcv,local128)))); + Long128 posnd = NTPTimestamps.ntp128BitToNanosTimestamp( + NTPTimestamps.inferNtp64To128(png.getTimepongsnd(), local128)); + // System.out.println("posnd"+NTPTimestamps.nanosSince1970ToString( + // NTPTimestamps.ntp128BitToNanosTimestamp( NTPTimestamps.inferNtp64To128( + // posnd,local128)))); - adjnc.getLock().lock(); - try { - adjnc.calibrate(png.getTimepongsnd() + ul / 2 - png.getRcvtime(), ul / 2); - DsndDelayFactor = DsndDelayFactor0 - adjnc.getDelta(); - DrcvDelayFactor = DrcvDelayFactor0 + adjnc.getDelta(); - if (DsndDelayFactor < 0) { - if (DrcvDelayFactor < 0) { + long DsndDelay = prcv.subtract(psnd).longValue(); + long DrcvDelay = local.subtract(posnd).longValue(); + // System.out.println(DsndDelay+" "+DrcvDelay); - } else { - adjnc.setDelta(adjnc.getDelta() + DsndDelayFactor); - DsndDelayFactor = DsndDelayFactor0 - adjnc.getDelta(); - DrcvDelayFactor = DrcvDelayFactor0 + adjnc.getDelta(); - } - } else { - if (DrcvDelayFactor < 0) { - adjnc.setDelta(adjnc.getDelta() - DrcvDelayFactor); - DsndDelayFactor = DsndDelayFactor0 - adjnc.getDelta(); - DrcvDelayFactor = DrcvDelayFactor0 + adjnc.getDelta(); - } + long rtt = DsndDelay + DrcvDelay; + if (DsndDelay >= 0 && DrcvDelay >= 0) { + monitor.recordUploadDelay(DsndDelay); + monitor.recordDownloadDelay(DrcvDelay); + //sysclk.adjustClock(Long128.valueOf( DsndDelay-DrcvDelay), 0.001); + } else { + long Dfallback = rtt / 2; + if (Dfallback >= 0) { + monitor.recordUploadDelay(Dfallback); + monitor.recordDownloadDelay(Dfallback); } - } finally { - adjnc.getLock().unlock(); + // sysclk.adjustClock(Long128.valueOf( DsndDelay-DrcvDelay), 0.1); } diff --git a/src/org/kne/cloud/network/srv6/SRv6Router.java b/src/org/kne/cloud/network/srv6/SRv6Router.java index d600670..d521a63 100644 --- a/src/org/kne/cloud/network/srv6/SRv6Router.java +++ b/src/org/kne/cloud/network/srv6/SRv6Router.java @@ -17,13 +17,13 @@ import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.CopyOnWriteArrayList; import java.util.concurrent.CopyOnWriteArraySet; import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.LongAdder; 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 org.jctools.counters.FixedSizeStripedLongCounter; import org.kne.cloud.clock.HighAccuracyClock; import org.kne.cloud.network.congestion.MessageBatcher; import org.kne.cloud.network.ipv6.ControlledIPv6NetworkLink; @@ -48,10 +48,8 @@ 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.HighPerformanceExecutor; import org.kne.concurrent.TimeoutConcurrentHashMap; -import org.kne.io.KNEChannels; import com.google.gson.internal.Pair; @@ -501,7 +499,7 @@ public class SRv6Router { } //System.out.println("Send failed:" + iPv6Packet); - LosscountAdder.inc(1); + LosscountAdder.add(1); if (debug) { dbg.append("miss.\n"); // 路由失败 @@ -747,8 +745,8 @@ public class SRv6Router { }); } - private static HighPerformanceExecutor2 defaultExecutor = new HighPerformanceExecutor2( - (int) (Runtime.getRuntime().availableProcessors()), Thread.ofVirtual().name("SRv6 Routing").factory()); + private static HighPerformanceExecutor defaultExecutor = new HighPerformanceExecutor( + (int) (Runtime.getRuntime().availableProcessors()), 32,Thread.ofVirtual().name("SRv6 Routing").factory()); //private static HighPerformanceExecutor2 defaultExecutor = new HighPerformanceExecutor2(4, Thread.ofVirtual().name("SRv6 Routing").factory()); @@ -766,9 +764,10 @@ public class SRv6Router { break; case MULTI_FILL: case MULTI_SCATTER: - defaultExecutor.executeWithCongestionReport((state) -> { - runPacketSendTask0(sendTask, state); - }, 20,performanceStrategy); + Thread curr2=Thread.currentThread(); + defaultExecutor.execute(() -> { + runPacketSendTask0(sendTask, Thread.currentThread()==curr2); + }); break; } @@ -798,20 +797,22 @@ public class SRv6Router { } public void enqueuePacketReceiveTask(IPv6NetworkLink link, Supplier pack) { - defaultExecutor.executeWithCongestionReport((state) -> { + + Thread curr2=Thread.currentThread(); + defaultExecutor.execute(() -> { try { long start=System.nanoTime(); IPv6Packet packet = pack.get(); - if(state) { + if(Thread.currentThread()==curr2) { packet.markCE(); } onReceive(link ,packet); long time=System.nanoTime()-start; - backplaneCount(time,state); + backplaneCount(time,Thread.currentThread()==curr2); } catch (Exception e) { e.printStackTrace(); } - },20,performanceStrategy); + }); } public void enqueuePacket(IPv6Packet packet) { @@ -821,15 +822,15 @@ public class SRv6Router { } //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 LongAdder timeAdder=new LongAdder(); + private LongAdder countAdder=new LongAdder(); + private LongAdder ECNcountAdder=new LongAdder(); + private LongAdder LosscountAdder=new LongAdder(); private void backplaneCount(long time,boolean ECN) { - timeAdder.inc(time); - countAdder.inc(1); + timeAdder.add(time); + countAdder.add(1); if(ECN) { - ECNcountAdder.inc(1); + ECNcountAdder.add(1); } } private volatile long backplaneTime=0; @@ -841,10 +842,10 @@ public class SRv6Router { private void updateBackplaneCounter() { long curr=System.nanoTime(); - long totalTime=timeAdder.getAndReset(); - long totalCount=countAdder.getAndReset(); - long ECNCount=ECNcountAdder.getAndReset(); - long LossCount=LosscountAdder.getAndReset(); + long totalTime=timeAdder.sumThenReset(); + long totalCount=countAdder.sumThenReset(); + long ECNCount=ECNcountAdder.sumThenReset(); + long LossCount=LosscountAdder.sumThenReset(); if(totalCount!=0) { backplaneTime=totalTime/totalCount; ECNRate=((double)ECNCount)/totalCount; diff --git a/src/org/kne/concurrent/HighPerformanceExecutor2.java b/src/org/kne/concurrent/HighPerformanceExecutor2.java deleted file mode 100644 index 52d30a7..0000000 --- a/src/org/kne/concurrent/HighPerformanceExecutor2.java +++ /dev/null @@ -1,133 +0,0 @@ -package org.kne.concurrent; - -import java.util.Queue; -import java.util.concurrent.Executor; -import java.util.concurrent.ThreadFactory; -import java.util.concurrent.atomic.AtomicInteger; -import java.util.function.Consumer; - -import org.jctools.queues.MpscArrayQueue; -import org.kne.cloud.network.klalb.PerformanceStrategy; - -public class HighPerformanceExecutor2 implements Executor { - - private ThreadElement[] threads; - - public HighPerformanceExecutor2(int threadcount) { - //this(threadcount,Executors.defaultThreadFactory()); - this(threadcount,Thread.ofVirtual().factory()); - } - public HighPerformanceExecutor2(int threadcount,ThreadFactory th) { - threads=new ThreadElement[threadcount]; - for (int i = 0; i < threadcount; i++) { - ThreadElement t= new ThreadElement(); - th.newThread(t).start(); - threads[i]=t; - } - } - - - private static class ThreadElement implements Runnable{ - private MpscArrayQueue queue=new MpscArrayQueue(2048); - private AtomicInteger size=new AtomicInteger(); - - private volatile ThreadParker parker=new ThreadParker(); - public Queue getQueue() { - return queue; - } - public int size() { - - return queue.size(); - } - public long prev=System.nanoTime(); - public boolean putTask(Runnable e) { - boolean b=queue.offer(e); - if(b) { - // size.incrementAndGet(); - long curr=System.nanoTime(); - if(curr-prev>1000L||queue.size()>=8) { - prev=curr; - parker.unpark(); - } - } - return b; - - } - @Override - public void run() { - while(true) { - Runnable r=queue.poll(); - if(r!=null) { - // size.decrementAndGet(); - try { - r.run(); - }catch(Throwable e) { - e.printStackTrace(); - } - ThreadYieldCheckpoint.yieldCheckpoint(1000000L); - }else { - parker.parkNanos(1000000L); - - } - } - } - } - @Override - public void execute(Runnable command) { - if(execute0((x)->{command.run();},1000, PerformanceStrategy.MULTI_FILL)) { - return; - } - System.out.println("loss!"); - //backup.execute(command); - - - } - - private long vl=0; - - private boolean execute0(Consumer command, int limit, PerformanceStrategy strategy) { - if(strategy.equals(PerformanceStrategy.MULTI_SCATTER)){ - long ord=vl++; - ThreadElement te= threads[(int) (ord%threads.length)]; - boolean b=te.size()>limit; - return te.putTask( ()->{command.accept(b);}); - }else { - for (int i = 0; i < threads.length; i++) { - ThreadElement te = threads[i]; - boolean b = te.size() > limit; - if (!b) { - if (te.putTask(() -> { - command.accept(false); - })) { - return true; - } - } - } - - for (int i = 0; i < threads.length; i++) { - ThreadElement te = threads[i]; - if (te.putTask(() -> { - command.accept(true); - })) { - return true; - } - } - return false; - } - } - - public void executeWithCongestionReport(Consumer command) { - if(execute0(command,1000, PerformanceStrategy.MULTI_FILL)) { - return; - } - System.out.println("loss!"); - } - public void executeWithCongestionReport(Consumer command, int limit, PerformanceStrategy strategy) { - if(execute0(command,limit,strategy)) { - return; - } - System.out.println("loss!"); - - - } -}