Geners/src/AbsArchive.cc

0001 #include <cassert>
0002 #include <sstream>
0003
0004 #include "Alignment/Geners/interface/AbsArchive.hh"
0005 #include "Alignment/Geners/interface/IOException.hh"
0006
0007 #define GS_STREAM_COPY_BUFFER_SIZE 65536
0008
0009 static void archive_stream_copy(std::istream &in, std::size_t count, std::ostream &out) {
0010   if (count) {
0011     const std::size_t bufsize = GS_STREAM_COPY_BUFFER_SIZE;
0012     char buffer[bufsize];
0013
0014     bool in_fail = in.fail();
0015     bool out_fail = out.fail();
0016     while (count > bufsize && !in_fail && !out_fail) {
0017       in.read(buffer, bufsize);
0018       in_fail = in.fail();
0019       if (!in_fail) {
0020         out.write(buffer, bufsize);
0021         out_fail = out.fail();
0022       }
0023       count -= bufsize;
0024     }
0025     if (!in_fail && !out_fail) {
0026       in.read(buffer, count);
0027       if (!in.fail())
0028         out.write(buffer, count);
0029     }
0030   }
0031 }
0032
0033 namespace {
0034   class NotWritableRecord : public gs::AbsRecord {
0035   public:
0036     inline NotWritableRecord(const gs::ClassId &classId, const char *ioPrototype, const char *name, const char *category)
0037         : gs::AbsRecord(classId, ioPrototype, name, category) {}
0038
0039     NotWritableRecord() = delete;
0040
0041   private:
0042     inline bool writeData(std::ostream &) const override { return false; }
0043   };
0044 }  // namespace
0045
0046 namespace gs {
0047   AbsArchive::AbsArchive(const char *name) : name_(name ? name : ""), lastItemId_(0), lastItemLength_(0) {}
0048
0049   void AbsArchive::addItemToReference(AbsReference &r, const unsigned long long id) const { r.addItemId(id); }
0050
0051   unsigned long long AbsArchive::copyItem(const unsigned long long id,
0052                                           AbsArchive *destination,
0053                                           const char *newName,
0054                                           const char *newCategory) {
0055     if (!isReadable())
0056       throw gs::IOInvalidArgument(
0057           "In gs::AbsArchive::copyItem: "
0058           "origin archive is not readable");
0059     assert(destination);
0060     if (this == destination)
0061       throw gs::IOInvalidArgument(
0062           "In gs::AbsArchive::copyItem: "
0063           "origin and destination archives are the same");
0064     AbsArchive &ar(*destination);
0065     if (!ar.isWritable())
0066       throw gs::IOInvalidArgument(
0067           "In gs::AbsArchive::copyItem: "
0068           "destination archive is not writable");
0069     std::shared_ptr<const CatalogEntry> entry(catalogEntry(id));
0070     if (!entry)
0071       throw gs::IOInvalidArgument(
0072           "In gs::AbsArchive::copyItem: no item "
0073           "in the archive with the given id");
0074
0075     // Position the input stream
0076     long long sz = 0;
0077     std::istream &is = inputStream(id, &sz);
0078
0079     // The following code is similar to the one in the "operator<<"
0080     // below w.r.t. operations with the output archive
0081     std::ostream &os = ar.outputStream();
0082     std::streampos base = os.tellp();
0083     std::ostream &compressed = ar.compressedStream(os);
0084
0085     // Transfer the data between the streams
0086     unsigned long long len;
0087     if (sz >= 0)
0088       len = sz;
0089     else
0090       len = entry->itemLength();
0091     archive_stream_copy(is, len, compressed);
0092     if (is.fail())
0093       throw IOReadFailure(
0094           "In gs::AbsArchive::copyItem: "
0095           "input stream failure");
0096     if (compressed.fail())
0097       throw IOWriteFailure(
0098           "In gs::AbsArchive::copyItem: "
0099           "output stream failure");
0100     const unsigned compressCode = ar.flushCompressedRecord(compressed);
0101     if (os.fail())
0102       throw IOWriteFailure(
0103           "In gs::AbsArchive::copyItem: "
0104           "failed to transfer compressed data");
0105
0106     // Figure out the record length. Naturally, can't have negative length.
0107     std::streamoff off = os.tellp() - base;
0108     const long long delta = off;
0109     assert(delta >= 0LL);
0110
0111     // We need to create a record out of the catalog entry we have found
0112     const char *name = newName;
0113     if (!name)
0114       name = entry->name().c_str();
0115     const char *category = newCategory;
0116     if (!category)
0117       category = entry->category().c_str();
0118     NotWritableRecord record(entry->type(), entry->ioPrototype().c_str(), name, category);
0119
0120     // Add the record to the catalog
0121     const unsigned long long id2 = ar.addToCatalog(record, compressCode, delta);
0122     if (id == 0ULL)
0123       throw IOWriteFailure(
0124           "In gs::AbsArchive::copyItem: "
0125           "failed to add catalog entry");
0126     ar.lastItemId_ = id2;
0127     ar.lastItemLength_ = delta;
0128     return id2;
0129   }
0130 }  // namespace gs
0131
0132 static std::string local_error_message(gs::AbsArchive &ar, const gs::AbsRecord &record, const char *failedAction) {
0133   std::ostringstream err;
0134   err << "In operator<<(gs::AbsArchive& ar, const gs::AbsRecord& record): "
0135       << "failed to " << failedAction << " to the archive \"" << ar.name() << "\" for item with type \""
0136       << record.type().name() << "\", name \"" << record.name() << "\", and category \"" << record.category() << '"';
0137   return err.str();
0138 }
0139
0140 gs::AbsArchive &operator<<(gs::AbsArchive &ar, const gs::AbsRecord &record) {
0141   // Do not reuse records
0142   if (record.id())
0143     throw gs::IOInvalidArgument(
0144         "In operator<<(gs::AbsArchive& ar, const gs::AbsRecord& record): "
0145         "records can not be reused");
0146
0147   // Get the relevant streams. Do not change the order
0148   // of the next three lines, some code which does
0149   // not implement compression may actually rely on
0150   // the fact that the "outputStream()" method
0151   // is called before "compressedStream()".
0152   std::ostream &os = ar.outputStream();
0153   std::streampos base = os.tellp();
0154   std::ostream &compressed = ar.compressedStream(os);
0155
0156   // Write the data
0157   if (!record.writeData(compressed))
0158     throw gs::IOWriteFailure(local_error_message(ar, record, "write data"));
0159
0160   const unsigned compressCode = ar.flushCompressedRecord(compressed);
0161   if (os.fail())
0162     throw gs::IOWriteFailure(local_error_message(ar, record, "transfer compressed data"));
0163
0164   // Figure out the record length. Naturally, can't have negative length.
0165   std::streamoff off = os.tellp() - base;
0166   const long long delta = off;
0167   assert(delta >= 0LL);
0168
0169   // Add the record to the catalog
0170   const unsigned long long id = ar.addToCatalog(record, compressCode, delta);
0171   if (id == 0ULL)
0172     throw gs::IOWriteFailure(local_error_message(ar, record, "add catalog entry"));
0173
0174   // Mark record as written and give feedback about item length
0175   record.itemId_ = id;
0176   record.itemLength_ = delta;
0177
0178   // Same thing for the archive
0179   ar.lastItemId_ = id;
0180   ar.lastItemLength_ = delta;
0181
0182   return ar;
0183 }