src/leveldb/db/log_writer.cc

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