[ceph.git] / ceph / src / test / journal / test_ObjectRecorder.cc

// -*- 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);
}
Commit	Line	Data
7c673cae FG	1	// -- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t --
	2	// vim: ts=8 sw=2 smarttab
	3
	4	#include "journal/ObjectRecorder.h"
	5	#include "common/Cond.h"
	6	#include "common/Mutex.h"
	7	#include "common/Timer.h"
	8	#include "gtest/gtest.h"
	9	#include "test/librados/test.h"
	10	#include "test/journal/RadosTestFixture.h"
	11	#include <limits>
	12
	13	using std::shared_ptr;
	14
	15	class TestObjectRecorder : public RadosTestFixture {
	16	public:
	17	TestObjectRecorder()
	18	: m_flush_interval(std::numeric_limits<uint32_t>::max()),
	19	m_flush_bytes(std::numeric_limits<uint64_t>::max()),
	20	m_flush_age(600)
	21	{
	22	}
	23
	24	struct Handler : public journal::ObjectRecorder::Handler {
	25	Mutex lock;
	26	shared_ptr<Mutex> object_lock;
	27	Cond cond;
	28	bool is_closed = false;
	29	uint32_t overflows = 0;
	30
	31	Handler() : lock("lock") {
	32	}
	33
	34	void closed(journal::ObjectRecorder *object_recorder) override {
	35	Mutex::Locker locker(lock);
	36	is_closed = true;
	37	cond.Signal();
	38	}
	39	void overflow(journal::ObjectRecorder *object_recorder) override {
	40	Mutex::Locker locker(lock);
	41	journal::AppendBuffers append_buffers;
	42	object_lock->Lock();
	43	object_recorder->claim_append_buffers(&append_buffers);
	44	object_lock->Unlock();
	45
	46	++overflows;
	47	cond.Signal();
	48	}
	49	};
	50
	51	typedef std::list<journal::ObjectRecorderPtr> ObjectRecorders;
	52	typedef std::map<std::string, shared_ptr<Mutex>> ObjectRecorderLocksMap;
	53
	54	ObjectRecorders m_object_recorders;
	55	ObjectRecorderLocksMap m_object_recorder_locks;
	56
	57	uint32_t m_flush_interval;
	58	uint64_t m_flush_bytes;
	59	double m_flush_age;
	60	Handler m_handler;
	61
	62	void TearDown() override {
	63	for (ObjectRecorders::iterator it = m_object_recorders.begin();
	64	it != m_object_recorders.end(); ++it) {
65	C_SaferCond cond;
66	(*it)->flush(&cond);
67	cond.wait();
68	}
69	m_object_recorders.clear();
70
71	RadosTestFixture::TearDown();
72	}
73
74	inline void set_flush_interval(uint32_t i) {
75	m_flush_interval = i;
76	}
77	inline void set_flush_bytes(uint64_t i) {
78	m_flush_bytes = i;
79	}
80	inline void set_flush_age(double i) {
81	m_flush_age = i;
82	}
83
84	journal::AppendBuffer create_append_buffer(uint64_t tag_tid, uint64_t entry_tid,
85	const std::string &payload) {
86	journal::FutureImplPtr future(new journal::FutureImpl(tag_tid, entry_tid,
87	456));
88	future->init(journal::FutureImplPtr());
89
90	bufferlist bl;
91	bl.append(payload);
92	return std::make_pair(future, bl);
93	}
94
95	journal::ObjectRecorderPtr create_object(const std::string &oid,
96	uint8_t order, shared_ptr<Mutex> lock) {
97	journal::ObjectRecorderPtr object(new journal::ObjectRecorder(
98	m_ioctx, oid, 0, lock, m_work_queue, *m_timer, m_timer_lock, &m_handler,
99	order, m_flush_interval, m_flush_bytes, m_flush_age));
100	m_object_recorders.push_back(object);
101	m_object_recorder_locks.insert(std::make_pair(oid, lock));
102	m_handler.object_lock = lock;
103	return object;
104	}
105	};
106
107	TEST_F(TestObjectRecorder, Append) {
108	std::string oid = get_temp_oid();
109	ASSERT_EQ(0, create(oid));
110	ASSERT_EQ(0, client_register(oid));
111	journal::JournalMetadataPtr metadata = create_metadata(oid);
112	ASSERT_EQ(0, init_metadata(metadata));
113
114	shared_ptr<Mutex> lock(new Mutex("object_recorder_lock"));
115	journal::ObjectRecorderPtr object = create_object(oid, 24, lock);
116
117	journal::AppendBuffer append_buffer1 = create_append_buffer(234, 123,
118	"payload");
119	journal::AppendBuffers append_buffers;
120	append_buffers = {append_buffer1};
121	lock->Lock();
122	ASSERT_FALSE(object->append_unlock(std::move(append_buffers)));
123	ASSERT_EQ(1U, object->get_pending_appends());
124
125	journal::AppendBuffer append_buffer2 = create_append_buffer(234, 124,
126	"payload");
127	append_buffers = {append_buffer2};
128	lock->Lock();
129	ASSERT_FALSE(object->append_unlock(std::move(append_buffers)));
130	ASSERT_EQ(2U, object->get_pending_appends());
131
132	C_SaferCond cond;
133	append_buffer2.first->flush(&cond);
134	ASSERT_EQ(0, cond.wait());
135	ASSERT_EQ(0U, object->get_pending_appends());
136	}
137
138	TEST_F(TestObjectRecorder, AppendFlushByCount) {
139	std::string oid = get_temp_oid();
140	ASSERT_EQ(0, create(oid));
141	ASSERT_EQ(0, client_register(oid));
142	journal::JournalMetadataPtr metadata = create_metadata(oid);
143	ASSERT_EQ(0, init_metadata(metadata));
144
145	set_flush_interval(2);
146	shared_ptr<Mutex> lock(new Mutex("object_recorder_lock"));
147	journal::ObjectRecorderPtr object = create_object(oid, 24, lock);
148
149	journal::AppendBuffer append_buffer1 = create_append_buffer(234, 123,
150	"payload");
151	journal::AppendBuffers append_buffers;
152	append_buffers = {append_buffer1};
153	lock->Lock();
154	ASSERT_FALSE(object->append_unlock(std::move(append_buffers)));
155	ASSERT_EQ(1U, object->get_pending_appends());
156
157	journal::AppendBuffer append_buffer2 = create_append_buffer(234, 124,
158	"payload");
159	append_buffers = {append_buffer2};
160	lock->Lock();
161	ASSERT_FALSE(object->append_unlock(std::move(append_buffers)));
162	ASSERT_EQ(0U, object->get_pending_appends());
163
164	C_SaferCond cond;
165	append_buffer2.first->wait(&cond);
166	ASSERT_EQ(0, cond.wait());
167	}
168
169	TEST_F(TestObjectRecorder, AppendFlushByBytes) {
170	std::string oid = get_temp_oid();
171	ASSERT_EQ(0, create(oid));
172	ASSERT_EQ(0, client_register(oid));
173	journal::JournalMetadataPtr metadata = create_metadata(oid);
174	ASSERT_EQ(0, init_metadata(metadata));
175
176	set_flush_bytes(10);
177	shared_ptr<Mutex> lock(new Mutex("object_recorder_lock"));
178	journal::ObjectRecorderPtr object = create_object(oid, 24, lock);
179
180	journal::AppendBuffer append_buffer1 = create_append_buffer(234, 123,
181	"payload");
182	journal::AppendBuffers append_buffers;
183	append_buffers = {append_buffer1};
184	lock->Lock();
185	ASSERT_FALSE(object->append_unlock(std::move(append_buffers)));
186	ASSERT_EQ(1U, object->get_pending_appends());
187
188	journal::AppendBuffer append_buffer2 = create_append_buffer(234, 124,
189	"payload");
190	append_buffers = {append_buffer2};
191	lock->Lock();
192	ASSERT_FALSE(object->append_unlock(std::move(append_buffers)));
193	ASSERT_EQ(0U, object->get_pending_appends());
194
195	C_SaferCond cond;
196	append_buffer2.first->wait(&cond);
197	ASSERT_EQ(0, cond.wait());
198	}
199
200	TEST_F(TestObjectRecorder, AppendFlushByAge) {
201	std::string oid = get_temp_oid();
202	ASSERT_EQ(0, create(oid));
203	ASSERT_EQ(0, client_register(oid));
204	journal::JournalMetadataPtr metadata = create_metadata(oid);
205	ASSERT_EQ(0, init_metadata(metadata));
206
207	set_flush_age(0.1);
208	shared_ptr<Mutex> lock(new Mutex("object_recorder_lock"));
209	journal::ObjectRecorderPtr object = create_object(oid, 24, lock);
210
211	journal::AppendBuffer append_buffer1 = create_append_buffer(234, 123,
212	"payload");
213	journal::AppendBuffers append_buffers;
214	append_buffers = {append_buffer1};
215	lock->Lock();
216	ASSERT_FALSE(object->append_unlock(std::move(append_buffers)));
217
218	journal::AppendBuffer append_buffer2 = create_append_buffer(234, 124,
219	"payload");
220	append_buffers = {append_buffer2};
221	lock->Lock();
222	ASSERT_FALSE(object->append_unlock(std::move(append_buffers)));
223
224	C_SaferCond cond;
225	append_buffer2.first->wait(&cond);
226	ASSERT_EQ(0, cond.wait());
227	ASSERT_EQ(0U, object->get_pending_appends());
228	}
229
230	TEST_F(TestObjectRecorder, AppendFilledObject) {
231	std::string oid = get_temp_oid();
232	ASSERT_EQ(0, create(oid));
233	ASSERT_EQ(0, client_register(oid));
234	journal::JournalMetadataPtr metadata = create_metadata(oid);
235	ASSERT_EQ(0, init_metadata(metadata));
236
237	shared_ptr<Mutex> lock(new Mutex("object_recorder_lock"));
238	journal::ObjectRecorderPtr object = create_object(oid, 12, lock);
239
240	std::string payload(2048, '1');
241	journal::AppendBuffer append_buffer1 = create_append_buffer(234, 123,
242	payload);
243	journal::AppendBuffers append_buffers;
244	append_buffers = {append_buffer1};
245	lock->Lock();
246	ASSERT_FALSE(object->append_unlock(std::move(append_buffers)));
247
248	journal::AppendBuffer append_buffer2 = create_append_buffer(234, 124,
249	payload);
250	append_buffers = {append_buffer2};
251	lock->Lock();
252	ASSERT_TRUE(object->append_unlock(std::move(append_buffers)));
253
254	C_SaferCond cond;
255	append_buffer2.first->wait(&cond);
256	ASSERT_EQ(0, cond.wait());
257	ASSERT_EQ(0U, object->get_pending_appends());
258	}
259
260	TEST_F(TestObjectRecorder, Flush) {
261	std::string oid = get_temp_oid();
262	ASSERT_EQ(0, create(oid));
263	ASSERT_EQ(0, client_register(oid));
264	journal::JournalMetadataPtr metadata = create_metadata(oid);
265	ASSERT_EQ(0, init_metadata(metadata));
266
267	shared_ptr<Mutex> lock(new Mutex("object_recorder_lock"));
268	journal::ObjectRecorderPtr object = create_object(oid, 24, lock);
269
270	journal::AppendBuffer append_buffer1 = create_append_buffer(234, 123,
271	"payload");
272	journal::AppendBuffers append_buffers;
273	append_buffers = {append_buffer1};
274	lock->Lock();
275	ASSERT_FALSE(object->append_unlock(std::move(append_buffers)));
276	ASSERT_EQ(1U, object->get_pending_appends());
277
278	C_SaferCond cond1;
279	object->flush(&cond1);
280	ASSERT_EQ(0, cond1.wait());
281
282	C_SaferCond cond2;
283	append_buffer1.first->wait(&cond2);
284	ASSERT_EQ(0, cond2.wait());
285	ASSERT_EQ(0U, object->get_pending_appends());
286	}
287
288	TEST_F(TestObjectRecorder, FlushFuture) {
289	std::string oid = get_temp_oid();
290	ASSERT_EQ(0, create(oid));
291	ASSERT_EQ(0, client_register(oid));
292	journal::JournalMetadataPtr metadata = create_metadata(oid);
293	ASSERT_EQ(0, init_metadata(metadata));
294
295	shared_ptr<Mutex> lock(new Mutex("object_recorder_lock"));
296	journal::ObjectRecorderPtr object = create_object(oid, 24, lock);
297
298	journal::AppendBuffer append_buffer = create_append_buffer(234, 123,
299	"payload");
300	journal::AppendBuffers append_buffers;
301	append_buffers = {append_buffer};
302	lock->Lock();
303	ASSERT_FALSE(object->append_unlock(std::move(append_buffers)));
304	ASSERT_EQ(1U, object->get_pending_appends());
305
306	C_SaferCond cond;
307	append_buffer.first->wait(&cond);
308	lock->Lock();
309	object->flush(append_buffer.first);
310	ASSERT_TRUE(lock->is_locked());
311	lock->Unlock();
312	ASSERT_TRUE(append_buffer.first->is_flush_in_progress() \|\|
313	append_buffer.first->is_complete());
314	ASSERT_EQ(0, cond.wait());
315	}
316
317	TEST_F(TestObjectRecorder, FlushDetachedFuture) {
318	std::string oid = get_temp_oid();
319	ASSERT_EQ(0, create(oid));
320	ASSERT_EQ(0, client_register(oid));
321	journal::JournalMetadataPtr metadata = create_metadata(oid);
322	ASSERT_EQ(0, init_metadata(metadata));
323
324	shared_ptr<Mutex> lock(new Mutex("object_recorder_lock"));
325	journal::ObjectRecorderPtr object = create_object(oid, 24, lock);
326
327	journal::AppendBuffer append_buffer = create_append_buffer(234, 123,
328	"payload");
329
330	journal::AppendBuffers append_buffers;
331	append_buffers = {append_buffer};
332
333	lock->Lock();
334	object->flush(append_buffer.first);
335	ASSERT_TRUE(lock->is_locked());
336	lock->Unlock();
337	ASSERT_FALSE(append_buffer.first->is_flush_in_progress());
338	lock->Lock();
339	ASSERT_FALSE(object->append_unlock(std::move(append_buffers)));
340
341	// should automatically flush once its attached to the object
342	C_SaferCond cond;
343	append_buffer.first->wait(&cond);
344	ASSERT_EQ(0, cond.wait());
345	}
346
347	TEST_F(TestObjectRecorder, Close) {
348	std::string oid = get_temp_oid();
349	ASSERT_EQ(0, create(oid));
350	ASSERT_EQ(0, client_register(oid));
351	journal::JournalMetadataPtr metadata = create_metadata(oid);
352	ASSERT_EQ(0, init_metadata(metadata));
353
354	set_flush_interval(2);
355	shared_ptr<Mutex> lock(new Mutex("object_recorder_lock"));
356	journal::ObjectRecorderPtr object = create_object(oid, 24, lock);
357
358	journal::AppendBuffer append_buffer1 = create_append_buffer(234, 123,
359	"payload");
360	journal::AppendBuffers append_buffers;
361	append_buffers = {append_buffer1};
362	lock->Lock();
363	ASSERT_FALSE(object->append_unlock(std::move(append_buffers)));
364	ASSERT_EQ(1U, object->get_pending_appends());
365
366	lock->Lock();
367	ASSERT_FALSE(object->close());
368	ASSERT_TRUE(lock->is_locked());
369	lock->Unlock();
370
371	{
372	Mutex::Locker locker(m_handler.lock);
373	while (!m_handler.is_closed) {
374	if (m_handler.cond.WaitInterval(
375	m_handler.lock, utime_t(10, 0)) != 0) {
376	break;
377	}
378	}
379	}
380
381	ASSERT_TRUE(m_handler.is_closed);
382	ASSERT_EQ(0U, object->get_pending_appends());
383	}
384
385	TEST_F(TestObjectRecorder, Overflow) {
386	std::string oid = get_temp_oid();
387	ASSERT_EQ(0, create(oid));
388	ASSERT_EQ(0, client_register(oid));
389	journal::JournalMetadataPtr metadata = create_metadata(oid);
390	ASSERT_EQ(0, init_metadata(metadata));
391
392	shared_ptr<Mutex> lock1(new Mutex("object_recorder_lock_1"));
393	journal::ObjectRecorderPtr object1 = create_object(oid, 12, lock1);
394	shared_ptr<Mutex> lock2(new Mutex("object_recorder_lock_2"));
395	journal::ObjectRecorderPtr object2 = create_object(oid, 12, lock2);
396
397	std::string payload(2048, '1');
398	journal::AppendBuffer append_buffer1 = create_append_buffer(234, 123,
399	payload);
400	journal::AppendBuffer append_buffer2 = create_append_buffer(234, 124,
401	payload);
402	journal::AppendBuffers append_buffers;
403	append_buffers = {append_buffer1, append_buffer2};
404	lock1->Lock();
405	ASSERT_TRUE(object1->append_unlock(std::move(append_buffers)));
406
407	C_SaferCond cond;
408	append_buffer2.first->wait(&cond);
409	ASSERT_EQ(0, cond.wait());
410	ASSERT_EQ(0U, object1->get_pending_appends());
411
412	journal::AppendBuffer append_buffer3 = create_append_buffer(456, 123,
413	payload);
414	append_buffers = {append_buffer3};
415
416	lock2->Lock();
417	ASSERT_FALSE(object2->append_unlock(std::move(append_buffers)));
418	append_buffer3.first->flush(NULL);
419
420	bool overflowed = false;
421	{
422	Mutex::Locker locker(m_handler.lock);
423	while (m_handler.overflows == 0) {
424	if (m_handler.cond.WaitInterval(
425	m_handler.lock, utime_t(10, 0)) != 0) {
426	break;
427	}
428	}
429	if (m_handler.overflows != 0) {
430	overflowed = true;
431	}
432	}
433
434	ASSERT_TRUE(overflowed);
435	}