1 // Copyright (c) 2011 The LevelDB Authors. All rights reserved.
2 // Use of this source code is governed by a BSD-style license that can be
3 // found in the LICENSE file. See the AUTHORS file for names of contributors.
4
5 #include "db/log_writer.h"
6
7 #include <stdint.h>
8 #include "leveldb/env.h"
9 #include "util/coding.h"
10 #include "util/crc32c.h"
11
12 namespace leveldb {
13 namespace log {
14
Writer(WritableFile * dest)15 Writer::Writer(WritableFile* dest)
16 : dest_(dest),
17 block_offset_(0) {
18 for (int i = 0; i <= kMaxRecordType; i++) {
19 char t = static_cast<char>(i);
20 type_crc_[i] = crc32c::Value(&t, 1);
21 }
22 }
23
~Writer()24 Writer::~Writer() {
25 }
26
AddRecord(const Slice & slice)27 Status Writer::AddRecord(const Slice& slice) {
28 const char* ptr = slice.data();
29 size_t left = slice.size();
30
31 // Fragment the record if necessary and emit it. Note that if slice
32 // is empty, we still want to iterate once to emit a single
33 // zero-length record
34 Status s;
35 bool begin = true;
36 do {
37 const int leftover = kBlockSize - block_offset_;
38 assert(leftover >= 0);
39 if (leftover < kHeaderSize) {
40 // Switch to a new block
41 if (leftover > 0) {
42 // Fill the trailer (literal below relies on kHeaderSize being 7)
43 assert(kHeaderSize == 7);
44 dest_->Append(Slice("\x00\x00\x00\x00\x00\x00", leftover));
45 }
46 block_offset_ = 0;
47 }
48
49 // Invariant: we never leave < kHeaderSize bytes in a block.
50 assert(kBlockSize - block_offset_ - kHeaderSize >= 0);
51
52 const size_t avail = kBlockSize - block_offset_ - kHeaderSize;
53 const size_t fragment_length = (left < avail) ? left : avail;
54
55 RecordType type;
56 const bool end = (left == fragment_length);
57 if (begin && end) {
58 type = kFullType;
59 } else if (begin) {
60 type = kFirstType;
61 } else if (end) {
62 type = kLastType;
63 } else {
64 type = kMiddleType;
65 }
66
67 s = EmitPhysicalRecord(type, ptr, fragment_length);
68 ptr += fragment_length;
69 left -= fragment_length;
70 begin = false;
71 } while (s.ok() && left > 0);
72 return s;
73 }
74
EmitPhysicalRecord(RecordType t,const char * ptr,size_t n)75 Status Writer::EmitPhysicalRecord(RecordType t, const char* ptr, size_t n) {
76 assert(n <= 0xffff); // Must fit in two bytes
77 assert(block_offset_ + kHeaderSize + n <= kBlockSize);
78
79 // Format the header
80 char buf[kHeaderSize];
81 buf[4] = static_cast<char>(n & 0xff);
82 buf[5] = static_cast<char>(n >> 8);
83 buf[6] = static_cast<char>(t);
84
85 // Compute the crc of the record type and the payload.
86 uint32_t crc = crc32c::Extend(type_crc_[t], ptr, n);
87 crc = crc32c::Mask(crc); // Adjust for storage
88 EncodeFixed32(buf, crc);
89
90 // Write the header and the payload
91 Status s = dest_->Append(Slice(buf, kHeaderSize));
92 if (s.ok()) {
93 s = dest_->Append(Slice(ptr, n));
94 if (s.ok()) {
95 s = dest_->Flush();
96 }
97 }
98 block_offset_ += kHeaderSize + n;
99 return s;
100 }
101
102 } // namespace log
103 } // namespace leveldb
104