Loading...
Searching...
No Matches
SiscoWriter.h
Go to the documentation of this file.
#ifndef SISCO_SISCO_WRITER_H_
#define SISCO_SISCO_WRITER_H_
#include <complex>
#include <fstream>
#include <list>
#include <map>
#include <set>
#include <span>
#include <string>
#include <thread>
#include <vector>
#include "ConditionalQueue.h"
#include "Deflate.h"
#include "Lane.h"
#include "Sisco.h"
namespace casacore::sisco {
class SiscoWriter {
public:
SiscoWriter(const std::string& filename, int predict_level,
int deflate_level);
SiscoWriter(SiscoWriter&&) = delete;
~SiscoWriter() {
if (file_.is_open()) Close();
}
SiscoWriter& operator=(SiscoWriter&&) = delete;
void Open(std::span<const std::byte> header_data);
void Write(size_t baseline_index, std::span<const std::complex<float>> data);
void Close();
private:
struct Chunk {
size_t index;
std::vector<std::byte> mantisas_data;
std::vector<std::byte> exponent_data;
size_t allocated_chunk_size = kDefaultChunkSize;
// Whenever a thread finishes preprocessing, it will update this value. If
// the value is high enough, the thread knows the chunk is ready for
// deflate.
size_t n_allocated_items = 0;
size_t n_finished_items = 0;
size_t item_counter = 0;
size_t chunk_size = 0;
};
struct PreprocessingTask {
std::vector<float> real_data;
std::vector<float> imaginary_data;
size_t baseline_index;
std::byte* mantisa_position;
std::byte* exponent_position;
Chunk* chunk;
};
struct WriteTask {
size_t index;
size_t uncompressed_size;
std::vector<std::byte> data;
};
struct BaselineData {
CompressorState real_state_;
CompressorState imaginary_state_;
size_t n_sequences_ = 0;
};
struct ThreadData {};
void NewChunk(std::unique_lock<std::mutex>& lock);
void RemoveChunk(const Chunk* chunk);
void PreprocessLoop();
void WriteLoop();
void Preprocess(SiscoWriter::PreprocessingTask& task,
std::unique_lock<std::mutex>& lock);
void WriteChunk(size_t uncompressed_size, std::span<const std::byte> data);
void DeflateChunk(Chunk& chunk);
// Chunk data
constexpr static size_t kDefaultChunkSize = 1024 * 1024;
size_t chunk_write_position_ = 0;
std::byte* mantisa_position_;
std::byte* exponent_position_;
size_t n_chunks_ = 0;
// A list is used because we do not want to invalidate pointers when
// adding/removing members. Access requires holding the mutex.
std::list<Chunk> chunks_;
Chunk* write_chunk_;
// Indexed by baseline_index.
std::map<size_t, BaselineData> baseline_data_;
std::ofstream file_;
std::string filename_;
int predict_level_ = 2;
int deflate_level_ = 9;
ConditionalQueue<PreprocessingTask> preprocessing_tasks_{4096};
aocommon::Lane<WriteTask> write_tasks_{10};
std::mutex mutex_;
std::condition_variable chunk_finished_condition_;
std::set<size_t> busy_baselines_;
std::vector<std::thread> compression_threads_;
std::thread write_thread_;
};
} // namespace casacore::sisco
#endif
ConditionalQueue.h
Deflate.h
Lane.h
Internal header file for the Lane.
Sisco.h
casacore::aocommon::Lane
The Lane is an efficient cyclic buffer that is synchronized.
Definition Lane.h:100
casacore::sisco::ConditionalQueue
A queue with a limited size and the ability to query only specific values.
Definition ConditionalQueue.h:20
casacore::sisco::SiscoWriter::Close
void Close()
casacore::sisco::SiscoWriter::write_chunk_
Chunk * write_chunk_
Definition SiscoWriter.h:91
casacore::sisco::SiscoWriter::exponent_position_
std::byte * exponent_position_
Definition SiscoWriter.h:86
casacore::sisco::SiscoWriter::RemoveChunk
void RemoveChunk(const Chunk *chunk)
casacore::sisco::SiscoWriter::write_tasks_
aocommon::Lane< WriteTask > write_tasks_
Definition SiscoWriter.h:100
casacore::sisco::SiscoWriter::busy_baselines_
std::set< size_t > busy_baselines_
Holds a list of baseline ids currently being preprocessed.
Definition SiscoWriter.h:106
casacore::sisco::SiscoWriter::deflate_level_
int deflate_level_
Definition SiscoWriter.h:98
casacore::sisco::SiscoWriter::kDefaultChunkSize
static constexpr size_t kDefaultChunkSize
Chunk data.
Definition SiscoWriter.h:83
casacore::sisco::SiscoWriter::Preprocess
void Preprocess(SiscoWriter::PreprocessingTask &task, std::unique_lock< std::mutex > &lock)
casacore::sisco::SiscoWriter::PreprocessLoop
void PreprocessLoop()
casacore::sisco::SiscoWriter::chunk_finished_condition_
std::condition_variable chunk_finished_condition_
Definition SiscoWriter.h:103
casacore::sisco::SiscoWriter::mantisa_position_
std::byte * mantisa_position_
Definition SiscoWriter.h:85
casacore::sisco::SiscoWriter::n_chunks_
size_t n_chunks_
Definition SiscoWriter.h:87
casacore::sisco::SiscoWriter::SiscoWriter
SiscoWriter(SiscoWriter &&)=delete
casacore::sisco::SiscoWriter::preprocessing_tasks_
ConditionalQueue< PreprocessingTask > preprocessing_tasks_
Definition SiscoWriter.h:99
casacore::sisco::SiscoWriter::chunks_
std::list< Chunk > chunks_
A list is used because we do not want to invalidate pointers when adding/removing members.
Definition SiscoWriter.h:90
casacore::sisco::SiscoWriter::compression_threads_
std::vector< std::thread > compression_threads_
Definition SiscoWriter.h:107
casacore::sisco::SiscoWriter::Write
void Write(size_t baseline_index, std::span< const std::complex< float > > data)
casacore::sisco::SiscoWriter::filename_
std::string filename_
Definition SiscoWriter.h:96
casacore::sisco::SiscoWriter::file_
std::ofstream file_
Definition SiscoWriter.h:95
casacore::sisco::SiscoWriter::operator=
SiscoWriter & operator=(SiscoWriter &&)=delete
casacore::sisco::SiscoWriter::SiscoWriter
SiscoWriter(const std::string &filename, int predict_level, int deflate_level)
casacore::sisco::SiscoWriter::~SiscoWriter
~SiscoWriter()
Definition SiscoWriter.h:30
casacore::sisco::SiscoWriter::baseline_data_
std::map< size_t, BaselineData > baseline_data_
Indexed by baseline_index.
Definition SiscoWriter.h:94
casacore::sisco::SiscoWriter::NewChunk
void NewChunk(std::unique_lock< std::mutex > &lock)
casacore::sisco::SiscoWriter::write_thread_
std::thread write_thread_
Definition SiscoWriter.h:108
casacore::sisco::SiscoWriter::predict_level_
int predict_level_
Definition SiscoWriter.h:97
casacore::sisco::SiscoWriter::DeflateChunk
void DeflateChunk(Chunk &chunk)
casacore::sisco::SiscoWriter::WriteChunk
void WriteChunk(size_t uncompressed_size, std::span< const std::byte > data)
casacore::sisco::SiscoWriter::Open
void Open(std::span< const std::byte > header_data)
casacore::sisco::SiscoWriter::WriteLoop
void WriteLoop()
casacore::sisco::SiscoWriter::mutex_
std::mutex mutex_
Definition SiscoWriter.h:102
casacore::sisco::SiscoWriter::chunk_write_position_
size_t chunk_write_position_
Definition SiscoWriter.h:84
casacore::sisco
Definition ConditionalQueue.h:8
casacore::sisco::CompressorState
Definition Sisco.h:12
casacore::sisco::SiscoWriter::BaselineData
Definition SiscoWriter.h:66
casacore::sisco::SiscoWriter::BaselineData::n_sequences_
size_t n_sequences_
Definition SiscoWriter.h:69
casacore::sisco::SiscoWriter::BaselineData::imaginary_state_
CompressorState imaginary_state_
Definition SiscoWriter.h:68
casacore::sisco::SiscoWriter::BaselineData::real_state_
CompressorState real_state_
Definition SiscoWriter.h:67
casacore::sisco::SiscoWriter::Chunk
Definition SiscoWriter.h:40
casacore::sisco::SiscoWriter::Chunk::item_counter
size_t item_counter
Definition SiscoWriter.h:50
casacore::sisco::SiscoWriter::Chunk::n_finished_items
size_t n_finished_items
Definition SiscoWriter.h:49
casacore::sisco::SiscoWriter::Chunk::allocated_chunk_size
size_t allocated_chunk_size
Definition SiscoWriter.h:44
casacore::sisco::SiscoWriter::Chunk::mantisas_data
std::vector< std::byte > mantisas_data
Definition SiscoWriter.h:42
casacore::sisco::SiscoWriter::Chunk::index
size_t index
Definition SiscoWriter.h:41
casacore::sisco::SiscoWriter::Chunk::chunk_size
size_t chunk_size
Definition SiscoWriter.h:51
casacore::sisco::SiscoWriter::Chunk::n_allocated_items
size_t n_allocated_items
Whenever a thread finishes preprocessing, it will update this value.
Definition SiscoWriter.h:48
casacore::sisco::SiscoWriter::Chunk::exponent_data
std::vector< std::byte > exponent_data
Definition SiscoWriter.h:43
casacore::sisco::SiscoWriter::PreprocessingTask
Definition SiscoWriter.h:53
casacore::sisco::SiscoWriter::PreprocessingTask::imaginary_data
std::vector< float > imaginary_data
Definition SiscoWriter.h:55
casacore::sisco::SiscoWriter::PreprocessingTask::mantisa_position
std::byte * mantisa_position
Definition SiscoWriter.h:57
casacore::sisco::SiscoWriter::PreprocessingTask::baseline_index
size_t baseline_index
Definition SiscoWriter.h:56
casacore::sisco::SiscoWriter::PreprocessingTask::real_data
std::vector< float > real_data
Definition SiscoWriter.h:54
casacore::sisco::SiscoWriter::PreprocessingTask::chunk
Chunk * chunk
Definition SiscoWriter.h:59
casacore::sisco::SiscoWriter::PreprocessingTask::exponent_position
std::byte * exponent_position
Definition SiscoWriter.h:58
casacore::sisco::SiscoWriter::ThreadData
Definition SiscoWriter.h:71
casacore::sisco::SiscoWriter::WriteTask
Definition SiscoWriter.h:61
casacore::sisco::SiscoWriter::WriteTask::uncompressed_size
size_t uncompressed_size
Definition SiscoWriter.h:63
casacore::sisco::SiscoWriter::WriteTask::index
size_t index
Definition SiscoWriter.h:62
casacore::sisco::SiscoWriter::WriteTask::data
std::vector< std::byte > data
Definition SiscoWriter.h:64