Spacecraft model fix

Performance result computed

Author: dariol83

README.md

@@ -131,20 +131,23 @@ TM/TC setup:
 - TM packets: with packet CRC.
 
 Scenario 1 results:
-- Processing start-up time (as per logs - first run, no cache):
-- Processing start-up time (as per logs - with cache):
-- Max TM rate without backpressure: 
-- Nb. of TM frames per second:
-- Nb. of TM packets per second:
-- Nb. of TM parameter samples decoded per second:
-- Nb. of processed items generated per second:
-- Memory usage (Windows Task Monitor):
-- CPU load:
+- Processing start-up time (as per logs - first run, no cache): 26 seconds
+- Processing start-up time (as per logs - with cache): 10 seconds
+- Max TM rate: 23.1 Mbit/sec (SLE TML level - RAF complete mode - processing backpressure propagated to the data generator)
+- Nb. of TM frames per second: ca 2.300/sec
+- Nb. of TM packets per second: ca 8.000/sec
+- Nb. of TM parameter samples decoded per second: ca 435.000/sec
+- Nb. of processed items generated per second: ca 530.000/sec
+- Memory usage (heap size): between 2 and 4 GB, Windows reports 5 GB 
+- CPU load: between 35% and 45% (equivalent of almost 6 cores fully utilised)
+
+![Connector Performance](docs/images/reatmetric-test-all-in-a-box-01.PNG "Connector Performance")
+![System Performance](docs/images/reatmetric-test-all-in-a-box-02.PNG "System Performance")
 
 Scenario 2 results:
 - Processing start-up time (as per logs - first run, no cache):
 - Processing start-up time (as per logs - with cache):
-- Max TM rate without backpressure: 
+- Max TM rate: 
 - Nb. of TM frames per second:
 - Nb. of TM packets per second:
 - Nb. of TM parameter samples decoded per second:

docs/images/reatmetric-test-all-in-a-box-01.PNG

@@ -104,7 +104,7 @@ public class SpacecraftModel implements IVirtualChannelReceiverOutput, IServiceI
     public static final String PUS_SUBTYPE_FIELD_NAME = "I-PUS-SUBTYPE";
     public static final int TM_FRAME_LENGTH = 1115;
     public static final int PACKET_GENERATION_PERIOD_MS = 1;
-    public static final int CYCLE_MODULE_SLEEP = 1; // 10
+    public static final int CYCLE_MODULE_SLEEP = 10000; // 10
     public static final int SETTER_APID = 10;
     public static final int SETTER_PUS = 69;
 
@@ -125,7 +125,7 @@ public class SpacecraftModel implements IVirtualChannelReceiverOutput, IServiceI
     private final FarmEngine farm;
 
     // TM processing part
-    private final BlockingQueue<SpacePacket> packetsToSend = new ArrayBlockingQueue<>(1000);
+    private final BlockingQueue<SpacePacket> packetsToSend = new ArrayBlockingQueue<>(2000);
     private final TmMasterChannelMuxer tmMux;
     private final Map<Integer, TmSenderVirtualChannel> id2tmvc = new TreeMap<>();
     private IPacketEncoder encoder;
@@ -134,7 +134,6 @@ public class SpacecraftModel implements IVirtualChannelReceiverOutput, IServiceI
     private List<TmPacketTemplate> pus1Packets = new LinkedList<>();
     private Map<Integer, AtomicInteger> apid2counter = new HashMap<>();
     private final ProcessingModelBasedResolver resolver;
-    private final ExecutorService frameSender = Executors.newFixedThreadPool(1); // OK, test tool
 
     private final Timer performanceSampler = new Timer("Spacecraft Sampler", true);
     private Instant lastSampleGenerationTime;
@@ -317,7 +316,7 @@ private void generatePackets() {
             try {
                 SpacePacket sp = pkt.generate();
                 if(sp != null) {
-                    packetsToSend.put(sp);
+                    addPacketToQueue(sp, false);
                     synchronized (performanceSampler) {
                         ++packetsPerSecond;
                     }
@@ -332,7 +331,7 @@ private void generatePackets() {
                 try {
                     SpacePacket spev = evpkt.generate();
                     if(spev != null) {
-                        packetsToSend.put(spev);
+                        addPacketToQueue(spev, false);
                         synchronized (performanceSampler) {
                             ++packetsPerSecond;
                         }
@@ -344,10 +343,27 @@ private void generatePackets() {
         }
     }
 
+    private void addPacketToQueue(SpacePacket sp, boolean withPriority) throws InterruptedException {
+        synchronized (packetsToSend) {
+            while (!withPriority && packetsToSend.size() >= 1000) {
+                packetsToSend.wait();
+            }
+            packetsToSend.put(sp);
+            packetsToSend.notifyAll();
+        }
+    }
+
     private void sendPackets() {
         while (running) {
             try {
-                SpacePacket sp = this.packetsToSend.take();
+                SpacePacket sp = null;
+                synchronized (packetsToSend) {
+                    while (packetsToSend.isEmpty()) {
+                        packetsToSend.wait();
+                    }
+                    sp = packetsToSend.take();
+                    packetsToSend.notifyAll();
+                }
                 // Get the TM VC to use: 0 if configured, otherwise the first
                 TmSenderVirtualChannel vcToUse = this.id2tmvc.get(0);
                 if (vcToUse == null) {
@@ -395,19 +411,21 @@ private void processCltu(long cltuId, byte[] cltu) {
     }
 
     private void sendTmFrame(TmTransferFrame tmTransferFrame) {
-        this.frameSender.execute(() -> {
-            // Send the transfer frame
-            // if vcId == 0 and vcc == 0, generate time packet: generation rate is 256
-            if (tmTransferFrame.getVirtualChannelId() == 0 && tmTransferFrame.getVirtualChannelFrameCount() == 0) {
-                generateTimePacket(Instant.now());
-            }
-            if (rafProvider.getCurrentBindingState() == ServiceInstanceBindingStateEnum.ACTIVE) {
+        // Send the transfer frame
+        // if vcId == 0 and vcc == 0, generate time packet: generation rate is 256
+        if (tmTransferFrame.getVirtualChannelId() == 0 && tmTransferFrame.getVirtualChannelFrameCount() == 0) {
+            generateTimePacket(Instant.now());
+        }
+        if (rafProvider.getCurrentBindingState() == ServiceInstanceBindingStateEnum.ACTIVE) {
+            try {
                 boolean result = rafProvider.transferData(tmTransferFrame.getFrame(), 0, 1, Instant.now(), false, StringUtil.toHexDump("ANTENNA-TEST".getBytes(StandardCharsets.ISO_8859_1)), false, new byte[0]);
                 if (!result) {
                     System.out.println("Error transferring TF");
                 }
+            } catch (Exception e) {
+                e.printStackTrace();
             }
-        });
+        }
     }
 
     private void generateTimePacket(Instant now) {
@@ -422,7 +440,7 @@ private void generateTimePacket(Instant now) {
         builder.addData(obt);
         SpacePacket sp = builder.build();
         try {
-            this.packetsToSend.put(sp);
+            addPacketToQueue(sp, true);
             synchronized (performanceSampler) {
                 ++packetsPerSecond;
             }
@@ -548,7 +566,7 @@ private void queuePus1(SpacePacket packet, int subtype) {
                 SpacePacket sp = pt.generate(Arrays.copyOfRange(packet.getPacket(), 0, 4));
                 System.out.println("Queueing PUS 1 packet: " + StringUtil.toHexDump(sp.getPacket()));
                 try {
-                    this.packetsToSend.put(sp);
+                    addPacketToQueue(sp, false);
                     synchronized (performanceSampler) {
                         ++packetsPerSecond;
                     }
@@ -565,7 +583,7 @@ private void queuePus1(SpacePacket packet, int subtype) {
                 SpacePacket sp = pt.generate(Arrays.copyOfRange(packet.getPacket(), 0, 4));
                 System.out.println("Queueing PUS 1 packet (def): " + StringUtil.toHexDump(sp.getPacket()));
                 try {
-                    this.packetsToSend.put(sp);
+                    addPacketToQueue(sp, false);
                     synchronized (performanceSampler) {
                         ++packetsPerSecond;
                     }