Home | History | Annotate | Download | only in db 
      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
    104