leonarski_fandClaude Opus 5.5
0ee8897294
The rugnux library plans what runs side by side against ProcessConfig::host_memory_budget_bytes, by default 75% of the machine's memory. In the broker most of the machine is the image buffer, which has to hold the whole collection for the live run, so that default would plan against memory that is already taken. The receiver now sets the budget to the default less the image buffer and a 16 GiB margin for the rest of the broker (never 0, which would mean the default; at worst the run schedules in series), decided once and the same for the main run and the 60-degree probe, which share the process-wide HostWorkerPool. ImageBuffer gets a size getter. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
96 lines
3.0 KiB
C++
96 lines
3.0 KiB
C++
// SPDX-FileCopyrightText: 2024 Filip Leonarski, Paul Scherrer Institute <filip.leonarski@psi.ch>
|
|
// SPDX-License-Identifier: GPL-3.0-only
|
|
|
|
#pragma once
|
|
|
|
#include "ThreadSafeFIFO.h"
|
|
#include "DiffractionExperiment.h"
|
|
|
|
class ZeroCopyReturnValue;
|
|
|
|
enum class ImageBufferEntryStatus {Empty, InPreparation, Sending, Idle};
|
|
|
|
struct ImageBufferStatus {
|
|
std::vector<int64_t> images_in_the_buffer;
|
|
int64_t min_image_number;
|
|
int64_t max_image_number;
|
|
|
|
int64_t total_slots;
|
|
int64_t available_slots;
|
|
int64_t preparation_slots;
|
|
int64_t sending_slots;
|
|
std::optional<int64_t> current_counter;
|
|
};
|
|
|
|
struct ImageBufferEntry {
|
|
int64_t image_number = -1;
|
|
size_t image_size = 0;
|
|
int64_t readers = 0;
|
|
ImageBufferEntryStatus status = ImageBufferEntryStatus::Empty;
|
|
bool indexed = false;
|
|
};
|
|
|
|
class ImageBuffer {
|
|
mutable std::mutex m;
|
|
std::condition_variable cv_preview_done;
|
|
std::condition_variable cv_preparation_done;
|
|
std::condition_variable cv_send_done;
|
|
uint8_t *buffer;
|
|
const size_t buffer_size;
|
|
|
|
size_t slot_size = 0;
|
|
int64_t slot_count = 0;
|
|
|
|
std::queue<uint32_t> free_slots;
|
|
std::vector<ImageBufferEntry> v;
|
|
std::vector<ZeroCopyReturnValue> send_buffer_zero_copy_ret_val;
|
|
int64_t slots_preparation = 0;
|
|
int64_t slots_sending = 0;
|
|
int64_t slots_in_preview = 0;
|
|
int64_t slots_idle = 0;
|
|
|
|
bool disable_requests = true; // All requests fail in this state
|
|
int64_t counter = 0; // Counter is increased when something changes in the image buffer
|
|
|
|
std::mutex start_message_mutex;
|
|
std::vector<uint8_t> start_message;
|
|
|
|
// Assumes existing lock
|
|
bool FinalizeInternal(std::unique_lock<std::mutex> &ul,
|
|
std::chrono::microseconds timeout = std::chrono::milliseconds(2500));
|
|
|
|
std::optional<uint32_t> getHandle(int64_t image_number) const;
|
|
std::optional<uint32_t> getHandleMaxImage() const;
|
|
std::optional<uint32_t> getHandleMaxIndexedImage() const;
|
|
|
|
uint8_t *GetBufferLocation(size_t id); // image metadata + image
|
|
public:
|
|
constexpr static int64_t MaxImage = -1;
|
|
constexpr static int64_t MaxIndexedImage = -2;
|
|
|
|
explicit ImageBuffer(size_t buffer_size_bytes);
|
|
~ImageBuffer();
|
|
ImageBuffer(const ImageBuffer &) = delete;
|
|
ImageBuffer &operator=(const ImageBuffer &) = delete;
|
|
|
|
void StartMeasurement(size_t buffer_location_size);
|
|
void StartMeasurement(const DiffractionExperiment& experiment);
|
|
|
|
ZeroCopyReturnValue *GetImageSlot();
|
|
void ReadyToSend(uint32_t location, int64_t image_number, size_t image_size, bool indexed);
|
|
void ReleaseSlot(uint32_t location);
|
|
|
|
int64_t GetAvailSlots() const;
|
|
size_t GetBufferSize() const;
|
|
|
|
bool CheckIfBufferReturned(std::chrono::microseconds timeout);
|
|
bool Finalize(std::chrono::microseconds timeout);
|
|
|
|
bool GetImage(std::vector<uint8_t> &out_v, int64_t image_number = MaxImage);
|
|
|
|
void SaveStartMessage(const std::vector<uint8_t> &msg);
|
|
void GetStartMessage(std::vector<uint8_t> &out_v);
|
|
|
|
ImageBufferStatus GetStatus() const;
|
|
};
|