diff --git a/receiver/JFJochReceiver.cpp b/receiver/JFJochReceiver.cpp index ccaaf68ee..3886fdb13 100644 --- a/receiver/JFJochReceiver.cpp +++ b/receiver/JFJochReceiver.cpp @@ -5,6 +5,7 @@ #include "../common/CUDAWrapper.h" #include "../common/time_utc.h" #include "JFJochCompressor.h" +#include "../frame_serialize/CBORStream2Serializer.h" JFJochReceiver::JFJochReceiver(const DiffractionExperiment &in_experiment, ImageBuffer &in_image_buffer, @@ -161,6 +162,101 @@ void JFJochReceiver::SaveStartMessageToImageBuffer(const StartMessage &msg) { image_buffer.SaveStartMessage(buffer); } +// Serialize a processed image into its buffer slot. The slot holds the compressed +// image plus the per-image CBOR metadata (spot list, reflection list, ...); if an +// unusually rich frame makes that metadata too large to fit alongside the image, +// drop just this frame (log + recycle the slot) instead of letting the serializer +// throw, which would abort the whole data collection. Returns true when the image +// was serialized and the slot is ready for the caller to commit. +bool JFJochReceiver::SerializeImageOrDrop(CBORStream2Serializer &serializer, DataMessage &msg, + ZeroCopyReturnValue &loc, size_t slot_size, Logger &logger) { + try { + serializer.SerializeImage(msg); + return true; + } catch (const JFJochException &) { + // Confirm this is a slot-overflow (not some other CBOR error) by re-serializing + // the metadata alone with an empty image placeholder: that always fits when the + // image is what overflowed, and gives the exact metadata size for the log. If it + // throws too, it is not a simple overflow and the exception propagates. + const auto image = msg.image; + msg.image = CompressedImage(nullptr, 0, image.GetWidth(), image.GetHeight(), + image.GetMode(), image.GetCompressionAlgorithm(), + image.GetChannel()); + CBORStream2Serializer meas(static_cast(loc.GetImage()), slot_size); + meas.SerializeImage(msg); + const size_t metadata_size = meas.GetImageAppendOffset(); + msg.image = image; + + logger.Error("Dropping image {} - CBOR metadata too large to store image: " + "metadata {} B + image {} B do not fit in buffer slot {} B", + msg.number, metadata_size, image.GetCompressedSize(), slot_size); + loc.release(); + return false; + } +} + +void JFJochReceiver::CountImage(const DataMessage &message) { + ++images_collected; + compressed_size += message.image.GetCompressedSize(); + uncompressed_size += message.image.GetUncompressedSize(); + UpdateMaxImageReceived(message.number); +} + +void JFJochReceiver::StampImage(DataMessage &message, size_t fifo_delay) { + message.original_number = message.number; + message.user_data = experiment.GetImageAppendix(); + message.run_number = experiment.GetRunNumber(); + message.run_name = experiment.GetRunName(); + message.receiver_buf_available = image_buffer.GetAvailSlots(); + message.receiver_aq_dev_delay = fifo_delay; + // Uncompressed/compressed (e.g. 7x), as JFJochReceiverFPGA and JFJochReceiver::GetFinalStatistics + if (message.image.GetCompressedSize() > 0) + message.compression_ratio = static_cast(message.image.GetUncompressedSize()) + / static_cast(message.image.GetCompressedSize()); +} + +void JFJochReceiver::AddToAverages(const DataMessage &message) { + // A stream that does not state it (DECTRIS) delivers whole images + if (message.image_collection_efficiency.value_or(1.0f) != 1.0f) + return; + saturated_pixels.Add(message.saturated_pixel_count); + error_pixels.Add(message.error_pixel_count); + if (message.roi.contains("beam")) { + roi_beam_npixel.Add(message.roi.at("beam").pixels); + roi_beam_sum.Add(message.roi.at("beam").sum); + } +} + +bool JFJochReceiver::ForwardImage(DataMessage &message, ZeroCopyReturnValue *loc) { + if (loc == nullptr) { + writer_queue_full = true; + return false; + } + auto writer_buffer = (uint8_t *) loc->GetImage(); + CBORStream2Serializer serializer(writer_buffer, experiment.GetImageBufferLocationSize()); + if (!SerializeImageOrDrop(serializer, message, *loc, experiment.GetImageBufferLocationSize(), logger)) + return false; + loc->SetImageNumber(message.number); + loc->SetImageSize(serializer.GetBufferSize()); + loc->SetIndexed(message.indexing_result.value_or(false)); + loc->ReadyToSend(); + + if (zmq_preview_socket != nullptr) + zmq_preview_socket->SendImage(writer_buffer, serializer.GetBufferSize()); + + if (zmq_metadata_socket != nullptr) + zmq_metadata_socket->AddDataMessage(message); + + if (push_images_to_writer) { + // Only count images the pusher accepted; TCP refuses them on a broken/absent connection + if (image_pusher.SendImage(*loc)) + ++images_sent; + } else + loc->release(); + UpdateMaxImageSent(message.number); + return true; +} + void JFJochReceiver::SendEndMessage() { EndMessage message{}; message.max_image_number = max_image_number_sent; diff --git a/receiver/JFJochReceiver.h b/receiver/JFJochReceiver.h index 2afe52c31..761cafea3 100644 --- a/receiver/JFJochReceiver.h +++ b/receiver/JFJochReceiver.h @@ -100,6 +100,16 @@ protected: void SendStartMessage(StartMessage message); void SendEndMessage(); void SaveStartMessageToImageBuffer(const StartMessage &msg); + + static bool SerializeImageOrDrop(CBORStream2Serializer &serializer, DataMessage &msg, + ZeroCopyReturnValue &loc, size_t slot_size, Logger &logger); + void CountImage(const DataMessage &message); + // What the receiver states about an image from a stream + void StampImage(DataMessage &message, size_t fifo_delay); + void AddToAverages(const DataMessage &message); + // Serializes into the slot and sends it on; false if there is no slot or the image does not fit + bool ForwardImage(DataMessage &message, ZeroCopyReturnValue *loc); + // The last word on the end message before it is sent: a receiver whose analysis is not the // per-image one fills in what it found. virtual void AddToEndMessage(EndMessage &message) {} diff --git a/receiver/JFJochReceiverLite.cpp b/receiver/JFJochReceiverLite.cpp index 362c4bf22..8f1abc134 100644 --- a/receiver/JFJochReceiverLite.cpp +++ b/receiver/JFJochReceiverLite.cpp @@ -7,39 +7,6 @@ using namespace std::chrono_literals; -// Serialize a processed image into its buffer slot. The slot holds the compressed -// image plus the per-image CBOR metadata (spot list, reflection list, ...); if an -// unusually rich frame makes that metadata too large to fit alongside the image, -// drop just this frame (log + recycle the slot) instead of letting the serializer -// throw, which would abort the whole data collection. Returns true when the image -// was serialized and the slot is ready for the caller to commit. -bool JFJochReceiverLite::SerializeImageOrDrop(CBORStream2Serializer &serializer, DataMessage &msg, - ZeroCopyReturnValue &loc, size_t slot_size, Logger &logger) { - try { - serializer.SerializeImage(msg); - return true; - } catch (const JFJochException &) { - // Confirm this is a slot-overflow (not some other CBOR error) by re-serializing - // the metadata alone with an empty image placeholder: that always fits when the - // image is what overflowed, and gives the exact metadata size for the log. If it - // throws too, it is not a simple overflow and the exception propagates. - const auto image = msg.image; - msg.image = CompressedImage(nullptr, 0, image.GetWidth(), image.GetHeight(), - image.GetMode(), image.GetCompressionAlgorithm(), - image.GetChannel()); - CBORStream2Serializer meas(static_cast(loc.GetImage()), slot_size); - meas.SerializeImage(msg); - const size_t metadata_size = meas.GetImageAppendOffset(); - msg.image = image; - - logger.Error("Dropping image {} - CBOR metadata too large to store image: " - "metadata {} B + image {} B do not fit in buffer slot {} B", - msg.number, metadata_size, image.GetCompressedSize(), slot_size); - loc.release(); - return false; - } -} - int64_t JFJochReceiverLite::NumberOfDataAnalysisThreads(int64_t requested_thread_number, const DiffractionExperiment& in_experiment) { int64_t number_of_images = in_experiment.GetImageNum(); @@ -253,15 +220,11 @@ void JFJochReceiverLite::MaskThread(uint32_t id) { end_message_received = true; logger.Debug("Thread {} end message received in JFJochReceiverLite", id); } else if (msg->cbor->data_message) { - ++images_collected; DataMessage data_msg = msg->cbor->data_message.value(); + CountImage(data_msg); dark_mask_analysis.AnalyzeImage(data_msg, buffer); - compressed_size += data_msg.image.GetCompressedSize(); - uncompressed_size += data_msg.image.GetUncompressedSize(); - UpdateMaxImageReceived(data_msg.number); - auto loc = image_buffer.GetImageSlot(); if (loc == nullptr) @@ -327,17 +290,9 @@ void JFJochReceiverLite::DataAnalysisThread(uint32_t id) { auto start = std::chrono::high_resolution_clock::now(); DataMessage data_msg = msg->cbor->data_message.value(); - ++images_collected; + CountImage(data_msg); collection_efficiency_sum += data_msg.image_collection_efficiency.value_or(1.0f); - auto compressed_size_img = data_msg.image.GetCompressedSize(); - auto uncompressed_size_img = data_msg.image.GetUncompressedSize(); - - compressed_size += compressed_size_img; - uncompressed_size += uncompressed_size_img; - - UpdateMaxImageReceived(data_msg.number); - auto image_start_time = std::chrono::high_resolution_clock::now(); AzimuthalIntegrationProfile profile(*az_int_mapping); @@ -348,65 +303,16 @@ void JFJochReceiverLite::DataAnalysisThread(uint32_t id) { std::chrono::duration image_duration = image_end_time - image_start_time; data_msg.processing_time_s = image_duration.count(); - data_msg.original_number = data_msg.number; - data_msg.user_data = experiment.GetImageAppendix(); - data_msg.run_number = experiment.GetRunNumber(); - data_msg.run_name = experiment.GetRunName(); - data_msg.receiver_buf_available = image_buffer.GetAvailSlots(); - data_msg.receiver_aq_dev_delay = image_puller.GetCurrentFifoUtilization(); - - // Compression ratio is uncompressed/compressed (e.g. 7x), matching the FPGA path - // (JFJochReceiverFPGA) and the aggregate (JFJochReceiver::GetFinalStatistics) - not the - // reciprocal. - if (compressed_size_img > 0) - data_msg.compression_ratio = static_cast(uncompressed_size_img) - / static_cast(compressed_size_img); - - saturated_pixels.Add(data_msg.saturated_pixel_count); - error_pixels.Add(data_msg.error_pixel_count); - - if (data_msg.roi.contains("beam")) { - roi_beam_npixel.Add(data_msg.roi["beam"].pixels); - roi_beam_sum.Add(data_msg.roi["beam"].sum); - } + StampImage(data_msg, image_puller.GetCurrentFifoUtilization()); + AddToAverages(data_msg); plots.Add(data_msg, profile); scan_result.Add(data_msg); if (!serialmx_filter.ApplyFilter(data_msg)) ++images_skipped; - else { - auto loc = image_buffer.GetImageSlot(); - - if (loc == nullptr) - writer_queue_full = true; - else { - auto writer_buffer = (uint8_t *) loc->GetImage(); - CBORStream2Serializer serializer(writer_buffer, experiment.GetImageBufferLocationSize()); - if (SerializeImageOrDrop(serializer, data_msg, *loc, - experiment.GetImageBufferLocationSize(), logger)) { - loc->SetImageNumber(data_msg.number); - loc->SetImageSize(serializer.GetBufferSize()); - loc->SetIndexed(data_msg.indexing_result.value_or(false)); - loc->ReadyToSend(); - - if (zmq_preview_socket != nullptr) - zmq_preview_socket->SendImage(writer_buffer, serializer.GetBufferSize()); - - if (zmq_metadata_socket != nullptr) - zmq_metadata_socket->AddDataMessage(data_msg); - - if (push_images_to_writer) { - // Only count images the pusher accepted; TCP refuses them on a broken/absent - // connection. Images actually written are GetImagesWritten() (ACK-based). - if (image_pusher.SendImage(*loc)) - ++images_sent; - } else - loc->release(); - UpdateMaxImageSent(data_msg.number); - } - } - } + else + ForwardImage(data_msg, image_buffer.GetImageSlot()); auto end = std::chrono::high_resolution_clock::now(); std::chrono::duration duration_total = end - start; diff --git a/receiver/JFJochReceiverLite.h b/receiver/JFJochReceiverLite.h index 277a25d90..fdaf32291 100644 --- a/receiver/JFJochReceiverLite.h +++ b/receiver/JFJochReceiverLite.h @@ -64,9 +64,6 @@ class JFJochReceiverLite : public JFJochReceiver { void MeasurementThread(); void DataAnalysisThread(uint32_t id); public: - // Shared with JFJochReceiverRugnux, which forwards a collection the same way. - static bool SerializeImageOrDrop(CBORStream2Serializer &serializer, DataMessage &msg, - ZeroCopyReturnValue &loc, size_t slot_size, Logger &logger); // Takes what the detector's start message decides into the experiment. static void ApplyStartMessage(DiffractionExperiment &experiment, const StartMessage &message); diff --git a/receiver/JFJochReceiverRugnux.cpp b/receiver/JFJochReceiverRugnux.cpp index c84b6cd13..e9941b1eb 100644 --- a/receiver/JFJochReceiverRugnux.cpp +++ b/receiver/JFJochReceiverRugnux.cpp @@ -13,7 +13,6 @@ #include "../common/HostMemory.h" #include "../common/time_utc.h" #include "../frame_serialize/CBORStream2Deserializer.h" -#include "../frame_serialize/CBORStream2Serializer.h" #include "../image_analysis/indexing/IndexerThreadPool.h" #include "../rugnux/RugnuxDefaults.h" #include "../rugnux/RugnuxLive.h" @@ -377,11 +376,8 @@ void JFJochReceiverRugnux::HoldStillsImage(const DataMessage &message) { logger.Error("Dropping image {} - the collection has {} images", message.number, experiment.GetImageNum()); return; } - ++images_collected; + CountImage(message); const size_t size = message.image.GetCompressedSize(); - compressed_size += size; - uncompressed_size += message.image.GetUncompressedSize(); - UpdateMaxImageReceived(message.number); // Copied into a slot of the image buffer, where it waits for the analysis; SendAnalysedImage writes it, with // its analysis, to a second slot, which is sent. A frame that finds the buffer full, but for a slot kept @@ -402,11 +398,7 @@ void JFJochReceiverRugnux::HoldStillsImage(const DataMessage &message) { held->message.image = CompressedImage(held->slot->GetImage(), size, message.image.GetWidth(), message.image.GetHeight(), message.image.GetMode(), message.image.GetCompressionAlgorithm(), message.image.GetChannel()); - held->message.original_number = message.number; - held->message.user_data = experiment.GetImageAppendix(); - held->message.run_number = experiment.GetRunNumber(); - held->message.run_name = experiment.GetRunName(); - held->message.receiver_aq_dev_delay = image_puller.GetCurrentFifoUtilization(); + StampImage(held->message, image_puller.GetCurrentFifoUtilization()); stills->Hold(held); } @@ -430,49 +422,13 @@ void JFJochReceiverRugnux::SendAnalysedImage(DataMessage &message, const Azimuth message.receiver_buf_available = status.available_slots; message.receiver_buf_in_preparation = status.preparation_slots; message.receiver_buf_in_sending = status.sending_slots; - if (message.image.GetCompressedSize() > 0) - message.compression_ratio = static_cast(experiment.GetPixelsNum() * experiment.GetByteDepthImage()) - / static_cast(message.image.GetCompressedSize()); - - // A stream that does not state it (DECTRIS) delivers whole images. - if (message.image_collection_efficiency.value_or(1.0f) == 1.0f) { - saturated_pixels.Add(message.saturated_pixel_count); - error_pixels.Add(message.error_pixel_count); - if (message.roi.contains("beam")) { - roi_beam_npixel.Add(message.roi["beam"].pixels); - roi_beam_sum.Add(message.roi["beam"].sum); - } - } - + AddToAverages(message); scan_result.Add(message); if (!serialmx_filter.ApplyFilter(message)) ++images_skipped; - else if (auto loc = AnalysedImageSlot(); loc == nullptr) - writer_queue_full = true; - else { - auto writer_buffer = (uint8_t *) loc->GetImage(); - CBORStream2Serializer serializer(writer_buffer, experiment.GetImageBufferLocationSize()); - if (JFJochReceiverLite::SerializeImageOrDrop(serializer, message, *loc, - experiment.GetImageBufferLocationSize(), logger)) { - loc->SetImageNumber(message.number); - loc->SetImageSize(serializer.GetBufferSize()); - loc->SetIndexed(message.indexing_result.value_or(false)); - loc->ReadyToSend(); - - if (zmq_preview_socket != nullptr) - zmq_preview_socket->SendImage(writer_buffer, serializer.GetBufferSize()); - if (zmq_metadata_socket != nullptr) - zmq_metadata_socket->AddDataMessage(message); - - if (push_images_to_writer) { - if (image_pusher.SendImage(*loc)) - ++images_sent; - } else - loc->release(); - UpdateMaxImageSent(message.number); - } - } + else + ForwardImage(message, AnalysedImageSlot()); plots.Add(message, profile); current_status.SetProgress(GetProgress()); @@ -699,49 +655,9 @@ void JFJochReceiverRugnux::ForwardThread(uint32_t id) { } else if (msg->cbor->data_message) { live_status.ImageReceived(msg->received); DataMessage data_msg = msg->cbor->data_message.value(); - ++images_collected; - - const auto compressed_size_img = data_msg.image.GetCompressedSize(); - const auto uncompressed_size_img = data_msg.image.GetUncompressedSize(); - compressed_size += compressed_size_img; - uncompressed_size += uncompressed_size_img; - UpdateMaxImageReceived(data_msg.number); - - data_msg.original_number = data_msg.number; - data_msg.user_data = experiment.GetImageAppendix(); - data_msg.run_number = experiment.GetRunNumber(); - data_msg.run_name = experiment.GetRunName(); - data_msg.receiver_buf_available = image_buffer.GetAvailSlots(); - data_msg.receiver_aq_dev_delay = image_puller.GetCurrentFifoUtilization(); - if (compressed_size_img > 0) - data_msg.compression_ratio = static_cast(uncompressed_size_img) - / static_cast(compressed_size_img); - - auto loc = image_buffer.GetImageSlot(); - if (loc == nullptr) - writer_queue_full = true; - else { - auto writer_buffer = (uint8_t *) loc->GetImage(); - CBORStream2Serializer serializer(writer_buffer, experiment.GetImageBufferLocationSize()); - if (JFJochReceiverLite::SerializeImageOrDrop(serializer, data_msg, *loc, - experiment.GetImageBufferLocationSize(), logger)) { - loc->SetImageNumber(data_msg.number); - loc->SetImageSize(serializer.GetBufferSize()); - loc->SetIndexed(false); - loc->ReadyToSend(); - - if (zmq_preview_socket != nullptr) - zmq_preview_socket->SendImage(writer_buffer, serializer.GetBufferSize()); - if (zmq_metadata_socket != nullptr) - zmq_metadata_socket->AddDataMessage(data_msg); - if (push_images_to_writer) { - if (image_pusher.SendImage(*loc)) - ++images_sent; - } else - loc->release(); - UpdateMaxImageSent(data_msg.number); - } - } + CountImage(data_msg); + StampImage(data_msg, image_puller.GetCurrentFifoUtilization()); + ForwardImage(data_msg, image_buffer.GetImageSlot()); } UpdateMaxDelay(image_puller.GetCurrentFifoUtilization()); current_status.SetProgress(GetProgress());