Commit 7cb8fb60 authored by Stezowski Olivier's avatar Stezowski Olivier
Browse files

DEmulator operationnal and First version of the BasicReplicator

git-svn-id: svn+ssh://anonsvn.in2p3.fr/agata/gammaware/trunk@1739 c3abf467-7ff2-0310-b516-c5fb849bdde0
parent e6c950ef
......@@ -8,6 +8,7 @@
#pragma link C++ namespace ;
#pragma link C++ class BaseDanteWatcher;
#pragma link C++ class DisplayDante;
#pragma link C++ class MyDanteTree;
#pragma link C++ class DoDanteDoppler;
#pragma link C++ nestedclass;
......
......@@ -39,6 +39,7 @@ BasicAFC::BasicAFC() :
fPath("./"),
fBaseForName("AFC_"),
fCurrentFile(0x0),
fCurrentFileNumber(0u),
fMaxSize(kMaxInt_t),
fVertex(0x0)
{
......@@ -71,28 +72,6 @@ void BasicAFC::process_config (const Char_t *directory_path, UInt_t *error_code)
}
}
void BasicAFC::SyncAgent()
{
Int_t run, subrun; // std::string expname;
// agent to be synchronized
ConfAgent *agent = GetFrameIO().GetConfAgent();
// ask Narval for different parameters
if ( NarvalInterface::GetGlobalParameter("run_number",run) != 0 )
run = agent->GetRunAgent()->GetRun(); // something goes wrong, does not change the current run number
// GetGlobalParameter("",expname); ??
// a new run is open, go back to zero for subrun otherwise keep the current value
if ( run != agent->GetRunAgent()->GetRun() )
subrun = 0;
else
subrun = agent->GetRunAgent()->GetSubRun();
// now change the status of the conf agent
agent->GetRunAgent()->SetRun(run, subrun);
}
Bool_t BasicAFC::NewFile()
{
Log.SetProcessMethod("NewFile");
......@@ -102,9 +81,6 @@ Bool_t BasicAFC::NewFile()
// close if already open
if ( fCurrentFile )
::fclose(fCurrentFile);
// ask NarvalInterface for exp name, run number etc ...
SyncAgent();
// open the file
std::string tmp = fPath;
......@@ -113,7 +89,7 @@ Bool_t BasicAFC::NewFile()
std::ostringstream filename;
filename << tmp << fBaseForName
<< std::setfill('0') << std::setw(4) << GetFrameIO().GetConfAgent()->GetRunAgent()->GetSubRun() << std::setfill(' ')
<< std::setfill('0') << std::setw(4) << fCurrentFileNumber++ << std::setfill(' ')
<< ".adf";
fCurrentFile = ::fopen(filename.str().data(),"wb");
......@@ -128,28 +104,6 @@ Bool_t BasicAFC::NewFile()
// add conf frame at the beginning of the file and a vertex frame.
GetFrameIO().RecordGlobalConfiguration();
if ( fVertex ) {
// interface to the vertex frame
VertexInterface *v = GetDataPointer<VertexInterface>(fVertex->GetOutputFrame());
if ( v ) {
// global doppler correction is in RunAgent
RunAgent *ragent = GetFrameIO().GetConfAgent()->GetRunAgent();
// copy from agent to frame then dump the frame on output
v->SetPosition(ragent->GetX(),ragent->GetY(),ragent->GetZ());
v->SetDirection(ragent->GetDX(),ragent->GetDY(),ragent->GetDZ());
v->SetBeta(ragent->GetBeta());
fVertex->GetOutputFrame()->Write();
// set to true
fVertex->Fired(true);
// and record
GetFrameIO().Record();
}
}
GetFrameIO().GetConfAgent()->GetRunAgent()->SetSubRun(GetFrameIO().GetConfAgent()->GetRunAgent()->GetSubRun()+1);
ok = true;
}
......@@ -217,7 +171,8 @@ void BasicAFC::process_initialise (UInt_t *error_code)
if ( filein.good() ) {
fPath = pathforfiles;
fBaseForName = basename;
GetFrameIO().GetConfAgent()->GetRunAgent()->SetSubRun(starting_number);
fCurrentFileNumber = starting_number;
// GetFrameIO().GetConfAgent()->GetRunAgent()->SetSubRun(starting_number);
}
filein.close();
}
......
......@@ -48,6 +48,8 @@ private:
std::string fBaseForName;
//! current file
FILE *fCurrentFile;
//! current file number
UInt_t fCurrentFileNumber;
//! max size for a file (in bytes)
UInt_t fMaxSize;
......@@ -57,9 +59,7 @@ private:
protected:
//! open a new file
Bool_t NewFile();
//! synchronize the current agent asking Narval (through NarvalInterface)
void SyncAgent();
public:
BasicAFC();
virtual ~BasicAFC();
......
......@@ -213,19 +213,26 @@ Bool_t BufferIO::Reserve(UInt_t add)
Bool_t BufferIO::Link(const Char_t *extbuf, UInt_t size_ext_buf)
{
Bool_t ok = true;
if ( extbuf == NULL )
return false;
// new buffer attached
fOuter = const_cast<Char_t *>(extbuf); fRealOuterSize = size_ext_buf;
// the new attached buffer is ready to be processed
Bool_t ok = true;
if ( fOuter ) {
SetCurrent('o');
fCurrent = 0u; fStatus = BaseBuffer::kGood;
}
else { ok = false; }
if ( extbuf == fInner ) { // try to link again the inner
if ( size_ext_buf < fRealInnerSize ) {
SetCurrent('i');
fCurrent = 0u; fSize = size_ext_buf; fStatus = BaseBuffer::kGood; // the new attached buffer is ready to be processed
}
else { ok = false; }
}
else {
fOuter = const_cast<Char_t *>(extbuf); fRealOuterSize = size_ext_buf;
if ( fOuter ) {
SetCurrent('o');
fCurrent = 0u; fSize = size_ext_buf; fStatus = BaseBuffer::kGood; // the new attached buffer is ready to be processed
}
else { ok = false; }
}// new buffer attached
return ok;
}
......
......@@ -152,8 +152,9 @@ Bool_t ExpandableBlock::Expand(UInt_t size)
FileBlock::FileBlock(UInt_t size, ConfAgent::EMode mode):
OneBlock(size,mode),
fCurrentSize(0u),
fCurrentSize(0L),
fMaxSize(kMaxUInt_t),
fFileSnapshot(0L),
fFile(0x0)
{
}
......
......@@ -138,11 +138,14 @@ public:
{ return false; }
//! size of the current block
virtual UInt_t GetSize(UInt_t = 0u) const
{ return 0u; }
virtual Long64_t GetSize(UInt_t = 0u) const
{ return 0; }
//! max size of the block
virtual UInt_t GetMaxSize(UInt_t = 0u) const
{ return 0u; }
virtual Long64_t GetMaxSize(UInt_t = 0u) const
{ return 0; }
//! Current position
virtual Long64_t GetOffset(UInt_t = 0u) const
{ return 0; }
//! Add a Frame to the block.
/*!
......@@ -228,11 +231,24 @@ public:
virtual ~OneBlock()
{;}
virtual UInt_t GetSize(UInt_t = 0u) const
{ return fBuffer.Offset(); }
virtual UInt_t GetMaxSize(UInt_t = 0u) const
{ return fBuffer.Size(); }
virtual Long64_t GetSize(UInt_t = 0u) const
{
if (GetModeIO()==ConfAgent::kRead) {
return fBuffer.Size();
}
else return fBuffer.Offset();
}
virtual Long64_t GetMaxSize(UInt_t = 0u) const
{
return fBuffer.Size();
}
//! Current position
virtual Long64_t GetOffset(UInt_t = 0u) const
{
return fBuffer.Offset();
}
virtual BufferIO *GetBufferIO(UInt_t /*which*/ = 0u)
{ return &fBuffer; }
......@@ -351,9 +367,11 @@ public:
class FileBlock : public OneBlock
{
private:
UInt_t fCurrentSize;
UInt_t fMaxSize;
Long64_t fCurrentSize;
Long64_t fMaxSize;
Long64_t fFileSnapshot;
private:
FILE *fFile;
......@@ -365,16 +383,38 @@ public:
virtual ~FileBlock()
{;}
virtual UInt_t GetSize(UInt_t = 0u) const
{ return fCurrentSize; }
virtual UInt_t GetMaxSize(UInt_t = 0u) const
{ return fMaxSize; }
virtual Long64_t GetSize(UInt_t = 0u) const
{
if (GetModeIO()==ConfAgent::kRead) {
return fMaxSize;
}
else return fCurrentSize;
}
virtual Long64_t GetMaxSize(UInt_t = 0u) const
{
return fMaxSize;
}
//! Current position
virtual Long64_t GetOffset(UInt_t = 0u) const
{
return fCurrentSize;
}
virtual void SetFile(FILE *file, UInt_t max_size = kMaxUInt_t)
virtual void SetFile(FILE *file, Long64_t max_size = kMaxUInt_t)
{
Reset();
fMaxSize = max_size;
if (GetModeIO()==ConfAgent::kRead) { // reading mode, take the value from the file
// keep current position
Long64_t c = ::ftell(file);
// now get the end of file
::fseek(file, 0L, SEEK_END);
fMaxSize = ::ftell(file);
// back to current position
::fseek(file, c, SEEK_SET);
}
else fMaxSize = max_size;
fFile = file;
}
......@@ -388,13 +428,23 @@ public:
//! keep the position
virtual void Snapshot()
{ fLastSnapshot = fCurrentSize; }
{
fFileSnapshot = fCurrentSize;
}
//! back to the last position
virtual void BackToSnapshot()
{ if ( fFile ) ::fseek(fFile,fLastSnapshot,SEEK_SET); fCurrentSize = fLastSnapshot; }
{
if ( fFile )
::fseek(fFile,fFileSnapshot,SEEK_SET);
fCurrentSize = fFileSnapshot;
}
virtual void Rewind(ConfAgent::EMode = ConfAgent::kRead)
{ fCurrentSize = 0u; if ( fFile ) ::fseek(fFile,0u,SEEK_SET); }
{
fCurrentSize = 0u;
if ( fFile )
::fseek(fFile,0u,SEEK_SET);
}
virtual Bool_t SetBlock(Char_t */*buf*/= 0x0, UInt_t/*size*/= 0u, char /*opt_rewind*/ = 'r', char /*opt_link_or_copy*/ = 'l', UInt_t/*which*/ = 0u)
{ return false; }
......
......@@ -454,10 +454,19 @@ Bool_t FrameIO::Notify(Bool_t with_inner_trigger)
return false;
// if true a new frame is coming and fKeyIn already linked
Long64_t pos = fInFB->GetOffset();
while ( NextFrame() ) {
// a new frame has been read
fRead++;
if ( pos == fInFB->GetSize() ) {
Log << error << " A New Frame found but the position in the input block has not changed " << pos << " " << fInFB->GetSize() << dolog ;
break;
}
else {
// Log << info << " changed position " << pos << " " << fInFB->GetSize() << dolog ;
pos = fInFB->GetOffset();
}
Bool_t is_treated = false;
......
......@@ -502,8 +502,8 @@ UInt_t NarvalFilter::ProcessBlocks()
if ( fAttachedInputBlock->IsEoB() ) {
fAttachedInputBlock->SetOwner(0x0);
}
if ( fAttachedInputBlock->IsEoB() ) {
fAttachedInputBlock->SetOwner(0x0);
if ( fAttachedOutputBlock->IsEoB() ) {
fAttachedOutputBlock->SetOwner(0x0);
}
Log << dolog;
......
......@@ -115,21 +115,23 @@ public:
}
//! Attach a block as input for a given slot number
/*!
-1 means attach it on the top of the stack in case the actor can handle many input. As a consequence, GetNbInput is increased by 1.
if slot > 0 and < GetNbInput, the block cannot be attached.
slot < 0 means attach it on the top of the stack in case the actor can handle many input. As a consequence, GetNbInput is increased by 1.
The returned value is the slot position
if slot < 0 and >= GetNbInput, the block cannot be attached and it returns -1
*/
virtual Bool_t AttachInput(FrameBlock * /*block*/, Int_t /*slot*/ = -1)
virtual Int_t AttachInput(FrameBlock * /*block*/, Int_t /*slot*/ = -1)
{
return false;
return -1;
}
//! Attach a block as output for a given slot number
/*!
-1 means attach it on the top of the stack in case the actor can handle many output. As a consequence, GetNbOutput is increased by 1.
if slot > 0 and < GetNbOutput, the block cannot be attached.
slot < 0 means attach it on the top of the stack in case the actor can handle many output. As a consequence, GetNbOutput is increased by 1.
The returned value is the slot position
if slot < 0 and >= GetNbOutput, the block cannot be attached and it returns -1
*/
virtual Bool_t AttachOutput(FrameBlock * /*block*/, Int_t /*slot*/ = -1)
virtual Int_t AttachOutput(FrameBlock * /*block*/, Int_t /*slot*/ = -1)
{
return false;
return -1;
}
//! generic ProcessBlocks method for any kind of actor. Intput/Output should be attached using the AttachInput/AttachOutput methods
/*!
......@@ -237,15 +239,17 @@ public:
//! Attach a block as input for a given slot number
/*!
slot has to be < 1 to attach the block. With block = 0x0, it detaches the block
slot < 0 means attach it on the top of the stack in case the actor can handle many input. As a consequence, GetNbInput is increased by 1.
The returned value is the slot position
if slot < 0 and >= GetNbInput, the block cannot be attached and it returns -1
*/
virtual Bool_t AttachInput(FrameBlock *block, Int_t slot = -1)
virtual Int_t AttachInput(FrameBlock *block, Int_t slot = -1)
{
if ( slot < 1 ) {
fAttachedInputBlock = block;
return true;
return 0;
}
return true;
return -1;
}
//! for this actor, it gives the current number of entry lines attached.
virtual Int_t GetNbInput() const
......@@ -277,6 +281,7 @@ public:
//! Base class for an actor with N input and M output : main purpose is to structure the dataflow
/*!
\warning The max number of input/output should be defined in the constructor and not in the init method.
*/
class NarvalShunter : public NarvalInterface
{
......@@ -307,6 +312,16 @@ public:
{
return 0;
}
//! generic ProcessBlocks method for any kind of actor. Intput/Output should be attached using the AttachInput/AttachOutput methods
/*!
It allows moreover to interrupt and restart without waiting for the end of one block.
It should :
- Check the actor is the owner of the block through the SetOwner method
- Process the block(s) in case
- Free the block once fully consumed/produced using again the SetOwner ( 0x0 ) method
*/
virtual UInt_t ProcessBlocks() = 0 ;
};
//! It defines the interface needed to be a narval actor (Filter).
......@@ -343,15 +358,17 @@ public:
//! Attach a block as input for a given slot number
/*!
slot has to be < 1 to attach the block. With block = 0x0, it detaches the block
*/
virtual Bool_t AttachInput(FrameBlock *block, Int_t slot = -1)
slot < 0 means attach it on the top of the stack in case the actor can handle many input. As a consequence, GetNbInput is increased by 1.
The returned value is the slot position
if slot < 0 and >= GetNbInput, the block cannot be attached and it returns -1
*/
virtual Int_t AttachInput(FrameBlock *block, Int_t slot = -1)
{
if ( slot < 1 ) {
fAttachedInputBlock = block;
return true;
return 0;
}
return true;
return -1;
}
//! for this actor, it gives the current number of entry lines attached.
virtual Int_t GetNbInput() const
......@@ -361,17 +378,19 @@ public:
}
return 0;
}
//! Attach a block as input for a given slot number
//! Attach a block as output for a given slot number
/*!
slot has to be < 1 to attach the block. With block = 0x0, it detaches the block
*/
virtual Bool_t AttachOutput(FrameBlock *block, Int_t slot = -1)
slot < 0 means attach it on the top of the stack in case the actor can handle many output. As a consequence, GetNbOutput is increased by 1.
The returned value is the slot position
if slot < 0 and >= GetNbOutput, the block cannot be attached and it returns -1
*/
virtual Int_t AttachOutput(FrameBlock *block, Int_t slot = -1)
{
if ( slot < 1 ) {
fAttachedOutputBlock = block;
return true;
return 0;
}
return true;
return -1;
}
//! for this actor, it gives the current number of entry lines attached.
virtual Int_t GetNbOutput() const
......@@ -432,17 +451,19 @@ public:
virtual Int_t GetMaxOutput() const
{ return 1; }
//! Attach a block as input for a given slot number
//! Attach a block as output for a given slot number
/*!
slot has to be < 1 to attach the block. With block = 0x0, it detaches the block
*/
virtual Bool_t AttachOutput(FrameBlock *block, Int_t slot = -1)
slot < 0 means attach it on the top of the stack in case the actor can handle many output. As a consequence, GetNbOutput is increased by 1.
The returned value is the slot position
if slot < 0 and > GetNbOutput, the block cannot be attached and it returns -1
*/
virtual Int_t AttachOutput(FrameBlock *block, Int_t slot = -1)
{
if ( slot < 1 ) {
fAttachedOutputBlock = block;
return true;
return 0;
}
return true;
return -1;
}
//! for this actor, it gives the current number of entry lines attached.
virtual Int_t GetNbOutput() const
......
......@@ -47,6 +47,7 @@ void ADF::AConfigurator::Configure()
ADF::ALoadedActor::ALoadedActor(const Char_t *name) :
fLog("ALoadedActor"),
fName(name),
fGlobals(),
fGeneration(0x0),
......@@ -55,7 +56,8 @@ ADF::ALoadedActor::ALoadedActor(const Char_t *name) :
fOutputs(),
fDepth(-1),
fActor(0x0),
fActorType( kUndefined ),
fActorType( kUnknown ),
fActorStatus( kUndefined ),
fLastError(0u)
{
;
......@@ -215,32 +217,98 @@ void ADF::ALoadedActor::ConfigureActor()
void ADF::ALoadedActor::RegisterActor()
{
fLog.SetProcessMethod("RegisterActor()");
if ( fConfigurator == 0x0 )
throw std::bad_alloc();
fConfigurator->SetError(0u);
fActor = fConfigurator->Register();
fActor =
fConfigurator->Register();
if ( fActor == 0x0 )
if ( fActor == 0x0 || fConfigurator->GetError() != 0u ) {
SetLastError(fConfigurator->GetError());
//
throw std::bad_alloc();
}
if ( fActor->GetMaxInput() == 0 && fActor->GetMaxOutput() == 0 ) { // virtual or not well defined
//
fLog << error << "The registered actor " << GetName() << " cannot have any input and output " << dolog;
//
fActorType = kUnknown;
SetLastError(fConfigurator->GetError());
//
return;
}
SetLastError(0u);
if ( fActor->GetMaxInput() > 1 || fActor->GetMaxOutput() > 1 ) { // Shunter
fActorType = kShunter;
UInt_t wrong = 0u;
// input and output are attached only for generation > 2 for which the proper narval interface is defined and the Shunter
if ( fGeneration->Get() > 2 ) {
for (size_t i = 0; i < fInputs.size(); i++) {
if ( fActor->AttachInput(fInputs[i]->GetBlock('c') ) < 0 ) {// cannot be attached
wrong = 1;
break;
}
}
for (size_t i = 0; i < fOutputs.size(); i++) {
if ( fActor->AttachOutput(fOutputs[i]->GetBlock('p') ) < 0 ) {// cannot be attached
wrong = 2;
break;
}
}
}
else wrong = 1; // to low generation for that
//
if ( wrong == 0 )
fActorType = kShunter;
else {
SetLastError(1u);
}
}
else {
if ( fActor->GetMaxInput() == 0 ) { // producer
fActorType = kProducer;
if ( fInputs.size() != 0 || fOutputs.size() != 1 ) {
fLog << error << "The registered producer " << GetName() << " has not the right number of [input and/or output] [" << fInputs.size() << "," << fOutputs.size() << "]" << nline;
SetLastError(1u);
}
else {
if ( fGeneration->Get() > 2 ) {
fActor->AttachOutput(fOutputs[0]->GetBlock('p'));
}
fActorType = kProducer;
}
}
else {
if ( fActor->GetMaxOutput() == 0 ) { //consumer
fActorType = kConsumer;
if ( fInputs.size() != 1 || fOutputs.size() != 0 ) {
fLog << error << "The registered consumer " << GetName() << " has not the right number of [input and/or output] [" << fInputs.size() << "," << fOutputs.size() << "]" << nline;
SetLastError(1u);
}
else {
if ( fGeneration->Get() > 2 ) {
fActor->AttachInput(fInputs[0]->GetBlock('c'));
}
fActorType = kConsumer;
}
}
else {
fActorType = kFilter; // filter
if ( fInputs.size() != 1 || fOutputs.size() != 1 ) {
fLog << error << "The registered filter " << GetName() << " has not the right number of [input and/or output] [" << fInputs.size() << "," << fOutputs.size() << "]" << nline;
SetLastError(1u);
}
else {
if ( fGeneration->Get() > 2 ) {
fActor->AttachInput(fInputs[0]->GetBlock('c'));
fActor->AttachOutput(fOutputs[0]->GetBlock('p'));
}
fActorType = kFilter;
}
}
}
}
SetLastError(fConfigurator->GetError());
fLog << dolog;
}
void ADF::ALoadedActor::InitActor()
......@@ -253,24 +321,30 @@ void ADF::ALoadedActor::InitActor()
void ADF::ALoadedActor::ProcessActorGeneration0_1_2 (Short_t generation)