2 * Copyright 2013 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/ScopeGuard.h"
26 #include "folly/String.h"
30 using namespace recordio_helpers;
32 RecordIOWriter::RecordIOWriter(File file, uint32_t fileId)
33 : file_(std::move(file)),
35 writeLock_(file_, std::defer_lock),
37 if (!writeLock_.try_lock()) {
38 throw std::runtime_error("RecordIOWriter: file locked by another process");
42 checkUnixError(fstat(file_.fd(), &st), "fstat() failed");
44 filePos_ = st.st_size;
47 void RecordIOWriter::write(std::unique_ptr<IOBuf> buf) {
48 size_t totalLength = prependHeader(buf, fileId_);
49 if (totalLength == 0) {
50 return; // nothing to do
53 DCHECK_EQ(buf->computeChainDataLength(), totalLength);
54 auto iov = buf->getIov();
56 // We're going to write. Reserve space for ourselves.
57 off_t pos = filePos_.fetch_add(totalLength);
58 ssize_t bytes = pwritevFull(file_.fd(), iov.data(), iov.size(), pos);
59 checkUnixError(bytes, "pwrite() failed");
60 DCHECK_EQ(bytes, totalLength);
63 RecordIOReader::RecordIOReader(File file, uint32_t fileId)
64 : map_(std::move(file)),
68 RecordIOReader::Iterator::Iterator(ByteRange range, uint32_t fileId, off_t pos)
71 recordAndPos_(ByteRange(), 0) {
72 if (pos >= range_.size()) {
73 recordAndPos_.second = off_t(-1);
76 recordAndPos_.second = pos;
82 void RecordIOReader::Iterator::advanceToValid() {
83 ByteRange record = findRecord(range_, fileId_).record;
85 recordAndPos_ = std::make_pair(ByteRange(), off_t(-1));
86 range_.clear(); // at end
88 size_t skipped = record.begin() - range_.begin();
89 DCHECK_GE(skipped, headerSize());
90 skipped -= headerSize();
91 range_.advance(skipped);
92 recordAndPos_.first = record;
93 recordAndPos_.second += skipped;
97 namespace recordio_helpers {
99 using namespace detail;
103 constexpr uint32_t kHashSeed = 0xdeadbeef; // for mcurtiss
105 uint32_t headerHash(const Header& header) {
106 return hash::SpookyHashV2::Hash32(&header, offsetof(Header, headerHash),
110 std::pair<size_t, uint64_t> dataLengthAndHash(const IOBuf* buf) {
112 hash::SpookyHashV2 hasher;
113 hasher.Init(kHashSeed, kHashSeed);
114 for (auto br : *buf) {
116 hasher.Update(br.data(), br.size());
120 hasher.Final(&hash1, &hash2);
121 if (len + headerSize() >= std::numeric_limits<uint32_t>::max()) {
122 throw std::invalid_argument("Record length must fit in 32 bits");
124 return std::make_pair(len, hash1);
127 uint64_t dataHash(ByteRange range) {
128 return hash::SpookyHashV2::Hash64(range.data(), range.size(), kHashSeed);
133 size_t prependHeader(std::unique_ptr<IOBuf>& buf, uint32_t fileId) {
135 throw std::invalid_argument("invalid file id");
137 auto lengthAndHash = dataLengthAndHash(buf.get());
138 if (lengthAndHash.first == 0) {
139 return 0; // empty, nothing to do, no zero-length records
142 // Prepend to the first buffer in the chain if we have room, otherwise
143 // prepend a new buffer.
144 if (buf->headroom() >= headerSize()) {
146 buf->prepend(headerSize());
148 auto b = IOBuf::create(headerSize());
149 b->append(headerSize());
150 b->appendChain(std::move(buf));
153 detail::Header* header =
154 reinterpret_cast<detail::Header*>(buf->writableData());
155 memset(header, 0, sizeof(Header));
156 header->magic = detail::Header::kMagic;
157 header->fileId = fileId;
158 header->dataLength = lengthAndHash.first;
159 header->dataHash = lengthAndHash.second;
160 header->headerHash = headerHash(*header);
162 return lengthAndHash.first + headerSize();
165 RecordInfo validateRecord(ByteRange range, uint32_t fileId) {
166 if (range.size() <= headerSize()) { // records may not be empty
169 const Header* header = reinterpret_cast<const Header*>(range.begin());
170 range.advance(sizeof(Header));
171 if (header->magic != Header::kMagic ||
172 header->version != 0 ||
173 header->hashFunction != 0 ||
174 header->flags != 0 ||
175 (fileId != 0 && header->fileId != fileId) ||
176 header->dataLength > range.size()) {
179 if (headerHash(*header) != header->headerHash) {
182 range.reset(range.begin(), header->dataLength);
183 if (dataHash(range) != header->dataHash) {
186 return {header->fileId, range};
189 RecordInfo findRecord(ByteRange searchRange,
190 ByteRange wholeRange,
192 static const uint32_t magic = Header::kMagic;
193 static const ByteRange magicRange(reinterpret_cast<const uint8_t*>(&magic),
195 static constexpr size_t headerTail = sizeof(Header) - sizeof(magic);
197 DCHECK_GE(searchRange.begin(), wholeRange.begin());
198 DCHECK_LE(searchRange.end(), wholeRange.end());
200 const uint8_t* start = searchRange.begin();
201 const uint8_t* end = std::min(searchRange.end(),
202 wholeRange.end() - sizeof(Header));
203 // end-1: the last place where a Header could start
204 while (start < end) {
205 auto p = ByteRange(start, end + sizeof(magic)).find(magicRange);
206 if (p == ByteRange::npos) {
211 auto r = validateRecord(ByteRange(start, wholeRange.end()), fileId);
212 if (!r.record.empty()) {
216 // No repeated prefix in magic, so we can do better than start++
217 start += sizeof(magic);