1 enum RecordType {
 2   // Zero is reserved for preallocated files
 3   kZeroType = 0,
 4
 5   kFullType = 1,
 6
 7   // For fragments
 8   kFirstType = 2,
 9   kMiddleType = 3,
10   kLastType = 4
11 };
12 static const int kMaxRecordType = kLastType;
13
14 static const int kBlockSize = 32768;
15
16 // Header is checksum (4 bytes), type (1 byte), length (2 bytes).
17 static const int kHeaderSize = 4 + 1 + 2;
18
19 }
20 }

 1 class Writer {
 2  public:
 3   // Create a writer that will append data to "*dest".
 4   // "*dest" must be initially empty.
 5   // "*dest" must remain live while this Writer is in use.
 6   explicit Writer(WritableFile* dest);
 7   ~Writer();
 8
 9   Status AddRecord(const Slice& slice);
10
11  private:
12   WritableFile* dest_;
13   int block_offset_;       // Current offset in block
14
15   // crc32c values for all supported record types.  These are
16   // pre-computed to reduce the overhead of computing the crc of the
17   // record type stored in the header.
18   uint32_t type_crc_[kMaxRecordType + 1];
19
20   Status EmitPhysicalRecord(RecordType type, const char* ptr, size_t length);
21
22   // No copying allowed
23   Writer(const Writer&);
24   void operator=(const Writer&);
25 };

 1 Status Writer::AddRecord(const Slice& slice) {
 2   const char* ptr = slice.data();
 3   size_t left = slice.size();
 4
 5   // Fragment the record if necessary and emit it.  Note that if slice
 6   // is empty, we still want to iterate once to emit a single
 7   // zero-length record
 8   Status s;
 9   bool begin = true;
10   do {
11     const int leftover = kBlockSize - block_offset_;        //剩余要填充的数据长度 是一个BLOCK的长度减去块内已经填充的长度
12     assert(leftover >= 0);
13     if (leftover < kHeaderSize) {                            //要填充的长度大于7 则在下一个BLOCK进行记录 (因为checksum 4字节  length2字节  type 1字节，光是记录信息已经需要7个字节)
14       // Switch to a new block
15       if (leftover > 0) {
16         // Fill the trailer (literal below relies on kHeaderSize being 7)
17         assert(kHeaderSize == 7);
18         dest_->Append(Slice("\x00\x00\x00\x00\x00\x00", leftover));
19       }
20       block_offset_ = 0;
21     }
22
23     // Invariant: we never leave < kHeaderSize bytes in a block.
24     assert(kBlockSize - block_offset_ - kHeaderSize >= 0);
25
26     const size_t avail = kBlockSize - block_offset_ - kHeaderSize;
27     const size_t fragment_length = (left < avail) ? left : avail;        //根据能否在本BLOCK填充完毕 选择填充长度为left 或者 avail
28
29     RecordType type;
30     const bool end = (left == fragment_length);
31     if (begin && end) {                                //beg end在用一个BLOCK里 record的type肯定是FULL
32       type = kFullType;
33     } else if (begin) {                                //本BLOCK只有beg 那么record的type    就是FIRST
34       type = kFirstType;
35     } else if (end) {                                //本BLOCK只有end 那么record的TYPE就是last
36       type = kLastType;
37     } else {
38       type = kMiddleType;                            //本BLOCK 没有beg end  那么record填充了整个BLOCK type是MIDDLE
39     }
40
41     s = EmitPhysicalRecord(type, ptr, fragment_length);        //提交到log文件记录
42     ptr += fragment_length;
43     left -= fragment_length;
44     begin = false;
45   } while (s.ok() && left > 0);
46   return s;
47 }
48
49 Status Writer::EmitPhysicalRecord(RecordType t, const char* ptr, size_t n) {
50   assert(n <= 0xffff);  // Must fit in two bytes
51   assert(block_offset_ + kHeaderSize + n <= kBlockSize);
52
53   // Format the header
54   char buf[kHeaderSize];
55   buf[4] = static_cast<char>(n & 0xff);                //长度低8位
56   buf[5] = static_cast<char>(n >> 8);                //长度高8位
57   buf[6] = static_cast<char>(t);                    //type
58
59   // Compute the crc of the record type and the payload.
60   uint32_t crc = crc32c::Extend(type_crc_[t], ptr, n);                //校验和
61   crc = crc32c::Mask(crc);                 // Adjust for storage
62   EncodeFixed32(buf, crc);
63
64   // Write the header and the payload
65   Status s = dest_->Append(Slice(buf, kHeaderSize));        //数据信息写入
66   if (s.ok()) {
67     s = dest_->Append(Slice(ptr, n));                        //数据写入
68     if (s.ok()) {
69       s = dest_->Flush();
70     }
71   }
72   block_offset_ += kHeaderSize + n;
73   return s;
74 }

 1 class Reader {
 2  public:
 3   // Interface for reporting errors.
 4   class Reporter {
 5    public:
 6     virtual ~Reporter();
 7
 8     // Some corruption was detected.  "size" is the approximate number
 9     // of bytes dropped due to the corruption.
10     virtual void Corruption(size_t bytes, const Status& status) = 0;
11   };
12
13   // Create a reader that will return log records from "*file".
14   // "*file" must remain live while this Reader is in use.
15   //
16   // If "reporter" is non-NULL, it is notified whenever some data is
17   // dropped due to a detected corruption.  "*reporter" must remain
18   // live while this Reader is in use.
19   //
20   // If "checksum" is true, verify checksums if available.
21   //
22   // The Reader will start reading at the first record located at physical
23   // position >= initial_offset within the file.
24   Reader(SequentialFile* file, Reporter* reporter, bool checksum,
25          uint64_t initial_offset);
26
27   ~Reader();
28
29   // Read the next record into *record.  Returns true if read
30   // successfully, false if we hit end of the input.  May use
31   // "*scratch" as temporary storage.  The contents filled in *record
32   // will only be valid until the next mutating operation on this
33   // reader or the next mutation to *scratch.
34   bool ReadRecord(Slice* record, std::string* scratch);
35
36   // Returns the physical offset of the last record returned by ReadRecord.
37   //
38   // Undefined before the first call to ReadRecord.
39   uint64_t LastRecordOffset();
40
41  private:
42   SequentialFile* const file_;
43   Reporter* const reporter_;
44   bool const checksum_;
45   char* const backing_store_;
46   Slice buffer_;
47   bool eof_;   // Last Read() indicated EOF by returning < kBlockSize
48
49   // Offset of the last record returned by ReadRecord.
50   uint64_t last_record_offset_;
51   // Offset of the first location past the end of buffer_.
52   uint64_t end_of_buffer_offset_;
53
54   // Offset at which to start looking for the first record to return
55   uint64_t const initial_offset_;
56
57   // Extend record types with the following special values
58   enum {
59     kEof = kMaxRecordType + 1,
60     // Returned whenever we find an invalid physical record.
61     // Currently there are three situations in which this happens:
62     // * The record has an invalid CRC (ReadPhysicalRecord reports a drop)
63     // * The record is a 0-length record (No drop is reported)
64     // * The record is below constructor's initial_offset (No drop is reported)
65     kBadRecord = kMaxRecordType + 2
66   };
67
68   // Skips all blocks that are completely before "initial_offset_".
69   //
70   // Returns true on success. Handles reporting.
71   bool SkipToInitialBlock();
72
73   // Return type, or one of the preceding special values
74   unsigned int ReadPhysicalRecord(Slice* result);
75
76   // Reports dropped bytes to the reporter.
77   // buffer_ must be updated to remove the dropped bytes prior to invocation.
78   void ReportCorruption(size_t bytes, const char* reason);
79   void ReportDrop(size_t bytes, const Status& reason);
80
81   // No copying allowed
82   Reader(const Reader&);
83   void operator=(const Reader&);
84 };

bool Reader::ReadRecord(Slice* record, std::string* scratch) {if (last_record_offset_ < initial_offset_) {                    //实际上整个工程中initial_offset_一直为0  ,if (!SkipToInitialBlock()) {                                //block_start_location圆整为包含initial_offset_的BLOCK的偏移return false;}}scratch->clear();record->clear();bool in_fragmented_record = false;// Record offset of the logical record that we're reading// 0 is a dummy value to make compilers happyuint64_t prospective_record_offset = 0;Slice fragment;while (true) {uint64_t physical_record_offset = end_of_buffer_offset_ - buffer_.size();const unsigned int record_type = ReadPhysicalRecord(&fragment);switch (record_type) {case kFullType:                                //一次性读取FULL类型的record 直接返回成功if (in_fragmented_record) {// Handle bug in earlier versions of log::Writer where// it could emit an empty kFirstType record at the tail end// of a block followed by a kFullType or kFirstType record// at the beginning of the next block.if (scratch->empty()) {in_fragmented_record = false;} else {ReportCorruption(scratch->size(), "partial record without end(1)");}}prospective_record_offset = physical_record_offset;scratch->clear();*record = fragment;last_record_offset_ = prospective_record_offset;return true;case kFirstType:                        //读取到FIRST类型的record  string.assign  然后继续if (in_fragmented_record) {// Handle bug in earlier versions of log::Writer where// it could emit an empty kFirstType record at the tail end// of a block followed by a kFullType or kFirstType record// at the beginning of the next block.if (scratch->empty()) {in_fragmented_record = false;} else {ReportCorruption(scratch->size(), "partial record without end(2)");}}prospective_record_offset = physical_record_offset;scratch->assign(fragment.data(), fragment.size());in_fragmented_record = true;break;case kMiddleType:                            //读取到MIDDLE类型的record  string.append  然后继续if (!in_fragmented_record) {ReportCorruption(fragment.size(),"missing start of fragmented record(1)");} else {scratch->append(fragment.data(), fragment.size());}break;case kLastType:                            //读取到LAST 类型record string.appendif (!in_fragmented_record) {ReportCorruption(fragment.size(),"missing start of fragmented record(2)");} else {scratch->append(fragment.data(), fragment.size());*record = Slice(*scratch);last_record_offset_ = prospective_record_offset;return true;}break;case kEof:if (in_fragmented_record) {ReportCorruption(scratch->size(), "partial record without end(3)");scratch->clear();}return false;case kBadRecord:if (in_fragmented_record) {ReportCorruption(scratch->size(), "error in middle of record");in_fragmented_record = false;scratch->clear();}break;default: {char buf[40];snprintf(buf, sizeof(buf), "unknown record type %u", record_type);ReportCorruption((fragment.size() + (in_fragmented_record ? scratch->size() : 0)),buf);in_fragmented_record = false;scratch->clear();break;}}}return false;
}