14#include "arcane/std/internal/BasicWriter.h"
16#include "arcane/utils/FatalErrorException.h"
17#include "arcane/utils/StringBuilder.h"
18#include "arcane/utils/PlatformUtils.h"
19#include "arcane/utils/JSONWriter.h"
20#include "arcane/utils/IDataCompressor.h"
21#include "arcane/utils/MemoryView.h"
22#include "arcane/utils/Ref.h"
23#include "arcane/utils/IHashAlgorithm.h"
25#include "arcane/core/IParallelMng.h"
26#include "arcane/core/ItemGroup.h"
27#include "arcane/core/IVariable.h"
28#include "arcane/core/IItemFamily.h"
29#include "arcane/core/IData.h"
30#include "arcane/core/internal/IVariableInternal.h"
32#include "arcane/std/internal/ParallelDataWriter.h"
46BasicWriter(IApplication* app, IParallelMng* pm,
const String& path,
47 eOpenMode open_mode,
Int32 version,
bool want_parallel)
49, m_want_parallel(want_parallel)
62 Int32 rank = m_parallel_mng->commRank();
63 if (m_open_mode == OpenModeTruncate && m_parallel_mng->isMasterIO())
64 platform::recursiveCreateDirectory(m_path);
65 m_parallel_mng->barrier();
66 String filename = _getBasicVariableFile(m_version, m_path, rank);
67 m_text_writer =
makeRef(
new KeyValueTextWriter(traceMng(), filename, m_version));
68 m_text_writer->setDataCompressor(m_data_compressor);
69 m_text_writer->setHashAlgorithm(m_hash_algorithm);
73 if (!m_data_compressor.get()) {
74 String data_compressor_name = platform::getEnvironmentVariable(
"ARCANE_DEFLATER");
75 if (!data_compressor_name.null()) {
76 data_compressor_name = data_compressor_name +
"DataCompressor";
77 auto bc = _createDeflater(m_application, data_compressor_name);
78 info() <<
"Use data_compressor from environment variable ARCANE_DEFLATER name=" << data_compressor_name;
79 m_data_compressor = bc;
80 m_text_writer->setDataCompressor(bc);
85 if (!m_hash_algorithm.get()) {
86 String hash_algorithm_name = platform::getEnvironmentVariable(
"ARCANE_HASHALGORITHM");
87 if (hash_algorithm_name.null())
88 hash_algorithm_name =
"SHA3_256";
90 info() <<
"Use hash algorithm from environment variable ARCANE_HASHALGORITHM name=" << hash_algorithm_name;
91 hash_algorithm_name = hash_algorithm_name +
"HashAlgorithm";
92 auto v = _createHashAlgorithm(m_application, hash_algorithm_name);
94 m_text_writer->setHashAlgorithm(v);
98 if (!m_compare_hash_algorithm.get()) {
99 String algo_name = platform::getEnvironmentVariable(
"ARCANE_COMPAREHASHALGORITHM");
100 if (!algo_name.empty()) {
101 info() <<
"Use global hash algorithm from environment variable ARCANE_COMPAREHASHALGORITHM name=" << algo_name;
102 algo_name = algo_name +
"HashAlgorithm";
103 auto v = _createHashAlgorithm(m_application, algo_name);
104 m_compare_hash_algorithm = v;
108 m_global_writer =
new BasicGenericWriter(m_application, m_version, m_text_writer);
109 if (m_verbose_level > 0)
110 info() <<
"** OPEN MODE = " << m_open_mode;
126Ref<ParallelDataWriter> BasicWriter::
127_getWriter(IVariable* var)
129 return m_parallel_data_writers.getOrCreateWriter(var->itemGroup());
136_directWriteVal(IVariable* var, IData* data)
138 info(4) <<
"DIRECT WRITE VAL v=" << var->fullName();
140 IData* write_data = data;
145 Ref<IData> allocated_write_data;
146 const bool is_mesh_variable = (var->itemKind() != IK_Unknown);
147 if (is_mesh_variable) {
148 ItemGroup group = var->itemGroup();
149 if (m_want_parallel) {
150 Ref<ParallelDataWriter> writer = _getWriter(var);
151 written_unique_ids = writer->sortedUniqueIds();
152 allocated_write_data = writer->getSortedValues(data);
153 write_data = allocated_write_data.get();
158 _fillUniqueIds(group, sequential_written_unique_ids);
159 written_unique_ids = sequential_written_unique_ids.view();
162 if (m_written_groups.find(group) == m_written_groups.end()) {
163 info(5) <<
"WRITE GROUP " << group.name();
164 const IItemFamily* item_family = group.itemFamily();
165 const String& gname = group.name();
166 String group_full_name = item_family->fullName() +
"_" + gname;
167 _fillUniqueIds(group, wanted_unique_ids);
168 if (m_is_save_values)
169 m_global_writer->writeItemGroup(group_full_name, written_unique_ids, wanted_unique_ids.view());
170 m_written_groups.insert(group);
174 Ref<ISerializedData> sdata(write_data->createSerializedDataRef(
false));
176 if (is_mesh_variable) {
177 compare_hash = _computeCompareHash(var, write_data);
179 m_global_writer->writeData(var->fullName(), sdata.get(), compare_hash, m_is_save_values);
211 info() <<
"** WARNING: partial variable not implemented in BasicWriter";
214 _directWriteVal(var, data);
221setMetaData(
const String& meta_data)
224 if (m_version >= 3) {
226 Int64 length = bytes.length();
227 String key_name =
"Global:CheckpointMetadata";
229 m_text_writer->write(key_name,
asBytes(bytes));
232 Int32 my_rank = m_parallel_mng->commRank();
233 String filename = _getMetaDataFileName(my_rank);
234 std::ofstream ofile(filename.
localstr(), ios::binary);
246 Int32 my_rank = m_parallel_mng->commRank();
247 m_global_writer->initialize(m_path, my_rank);
256 const Int64 nb_part = m_parallel_mng->commSize();
263 jsw.writeKey(_getArcaneDBTag());
266 jsw.write(
"Version", (
Int64)m_version);
267 jsw.write(
"NbPart", nb_part);
268 jsw.write(
"HasValues", m_is_save_values);
270 String data_compressor_name;
271 Int64 data_compressor_min_size = 0;
272 if (m_data_compressor.get()) {
273 data_compressor_name = m_data_compressor->name();
274 data_compressor_min_size = m_data_compressor->minCompressSize();
276 jsw.write(
"DataCompressor", data_compressor_name);
277 jsw.write(
"DataCompressorMinSize", String::fromNumber(data_compressor_min_size));
282 if (m_hash_algorithm.get())
283 name = m_hash_algorithm->name();
284 jsw.write(
"HashAlgorithm", name);
290 if (m_compare_hash_algorithm.get())
291 name = m_compare_hash_algorithm->name();
292 jsw.write(
"ComparisonHashAlgorithm", name);
297 StringBuilder filename = m_path;
298 filename +=
"/arcane_acr_db.json";
299 String fn = filename.toString();
300 std::ofstream ofile(fn.localstr());
301 ofile << jsw.getBuffer();
310 const IParallelMng* pm = m_parallel_mng;
311 if (pm->isMasterIO()) {
312 if (m_version >= 3) {
316 Int64 nb_part = pm->commSize();
317 StringBuilder filename = m_path;
318 filename +=
"/infos.txt";
319 String fn = filename.toString();
320 std::ofstream ofile(fn.localstr());
321 ofile << nb_part <<
'\n';
324 m_global_writer->endWrite();
#define ARCANE_FATAL(...)
Macro throwing a FatalErrorException.
Interface of a data item.
Interface of a hashing algorithm.
virtual String computeComparisonHashCollective(IHashAlgorithm *hash_algo, IData *sorted_data)=0
Calculates the comparison hash for the variable.
virtual bool isPartial() const =0
Indicates if the variable is partial.
virtual IVariableInternal * _internalApi()=0
Internal Arcane API.
View of an array of elements of type T.
Unicode character string.
void writeBytes(std::ostream &o) const
Writes the string in UTF-8 format to the stream o.
const char * localstr() const
Returns the conversion of the instance into UTF-8 encoding.
ByteConstArrayView utf8() const
Returns the conversion of the instance into UTF-8 encoding.
TraceMessage info() const
Flow for an information message.
UniqueArray< Int64 > Int64UniqueArray
Dynamic 1D array of 64-bit integers.
std::int64_t Int64
Signed integer type of 64 bits.
ConstArrayView< Int64 > Int64ConstArrayView
C equivalent of a 1D array of 64-bit integers.
Impl::SpanTypeFromSize< conststd::byte, SizeType >::SpanType asBytes(const SpanImpl< DataType, SizeType, Extent > &s)
Converts the view into an array of non-modifiable bytes.
auto makeRef(InstanceType *t) -> Ref< InstanceType >
Creates a reference on a pointer.
std::int32_t Int32
Signed integer type of 32 bits.