1 //  Copyright (c) 2011-present, Facebook, Inc.  All rights reserved.
2 //  This source code is licensed under both the GPLv2 (found in the
3 //  COPYING file in the root directory) and Apache 2.0 License
4 //  (found in the LICENSE.Apache file in the root directory).
5 //
6 // Copyright (c) 2011 The LevelDB Authors. All rights reserved.
7 // Use of this source code is governed by a BSD-style license that can be
8 // found in the LICENSE file. See the AUTHORS file for names of contributors.
9 
10 #include "db/log_writer.h"
11 
12 #include <stdint.h>
13 #include "file/writable_file_writer.h"
14 #include "rocksdb/env.h"
15 #include "util/coding.h"
16 #include "util/crc32c.h"
17 
18 namespace ROCKSDB_NAMESPACE {
19 namespace log {
20 
Writer(std::unique_ptr<WritableFileWriter> && dest,uint64_t log_number,bool recycle_log_files,bool manual_flush)21 Writer::Writer(std::unique_ptr<WritableFileWriter>&& dest, uint64_t log_number,
22                bool recycle_log_files, bool manual_flush)
23     : dest_(std::move(dest)),
24       block_offset_(0),
25       log_number_(log_number),
26       recycle_log_files_(recycle_log_files),
27       manual_flush_(manual_flush) {
28   for (int i = 0; i <= kMaxRecordType; i++) {
29     char t = static_cast<char>(i);
30     type_crc_[i] = crc32c::Value(&t, 1);
31   }
32 }
33 
~Writer()34 Writer::~Writer() {
35   if (dest_) {
36     WriteBuffer();
37   }
38 }
39 
WriteBuffer()40 IOStatus Writer::WriteBuffer() { return dest_->Flush(); }
41 
Close()42 IOStatus Writer::Close() {
43   IOStatus s;
44   if (dest_) {
45     s = dest_->Close();
46     dest_.reset();
47   }
48   return s;
49 }
50 
AddRecord(const Slice & slice)51 IOStatus Writer::AddRecord(const Slice& slice) {
52   const char* ptr = slice.data();
53   size_t left = slice.size();
54 
55   // Header size varies depending on whether we are recycling or not.
56   const int header_size =
57       recycle_log_files_ ? kRecyclableHeaderSize : kHeaderSize;
58 
59   // Fragment the record if necessary and emit it.  Note that if slice
60   // is empty, we still want to iterate once to emit a single
61   // zero-length record
62   IOStatus s;
63   bool begin = true;
64   do {
65     const int64_t leftover = kBlockSize - block_offset_;
66     assert(leftover >= 0);
67     if (leftover < header_size) {
68       // Switch to a new block
69       if (leftover > 0) {
70         // Fill the trailer (literal below relies on kHeaderSize and
71         // kRecyclableHeaderSize being <= 11)
72         assert(header_size <= 11);
73         s = dest_->Append(Slice("\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00",
74                                 static_cast<size_t>(leftover)));
75         if (!s.ok()) {
76           break;
77         }
78       }
79       block_offset_ = 0;
80     }
81 
82     // Invariant: we never leave < header_size bytes in a block.
83     assert(static_cast<int64_t>(kBlockSize - block_offset_) >= header_size);
84 
85     const size_t avail = kBlockSize - block_offset_ - header_size;
86     const size_t fragment_length = (left < avail) ? left : avail;
87 
88     RecordType type;
89     const bool end = (left == fragment_length);
90     if (begin && end) {
91       type = recycle_log_files_ ? kRecyclableFullType : kFullType;
92     } else if (begin) {
93       type = recycle_log_files_ ? kRecyclableFirstType : kFirstType;
94     } else if (end) {
95       type = recycle_log_files_ ? kRecyclableLastType : kLastType;
96     } else {
97       type = recycle_log_files_ ? kRecyclableMiddleType : kMiddleType;
98     }
99 
100     s = EmitPhysicalRecord(type, ptr, fragment_length);
101     ptr += fragment_length;
102     left -= fragment_length;
103     begin = false;
104   } while (s.ok() && left > 0);
105 
106   if (s.ok()) {
107     if (!manual_flush_) {
108       s = dest_->Flush();
109     }
110   }
111 
112   return s;
113 }
114 
TEST_BufferIsEmpty()115 bool Writer::TEST_BufferIsEmpty() { return dest_->TEST_BufferIsEmpty(); }
116 
EmitPhysicalRecord(RecordType t,const char * ptr,size_t n)117 IOStatus Writer::EmitPhysicalRecord(RecordType t, const char* ptr, size_t n) {
118   assert(n <= 0xffff);  // Must fit in two bytes
119 
120   size_t header_size;
121   char buf[kRecyclableHeaderSize];
122 
123   // Format the header
124   buf[4] = static_cast<char>(n & 0xff);
125   buf[5] = static_cast<char>(n >> 8);
126   buf[6] = static_cast<char>(t);
127 
128   uint32_t crc = type_crc_[t];
129   if (t < kRecyclableFullType) {
130     // Legacy record format
131     assert(block_offset_ + kHeaderSize + n <= kBlockSize);
132     header_size = kHeaderSize;
133   } else {
134     // Recyclable record format
135     assert(block_offset_ + kRecyclableHeaderSize + n <= kBlockSize);
136     header_size = kRecyclableHeaderSize;
137 
138     // Only encode low 32-bits of the 64-bit log number.  This means
139     // we will fail to detect an old record if we recycled a log from
140     // ~4 billion logs ago, but that is effectively impossible, and
141     // even if it were we'dbe far more likely to see a false positive
142     // on the 32-bit CRC.
143     EncodeFixed32(buf + 7, static_cast<uint32_t>(log_number_));
144     crc = crc32c::Extend(crc, buf + 7, 4);
145   }
146 
147   // Compute the crc of the record type and the payload.
148   crc = crc32c::Extend(crc, ptr, n);
149   crc = crc32c::Mask(crc);  // Adjust for storage
150   EncodeFixed32(buf, crc);
151 
152   // Write the header and the payload
153   IOStatus s = dest_->Append(Slice(buf, header_size));
154   if (s.ok()) {
155     s = dest_->Append(Slice(ptr, n));
156   }
157   block_offset_ += header_size + n;
158   return s;
159 }
160 
161 }  // namespace log
162 }  // namespace ROCKSDB_NAMESPACE
163