+++ /dev/null
-// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
-// vim: ts=8 sw=2 smarttab
-
-#include "journal/ObjectRecorder.h"
-#include "common/Cond.h"
-#include "common/Mutex.h"
-#include "common/Timer.h"
-#include "gtest/gtest.h"
-#include "test/librados/test.h"
-#include "test/journal/RadosTestFixture.h"
-#include <limits>
-
-using std::shared_ptr;
-
-class TestObjectRecorder : public RadosTestFixture {
-public:
- TestObjectRecorder()
- : m_flush_interval(std::numeric_limits<uint32_t>::max()),
- m_flush_bytes(std::numeric_limits<uint64_t>::max()),
- m_flush_age(600)
- {
- }
-
- struct Handler : public journal::ObjectRecorder::Handler {
- Mutex lock;
- shared_ptr<Mutex> object_lock;
- Cond cond;
- bool is_closed = false;
- uint32_t overflows = 0;
-
- Handler() : lock("lock") {
- }
-
- void closed(journal::ObjectRecorder *object_recorder) override {
- Mutex::Locker locker(lock);
- is_closed = true;
- cond.Signal();
- }
- void overflow(journal::ObjectRecorder *object_recorder) override {
- Mutex::Locker locker(lock);
- journal::AppendBuffers append_buffers;
- object_lock->Lock();
- object_recorder->claim_append_buffers(&append_buffers);
- object_lock->Unlock();
-
- ++overflows;
- cond.Signal();
- }
- };
-
- typedef std::list<journal::ObjectRecorderPtr> ObjectRecorders;
- typedef std::map<std::string, shared_ptr<Mutex>> ObjectRecorderLocksMap;
-
- ObjectRecorders m_object_recorders;
- ObjectRecorderLocksMap m_object_recorder_locks;
-
- uint32_t m_flush_interval;
- uint64_t m_flush_bytes;
- double m_flush_age;
- Handler m_handler;
-
- void TearDown() override {
- for (ObjectRecorders::iterator it = m_object_recorders.begin();
- it != m_object_recorders.end(); ++it) {
- C_SaferCond cond;
- (*it)->flush(&cond);
- cond.wait();
- }
- m_object_recorders.clear();
-
- RadosTestFixture::TearDown();
- }
-
- inline void set_flush_interval(uint32_t i) {
- m_flush_interval = i;
- }
- inline void set_flush_bytes(uint64_t i) {
- m_flush_bytes = i;
- }
- inline void set_flush_age(double i) {
- m_flush_age = i;
- }
-
- journal::AppendBuffer create_append_buffer(uint64_t tag_tid, uint64_t entry_tid,
- const std::string &payload) {
- journal::FutureImplPtr future(new journal::FutureImpl(tag_tid, entry_tid,
- 456));
- future->init(journal::FutureImplPtr());
-
- bufferlist bl;
- bl.append(payload);
- return std::make_pair(future, bl);
- }
-
- journal::ObjectRecorderPtr create_object(const std::string &oid,
- uint8_t order, shared_ptr<Mutex> lock) {
- journal::ObjectRecorderPtr object(new journal::ObjectRecorder(
- m_ioctx, oid, 0, lock, m_work_queue, *m_timer, m_timer_lock, &m_handler,
- order, m_flush_interval, m_flush_bytes, m_flush_age));
- m_object_recorders.push_back(object);
- m_object_recorder_locks.insert(std::make_pair(oid, lock));
- m_handler.object_lock = lock;
- return object;
- }
-};
-
-TEST_F(TestObjectRecorder, Append) {
- std::string oid = get_temp_oid();
- ASSERT_EQ(0, create(oid));
- ASSERT_EQ(0, client_register(oid));
- journal::JournalMetadataPtr metadata = create_metadata(oid);
- ASSERT_EQ(0, init_metadata(metadata));
-
- shared_ptr<Mutex> lock(new Mutex("object_recorder_lock"));
- journal::ObjectRecorderPtr object = create_object(oid, 24, lock);
-
- journal::AppendBuffer append_buffer1 = create_append_buffer(234, 123,
- "payload");
- journal::AppendBuffers append_buffers;
- append_buffers = {append_buffer1};
- lock->Lock();
- ASSERT_FALSE(object->append_unlock(std::move(append_buffers)));
- ASSERT_EQ(1U, object->get_pending_appends());
-
- journal::AppendBuffer append_buffer2 = create_append_buffer(234, 124,
- "payload");
- append_buffers = {append_buffer2};
- lock->Lock();
- ASSERT_FALSE(object->append_unlock(std::move(append_buffers)));
- ASSERT_EQ(2U, object->get_pending_appends());
-
- C_SaferCond cond;
- append_buffer2.first->flush(&cond);
- ASSERT_EQ(0, cond.wait());
- ASSERT_EQ(0U, object->get_pending_appends());
-}
-
-TEST_F(TestObjectRecorder, AppendFlushByCount) {
- std::string oid = get_temp_oid();
- ASSERT_EQ(0, create(oid));
- ASSERT_EQ(0, client_register(oid));
- journal::JournalMetadataPtr metadata = create_metadata(oid);
- ASSERT_EQ(0, init_metadata(metadata));
-
- set_flush_interval(2);
- shared_ptr<Mutex> lock(new Mutex("object_recorder_lock"));
- journal::ObjectRecorderPtr object = create_object(oid, 24, lock);
-
- journal::AppendBuffer append_buffer1 = create_append_buffer(234, 123,
- "payload");
- journal::AppendBuffers append_buffers;
- append_buffers = {append_buffer1};
- lock->Lock();
- ASSERT_FALSE(object->append_unlock(std::move(append_buffers)));
- ASSERT_EQ(1U, object->get_pending_appends());
-
- journal::AppendBuffer append_buffer2 = create_append_buffer(234, 124,
- "payload");
- append_buffers = {append_buffer2};
- lock->Lock();
- ASSERT_FALSE(object->append_unlock(std::move(append_buffers)));
- ASSERT_EQ(0U, object->get_pending_appends());
-
- C_SaferCond cond;
- append_buffer2.first->wait(&cond);
- ASSERT_EQ(0, cond.wait());
-}
-
-TEST_F(TestObjectRecorder, AppendFlushByBytes) {
- std::string oid = get_temp_oid();
- ASSERT_EQ(0, create(oid));
- ASSERT_EQ(0, client_register(oid));
- journal::JournalMetadataPtr metadata = create_metadata(oid);
- ASSERT_EQ(0, init_metadata(metadata));
-
- set_flush_bytes(10);
- shared_ptr<Mutex> lock(new Mutex("object_recorder_lock"));
- journal::ObjectRecorderPtr object = create_object(oid, 24, lock);
-
- journal::AppendBuffer append_buffer1 = create_append_buffer(234, 123,
- "payload");
- journal::AppendBuffers append_buffers;
- append_buffers = {append_buffer1};
- lock->Lock();
- ASSERT_FALSE(object->append_unlock(std::move(append_buffers)));
- ASSERT_EQ(1U, object->get_pending_appends());
-
- journal::AppendBuffer append_buffer2 = create_append_buffer(234, 124,
- "payload");
- append_buffers = {append_buffer2};
- lock->Lock();
- ASSERT_FALSE(object->append_unlock(std::move(append_buffers)));
- ASSERT_EQ(0U, object->get_pending_appends());
-
- C_SaferCond cond;
- append_buffer2.first->wait(&cond);
- ASSERT_EQ(0, cond.wait());
-}
-
-TEST_F(TestObjectRecorder, AppendFlushByAge) {
- std::string oid = get_temp_oid();
- ASSERT_EQ(0, create(oid));
- ASSERT_EQ(0, client_register(oid));
- journal::JournalMetadataPtr metadata = create_metadata(oid);
- ASSERT_EQ(0, init_metadata(metadata));
-
- set_flush_age(0.1);
- shared_ptr<Mutex> lock(new Mutex("object_recorder_lock"));
- journal::ObjectRecorderPtr object = create_object(oid, 24, lock);
-
- journal::AppendBuffer append_buffer1 = create_append_buffer(234, 123,
- "payload");
- journal::AppendBuffers append_buffers;
- append_buffers = {append_buffer1};
- lock->Lock();
- ASSERT_FALSE(object->append_unlock(std::move(append_buffers)));
-
- journal::AppendBuffer append_buffer2 = create_append_buffer(234, 124,
- "payload");
- append_buffers = {append_buffer2};
- lock->Lock();
- ASSERT_FALSE(object->append_unlock(std::move(append_buffers)));
-
- C_SaferCond cond;
- append_buffer2.first->wait(&cond);
- ASSERT_EQ(0, cond.wait());
- ASSERT_EQ(0U, object->get_pending_appends());
-}
-
-TEST_F(TestObjectRecorder, AppendFilledObject) {
- std::string oid = get_temp_oid();
- ASSERT_EQ(0, create(oid));
- ASSERT_EQ(0, client_register(oid));
- journal::JournalMetadataPtr metadata = create_metadata(oid);
- ASSERT_EQ(0, init_metadata(metadata));
-
- shared_ptr<Mutex> lock(new Mutex("object_recorder_lock"));
- journal::ObjectRecorderPtr object = create_object(oid, 12, lock);
-
- std::string payload(2048, '1');
- journal::AppendBuffer append_buffer1 = create_append_buffer(234, 123,
- payload);
- journal::AppendBuffers append_buffers;
- append_buffers = {append_buffer1};
- lock->Lock();
- ASSERT_FALSE(object->append_unlock(std::move(append_buffers)));
-
- journal::AppendBuffer append_buffer2 = create_append_buffer(234, 124,
- payload);
- append_buffers = {append_buffer2};
- lock->Lock();
- ASSERT_TRUE(object->append_unlock(std::move(append_buffers)));
-
- C_SaferCond cond;
- append_buffer2.first->wait(&cond);
- ASSERT_EQ(0, cond.wait());
- ASSERT_EQ(0U, object->get_pending_appends());
-}
-
-TEST_F(TestObjectRecorder, Flush) {
- std::string oid = get_temp_oid();
- ASSERT_EQ(0, create(oid));
- ASSERT_EQ(0, client_register(oid));
- journal::JournalMetadataPtr metadata = create_metadata(oid);
- ASSERT_EQ(0, init_metadata(metadata));
-
- shared_ptr<Mutex> lock(new Mutex("object_recorder_lock"));
- journal::ObjectRecorderPtr object = create_object(oid, 24, lock);
-
- journal::AppendBuffer append_buffer1 = create_append_buffer(234, 123,
- "payload");
- journal::AppendBuffers append_buffers;
- append_buffers = {append_buffer1};
- lock->Lock();
- ASSERT_FALSE(object->append_unlock(std::move(append_buffers)));
- ASSERT_EQ(1U, object->get_pending_appends());
-
- C_SaferCond cond1;
- object->flush(&cond1);
- ASSERT_EQ(0, cond1.wait());
-
- C_SaferCond cond2;
- append_buffer1.first->wait(&cond2);
- ASSERT_EQ(0, cond2.wait());
- ASSERT_EQ(0U, object->get_pending_appends());
-}
-
-TEST_F(TestObjectRecorder, FlushFuture) {
- std::string oid = get_temp_oid();
- ASSERT_EQ(0, create(oid));
- ASSERT_EQ(0, client_register(oid));
- journal::JournalMetadataPtr metadata = create_metadata(oid);
- ASSERT_EQ(0, init_metadata(metadata));
-
- shared_ptr<Mutex> lock(new Mutex("object_recorder_lock"));
- journal::ObjectRecorderPtr object = create_object(oid, 24, lock);
-
- journal::AppendBuffer append_buffer = create_append_buffer(234, 123,
- "payload");
- journal::AppendBuffers append_buffers;
- append_buffers = {append_buffer};
- lock->Lock();
- ASSERT_FALSE(object->append_unlock(std::move(append_buffers)));
- ASSERT_EQ(1U, object->get_pending_appends());
-
- C_SaferCond cond;
- append_buffer.first->wait(&cond);
- lock->Lock();
- object->flush(append_buffer.first);
- ASSERT_TRUE(lock->is_locked());
- lock->Unlock();
- ASSERT_TRUE(append_buffer.first->is_flush_in_progress() ||
- append_buffer.first->is_complete());
- ASSERT_EQ(0, cond.wait());
-}
-
-TEST_F(TestObjectRecorder, FlushDetachedFuture) {
- std::string oid = get_temp_oid();
- ASSERT_EQ(0, create(oid));
- ASSERT_EQ(0, client_register(oid));
- journal::JournalMetadataPtr metadata = create_metadata(oid);
- ASSERT_EQ(0, init_metadata(metadata));
-
- shared_ptr<Mutex> lock(new Mutex("object_recorder_lock"));
- journal::ObjectRecorderPtr object = create_object(oid, 24, lock);
-
- journal::AppendBuffer append_buffer = create_append_buffer(234, 123,
- "payload");
-
- journal::AppendBuffers append_buffers;
- append_buffers = {append_buffer};
-
- lock->Lock();
- object->flush(append_buffer.first);
- ASSERT_TRUE(lock->is_locked());
- lock->Unlock();
- ASSERT_FALSE(append_buffer.first->is_flush_in_progress());
- lock->Lock();
- ASSERT_FALSE(object->append_unlock(std::move(append_buffers)));
-
- // should automatically flush once its attached to the object
- C_SaferCond cond;
- append_buffer.first->wait(&cond);
- ASSERT_EQ(0, cond.wait());
-}
-
-TEST_F(TestObjectRecorder, Close) {
- std::string oid = get_temp_oid();
- ASSERT_EQ(0, create(oid));
- ASSERT_EQ(0, client_register(oid));
- journal::JournalMetadataPtr metadata = create_metadata(oid);
- ASSERT_EQ(0, init_metadata(metadata));
-
- set_flush_interval(2);
- shared_ptr<Mutex> lock(new Mutex("object_recorder_lock"));
- journal::ObjectRecorderPtr object = create_object(oid, 24, lock);
-
- journal::AppendBuffer append_buffer1 = create_append_buffer(234, 123,
- "payload");
- journal::AppendBuffers append_buffers;
- append_buffers = {append_buffer1};
- lock->Lock();
- ASSERT_FALSE(object->append_unlock(std::move(append_buffers)));
- ASSERT_EQ(1U, object->get_pending_appends());
-
- lock->Lock();
- ASSERT_FALSE(object->close());
- ASSERT_TRUE(lock->is_locked());
- lock->Unlock();
-
- {
- Mutex::Locker locker(m_handler.lock);
- while (!m_handler.is_closed) {
- if (m_handler.cond.WaitInterval(
- m_handler.lock, utime_t(10, 0)) != 0) {
- break;
- }
- }
- }
-
- ASSERT_TRUE(m_handler.is_closed);
- ASSERT_EQ(0U, object->get_pending_appends());
-}
-
-TEST_F(TestObjectRecorder, Overflow) {
- std::string oid = get_temp_oid();
- ASSERT_EQ(0, create(oid));
- ASSERT_EQ(0, client_register(oid));
- journal::JournalMetadataPtr metadata = create_metadata(oid);
- ASSERT_EQ(0, init_metadata(metadata));
-
- shared_ptr<Mutex> lock1(new Mutex("object_recorder_lock_1"));
- journal::ObjectRecorderPtr object1 = create_object(oid, 12, lock1);
- shared_ptr<Mutex> lock2(new Mutex("object_recorder_lock_2"));
- journal::ObjectRecorderPtr object2 = create_object(oid, 12, lock2);
-
- std::string payload(2048, '1');
- journal::AppendBuffer append_buffer1 = create_append_buffer(234, 123,
- payload);
- journal::AppendBuffer append_buffer2 = create_append_buffer(234, 124,
- payload);
- journal::AppendBuffers append_buffers;
- append_buffers = {append_buffer1, append_buffer2};
- lock1->Lock();
- ASSERT_TRUE(object1->append_unlock(std::move(append_buffers)));
-
- C_SaferCond cond;
- append_buffer2.first->wait(&cond);
- ASSERT_EQ(0, cond.wait());
- ASSERT_EQ(0U, object1->get_pending_appends());
-
- journal::AppendBuffer append_buffer3 = create_append_buffer(456, 123,
- payload);
- append_buffers = {append_buffer3};
-
- lock2->Lock();
- ASSERT_FALSE(object2->append_unlock(std::move(append_buffers)));
- append_buffer3.first->flush(NULL);
-
- bool overflowed = false;
- {
- Mutex::Locker locker(m_handler.lock);
- while (m_handler.overflows == 0) {
- if (m_handler.cond.WaitInterval(
- m_handler.lock, utime_t(10, 0)) != 0) {
- break;
- }
- }
- if (m_handler.overflows != 0) {
- overflowed = true;
- }
- }
-
- ASSERT_TRUE(overflowed);
-}
Code Review / stor4nfv.git / blobdiff