root/trunk/whisperlib/common/io/logio/test/logio_test.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

// Simple tests for logio module

#include <unistd.h>
#include <sys/stat.h>
#include <fcntl.h>

#include "common/base/types.h"
#include "common/base/log.h"
#include "common/base/timer.h"
#include "common/base/system.h"
#include "common/sync/thread.h"

#include "common/io/buffer/memory_stream.h"
#include "common/io/logio/logio.h"

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

DEFINE_string(test_dir,
              "/tmp",
              "Where to write test logs");

DEFINE_string(test_filebase,
              "testlog",
              "Write logs w/ this prefix");

DEFINE_int32(rand_seed,
             3274,
             "Seed the random number generator w/ this");

DEFINE_int64(num_records,
             20000,
             "Number of records for the test");

DEFINE_int32(block_size,
             16385,
             "Create blocks of this size");

DEFINE_int32(blocks_per_file,
             1003,
             "Create files of this size (in blocks)");

DEFINE_int32(record_size,
             200,
             "Generate records of this size");

DEFINE_int32(writer_mess_probability,
             10000,
             "We mess-up the log w/ this probability :)");

DEFINE_int32(writer_stop_probability,
             100000,
             "Reintiialize the writer w/ this probability");

DEFINE_int32(reader_stop_probability,
             100000,
             "Reintiialize the reader w/ this probability");

DEFINE_int32(num_writers,
             5,
             "Use these many concurrent writers");

DEFINE_bool(deflate,
            false,
            "Zip records for writing");

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

static unsigned int rand_seed;

void GenerateRecord(int64 rid,
                    int32 tid,
                    io::MemoryStream* rec) {
  char* buf = new char[FLAGS_record_size];
  memset(buf, 0, FLAGS_record_size);
  *(reinterpret_cast<int32*>(buf)) = tid;
  *(reinterpret_cast<int64*>(buf + sizeof(tid))) = rid;
  rec->AppendRaw(buf, FLAGS_record_size);
}

int32 VerifyRecord(bool fresh_reader,
                   int64* rid,
                   int32 num_tids,
                   io::MemoryStream* rec) {
  CHECK_EQ(rec->Size(), FLAGS_record_size);
  string s;
  rec->ReadString(&s, FLAGS_record_size);
  const int32 tid = *(reinterpret_cast<const int32*>(s.data()));
  CHECK_LE(tid, num_tids);
  CHECK_GE(tid, 0);
  const int64 crt_rid =
      *(reinterpret_cast<const int64*>(s.data() + sizeof(tid)));
  if ( rid[tid] >= 0 ) {
    if ( FLAGS_writer_mess_probability == 0 ) {
      CHECK_EQ(rid[tid] + 1, crt_rid);
    } else {
      CHECK_LE(rid[tid] + 1, crt_rid);
      if ( rid[tid] + 1 < crt_rid ) {
        if ( !fresh_reader ) {
          CHECK_EQ((crt_rid - rid[tid]) / (FLAGS_block_size - 4) %
                   FLAGS_record_size,
                   0) << " crt_rid: " << crt_rid
                      << " rid[tid]: " << rid[tid];
        }
      }
    }
  }
  rid[tid] = crt_rid;
  return tid;
}

void WriterThread(int32 tid) {
  io::LogWriter* writer =  new io::LogWriter(FLAGS_test_dir.c_str(),
                                             FLAGS_test_filebase.c_str(),
                                             false,
                                             FLAGS_block_size,
                                             FLAGS_blocks_per_file,
                                             false,
                                             FLAGS_deflate);
  {
    io::MemoryStream rec;
    for ( int64 i = 0; i < FLAGS_num_records; ++i ) {
      LOG_INFO_IF((i % 1000) == 0) << "tid: " << tid
                                   << " Writing record: " << i;
      GenerateRecord(i, tid, &rec);
      CHECK(writer->WriteRecord(&rec));
      const bool should_reopen =
          (rand_r(&rand_seed) % FLAGS_writer_stop_probability) == 0;
      if ( should_reopen ) {
        LOG_INFO << " Re-opening the writer: " << tid << " @" << i;
        delete writer;
        writer =  new io::LogWriter(FLAGS_test_dir.c_str(),
                                    FLAGS_test_filebase.c_str(),
                                    false,
                                    FLAGS_block_size,
                                    FLAGS_blocks_per_file,
                                    false,
                                    FLAGS_deflate);
      }
      if ( FLAGS_writer_mess_probability > 0 ) {
        const bool should_mess = (
            rand_r(&rand_seed) % FLAGS_writer_mess_probability) == 0;
        if ( should_mess && i < (FLAGS_num_records -
                                 5 * (FLAGS_block_size / FLAGS_record_size)) ) {
          LOG_INFO << " Messing file: " << writer->file_name();
          int fd = ::open(writer->file_name().c_str(),
                          O_WRONLY | O_APPEND | O_SYNC | O_CREAT | O_LARGEFILE,
                          0644);
          if ( fd >= 0 ) {
            const int32 size = FLAGS_block_size * 3 / 4;
            char* buf = new char[size];
            CHECK_EQ(size, ::write(fd, buf, size));
            delete [] buf;
            ::close(fd);
          }
        }
      }
    }
  }
  delete writer;
}

void ReaderThread(io::LogPos pos, int32 num_tids) {
  io::LogReader* reader = new io::LogReader(FLAGS_test_dir.c_str(),
                                            FLAGS_test_filebase.c_str(),
                                            FLAGS_block_size,
                                            FLAGS_blocks_per_file);
  int64* rids = new int64[num_tids];
  for ( int32 i = 0; i < num_tids; ++i ) {
    rids[i] = -1LL;
  }
  if ( !pos.IsNull() ) {
    CHECK(reader->Seek(pos));
  }
  io::MemoryStream rec;
  int32 tids_solved = 0;
  bool fresh_reader = true;
  while ( tids_solved < num_tids ) {
    const bool should_reopen = (
        FLAGS_reader_stop_probability > 0 &&
        ((rand_r(&rand_seed) % FLAGS_reader_stop_probability) == 0));
    if ( should_reopen ) {
      io::LogPos pos(reader->Tell());
      LOG_INFO << "Reopening Reader at: " << pos.ToString()
               << " Vs. : " << reader->crt_pos().ToString();
      delete reader;
      reader = new io::LogReader(FLAGS_test_dir.c_str(),
                                 FLAGS_test_filebase.c_str(),
                                 FLAGS_block_size,
                                 FLAGS_blocks_per_file);
      if ( !pos.IsNull() ) {
        CHECK(reader->Seek(pos));
        CHECK(reader->Tell() == pos) << " AT: " << reader->Tell().ToString()
                                     << " vs. " << pos.ToString();
        LOG_INFO << " OK - > AT: " << reader->Tell().ToString()
                 << " vs: " << reader->crt_pos().ToString();
      }
      fresh_reader = true;
    }
    if ( !reader->GetNextRecord(&rec) ) {
      if ( FLAGS_writer_mess_probability == 0 ) {
        CHECK_EQ(reader->num_errors(), 0);
      }
      sleep(1);
    } else {
      int32 tid = VerifyRecord(fresh_reader, rids, num_tids, &rec);
      LOG_INFO_IF((rids[tid] % 1000) == 0) << " Reader: " << tid
                                           << " @ " << rids[tid];
      fresh_reader = false;
      if ( rids[tid] > FLAGS_num_records - 50 ) {
        if ( rids[tid] >= FLAGS_num_records - 1 ) {
          tids_solved++;
          LOG_INFO << " ===========> Solved TID: " << tid;
        }
      }
      rec.Clear();
    }
  }
}

int main(int argc, char* argv[]) {
  common::Init(argc, argv);
  rand_seed = FLAGS_rand_seed;
  srand(rand_seed);

  LOG_INFO << "Clearing old files.."
           << system(strutil::StringPrintf(
                         "rm -f %s/%s_??????????_??????????",
                         FLAGS_test_dir.c_str(),
                         FLAGS_test_filebase.c_str()).c_str());
  LOG_INFO << "Creating threads..";
  thread::Thread** iothreads = new thread::Thread*[FLAGS_num_writers];
  for ( int32 i = 0; i < FLAGS_num_writers; ++i ) {
    iothreads[i] = new thread::Thread(NewCallback(&WriterThread, i));
    iothreads[i]->SetJoinable();
    iothreads[i]->Start();
  }
  sleep(2);

  thread::Thread reader(NewCallback(&ReaderThread,
                                    io::LogPos(),
                                    FLAGS_num_writers));
  reader.SetJoinable();
  reader.Start();

  LOG_INFO << " Waiting to join !";
  for ( int32 i = 0; i < FLAGS_num_writers; ++i ) {
    iothreads[i]->Join();
    delete iothreads[i];
    LOG_INFO << " Join: " << i;
  }
  delete [] iothreads;
  reader.Join();
  LOG_INFO << " Reader Join";
}