/* Current API by design, incorrect as code but describes all the ideas */
#include "ADIOS.h"
namespace adios
{
typedef enum {
VARYING_DIMENSION = -1,
LOCAL_VALUE = 0,
GLOBAL_VALUE = 1
};
typedef enum { ERROR = 0, WARN = 1, INFO = 2, DEBUG = 3 } VerboseFlag;
typedef enum {
COLLECTIVE_WRITERS = 0, INDEPENDENT_WRITERS = 1,
GLOBAL_READERS = 2, ROUNDROBIN_READERS = 3, FIFO_READERS = 4,
OPEN_ALL_STEPS = 5
} OpenMode;
typedef enum { NOWAITFORSTREAM = 0, WAITFORSTREAM = 1 } StreamOpenMode; // default: wait for stream
typedef enum {
APPEND = 0, UPDATE = 1, // writer advance modes
NEXT_AVAILABLE = 2, LATEST_AVAILABLE = 3, // reader advance modes
} AdvanceMode;
typedef enum { NONBLOCKINGREAD = 0, BLOCKINGREAD = 1 } ReadMode;
} // namespace adios
void dummy()
{
//Application variables
std::vector<double> Temperature;
std::vector<float> RaggedArray;
unsigned int Nx;
int Nparts, nproc;
// Global class/object that serves for init, finalize and provides ADIOS functions
adios::ADIOS adios( std::string configfile, MPI_Comm comm, adios::verboseflag flag, bool debugflag );
/*************
* WRITE API
*************/
/* Method
* We associate Engines and Transports and user settings into an object called Method.
* ADIOS check if it is defined by the user in the config file, and fills it out if it is.
*/
std::shared_ptr<adios::Method> method = adios.DeclareMethod( "MethodName" );
if( ! method.isDefinedInConfig() )
{
// if not defined by user, we can change the default settings
method.SetEngine( "BP" ); // BP is the default engine
method.AddTransport( "File", "lucky=yes" ); // ISO-POSIX file is the default transport
method.AddTransport( "Staging" ); //"The" staging method developed in ECP
method.SetParameters("have_metadata_file","yes" ); // Passing parameters to the engine
method.SetParameters( "Aggregation", (nproc+1)/2 ); // number of aggregators
method.SetParameters( "verbose", adios::WARN ); // Verbosity level for this engine and what it calls
}
//Define variables with transformations.
adios::Variable<unsigned int>& varNX = adios.DefineVariable<unsigned int>( "NX" ); // global single-value across processes
adios::Variable<int>& varNproc = adios.DefineVariable<int>( "nproc", adios::GLOBAL_VALUE ); // same def for global value
adios::Variable<int>& varNparts = adios.DefineVariable<int>( "Nparts", adios::LOCAL_VALUE ); // a single-value different on every process
adios::Variable<double>& var1D = adios.DefineVariable<double>( "Temperature", {nproc*Nx} ); // 1D global array
adios::Variable<float>& varRagged = adios.DefineVariable<float>( "Ragged", {nproc,adios::VARYING_DIMENSION} ); // ragged array
// Define a variable with local dimensions now, and make ADIOS allocate it inside its buffers (zero-copy API)
adios::Variable<double>& varZeroCopy = adios.DefineVariable<double>( "ZC", {nproc,Nx}, {1,NX}, {rank,0} ); // 2D global array, 1D decomposition
double *myVarZC = adios.AllocateVar( varZeroCopy );
//add transform to variable
adios::Transform bzip2 = adios::transform::BZIP2( );
var1D->AddTransform( bzip2, 1 );
// open...write.write.write...advance...write.write.write...advance... ...close cycle
// "w" create/overwrite on open, "a" append at open, "u" open for update (does not increase step), "r" open for read.
std::shared_ptr<adios::Engine> writer = adios.Open( "myNumbers.bp", "w", method, adios::INDEPENDENT_WRITERS );
if( writer == nullptr )
throw std::ios_base::failure( "ERROR: failed to open ADIOS writer\n" );
for (int step = 0; step < 10; ++step) {
// write scalar value
writer->Write<int>( varNparts, Nparts );
// Make a selection to describe the local dimensions of the variable we write and
// its offsets in the global spaces
adios::Selection& sel = adios.SelectionBoundingBox( {Nx}, {rank*Nx} ); // local dims and offsets; both as list
var1D.SetSelection( sel );
writer->Write<double>( var1D, Temperature.data() );
// Indicate we are done for this step.
// N-to-M Aggregation, disk I/O will be performed during this call, unless
// time aggregation postpones all of that to some later step.
// When Advance() returns, user can overwrite its Zero Copy variables.
// Internal buffer is freed only if there are no Zero Copy variables and there is no time aggregation going on
writer->Advance(); // same as AppendMode
writer->Advance( adios::APPEND ); // append new step at next write
writer->Advance( adios::UPDATE ); // do not increase step; ? should this cause error in staging ?
// When AdvanceAsync returns, user need to wait for notification that he can overwrite the Zero Copy variables.
writer->AdvanceAsync( callback_func_to_notify_me() );
}
// Called once: indicate that we are done with this output for the run
// Zero Copy variables will be deallocated
writer->Close();
/*************
* READ API
*************/
// 1. Open a stream, where every reader can see everything in a stream (so that they can read a global array)
// Blocking read of a variable
try
{
// Open a stream
std::shared_ptr<adios::Engine> reader =
adios.Open( "filename.bp", "r", method,
adios::GLOBAL_READERS, // Each reader process sees everything from the stream
adios::WAITFORSTREAM, // wait for the first step appear (default)
timeout ); // wait this long for the stream, return error afterwards
/* Variable names are available as a vector of strings */
std::cout << "List of variables in file: " << reader->VariableNames << "\n";
/* read a Global scalar which has a single value in a step */
reader->Read<unsigned int>( "NX", Nx );
// inquiry about a variable, whose name we know
adios::Variable var1D = reader.InquiryVariable( "Temperature" );
vector<uint64_t> gdims = var1D->GetGlobalDimensions();
int step = varID->GetStep();
struct adios::BlockInfo blocks = reader.InquiryVariableBlockInfo( reader, var1D ); // get per-writer size info
// this is adios1 ADIOS_VARBLOCK
struct adios::Statistics stats = reader.InquiryVariableStat( reader, var1D, perstepstat, perblockstat ); // get min/max statistics
// this is adios1 ADIOS_VARSTAT
while( true )
{
// Make a selection to describe the local dimensions of the variable we READ and
// its offsets in the global spaces
adios::Selection bbsel = adios.SelectionBoundingBox( {ldim}, {offs} ); // local dims and offsets; both as list
var1D->SetSelection( bbsel );
reader->Read<double>( var1D, Temperature.data() );
// Better for staging to schedule several reads at once
reader->ScheduleRead<double>( var1D, Temperature.data() );
reader->PerformRead( adios::BLOCKINGREAD );
// promise to not read more from this step/item
reader->Release();
// want to move on to the next available step/item
reader->Advance(adios::NEXT_AVAILABLE); // default
reader->Advance(adios::LATEST_AVAILABLE); // interested only in the latest data
}
// Close file/stream
reader->Close();
}
catch( adios::end_of_stream& e )
{
// Reached end of stream, end processing loop
// Close file/stream
bpReader->Close();
}
catch( adios::file_not_found& e )
{
// File/stream does not exist, quit
}
// 2. Open a stream, where each item from the writers will get to a single reader only
// If the writers are collective, that means a whole steps go to different readers
// If the writers are independent, that means each writer's output goes to different readers
// Also show here ScheduleRead/PerformRead
//try
{
// Open a stream
std::shared_ptr<adios::Engine> reader =
adios.Open( "filename.bp", "r", method,
adios::FIFO_READERS, // Each reader process sees everything from the stream
adios::WAITFORSTREAM, // wait for the first step appear (default)
timeout ); // wait this long for the stream, return error afterwards
while( true )
{
// Make a selection to describe the local dimensions of the variable we READ and
// its offsets in the global spaces if we know this somehow
adios::Selection bbsel = adios.SelectionBoundingBox( {ldim}, {offs} ); // local dims and offsets; both as list
var1D->SetSelection( bbsel );
reader->Read<double>( var1D, Temperature.data() );
// Let ADIOS allocate space for the incoming (per-writer) item
double * data = reader->Read<double>( var1D );
// promise to not read more from this step/item
reader->Release();
// want to move on to the next available step/item
reader->Advance(); // default
reader->Advance(adios::LATEST_AVAILABLE); // This should be an error, or could it make sense?
}
reader->Close();
}
// 3. Open a stream and return immediately, not waiting for data to appear
// In this mode we cannot inquiry variables, but can schedule reads
//try
{
// Open a stream
std::shared_ptr<adios::Engine> reader =
adios.Open( "filename.bp", "r", method,
adios::GLOBAL_READERS, // Each reader process sees everything from the stream
adios::NOWAITFORSTREAM // wait for the first step appear (default)
);
while( true )
{
// Let ADIOS allocate space for the incoming (per-writer) item
reader->ScheduleRead<void>(); // read whatever comes
// One way is to handle the incoming data through a callback (which will be called in a thread)
// void cb( const void *data, std::string doid, std::string var, std::string dtype, std::vector<std::size_t> varshape );
// void cb( adios::VARCHUNK * chunk ); // see adios1 for VARCHUNK
reader->SetReadCallback( cb );
reader->PerformRead( adios::NONBLOCKINGREAD );
// Another way is checking back manually like in adios1 and processing chunks
reader->PerformRead( adios::NONBLOCKINGREAD );
int ck;
adios::VARCHUNK * chunk;
try
{
while ( (ck = reader->CheckReads( &chunk )) > 0) {
if (chunk) {
// process the chunk first
// ...
// free memory of chunk (not the data!)
adios::FreeChunk(chunk);
} else {
// no chunk was returned, slow down a little
sleep(1);
}
}
}
catch( std::exception& e )
{
// some error happened while getting a chunk
}
reader->Release();
reader->Advance();
}
reader->Close();
}
// 4. Open it as file and see all steps at once.
// Allow for reading multiple steps of a variable into a contiguous array
try
{
// Open a stream
std::shared_ptr<adios::Engine> reader =
adios.Open( "filename.bp", "r", method,
adios::OPEN_ALL_STEPS, //File mode, no advance
);
/* NX */
/* There is a single value for each step. We can read all into a 1D array with a step selection.
* Steps are not automatically presented as an array dimension and read does not read it as array.
*/
// We can also just conveniently get the first step with a simple read statement.
reader->Read<unsigned int>( "NX", &Nx ); // read a Global scalar which has a single value in a step
adios::Variable<void> varNx = bpReader.InquiryVariable("Nx");
std::vector<int> Nxs( varNx->GetSteps() ); // number of steps available
// make a StepSelection to select multiple steps. Args: From, #of consecutive steps
adios::StepSelection stepsNx( 0, varNx->GetSteps() );
// ? How do we make a selection for an arbitrary list of steps ?
varNX.SetStepSelection( stepsNx );
reader->Read<unsigned int>( varNx, Nxs.data() );
auto itmax = std::max_element(std::begin(Nxs), std::end(Nxs));
auto itmin = std::min_element(std::begin(Nxs), std::end(Nxs));
if (*itmin != *itmax)
{
throw std::ios_base::failure( "ERROR: NX is not the same at all steps!\n" );
}
/* Nparts */
// Nparts local scalar is presented as a 1D array of nproc elements.
// We can read all steps into a 2D array of nproc * nsteps
adios::Variable<void> varNparts = bpReader.InquiryVariable("Nparts");
std::vector<int> partsV( Nproc * varNparts->GetSteps() );
varNparts->SetStepSelection(
adios.StepSelection( 0, varNparts->GetSteps() )
);
bpReader->Read<int>( varNparts, partsV.data() ); // missing spatial selection = whole array at each step
// Close file/stream
reader->Close();
}
catch( adios::end_of_stream& e )
{
// Reached end of stream, end processing loop
// Close file/stream
bpReader->Close();
}
catch( adios::file_not_found& e )
{
// File/stream does not exist, quit
}
}