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 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
  24 Writer::~Writer() {
  25 }
  26
  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
  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