root/trunk/whisperlib/common/io/logio/logio.h

// 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

// Logging interface - one for write and one for read .
//
// Logs are composed from files of (approximately) fixed length,
// sequentialy numbered. These file are composed from fixed sized
// blocks of variable length records.
//
// The position in a log is a tripet: (file_num, block_num, record_num)
//

#ifndef __COMMON_IO_LOGIO_LOGIO_H__
#define __COMMON_IO_LOGIO_LOGIO_H__

#include <string>
#include <whisperlib/common/base/types.h>
#include <whisperlib/common/io/buffer/memory_stream.h>
#include <whisperlib/common/io/logio/recordio.h>
#include <whisperlib/common/base/re.h>

namespace io {

static const int32 kDefaultBlocksPerFile = 1 << 14;

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

struct LogPos {
  int32 file_num_;
  int32 block_num_;
  int32 record_num_;
  LogPos()
    : file_num_(-1), block_num_(-1), record_num_(-1) {
  }
  LogPos(int32 file_num, int32 block_num, int32 record_num)
    : file_num_(file_num), block_num_(block_num), record_num_(record_num) {
  }
  LogPos(const LogPos& l)
    : file_num_(l.file_num_),
      block_num_(l.block_num_),
      record_num_(l.record_num_) {
  };

  const LogPos& operator=(const LogPos& l) {
    file_num_ = l.file_num_;
    block_num_ = l.block_num_;
    record_num_ = l.record_num_;
    return *this;
  }
  bool IsNull() const {
    return file_num_ == -1;
  }
  string ToString() const {
    return strutil::StringPrintf(
      " LogPos(file#: %5d; block#: %5d; record#: %5d)",
      static_cast<int32>(file_num_),
      static_cast<int32>(block_num_),
      static_cast<int32>(record_num_));
  }
  bool operator==(const LogPos& l) const {
    return (file_num_ == l.file_num_ &&
            block_num_ == l.block_num_ &&
            record_num_ == l.record_num_);
  }
};

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

class LogWriter {
 public:
  LogWriter(const char* log_dir,
            const char* file_base,
            bool sync_on_write,
            int32 block_size = kDefaultRecordBlockSize,
            int32 blocks_per_file = kDefaultBlocksPerFile,
            bool temporary_incomplete_file = false,
            bool deflate = false);
  ~LogWriter();

  bool WriteRecord(io::MemoryStream* in,
                   bool force_flush = false);
  bool WriteRecord(const char* buffer, int32 size,
                   bool force_flush = false);

  bool Flush();

  const string& file_name() const { return file_name_; }

  // This makes sense after a Flush() call
  LogPos Tell() const;

 private:
  // Flushes the buffer after appending a record in the recorder
  bool ProcessWriteRecord(bool result, bool force_flush);
  // Name of the current file to open (according to our internal members)
  // If temp == true and the real file does not exist:
  //   => generates a temporary filename
  // else => returns the real file name
  string ComposeFileName(bool temp) const;
  // Puts the current buf_ into the log file
  bool WriteBuffer(bool force_flush);
  // Determines and opens the next log file
  bool OpenLog();
  // Close the current log file. Moves temporary file in final place.
  void CloseLog();
  // Makes the file size correct for the current log file
  bool NormalizeFileSize(int64 pos);

  const string log_dir_;         // log files are here..
  const string file_base_;       // and begin w/ this name
  const bool   sync_on_write_;   // we synch th files
  re::RE       re_;              // matches log files
  const int64  max_file_size_;   // we stop at logs files of this size
                                 // (and continue to the next file)
  const int32 block_size_;       // we create block of records of this size
  const int32 blocks_per_file_;  // and we write these many blocks per file
  const bool temporary_incomplete_file_;
                                 // incomplete files have a temporary name.
                                 // Once a file is complete it is moved in
                                 // the right place.
  const bool deflate_;           // deflate records on writing
                                 // (makes sense for txt)

  int32 file_num_;               // the current file number
                                 //  (for the opened file)
  int   fd_;                     // current file descriptor
  string file_name_;             // name for the current file
  MemoryStream buf_;             // used to accumulate records
  RecordWriter recorder_;        // encapsulates records for us


  DISALLOW_EVIL_CONSTRUCTORS(LogWriter);
};

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

class LogReader {
 public:
  LogReader(const char* log_dir,
            const char* file_base,
            int32 block_size = kDefaultRecordBlockSize,
            int32 blocks_per_file = kDefaultBlocksPerFile);
  ~LogReader();

  bool GetNextRecord(io::MemoryStream* out);

  int32 num_errors() const   { return num_errors_; }
  int   last_errno() const   { return last_errno_; }

  const string& file_name() const { return file_name_; }

  const LogPos& Tell() const { return last_read_pos_;    }
  const LogPos& crt_pos() const { return crt_pos_;    }
  bool Seek(const LogPos& log_pos);

  bool IsValid() const {
    return fd_ != INVALID_FD_VALUE;
  }
  void Close() {
    CloseInternal(false);
  }
  bool Rewind();

  string ComposeFileName(int32 file_num) const;

 private:
  void CloseInternal(bool is_error);
  bool OpenFile(int32 file_num);
  bool AdvanceToPosInFile(const LogPos& log_pos);
  bool ReadNextBlock();


  const string log_dir_;         // log files are here..
  const string file_base_;       // and begin w/ this name
  re::RE       re_;              // matches log files
  const int32 block_size_;       // we create block of records of this size
  const int32 blocks_per_file_;  // and we write these many blocks per file

  int   fd_;                     // current file descriptor
  string file_name_;             // the name of the current file
  LogPos crt_pos_;               // our position in log
  LogPos last_read_pos_;         // our position in log - w/
                                 // a valid block to read
  RecordReader reader_;          // splits records for us
  int32 num_errors_;
  int32 last_errno_;
  io::MemoryStream buf_;

  DISALLOW_EVIL_CONSTRUCTORS(LogReader);
};

// Clean log files before the given position (i.e. you can seek to the
// given position, but probably not before .. )
//
// We return how many files we have deleted.
int32 CleanLog(const char* log_dir, const char* file_base, LogPos first_pos,
               int32 block_size = kDefaultRecordBlockSize);
}

#endif  // __COMMON_IO_LOGIO_LOGIO_H__