2 * Copyright 2014 Facebook, Inc.
4 * Licensed under the Apache License, Version 2.0 (the "License");
5 * you may not use this file except in compliance with the License.
6 * You may obtain a copy of the License at
8 * http://www.apache.org/licenses/LICENSE-2.0
10 * Unless required by applicable law or agreed to in writing, software
11 * distributed under the License is distributed on an "AS IS" BASIS,
12 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13 * See the License for the specific language governing permissions and
14 * limitations under the License.
17 #include <folly/io/RecordIO.h>
19 #include <sys/types.h>
22 #include <folly/Exception.h>
23 #include <folly/FileUtil.h>
24 #include <folly/Memory.h>
25 #include <folly/Portability.h>
26 #include <folly/ScopeGuard.h>
27 #include <folly/String.h>
31 using namespace recordio_helpers;
33 RecordIOWriter::RecordIOWriter(File file, uint32_t fileId)
34 : file_(std::move(file)),
36 writeLock_(file_, std::defer_lock),
38 if (!writeLock_.try_lock()) {
39 throw std::runtime_error("RecordIOWriter: file locked by another process");
43 checkUnixError(fstat(file_.fd(), &st), "fstat() failed");
45 filePos_ = st.st_size;
48 void RecordIOWriter::write(std::unique_ptr<IOBuf> buf) {
49 size_t totalLength = prependHeader(buf, fileId_);
50 if (totalLength == 0) {
51 return; // nothing to do
54 DCHECK_EQ(buf->computeChainDataLength(), totalLength);
56 // We're going to write. Reserve space for ourselves.
57 off_t pos = filePos_.fetch_add(totalLength);
59 #if FOLLY_HAVE_PWRITEV
60 auto iov = buf->getIov();
61 ssize_t bytes = pwritevFull(file_.fd(), iov.data(), iov.size(), pos);
65 ssize_t bytes = pwriteFull(file_.fd(), buf->data(), buf->length(), pos);
68 checkUnixError(bytes, "pwrite() failed");
69 DCHECK_EQ(bytes, totalLength);
72 RecordIOReader::RecordIOReader(File file, uint32_t fileId)
73 : map_(std::move(file)),
77 RecordIOReader::Iterator::Iterator(ByteRange range, uint32_t fileId, off_t pos)
80 recordAndPos_(ByteRange(), 0) {
81 if (pos >= range_.size()) {
82 recordAndPos_.second = off_t(-1);
85 recordAndPos_.second = pos;
91 void RecordIOReader::Iterator::advanceToValid() {
92 ByteRange record = findRecord(range_, fileId_).record;
94 recordAndPos_ = std::make_pair(ByteRange(), off_t(-1));
95 range_.clear(); // at end
97 size_t skipped = record.begin() - range_.begin();
98 DCHECK_GE(skipped, headerSize());
99 skipped -= headerSize();
100 range_.advance(skipped);
101 recordAndPos_.first = record;
102 recordAndPos_.second += skipped;
106 namespace recordio_helpers {
108 using namespace detail;
112 constexpr uint32_t kHashSeed = 0xdeadbeef; // for mcurtiss
114 uint32_t headerHash(const Header& header) {
115 return hash::SpookyHashV2::Hash32(&header, offsetof(Header, headerHash),
119 std::pair<size_t, uint64_t> dataLengthAndHash(const IOBuf* buf) {
121 hash::SpookyHashV2 hasher;
122 hasher.Init(kHashSeed, kHashSeed);
123 for (auto br : *buf) {
125 hasher.Update(br.data(), br.size());
129 hasher.Final(&hash1, &hash2);
130 if (len + headerSize() >= std::numeric_limits<uint32_t>::max()) {
131 throw std::invalid_argument("Record length must fit in 32 bits");
133 return std::make_pair(len, hash1);
136 uint64_t dataHash(ByteRange range) {
137 return hash::SpookyHashV2::Hash64(range.data(), range.size(), kHashSeed);
142 size_t prependHeader(std::unique_ptr<IOBuf>& buf, uint32_t fileId) {
144 throw std::invalid_argument("invalid file id");
146 auto lengthAndHash = dataLengthAndHash(buf.get());
147 if (lengthAndHash.first == 0) {
148 return 0; // empty, nothing to do, no zero-length records
151 // Prepend to the first buffer in the chain if we have room, otherwise
152 // prepend a new buffer.
153 if (buf->headroom() >= headerSize()) {
155 buf->prepend(headerSize());
157 auto b = IOBuf::create(headerSize());
158 b->append(headerSize());
159 b->appendChain(std::move(buf));
162 detail::Header* header =
163 reinterpret_cast<detail::Header*>(buf->writableData());
164 memset(header, 0, sizeof(Header));
165 header->magic = detail::Header::kMagic;
166 header->fileId = fileId;
167 header->dataLength = lengthAndHash.first;
168 header->dataHash = lengthAndHash.second;
169 header->headerHash = headerHash(*header);
171 return lengthAndHash.first + headerSize();
174 RecordInfo validateRecord(ByteRange range, uint32_t fileId) {
175 if (range.size() <= headerSize()) { // records may not be empty
178 const Header* header = reinterpret_cast<const Header*>(range.begin());
179 range.advance(sizeof(Header));
180 if (header->magic != Header::kMagic ||
181 header->version != 0 ||
182 header->hashFunction != 0 ||
183 header->flags != 0 ||
184 (fileId != 0 && header->fileId != fileId) ||
185 header->dataLength > range.size()) {
188 if (headerHash(*header) != header->headerHash) {
191 range.reset(range.begin(), header->dataLength);
192 if (dataHash(range) != header->dataHash) {
195 return {header->fileId, range};
198 RecordInfo findRecord(ByteRange searchRange,
199 ByteRange wholeRange,
201 static const uint32_t magic = Header::kMagic;
202 static const ByteRange magicRange(reinterpret_cast<const uint8_t*>(&magic),
205 DCHECK_GE(searchRange.begin(), wholeRange.begin());
206 DCHECK_LE(searchRange.end(), wholeRange.end());
208 const uint8_t* start = searchRange.begin();
209 const uint8_t* end = std::min(searchRange.end(),
210 wholeRange.end() - sizeof(Header));
211 // end-1: the last place where a Header could start
212 while (start < end) {
213 auto p = ByteRange(start, end + sizeof(magic)).find(magicRange);
214 if (p == ByteRange::npos) {
219 auto r = validateRecord(ByteRange(start, wholeRange.end()), fileId);
220 if (!r.record.empty()) {
224 // No repeated prefix in magic, so we can do better than start++
225 start += sizeof(magic);