faspplugin/src/FileInput.cxx (r4864/r3636)

Go to the documentation of this file.

#include "fasp/FileInput.h"

#include <string.h>
#include <stdlib.h>

#include "dabc/logging.h"
#include "dabc/Buffer.h"
#include "dabc/FileIO.h"
#include "dabc/Manager.h"
#include "dabc/Port.h"


#include "base/commons.h"
#include "mbs/MbsTypeDefs.h"
#include "mbs/Iterator.h"

fasp::FileInput::FileInput(const char* fname, uint32_t bufsize) :
   dabc::DataInput(),
   fFileName(fname ? fname : ""),
   fBufferSize(bufsize),
fFilesList(0),
fFile(0),
fCurrentFileName(),
fCurrentRead(0),
fRaw(0),
fRawSize(0),
fRawPos(0),
fRawSync(DummySync)
{
   DOUT2(("Create FASP input"));
}

fasp::FileInput::~FileInput()
{
   // FIXME: cleanup should be done much earlier
   CloseFile();
   if (fFilesList) {
      dabc::Object::Destroy(fFilesList);
      fFilesList = 0;
   }

   if (fRaw) { delete [] fRaw; fRaw = 0; }
}

bool fasp::FileInput::Read_Init(const dabc::WorkerRef& wrk, const dabc::Command& cmd)
{
   fFileName = wrk.Cfg(dabc::xmlFileName, cmd).AsStdStr(fFileName);
   fBufferSize = wrk.Cfg(dabc::xmlBufferSize, cmd).AsInt(fBufferSize);

   DOUT0(("BufferSize = %d Filename = %s", fBufferSize, fFileName.c_str()));

   fRawSize = fBufferSize / FaspBlockSize;
   if (fRawSize < 100) fRawSize = 100;
   fRawSize *= FaspBlockSize;

   return Init();
}

bool fasp::FileInput::Init()
{
   if (fFileName.length()==0) return false;

   if (fFilesList!=0) {
      EOUT(("Files list already exists"));
      return false;
   }

   if (fBufferSize==0) {
      EOUT(("Buffer size not specified !!!!"));
      return false;
   }

   if (strpbrk(fFileName.c_str(),"*?")!=0)
      fFilesList = dabc::mgr()->ListMatchFiles("", fFileName.c_str());
   else {
      fFilesList = new dabc::Object(0, "FilesList", true);
      new dabc::Object(fFilesList, fFileName.c_str());
   }

   return OpenNextFile();
}

bool fasp::FileInput::OpenNextFile()
{
   CloseFile();

   if ((fFilesList==0) || (fFilesList->NumChilds()==0)) return false;

   const char* nextfilename = fFilesList->GetChild(0)->GetName();

   fFile = fopen(nextfilename,"r");

   bool res = true;

   if (fFile==0) {
      EOUT(("Cannot open FASP file %s for reading", nextfilename));
      res = false;
   } else {
      fCurrentFileName = nextfilename;
      DOUT1(("Open FASP file %s for reading", fCurrentFileName.c_str()));

      if (fRaw==0) fRaw = new uint8_t [fRawSize];
   }

   fFilesList->DeleteChild(0);

   return res;
}


bool fasp::FileInput::CloseFile()
{
   if (fFile) { fclose(fFile); fFile = 0; }
   fCurrentFileName = "";
   fCurrentRead = 0;
   return true;
}

unsigned fasp::FileInput::Read_Size()
{
   // get size of the buffer which should be read from the file

   /*   if ((fFile==0) || feof(fFile))
      if (!OpenNextFile()) {
         DOUT0(("Indicate end of stream"));
         return dabc::di_EndOfStream;
      }
    */
   return fBufferSize;
}

unsigned fasp::FileInput::Read_Complete(dabc::Buffer& buf)
{
   mbs::WriteIterator iter(buf);

   if (!iter.IsPlaceForEvent(fRawPos + sizeof(mbs::SubeventHeader))) {
      EOUT(("Failure - cannot put FASP data for SYNC %u in single MBS event - skip it", fRawSync));
      fRawPos = 0;
      fRawSync = DummySync;
   }

   int numev = 0;

   while (iter.IsPlaceForEvent(fRawPos + sizeof(mbs::SubeventHeader))) {

      bool is_end_of_stream(false);

      if ((fFile==0) || feof(fFile))
         if (!OpenNextFile()) is_end_of_stream = true;

      if (is_end_of_stream && (fRawPos==0)) break;

      uint8_t* rawbuf = fRaw + fRawPos;
      unsigned id(0);

      if (!is_end_of_stream) {

         size_t rz = fread(rawbuf, 1, FaspBlockSize, fFile);

         if (rz<FaspBlockSize) {
            if (rz>0) EOUT(("Cannot read next %u bytes from FASP file - actual only %d", FaspBlockSize, rz));
            CloseFile();
            continue;
         }

         id = rawbuf[FaspSyncPos] * 0x10000 + rawbuf[FaspSyncPos+1] * 0x100 + rawbuf[FaspSyncPos+2];

         if ((fRawSync == DummySync) || (fRawSync == id)) {
            if (fRawPos < fRawSize - FaspBlockSize) {
               fRawPos += FaspBlockSize;
               fRawSync = id;
            } else {
               EOUT(("Not enough place in raw buffer - skip data completely"));
               fRawPos = 0;
               fRawSync = DummySync;
            }
            continue;
         }
      }

      // DOUT0(("Producing event %u size %u", fRawSync, fRawPos));

      if (!iter.NewEvent(fRawSync, fRawPos + sizeof(mbs::SubeventHeader))) { EOUT(("No place for new event")); exit(1); }
      if (!iter.NewSubevent(fRawPos, 0, roc::proc_FASP, 0)) { EOUT(("No place for subevent")); exit(1); }

      memcpy(iter.rawdata(), fRaw, fRawPos);

      if (!iter.FinishSubEvent(fRawPos)) { EOUT(("Not possible to finish subevent")); exit(1); }
      if (!iter.FinishEvent()) { EOUT(("Not possible to finish event")); exit(1); }

      numev++;
      fRawPos = 0;
      fRawSync = DummySync;

      if (is_end_of_stream) break;

      // DOUT0(("Event id = 0x%08x %10u", id ,id));

      memcpy(fRaw, rawbuf, FaspBlockSize);
      fRawPos = FaspBlockSize;
      fRawSync = id;
   }

   if (numev==0) return dabc::di_EndOfStream;

   iter.Close();

   return dabc::di_Ok;
}