--- /dev/null
+// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
+#include <gtest/gtest.h>
+#include <stdlib.h>
+#include <limits.h>
+
+#include "common/ceph_argparse.h"
+#include "common/common_init.h"
+#include "global/global_init.h"
+#include "common/config.h"
+#include "common/Finisher.h"
+#include "os/filestore/FileJournal.h"
+#include "include/Context.h"
+#include "common/Mutex.h"
+#include "common/safe_io.h"
+#include "os/filestore/JournalingObjectStore.h"
+
+Finisher *finisher;
+Cond sync_cond;
+char path[200];
+uuid_d fsid;
+struct test_info {
+ bool directio, aio, faio;
+ const char *description;
+} subtests[3] = {
+ { false, false, false, "DIRECTIO OFF AIO OFF" },
+ { true, false, false, "DIRECTIO ON AIO OFF" },
+ { true, true, true, "DIRECTIO ON AIO ON"}
+};
+
+// ----
+Cond cond;
+Mutex wait_lock("lock");
+bool done;
+
+void wait()
+{
+ wait_lock.Lock();
+ while (!done)
+ cond.Wait(wait_lock);
+ wait_lock.Unlock();
+}
+
+// ----
+class C_Sync {
+public:
+ Cond cond;
+ Mutex lock;
+ bool done;
+ C_SafeCond *c;
+
+ C_Sync()
+ : lock("C_Sync::lock"), done(false) {
+ c = new C_SafeCond(&lock, &cond, &done);
+ }
+ ~C_Sync() {
+ lock.Lock();
+ //cout << "wait" << std::endl;
+ while (!done)
+ cond.Wait(lock);
+ //cout << "waited" << std::endl;
+ lock.Unlock();
+ }
+};
+
+unsigned size_mb = 200;
+//Gtest argument prefix
+const char GTEST_PRFIX[] = "--gtest_";
+
+int main(int argc, char **argv) {
+ vector<const char*> args;
+ argv_to_vec(argc, (const char **)argv, args);
+
+ auto cct = global_init(NULL, args, CEPH_ENTITY_TYPE_CLIENT,
+ CODE_ENVIRONMENT_UTILITY, 0);
+ common_init_finish(g_ceph_context);
+
+ char mb[10];
+ sprintf(mb, "%u", size_mb);
+ g_ceph_context->_conf->set_val("osd_journal_size", mb);
+ g_ceph_context->_conf->apply_changes(NULL);
+
+ finisher = new Finisher(g_ceph_context);
+
+ path[0] = '\0';
+ if (!args.empty()) {
+ for ( unsigned int i = 0; i < args.size(); ++i) {
+ if (strncmp(args[i], GTEST_PRFIX, sizeof(GTEST_PRFIX) - 1)) {
+ //Non gtest argument, set to path.
+ size_t copy_len = std::min(sizeof(path) - 1, strlen(args[i]));
+ strncpy(path, args[i], copy_len);
+ path[copy_len] = '\0';
+ break;
+ }
+ }
+ }
+ if ( path[0] == '\0') {
+ srand(getpid() + time(0));
+ snprintf(path, sizeof(path), "/var/tmp/ceph_test_filejournal.tmp.%d", rand());
+ }
+ cout << "path " << path << std::endl;
+
+ ::testing::InitGoogleTest(&argc, argv);
+
+ finisher->start();
+
+ int r = RUN_ALL_TESTS();
+
+ finisher->stop();
+
+ unlink(path);
+
+ return r;
+}
+
+TEST(TestFileJournal, Create) {
+ g_ceph_context->_conf->set_val("journal_ignore_corruption", "false");
+ g_ceph_context->_conf->set_val("journal_write_header_frequency", "0");
+ g_ceph_context->_conf->apply_changes(NULL);
+
+ for (unsigned i = 0 ; i < 3; ++i) {
+ SCOPED_TRACE(subtests[i].description);
+ fsid.generate_random();
+ FileJournal fj(g_ceph_context, fsid, finisher, &sync_cond, path,
+ subtests[i].directio, subtests[i].aio, subtests[i].faio);
+ ASSERT_EQ(0, fj.create());
+ }
+}
+
+TEST(TestFileJournal, WriteSmall) {
+ g_ceph_context->_conf->set_val("journal_ignore_corruption", "false");
+ g_ceph_context->_conf->set_val("journal_write_header_frequency", "0");
+ g_ceph_context->_conf->apply_changes(NULL);
+
+ for (unsigned i = 0 ; i < 3; ++i) {
+ SCOPED_TRACE(subtests[i].description);
+ fsid.generate_random();
+ FileJournal fj(g_ceph_context, fsid, finisher, &sync_cond, path,
+ subtests[i].directio, subtests[i].aio, subtests[i].faio);
+ ASSERT_EQ(0, fj.create());
+ ASSERT_EQ(0, fj.make_writeable());
+
+ vector<ObjectStore::Transaction> tls;
+ bufferlist bl;
+ bl.append("small");
+ int orig_len = fj.prepare_entry(tls, &bl);
+ fj.reserve_throttle_and_backoff(bl.length());
+ fj.submit_entry(1, bl, orig_len, new C_SafeCond(&wait_lock, &cond, &done));
+ wait();
+
+ fj.close();
+ }
+}
+
+TEST(TestFileJournal, WriteBig) {
+ g_ceph_context->_conf->set_val("journal_ignore_corruption", "false");
+ g_ceph_context->_conf->set_val("journal_write_header_frequency", "0");
+ g_ceph_context->_conf->apply_changes(NULL);
+
+ for (unsigned i = 0 ; i < 3; ++i) {
+ SCOPED_TRACE(subtests[i].description);
+ fsid.generate_random();
+ FileJournal fj(g_ceph_context, fsid, finisher, &sync_cond, path,
+ subtests[i].directio, subtests[i].aio, subtests[i].faio);
+ ASSERT_EQ(0, fj.create());
+ ASSERT_EQ(0, fj.make_writeable());
+
+ bufferlist bl;
+ while (bl.length() < size_mb*1000/2) {
+ char foo[1024*1024];
+ memset(foo, 1, sizeof(foo));
+ bl.append(foo, sizeof(foo));
+ }
+ vector<ObjectStore::Transaction> tls;
+ int orig_len = fj.prepare_entry(tls, &bl);
+ fj.reserve_throttle_and_backoff(bl.length());
+ fj.submit_entry(1, bl, orig_len, new C_SafeCond(&wait_lock, &cond, &done));
+ wait();
+ fj.close();
+ }
+}
+
+TEST(TestFileJournal, WriteMany) {
+ g_ceph_context->_conf->set_val("journal_ignore_corruption", "false");
+ g_ceph_context->_conf->set_val("journal_write_header_frequency", "0");
+ g_ceph_context->_conf->apply_changes(NULL);
+
+ for (unsigned i = 0 ; i < 3; ++i) {
+ SCOPED_TRACE(subtests[i].description);
+ fsid.generate_random();
+ FileJournal fj(g_ceph_context, fsid, finisher, &sync_cond, path,
+ subtests[i].directio, subtests[i].aio, subtests[i].faio);
+ ASSERT_EQ(0, fj.create());
+ ASSERT_EQ(0, fj.make_writeable());
+
+ C_GatherBuilder gb(g_ceph_context, new C_SafeCond(&wait_lock, &cond, &done));
+
+ vector<ObjectStore::Transaction> tls;
+ bufferlist bl;
+ bl.append("small");
+ uint64_t seq = 1;
+ for (int i=0; i<100; i++) {
+ bl.append("small");
+ int orig_len = fj.prepare_entry(tls, &bl);
+ fj.reserve_throttle_and_backoff(bl.length());
+ fj.submit_entry(seq++, bl, orig_len, gb.new_sub());
+ }
+ gb.activate();
+
+ wait();
+
+ fj.close();
+ }
+}
+
+TEST(TestFileJournal, WriteManyVecs) {
+ g_ceph_context->_conf->set_val("journal_ignore_corruption", "false");
+ g_ceph_context->_conf->set_val("journal_write_header_frequency", "0");
+ g_ceph_context->_conf->apply_changes(NULL);
+
+ for (unsigned i = 0 ; i < 3; ++i) {
+ SCOPED_TRACE(subtests[i].description);
+ fsid.generate_random();
+ FileJournal fj(g_ceph_context, fsid, finisher, &sync_cond, path,
+ subtests[i].directio, subtests[i].aio, subtests[i].faio);
+ ASSERT_EQ(0, fj.create());
+ ASSERT_EQ(0, fj.make_writeable());
+
+ C_GatherBuilder gb(g_ceph_context, new C_SafeCond(&wait_lock, &cond, &done));
+
+ bufferlist first;
+ first.append("small");
+ vector<ObjectStore::Transaction> tls;
+ int orig_len = fj.prepare_entry(tls, &first);
+ fj.reserve_throttle_and_backoff(first.length());
+ fj.submit_entry(1, first, orig_len, gb.new_sub());
+
+ bufferlist bl;
+ for (int i=0; i<IOV_MAX * 2; i++) {
+ bufferptr bp = buffer::create_page_aligned(4096);
+ memset(bp.c_str(), (char)i, 4096);
+ bl.append(bp);
+ }
+ bufferlist origbl = bl;
+ orig_len = fj.prepare_entry(tls, &bl);
+ fj.reserve_throttle_and_backoff(bl.length());
+ fj.submit_entry(2, bl, orig_len, gb.new_sub());
+ gb.activate();
+ wait();
+
+ fj.close();
+
+ fj.open(1);
+ bufferlist inbl;
+ string v;
+ uint64_t seq = 0;
+ ASSERT_EQ(true, fj.read_entry(inbl, seq));
+ ASSERT_EQ(seq, 2ull);
+ ASSERT_TRUE(inbl.contents_equal(origbl));
+ ASSERT_EQ(0, fj.make_writeable());
+ fj.close();
+
+ }
+}
+
+TEST(TestFileJournal, ReplaySmall) {
+ g_ceph_context->_conf->set_val("journal_ignore_corruption", "false");
+ g_ceph_context->_conf->set_val("journal_write_header_frequency", "0");
+ g_ceph_context->_conf->apply_changes(NULL);
+
+ vector<ObjectStore::Transaction> tls;
+
+ for (unsigned i = 0 ; i < 3; ++i) {
+ SCOPED_TRACE(subtests[i].description);
+ fsid.generate_random();
+ FileJournal fj(g_ceph_context, fsid, finisher, &sync_cond, path,
+ subtests[i].directio, subtests[i].aio, subtests[i].faio);
+ ASSERT_EQ(0, fj.create());
+ ASSERT_EQ(0, fj.make_writeable());
+
+ C_GatherBuilder gb(g_ceph_context, new C_SafeCond(&wait_lock, &cond, &done));
+
+ bufferlist bl;
+ bl.append("small");
+ int orig_len = fj.prepare_entry(tls, &bl);
+ fj.reserve_throttle_and_backoff(bl.length());
+ fj.submit_entry(1, bl, orig_len, gb.new_sub());
+ bl.append("small");
+ orig_len = fj.prepare_entry(tls, &bl);
+ fj.reserve_throttle_and_backoff(bl.length());
+ fj.submit_entry(2, bl, orig_len, gb.new_sub());
+ bl.append("small");
+ orig_len = fj.prepare_entry(tls, &bl);
+ fj.reserve_throttle_and_backoff(bl.length());
+ fj.submit_entry(3, bl, orig_len, gb.new_sub());
+ gb.activate();
+ wait();
+
+ fj.close();
+
+ fj.open(1);
+
+ bufferlist inbl;
+ string v;
+ uint64_t seq = 0;
+ ASSERT_EQ(true, fj.read_entry(inbl, seq));
+ ASSERT_EQ(seq, 2ull);
+ inbl.copy(0, inbl.length(), v);
+ ASSERT_EQ("small", v);
+ inbl.clear();
+ v.clear();
+
+ ASSERT_EQ(true, fj.read_entry(inbl, seq));
+ ASSERT_EQ(seq, 3ull);
+ inbl.copy(0, inbl.length(), v);
+ ASSERT_EQ("small", v);
+ inbl.clear();
+ v.clear();
+
+ ASSERT_TRUE(!fj.read_entry(inbl, seq));
+
+ ASSERT_EQ(0, fj.make_writeable());
+ fj.close();
+ }
+}
+
+TEST(TestFileJournal, ReplayCorrupt) {
+ g_ceph_context->_conf->set_val("journal_ignore_corruption", "true");
+ g_ceph_context->_conf->set_val("journal_write_header_frequency", "0");
+ g_ceph_context->_conf->apply_changes(NULL);
+
+ vector<ObjectStore::Transaction> tls;
+ for (unsigned i = 0 ; i < 3; ++i) {
+ SCOPED_TRACE(subtests[i].description);
+ fsid.generate_random();
+ FileJournal fj(g_ceph_context, fsid, finisher, &sync_cond, path,
+ subtests[i].directio, subtests[i].aio, subtests[i].faio);
+ ASSERT_EQ(0, fj.create());
+ ASSERT_EQ(0, fj.make_writeable());
+
+ C_GatherBuilder gb(g_ceph_context, new C_SafeCond(&wait_lock, &cond, &done));
+
+ const char *needle = "i am a needle";
+ const char *newneedle = "in a haystack";
+ bufferlist bl;
+ bl.append(needle);
+ int orig_len = fj.prepare_entry(tls, &bl);
+ fj.reserve_throttle_and_backoff(bl.length());
+ fj.submit_entry(1, bl, orig_len, gb.new_sub());
+ bl.append(needle);
+ orig_len = fj.prepare_entry(tls, &bl);
+ fj.reserve_throttle_and_backoff(bl.length());
+ fj.submit_entry(2, bl, orig_len, gb.new_sub());
+ bl.append(needle);
+ orig_len = fj.prepare_entry(tls, &bl);
+ fj.reserve_throttle_and_backoff(bl.length());
+ fj.submit_entry(3, bl, orig_len, gb.new_sub());
+ bl.append(needle);
+ orig_len = fj.prepare_entry(tls, &bl);
+ fj.reserve_throttle_and_backoff(bl.length());
+ fj.submit_entry(4, bl, orig_len, gb.new_sub());
+ gb.activate();
+ wait();
+
+ fj.close();
+
+ cout << "corrupting journal" << std::endl;
+ char buf[1024*128];
+ int fd = open(path, O_RDONLY);
+ ASSERT_GE(fd, 0);
+ int r = safe_read_exact(fd, buf, sizeof(buf));
+ ASSERT_EQ(0, r);
+ int n = 0;
+ for (unsigned o=0; o < sizeof(buf) - strlen(needle); o++) {
+ if (memcmp(buf+o, needle, strlen(needle)) == 0) {
+ if (n >= 2) {
+ cout << "replacing at offset " << o << std::endl;
+ memcpy(buf+o, newneedle, strlen(newneedle));
+ } else {
+ cout << "leaving at offset " << o << std::endl;
+ }
+ n++;
+ }
+ }
+ ASSERT_EQ(n, 4);
+ close(fd);
+ fd = open(path, O_WRONLY);
+ ASSERT_GE(fd, 0);
+ r = safe_write(fd, buf, sizeof(buf));
+ ASSERT_EQ(r, 0);
+ close(fd);
+
+ fj.open(1);
+
+ bufferlist inbl;
+ string v;
+ uint64_t seq = 0;
+ ASSERT_EQ(true, fj.read_entry(inbl, seq));
+ ASSERT_EQ(seq, 2ull);
+ inbl.copy(0, inbl.length(), v);
+ ASSERT_EQ(needle, v);
+ inbl.clear();
+ v.clear();
+ bool corrupt;
+ ASSERT_FALSE(fj.read_entry(inbl, seq, &corrupt));
+ ASSERT_TRUE(corrupt);
+
+ ASSERT_EQ(0, fj.make_writeable());
+ fj.close();
+ }
+}
+
+TEST(TestFileJournal, WriteTrim) {
+ g_ceph_context->_conf->set_val("journal_ignore_corruption", "false");
+ g_ceph_context->_conf->set_val("journal_write_header_frequency", "0");
+ g_ceph_context->_conf->apply_changes(NULL);
+
+ for (unsigned i = 0 ; i < 3; ++i) {
+ SCOPED_TRACE(subtests[i].description);
+ fsid.generate_random();
+ FileJournal fj(g_ceph_context, fsid, finisher, &sync_cond, path,
+ subtests[i].directio, subtests[i].aio, subtests[i].faio);
+ ASSERT_EQ(0, fj.create());
+ ASSERT_EQ(0, fj.make_writeable());
+
+ list<C_Sync*> ls;
+
+ bufferlist bl;
+ char foo[1024*1024];
+ memset(foo, 1, sizeof(foo));
+
+ uint64_t seq = 1, committed = 0;
+ vector<ObjectStore::Transaction> tls;
+
+ for (unsigned i=0; i<size_mb*2; i++) {
+ bl.clear();
+ bl.push_back(buffer::copy(foo, sizeof(foo)));
+ bl.zero();
+ ls.push_back(new C_Sync);
+ int orig_len = fj.prepare_entry(tls, &bl);
+ fj.reserve_throttle_and_backoff(bl.length());
+ fj.submit_entry(seq++, bl, orig_len, ls.back()->c);
+
+ while (ls.size() > size_mb/2) {
+ delete ls.front();
+ ls.pop_front();
+ committed++;
+ fj.committed_thru(committed);
+ }
+ }
+
+ while (ls.size()) {
+ delete ls.front();
+ ls.pop_front();
+ fj.committed_thru(++committed);
+ }
+
+ ASSERT_TRUE(fj.journalq_empty());
+
+ fj.close();
+ }
+}
+
+TEST(TestFileJournal, WriteTrimSmall) {
+ g_ceph_context->_conf->set_val("journal_ignore_corruption", "false");
+ g_ceph_context->_conf->set_val("journal_write_header_frequency", "0");
+ g_ceph_context->_conf->apply_changes(NULL);
+ vector<ObjectStore::Transaction> tls;
+
+ for (unsigned i = 0 ; i < 3; ++i) {
+ SCOPED_TRACE(subtests[i].description);
+ fsid.generate_random();
+ FileJournal fj(g_ceph_context, fsid, finisher, &sync_cond, path,
+ subtests[i].directio, subtests[i].aio, subtests[i].faio);
+ ASSERT_EQ(0, fj.create());
+ ASSERT_EQ(0, fj.make_writeable());
+
+ list<C_Sync*> ls;
+
+ bufferlist bl;
+ char foo[1024*1024];
+ memset(foo, 1, sizeof(foo));
+
+ uint64_t seq = 1, committed = 0;
+
+ for (unsigned i=0; i<size_mb*2; i++) {
+ bl.clear();
+ for (int k=0; k<128; k++)
+ bl.push_back(buffer::copy(foo, sizeof(foo) / 128));
+ bl.zero();
+ ls.push_back(new C_Sync);
+ int orig_len = fj.prepare_entry(tls, &bl);
+ fj.reserve_throttle_and_backoff(bl.length());
+ fj.submit_entry(seq++, bl, orig_len, ls.back()->c);
+
+ while (ls.size() > size_mb/2) {
+ delete ls.front();
+ ls.pop_front();
+ committed++;
+ fj.committed_thru(committed);
+ }
+ }
+
+ while (ls.size()) {
+ delete ls.front();
+ ls.pop_front();
+ fj.committed_thru(committed);
+ }
+
+ fj.close();
+ }
+}
+
+TEST(TestFileJournal, ReplayDetectCorruptFooterMagic) {
+ g_ceph_context->_conf->set_val("journal_ignore_corruption", "true");
+ g_ceph_context->_conf->set_val("journal_write_header_frequency", "1");
+ g_ceph_context->_conf->apply_changes(NULL);
+
+ vector<ObjectStore::Transaction> tls;
+ for (unsigned i = 0 ; i < 3; ++i) {
+ SCOPED_TRACE(subtests[i].description);
+ fsid.generate_random();
+ FileJournal fj(g_ceph_context, fsid, finisher, &sync_cond, path,
+ subtests[i].directio, subtests[i].aio, subtests[i].faio);
+ ASSERT_EQ(0, fj.create());
+ ASSERT_EQ(0, fj.make_writeable());
+
+ C_GatherBuilder gb(g_ceph_context, new C_SafeCond(&wait_lock, &cond, &done));
+
+ const char *needle = "i am a needle";
+ for (unsigned i = 1; i <= 4; ++i) {
+ bufferlist bl;
+ bl.append(needle);
+ int orig_len = fj.prepare_entry(tls, &bl);
+ fj.reserve_throttle_and_backoff(bl.length());
+ fj.submit_entry(i, bl, orig_len, gb.new_sub());
+ }
+ gb.activate();
+ wait();
+
+ bufferlist bl;
+ bl.append("needle");
+ int orig_len = fj.prepare_entry(tls, &bl);
+ fj.reserve_throttle_and_backoff(bl.length());
+ fj.submit_entry(5, bl, orig_len, new C_SafeCond(&wait_lock, &cond, &done));
+ wait();
+
+ fj.close();
+ int fd = open(path, O_WRONLY);
+
+ cout << "corrupting journal" << std::endl;
+ fj.open(0);
+ fj.corrupt_footer_magic(fd, 2);
+
+ uint64_t seq = 0;
+ bl.clear();
+ bool corrupt = false;
+ bool result = fj.read_entry(bl, seq, &corrupt);
+ ASSERT_TRUE(result);
+ ASSERT_EQ(seq, 1UL);
+ ASSERT_FALSE(corrupt);
+
+ result = fj.read_entry(bl, seq, &corrupt);
+ ASSERT_FALSE(result);
+ ASSERT_TRUE(corrupt);
+
+ ASSERT_EQ(0, fj.make_writeable());
+ fj.close();
+ ::close(fd);
+ }
+}
+
+TEST(TestFileJournal, ReplayDetectCorruptPayload) {
+ g_ceph_context->_conf->set_val("journal_ignore_corruption", "true");
+ g_ceph_context->_conf->set_val("journal_write_header_frequency", "1");
+ g_ceph_context->_conf->apply_changes(NULL);
+
+ vector<ObjectStore::Transaction> tls;
+ for (unsigned i = 0 ; i < 3; ++i) {
+ SCOPED_TRACE(subtests[i].description);
+ fsid.generate_random();
+ FileJournal fj(g_ceph_context, fsid, finisher, &sync_cond, path,
+ subtests[i].directio, subtests[i].aio, subtests[i].faio);
+ ASSERT_EQ(0, fj.create());
+ ASSERT_EQ(0, fj.make_writeable());
+
+ C_GatherBuilder gb(g_ceph_context, new C_SafeCond(&wait_lock, &cond, &done));
+
+ const char *needle = "i am a needle";
+ for (unsigned i = 1; i <= 4; ++i) {
+ bufferlist bl;
+ bl.append(needle);
+ int orig_len = fj.prepare_entry(tls, &bl);
+ fj.reserve_throttle_and_backoff(bl.length());
+ fj.submit_entry(i, bl, orig_len, gb.new_sub());
+ }
+ gb.activate();
+ wait();
+
+ bufferlist bl;
+ bl.append("needle");
+ int orig_len = fj.prepare_entry(tls, &bl);
+ fj.reserve_throttle_and_backoff(bl.length());
+ fj.submit_entry(5, bl, orig_len, new C_SafeCond(&wait_lock, &cond, &done));
+ wait();
+
+ fj.close();
+ int fd = open(path, O_WRONLY);
+
+ cout << "corrupting journal" << std::endl;
+ fj.open(0);
+ fj.corrupt_payload(fd, 2);
+
+ uint64_t seq = 0;
+ bl.clear();
+ bool corrupt = false;
+ bool result = fj.read_entry(bl, seq, &corrupt);
+ ASSERT_TRUE(result);
+ ASSERT_EQ(seq, 1UL);
+ ASSERT_FALSE(corrupt);
+
+ result = fj.read_entry(bl, seq, &corrupt);
+ ASSERT_FALSE(result);
+ ASSERT_TRUE(corrupt);
+
+ ASSERT_EQ(0, fj.make_writeable());
+ fj.close();
+ ::close(fd);
+ }
+}
+
+TEST(TestFileJournal, ReplayDetectCorruptHeader) {
+ g_ceph_context->_conf->set_val("journal_ignore_corruption", "true");
+ g_ceph_context->_conf->set_val("journal_write_header_frequency", "1");
+ g_ceph_context->_conf->apply_changes(NULL);
+
+ vector<ObjectStore::Transaction> tls;
+ for (unsigned i = 0 ; i < 3; ++i) {
+ SCOPED_TRACE(subtests[i].description);
+ fsid.generate_random();
+ FileJournal fj(g_ceph_context, fsid, finisher, &sync_cond, path,
+ subtests[i].directio, subtests[i].aio, subtests[i].faio);
+ ASSERT_EQ(0, fj.create());
+ ASSERT_EQ(0, fj.make_writeable());
+
+ C_GatherBuilder gb(g_ceph_context, new C_SafeCond(&wait_lock, &cond, &done));
+
+ const char *needle = "i am a needle";
+ for (unsigned i = 1; i <= 4; ++i) {
+ bufferlist bl;
+ bl.append(needle);
+ int orig_len = fj.prepare_entry(tls, &bl);
+ fj.reserve_throttle_and_backoff(bl.length());
+ fj.submit_entry(i, bl, orig_len, gb.new_sub());
+ }
+ gb.activate();
+ wait();
+
+ bufferlist bl;
+ bl.append("needle");
+ int orig_len = fj.prepare_entry(tls, &bl);
+ fj.reserve_throttle_and_backoff(bl.length());
+ fj.submit_entry(5, bl, orig_len, new C_SafeCond(&wait_lock, &cond, &done));
+ wait();
+
+ fj.close();
+ int fd = open(path, O_WRONLY);
+
+ cout << "corrupting journal" << std::endl;
+ fj.open(0);
+ fj.corrupt_header_magic(fd, 2);
+
+ uint64_t seq = 0;
+ bl.clear();
+ bool corrupt = false;
+ bool result = fj.read_entry(bl, seq, &corrupt);
+ ASSERT_TRUE(result);
+ ASSERT_EQ(seq, 1UL);
+ ASSERT_FALSE(corrupt);
+
+ result = fj.read_entry(bl, seq, &corrupt);
+ ASSERT_FALSE(result);
+ ASSERT_TRUE(corrupt);
+
+ ASSERT_EQ(0, fj.make_writeable());
+ fj.close();
+ ::close(fd);
+ }
+}
Code Review / stor4nfv.git / blobdiff