Improvement in internal frame generator

Improve way how internal packet generator is executed (mostly in HLS simulation), allowing to make it more similar to "standard" network collection. This required to modify network stack cores, so they can be executed in parallel in HLS simulation.

acquisition_device/HLSSimulatedDevice.cpp

@@ -145,31 +145,36 @@ void HLSSimulatedDevice::FPGA_StartAction(const DiffractionExperiment &experimen
     run_data_collection = 1;
     cancel_data_collection = 0;
     idle = false;
+
+    while (!din_frame_generator.empty())
+        din_frame_generator.read();
+
     datamover_out.ClearCompletedDescriptors();
-    if (experiment.IsUsingInternalPacketGen()) {
-        auto ret = frame_generator(din_frame_generator,
-                                   hbm.data(),
-                                   hbm.data(),
-                                   hbm_if_size,
-                                   experiment.GetFrameNum() + DELAY_FRAMES_STOP_AND_QUIT + 1,
-                                   experiment.GetModulesNum(data_stream),
-                                   mac_addr,
-                                   mac_addr,
-                                   ipv4_addr,
-                                   ipv4_addr,
-                                   INT_PKT_GEN_BUNCHID,
-                                   INT_PKT_GEN_EXPTTIME,
-                                   INT_PKT_GEN_DEBUG,
-                                   cancel_data_collection);
-        if (ret)
-            throw JFJochException(JFJochExceptionCategory::AcquisitionDeviceError,
-                                  "Error running internal packet generator");
-    } else {
-        data_source = STREAM_MERGE_SRC_100G;
-    }
+
     action_thread = std::thread(&HLSSimulatedDevice::HLSMainThread, this );
 }
 
+void HLSSimulatedDevice::FrameGeneratorFuture(FrameGeneratorConfig config) {
+    frame_generator(din_frame_generator,
+                    hbm.data(),
+                    hbm.data(),
+                    hbm_if_size,
+                    config.frames,
+                    config.modules,
+                    mac_addr,
+                    config.dest_mac_addr,
+                    ipv4_addr,
+                    config.dest_ipv4_addr,
+                    config.bunchid,
+                    config.exptime,
+                    config.debug,
+                    cancel_data_collection);
+}
+
+void HLSSimulatedDevice::HW_RunInternalGenerator(const FrameGeneratorConfig &config) {
+    frame_generator_future = std::async(std::launch::async, &HLSSimulatedDevice::FrameGeneratorFuture, this, config);
+}
+
 void HLSSimulatedDevice::FPGA_EndAction() {
     if (action_thread.joinable())
         action_thread.join();
@@ -279,64 +284,106 @@ void HLSSimulatedDevice::HLSMainThread() {
     hls::stream<ap_uint<512>> spot_finder_result_1;
 
     hls::stream<ap_uint<UDP_METADATA_STREAM_WIDTH> > udp_metadata;
-    ap_uint<1> idle_data_collection;
+    volatile ap_uint<1> idle_data_collection = 1;
     ap_uint<1> load_to_hbm_idle;
     ap_uint<1> save_to_hbm_idle;
     ap_uint<1> integration_idle;
     ap_uint<1> stream_conv_idle;
     ap_uint<1> frame_summation_idle;
 
-    if (data_source == STREAM_MERGE_SRC_FRAME_GEN) {
-        while (!din_frame_generator.empty())
-            stream_merge(din_eth, din_eth_4x10G, din_frame_generator, network0, data_source);
-    } else if (data_source == STREAM_MERGE_SRC_100G) {
-        while (!din_eth.empty())
-            stream_merge(din_eth, din_eth_4x10G, din_frame_generator, network0, data_source);
-    } else if (data_source == STREAM_MERGE_SRC_4x10G) {
-        while (!din_eth_4x10G.empty())
-            stream_merge(din_eth, din_eth_4x10G, din_frame_generator, network0, data_source);
-    }
+    volatile bool done = false;
+    volatile bool udp_done = false; // done AND udp_idle
+    volatile bool sls_done = false; // done AND sls_idle
+
+    // Sent gratuitous ARP message
+    arp(arp1, dout_eth, mac_addr, ipv4_addr, 1, 1);
+
+    Logger logger_hls("HLS");
+
+    volatile rcv_state_t state = RCV_INIT;
+    // Start data collection
+    data_collection_fsm(raw0, raw1,
+                        addr0, addr1,
+                        run_data_collection,
+                        cancel_data_collection,
+                        idle_data_collection,
+                        cfg.mode,
+                        float2uint(cfg.energy_kev),
+                        cfg.nframes,
+                        cfg.nmodules,
+                        cfg.nstorage_cells,
+                        cfg.nsummation , state);
+
+    run_data_collection = 0;
+    data_collection_fsm(raw0, raw1,
+                        addr0, addr1,
+                        run_data_collection,
+                        cancel_data_collection,
+                        idle_data_collection,
+                        cfg.mode,
+                        float2uint(cfg.energy_kev),
+                        cfg.nframes,
+                        cfg.nmodules,
+                        cfg.nstorage_cells,
+                        cfg.nsummation, state);
 
-    while(!network0.empty())
-        ethernet(network0, ip1, arp1, mac_addr, eth_packets, clear_counters);
-
-    while(!ip1.empty())
-        ipv4(ip1, udp1, icmp1, ipv4_addr);
-
-    arp(arp1,
-        dout_eth,
-        mac_addr,
-        ipv4_addr,
-        1, run_data_collection);
-
-    while (!arp1.empty()) {
-        arp(arp1,
-            dout_eth,
-            mac_addr,
-            ipv4_addr,
-            1, run_data_collection);
-    }
+    hls_cores.emplace_back([&] {
+        while (!udp_done) {
+            if (din_eth.empty() && din_eth_4x10G.empty() && din_frame_generator.empty())
+                std::this_thread::sleep_for(std::chrono::microseconds(10));
+            else
+                stream_merge(din_eth, din_eth_4x10G, din_frame_generator, network0, data_source);
+        }
+        logger_hls.Info("Stream_merge done");
+    });
 
-    // reset static counter
-    arp(arp1,
-        dout_eth,
-        mac_addr,
-        ipv4_addr,
-        0, run_data_collection);
+    hls_cores.emplace_back([&] {
+        while (!udp_done) {
+            if (network0.empty())
+                std::this_thread::sleep_for(std::chrono::microseconds(10));
+            else
+                ethernet(network0, ip1, arp1, mac_addr, eth_packets, clear_counters);
+        }
+        logger_hls.Info("ethernet done");
+    });
 
-    while(!icmp1.empty())
-        icmp(icmp1, dout_eth, icmp_packets, clear_counters);
+    hls_cores.emplace_back([&] {
+        while (!udp_done) {
+            if (ip1.empty())
+                std::this_thread::sleep_for(std::chrono::microseconds(10));
+            else
+                ipv4(ip1, udp1, icmp1, ipv4_addr);
+        }
+        logger_hls.Info("ipv4 done");
+    });
 
-    while (!udp1.empty())
-        udp(udp1, udp2, udp_metadata, udp_packets, clear_counters);
+    hls_cores.emplace_back([&] {
+        ap_uint<1> udp_idle = 1;
+        while (!done || !udp_idle) {
+            if (udp1.empty())
+                std::this_thread::sleep_for(std::chrono::microseconds(10));
+            else
+                udp(udp1, udp2, udp_metadata, udp_packets, clear_counters, udp_idle);
+        }
+        udp_done = true;
+        logger_hls.Info("udp done");
+    });
 
-    while (!udp2.empty())
-        sls_detector(udp2, udp_metadata, raw0, addr0, sls_packets, udp_eth_err, udp_len_err, clear_counters);
+    hls_cores.emplace_back([&] {
+        ap_uint<1> sls_idle = 1;
+        while (!done || !sls_idle) {
+            if (udp2.empty())
+                std::this_thread::sleep_for(std::chrono::microseconds (10));
+            else
+                sls_detector(udp2, udp_metadata, raw0, addr0, sls_packets, udp_eth_err, udp_len_err, clear_counters, sls_idle);
+        }
+        sls_done = true;
+        logger_hls.Info("sls_detector done");
+    });
 
     // 1. Parse incoming UDP packets
-    idle_data_collection = 0;
     hls_cores.emplace_back([&] {
-        while ((idle_data_collection == 0) || (!raw0.empty()))  {
+        while ((state != RCV_WAIT_FOR_START) || (idle_data_collection.read() == 0) || (!raw0.empty()))  {
             data_collection_fsm(raw0, raw1,
                                 addr0, addr1,
                                 run_data_collection,
@@ -347,9 +394,12 @@ void HLSSimulatedDevice::HLSMainThread() {
                                 cfg.nframes,
                                 cfg.nmodules,
                                 cfg.nstorage_cells,
-                                cfg.nsummation);
-            run_data_collection = 0;
+                                cfg.nsummation,
+                                state);
         }
+        done = true;
+
+        logger_hls.Info("data collection done");
     });
 
     // 2. Cache images in HBM
@@ -430,102 +480,112 @@ void HLSSimulatedDevice::HLSMainThread() {
     for (auto &i : hls_cores)
         i.join();
 
-    if (!din_eth.empty())
-        throw std::runtime_error("din_eth queue not empty");
+    if (frame_generator_future.valid())
+        frame_generator_future.get();
 
-    if (!addr1.empty())
-        throw std::runtime_error("Addr1 queue not empty");
+    // reset static counter
+    arp(arp1, dout_eth, mac_addr, ipv4_addr, 0, 1);
+    try {
+        while (!din_eth.empty())
+            din_eth.read();
 
-    if (!addr2.empty())
-        throw std::runtime_error("Addr2 queue not empty");
+        if (!addr1.empty())
+            throw std::runtime_error("Addr1 queue not empty");
 
-    if (!addr3.empty())
-        throw std::runtime_error("Addr3 queue not empty");
+        if (!addr2.empty())
+            throw std::runtime_error("Addr2 queue not empty");
 
-    if (!raw1.empty())
-        throw std::runtime_error("Raw1 queue not empty");
+        if (!addr3.empty())
+            throw std::runtime_error("Addr3 queue not empty");
 
-    if (!raw2.empty())
-        throw std::runtime_error("Raw2 queue not empty");
+        if (!raw1.empty())
+            throw std::runtime_error("Raw1 queue not empty");
 
-    if (!raw3.empty())
-        throw std::runtime_error("Raw3 queue not empty");
+        if (!raw2.empty())
+            throw std::runtime_error("Raw2 queue not empty");
 
-    if (!converted_0.empty())
-        throw std::runtime_error("Converted_0 queue not empty");
+        if (!raw3.empty())
+            throw std::runtime_error("Raw3 queue not empty");
 
-    if (!converted_1.empty())
-        throw std::runtime_error("Converted_1 queue not empty");
+        if (!converted_0.empty())
+            throw std::runtime_error("Converted_0 queue not empty");
 
-    if (!converted_2.empty())
-        throw std::runtime_error("Converted_2 queue not empty");
+        if (!converted_1.empty())
+            throw std::runtime_error("Converted_1 queue not empty");
 
-    if (!converted_3.empty())
-        throw std::runtime_error("Converted_3 queue not empty");
+        if (!converted_2.empty())
+            throw std::runtime_error("Converted_2 queue not empty");
 
-    if (!converted_4.empty())
-        throw std::runtime_error("Converted_4 queue not empty");
+        if (!converted_3.empty())
+            throw std::runtime_error("Converted_3 queue not empty");
 
-    if (!converted_5.empty())
-        throw std::runtime_error("Converted_5 queue not empty");
+        if (!converted_4.empty())
+            throw std::runtime_error("Converted_4 queue not empty");
 
-    if (!converted_6.empty())
-        throw std::runtime_error("Converted_6 queue not empty");
+        if (!converted_5.empty())
+            throw std::runtime_error("Converted_5 queue not empty");
 
-    if (!converted_7.empty())
-        throw std::runtime_error("Converted_7 queue not empty");
+        if (!converted_6.empty())
+            throw std::runtime_error("Converted_6 queue not empty");
 
-    if (!converted_8.empty())
-        throw std::runtime_error("Converted_8 queue not empty");
+        if (!converted_7.empty())
+            throw std::runtime_error("Converted_7 queue not empty");
 
-    if (!compl0.empty())
-        throw std::runtime_error("Compl0 queue not empty");
+        if (!converted_8.empty())
+            throw std::runtime_error("Converted_8 queue not empty");
 
-    if (!compl1.empty())
-        throw std::runtime_error("Compl1 queue not empty");
+        if (!compl0.empty())
+            throw std::runtime_error("Compl0 queue not empty");
 
-    if (!compl2.empty())
-        throw std::runtime_error("Compl2 queue not empty");
+        if (!compl1.empty())
+            throw std::runtime_error("Compl1 queue not empty");
 
-    if (!compl3.empty())
-        throw std::runtime_error("Compl3 queue not empty");
+        if (!compl2.empty())
+            throw std::runtime_error("Compl2 queue not empty");
 
-    if (!compl4.empty())
-        throw std::runtime_error("Compl4 queue not empty");
+        if (!compl3.empty())
+            throw std::runtime_error("Compl3 queue not empty");
 
-    if (!compl5.empty())
-        throw std::runtime_error("Compl5 queue not empty");
+        if (!compl4.empty())
+            throw std::runtime_error("Compl4 queue not empty");
 
-    if (!compl6.empty())
-        throw std::runtime_error("Compl6 queue not empty");
+        if (!compl5.empty())
+            throw std::runtime_error("Compl5 queue not empty");
 
-    if (!compl7.empty())
-        throw std::runtime_error("Compl7 queue not empty");
+        if (!compl6.empty())
+            throw std::runtime_error("Compl6 queue not empty");
 
-    if (!stream_768_0.empty())
-        throw std::runtime_error("Stream_768_0 queue not empty");
+        if (!compl7.empty())
+            throw std::runtime_error("Compl7 queue not empty");
 
-    if (!stream_768_1.empty())
-        throw std::runtime_error("Stream_768_1 queue not empty");
+        if (!stream_768_0.empty())
+            throw std::runtime_error("Stream_768_0 queue not empty");
 
-    if (!stream_768_2.empty())
-        throw std::runtime_error("Stream_768_2 queue not empty");
+        if (!stream_768_1.empty())
+            throw std::runtime_error("Stream_768_1 queue not empty");
 
-    if (!hbm_handles.empty())
-        throw std::runtime_error("Handles queue not empty");
+        if (!stream_768_2.empty())
+            throw std::runtime_error("Stream_768_2 queue not empty");
 
-    if (!integration_result_0.empty())
-        throw std::runtime_error("Integration result queue not empty");
+        if (!hbm_handles.empty())
+            throw std::runtime_error("Handles queue not empty");
 
-    if (!integration_result_1.empty())
-        throw std::runtime_error("Integration result queue not empty");
+        if (!integration_result_0.empty())
+            throw std::runtime_error("Integration result queue not empty");
 
-    if (!datamover_in.GetDataStream().empty())
-        throw std::runtime_error("Datamover queue is not empty");
+        if (!integration_result_1.empty())
+            throw std::runtime_error("Integration result queue not empty");
 
-    while (!datamover_out.IsIdle())
-        std::this_thread::sleep_for(std::chrono::milliseconds(100));
+        if (!datamover_in.GetDataStream().empty())
+            throw std::runtime_error("Datamover queue is not empty");
 
+        while (!datamover_out.IsIdle())
+            std::this_thread::sleep_for(std::chrono::milliseconds(100));
+    } catch (const std::runtime_error &e) {
+        if (logger)
+            logger->ErrorException(e);
+        throw e;
+    }
     if (logger)
         logger->Info("Packets Eth {} UDP {} SLS {} Proc {}", eth_packets, udp_packets, sls_packets, packets_processed);
 
@@ -567,4 +627,4 @@ void HLSSimulatedDevice::HW_SetDataSource(uint32_t val) {
 
 uint32_t HLSSimulatedDevice::HW_GetDataSource() {
     return data_source;
-}
\ No newline at end of file
+}