root/trunk/whisperlib/common/io/logio/recordio.cc

// Copyright (c) 2009, Whispersoft s.r.l.
// All rights reserved.
//
// Redistribution and use in source and binary forms, with or without
// modification, are permitted provided that the following conditions are
// met:
//
// * Redistributions of source code must retain the above copyright
// notice, this list of conditions and the following disclaimer.
// * Redistributions in binary form must reproduce the above
// copyright notice, this list of conditions and the following disclaimer
// in the documentation and/or other materials provided with the
// distribution.
// * Neither the name of Whispersoft s.r.l. nor the names of its
// contributors may be used to endorse or promote products derived from
// this software without specific prior written permission.
//
// THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
// "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
// LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
// A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
// OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
// SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
// LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
// DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
// THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
// (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
// OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
//
// Author: Catalin Popescu

#include "common/io/logio/recordio.h"

namespace {
int32 ComputeCRC(io::MemoryStream* buf) {
  int32 crc = static_cast<int32>(crc32(0L, Z_NULL, 0));
  buf->MarkerSet();
  while ( !buf->IsEmpty() ) {
    int32 crt_size = buf->Size();
    const char* crt_buf = NULL;
    CHECK(buf->ReadNext(&crt_buf, &crt_size));
    crc = crc32(crc, reinterpret_cast<const Bytef*>(crt_buf), crt_size);
  }
  buf->MarkerRestore();
  return crc;
}
}

// COMPILE_ASSERT(sizeof(uLong) == sizeof(int32),  zlib_uLong_should_be_32_bit);

namespace io {

//////////////////////////////////////////////////////////////////////

RecordWriter::RecordWriter(int32 block_size,
                           bool deflate,
                           float dumpable_percent)
  : block_size_(block_size),
    dumpable_size_(static_cast<int32>(dumpable_percent * block_size)),
    content_(common::BIGENDIAN, block_size_),
    padding_(new char[block_size_]),
    zlib_(deflate ? new ZlibDeflateWrapper() : NULL),
    zlib_content_(common::BIGENDIAN, block_size_) {
  CHECK_LT(block_size_, kMaximumRecordBlockSize);
  bzero(padding_, block_size_);
}


RecordWriter::~RecordWriter() {
  CHECK(content_.IsEmpty());
  delete [] padding_;
  delete zlib_;
  zlib_ = NULL;
}

bool RecordWriter::AppendRecord(io::MemoryStream* in,
                                io::MemoryStream* out,
                                bool is_zipped) {
  bool retval = false;
  if ( zlib_ && !is_zipped ) {
    DCHECK(zlib_content_.IsEmpty());
    CHECK(zlib_->Deflate(in, &zlib_content_));
    return AppendRecord(&zlib_content_, out, true);
  }
  do {
    const int32 max_to_write = max_block_record_size() - content_.Size();
    if ( max_to_write <= 0 ) {
      retval = true;
      FinalizeContent(out);
    } else if ( max_to_write >= in->Size() ) {
      io::NumStreamer::WriteByte(&content_,
                                 is_zipped ? IS_ZIPPED : 0,
                                 common::BIGENDIAN);
      io::NumStreamer::WriteUInt24(&content_, in->Size(), common::BIGENDIAN);
      content_.AppendStream(in);
    } else if ( content_.Size() > dumpable_size_ ) {
      retval = true;
      FinalizeContent(out);
    } else {
      io::NumStreamer::WriteByte(&content_,
                                 is_zipped ? IS_ZIPPED | HAS_CONT : HAS_CONT,
                                 common::BIGENDIAN);
      io::NumStreamer::WriteUInt24(&content_, max_to_write, common::BIGENDIAN);
      content_.AppendStream(in, max_to_write);
    }
  } while ( !in->IsEmpty() );

  return retval;
}

bool RecordWriter::AppendRecord(const char* buffer, int32 size,
                                io::MemoryStream* out) {
  if ( zlib_ ) {
    CHECK(zlib_->Deflate(buffer, size, &zlib_content_));
    return AppendRecord(&zlib_content_, out, true);
  }
  bool retval = false;
  const char* p = buffer;
  do {
    const int32 max_to_write = max_block_record_size() - content_.Size();
    if ( max_to_write <= 0 ) {
      retval = true;
      FinalizeContent(out);
    } else if ( max_to_write >= size ) {
      io::NumStreamer::WriteByte(&content_, 0, common::BIGENDIAN);
      io::NumStreamer::WriteUInt24(&content_, size, common::BIGENDIAN);
      content_.Write(p, size);
      p += size;
      size = 0;
    } else if ( content_.Size() > dumpable_size_ ) {
      retval = true;
      FinalizeContent(out);
    } else {
      io::NumStreamer::WriteByte(&content_, HAS_CONT, common::BIGENDIAN);
      io::NumStreamer::WriteUInt24(&content_, max_to_write, common::BIGENDIAN);
      content_.Write(p, max_to_write);
      p += max_to_write;
      size -= max_to_write;
    }
  } while ( size > 0 );

  return retval;
}

void RecordWriter::FinalizeContent(io::MemoryStream* out) {
  if ( content_.IsEmpty() ) return;
  const int32 content_size = content_.Size();
  const int32 padding_size = block_size_ - content_size - kTrailerEnd;
  CHECK_GE(padding_size, 0) << " content: " << content_.DetailedContent();
  if ( padding_size > 0 ) {
    content_.Write(padding_, padding_size);
  }
  io::NumStreamer::WriteInt32(&content_, content_size, common::BIGENDIAN);
  io::NumStreamer::WriteInt32(&content_, ComputeCRC(&content_),
                              common::BIGENDIAN);
  out->AppendStream(&content_);

  DCHECK(content_.IsEmpty());
}

//////////////////////////////////////////////////////////////////////

RecordReader::RecordReader(int32 block_size)
  : block_size_(block_size),
    temp_(common::BIGENDIAN, block_size_),
    content_(common::BIGENDIAN, block_size_),
    zip_content_(common::BIGENDIAN, block_size_) {
}

RecordReader::~RecordReader() {
}

RecordReader::ReadResult RecordReader::ReadNextBlock(io::MemoryStream* in) {
  if ( in->Size() < block_size_ ) {
    return READ_NO_DATA;
  }
  temp_.Clear();
  temp_.AppendStream(in, block_size_ - kTrailerEnd);
  const int32 content_size = io::NumStreamer::ReadInt32(in, common::BIGENDIAN);
  io::NumStreamer::WriteInt32(&temp_, content_size, common::BIGENDIAN);
  const int32 crc = io::NumStreamer::ReadInt32(in, common::BIGENDIAN);
  if ( ComputeCRC(&temp_) != crc ) {
    temp_.Clear();
    return READ_CRC_CORRUPTED;
  }
  CHECK_LE(content_size, block_size_ - kTrailerEnd);
  content_.AppendStream(&temp_, content_size);
  return READ_OK;
}

RecordReader::ReadResult RecordReader::ReadRecord(io::MemoryStream* in,
                                                  io::MemoryStream* out) {
  if ( content_.IsEmpty() ) {
    RecordReader::ReadResult ret = ReadNextBlock(in);
    if ( ret != READ_OK )
      return ret;
  }
  // At this point we have verified data in content_.
  uint8 flags = 0;
  int zip_error = Z_STREAM_END;
  bool unzip = false;
  zlib_.Clear();

  do {
    // TODO(cpopescu): probably this meas a bug somewhere at reading or writing
    //                 shall we mark a corruption and live ?
    CHECK_GE(content_.Size(), sizeof(int32));
    flags = io::NumStreamer::ReadByte(&content_,
                                      common::BIGENDIAN);
    int32 crt_len = io::NumStreamer::ReadUInt24(&content_,
                                                common::BIGENDIAN);
    if ( crt_len > 0 ) {
      if ( out ) {
        if ( !(flags & RecordWriter::IS_ZIPPED) ) {
          out->AppendStream(&content_, crt_len);
        } else {
          zip_content_.AppendStream(&content_, crt_len);
          unzip = true;
        }
      } else {
        content_.Skip(crt_len);
      }
    }
    if ( (flags & RecordWriter::HAS_CONT) && content_.IsEmpty() ) {
      RecordReader::ReadResult ret = ReadNextBlock(in);
      if ( ret != READ_OK )
        return ret;
    }
  } while ( flags & RecordWriter::HAS_CONT );

  if ( !zip_content_.IsEmpty() ) {
    if ( out ) {
      zip_error = zlib_.Inflate(&zip_content_, out);
      if ( zip_error != Z_STREAM_END ) {
        zip_content_.Clear();
        return READ_ZIP_CORRUPTED;
      }
    }
    zip_content_.Clear();
  }
  return READ_OK;
}
}