From c1469d1e46cae73f1df23704b0be7013c3899ff4 Mon Sep 17 00:00:00 2001 From: Filip Leonarski Date: Sun, 22 Oct 2023 14:36:53 +0200 Subject: [PATCH] JFJochReceiver: Skip frames if acquisition finished and frames stopped earlier on the first acquisition device --- receiver/AcquisitionCounters.cpp | 46 ++++++++++++++++++------------ receiver/AcquisitionCounters.h | 10 ++++--- receiver/FPGAAcquisitionDevice.cpp | 2 +- receiver/JFJochReceiver.cpp | 9 +++++- tests/AcquisitionCountersTest.cpp | 30 ++++++++++--------- tests/FPGAIntegrationTest.cpp | 10 +++---- 6 files changed, 65 insertions(+), 42 deletions(-) diff --git a/receiver/AcquisitionCounters.cpp b/receiver/AcquisitionCounters.cpp index 126bede8..d5394fa3 100644 --- a/receiver/AcquisitionCounters.cpp +++ b/receiver/AcquisitionCounters.cpp @@ -6,13 +6,15 @@ #include "../common/JFJochException.h" AcquisitionCounters::AcquisitionCounters() - : head(max_modules, 0), slowest_head(0), total_packets(0), expected_frames(0), acquisition_finished(false) {} + : curr_frame_number(max_modules, 0), slowest_frame_number(0), fastest_frame_number(0), + total_packets(0), expected_frames(0), acquisition_finished(false) {} void AcquisitionCounters::Reset(const DiffractionExperiment &experiment, uint16_t data_stream) { std::unique_lock ul(m); acquisition_finished = false; - slowest_head = 0; + slowest_frame_number = 0; + fastest_frame_number = 0; if ((experiment.GetDetectorMode() == DetectorMode::PedestalG0) || (experiment.GetDetectorMode() == DetectorMode::PedestalG1) || @@ -25,7 +27,7 @@ void AcquisitionCounters::Reset(const DiffractionExperiment &experiment, uint16_ if (nmodules > max_modules) throw JFJochException(JFJochExceptionCategory::InputParameterInvalid, "Acquisition counter cannot support that many modules"); for (int i = 0; i < max_modules; i++) - head[i] = 0; + curr_frame_number[i] = 0; handle_for_frame = std::vector((expected_frames+1) * nmodules, HandleNotFound); packets_collected = std::vector(expected_frames * nmodules); @@ -44,14 +46,18 @@ void AcquisitionCounters::UpdateCounters(const Completion *c) { throw JFJochException(JFJochExceptionCategory::InputParameterInvalid, "UpdateCounters frame number is out of bounds"); else { - if (head.at(c->module_number) < c->frame_number) - head.at(c->module_number) = c->frame_number; + if (curr_frame_number.at(c->module_number) < c->frame_number) + curr_frame_number.at(c->module_number) = c->frame_number; - if (c->frame_number > slowest_head) { - slowest_head = head[0]; + if (c->frame_number > slowest_frame_number) { + slowest_frame_number = curr_frame_number[0]; for (int i = 1; i < nmodules; i++) - if (head[i] < slowest_head) - slowest_head = head[i]; + if (curr_frame_number[i] < slowest_frame_number) + slowest_frame_number = curr_frame_number[i]; + } + + if (c->frame_number > fastest_frame_number) { + fastest_frame_number = c->frame_number; } packets_collected.at(c->frame_number * nmodules + c->module_number) = c->packet_count; @@ -96,22 +102,26 @@ uint64_t AcquisitionCounters::GetBufferHandleAndClear(size_t frame, uint16_t mod return ret_val; } -uint64_t AcquisitionCounters::GetHead(uint16_t module_number) const { +uint64_t AcquisitionCounters::GetCurrFrameNumber(uint16_t module_number) const { if (module_number >= max_modules) throw JFJochException(JFJochExceptionCategory::ArrayOutOfBounds, - "GetHead Wrong module number: " + std::to_string(module_number)); - return head[module_number]; + "GetCurrFrameNumber Wrong module number: " + std::to_string(module_number)); + return curr_frame_number[module_number]; } -uint64_t AcquisitionCounters::GetSlowestHead() const { - return slowest_head; +uint64_t AcquisitionCounters::GetSlowestFrameNumber() const { + return slowest_frame_number; +} + +uint64_t AcquisitionCounters::GetFastestFrameNumber() const { + return fastest_frame_number; } void AcquisitionCounters::WaitForFrame(size_t curr_frame, uint16_t module_number) const { - uint64_t slowest_head_tmp = (module_number == UINT16_MAX) ? GetSlowestHead() : GetHead(module_number); + uint64_t slowest_head_tmp = (module_number == UINT16_MAX) ? GetSlowestFrameNumber() : GetCurrFrameNumber(module_number); while (!acquisition_finished && (slowest_head_tmp < curr_frame)) { std::this_thread::sleep_for(std::chrono::microseconds(100)); - slowest_head_tmp = (module_number == UINT16_MAX) ? GetSlowestHead() : GetHead(module_number); + slowest_head_tmp = (module_number == UINT16_MAX) ? GetSlowestFrameNumber() : GetCurrFrameNumber(module_number); } } @@ -119,9 +129,9 @@ int64_t AcquisitionCounters::CalculateDelay(size_t curr_frame, uint16_t module_n uint64_t slowest_head_tmp; if (module_number == UINT16_MAX) - slowest_head_tmp = GetSlowestHead(); + slowest_head_tmp = GetSlowestFrameNumber(); else - slowest_head_tmp = GetHead(module_number); + slowest_head_tmp = GetCurrFrameNumber(module_number); return slowest_head_tmp - curr_frame; } diff --git a/receiver/AcquisitionCounters.h b/receiver/AcquisitionCounters.h index 7175da3f..3540ab73 100644 --- a/receiver/AcquisitionCounters.h +++ b/receiver/AcquisitionCounters.h @@ -29,8 +29,9 @@ class AcquisitionCounters { uint64_t total_packets; std::vector packets_per_module; - uint64_t slowest_head; - std::vector head; + uint64_t slowest_frame_number; + uint64_t fastest_frame_number; + std::vector curr_frame_number; bool acquisition_finished; uint64_t expected_frames; uint64_t nmodules = max_modules; @@ -45,8 +46,9 @@ public: uint64_t GetBufferHandleAndClear(size_t frame, uint16_t module_number); - uint64_t GetHead(uint16_t module_number) const; - uint64_t GetSlowestHead() const; + uint64_t GetCurrFrameNumber(uint16_t module_number) const; + uint64_t GetSlowestFrameNumber() const; + uint64_t GetFastestFrameNumber() const; void WaitForFrame(size_t curr_frame, uint16_t module_number = UINT16_MAX) const; int64_t CalculateDelay(size_t curr_frame, uint16_t module_number = UINT16_MAX) const; // mutex acquired indirectly diff --git a/receiver/FPGAAcquisitionDevice.cpp b/receiver/FPGAAcquisitionDevice.cpp index 2e97c673..daabac6a 100644 --- a/receiver/FPGAAcquisitionDevice.cpp +++ b/receiver/FPGAAcquisitionDevice.cpp @@ -315,7 +315,7 @@ JFJochProtoBuf::FPGAStatus FPGAAcquisitionDevice::GetStatus() const { ret.set_ethernet_rx_aligned(env.ethernet_aligned); ret.set_hbm_temp_0_degc(env.hbm_0_temp_C); ret.set_hbm_temp_1_degc(env.hbm_1_temp_C); - ret.set_slowest_head(counters.GetSlowestHead()); + ret.set_slowest_head(counters.GetSlowestFrameNumber()); return ret; } diff --git a/receiver/JFJochReceiver.cpp b/receiver/JFJochReceiver.cpp index 9321d75d..5634e0cb 100644 --- a/receiver/JFJochReceiver.cpp +++ b/receiver/JFJochReceiver.cpp @@ -417,6 +417,13 @@ void JFJochReceiver::FrameTransformationThread() { uint64_t image_number; while (images_to_go.Get(image_number) != 0) { try { + // If data acquisition is finished and fastest frame for the first device is behind + size_t frame_number = image_number * experiment.GetSummation(); + acquisition_device[0]->Counters().WaitForFrame(frame_number + 2); + if (acquisition_device[0]->Counters().IsAcquisitionFinished() && + (acquisition_device[0]->Counters().GetFastestFrameNumber() < frame_number)) + continue; + DataMessage message{}; message.number = image_number; message.timestamp_base = 10*1000*1000; @@ -450,7 +457,7 @@ void JFJochReceiver::FrameTransformationThread() { message.receiver_aq_dev_delay = max_delay; } else { for (int j = 0; j < experiment.GetSummation(); j++) { - size_t frame_number = image_number * experiment.GetSummation() + j; + frame_number = image_number * experiment.GetSummation() + j; for (int d = 0; d < ndatastreams; d++) { acquisition_device[d]->Counters().WaitForFrame(frame_number + 2); diff --git a/tests/AcquisitionCountersTest.cpp b/tests/AcquisitionCountersTest.cpp index 453e0620..d7e38147 100644 --- a/tests/AcquisitionCountersTest.cpp +++ b/tests/AcquisitionCountersTest.cpp @@ -9,10 +9,11 @@ TEST_CASE("AcquisitionCountersTest","[AcquisitionDeviceCounters]") { AcquisitionCounters counters; counters.Reset(x, 0); - REQUIRE(counters.GetSlowestHead() == 0); - REQUIRE(counters.GetHead(0) == 0); - REQUIRE(counters.GetHead(1) == 0); - REQUIRE_THROWS(counters.GetHead(32)); + REQUIRE(counters.GetSlowestFrameNumber() == 0); + REQUIRE(counters.GetFastestFrameNumber() == 0); + REQUIRE(counters.GetCurrFrameNumber(0) == 0); + REQUIRE(counters.GetCurrFrameNumber(1) == 0); + REQUIRE_THROWS(counters.GetCurrFrameNumber(32)); REQUIRE(counters.CalculateDelay(2) == -2); REQUIRE(!counters.IsAcquisitionFinished()); @@ -26,9 +27,10 @@ TEST_CASE("AcquisitionCountersTest","[AcquisitionDeviceCounters]") { c.timestamp = 456; counters.UpdateCounters(&c); - REQUIRE(counters.GetSlowestHead() == 0); - REQUIRE(counters.GetHead(0) == 0); - REQUIRE(counters.GetHead(1) == 32); + REQUIRE(counters.GetSlowestFrameNumber() == 0); + REQUIRE(counters.GetFastestFrameNumber() == 32); + REQUIRE(counters.GetCurrFrameNumber(0) == 0); + REQUIRE(counters.GetCurrFrameNumber(1) == 32); REQUIRE(counters.CalculateDelay(31, 1) == 1); REQUIRE(counters.CalculateDelay(33, 1) == -1); @@ -42,9 +44,10 @@ TEST_CASE("AcquisitionCountersTest","[AcquisitionDeviceCounters]") { c.module_number = 0; counters.UpdateCounters(&c); - REQUIRE(counters.GetSlowestHead() == 15); - REQUIRE(counters.GetHead(0) == 15); - REQUIRE(counters.GetHead(1) == 32); + REQUIRE(counters.GetSlowestFrameNumber() == 15); + REQUIRE(counters.GetFastestFrameNumber() == 32); + REQUIRE(counters.GetCurrFrameNumber(0) == 15); + REQUIRE(counters.GetCurrFrameNumber(1) == 32); REQUIRE(counters.CalculateDelay(14) == 1); REQUIRE(counters.CalculateDelay(16) == -1); @@ -57,9 +60,10 @@ TEST_CASE("AcquisitionCountersTest","[AcquisitionDeviceCounters]") { counters.SetAcquisitionFinished(); - REQUIRE(counters.GetSlowestHead() == 15); - REQUIRE(counters.GetHead(0) == 15); - REQUIRE(counters.GetHead(1) == 32); + REQUIRE(counters.GetFastestFrameNumber() == 32); + REQUIRE(counters.GetSlowestFrameNumber() == 15); + REQUIRE(counters.GetCurrFrameNumber(0) == 15); + REQUIRE(counters.GetCurrFrameNumber(1) == 32); REQUIRE(counters.CalculateDelay(0) == 15); REQUIRE(counters.CalculateDelay(50) == 15-50); } diff --git a/tests/FPGAIntegrationTest.cpp b/tests/FPGAIntegrationTest.cpp index 45beea0a..956ad33d 100644 --- a/tests/FPGAIntegrationTest.cpp +++ b/tests/FPGAIntegrationTest.cpp @@ -99,7 +99,7 @@ TEST_CASE("HLS_C_Simulation_check_raw", "[FPGA][Full]") { REQUIRE_NOTHROW(test.StartAction(x)); REQUIRE_NOTHROW(test.WaitForActionComplete()); - REQUIRE(test.Counters().GetSlowestHead() == 0); + REQUIRE(test.Counters().GetSlowestFrameNumber() == 0); REQUIRE_NOTHROW(test.OutputStream().read()); REQUIRE(test.OutputStream().size() == 0); @@ -138,7 +138,7 @@ TEST_CASE("HLS_C_Simulation_check_cancel", "[FPGA][Full]") { REQUIRE_NOTHROW(test.WaitForActionComplete()); - REQUIRE(test.Counters().GetSlowestHead() == 0); + REQUIRE(test.Counters().GetSlowestFrameNumber() == 0); REQUIRE_NOTHROW(test.OutputStream().read()); REQUIRE(test.OutputStream().size() == 0); @@ -163,7 +163,7 @@ TEST_CASE("HLS_C_Simulation_check_cancel_conversion", "[FPGA][Full]") { REQUIRE_NOTHROW(test.WaitForActionComplete()); - REQUIRE(test.Counters().GetSlowestHead() == 0); + REQUIRE(test.Counters().GetSlowestFrameNumber() == 0); REQUIRE_NOTHROW(test.OutputStream().read()); REQUIRE(test.OutputStream().size() == 0); @@ -614,8 +614,8 @@ TEST_CASE("HLS_C_Simulation_check_2_trigger_convert", "[FPGA][Full]") { // address properly aligned REQUIRE((uint64_t) test.GetDeviceOutput(0,0)->pixels % 128 == 0); - REQUIRE(test.Counters().GetSlowestHead() == 0); - REQUIRE(test.Counters().GetHead(0) == 9); + REQUIRE(test.Counters().GetSlowestFrameNumber() == 0); + REQUIRE(test.Counters().GetCurrFrameNumber(0) == 9); REQUIRE_NOTHROW(test.OutputStream().read()); REQUIRE(test.OutputStream().size() == 0);