Line data    Source code

       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             : 
      15         617 : Writer::Writer(WritableFile* dest)
      16             :     : dest_(dest),
      17         617 :       block_offset_(0) {
      18        3702 :   for (int i = 0; i <= kMaxRecordType; i++) {
      19        3085 :     char t = static_cast<char>(i);
      20        3085 :     type_crc_[i] = crc32c::Value(&t, 1);
      21             :   }
      22         617 : }
      23             : 
      24         617 : Writer::~Writer() {
      25         617 : }
      26             : 
      27        1175 : Status Writer::AddRecord(const Slice& slice) {
      28        1175 :   const char* ptr = slice.data();
      29        1175 :   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        1175 :   bool begin = true;
      36        1207 :   do {
      37        1207 :     const int leftover = kBlockSize - block_offset_;
      38        1207 :     assert(leftover >= 0);
      39        1207 :     if (leftover < kHeaderSize) {
      40             :       // Switch to a new block
      41          32 :       if (leftover > 0) {
      42             :         // Fill the trailer (literal below relies on kHeaderSize being 7)
      43             :         assert(kHeaderSize == 7);
      44           0 :         dest_->Append(Slice("\x00\x00\x00\x00\x00\x00", leftover));
      45             :       }
      46          32 :       block_offset_ = 0;
      47             :     }
      48             : 
      49             :     // Invariant: we never leave < kHeaderSize bytes in a block.
      50        1207 :     assert(kBlockSize - block_offset_ - kHeaderSize >= 0);
      51             : 
      52        1207 :     const size_t avail = kBlockSize - block_offset_ - kHeaderSize;
      53        1207 :     const size_t fragment_length = (left < avail) ? left : avail;
      54             : 
      55             :     RecordType type;
      56        1207 :     const bool end = (left == fragment_length);
      57        1207 :     if (begin && end) {
      58             :       type = kFullType;
      59          33 :     } else if (begin) {
      60             :       type = kFirstType;
      61          32 :     } else if (end) {
      62             :       type = kLastType;
      63             :     } else {
      64          31 :       type = kMiddleType;
      65             :     }
      66             : 
      67        2414 :     s = EmitPhysicalRecord(type, ptr, fragment_length);
      68        1207 :     ptr += fragment_length;
      69        1207 :     left -= fragment_length;
      70        1207 :     begin = false;
      71        1207 :   } while (s.ok() && left > 0);
      72        1175 :   return s;
      73             : }
      74             : 
      75        1207 : Status Writer::EmitPhysicalRecord(RecordType t, const char* ptr, size_t n) {
      76        1207 :   assert(n <= 0xffff);  // Must fit in two bytes
      77        1207 :   assert(block_offset_ + kHeaderSize + n <= kBlockSize);
      78             : 
      79             :   // Format the header
      80             :   char buf[kHeaderSize];
      81        1207 :   buf[4] = static_cast<char>(n & 0xff);
      82        1207 :   buf[5] = static_cast<char>(n >> 8);
      83        1207 :   buf[6] = static_cast<char>(t);
      84             : 
      85             :   // Compute the crc of the record type and the payload.
      86        1207 :   uint32_t crc = crc32c::Extend(type_crc_[t], ptr, n);
      87        1207 :   crc = crc32c::Mask(crc);                 // Adjust for storage
      88        1207 :   EncodeFixed32(buf, crc);
      89             : 
      90             :   // Write the header and the payload
      91        2414 :   Status s = dest_->Append(Slice(buf, kHeaderSize));
      92        1207 :   if (s.ok()) {
      93        3621 :     s = dest_->Append(Slice(ptr, n));
      94        1207 :     if (s.ok()) {
      95        2414 :       s = dest_->Flush();
      96             :     }
      97             :   }
      98        1207 :   block_offset_ += kHeaderSize + n;
      99        1207 :   return s;
     100             : }
     101             : 
     102             : }  // namespace log
     103             : }  // namespace leveldb