log_writer.cc 4.8 KB

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