+++ /dev/null
-// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
-// vim: ts=8 sw=2 smarttab
-
-#include "journal/JournalPlayer.h"
-#include "journal/Entry.h"
-#include "journal/JournalMetadata.h"
-#include "journal/ReplayHandler.h"
-#include "include/stringify.h"
-#include "common/Cond.h"
-#include "common/Mutex.h"
-#include "gtest/gtest.h"
-#include "test/journal/RadosTestFixture.h"
-#include <list>
-#include <boost/scope_exit.hpp>
-
-typedef std::list<journal::Entry> Entries;
-
-template <typename T>
-class TestJournalPlayer : public RadosTestFixture {
-public:
- typedef std::list<journal::JournalPlayer *> JournalPlayers;
-
- static const uint64_t max_fetch_bytes = T::max_fetch_bytes;
-
- struct ReplayHandler : public journal::ReplayHandler {
- Mutex lock;
- Cond cond;
- bool entries_available;
- bool complete;
- int complete_result;
-
- ReplayHandler()
- : lock("lock"), entries_available(false), complete(false),
- complete_result(0) {}
-
- void get() override {}
- void put() override {}
-
- void handle_entries_available() override {
- Mutex::Locker locker(lock);
- entries_available = true;
- cond.Signal();
- }
-
- void handle_complete(int r) override {
- Mutex::Locker locker(lock);
- complete = true;
- complete_result = r;
- cond.Signal();
- }
- };
-
- void TearDown() override {
- for (JournalPlayers::iterator it = m_players.begin();
- it != m_players.end(); ++it) {
- delete *it;
- }
- RadosTestFixture::TearDown();
- }
-
- journal::JournalMetadataPtr create_metadata(const std::string &oid) {
- return RadosTestFixture::create_metadata(oid, "client", 0.1,
- max_fetch_bytes);
- }
-
- int client_commit(const std::string &oid,
- journal::JournalPlayer::ObjectSetPosition position) {
- return RadosTestFixture::client_commit(oid, "client", position);
- }
-
- journal::Entry create_entry(uint64_t tag_tid, uint64_t entry_tid) {
- std::string payload(128, '0');
- bufferlist payload_bl;
- payload_bl.append(payload);
- return journal::Entry(tag_tid, entry_tid, payload_bl);
- }
-
- journal::JournalPlayer *create_player(const std::string &oid,
- const journal::JournalMetadataPtr &metadata) {
- journal::JournalPlayer *player(new journal::JournalPlayer(
- m_ioctx, oid + ".", metadata, &m_replay_hander));
- m_players.push_back(player);
- return player;
- }
-
- bool wait_for_entries(journal::JournalPlayer *player, uint32_t count,
- Entries *entries) {
- entries->clear();
- while (entries->size() < count) {
- journal::Entry entry;
- uint64_t commit_tid;
- while (entries->size() < count &&
- player->try_pop_front(&entry, &commit_tid)) {
- entries->push_back(entry);
- }
- if (entries->size() == count) {
- break;
- }
-
- Mutex::Locker locker(m_replay_hander.lock);
- if (m_replay_hander.entries_available) {
- m_replay_hander.entries_available = false;
- } else if (m_replay_hander.cond.WaitInterval(
- m_replay_hander.lock, utime_t(10, 0)) != 0) {
- break;
- }
- }
- return entries->size() == count;
- }
-
- bool wait_for_complete(journal::JournalPlayer *player) {
- Mutex::Locker locker(m_replay_hander.lock);
- while (!m_replay_hander.complete) {
- journal::Entry entry;
- uint64_t commit_tid;
- player->try_pop_front(&entry, &commit_tid);
-
- if (m_replay_hander.cond.WaitInterval(
- m_replay_hander.lock, utime_t(10, 0)) != 0) {
- return false;
- }
- }
- m_replay_hander.complete = false;
- return true;
- }
-
- int write_entry(const std::string &oid, uint64_t object_num,
- uint64_t tag_tid, uint64_t entry_tid) {
- bufferlist bl;
- ::encode(create_entry(tag_tid, entry_tid), bl);
- return append(oid + "." + stringify(object_num), bl);
- }
-
- JournalPlayers m_players;
- ReplayHandler m_replay_hander;
-};
-
-template <uint64_t _max_fetch_bytes>
-class TestJournalPlayerParams {
-public:
- static const uint64_t max_fetch_bytes = _max_fetch_bytes;
-};
-
-typedef ::testing::Types<TestJournalPlayerParams<0>,
- TestJournalPlayerParams<16> > TestJournalPlayerTypes;
-TYPED_TEST_CASE(TestJournalPlayer, TestJournalPlayerTypes);
-
-TYPED_TEST(TestJournalPlayer, Prefetch) {
- std::string oid = this->get_temp_oid();
-
- journal::JournalPlayer::ObjectPositions positions;
- positions = {
- cls::journal::ObjectPosition(0, 234, 122) };
- cls::journal::ObjectSetPosition commit_position(positions);
-
- ASSERT_EQ(0, this->create(oid));
- ASSERT_EQ(0, this->client_register(oid));
- ASSERT_EQ(0, this->client_commit(oid, commit_position));
-
- journal::JournalMetadataPtr metadata = this->create_metadata(oid);
- ASSERT_EQ(0, this->init_metadata(metadata));
-
- journal::JournalPlayer *player = this->create_player(oid, metadata);
- BOOST_SCOPE_EXIT_ALL( (player) ) {
- C_SaferCond unwatch_ctx;
- player->shut_down(&unwatch_ctx);
- ASSERT_EQ(0, unwatch_ctx.wait());
- };
-
- ASSERT_EQ(0, this->write_entry(oid, 0, 234, 122));
- ASSERT_EQ(0, this->write_entry(oid, 1, 234, 123));
- ASSERT_EQ(0, this->write_entry(oid, 0, 234, 124));
- ASSERT_EQ(0, this->write_entry(oid, 1, 234, 125));
-
- player->prefetch();
-
- Entries entries;
- ASSERT_TRUE(this->wait_for_entries(player, 3, &entries));
- ASSERT_TRUE(this->wait_for_complete(player));
-
- Entries expected_entries;
- expected_entries = {
- this->create_entry(234, 123),
- this->create_entry(234, 124),
- this->create_entry(234, 125)};
- ASSERT_EQ(expected_entries, entries);
-
- uint64_t last_tid;
- ASSERT_TRUE(metadata->get_last_allocated_entry_tid(234, &last_tid));
- ASSERT_EQ(125U, last_tid);
-}
-
-TYPED_TEST(TestJournalPlayer, PrefetchSkip) {
- std::string oid = this->get_temp_oid();
-
- journal::JournalPlayer::ObjectPositions positions;
- positions = {
- cls::journal::ObjectPosition(0, 234, 125),
- cls::journal::ObjectPosition(1, 234, 124) };
- cls::journal::ObjectSetPosition commit_position(positions);
-
- ASSERT_EQ(0, this->create(oid));
- ASSERT_EQ(0, this->client_register(oid));
- ASSERT_EQ(0, this->client_commit(oid, commit_position));
-
- journal::JournalMetadataPtr metadata = this->create_metadata(oid);
- ASSERT_EQ(0, this->init_metadata(metadata));
-
- journal::JournalPlayer *player = this->create_player(oid, metadata);
- BOOST_SCOPE_EXIT_ALL( (player) ) {
- C_SaferCond unwatch_ctx;
- player->shut_down(&unwatch_ctx);
- ASSERT_EQ(0, unwatch_ctx.wait());
- };
-
- ASSERT_EQ(0, this->write_entry(oid, 0, 234, 122));
- ASSERT_EQ(0, this->write_entry(oid, 1, 234, 123));
- ASSERT_EQ(0, this->write_entry(oid, 0, 234, 124));
- ASSERT_EQ(0, this->write_entry(oid, 1, 234, 125));
-
- player->prefetch();
-
- Entries entries;
- ASSERT_TRUE(this->wait_for_entries(player, 0, &entries));
- ASSERT_TRUE(this->wait_for_complete(player));
-
- uint64_t last_tid;
- ASSERT_TRUE(metadata->get_last_allocated_entry_tid(234, &last_tid));
- ASSERT_EQ(125U, last_tid);
-}
-
-TYPED_TEST(TestJournalPlayer, PrefetchWithoutCommit) {
- std::string oid = this->get_temp_oid();
-
- cls::journal::ObjectSetPosition commit_position;
-
- ASSERT_EQ(0, this->create(oid));
- ASSERT_EQ(0, this->client_register(oid));
- ASSERT_EQ(0, this->client_commit(oid, commit_position));
-
- journal::JournalMetadataPtr metadata = this->create_metadata(oid);
- ASSERT_EQ(0, this->init_metadata(metadata));
-
- journal::JournalPlayer *player = this->create_player(oid, metadata);
- BOOST_SCOPE_EXIT_ALL( (player) ) {
- C_SaferCond unwatch_ctx;
- player->shut_down(&unwatch_ctx);
- ASSERT_EQ(0, unwatch_ctx.wait());
- };
-
- ASSERT_EQ(0, this->write_entry(oid, 0, 234, 122));
- ASSERT_EQ(0, this->write_entry(oid, 1, 234, 123));
-
- player->prefetch();
-
- Entries entries;
- ASSERT_TRUE(this->wait_for_entries(player, 2, &entries));
- ASSERT_TRUE(this->wait_for_complete(player));
-
- Entries expected_entries;
- expected_entries = {
- this->create_entry(234, 122),
- this->create_entry(234, 123)};
- ASSERT_EQ(expected_entries, entries);
-}
-
-TYPED_TEST(TestJournalPlayer, PrefetchMultipleTags) {
- std::string oid = this->get_temp_oid();
-
- journal::JournalPlayer::ObjectPositions positions;
- positions = {
- cls::journal::ObjectPosition(2, 234, 122),
- cls::journal::ObjectPosition(1, 234, 121),
- cls::journal::ObjectPosition(0, 234, 120)};
- cls::journal::ObjectSetPosition commit_position(positions);
-
- ASSERT_EQ(0, this->create(oid, 14, 3));
- ASSERT_EQ(0, this->client_register(oid));
- ASSERT_EQ(0, this->client_commit(oid, commit_position));
-
- journal::JournalMetadataPtr metadata = this->create_metadata(oid);
- ASSERT_EQ(0, this->init_metadata(metadata));
-
- journal::JournalPlayer *player = this->create_player(oid, metadata);
- BOOST_SCOPE_EXIT_ALL( (player) ) {
- C_SaferCond unwatch_ctx;
- player->shut_down(&unwatch_ctx);
- ASSERT_EQ(0, unwatch_ctx.wait());
- };
-
- ASSERT_EQ(0, this->write_entry(oid, 0, 234, 120));
- ASSERT_EQ(0, this->write_entry(oid, 1, 234, 121));
- ASSERT_EQ(0, this->write_entry(oid, 2, 234, 122));
- ASSERT_EQ(0, this->write_entry(oid, 0, 234, 123));
- ASSERT_EQ(0, this->write_entry(oid, 1, 234, 124));
- ASSERT_EQ(0, this->write_entry(oid, 0, 236, 0)); // new tag allocated
-
- player->prefetch();
-
- Entries entries;
- ASSERT_TRUE(this->wait_for_entries(player, 3, &entries));
- ASSERT_TRUE(this->wait_for_complete(player));
-
- uint64_t last_tid;
- ASSERT_TRUE(metadata->get_last_allocated_entry_tid(234, &last_tid));
- ASSERT_EQ(124U, last_tid);
- ASSERT_TRUE(metadata->get_last_allocated_entry_tid(236, &last_tid));
- ASSERT_EQ(0U, last_tid);
-}
-
-TYPED_TEST(TestJournalPlayer, PrefetchCorruptSequence) {
- std::string oid = this->get_temp_oid();
-
- cls::journal::ObjectSetPosition commit_position;
-
- ASSERT_EQ(0, this->create(oid));
- ASSERT_EQ(0, this->client_register(oid));
- ASSERT_EQ(0, this->client_commit(oid, commit_position));
-
- journal::JournalMetadataPtr metadata = this->create_metadata(oid);
- ASSERT_EQ(0, this->init_metadata(metadata));
-
- journal::JournalPlayer *player = this->create_player(oid, metadata);
- BOOST_SCOPE_EXIT_ALL( (player) ) {
- C_SaferCond unwatch_ctx;
- player->shut_down(&unwatch_ctx);
- ASSERT_EQ(0, unwatch_ctx.wait());
- };
-
- ASSERT_EQ(0, this->write_entry(oid, 0, 234, 120));
- ASSERT_EQ(0, this->write_entry(oid, 1, 234, 121));
- ASSERT_EQ(0, this->write_entry(oid, 0, 234, 124));
-
- player->prefetch();
- Entries entries;
- ASSERT_TRUE(this->wait_for_entries(player, 2, &entries));
-
- journal::Entry entry;
- uint64_t commit_tid;
- ASSERT_FALSE(player->try_pop_front(&entry, &commit_tid));
- ASSERT_TRUE(this->wait_for_complete(player));
- ASSERT_EQ(-ENOMSG, this->m_replay_hander.complete_result);
-}
-
-TYPED_TEST(TestJournalPlayer, PrefetchMissingSequence) {
- std::string oid = this->get_temp_oid();
-
- cls::journal::ObjectSetPosition commit_position;
-
- ASSERT_EQ(0, this->create(oid, 14, 4));
- ASSERT_EQ(0, this->client_register(oid));
- ASSERT_EQ(0, this->client_commit(oid, commit_position));
-
- journal::JournalMetadataPtr metadata = this->create_metadata(oid);
- ASSERT_EQ(0, this->init_metadata(metadata));
-
- journal::JournalPlayer *player = this->create_player(oid, metadata);
- BOOST_SCOPE_EXIT_ALL( (player) ) {
- C_SaferCond unwatch_ctx;
- player->shut_down(&unwatch_ctx);
- ASSERT_EQ(0, unwatch_ctx.wait());
- };
-
- ASSERT_EQ(0, metadata->set_active_set(1));
- ASSERT_EQ(0, this->write_entry(oid, 0, 2, 852));
- ASSERT_EQ(0, this->write_entry(oid, 0, 2, 856));
- ASSERT_EQ(0, this->write_entry(oid, 0, 2, 860));
- ASSERT_EQ(0, this->write_entry(oid, 1, 2, 853));
- ASSERT_EQ(0, this->write_entry(oid, 1, 2, 857));
- ASSERT_EQ(0, this->write_entry(oid, 5, 2, 861));
- ASSERT_EQ(0, this->write_entry(oid, 2, 2, 854));
- ASSERT_EQ(0, this->write_entry(oid, 0, 3, 0));
- ASSERT_EQ(0, this->write_entry(oid, 5, 3, 1));
- ASSERT_EQ(0, this->write_entry(oid, 2, 3, 2));
- ASSERT_EQ(0, this->write_entry(oid, 3, 3, 3));
-
- player->prefetch();
- Entries entries;
- ASSERT_TRUE(this->wait_for_entries(player, 7, &entries));
-
- Entries expected_entries = {
- this->create_entry(2, 852),
- this->create_entry(2, 853),
- this->create_entry(2, 854),
- this->create_entry(3, 0),
- this->create_entry(3, 1),
- this->create_entry(3, 2),
- this->create_entry(3, 3)};
- ASSERT_EQ(expected_entries, entries);
-
- ASSERT_TRUE(this->wait_for_complete(player));
- ASSERT_EQ(0, this->m_replay_hander.complete_result);
-}
-
-TYPED_TEST(TestJournalPlayer, PrefetchLargeMissingSequence) {
- std::string oid = this->get_temp_oid();
-
- cls::journal::ObjectSetPosition commit_position;
-
- ASSERT_EQ(0, this->create(oid));
- ASSERT_EQ(0, this->client_register(oid));
- ASSERT_EQ(0, this->client_commit(oid, commit_position));
-
- journal::JournalMetadataPtr metadata = this->create_metadata(oid);
- ASSERT_EQ(0, this->init_metadata(metadata));
-
- journal::JournalPlayer *player = this->create_player(oid, metadata);
- BOOST_SCOPE_EXIT_ALL( (player) ) {
- C_SaferCond unwatch_ctx;
- player->shut_down(&unwatch_ctx);
- ASSERT_EQ(0, unwatch_ctx.wait());
- };
-
- ASSERT_EQ(0, metadata->set_active_set(2));
- ASSERT_EQ(0, this->write_entry(oid, 0, 0, 0));
- ASSERT_EQ(0, this->write_entry(oid, 1, 0, 1));
- ASSERT_EQ(0, this->write_entry(oid, 3, 0, 3));
- ASSERT_EQ(0, this->write_entry(oid, 4, 1, 0));
-
- player->prefetch();
- Entries entries;
- ASSERT_TRUE(this->wait_for_entries(player, 3, &entries));
-
- Entries expected_entries = {
- this->create_entry(0, 0),
- this->create_entry(0, 1),
- this->create_entry(1, 0)};
- ASSERT_EQ(expected_entries, entries);
-}
-
-TYPED_TEST(TestJournalPlayer, PrefetchBlockedNewTag) {
- std::string oid = this->get_temp_oid();
-
- cls::journal::ObjectSetPosition commit_position;
-
- ASSERT_EQ(0, this->create(oid));
- ASSERT_EQ(0, this->client_register(oid));
- ASSERT_EQ(0, this->client_commit(oid, commit_position));
-
- journal::JournalMetadataPtr metadata = this->create_metadata(oid);
- ASSERT_EQ(0, this->init_metadata(metadata));
-
- journal::JournalPlayer *player = this->create_player(oid, metadata);
- BOOST_SCOPE_EXIT_ALL( (player) ) {
- C_SaferCond unwatch_ctx;
- player->shut_down(&unwatch_ctx);
- ASSERT_EQ(0, unwatch_ctx.wait());
- };
-
- ASSERT_EQ(0, this->write_entry(oid, 0, 0, 0));
- ASSERT_EQ(0, this->write_entry(oid, 1, 0, 1));
- ASSERT_EQ(0, this->write_entry(oid, 0, 0, 2));
- ASSERT_EQ(0, this->write_entry(oid, 0, 0, 4));
- ASSERT_EQ(0, this->write_entry(oid, 0, 1, 0));
-
- player->prefetch();
- Entries entries;
- ASSERT_TRUE(this->wait_for_entries(player, 4, &entries));
-
- Entries expected_entries = {
- this->create_entry(0, 0),
- this->create_entry(0, 1),
- this->create_entry(0, 2),
- this->create_entry(1, 0)};
- ASSERT_EQ(expected_entries, entries);
-}
-
-TYPED_TEST(TestJournalPlayer, PrefetchStaleEntries) {
- std::string oid = this->get_temp_oid();
-
- journal::JournalPlayer::ObjectPositions positions = {
- cls::journal::ObjectPosition(0, 1, 0) };
- cls::journal::ObjectSetPosition commit_position(positions);
-
- ASSERT_EQ(0, this->create(oid));
- ASSERT_EQ(0, this->client_register(oid));
- ASSERT_EQ(0, this->client_commit(oid, commit_position));
-
- journal::JournalMetadataPtr metadata = this->create_metadata(oid);
- ASSERT_EQ(0, this->init_metadata(metadata));
-
- journal::JournalPlayer *player = this->create_player(oid, metadata);
- BOOST_SCOPE_EXIT_ALL( (player) ) {
- C_SaferCond unwatch_ctx;
- player->shut_down(&unwatch_ctx);
- ASSERT_EQ(0, unwatch_ctx.wait());
- };
-
- ASSERT_EQ(0, this->write_entry(oid, 1, 0, 1));
- ASSERT_EQ(0, this->write_entry(oid, 1, 0, 3));
- ASSERT_EQ(0, this->write_entry(oid, 0, 1, 0));
- ASSERT_EQ(0, this->write_entry(oid, 1, 1, 1));
-
- player->prefetch();
- Entries entries;
- ASSERT_TRUE(this->wait_for_entries(player, 1, &entries));
-
- Entries expected_entries = {
- this->create_entry(1, 1)};
- ASSERT_EQ(expected_entries, entries);
-
- ASSERT_TRUE(this->wait_for_complete(player));
- ASSERT_EQ(0, this->m_replay_hander.complete_result);
-}
-
-TYPED_TEST(TestJournalPlayer, PrefetchUnexpectedTag) {
- std::string oid = this->get_temp_oid();
-
- cls::journal::ObjectSetPosition commit_position;
-
- ASSERT_EQ(0, this->create(oid));
- ASSERT_EQ(0, this->client_register(oid));
- ASSERT_EQ(0, this->client_commit(oid, commit_position));
-
- journal::JournalMetadataPtr metadata = this->create_metadata(oid);
- ASSERT_EQ(0, this->init_metadata(metadata));
-
- journal::JournalPlayer *player = this->create_player(oid, metadata);
- BOOST_SCOPE_EXIT_ALL( (player) ) {
- C_SaferCond unwatch_ctx;
- player->shut_down(&unwatch_ctx);
- ASSERT_EQ(0, unwatch_ctx.wait());
- };
-
- ASSERT_EQ(0, this->write_entry(oid, 0, 234, 120));
- ASSERT_EQ(0, this->write_entry(oid, 1, 235, 121));
- ASSERT_EQ(0, this->write_entry(oid, 0, 234, 124));
-
- player->prefetch();
- Entries entries;
- ASSERT_TRUE(this->wait_for_entries(player, 1, &entries));
-
- journal::Entry entry;
- uint64_t commit_tid;
- ASSERT_FALSE(player->try_pop_front(&entry, &commit_tid));
- ASSERT_TRUE(this->wait_for_complete(player));
- ASSERT_EQ(0, this->m_replay_hander.complete_result);
-}
-
-TYPED_TEST(TestJournalPlayer, PrefetchAndWatch) {
- std::string oid = this->get_temp_oid();
-
- journal::JournalPlayer::ObjectPositions positions;
- positions = {
- cls::journal::ObjectPosition(0, 234, 122)};
- cls::journal::ObjectSetPosition commit_position(positions);
-
- ASSERT_EQ(0, this->create(oid));
- ASSERT_EQ(0, this->client_register(oid));
- ASSERT_EQ(0, this->client_commit(oid, commit_position));
-
- journal::JournalMetadataPtr metadata = this->create_metadata(oid);
- ASSERT_EQ(0, this->init_metadata(metadata));
-
- journal::JournalPlayer *player = this->create_player(oid, metadata);
- BOOST_SCOPE_EXIT_ALL( (player) ) {
- C_SaferCond unwatch_ctx;
- player->shut_down(&unwatch_ctx);
- ASSERT_EQ(0, unwatch_ctx.wait());
- };
-
- ASSERT_EQ(0, this->write_entry(oid, 0, 234, 122));
-
- player->prefetch_and_watch(0.25);
-
- Entries entries;
- ASSERT_EQ(0, this->write_entry(oid, 1, 234, 123));
- ASSERT_TRUE(this->wait_for_entries(player, 1, &entries));
-
- Entries expected_entries;
- expected_entries = {this->create_entry(234, 123)};
- ASSERT_EQ(expected_entries, entries);
-
- ASSERT_EQ(0, this->write_entry(oid, 0, 234, 124));
- ASSERT_TRUE(this->wait_for_entries(player, 1, &entries));
-
- expected_entries = {this->create_entry(234, 124)};
- ASSERT_EQ(expected_entries, entries);
-}
-
-TYPED_TEST(TestJournalPlayer, PrefetchSkippedObject) {
- std::string oid = this->get_temp_oid();
-
- cls::journal::ObjectSetPosition commit_position;
-
- ASSERT_EQ(0, this->create(oid, 14, 3));
- ASSERT_EQ(0, this->client_register(oid));
- ASSERT_EQ(0, this->client_commit(oid, commit_position));
-
- journal::JournalMetadataPtr metadata = this->create_metadata(oid);
- ASSERT_EQ(0, this->init_metadata(metadata));
- ASSERT_EQ(0, metadata->set_active_set(2));
-
- journal::JournalPlayer *player = this->create_player(oid, metadata);
- BOOST_SCOPE_EXIT_ALL( (player) ) {
- C_SaferCond unwatch_ctx;
- player->shut_down(&unwatch_ctx);
- ASSERT_EQ(0, unwatch_ctx.wait());
- };
-
- ASSERT_EQ(0, this->write_entry(oid, 0, 234, 122));
- ASSERT_EQ(0, this->write_entry(oid, 1, 234, 123));
- ASSERT_EQ(0, this->write_entry(oid, 5, 234, 124));
- ASSERT_EQ(0, this->write_entry(oid, 6, 234, 125));
- ASSERT_EQ(0, this->write_entry(oid, 7, 234, 126));
-
- player->prefetch();
-
- Entries entries;
- ASSERT_TRUE(this->wait_for_entries(player, 5, &entries));
- ASSERT_TRUE(this->wait_for_complete(player));
-
- Entries expected_entries;
- expected_entries = {
- this->create_entry(234, 122),
- this->create_entry(234, 123),
- this->create_entry(234, 124),
- this->create_entry(234, 125),
- this->create_entry(234, 126)};
- ASSERT_EQ(expected_entries, entries);
-
- uint64_t last_tid;
- ASSERT_TRUE(metadata->get_last_allocated_entry_tid(234, &last_tid));
- ASSERT_EQ(126U, last_tid);
-}
-
-TYPED_TEST(TestJournalPlayer, ImbalancedJournal) {
- std::string oid = this->get_temp_oid();
-
- journal::JournalPlayer::ObjectPositions positions = {
- cls::journal::ObjectPosition(9, 300, 1),
- cls::journal::ObjectPosition(8, 300, 0),
- cls::journal::ObjectPosition(10, 200, 4334),
- cls::journal::ObjectPosition(11, 200, 4331) };
- cls::journal::ObjectSetPosition commit_position(positions);
-
- ASSERT_EQ(0, this->create(oid, 14, 4));
- ASSERT_EQ(0, this->client_register(oid));
- ASSERT_EQ(0, this->client_commit(oid, commit_position));
-
- journal::JournalMetadataPtr metadata = this->create_metadata(oid);
- ASSERT_EQ(0, this->init_metadata(metadata));
- ASSERT_EQ(0, metadata->set_active_set(2));
- metadata->set_minimum_set(2);
-
- journal::JournalPlayer *player = this->create_player(oid, metadata);
- BOOST_SCOPE_EXIT_ALL( (player) ) {
- C_SaferCond unwatch_ctx;
- player->shut_down(&unwatch_ctx);
- ASSERT_EQ(0, unwatch_ctx.wait());
- };
-
- ASSERT_EQ(0, this->write_entry(oid, 8, 300, 0));
- ASSERT_EQ(0, this->write_entry(oid, 8, 301, 0));
- ASSERT_EQ(0, this->write_entry(oid, 9, 300, 1));
- ASSERT_EQ(0, this->write_entry(oid, 9, 301, 1));
- ASSERT_EQ(0, this->write_entry(oid, 10, 200, 4334));
- ASSERT_EQ(0, this->write_entry(oid, 10, 301, 2));
- ASSERT_EQ(0, this->write_entry(oid, 11, 200, 4331));
- ASSERT_EQ(0, this->write_entry(oid, 11, 301, 3));
-
- player->prefetch();
-
- Entries entries;
- ASSERT_TRUE(this->wait_for_entries(player, 4, &entries));
- ASSERT_TRUE(this->wait_for_complete(player));
-
- Entries expected_entries;
- expected_entries = {
- this->create_entry(301, 0),
- this->create_entry(301, 1),
- this->create_entry(301, 2),
- this->create_entry(301, 3)};
- ASSERT_EQ(expected_entries, entries);
-
- uint64_t last_tid;
- ASSERT_TRUE(metadata->get_last_allocated_entry_tid(301, &last_tid));
- ASSERT_EQ(3U, last_tid);
-}
-
-TYPED_TEST(TestJournalPlayer, LiveReplayLaggyAppend) {
- std::string oid = this->get_temp_oid();
-
- cls::journal::ObjectSetPosition commit_position;
-
- ASSERT_EQ(0, this->create(oid));
- ASSERT_EQ(0, this->client_register(oid));
- ASSERT_EQ(0, this->client_commit(oid, commit_position));
-
- journal::JournalMetadataPtr metadata = this->create_metadata(oid);
- ASSERT_EQ(0, this->init_metadata(metadata));
-
- journal::JournalPlayer *player = this->create_player(oid, metadata);
- BOOST_SCOPE_EXIT_ALL( (player) ) {
- C_SaferCond unwatch_ctx;
- player->shut_down(&unwatch_ctx);
- ASSERT_EQ(0, unwatch_ctx.wait());
- };
-
- ASSERT_EQ(0, this->write_entry(oid, 0, 0, 0));
- ASSERT_EQ(0, this->write_entry(oid, 1, 0, 1));
- ASSERT_EQ(0, this->write_entry(oid, 0, 0, 2));
- ASSERT_EQ(0, this->write_entry(oid, 0, 0, 4));
- ASSERT_EQ(0, this->write_entry(oid, 3, 0, 5)); // laggy entry 0/3 in object 1
- player->prefetch_and_watch(0.25);
-
- Entries entries;
- ASSERT_TRUE(this->wait_for_entries(player, 3, &entries));
-
- Entries expected_entries = {
- this->create_entry(0, 0),
- this->create_entry(0, 1),
- this->create_entry(0, 2)};
- ASSERT_EQ(expected_entries, entries);
-
- journal::Entry entry;
- uint64_t commit_tid;
- ASSERT_FALSE(player->try_pop_front(&entry, &commit_tid));
-
- ASSERT_EQ(0, this->write_entry(oid, 1, 0, 3));
- ASSERT_EQ(0, metadata->set_active_set(1));
- ASSERT_TRUE(this->wait_for_entries(player, 3, &entries));
-
- expected_entries = {
- this->create_entry(0, 3),
- this->create_entry(0, 4),
- this->create_entry(0, 5)};
- ASSERT_EQ(expected_entries, entries);
-}
-
-TYPED_TEST(TestJournalPlayer, LiveReplayMissingSequence) {
- std::string oid = this->get_temp_oid();
-
- cls::journal::ObjectSetPosition commit_position;
-
- ASSERT_EQ(0, this->create(oid, 14, 4));
- ASSERT_EQ(0, this->client_register(oid));
- ASSERT_EQ(0, this->client_commit(oid, commit_position));
-
- journal::JournalMetadataPtr metadata = this->create_metadata(oid);
- ASSERT_EQ(0, this->init_metadata(metadata));
-
- journal::JournalPlayer *player = this->create_player(oid, metadata);
- BOOST_SCOPE_EXIT_ALL( (player) ) {
- C_SaferCond unwatch_ctx;
- player->shut_down(&unwatch_ctx);
- ASSERT_EQ(0, unwatch_ctx.wait());
- };
-
- ASSERT_EQ(0, this->write_entry(oid, 0, 2, 852));
- ASSERT_EQ(0, this->write_entry(oid, 0, 2, 856));
- ASSERT_EQ(0, this->write_entry(oid, 0, 2, 860));
- ASSERT_EQ(0, this->write_entry(oid, 1, 2, 853));
- ASSERT_EQ(0, this->write_entry(oid, 1, 2, 857));
- ASSERT_EQ(0, this->write_entry(oid, 2, 2, 854));
- ASSERT_EQ(0, this->write_entry(oid, 0, 2, 856));
- player->prefetch_and_watch(0.25);
-
- Entries entries;
- ASSERT_TRUE(this->wait_for_entries(player, 3, &entries));
-
- Entries expected_entries = {
- this->create_entry(2, 852),
- this->create_entry(2, 853),
- this->create_entry(2, 854)};
- ASSERT_EQ(expected_entries, entries);
-
- journal::Entry entry;
- uint64_t commit_tid;
- ASSERT_FALSE(player->try_pop_front(&entry, &commit_tid));
-
- ASSERT_EQ(0, this->write_entry(oid, 3, 3, 3));
- ASSERT_EQ(0, this->write_entry(oid, 2, 3, 2));
- ASSERT_EQ(0, this->write_entry(oid, 1, 3, 1));
- ASSERT_EQ(0, this->write_entry(oid, 0, 3, 0));
- ASSERT_TRUE(this->wait_for_entries(player, 4, &entries));
-
- expected_entries = {
- this->create_entry(3, 0),
- this->create_entry(3, 1),
- this->create_entry(3, 2),
- this->create_entry(3, 3)};
- ASSERT_EQ(expected_entries, entries);
-}
-
-TYPED_TEST(TestJournalPlayer, LiveReplayLargeMissingSequence) {
- std::string oid = this->get_temp_oid();
-
- cls::journal::ObjectSetPosition commit_position;
-
- ASSERT_EQ(0, this->create(oid));
- ASSERT_EQ(0, this->client_register(oid));
- ASSERT_EQ(0, this->client_commit(oid, commit_position));
-
- journal::JournalMetadataPtr metadata = this->create_metadata(oid);
- ASSERT_EQ(0, this->init_metadata(metadata));
-
- journal::JournalPlayer *player = this->create_player(oid, metadata);
- BOOST_SCOPE_EXIT_ALL( (player) ) {
- C_SaferCond unwatch_ctx;
- player->shut_down(&unwatch_ctx);
- ASSERT_EQ(0, unwatch_ctx.wait());
- };
-
- ASSERT_EQ(0, metadata->set_active_set(2));
- ASSERT_EQ(0, this->write_entry(oid, 0, 0, 0));
- ASSERT_EQ(0, this->write_entry(oid, 1, 0, 1));
- ASSERT_EQ(0, this->write_entry(oid, 3, 0, 3));
- ASSERT_EQ(0, this->write_entry(oid, 4, 1, 0));
- player->prefetch_and_watch(0.25);
-
- Entries entries;
- ASSERT_TRUE(this->wait_for_entries(player, 3, &entries));
-
- Entries expected_entries = {
- this->create_entry(0, 0),
- this->create_entry(0, 1),
- this->create_entry(1, 0)};
- ASSERT_EQ(expected_entries, entries);
-}
-
-TYPED_TEST(TestJournalPlayer, LiveReplayBlockedNewTag) {
- std::string oid = this->get_temp_oid();
-
- cls::journal::ObjectSetPosition commit_position;
-
- ASSERT_EQ(0, this->create(oid));
- ASSERT_EQ(0, this->client_register(oid));
- ASSERT_EQ(0, this->client_commit(oid, commit_position));
-
- journal::JournalMetadataPtr metadata = this->create_metadata(oid);
- ASSERT_EQ(0, this->init_metadata(metadata));
-
- journal::JournalPlayer *player = this->create_player(oid, metadata);
- BOOST_SCOPE_EXIT_ALL( (player) ) {
- C_SaferCond unwatch_ctx;
- player->shut_down(&unwatch_ctx);
- ASSERT_EQ(0, unwatch_ctx.wait());
- };
-
- C_SaferCond ctx1;
- cls::journal::Tag tag1;
- metadata->allocate_tag(cls::journal::Tag::TAG_CLASS_NEW, {}, &tag1, &ctx1);
- ASSERT_EQ(0, ctx1.wait());
-
- ASSERT_EQ(0, metadata->set_active_set(0));
- ASSERT_EQ(0, this->write_entry(oid, 0, tag1.tid, 0));
- ASSERT_EQ(0, this->write_entry(oid, 1, tag1.tid, 1));
- ASSERT_EQ(0, this->write_entry(oid, 0, tag1.tid, 2));
- ASSERT_EQ(0, this->write_entry(oid, 0, tag1.tid, 4));
- player->prefetch_and_watch(0.25);
-
- Entries entries;
- ASSERT_TRUE(this->wait_for_entries(player, 3, &entries));
-
- Entries expected_entries = {
- this->create_entry(tag1.tid, 0),
- this->create_entry(tag1.tid, 1),
- this->create_entry(tag1.tid, 2)};
- ASSERT_EQ(expected_entries, entries);
-
- journal::Entry entry;
- uint64_t commit_tid;
- ASSERT_FALSE(player->try_pop_front(&entry, &commit_tid));
-
- C_SaferCond ctx2;
- cls::journal::Tag tag2;
- metadata->allocate_tag(tag1.tag_class, {}, &tag2, &ctx2);
- ASSERT_EQ(0, ctx2.wait());
-
- ASSERT_EQ(0, this->write_entry(oid, 0, tag2.tid, 0));
- ASSERT_TRUE(this->wait_for_entries(player, 1, &entries));
-
- expected_entries = {
- this->create_entry(tag2.tid, 0)};
- ASSERT_EQ(expected_entries, entries);
-}
-
-TYPED_TEST(TestJournalPlayer, LiveReplayStaleEntries) {
- std::string oid = this->get_temp_oid();
-
- journal::JournalPlayer::ObjectPositions positions = {
- cls::journal::ObjectPosition(0, 1, 0) };
- cls::journal::ObjectSetPosition commit_position(positions);
-
- ASSERT_EQ(0, this->create(oid));
- ASSERT_EQ(0, this->client_register(oid));
- ASSERT_EQ(0, this->client_commit(oid, commit_position));
-
- journal::JournalMetadataPtr metadata = this->create_metadata(oid);
- ASSERT_EQ(0, this->init_metadata(metadata));
-
- journal::JournalPlayer *player = this->create_player(oid, metadata);
- BOOST_SCOPE_EXIT_ALL( (player) ) {
- C_SaferCond unwatch_ctx;
- player->shut_down(&unwatch_ctx);
- ASSERT_EQ(0, unwatch_ctx.wait());
- };
-
- ASSERT_EQ(0, this->write_entry(oid, 1, 0, 1));
- ASSERT_EQ(0, this->write_entry(oid, 1, 0, 3));
- ASSERT_EQ(0, this->write_entry(oid, 0, 1, 0));
- ASSERT_EQ(0, this->write_entry(oid, 1, 1, 1));
- player->prefetch_and_watch(0.25);
-
- Entries entries;
- ASSERT_TRUE(this->wait_for_entries(player, 1, &entries));
-
- Entries expected_entries = {
- this->create_entry(1, 1)};
- ASSERT_EQ(expected_entries, entries);
-}
-
-TYPED_TEST(TestJournalPlayer, LiveReplayRefetchRemoveEmpty) {
- std::string oid = this->get_temp_oid();
-
- journal::JournalPlayer::ObjectPositions positions = {
- cls::journal::ObjectPosition(1, 0, 1),
- cls::journal::ObjectPosition(0, 0, 0)};
- cls::journal::ObjectSetPosition commit_position(positions);
-
- ASSERT_EQ(0, this->create(oid));
- ASSERT_EQ(0, this->client_register(oid));
- ASSERT_EQ(0, this->client_commit(oid, commit_position));
-
- journal::JournalMetadataPtr metadata = this->create_metadata(oid);
- ASSERT_EQ(0, this->init_metadata(metadata));
-
- journal::JournalPlayer *player = this->create_player(oid, metadata);
- BOOST_SCOPE_EXIT_ALL( (player) ) {
- C_SaferCond unwatch_ctx;
- player->shut_down(&unwatch_ctx);
- ASSERT_EQ(0, unwatch_ctx.wait());
- };
-
- ASSERT_EQ(0, metadata->set_active_set(1));
- ASSERT_EQ(0, this->write_entry(oid, 0, 0, 0));
- ASSERT_EQ(0, this->write_entry(oid, 1, 0, 1));
- ASSERT_EQ(0, this->write_entry(oid, 3, 0, 3));
- ASSERT_EQ(0, this->write_entry(oid, 2, 1, 0));
- player->prefetch_and_watch(0.25);
-
- Entries entries;
- ASSERT_TRUE(this->wait_for_entries(player, 1, &entries));
-
- Entries expected_entries = {
- this->create_entry(1, 0)};
- ASSERT_EQ(expected_entries, entries);
-
- // should remove player for offset 3 after refetching
- ASSERT_EQ(0, metadata->set_active_set(3));
- ASSERT_EQ(0, this->write_entry(oid, 7, 1, 1));
-
- ASSERT_TRUE(this->wait_for_entries(player, 1, &entries));
-
- expected_entries = {
- this->create_entry(1, 1)};
- ASSERT_EQ(expected_entries, entries);
-}
-
-TYPED_TEST(TestJournalPlayer, PrefechShutDown) {
- std::string oid = this->get_temp_oid();
-
- ASSERT_EQ(0, this->create(oid));
- ASSERT_EQ(0, this->client_register(oid));
- ASSERT_EQ(0, this->client_commit(oid, {}));
-
- journal::JournalMetadataPtr metadata = this->create_metadata(oid);
- ASSERT_EQ(0, this->init_metadata(metadata));
-
- journal::JournalPlayer *player = this->create_player(oid, metadata);
- BOOST_SCOPE_EXIT_ALL( (player) ) {
- C_SaferCond unwatch_ctx;
- player->shut_down(&unwatch_ctx);
- ASSERT_EQ(0, unwatch_ctx.wait());
- };
- player->prefetch();
-}
-
-TYPED_TEST(TestJournalPlayer, LiveReplayShutDown) {
- std::string oid = this->get_temp_oid();
-
- ASSERT_EQ(0, this->create(oid));
- ASSERT_EQ(0, this->client_register(oid));
- ASSERT_EQ(0, this->client_commit(oid, {}));
-
- journal::JournalMetadataPtr metadata = this->create_metadata(oid);
- ASSERT_EQ(0, this->init_metadata(metadata));
-
- journal::JournalPlayer *player = this->create_player(oid, metadata);
- BOOST_SCOPE_EXIT_ALL( (player) ) {
- C_SaferCond unwatch_ctx;
- player->shut_down(&unwatch_ctx);
- ASSERT_EQ(0, unwatch_ctx.wait());
- };
- player->prefetch_and_watch(0.25);
-}
-
Code Review / stor4nfv.git / blobdiff