HierarchyStore.cxx

Go to the documentation of this file.

// $Id: HierarchyStore.cxx 4476 2020-04-15 14:12:38Z linev $
 
/************************************************************
 * The Data Acquisition Backbone Core (DABC)                *
 ************************************************************
 * Copyright (C) 2009 -                                     *
 * GSI Helmholtzzentrum fuer Schwerionenforschung GmbH      *
 * Planckstr. 1, 64291 Darmstadt, Germany                   *
 * Contact:  http://dabc.gsi.de                             *
 ************************************************************
 * This software can be used under the GPL license          *
 * agreements as stated in LICENSE.txt file                 *
 * which is part of the distribution.                       *
 ************************************************************/
 
#include "dabc/HierarchyStore.h"
 
#include <algorithm>
 
dabc::HierarchyStore::HierarchyStore() :
   fBasePath(),
   fIO(0),
   fFile(),
   fLastStoreTm(),
   fLastFlushTm(),
   fDoStore(false),
   fDoFlush(false),
   fLastVersion(0),
   fStoreBuf(),
   fFlushBuf()
{
}
 
dabc::HierarchyStore::~HierarchyStore()
{
   CloseFile();
 
   if (fIO!=0) {
      delete fIO;
      fIO = 0;
   }
}
 
bool dabc::HierarchyStore::SetBasePath(const std::string &path)
{
   if (fFile.isOpened()) {
      EOUT("Cannot change base path when current file is opened");
      return false;
   }
 
   fBasePath = path;
   return true;
}
 
bool dabc::HierarchyStore::StartFile(dabc::Buffer buf)
{
   if (!CloseFile()) return false;
 
   std::string path = fBasePath;
   if ((path.length()>0) && (path[path.length()-1]!='/')) path.append("/");
 
   dabc::DateTime tm;
   tm.GetNow();
 
   std::string strdate = tm.OnlyDateAsString();
   std::string strtime = tm.OnlyTimeAsString();
 
   path.append(strdate);
 
   if (fIO==0) fIO = new FileInterface;
 
   if (!fIO->mkdir(path.c_str())) {
      EOUT("Cannot create path %s for hierarchy storage", path.c_str());
      return false;
   }
 
   DOUT0("HierarchyStore:: CREATE %s", path.c_str());
 
   path.append("/");
   path.append(strtime);
   path.append(".dabc");
 
   fFile.SetIO(fIO);
   if (!fFile.OpenWriting(path.c_str())) {
      EOUT("Cannot open file %s for hierarchy storage", path.c_str());
      return false;
   }
 
   return WriteDiff(buf);
}
 
bool dabc::HierarchyStore::WriteDiff(dabc::Buffer buf)
{
   if (!fFile.isWriting()) return false;
 
   BufferSize_t fullsz = buf.GetTotalSize();
 
   DOUT0("HierarchyStore::  WRITE %u", (unsigned) buf.GetTotalSize());
 
   if (!fFile.WriteBufHeader(fullsz, buf.GetTypeId())) {
      EOUT("Cannot write buffer header");
      return false;
   }
 
   for (unsigned n=0;n<buf.NumSegments();n++)
      if (!fFile.WriteBufPayload(buf.SegmentPtr(n), buf.SegmentSize(n))) {
         EOUT("Cannot write buffer segment");
         return false;
      }
 
   return true;
}
 
bool dabc::HierarchyStore::CloseFile()
{
   if (!fFile.isOpened()) return true;
 
   fFile.Close();
 
   return true;
}
 
bool dabc::HierarchyStore::CheckForNextStore(DateTime& now, double store_period, double flush_period)
{
   // if flags already set, do not touch them here
   if (fDoStore || fDoFlush) return true;
 
   if (now.null()) now.GetNow();
 
   // DOUT0("SINCE LAST STORE = %5.1f", now - fLastStoreTm);
 
   // in the very first call trigger only write of base structure
   if (fLastStoreTm.null()) {
      fLastStoreTm = now;
      fLastFlushTm = now;
      fDoStore = false;
      fDoFlush = true;
   } else
   if (now - fLastStoreTm > store_period) {
      fLastStoreTm = now;
      fDoStore = true;
      if (now - fLastFlushTm >= flush_period) {
         fDoFlush = true;
         fLastFlushTm = now;
      }
   }
 
   return fDoStore || fDoFlush;
}
 
bool dabc::HierarchyStore::ExtractData(dabc::Hierarchy& h)
{
   if (fDoStore) {
      // we record diff to previous version, including all history entries
 
      fStoreBuf = h.SaveToBuffer(dabc::stream_Full, fLastVersion, 1000000000);
 
      DOUT0("HierarchyStore::  EXTRACT prev %u now %u chlds %u size %u",
                   (unsigned) fLastVersion, (unsigned) h.GetVersion(),
                   (unsigned) h()->GetChildsVersion(), (unsigned) fStoreBuf.GetTotalSize());
 
      fLastVersion = h.GetVersion();
   }
 
   if (fDoFlush) {
      // we record complete hierarchy without any history entry
      fFlushBuf = h.SaveToBuffer(dabc::stream_Full, 0, 0);
      fLastVersion = h.GetVersion();
   }
 
   if (fDoFlush || fDoStore) {
      // this time will be used as raster when reading files
      // it can be read very efficient - most decoding can be skipped
      h.SetField("storetm", fLastStoreTm);
      h.MarkChangedItems(fLastStoreTm.AsJSDate());
   }
 
   return true;
}
 
bool dabc::HierarchyStore::WriteExtractedData()
{
   if (fDoStore) {
      if (!fStoreBuf.null()) WriteDiff(fStoreBuf);
      fStoreBuf.Release();
      fDoStore = false;
   }
 
   if (fDoFlush) {
      if (!fFlushBuf.null()) StartFile(fFlushBuf);
      fFlushBuf.Release();
      fDoFlush = false;
   }
 
   return true;
}
 
// =================================================================================
 
 
 
dabc::HierarchyReading::HierarchyReading() :
   fBasePath(),
   fIO(0)
{
}
 
dabc::HierarchyReading::~HierarchyReading()
{
   if (fIO!=0) {
      delete fIO;
      fIO = 0;
   }
}
 
void dabc::HierarchyReading::SetBasePath(const std::string &path)
{
   fBasePath = path;
   if ((fBasePath.length()>0) && (fBasePath[fBasePath.length()-1] != '/')) fBasePath.append("/");
}
 
std::string dabc::HierarchyReading::MakeFileName(const std::string &fpath, const DateTime& dt)
{
   std::string res = fpath;
   if ((res.length()>0) && (res[res.length()-1] != '/')) res.append("/");
 
   std::string strdate = dt.OnlyDateAsString();
   std::string strtime = dt.OnlyTimeAsString();
 
   res += dabc::format("%s/%s.dabc", strdate.c_str(), strtime.c_str());
 
   return res;
}
 
bool dabc::HierarchyReading::ScanFiles(const std::string &dirname, const DateTime& onlydate, std::vector<uint64_t>& vect)
{
   std::string mask = dirname + "*.dabc";
 
   Reference files = fIO->fmatch(mask.c_str(), true);
 
   for (unsigned n=0;n<files.NumChilds();n++) {
      std::string fname = files.GetChild(n).GetName();
      fname.erase(0, dirname.length());
 
      size_t pos = fname.find(".dabc");
      if (pos == std::string::npos) {
         EOUT("Wrong time file name %s", fname.c_str());
         continue;
      }
      fname.erase(pos);
 
      DateTime dt = onlydate;
      if (!dt.SetOnlyTime(fname.c_str())) {
         EOUT("Wrong time file name %s", fname.c_str());
         continue;
      }
 
      vect.push_back(dt.AsJSDate());
   }
 
   return true;
}
 
 
bool dabc::HierarchyReading::ScanTreeDir(dabc::Hierarchy& h, const std::string &dirname)
{
   if (h.null() || dirname.empty()) return false;
 
   DOUT0("SCAN %s", dirname.c_str());
 
   std::string mask = dirname + "*";
 
   Reference dirs = fIO->fmatch(mask.c_str(), false);
 
   bool isanysubdir(false), isanydatedir(false);
 
   std::vector<uint64_t> files;
 
   for (unsigned n=0;n<dirs.NumChilds();n++) {
      std::string fullsubname = dirs.GetChild(n).GetName();
 
      std::string itemname = fullsubname;
      itemname.erase(0, dirname.length());
 
      DateTime dt;
 
      DOUT0("FOUND %s PART %s", fullsubname.c_str(), itemname.c_str());
 
      if (dt.SetOnlyDate(itemname.c_str())) {
         isanydatedir = true;
         if (!ScanFiles(fullsubname + "/", dt, files)) return false;
      } else {
         isanysubdir = true;
         dabc::Hierarchy subh = h.CreateHChild(itemname);
         if (!ScanTreeDir(subh, fullsubname + "/")) return false;
      }
   }
 
   dirs.Release();
 
   if (isanysubdir && isanydatedir) {
      EOUT("DIR %s Cannot combine subdirs with date dirs!!!", dirname.c_str());
      return false;
   }
 
   if (isanydatedir && (files.size()>0)) {
 
      std::sort(files.begin(), files.end());
 
      dabc::DateTime mindt(files.front()), maxdt(files.back());
 
      DOUT0("DIR: %s mintm: %s maxtm: %s files %u", dirname.c_str(), mindt.AsJSString().c_str(), maxdt.AsJSString().c_str(), files.size());
 
      h.SetField("dabc:path", dirname);
      h.SetField("dabc:mindt", mindt);
      h.SetField("dabc:maxdt" ,maxdt);
      h.SetField("dabc:files", files);
   }
 
 
   return true;
}
 
 
bool dabc::HierarchyReading::ScanTree()
{
   fTree.Release();
   fTree.Create("TOP");
   if (fIO==0) fIO = new FileInterface;
   return ScanTreeDir(fTree, fBasePath);
}
 
dabc::Buffer dabc::HierarchyReading::ReadBuffer(dabc::BinaryFile& f)
{
 
   dabc::Buffer buf;
 
   if (f.eof()) return buf;
 
   uint64_t size(0), typ(0);
   if (!f.ReadBufHeader(&size, &typ)) return buf;
 
   buf = dabc::Buffer::CreateBuffer(size);
   if (buf.null()) return buf;
 
   if (!f.ReadBufPayload(buf.SegmentPtr(), size)) { buf.Release(); return buf; }
 
   buf.SetTotalSize(size);
   buf.SetTypeId(typ);
 
   return buf;
}
 
bool dabc::HierarchyReading::ProduceStructure(Hierarchy& tree, const DateTime& from_date, const DateTime& till_date, const std::string &entry, Hierarchy& tgt)
{
   if (tree.null()) return false;
 
   DOUT0("Produce structure for item %s", tree.ItemName().c_str());
 
   if (!tree.HasField("dabc:path")) {
      for (unsigned n=0;n<tree.NumChilds();n++) {
         Hierarchy tree_chld = tree.GetChild(n);
 
         if (!entry.empty())
            if (entry.find(tree_chld.ItemName())!=0) continue;
 
         Hierarchy tgt_chld = tgt.CreateHChild(tree_chld.GetName());
         if (!ProduceStructure(tree_chld, from_date, till_date, entry, tgt_chld)) return false;
      }
      return true;
   }
 
   std::string fpath = tree.Field("dabc:path").AsStr();
 
   int nfiles = tree.Field("dabc:files").GetArraySize();
   uint64_t* files = tree.Field("dabc:files").GetUIntArr();
 
   if ((files==0) || (nfiles<=0)) return false;
 
   DateTime dt;
   dt.SetJSDate(files[0]);
   if (!from_date.null()) {
      for (int n=0;n<nfiles;n++)
         if (files[n] <= from_date.AsJSDate())
            dt.SetJSDate(files[n]);
   }
 
   std::string fname = MakeFileName(fpath, dt);
 
   DOUT0("Opening file %s", fname.c_str());
 
   dabc::BinaryFile f;
   f.SetIO(fIO);
   if (!f.OpenReading(fname.c_str())) return false;
   dabc::Buffer buf = ReadBuffer(f);
   if (buf.null()) return false;
 
   if (!tgt.ReadFromBuffer(buf)) return false;
 
   if (!entry.empty()) {
 
      Hierarchy sel = tgt.GetTop().FindChild(entry.c_str());
 
      if (sel.null()) {
         EOUT("Cannot locate entry %s in the storage", entry.c_str());
         return false;
      }
 
      tgt.DisableReading();    // we ignore all data
      sel.EnableReading(tgt);  // beside selected element
      tgt.EnableReading();     // we enable reading of top - it stores time stamp
 
      // now we need enable only that entry
      sel.EnableHistory(100);
 
      while (!f.eof()) {
         buf = ReadBuffer(f);
         if (!buf.null())
            if (!tgt.UpdateFromBuffer(buf, stream_NoDelete)) return false;
      }
 
   }
 
   return true;
}
 
bool dabc::HierarchyReading::GetStrucutre(Hierarchy& tgt, const DateTime& dt)
{
   if (fTree.null()) {
      if (!ScanTree()) return false;
   }
 
   if (fIO==0) return false;
 
   tgt.Release();
   tgt.Create("TOP");
 
   return ProduceStructure(fTree, dt, 0, "", tgt);
}
 
dabc::Hierarchy dabc::HierarchyReading::GetSerie(const std::string &entry, const DateTime& from, const DateTime& till)
{
   dabc::Hierarchy h, res;
 
   if (fTree.null()) {
      if (!ScanTree()) return res;
   }
 
   if (fIO==0) return res;
   h.Create("TOP");
 
   if (!ProduceStructure(fTree, from, till, entry, h)) return res;
 
   res = h.FindChild(entry.c_str());
   if (res.null()) {
      EOUT("Cannot locate entry %s in the storage", entry.c_str());
      return res;
   }
 
   res.DettachFromParent();
 
   h.Release(); // rest of hierarchy is not interesting
 
   return res;
}