Geners/src/BinaryFileArchive.cc

0001 #include <sstream>
0002
0003 #include "Alignment/Geners/interface/BinaryFileArchive.hh"
0004
0005 #include "Alignment/Geners/interface/ContiguousCatalog.hh"
0006 #include "Alignment/Geners/interface/IOException.hh"
0007 #include "Alignment/Geners/interface/WriteOnlyCatalog.hh"
0008 #include "Alignment/Geners/interface/streamposIO.hh"
0009 #include "Alignment/Geners/interface/uriUtils.hh"
0010
0011 namespace gs {
0012   BinaryFileArchive::BinaryFileArchive(const char *basename,
0013                                        const char *mode,
0014                                        const char *ann,
0015                                        const unsigned dataFileBufferSize,
0016                                        const unsigned catalogFileBufferSize)
0017       : BinaryArchiveBase(basename, mode),
0018         filebuf_(nullptr),
0019         catabuf_(nullptr),
0020         annotation_(ann ? std::string(ann) : std::string("")),
0021         dataFileName_(AbsArchive::name() + ".0.gsbd"),    // binary data
0022         catalogFileName_(AbsArchive::name() + ".gsbmf"),  // binary metafile
0023         dataFileURI_(localFileURI(dataFileName_.c_str())),
0024         lastpos_(0),
0025         jumppos_(0),
0026         catalogMergeLevel_(1),
0027         annotationsMerged_(false),
0028         streamFlushed_(true) {
0029     if (!modeValid())
0030       return;
0031
0032     try {
0033       // Get a new buffer for the data and open the data stream
0034       if (dataFileBufferSize)
0035         filebuf_ = new char[dataFileBufferSize];
0036       dataStream_.rdbuf()->pubsetbuf(filebuf_, dataFileBufferSize);
0037       openDataFile(dataStream_, dataFileName_.c_str());
0038       dataStream_.seekp(0, std::ios_base::end);
0039
0040       // Get a new buffer for the catalog and open the catalog stream.
0041       // We may have to rewrite the complete catalog, so remove the flag
0042       // std::ios_base::app from the opening mode.
0043       if (catalogFileBufferSize)
0044         catabuf_ = new char[catalogFileBufferSize];
0045       catStream_.rdbuf()->pubsetbuf(catabuf_, catalogFileBufferSize);
0046       catStream_.open(catalogFileName_.c_str(), openmode() & ~std::ios_base::app);
0047       if (!catStream_.is_open())
0048         throw IOOpeningFailure("gs::BinaryFileArchive constructor", catalogFileName_);
0049
0050       // Can we use a write-only catalog?
0051       if (openmode() & std::ios_base::in) {
0052         // Reading is allowed. Have to use in-memory catalog.
0053         // If the file data already exists, get the catalog in.
0054         if (isEmptyFile(catStream_))
0055           setCatalog(new ContiguousCatalog());
0056         else
0057           readCatalog<ContiguousCatalog>();
0058       } else {
0059         // Yes, we can use a write-only catalog.
0060         // Is the catalog file empty? If so, write out
0061         // the stuff needed at the beginning of the file.
0062         // If not, assume that the necessary stuff is
0063         // already there. Note that in this case we will
0064         // not be able to add the annotation.
0065         if (isEmptyFile(catStream_)) {
0066           setCatalog(new WriteOnlyCatalog(catStream_));
0067           writeCatalog();
0068         } else {
0069           // Have to read in the catalog contents
0070           catStream_.close();
0071           catStream_.clear();
0072           catStream_.open(catalogFileName_.c_str(), openmode() | std::ios_base::in);
0073           if (!catStream_.is_open())
0074             throw IOOpeningFailure("gs::BinaryFileArchive constructor", catalogFileName_);
0075           readCatalog<WriteOnlyCatalog>();
0076           catStream_.seekp(0, std::ios_base::end);
0077         }
0078       }
0079     } catch (std::exception &e) {
0080       setCatalog(nullptr);
0081       releaseBuffers();
0082       errorStream() << e.what();
0083     }
0084   }
0085
0086   void BinaryFileArchive::releaseBuffers() {
0087     if (dataStream_.is_open())
0088       dataStream_.close();
0089     if (catStream_.is_open())
0090       catStream_.close();
0091     catStream_.rdbuf()->pubsetbuf(nullptr, 0);
0092     dataStream_.rdbuf()->pubsetbuf(nullptr, 0);
0093     delete[] catabuf_;
0094     catabuf_ = nullptr;
0095     delete[] filebuf_;
0096     filebuf_ = nullptr;
0097   }
0098
0099   BinaryFileArchive::~BinaryFileArchive() {
0100     flush();
0101     releaseBuffers();
0102   }
0103
0104   void BinaryFileArchive::writeCatalog() {
0105     if (isOpen()) {
0106       if (!annotationsMerged_) {
0107         if (!annotation_.empty())
0108           catalogAnnotations_.push_back(annotation_);
0109         annotationsMerged_ = true;
0110       }
0111       const unsigned compress = static_cast<unsigned>(compressionMode());
0112       if (!writeBinaryCatalog(catStream_, compress, catalogMergeLevel_, catalogAnnotations_, *catalog())) {
0113         std::ostringstream os;
0114         os << "In BinaryFileArchive::writeCatalog: "
0115            << "failed to write catalog data to file " << catalogFileName_;
0116         throw IOWriteFailure(os.str());
0117       }
0118     }
0119   }
0120
0121   std::istream &BinaryFileArchive::plainInputStream(const unsigned long long id,
0122                                                     unsigned *compressionCode,
0123                                                     unsigned long long *length) {
0124     if (isOpen()) {
0125       assert(openmode() & std::ios_base::in);
0126       if (!id)
0127         throw gs::IOInvalidArgument("In gs::BinaryFileArchive::plainInputStream: invalid item id");
0128       std::streampos pos;
0129       if (!catalog()->retrieveStreampos(id, compressionCode, length, &pos)) {
0130         std::ostringstream os;
0131         os << "In gs::BinaryFileArchive::plainInputStream: "
0132            << "failed to locate item with id " << id << "in the catalog stored in file " << catalogFileName_;
0133         throw gs::IOInvalidArgument(os.str());
0134       }
0135       if (!streamFlushed_) {
0136         dataStream_.flush();
0137         streamFlushed_ = true;
0138       }
0139       dataStream_.seekg(pos);
0140     }
0141     return dataStream_;
0142   }
0143
0144   unsigned long long BinaryFileArchive::addToCatalog(const AbsRecord &record,
0145                                                      const unsigned compressionCode,
0146                                                      const unsigned long long itemLength) {
0147     unsigned long long id = 0;
0148     if (isOpen()) {
0149       id = catalog()->makeEntry(record, compressionCode, itemLength, ItemLocation(lastpos_, dataFileURI_.c_str()));
0150       if (id && injectMetadata()) {
0151         const CatalogEntry *entry = catalog()->lastEntryMade();
0152         assert(entry);
0153         dataStream_.seekp(0, std::ios_base::end);
0154         std::streampos now = dataStream_.tellp();
0155         if (entry->write(dataStream_)) {
0156           dataStream_.seekp(jumppos_);
0157           write_pod(dataStream_, now);
0158           dataStream_.seekp(0, std::ios_base::end);
0159         } else
0160           id = 0;
0161       }
0162     }
0163     return id;
0164   }
0165
0166   std::ostream &BinaryFileArchive::plainOutputStream() {
0167     if (isOpen()) {
0168       assert(openmode() & std::ios_base::out);
0169       dataStream_.seekp(0, std::ios_base::end);
0170       if (injectMetadata()) {
0171         jumppos_ = dataStream_.tellp();
0172         std::streampos catpos(0);
0173         write_pod(dataStream_, catpos);
0174       }
0175       lastpos_ = dataStream_.tellp();
0176       streamFlushed_ = false;
0177     }
0178     return dataStream_;
0179   }
0180
0181   void BinaryFileArchive::flush() {
0182     if (isOpen()) {
0183       if (!streamFlushed_) {
0184         dataStream_.flush();
0185         streamFlushed_ = true;
0186       }
0187
0188       if (openmode() & std::ios_base::out) {
0189         if (dynamic_cast<WriteOnlyCatalog *>(catalog()) == nullptr)
0190           writeCatalog();
0191         catStream_.flush();
0192       }
0193     }
0194   }
0195 }  // namespace gs