leonarski_fandClaude Opus 5.5
f215300a07
Images reach the buffer in order give or take a few, so one still missing when the buffer holds an image 100 past it is lost, not late: the read returns false instead of waiting for the end of the collection. The newest image number comes from ImageBuffer::GetStatus, which scans every slot under the buffer's lock, so it is asked only every 100 polls (~10 Hz). The test puts CBOR images into an ImageBuffer the way the receiver does and checks read-back of the pixels, waiting for an image not yet there, the lag limit, the end of the collection and the bounds. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
75 lines
3.2 KiB
C++
75 lines
3.2 KiB
C++
// SPDX-FileCopyrightText: 2026 Filip Leonarski, Paul Scherrer Institute <filip.leonarski@psi.ch>
|
|
// SPDX-License-Identifier: GPL-3.0-only
|
|
|
|
#include "JFJochImageBufferReader.h"
|
|
|
|
#include <cstring>
|
|
#include <thread>
|
|
|
|
#include "JFJochStartMessageReader.h"
|
|
#include "../common/JFJochException.h"
|
|
#include "../frame_serialize/CBORStream2Deserializer.h"
|
|
|
|
JFJochImageBufferReader::JFJochImageBufferReader(ImageBuffer &buffer) : buffer_(buffer) {}
|
|
|
|
void JFJochImageBufferReader::Open(const StartMessage &start) {
|
|
number_of_images_ = start.number_of_images;
|
|
ended_ = false;
|
|
SetStartMessage(DatasetFromStartMessage(start, default_experiment));
|
|
}
|
|
|
|
void JFJochImageBufferReader::Ended() {
|
|
ended_ = true;
|
|
}
|
|
|
|
uint64_t JFJochImageBufferReader::GetNumberOfImages() const {
|
|
return number_of_images_;
|
|
}
|
|
|
|
void JFJochImageBufferReader::Close() {
|
|
SetStartMessage({});
|
|
}
|
|
|
|
bool JFJochImageBufferReader::WaitForMessage(int64_t image_number, std::vector<uint8_t> &cbor) {
|
|
if (image_number < 0 || static_cast<uint64_t>(image_number) >= number_of_images_)
|
|
throw JFJochException(JFJochExceptionCategory::ArrayOutOfBounds, "Image out of bounds");
|
|
// The buffer has no notification of its own; at 100 Hz a millisecond's poll costs nothing.
|
|
for (int poll = 1; !buffer_.GetImage(cbor, image_number); poll++) {
|
|
if (ended_)
|
|
return buffer_.GetImage(cbor, image_number); // one that came in just before the end
|
|
// Images arrive in order, give or take a few: one still missing when the buffer already holds
|
|
// an image MAX_IMAGE_LAG past it is lost, not late. GetStatus scans every slot under the
|
|
// buffer's lock, so it is asked only every 100 polls.
|
|
if (poll % 100 == 0 && buffer_.GetStatus().max_image_number >= image_number + MAX_IMAGE_LAG)
|
|
return buffer_.GetImage(cbor, image_number);
|
|
std::this_thread::sleep_for(std::chrono::milliseconds(1));
|
|
}
|
|
return true;
|
|
}
|
|
|
|
bool JFJochImageBufferReader::ReadRawImage(int64_t image_number, JFJochReaderRawImage &ret) {
|
|
if (!WaitForMessage(image_number, ret.read_buffer))
|
|
return false;
|
|
const auto msg = CBORStream2Deserialize(ret.read_buffer);
|
|
if (msg->msg_type != CBORImageType::IMAGE || !msg->data_message)
|
|
return false;
|
|
// Copied out of the message: read_buffer is scratch a caller may hand to the next read.
|
|
const CompressedImage &image = msg->data_message->image;
|
|
ret.image_buffer.resize(image.GetCompressedSize());
|
|
memcpy(ret.image_buffer.data(), image.GetCompressed(), image.GetCompressedSize());
|
|
ret.image = CompressedImage(ret.image_buffer.data(), ret.image_buffer.size(), image.GetWidth(),
|
|
image.GetHeight(), image.GetMode(), image.GetCompressionAlgorithm());
|
|
return true;
|
|
}
|
|
|
|
bool JFJochImageBufferReader::LoadImage_i(std::shared_ptr<JFJochReaderDataset> &dataset, DataMessage &message,
|
|
std::vector<uint8_t> &buffer, int64_t image_number, bool) {
|
|
if (!dataset || !WaitForMessage(image_number, buffer))
|
|
return false;
|
|
const auto msg = CBORStream2Deserialize(buffer);
|
|
if (msg->msg_type != CBORImageType::IMAGE || !msg->data_message)
|
|
return false;
|
|
message = *msg->data_message;
|
|
return true;
|
|
}
|