mx/Jungfraujoch
2
1
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 72 Wiki Activity

JFJochFrameSerializer: for the buffer return pointer + size, not vector reference (allows for more flexibility)

Browse Source
This commit is contained in:
leonarski_f
2023-05-01 20:26:15 +02:00
parent 60d95dc0d7
commit af2ed525a0
15 changed files with 50 additions and 76 deletions
Download Patch File Download Diff File Expand all files Collapse all files
common/ImagePusher.cpp
-12
View File
@@ -15,15 +15,3 @@ void PrepareCBORImage(DataMessage& message,
message.image.algorithm = experiment.GetCompressionAlgorithmEnum();
message.image.channel = "default";
}
void PrepareDataMessageSpots(DataMessage& message,
const std::vector<DiffractionSpot>& spots) {
message.spots.clear();
for (const auto & spot : spots)
message.spots.push_back(spot);
}
void ImagePusher::SendData(const std::vector<uint8_t> &serialized_image, int64_t image_number) {
SendDataInternal(serialized_image, image_number);
}
common/ImagePusher.h
+1 -3
View File
@@ -18,12 +18,10 @@ void PrepareCBORImage(DataMessage& message,
void *image, size_t image_size);
class ImagePusher {
protected:
virtual void SendDataInternal(const std::vector<uint8_t>& serialized_image, int64_t image_number) = 0;
public:
virtual void StartDataCollection(const StartMessage& message) = 0;
virtual void EndDataCollection(const EndMessage& message) = 0;
void SendData(const std::vector<uint8_t>& serialized_image, int64_t image_number);
virtual void SendImage(const uint8_t *image_data, size_t image_size, int64_t image_number) = 0;
};
common/TestImagePusher.cpp
+2 -2
View File
@@ -29,14 +29,14 @@ void TestImagePusher::EndDataCollection(const EndMessage& message) {
is_running = false;
}
void TestImagePusher::SendDataInternal(const std::vector<uint8_t> &serialized_image, int64_t image_number) {
void TestImagePusher::SendImage(const uint8_t *image_data, size_t image_size, int64_t image_number) {
std::unique_lock<std::mutex> ul(m);
frame_counter++;
if (image_number == image_id) {
JFJochFrameDeserializer deserializer;
deserializer.Process(serialized_image);
deserializer.Process(image_data, image_size);
auto image_array = deserializer.GetDataMessage();
receiver_generated_image.resize(image_array.image.size);
common/TestImagePusher.h
+1 -1
View File
@@ -19,8 +19,8 @@ class TestImagePusher : public ImagePusher {
bool correct_sequence = true;
bool is_running = false;
size_t frame_counter = 0;
void SendDataInternal(const std::vector<uint8_t> &serialized_image, int64_t image_number) override;
public:
void SendImage(const uint8_t *image_data, size_t image_size, int64_t image_number) override;
explicit TestImagePusher(int64_t image_number);
void StartDataCollection(const StartMessage& message) override;
void EndDataCollection(const EndMessage& message) override;
common/ZMQImagePusher.cpp
+5 -7
View File
@@ -40,11 +40,11 @@ ZMQImagePusher::ZMQImagePusher(const std::vector<std::string> &addr,
}
}
void ZMQImagePusher::SendDataInternal(const std::vector<uint8_t> &serialized_image, int64_t image_number) {
void ZMQImagePusher::SendImage(const uint8_t *image_data, size_t image_size, int64_t image_number) {
if (sockets.empty())
return;
auto socket_number = (image_number % file_count) % sockets.size();
sockets[socket_number]->Send(serialized_image.data(), serialized_image.size());
sockets[socket_number]->Send(image_data, image_size);
}
void ZMQImagePusher::StartDataCollection(const StartMessage& message) {
@@ -56,9 +56,9 @@ void ZMQImagePusher::StartDataCollection(const StartMessage& message) {
file_count = message.data_file_count;
serializer.SerializeSequenceStart(message);
auto &buffer = serializer.GetBuffer();
for (const auto &s: sockets)
s->Send(buffer.data(), buffer.size(), true);
s->Send(serializer.GetBuffer(), serializer.GetBufferSize(), true);
}
@@ -69,9 +69,7 @@ void ZMQImagePusher::EndDataCollection(const EndMessage& message) {
for (const auto &s: sockets) {
serializer.SerializeSequenceEnd(end_message);
auto &buffer = serializer.GetBuffer();
s->Send(buffer.data(), buffer.size(), true);
s->Send(serializer.GetBuffer(), serializer.GetBufferSize(), true);
end_message.write_master_file = false;
}
}
common/ZMQImagePusher.h
+1 -1
View File
@@ -16,8 +16,8 @@ class ZMQImagePusher : public ImagePusher {
std::vector<std::unique_ptr<ZMQContext>> contexts;
std::vector<std::unique_ptr<ZMQSocket>> sockets;
int64_t file_count = 1;
void SendDataInternal(const std::vector<uint8_t>& serialized_image, int64_t image_number) override;
public:
void SendImage(const uint8_t *image_data, size_t image_size, int64_t image_number) override;
ZMQImagePusher(ZMQContext &context, const std::vector<std::string>& addr,
int32_t send_buffer_high_watermark = -1, int32_t send_buffer_size = -1);
// High performance implementation, where each socket has dedicated ZMQ context
frame_serialize/JFJochFrameDeserializer.cpp
+5 -1
View File
@@ -603,12 +603,16 @@ void JFJochFrameDeserializer::ProcessImageData(CborValue &value) {
}
void JFJochFrameDeserializer::Process(const std::vector<uint8_t> &buffer) {
Process(buffer.data(), buffer.size());
}
void JFJochFrameDeserializer::Process(const uint8_t *msg, size_t msg_size) {
std::unique_lock<std::mutex> ul(m);
data_message = DataMessage();
CborParser parser;
CborValue value;
cborErr(cbor_parser_init(buffer.data(), buffer.size(), 0, &parser, &value));
cborErr(cbor_parser_init(msg, msg_size, 0, &parser, &value));
if (GetCBORTag(value) != CborSignatureTag)
throw JFJochException(JFJochExceptionCategory::CBORError, "CBOR must start with dedicated tag");
frame_serialize/JFJochFrameDeserializer.h
+1
View File
@@ -47,6 +47,7 @@ private:
bool ProcessEndMessageElement(CborValue &value);
public:
void Process(const std::vector<uint8_t>& buffer);
void Process(const uint8_t *msg, size_t msg_size);
[[nodiscard]] Type GetType() const;
[[nodiscard]] EndMessage GetEndMessage() const;
[[nodiscard]] StartMessage GetStartMessage() const;
frame_serialize/JFJochFrameSerializer.cpp
+13 -15
View File
@@ -52,9 +52,9 @@ void CBOR_ENC_COMPRESSED(CborEncoder &encoder,
const void *image, size_t image_size,
CompressionAlgorithm algorithm,
size_t elem_size) {
if (algorithm == CompressionAlgorithm::NO_COMPRESSION) {
if (algorithm == CompressionAlgorithm::NO_COMPRESSION)
cborErr(cbor_encode_byte_string(&encoder, (uint8_t *) image, image_size));
} else {
else {
cbor_encode_tag(&encoder, TagDECTRISCompression);
CborEncoder arrayEncoder;
@@ -288,16 +288,19 @@ inline void CBOR_ENC_USER_DATA(CborEncoder &encoder, const StartMessage& message
}
JFJochFrameSerializer::JFJochFrameSerializer(size_t in_max_buffer_size) :
max_buffer_size(in_max_buffer_size) {
buffer.reserve(in_max_buffer_size);
max_buffer_size(in_max_buffer_size), curr_size(0) {
buffer.resize(max_buffer_size);
}
const std::vector<uint8_t> &JFJochFrameSerializer::GetBuffer() const {
return buffer;
const uint8_t *JFJochFrameSerializer::GetBuffer() const {
return buffer.data();
}
size_t JFJochFrameSerializer::GetBufferSize() const {
return curr_size;
}
void JFJochFrameSerializer::SerializeSequenceStart(const StartMessage& message) {
buffer.resize(max_buffer_size);
CborEncoder encoder, mapEncoder;
cbor_encoder_init(&encoder, buffer.data(), buffer.size(), 0);
@@ -342,13 +345,10 @@ void JFJochFrameSerializer::SerializeSequenceStart(const StartMessage& message)
CBOR_ENC_CHANNELS(mapEncoder, "channels", message.channels);
cborErr(cbor_encoder_close_container(&encoder, &mapEncoder));
buffer.resize(cbor_encoder_get_buffer_size(&encoder, buffer.data()));
curr_size = cbor_encoder_get_buffer_size(&encoder, buffer.data());
}
void JFJochFrameSerializer::SerializeSequenceEnd(const EndMessage& message) {
buffer.resize(max_buffer_size);
CborEncoder encoder, mapEncoder;
cbor_encoder_init(&encoder, buffer.data(), buffer.size(), 0);
cborErr(cbor_encode_tag(&encoder,CborSignatureTag ));
@@ -365,12 +365,10 @@ void JFJochFrameSerializer::SerializeSequenceEnd(const EndMessage& message) {
cborErr(cbor_encoder_close_container(&encoder, &mapEncoder));
buffer.resize(cbor_encoder_get_buffer_size(&encoder, buffer.data()));
curr_size = cbor_encoder_get_buffer_size(&encoder, buffer.data());
}
void JFJochFrameSerializer::SerializeImage(const DataMessage& message) {
buffer.resize(max_buffer_size);
CborEncoder encoder, mapEncoder, userDataMapEncoder;
cbor_encoder_init(&encoder, buffer.data(), buffer.size(), 0);
@@ -397,5 +395,5 @@ void JFJochFrameSerializer::SerializeImage(const DataMessage& message) {
cborErr(cbor_encoder_close_container(&encoder, &mapEncoder));
buffer.resize(cbor_encoder_get_buffer_size(&encoder, buffer.data()));
curr_size = cbor_encoder_get_buffer_size(&encoder, buffer.data());
}
frame_serialize/JFJochFrameSerializer.h
+3 -1
View File
@@ -15,9 +15,11 @@
class JFJochFrameSerializer {
std::vector<uint8_t> buffer;
size_t max_buffer_size;
size_t curr_size;
public:
explicit JFJochFrameSerializer(size_t buffer_size);
[[nodiscard]] const std::vector<uint8_t> &GetBuffer() const;
[[nodiscard]] const uint8_t *GetBuffer() const;
[[nodiscard]] size_t GetBufferSize() const;
void SerializeSequenceStart(const StartMessage& message);
void SerializeSequenceEnd(const EndMessage& message);
void SerializeImage(const DataMessage& message);
image_analysis/fast-feedback-indexer
+1 -1
Submodule image_analysis/fast-feedback-indexer updated: 111b4975ca...6eccc70b52
receiver/JFJochReceiver.cpp
+1 -1
View File
@@ -413,7 +413,7 @@ int64_t JFJochReceiver::FrameTransformationThread() {
if (push_images_to_writer) {
PrepareCBORImage(message, experiment, writer_buffer.data(), image_size);
serializer.SerializeImage(message);
image_pusher.SendData(serializer.GetBuffer(), image_number);
image_pusher.SendImage(serializer.GetBuffer(), serializer.GetBufferSize(), image_number);
}
UpdateMaxImage(image_number);
tests/CBORTest.cpp
+12 -27
View File
@@ -63,10 +63,8 @@ TEST_CASE("CBORSerialize_Start", "[CBOR]") {
REQUIRE_NOTHROW(serializer.SerializeSequenceStart(message));
auto image = serializer.GetBuffer();
JFJochFrameDeserializer deserializer;
REQUIRE_NOTHROW(deserializer.Process(image));
REQUIRE_NOTHROW(deserializer.Process(serializer.GetBuffer(), serializer.GetBufferSize()));
REQUIRE(deserializer.GetType() == JFJochFrameDeserializer::Type::START);
StartMessage output_message;
@@ -137,11 +135,9 @@ TEST_CASE("CBORSerialize_End", "[CBOR]") {
};
REQUIRE_NOTHROW(serializer.SerializeSequenceEnd(message));
auto image = serializer.GetBuffer();
JFJochFrameDeserializer deserializer;
REQUIRE_NOTHROW(deserializer.Process(image));
REQUIRE_NOTHROW(deserializer.Process(serializer.GetBuffer(), serializer.GetBufferSize()));
REQUIRE(deserializer.GetType() == JFJochFrameDeserializer::Type::END);
EndMessage output_message{};
@@ -188,10 +184,9 @@ TEST_CASE("CBORSerialize_Image", "[CBOR]") {
};
REQUIRE_NOTHROW(serializer.SerializeImage(message));
auto serialized = serializer.GetBuffer();
JFJochFrameDeserializer deserializer;
REQUIRE_NOTHROW(deserializer.Process(serialized));
REQUIRE_NOTHROW(deserializer.Process(serializer.GetBuffer(), serializer.GetBufferSize()));
REQUIRE(deserializer.GetType() == JFJochFrameDeserializer::Type::IMAGE);
auto image_array = deserializer.GetDataMessage();
@@ -241,13 +236,11 @@ TEST_CASE("CBORSerialize_Image_2", "[CBOR]") {
};
REQUIRE_NOTHROW(serializer.SerializeImage(message));
auto serialized = serializer.GetBuffer();
JFJochFrameDeserializer deserializer;
REQUIRE_NOTHROW(deserializer.Process(serialized));
REQUIRE_NOTHROW(deserializer.Process(serializer.GetBuffer(), serializer.GetBufferSize()));
REQUIRE(deserializer.GetType() == JFJochFrameDeserializer::Type::IMAGE);
auto image_array = deserializer.GetDataMessage();
REQUIRE(image_array.image.algorithm == CompressionAlgorithm::NO_COMPRESSION);
REQUIRE(image_array.image.xpixel == 1024);
@@ -288,10 +281,9 @@ TEST_CASE("CBORSerialize_Image_Compressed", "[CBOR]") {
};
REQUIRE_NOTHROW(serializer.SerializeImage(message));
auto serialized = serializer.GetBuffer();
JFJochFrameDeserializer deserializer;
REQUIRE_NOTHROW(deserializer.Process(serialized));
REQUIRE_NOTHROW(deserializer.Process(serializer.GetBuffer(), serializer.GetBufferSize()));
REQUIRE(deserializer.GetType() == JFJochFrameDeserializer::Type::IMAGE);
auto image_array = deserializer.GetDataMessage();
@@ -331,10 +323,8 @@ TEST_CASE("CBORSerialize_Image_Rad_Int_Profile", "[CBOR]") {
REQUIRE_NOTHROW(serializer.SerializeImage(message));
auto serialized = serializer.GetBuffer();
JFJochFrameDeserializer deserializer;
REQUIRE_NOTHROW(deserializer.Process(serialized));
REQUIRE_NOTHROW(deserializer.Process(serializer.GetBuffer(), serializer.GetBufferSize()));
REQUIRE(deserializer.GetType() == JFJochFrameDeserializer::Type::IMAGE);
auto image_array = deserializer.GetDataMessage();
@@ -374,10 +364,8 @@ TEST_CASE("CBORSerialize_Image_Spots", "[CBOR]") {
REQUIRE_NOTHROW(serializer.SerializeImage(message));
auto serialized = serializer.GetBuffer();
JFJochFrameDeserializer deserializer;
REQUIRE_NOTHROW(deserializer.Process(serialized));
REQUIRE_NOTHROW(deserializer.Process(serializer.GetBuffer(), serializer.GetBufferSize()));
REQUIRE(deserializer.GetType() == JFJochFrameDeserializer::Type::IMAGE);
auto image_array = deserializer.GetDataMessage();
@@ -452,10 +440,9 @@ TEST_CASE("CBORSerialize_Start_stream2", "[CBOR]") {
REQUIRE_NOTHROW(serializer.SerializeSequenceStart(message));
auto image = serializer.GetBuffer();
stream2_msg *msg;
auto ret = stream2_parse_msg(image.data(), image.size(), &msg);
auto ret = stream2_parse_msg(serializer.GetBuffer(), serializer.GetBufferSize(), &msg);
REQUIRE(ret == STREAM2_OK);
CHECK(msg->type == STREAM2_MSG_START);
auto msg2 = (stream2_start_msg *) msg;
@@ -499,10 +486,9 @@ TEST_CASE("CBORSerialize_End_stream2", "[CBOR]") {
REQUIRE_NOTHROW(serializer.SerializeSequenceEnd(message));
auto image = serializer.GetBuffer();
stream2_msg *msg;
auto ret = stream2_parse_msg(image.data(), image.size(), &msg);
auto ret = stream2_parse_msg(serializer.GetBuffer(), serializer.GetBufferSize(), &msg);
REQUIRE(ret == STREAM2_OK);
CHECK(msg->type == STREAM2_MSG_END);
auto msg2 = (stream2_end_msg *) msg;
@@ -546,11 +532,10 @@ TEST_CASE("CBORSerialize_Image_compressed_stream2", "[CBOR]") {
};
REQUIRE_NOTHROW(serializer.SerializeImage(message));
auto cbor_image = serializer.GetBuffer();
stream2_msg *msg;
auto ret = stream2_parse_msg(cbor_image.data(), cbor_image.size(), &msg);
auto ret = stream2_parse_msg(serializer.GetBuffer(), serializer.GetBufferSize(), &msg);
REQUIRE(ret == STREAM2_OK);
CHECK(msg->type == STREAM2_MSG_IMAGE);
auto msg2 = (stream2_image_msg *) msg;
@@ -601,7 +586,7 @@ TEST_CASE("CBORSerialize_Image_uncompressed_stream2", "[CBOR]") {
stream2_msg *msg;
auto ret = stream2_parse_msg(cbor_image.data(), cbor_image.size(), &msg);
auto ret = stream2_parse_msg(serializer.GetBuffer(), serializer.GetBufferSize(), &msg);
REQUIRE(ret == STREAM2_OK);
CHECK(msg->type == STREAM2_MSG_IMAGE);
auto msg2 = (stream2_image_msg *) msg;
tests/ZMQImagePusherTest.cpp
+3 -3
View File
@@ -97,7 +97,7 @@ TEST_CASE("ZMQImageCommTest_1Writer","[ZeroMQ]") {
data_message.number = i;
PrepareCBORImage(data_message, x, image1.data() + i * x.GetPixelsNum(), x.GetPixelsNum() * sizeof(uint16_t));
serializer.SerializeImage(data_message);
pusher.SendData(serializer.GetBuffer(), i);
pusher.SendImage(serializer.GetBuffer(), serializer.GetBufferSize(), i);
}
pusher.EndDataCollection(end_message);
@@ -172,7 +172,7 @@ TEST_CASE("ZMQImageCommTest_2Writers","[ZeroMQ]") {
data_message.number = i;
PrepareCBORImage(data_message, x, image1.data() + i * x.GetPixelsNum(), x.GetPixelsNum() * sizeof(uint16_t));
serializer.SerializeImage(data_message);
pusher.SendData(serializer.GetBuffer(), i);
pusher.SendImage(serializer.GetBuffer(), serializer.GetBufferSize(), i);
}
pusher.EndDataCollection(end_message);
@@ -260,7 +260,7 @@ TEST_CASE("ZMQImageCommTest_4Writers","[ZeroMQ]") {
data_message.number = i;
PrepareCBORImage(data_message, x, image1.data() + i * x.GetPixelsNum(), x.GetPixelsNum() * sizeof(uint16_t));
serializer.SerializeImage(data_message);
pusher.SendData(serializer.GetBuffer(), i);
pusher.SendImage(serializer.GetBuffer(), serializer.GetBufferSize(), i);
}
pusher.EndDataCollection(end_message);
writer/jfjoch_writer_test.cpp
+1 -1
View File
@@ -102,7 +102,7 @@ int main(int argc, char **argv) {
data_message.number = i;
PrepareCBORImage(data_message, x, output[i % nimages_in_file].data(), output_size[i % nimages_in_file]);
serializer.SerializeImage(data_message);
pusher.SendData(serializer.GetBuffer(), i);
pusher.SendImage(serializer.GetBuffer(), serializer.GetBufferSize(), i);
}
EndMessage end_message{};