v1.0.0-rc.140 (#50)

This is an UNSTABLE release. The release has significant modifications and bug fixes, if things go wrong, it is better to revert to 1.0.0-rc.132.

* jfjoch_broker: For DECTRIS detectors, ZeroMQ link is persistent, to save time for establishing new connection
* jfjoch_broker: Minor bug fixes for rare conditions

Reviewed-on: #50

tools/jfjoch_process.cpp

@@ -308,6 +308,7 @@ int main(int argc, char **argv) {
     experiment.PixelSigned(true);
     experiment.OverwriteExistingFiles(true);
     experiment.PolarizationFactor(0.99);
+    experiment.SetFileWriterFormat(FileWriterFormat::NXmxLegacy);
 
     if (fixed_reference_unit_cell.has_value())
         experiment.SetUnitCell(*fixed_reference_unit_cell);
@@ -389,9 +390,6 @@ int main(int argc, char **argv) {
     logger.Info("Starting analysis of {} images (range {}-{}) using {} threads",
                 images_to_process, start_image, end_image, nthreads);
 
-    std::mutex reader_mutex;
-    std::mutex plots_mutex; // Protect shared plot/stats updates if they aren't thread-safe
-
     std::atomic<int> processed_count = 0;
     std::atomic<uint64_t> total_uncompressed_bytes = 0;
     std::atomic<uint64_t> max_image_number_sent = 0;
@@ -407,13 +405,6 @@ int main(int argc, char **argv) {
     std::atomic<int> finished_count = 0;
 
     auto worker = [&](int thread_id) {
-        JFJochBitShuffleCompressor compressor(experiment.GetCompressionAlgorithm());
-        std::vector<uint8_t> compressed_buffer;
-
-        compressed_buffer.resize(MaxCompressedSize(experiment.GetCompressionAlgorithm(),
-                                                   experiment.GetPixelsNum(),
-                                                   experiment.GetByteDepthImage()));
-
         // Thread-local analysis resources
         MXAnalysisWithoutFPGA analysis(experiment, mapping, pixel_mask, indexer);
 
@@ -426,27 +417,22 @@ int main(int argc, char **argv) {
             if (image_idx >= end_image) break;
 
             // Load Image
-            std::shared_ptr<JFJochReaderImage> img; {
-                std::lock_guard<std::mutex> lock(reader_mutex);
-                try {
-                    img = reader.LoadImage(image_idx);
-                } catch (const std::exception &e) {
-                    logger.Error("Failed to load image {}: {}", image_idx, e.what());
-                    continue;
-                }
+            std::shared_ptr<JFJochReaderRawImage> img;
+            try {
+                img = reader.GetRawImage(image_idx);
+            } catch (const std::exception &e) {
+                logger.Error("Failed to load image {}: {}", image_idx, e.what());
+                continue;
             }
 
             if (!img) continue;
 
-            DataMessage msg;
+            DataMessage msg{};
+
+            msg.image = img->image;
+            msg.number = image_idx;
+            msg.image_collection_efficiency = dataset->efficiency[image_idx];
 
-            msg.image = img->ImageData().image;
-            msg.number = img->ImageData().number;
-            msg.error_pixel_count = img->ImageData().error_pixel_count;
-            msg.image_collection_efficiency = img->ImageData().image_collection_efficiency;
-            msg.storage_cell = img->ImageData().storage_cell;
-            msg.user_data = img->ImageData().user_data;
-            msg.compression_time_s = img->ImageData().compression_time_s;
             total_uncompressed_bytes += msg.image.GetUncompressedSize();
 
             auto image_start_time = std::chrono::high_resolution_clock::now();
@@ -462,33 +448,15 @@ int main(int argc, char **argv) {
             auto image_end_time = std::chrono::high_resolution_clock::now();
             std::chrono::duration<float> image_duration = image_end_time - image_start_time;
 
-            auto size = compressor.Compress(compressed_buffer.data(),
-                                            img->Image().data(),
-                                            experiment.GetPixelsNum(),
-                                            sizeof(int32_t));
-
-            msg.image = CompressedImage(compressed_buffer.data(),
-                                        size, experiment.GetXPixelsNum(),
-                                        experiment.GetYPixelsNum(),
-                                        CompressedImageMode::Int32,
-                                        experiment.GetCompressionAlgorithm());
-
-            // Mimic DataMessage stats from Receiver
             msg.processing_time_s = image_duration.count();
             msg.original_number = msg.number;
             msg.run_number = experiment.GetRunNumber();
             msg.run_name = experiment.GetRunName();
-            // msg.receiver_free_send_buf  <- Not relevant for file analysis
-            // msg.receiver_aq_dev_delay   <- Not relevant for file analysis
 
-            // Update Plots/Stats (Thread safe update needed)
-            {
-                std::lock_guard<std::mutex> lock(plots_mutex);
-                plots.Add(msg, profile);
-                // Write Result
-                if (writer)
-                    writer->Write(msg);
-            }
+            plots.Add(msg, profile);
+            // Write Result
+            if (writer)
+                writer->Write(msg);
 
             // Update max sent tracking
             uint64_t current_max = max_image_number_sent.load();
@@ -501,11 +469,7 @@ int main(int argc, char **argv) {
 
             // Progress log
             if (current_idx_offset > 0 && current_idx_offset % 100 == 0) {
-                std::optional<float> indexing_rate;
-                {
-                    std::lock_guard<std::mutex> lock(plots_mutex);
-                    indexing_rate = plots.GetIndexingRate();
-                }
+                std::optional<float> indexing_rate = plots.GetIndexingRate();
 
                 const auto now = std::chrono::steady_clock::now();
                 const double elapsed_s = std::chrono::duration<double>(now - start_time).count();