root/trunk/whisperlib/common/io/logio/logio.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 <unistd.h>
#include <sys/stat.h>
#include <fcntl.h>
#include "common/io/logio/logio.h"
#include "common/base/strutil.h"
#include "common/io/ioutil.h"
#include "common/base/log.h"

namespace {
//////////////////////////////////////////////////////////////////////

const string ComposeFileName(const char* log_dir,
                             const char* file_base,
                             int32 block_size,
                             int32 file_num) {
  return strutil::NormalizePath(
      strutil::StringPrintf("%s/%s_%010d_%010d",
                            log_dir,
                            file_base,
                            static_cast<int32>(block_size),
                            static_cast<int32>(file_num)));
}
}

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

namespace io {

LogWriter::LogWriter(const char* log_dir,
                     const char* file_base,
                     bool sync_on_write,
                     int32 block_size,
                     int32 blocks_per_file,
                     bool temporary_incomplete_file,
                     bool deflate)
  : log_dir_(log_dir),
    file_base_(file_base),
    sync_on_write_(sync_on_write),
    re_(string("^") + string(file_base_) +
        strutil::StringPrintf("_%010d", static_cast<int32>(block_size)) +
        "_[0-9][0-9][0-9][0-9][0-9][0-9][0-9][0-9][0-9][0-9]$"),
    max_file_size_(block_size * blocks_per_file),
    block_size_(block_size),
    blocks_per_file_(blocks_per_file),
    temporary_incomplete_file_(temporary_incomplete_file),
    deflate_(deflate),
    file_num_(-1),
    fd_(INVALID_FD_VALUE),
    recorder_(block_size, deflate) {
}
LogWriter::~LogWriter() {
  Flush();
  CloseLog();
}


bool LogWriter::ProcessWriteRecord(bool result, bool force_flush) {
  const int32 leftover = recorder_.leftover();
  if ( result ) {
    if ( leftover == 0 || buf_.Size() > block_size_ ) {
      if ( buf_.Size() % block_size_ != 0 ) {
        recorder_.FinalizeContent(&buf_);
      }
      return WriteBuffer(force_flush);
    }
  }
  if ( force_flush ) {
    recorder_.FinalizeContent(&buf_);
    WriteBuffer(force_flush);
  }
  return true;
}

bool LogWriter::WriteRecord(io::MemoryStream* in, bool force_flush) {
  return ProcessWriteRecord(recorder_.AppendRecord(in, &buf_),
                            force_flush);
}

bool LogWriter::WriteRecord(const char* buffer, int32 size, bool force_flush) {
  return ProcessWriteRecord(recorder_.AppendRecord(buffer, size, &buf_),
                            force_flush);
}

bool LogWriter::Flush() {
  recorder_.FinalizeContent(&buf_);
  if ( buf_.IsEmpty() ) {
    return true;
  }
  return WriteBuffer(true);
}

string LogWriter::ComposeFileName(bool temp) const {
  if ( file_num_ < 0 ) return "";
  const string normal_file_name = ::ComposeFileName(
      log_dir_.c_str(), file_base_.c_str(), block_size_, file_num_);
  if ( !temp || io::Exists(normal_file_name) ) {
    return normal_file_name;
  }
  return ::ComposeFileName(strutil::JoinPaths(log_dir_, "temp").c_str(),
                           file_base_.c_str(), block_size_, file_num_);
}

LogPos LogWriter::Tell() const {
  if ( fd_ == INVALID_FD_VALUE ) {
    return LogPos();
  }
  int64 pos = ::lseek64(fd_, 0, SEEK_END);
  if ( pos < 0 ) {
    return LogPos(file_num_, 0, 0);
  }
  return LogPos(file_num_, pos / block_size_ + 1, 0);
}

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

bool LogWriter::WriteBuffer(bool force_flush) {
  if ( fd_ == INVALID_FD_VALUE ) {
    if ( !OpenLog() ) return false;
  }
  int64 pos = 0;
  do {
    pos = ::lseek64(fd_, 0, SEEK_END);
    if ( pos < 0 ) {
      LOG_ERROR << " Error in lseek64: " << errno
                << " - " << ::GetLastSystemErrorDescription();
      return false;
    }
    if ( pos >= max_file_size_ ) {
      LOG_INFO << "Over the position: "
               << file_name_ << " pos: "
               << pos << " max_file_size_: " << max_file_size_;
      if ( !OpenLog() ) return false;
      continue;
    }
  } while ( pos >= max_file_size_ );
  if ( pos % block_size_ != 0 ) {
    if ( !NormalizeFileSize(pos) ) {
      return false;
    }
  }
  string content;
  buf_.MarkerSet();
  buf_.ReadString(&content);
  CHECK_EQ(content.size() % block_size_, 0)
    << " - content.size(): " << content.size()
    << " - block_size_: " << block_size_;

  // We are in a good position - do the write
  const int cb = ::write(fd_, content.data(), content.size());
  if ( cb < 0 ) {
    buf_.MarkerRestore();
    LOG_ERROR << " Error writing the log" << errno
              << " - " << ::GetLastSystemErrorDescription();
    return false;
  }
  if ( content.size() != cb ) {
    buf_.MarkerRestore();
    LOG_ERROR << " Trouble in log write - succeded only" << cb
              << " out of " << content.size();
    return false;
  }
  buf_.MarkerClear();
  if ( force_flush ) {
    fdatasync(fd_);
  }
  return true;
}

bool LogWriter::OpenLog() {
  if ( fd_ == INVALID_FD_VALUE ) {
    file_num_ = io::GetLastNumberedFile(log_dir_, &re_, 10);
    if ( file_num_ < -1 ) {
      return false;
    }
    if ( file_num_  < 0 ) {
      file_num_ = 0;
    }
    if ( temporary_incomplete_file_ ) {
      if ( !io::Mkdir(strutil::JoinPaths(log_dir_, "temp")) ) {
        LOG_ERROR << "Failed to create temporary directory";
        return false;
      }
    }
  } else {
    CloseLog();
    CHECK_GE(file_num_, 0);
    ++file_num_;
  }
  const string filename(ComposeFileName(temporary_incomplete_file_));
  DLOG_INFO << "New log file: " << filename;
  fd_ = ::open(filename.c_str(),
               (sync_on_write_ ? O_SYNC : 0) |
               O_WRONLY | O_APPEND | O_CREAT | O_LARGEFILE,
               0644);
  if ( fd_ < 0 ) {
    LOG_ERROR << " Error opening log writer for file : "<< filename
              << " err: " << errno << " - "
              << ::GetLastSystemErrorDescription();
    fd_ = INVALID_FD_VALUE;
    file_num_ = -1;
    file_name_ = "";
    return false;
  }
  file_name_ = filename;
  return true;
}
void LogWriter::CloseLog() {
  if ( fd_ == INVALID_FD_VALUE ) {
    return;
  }
  ::close(fd_);
  fd_ = INVALID_FD_VALUE;
  if ( temporary_incomplete_file_ ) {
    const string final_file_name = ComposeFileName(false);
    if ( final_file_name != file_name_ ) {
      if ( !io::Rename(file_name_, final_file_name, false) ) {
        LOG_ERROR << "Error renaming temporary file: [" << file_name_ << "]"
                     " to: [" << final_file_name << "]";
      }
    }
  }
  file_name_ = "";
}

bool LogWriter::NormalizeFileSize(int64 pos) {
  const int32 delta = pos % block_size_;
  if ( delta == 0 ) return true;
  CHECK_NE(fd_, INVALID_FD_VALUE);

  const int fd_copy = ::open(file_name_.c_str(),
                             O_WRONLY | O_SYNC | O_LARGEFILE,
                             0644);
  if ( fd_copy < 0 ) {
    LOG_ERROR << " Error opening a file for normalization: "<< file_name_
              << " err: " << errno << " - "
              << ::GetLastSystemErrorDescription();
    return false;
  }
  LOG_WARNING << " Normalizing a file: " << file_name_;
  if ( ::lseek64(fd_copy, pos - delta, SEEK_SET) < 0 ) {
    LOG_ERROR << " Error seeking while normalizing a file: "<< file_name_
              << " err: " << errno << " - "
              << ::GetLastSystemErrorDescription();
    ::close(fd_copy);
    return false;
  }
  if ( ::lockf(fd_copy, F_LOCK, block_size_) ) {
    LOG_ERROR << " Error locking while normalizing a file: "<< file_name_
              << " err: " << errno << " - "
              << ::GetLastSystemErrorDescription();
    return false;
  }
  const int64 locked_pos = ::lseek64(fd_copy, 0, SEEK_END);
  if ( locked_pos < 0 ) {
    LOG_ERROR << "Everything got screwed. Cannot recover .."
              << " err: " << errno << " - "
              << ::GetLastSystemErrorDescription();
    ::close(fd_copy);   // unlocks
    return false;
  }
  const int32 locked_delta = locked_pos % block_size_;
  bool success = false;
  if ( locked_delta == 0 ) {
    LOG_INFO << " Already normalized !!";
    success = true;
  } else if ( locked_pos != pos ) {
    // This is really strange ... btw
    LOG_WARNING << "Amazingly strange situation: pos: " << pos
               << " locked_pos: " << locked_pos
               << " filename: " << file_name_;
    success = NormalizeFileSize(locked_pos);
  } else {
    if ( ::lseek64(fd_copy, pos - delta, SEEK_SET) < 0 ) {
      LOG_ERROR << " Error re-seeking while normalizing a file: "<< file_name_
                << " err: " << errno << " - "
                << ::GetLastSystemErrorDescription();
      ::close(fd_copy);  // unlocks
      return false;
    }
    if ( ::lseek64(fd_copy, delta, SEEK_CUR) > 0 ) {
      char* const buf = new char[block_size_ - delta];
      if ( write(fd_copy, buf, block_size_ - delta) != block_size_ - delta ) {
        LOG_ERROR << " Error writing while normalizing a file: "<< file_name_
                  << " err: " << errno << " - "
                  << ::GetLastSystemErrorDescription();
      } else {
        LOG_INFO << " Normalized file: " << file_name_
                 << "@: " << pos << " w/ " << block_size_ - delta << " bytes.";
        success = true;
      }
    } else {
      LOG_ERROR << " Error seeking while preparing to normalize a file: "
                << file_name_ << " err: " << errno << " - "
                << ::GetLastSystemErrorDescription();
    }
  }
  if ( ::lseek64(fd_copy, pos - delta, SEEK_SET) < 0 ) {
    LOG_ERROR << " Error seeking after a file normalization: "<< file_name_
              << " err: " << errno << " - "
              << ::GetLastSystemErrorDescription()
              << " - we must fail (as the file is locked..";
  } else {
    if ( ::lockf(fd_copy, F_ULOCK, block_size_) ) {
      LOG_ERROR << " Error unlocking after normalization: "<< file_name_
                << " err: " << errno << " - "
                << ::GetLastSystemErrorDescription()
                << " - we must fail (as the file is locked..";
    }
  }
  ::close(fd_copy);   // this should unlock everything anyway
  return success;
}

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


LogReader::LogReader(const char* log_dir,
                     const char* file_base,
                     int32 block_size,
                     int32 blocks_per_file)
  : log_dir_(log_dir),
    file_base_(file_base),
    re_(string("^") + string(file_base_) +
        strutil::StringPrintf("_%010d", static_cast<int32>(block_size)) +
        "_[0-9][0-9][0-9][0-9][0-9][0-9][0-9][0-9][0-9][0-9]$"),
    block_size_(block_size),
    blocks_per_file_(blocks_per_file),
    fd_(INVALID_FD_VALUE),
    crt_pos_(),
    last_read_pos_(),
    reader_(block_size),
    num_errors_(0),
    last_errno_(0) {
}

LogReader::~LogReader() {
  Close();
}

bool LogReader::Seek(const LogPos& log_pos) {
  last_errno_ = 0;
  if ( log_pos.IsNull() ) {
    return Rewind();
  }
  reader_.Clear();
  buf_.Clear();
  if ( !OpenFile(log_pos.file_num_) ) {
    return false;
  }
  return AdvanceToPosInFile(log_pos);
}

bool LogReader::GetNextRecord(io::MemoryStream* out) {
  last_errno_ = 0;
  if ( fd_ == INVALID_FD_VALUE ) {
    if ( !Rewind() ) {
      return false;
    }
  }
  while ( true ) {
    CHECK(!crt_pos_.IsNull());
    RecordReader::ReadResult res = reader_.ReadRecord(&buf_, out);
    if ( res == RecordReader::READ_NO_DATA ) {
      // LOG_ERROR << " No more data=========:";
      if ( !ReadNextBlock() ) {
        return false;
      }
    } else {
      if ( res == RecordReader::READ_OK ) {
        crt_pos_.record_num_++;
        last_read_pos_ = crt_pos_;
        return true;
      }
      LOG_ERROR << " Log error: CRC in [" << file_name_
                << " @: " << crt_pos_.ToString();
      buf_.Clear();
      reader_.Clear();  // force the read of the next block
      num_errors_++;
    }
  }
  return false;
}

bool LogReader::Rewind() {
  CloseInternal(false);
  vector<string> files;
  // TODO(cpopescu): well .. this is almost there ..
  //                 as we don't count in the block_size
  if ( !DirList(log_dir_ + "/", &files, false, &re_) ) {
    LOG_WARNING << " No file to open for reader: "
                << log_dir_ << "/" << file_base_ << "_x_x";
    return false;
  }
  if ( files.empty() ) {
    // Not an error though ..
    LOG_INFO << " No log file to read found .. "
             << log_dir_ << "/" << file_base_ << "_x_x";
    return false;
  }
  sort(files.begin(), files.end());

  CHECK_GT(files.front().size(), 10);
  errno = 0;  // essential as strtol would not set a 0 errno
  const int32 file_num = strtol(
    files.front().c_str() + files.front().size() - 10, NULL, 10);
  if ( errno || file_num < 0 ) {
    LOG_ERROR << " Invalid file found: " << files.back();
    return false;
  }
  return OpenFile(file_num);
}

string LogReader::ComposeFileName(int32 file_num) const {
  if ( file_num < 0 ) return "";
  return ::ComposeFileName(log_dir_.c_str(), file_base_.c_str(),
                           block_size_, file_num);
}

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

void LogReader::CloseInternal(bool on_error) {
  if ( fd_ != INVALID_FD_VALUE ) {
    ::close(fd_);
  }
  fd_ = INVALID_FD_VALUE;
  file_name_ = "";
  crt_pos_ = LogPos();
  last_read_pos_ = LogPos();
  if ( on_error ) {
    last_errno_ = errno;
    num_errors_++;
  }
  reader_.Clear();
  buf_.Clear();
}

bool LogReader::OpenFile(int32 file_num) {
  if ( fd_ != INVALID_FD_VALUE ) {
    if ( file_num == crt_pos_.file_num_ ) {
      if ( ::lseek64(fd_, 0, SEEK_SET) < 0 ) {
        CloseInternal(true);
        return false;
      }
      return true;
    }
    CloseInternal(false);
  }
  const string filename = ComposeFileName(file_num);
  fd_ = ::open(filename.c_str(), O_RDONLY | O_LARGEFILE);
  if ( fd_ < 0 ) {
    LOG_ERROR << " Error opening log reader for file : "<< filename
              << " err: " << errno << " - "
              << ::GetLastSystemErrorDescription();
    CloseInternal(true);
    return false;
  }
  crt_pos_ = LogPos(file_num, 0, 0);
  last_read_pos_ = crt_pos_;
  file_name_ = filename;
  return true;
}

bool LogReader::AdvanceToPosInFile(const LogPos& log_pos) {
  CHECK_EQ(log_pos.file_num_, crt_pos_.file_num_);
  CHECK_NE(fd_, INVALID_FD_VALUE);
  crt_pos_.block_num_ = log_pos.block_num_ - 1;
  crt_pos_.record_num_ = 0;


  if ( ::lseek64(fd_, crt_pos_.block_num_ * block_size_, SEEK_SET) < 0 ) {
    CloseInternal(true);
    return false;
  }
  reader_.Clear();
  buf_.Clear();
  crt_pos_.record_num_ = 0;
  if ( !ReadNextBlock() ) {
    if ( log_pos.record_num_ > 0 ) {
      CloseInternal(true);
      return false;
    }
    if ( crt_pos_.block_num_ < 0 ) {
      crt_pos_.block_num_ = 0;
    }
    last_read_pos_ = crt_pos_;
    return true;
  }
  last_read_pos_ = crt_pos_;
  while ( crt_pos_.record_num_ < log_pos.record_num_ ) {
    RecordReader::ReadResult res = reader_.ReadRecord(&buf_, NULL);
    if ( res != RecordReader::READ_OK ) {
      CloseInternal(true);
      return false;
    }
    ++crt_pos_.record_num_;
  }
  last_read_pos_ = crt_pos_;
  return true;
}

bool LogReader::ReadNextBlock() {
  CHECK_NE(fd_, INVALID_FD_VALUE);

  char* buffer = new char[block_size_];
  const int32 cb = ::read(fd_, buffer, block_size_);
  if ( cb < 0 ) {
    delete [] buffer;
    LOG_ERROR << " Error reading: " << file_name_
              << " errno: " << errno
              << " - " << ::GetLastSystemErrorDescription();
    CloseInternal(true);
    return false;
  }

  // Incomlete read..
  if ( cb < block_size_ ) {
    delete [] buffer;
    // It is a possibility to be in the new file !
    if ( crt_pos_.block_num_ >= blocks_per_file_ ) {
      const string filename = ComposeFileName(crt_pos_.file_num_ + 1);
      const int fd = ::open(filename.c_str(), O_RDONLY | O_LARGEFILE);
      if ( fd < 0 ) {
        // A new file not available yet..
        return false;
      }
      // Cool - advance to the new file
      ::close(fd_);
      fd_ = fd;
      crt_pos_ = LogPos(crt_pos_.file_num_ + 1, 0, 0);
      file_name_ = filename;
      return ReadNextBlock();
    }
    if ( cb > 0 ) {
      // Strange - but possible.. we need to revert
      LOG_WARNING << "Strange block read from : " << file_name_
                  << " size: " << cb
                  << " @: " << crt_pos_.ToString();
      if ( ::lseek64(fd_, -cb, SEEK_CUR) < 0 ) {
        LOG_ERROR << " Error seeking back: " << file_name_
                  << " errno: " << errno
                  << " - " << ::GetLastSystemErrorDescription();
        CloseInternal(true);
      }
    }
    return false;
  }
  crt_pos_.record_num_ = 0;
  crt_pos_.block_num_++;

  buf_.AppendRaw(buffer, block_size_);

  return true;
}

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

int32 CleanLog(const char* log_dir, const char* file_base, LogPos first_pos,
               int32 block_size) {
  re::RE re(string("^") + string(file_base) +
            strutil::StringPrintf("_%010d", static_cast<int32>(block_size)) +
            "_[0-9][0-9][0-9][0-9][0-9][0-9][0-9][0-9][0-9][0-9]$");
  vector<string> files;
  if ( !DirList(string(log_dir) + "/", &files, false, &re) ) {
    return 0;
  }
  if ( files.empty() ) {
    return 0;
  }
  sort(files.begin(), files.end());

  int32 num_deleted = 0;
  for ( int32 i = 0; i < files.size(); ++i ) {
    CHECK_GT(files[i].size(), 10);
    errno = 0;  // essential as strtol would not set a 0 errno
    const int32 file_num = strtol(files[i].c_str() + files[i].size() - 10,
                                  NULL, 10);
    if ( errno || file_num < 0 ) {
      LOG_ERROR << " FileNum : " << file_num <<  " for " << files[i];
      continue;
    }
    if ( file_num >= first_pos.file_num_ ) {
      return num_deleted;
    }
    const string filename = ComposeFileName(log_dir, file_base,
                                            block_size, file_num);
    if ( unlink(filename.c_str()) ) {
      LOG_ERROR << "Error removing file: " << filename << " : " << errno
                << " - " << ::GetLastSystemErrorDescription();
    } else {
      ++num_deleted;
    }
  }
  return num_deleted;
}

//////////////////////////////////////////////////////////////////////
}