--- /dev/null
+// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
+// vim: ts=8 sw=2 smarttab
+
+#include "journal/JournalMetadata.h"
+#include "test/journal/RadosTestFixture.h"
+#include "common/Cond.h"
+#include "common/Mutex.h"
+#include <map>
+
+class TestJournalMetadata : public RadosTestFixture {
+public:
+ void TearDown() override {
+ for (MetadataList::iterator it = m_metadata_list.begin();
+ it != m_metadata_list.end(); ++it) {
+ (*it)->remove_listener(&m_listener);
+ }
+ m_metadata_list.clear();
+
+ RadosTestFixture::TearDown();
+ }
+
+ journal::JournalMetadataPtr create_metadata(const std::string &oid,
+ const std::string &client_id,
+ double commit_interval = 0.1,
+ uint64_t max_fetch_bytes = 0,
+ int max_concurrent_object_sets = 0) {
+ journal::JournalMetadataPtr metadata = RadosTestFixture::create_metadata(
+ oid, client_id, commit_interval, max_fetch_bytes,
+ max_concurrent_object_sets);
+ m_metadata_list.push_back(metadata);
+ metadata->add_listener(&m_listener);
+ return metadata;
+ }
+
+ typedef std::list<journal::JournalMetadataPtr> MetadataList;
+ MetadataList m_metadata_list;
+};
+
+TEST_F(TestJournalMetadata, JournalDNE) {
+ std::string oid = get_temp_oid();
+
+ journal::JournalMetadataPtr metadata1 = create_metadata(oid, "client1");
+ ASSERT_EQ(-ENOENT, init_metadata(metadata1));
+}
+
+TEST_F(TestJournalMetadata, ClientDNE) {
+ std::string oid = get_temp_oid();
+
+ ASSERT_EQ(0, create(oid, 14, 2));
+ ASSERT_EQ(0, client_register(oid, "client1", ""));
+
+ journal::JournalMetadataPtr metadata1 = create_metadata(oid, "client1");
+ ASSERT_EQ(0, init_metadata(metadata1));
+
+ journal::JournalMetadataPtr metadata2 = create_metadata(oid, "client2");
+ ASSERT_EQ(-ENOENT, init_metadata(metadata2));
+}
+
+TEST_F(TestJournalMetadata, Committed) {
+ std::string oid = get_temp_oid();
+
+ ASSERT_EQ(0, create(oid, 14, 2));
+ ASSERT_EQ(0, client_register(oid, "client1", ""));
+
+ journal::JournalMetadataPtr metadata1 = create_metadata(oid, "client1", 600);
+ ASSERT_EQ(0, init_metadata(metadata1));
+
+ journal::JournalMetadataPtr metadata2 = create_metadata(oid, "client1");
+ ASSERT_EQ(0, init_metadata(metadata2));
+ ASSERT_TRUE(wait_for_update(metadata2));
+
+ journal::JournalMetadata::ObjectSetPosition expect_commit_position;
+ journal::JournalMetadata::ObjectSetPosition read_commit_position;
+ metadata1->get_commit_position(&read_commit_position);
+ ASSERT_EQ(expect_commit_position, read_commit_position);
+
+ uint64_t commit_tid1 = metadata1->allocate_commit_tid(0, 0, 0);
+ uint64_t commit_tid2 = metadata1->allocate_commit_tid(0, 1, 0);
+ uint64_t commit_tid3 = metadata1->allocate_commit_tid(1, 0, 1);
+ uint64_t commit_tid4 = metadata1->allocate_commit_tid(0, 0, 2);
+
+ // cannot commit until tid1 + 2 committed
+ metadata1->committed(commit_tid2, []() { return nullptr; });
+ metadata1->committed(commit_tid3, []() { return nullptr; });
+
+ C_SaferCond cond1;
+ metadata1->committed(commit_tid1, [&cond1]() { return &cond1; });
+
+ // given our 10 minute commit internal, this should override the
+ // in-flight commit
+ C_SaferCond cond2;
+ metadata1->committed(commit_tid4, [&cond2]() { return &cond2; });
+
+ ASSERT_EQ(-ESTALE, cond1.wait());
+ metadata1->flush_commit_position();
+ ASSERT_EQ(0, cond2.wait());
+
+ ASSERT_TRUE(wait_for_update(metadata2));
+ metadata2->get_commit_position(&read_commit_position);
+ expect_commit_position = {{{0, 0, 2}, {1, 0, 1}}};
+ ASSERT_EQ(expect_commit_position, read_commit_position);
+}
+
+TEST_F(TestJournalMetadata, UpdateActiveObject) {
+ std::string oid = get_temp_oid();
+
+ ASSERT_EQ(0, create(oid, 14, 2));
+ ASSERT_EQ(0, client_register(oid, "client1", ""));
+
+ journal::JournalMetadataPtr metadata1 = create_metadata(oid, "client1");
+ ASSERT_EQ(0, init_metadata(metadata1));
+ ASSERT_TRUE(wait_for_update(metadata1));
+
+ ASSERT_EQ(0U, metadata1->get_active_set());
+
+ ASSERT_EQ(0, metadata1->set_active_set(123));
+ ASSERT_TRUE(wait_for_update(metadata1));
+
+ ASSERT_EQ(123U, metadata1->get_active_set());
+}
+
+TEST_F(TestJournalMetadata, DisconnectLaggyClient) {
+ std::string oid = get_temp_oid();
+
+ ASSERT_EQ(0, create(oid));
+ ASSERT_EQ(0, client_register(oid, "client1", ""));
+ ASSERT_EQ(0, client_register(oid, "client2", "laggy"));
+
+ int max_concurrent_object_sets = 100;
+ journal::JournalMetadataPtr metadata =
+ create_metadata(oid, "client1", 0.1, 0, max_concurrent_object_sets);
+ ASSERT_EQ(0, init_metadata(metadata));
+ ASSERT_TRUE(wait_for_update(metadata));
+
+ ASSERT_EQ(0U, metadata->get_active_set());
+
+ journal::JournalMetadata::RegisteredClients clients;
+
+#define ASSERT_CLIENT_STATES(s1, s2) \
+ ASSERT_EQ(2U, clients.size()); \
+ for (auto &c : clients) { \
+ if (c.id == "client1") { \
+ ASSERT_EQ(c.state, s1); \
+ } else if (c.id == "client2") { \
+ ASSERT_EQ(c.state, s2); \
+ } else { \
+ ASSERT_TRUE(false); \
+ } \
+ }
+
+ metadata->get_registered_clients(&clients);
+ ASSERT_CLIENT_STATES(cls::journal::CLIENT_STATE_CONNECTED,
+ cls::journal::CLIENT_STATE_CONNECTED);
+
+ // client2 is connected when active set <= max_concurrent_object_sets
+ ASSERT_EQ(0, metadata->set_active_set(max_concurrent_object_sets));
+ ASSERT_TRUE(wait_for_update(metadata));
+ uint64_t commit_tid = metadata->allocate_commit_tid(0, 0, 0);
+ C_SaferCond cond1;
+ metadata->committed(commit_tid, [&cond1]() { return &cond1; });
+ ASSERT_EQ(0, cond1.wait());
+ metadata->flush_commit_position();
+ ASSERT_TRUE(wait_for_update(metadata));
+ ASSERT_EQ(100U, metadata->get_active_set());
+ clients.clear();
+ metadata->get_registered_clients(&clients);
+ ASSERT_CLIENT_STATES(cls::journal::CLIENT_STATE_CONNECTED,
+ cls::journal::CLIENT_STATE_CONNECTED);
+
+ // client2 is disconnected when active set > max_concurrent_object_sets
+ ASSERT_EQ(0, metadata->set_active_set(max_concurrent_object_sets + 1));
+ ASSERT_TRUE(wait_for_update(metadata));
+ commit_tid = metadata->allocate_commit_tid(0, 0, 1);
+ C_SaferCond cond2;
+ metadata->committed(commit_tid, [&cond2]() { return &cond2; });
+ ASSERT_EQ(0, cond2.wait());
+ metadata->flush_commit_position();
+ ASSERT_TRUE(wait_for_update(metadata));
+ ASSERT_EQ(101U, metadata->get_active_set());
+ clients.clear();
+ metadata->get_registered_clients(&clients);
+ ASSERT_CLIENT_STATES(cls::journal::CLIENT_STATE_CONNECTED,
+ cls::journal::CLIENT_STATE_DISCONNECTED);
+}
+
+TEST_F(TestJournalMetadata, AssertActiveTag) {
+ std::string oid = get_temp_oid();
+
+ ASSERT_EQ(0, create(oid));
+ ASSERT_EQ(0, client_register(oid, "client1", ""));
+
+ journal::JournalMetadataPtr metadata = create_metadata(oid, "client1");
+ ASSERT_EQ(0, init_metadata(metadata));
+ ASSERT_TRUE(wait_for_update(metadata));
+
+ C_SaferCond ctx1;
+ cls::journal::Tag tag1;
+ metadata->allocate_tag(cls::journal::Tag::TAG_CLASS_NEW, {}, &tag1, &ctx1);
+ ASSERT_EQ(0, ctx1.wait());
+
+ C_SaferCond ctx2;
+ metadata->assert_active_tag(tag1.tid, &ctx2);
+ ASSERT_EQ(0, ctx2.wait());
+
+ C_SaferCond ctx3;
+ cls::journal::Tag tag2;
+ metadata->allocate_tag(tag1.tag_class, {}, &tag2, &ctx3);
+ ASSERT_EQ(0, ctx3.wait());
+
+ C_SaferCond ctx4;
+ metadata->assert_active_tag(tag1.tid, &ctx4);
+ ASSERT_EQ(-ESTALE, ctx4.wait());
+}
Code Review / stor4nfv.git / blobdiff