MultiFileArchive.cc

Go to the documentation of this file.

#include <cstdio>
#include <sstream>
 
#include "Alignment/Geners/interface/MultiFileArchive.hh"
 
#include "Alignment/Geners/interface/ContiguousCatalog.hh"
#include "Alignment/Geners/interface/IOException.hh"
#include "Alignment/Geners/interface/WriteOnlyCatalog.hh"
#include "Alignment/Geners/interface/streamposIO.hh"
#include "Alignment/Geners/interface/uriUtils.hh"
 
namespace gs {
  MultiFileArchive::MultiFileArchive(const char *basename,
                                     const char *mode,
                                     const char *ann,
                                     const unsigned typicalFileSizeInMB,
                                     const unsigned dataFileBufferSize,
                                     const unsigned catalogFileBufferSize)
      : BinaryArchiveBase(basename, mode),
        filebuf_(nullptr),
        readbuf_(nullptr),
        catabuf_(nullptr),
        annotation_(ann ? std::string(ann) : std::string("")),
        catalogFileName_(AbsArchive::name() + ".gsbmf"),  // binary metafile
        writeFileURI_("/ / / / / / /\\ \\ \\ \\"),
        readFileURI_(writeFileURI_),
        lastpos_(0),
        jumppos_(0),
        maxpos_(std::streamoff(1048576LL * typicalFileSizeInMB)),
        writeFileNumber_(0),
        catalogMergeLevel_(1),
        annotationsMerged_(false),
        streamFlushed_(true) {
    if (!modeValid())
      return;
 
    try {
      // Get a new buffer for the output stream
      if (dataFileBufferSize)
        filebuf_ = new char[dataFileBufferSize];
      writeStream_.rdbuf()->pubsetbuf(filebuf_, dataFileBufferSize);
 
      // Get a new buffer for the input stream
      if (dataFileBufferSize)
        readbuf_ = new char[dataFileBufferSize];
      separateReadStream_.rdbuf()->pubsetbuf(readbuf_, dataFileBufferSize);
 
      // Get a new buffer for the catalog and open the catalog stream.
      // We may have to rewrite the complete catalog, so remove the flag
      // std::ios_base::app from the opening mode.
      if (catalogFileBufferSize)
        catabuf_ = new char[catalogFileBufferSize];
      catStream_.rdbuf()->pubsetbuf(catabuf_, catalogFileBufferSize);
      catStream_.open(catalogFileName_.c_str(), openmode() & ~std::ios_base::app);
      if (!catStream_.is_open())
        throw IOOpeningFailure("gs::MultiFileArchive constructor", catalogFileName_);
 
      // Can we use a write-only catalog?
      if (openmode() & std::ios_base::in) {
        // Reading is allowed. Have to use in-memory catalog.
        // If the file data already exists, get the catalog in.
        if (isEmptyFile(catStream_))
          setCatalog(new ContiguousCatalog());
        else
          readCatalog<ContiguousCatalog>();
      } else {
        // Yes, we can use a write-only catalog.
        // Is the catalog file empty? If so, write out
        // the stuff needed at the beginning of the file.
        // If not, assume that the necessary stuff is
        // already there. Note that in this case we will
        // not be able to add the annotation.
        if (isEmptyFile(catStream_)) {
          setCatalog(new WriteOnlyCatalog(catStream_));
          writeCatalog();
        } else {
          catStream_.close();
          catStream_.clear();
          catStream_.open(catalogFileName_.c_str(), openmode() | std::ios_base::in);
          if (!catStream_.is_open())
            throw IOOpeningFailure("gs::MultiFileArchive constructor", catalogFileName_);
          readCatalog<WriteOnlyCatalog>();
          catStream_.seekp(0, std::ios_base::end);
        }
      }
 
      // Open the write stream
      if (openmode() & std::ios_base::out) {
        setupWriteStream();
        const std::streampos pos1 = writeStream_.tellp();
        if (maxpos_ < pos1)
          maxpos_ = pos1;
      }
    } catch (std::exception &e) {
      setCatalog(nullptr);
      releaseBuffers();
      errorStream() << e.what();
    }
  }
 
  void MultiFileArchive::releaseBuffers() {
    if (writeStream_.is_open())
      writeStream_.close();
    if (separateReadStream_.is_open())
      separateReadStream_.close();
    if (catStream_.is_open())
      catStream_.close();
    catStream_.rdbuf()->pubsetbuf(nullptr, 0);
    writeStream_.rdbuf()->pubsetbuf(nullptr, 0);
    separateReadStream_.rdbuf()->pubsetbuf(nullptr, 0);
    delete[] catabuf_;
    catabuf_ = nullptr;
    delete[] readbuf_;
    readbuf_ = nullptr;
    delete[] filebuf_;
    filebuf_ = nullptr;
  }
 
  MultiFileArchive::~MultiFileArchive() {
    flush();
    releaseBuffers();
  }
 
  void MultiFileArchive::writeCatalog() {
    if (isOpen()) {
      if (!annotationsMerged_) {
        if (!annotation_.empty())
          catalogAnnotations_.push_back(annotation_);
        annotationsMerged_ = true;
      }
      const unsigned compress = static_cast<unsigned>(compressionMode());
      if (!writeBinaryCatalog(catStream_, compress, catalogMergeLevel_, catalogAnnotations_, *catalog())) {
        std::ostringstream os;
        os << "In MultiFileArchive::writeCatalog: "
           << "failed to write catalog data to file " << catalogFileName_;
        throw IOWriteFailure(os.str());
      }
    }
  }
 
  void MultiFileArchive::openWriteStream() {
    assert(openmode() & std::ios_base::out);
    assert(!writeStream_.is_open());
    {
      std::ostringstream os;
      os << AbsArchive::name() << '.' << writeFileNumber_ << ".gsbd";
      writeFileName_ = os.str();
    }
    writeFileURI_ = localFileURI(writeFileName_.c_str());
    openDataFile(writeStream_, writeFileName_.c_str());
  }
 
  std::ostream &MultiFileArchive::plainOutputStream() {
    if (isOpen()) {
      assert(openmode() & std::ios_base::out);
      if (writeStream_.is_open()) {
        writeStream_.seekp(0, std::ios_base::end);
        lastpos_ = writeStream_.tellp();
        if (lastpos_ > maxpos_) {
          writeStream_.close();
          // Don't have to clear. "openDataFile" will do it.
          // writeStream_.clear();
          ++writeFileNumber_;
        } else if (injectMetadata()) {
          jumppos_ = lastpos_;
          std::streampos catpos(0);
          write_pod(writeStream_, catpos);
          lastpos_ = writeStream_.tellp();
        }
      }
      if (!writeStream_.is_open()) {
        openWriteStream();
        writeStream_.seekp(0, std::ios_base::end);
        if (injectMetadata()) {
          jumppos_ = writeStream_.tellp();
          std::streampos catpos(0);
          write_pod(writeStream_, catpos);
        }
        lastpos_ = writeStream_.tellp();
      }
      streamFlushed_ = false;
    }
    return writeStream_;
  }
 
  void MultiFileArchive::flush() {
    if (isOpen()) {
      if (!streamFlushed_) {
        writeStream_.flush();
        streamFlushed_ = true;
      }
 
      if (openmode() & std::ios_base::out) {
        if (dynamic_cast<WriteOnlyCatalog *>(catalog()) == nullptr)
          writeCatalog();
        catStream_.flush();
      }
    }
  }
 
  void MultiFileArchive::setupWriteStream() {
    if (openmode() & std::ios_base::trunc) {
      bool removed = true;
      for (unsigned i = 0; removed; ++i) {
        std::ostringstream os;
        os << AbsArchive::name() << '.' << i << ".gsbd";
        std::string fname = os.str();
        removed = std::remove(fname.c_str()) == 0;
      }
      writeFileNumber_ = 0;
    } else {
      unsigned long firstNonExistent = 0;
      for (;; ++firstNonExistent) {
        std::ostringstream os;
        os << AbsArchive::name() << '.' << firstNonExistent << ".gsbd";
        std::string fname = os.str();
        std::ifstream f(fname.c_str());
        if (!f)
          break;
      }
      writeFileNumber_ = firstNonExistent ? firstNonExistent - 1UL : 0UL;
    }
    openWriteStream();
  }
 
  std::istream &MultiFileArchive::plainInputStream(const unsigned long long id,
                                                   unsigned *compressionCode,
                                                   unsigned long long *length) {
    std::fstream *readStream = &writeStream_;
    if (isOpen()) {
      assert(openmode() & std::ios_base::in);
      if (!id)
        throw gs::IOInvalidArgument("In gs::MultiFileArchive::plainInputStream: invalid item id");
 
      // If we have a write stream, and if the archive
      // has one file only, we should be able to retrieve
      // stream position quickly
      std::streampos pos(0);
      if ((openmode() & std::ios_base::out) && writeFileNumber_ == 0UL) {
        if (!catalog()->retrieveStreampos(id, compressionCode, length, &pos)) {
          std::ostringstream os;
          os << "In gs::MultiFileArchive::plainInputStream: "
             << "failed to locate item with id " << id << "in the catalog stored in file " << catalogFileName_;
          throw gs::IOInvalidArgument(os.str());
        }
      } else {
        // Here, we have to do a full catalog search
        std::shared_ptr<const CatalogEntry> sptr = catalog()->retrieveEntry(id);
        const CatalogEntry *pe = sptr.get();
        if (!pe) {
          std::ostringstream os;
          os << "In gs::MultiFileArchive::plainInputStream: "
             << "failed to locate item with id " << id << "in the catalog stored in file " << catalogFileName_;
          throw gs::IOInvalidArgument(os.str());
        }
        pos = pe->location().streamPosition();
        if (pe->location().URI() != writeFileURI_) {
          updateReadStream(pe->location().URI());
          readStream = &separateReadStream_;
        }
        *compressionCode = pe->compressionCode();
        *length = pe->itemLength();
      }
 
      // Flush the write stream if it will be used for reading
      if (readStream == &writeStream_) {
        assert(writeStream_.is_open());
        if (!streamFlushed_) {
          writeStream_.flush();
          streamFlushed_ = true;
        }
      }
 
      readStream->seekg(pos);
    }
    return *readStream;
  }
 
  void MultiFileArchive::updateReadStream(const std::string &uri) {
    if (uri == readFileURI_)
      return;
 
    assert(openmode() & std::ios_base::in);
    if (separateReadStream_.is_open()) {
      separateReadStream_.close();
      separateReadStream_.clear();
    }
 
    // We need to get the name of the local file from the URI.
    // We will assume that it belongs to the archive we are
    // working with right now.
    readFileName_ = joinDir1WithName2(AbsArchive::name().c_str(), uri.c_str());
    separateReadStream_.open(readFileName_.c_str(), std::ios_base::binary | std::ios_base::in);
    if (!separateReadStream_.is_open())
      throw IOOpeningFailure("gs::MultiFileArchive::updateReadStream", readFileName_);
    readFileURI_ = uri;
  }
 
  unsigned long long MultiFileArchive::addToCatalog(const AbsRecord &record,
                                                    const unsigned compressionCode,
                                                    const unsigned long long itemLength) {
    unsigned long long id = 0;
    if (isOpen()) {
      id = catalog()->makeEntry(record, compressionCode, itemLength, ItemLocation(lastpos_, writeFileURI_.c_str()));
      if (id && injectMetadata()) {
        const CatalogEntry *entry = catalog()->lastEntryMade();
        assert(entry);
        writeStream_.seekp(0, std::ios_base::end);
        std::streampos now = writeStream_.tellp();
        if (entry->write(writeStream_)) {
          writeStream_.seekp(jumppos_);
          write_pod(writeStream_, now);
          writeStream_.seekp(0, std::ios_base::end);
        } else
          id = 0;
      }
    }
    return id;
  }
}  // namespace gs