"add crc32 encoder"

Author: dzhwinter

paddle/fluid/recordio/CMakeLists.txt

@@ -2,3 +2,4 @@ cc_library(header SRCS header.cc)
 cc_test(header_test SRCS header_test.cc DEPS header)
 cc_library(io SRCS io.cc DEPS stringpiece)
 cc_test(io_test SRCS io_test.cc DEPS io)
+cc_library(chunk SRCS chunk.cc DEPS snappy)

paddle/fluid/recordio/chunk.cc

@@ -0,0 +1,105 @@
+//   Copyright (c) 2018 PaddlePaddle Authors. All Rights Reserved.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+#include "paddle/fluid/recordio/chunk.h"
+
+#include <cstring>
+#include <sstream>
+#include <utility>
+
+#include "snappy.h"
+
+#include "paddle/fluid/recordio/crc32.h"
+
+namespace paddle {
+namespace recordio {
+
+void Chunk::Add(const char* record, size_t length) {
+  records_.emplace_after(std::move(s));
+  num_bytes_ += s.size() * sizeof(char);
+}
+
+bool Chunk::Dump(Stream* fo, Compressor ct) {
+  // NOTE(dzhwinter): don't check records.numBytes instead, because
+  // empty records are allowed.
+  if (records_.size() == 0) return false;
+
+  // pack the record into consecutive memory for compress
+  std::ostringstream os;
+  for (auto& record : records_) {
+    os.write(record.size(), sizeof(size_t));
+    os.write(record.data(), static_cast<std::streamsize>(record.size()));
+  }
+
+  std::unique_ptr<char[]> buffer(new char[kDefaultMaxChunkSize]);
+  size_t compressed =
+      CompressData(os.str().c_str(), num_bytes_, ct, buffer.get());
+  uint32_t checksum = Crc32(buffer.get(), compressed);
+  Header hdr(records_.size(), checksum, ct, static_cast<uint32_t>(compressed));
+  hdr.Write(fo);
+  fo.Write(buffer.get(), compressed);
+  return true;
+}
+
+void Chunk::Parse(Stream* fi, size_t offset) {
+  fi->Seek(offset);
+  Header hdr;
+  hdr.Parse(fi);
+
+  std::unique_ptr<char[]> buffer(new char[kDefaultMaxChunkSize]);
+  fi->Read(buffer.get(), static_cast<size_t>(hdr.CompressSize()));
+  uint32_t deflated_size =
+      DeflateData(buffer.get(), hdr.CompressSize(), hdr.CompressType());
+  std::istringstream deflated(std::string(buffer.get(), deflated_size));
+  for (size_t i = 0; i < hdr.NumRecords(); ++i) {
+    uint32_t rs;
+    deflated >> rs;
+    std::string record(rs, '\0');
+    deflated.read(&record[0], rs);
+    records_.emplace_back(record);
+    num_bytes_ += record.size();
+  }
+}
+
+size_t CompressData(const char* in,
+                    size_t in_length,
+                    Compressor ct,
+                    char* out) {
+  size_t compressd_size = 0;
+  switch (ct) {
+    case Compressor::kNoCompress:
+      // do nothing
+      memcpy(out, in, in_length);
+      compressd_size = in_length;
+      break;
+    case Compressor::kSnappy:
+      snappy::RawCompress(in, in_length, out, &compressd_size);
+      break;
+  }
+  return compressd_size;
+}
+
+void DeflateData(const char* in, size_t in_length, Compressor ct, char* out) {
+  switch (c) {
+    case Compressor::kNoCompress:
+      memcpy(out, in, in_length);
+      break;
+    case Compressor::kSnappy:
+      snappy::RawUncompress(in, in_length, out);
+      break;
+  }
+}
+
+}  // namespace recordio
+}  // namespace paddle

paddle/fluid/recordio/chunk.h

@@ -13,109 +13,36 @@
 // limitations under the License.
 
 #pragma once
-
-#include <fstream>
-#include <memory>
-#include <sstream>
+#include <forward_list>
 #include <string>
-#include <utility>
-#include <vector>
 
-// Chunk
-// a chunk contains the Header and optionally compressed records.
+#include "paddle/fluid/recordio/header.h"
+#include "paddle/fluid/recordio/io.h"
+
+namespace paddle {
+namespace recordio {
+
+// A Chunk contains the Header and optionally compressed records.
 class Chunk {
 public:
-  Chunk() = default;
-  void Add(const char* record, size_t length);
-  void Add(const std::string&);
-
-  bool Dump(std::ostream& os, Compressor ct);
-  void Parse(std::istream& iss, int64_t offset);
-  const std::string Record(int i) { return records_[i]; }
+  Chunk() {}
+  void Add(const char* record, size_t size);
+  // dump the chunk into w, and clears the chunk and makes it ready for
+  // the next add invocation.
+  bool Dump(Stream* fo, Compressor ct);
+  void Parse(Stream* fi, size_t offset);
   size_t NumBytes() { return num_bytes_; }
 
 private:
-  std::vector<std::string> records_;
+  std::forward_list<std::string> records_;
   // sum of record lengths in bytes.
   size_t num_bytes_;
+  DISABLE_COPY_AND_ASSIGN(Chunk);
 };
 
-size_t CompressData(const std::stringstream& ss, Compressor ct, char* buffer);
-
-uint32_t DeflateData(char* buffer, uint32_t size, Compressor c);
-
-// implementation
-void Chunk::Add(const std::string& s) {
-  num_bytes_ += s.size() * sizeof(char);
-  records_.emplace_back(std::move(s));
-  // records_.resize(records_.size()+1);
-  // records_[records_.size()-1] = s;
-}
-
-void Chunk::Add(const char* record, size_t length) {
-  Add(std::string(record, length));
-}
-
-bool Chunk::Dump(std::ostream& os, Compressor ct) {
-  if (records_.size() == 0) return false;
-
-  // TODO(dzhwinter):
-  // we pack the string with same size buffer,
-  // then compress with another buffer.
-  // Here can be optimized if it is the bottle-neck.
-  std::ostringstream oss;
-  for (auto& record : records_) {
-    unsigned len = record.size();
-    oss << len;
-    oss << record;
-    // os.write(std::to_string(len).c_str(), sizeof(unsigned));
-    // os.write(record.c_str(), record.size());
-  }
-  std::unique_ptr<char[]> buffer(new char[kDefaultMaxChunkSize]);
-  size_t compressed = CompressData(oss.str(), ct, buffer.get());
-
-  // TODO(dzhwinter): crc32 checksum
-  size_t checksum = compressed;
-
-  Header hdr(records_.size(), checksum, ct, compressed);
-
-  return true;
-}
-
-void Chunk::Parse(std::istream& iss, int64_t offset) {
-  iss.seekg(offset, iss.beg);
-  Header hdr;
-  hdr.Parse(iss);
+size_t CompressData(const char* in, size_t in_length, Compressor ct, char* out);
 
-  std::unique_ptr<char[]> buffer(new char[kDefaultMaxChunkSize]);
-  iss.read(buffer.get(), static_cast<size_t>(hdr.CompressSize()));
-  // TODO(dzhwinter): checksum
-  uint32_t deflated_size =
-      DeflateData(buffer.get(), hdr.CompressSize(), hdr.CompressType());
-  std::istringstream deflated(std::string(buffer.get(), deflated_size));
-  for (size_t i = 0; i < hdr.NumRecords(); ++i) {
-    uint32_t rs;
-    deflated >> rs;
-    std::string record(rs, '\0');
-    deflated.read(&record[0], rs);
-    records_.emplace_back(record);
-    num_bytes_ += record.size();
-  }
-}
+void DeflateData(const char* in, size_t in_length, Compressor ct, char* out);
 
-uint32_t DeflateData(char* buffer, uint32_t size, Compressor c) {
-  uint32_t deflated_size = 0;
-  std::string uncompressed;
-  switch (c) {
-    case Compressor::kNoCompress:
-      deflated_size = size;
-      break;
-    case Compressor::kSnappy:
-      // snappy::Uncompress(buffer, size, &uncompressed);
-      // deflated_size = uncompressed.size();
-      // memcpy(buffer, uncompressed.data(), uncompressed.size() *
-      // sizeof(char));
-      break;
-  }
-  return deflated_size;
-}
+}  // namespace recordio
+}  // namespace paddle

paddle/fluid/recordio/chunk_test.cc

@@ -0,0 +1,23 @@
+//   Copyright (c) 2018 PaddlePaddle Authors. All Rights Reserved.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+#include "paddle/fluid/recordio/chunk.h"
+
+#include <sstream>
+
+#include "gtest/gtest.h"
+
+using namespace paddle::recordio;
+
+TEST(Chunk, SaveLoad) {}

paddle/fluid/recordio/crc32.h

@@ -0,0 +1,33 @@
+//   Copyright (c) 2018 PaddlePaddle Authors. All Rights Reserved.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+// A wrapper on crc library https://github.com/d-bahr/CRCpp
+#include <cstdint>
+
+#include "paddle/fluid/recordio/detail/crc.h"
+
+namespace paddle {
+namespace recordio {
+
+// usage
+// char data[] = "hello,world";
+// crc = Crc32(data, 12);
+// Assert_EQ(crc, 68a85159);
+
+uint32_t Crc32(const char* data, size_t size) {
+  return CRC::Calculate(data, size, CRC::CRC_32())
+}
+
+}  // namespace recordio
+}  // namespace paddle