ceph/src/test/test_filejournal.cc

   1 // -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
   2 #include <gtest/gtest.h>
   3 #include <stdlib.h>
   4 #include <limits.h>
   5
   6 #include "common/ceph_argparse.h"
   7 #include "common/common_init.h"
   8 #include "global/global_init.h"
   9 #include "common/config.h"
  10 #include "common/Finisher.h"
  11 #include "os/filestore/FileJournal.h"
  12 #include "include/Context.h"
  13 #include "common/Mutex.h"
  14 #include "common/safe_io.h"
  15 #include "os/filestore/JournalingObjectStore.h"
  16
  17 Finisher *finisher;
  18 Cond sync_cond;
  19 char path[200];
  20 uuid_d fsid;
  21 struct test_info {
  22     bool directio, aio, faio;
  23     const char *description;
  24 } subtests[3] = {
  25     { false, false, false, "DIRECTIO OFF  AIO OFF" },
  26     { true, false, false, "DIRECTIO ON  AIO OFF" },
  27     { true, true, true, "DIRECTIO ON  AIO ON"}
  28 };
  29
  30 // ----
  31 Cond cond;
  32 Mutex wait_lock("lock");
  33 bool done;
  34
  35 void wait()
  36 {
  37   wait_lock.Lock();
  38   while (!done)
  39     cond.Wait(wait_lock);
  40   wait_lock.Unlock();
  41 }
  42
  43 // ----
  44 class C_Sync {
  45 public:
  46   Cond cond;
  47   Mutex lock;
  48   bool done;
  49   C_SafeCond *c;
  50
  51   C_Sync()
  52     : lock("C_Sync::lock"), done(false) {
  53     c = new C_SafeCond(&lock, &cond, &done);
  54   }
  55   ~C_Sync() {
  56     lock.Lock();
  57     //cout << "wait" << std::endl;
  58     while (!done)
  59       cond.Wait(lock);
  60     //cout << "waited" << std::endl;
  61     lock.Unlock();
  62   }
  63 };
  64
  65 unsigned size_mb = 200;
  66 //Gtest argument prefix
  67 const char GTEST_PRFIX[] = "--gtest_";
  68
  69 int main(int argc, char **argv) {
  70   vector<const char*> args;
  71   argv_to_vec(argc, (const char **)argv, args);
  72
  73   auto cct = global_init(NULL, args, CEPH_ENTITY_TYPE_CLIENT,
  74                          CODE_ENVIRONMENT_UTILITY,
  75                          CINIT_FLAG_NO_DEFAULT_CONFIG_FILE);
  76   common_init_finish(g_ceph_context);
  77
  78   char mb[10];
  79   sprintf(mb, "%u", size_mb);
  80   g_ceph_context->_conf.set_val("osd_journal_size", mb);
  81   g_ceph_context->_conf.apply_changes(nullptr);
  82
  83   finisher = new Finisher(g_ceph_context);
  84
  85   path[0] = '\0';
  86   if (!args.empty()) {
  87     for ( unsigned int i = 0; i < args.size(); ++i) {
  88       if (strncmp(args[i], GTEST_PRFIX, sizeof(GTEST_PRFIX) - 1)) {
  89         //Non gtest argument, set to path.
  90         size_t copy_len = std::min(sizeof(path) - 1, strlen(args[i]));
  91         strncpy(path, args[i], copy_len);
  92         path[copy_len] = '\0';
  93         break;
  94       }
  95     }
  96   }
  97   if ( path[0] == '\0') {
  98     srand(getpid() + time(0));
  99     snprintf(path, sizeof(path), "/var/tmp/ceph_test_filejournal.tmp.%d", rand());
 100   }
 101   cout << "path " << path << std::endl;
 102
 103   ::testing::InitGoogleTest(&argc, argv);
 104
 105   finisher->start();
 106
 107   int r = RUN_ALL_TESTS();
 108
 109   finisher->stop();
 110
 111   unlink(path);
 112
 113   return r;
 114 }
 115
 116 TEST(TestFileJournal, Create) {
 117   g_ceph_context->_conf.set_val("journal_ignore_corruption", "false");
 118   g_ceph_context->_conf.set_val("journal_write_header_frequency", "0");
 119   g_ceph_context->_conf.apply_changes(nullptr);
 120
 121   for (unsigned i = 0 ; i < 3; ++i) {
 122     SCOPED_TRACE(subtests[i].description);
 123     fsid.generate_random();
 124     FileJournal fj(g_ceph_context, fsid, finisher, &sync_cond, path,
 125                   subtests[i].directio, subtests[i].aio, subtests[i].faio);
 126     ASSERT_EQ(0, fj.create());
 127   }
 128 }
 129
 130 TEST(TestFileJournal, WriteSmall) {
 131   g_ceph_context->_conf.set_val("journal_ignore_corruption", "false");
 132   g_ceph_context->_conf.set_val("journal_write_header_frequency", "0");
 133   g_ceph_context->_conf.apply_changes(nullptr);
 134
 135   for (unsigned i = 0 ; i < 3; ++i) {
 136     SCOPED_TRACE(subtests[i].description);
 137     fsid.generate_random();
 138     FileJournal fj(g_ceph_context, fsid, finisher, &sync_cond, path,
 139                   subtests[i].directio, subtests[i].aio, subtests[i].faio);
 140     ASSERT_EQ(0, fj.create());
 141     ASSERT_EQ(0, fj.make_writeable());
 142
 143     vector<ObjectStore::Transaction> tls;
 144     bufferlist bl;
 145     bl.append("small");
 146     int orig_len = fj.prepare_entry(tls, &bl);
 147     fj.reserve_throttle_and_backoff(bl.length());
 148     fj.submit_entry(1, bl, orig_len, new C_SafeCond(&wait_lock, &cond, &done));
 149     wait();
 150
 151     fj.close();
 152   }
 153 }
 154
 155 TEST(TestFileJournal, WriteBig) {
 156   g_ceph_context->_conf.set_val("journal_ignore_corruption", "false");
 157   g_ceph_context->_conf.set_val("journal_write_header_frequency", "0");
 158   g_ceph_context->_conf.apply_changes(nullptr);
 159
 160   for (unsigned i = 0 ; i < 3; ++i) {
 161     SCOPED_TRACE(subtests[i].description);
 162     fsid.generate_random();
 163     FileJournal fj(g_ceph_context, fsid, finisher, &sync_cond, path,
 164                   subtests[i].directio, subtests[i].aio, subtests[i].faio);
 165     ASSERT_EQ(0, fj.create());
 166     ASSERT_EQ(0, fj.make_writeable());
 167
 168     bufferlist bl;
 169     while (bl.length() < size_mb*1000/2) {
 170       char foo[1024*1024];
 171       memset(foo, 1, sizeof(foo));
 172       bl.append(foo, sizeof(foo));
 173     }
 174     vector<ObjectStore::Transaction> tls;
 175     int orig_len = fj.prepare_entry(tls, &bl);
 176     fj.reserve_throttle_and_backoff(bl.length());
 177     fj.submit_entry(1, bl, orig_len, new C_SafeCond(&wait_lock, &cond, &done));
 178     wait();
 179     fj.close();
 180   }
 181 }
 182
 183 TEST(TestFileJournal, WriteMany) {
 184   g_ceph_context->_conf.set_val("journal_ignore_corruption", "false");
 185   g_ceph_context->_conf.set_val("journal_write_header_frequency", "0");
 186   g_ceph_context->_conf.apply_changes(nullptr);
 187
 188   for (unsigned i = 0 ; i < 3; ++i) {
 189     SCOPED_TRACE(subtests[i].description);
 190     fsid.generate_random();
 191     FileJournal fj(g_ceph_context, fsid, finisher, &sync_cond, path,
 192                   subtests[i].directio, subtests[i].aio, subtests[i].faio);
 193     ASSERT_EQ(0, fj.create());
 194     ASSERT_EQ(0, fj.make_writeable());
 195
 196     C_GatherBuilder gb(g_ceph_context, new C_SafeCond(&wait_lock, &cond, &done));
 197
 198     vector<ObjectStore::Transaction> tls;
 199     bufferlist bl;
 200     bl.append("small");
 201     uint64_t seq = 1;
 202     for (int i=0; i<100; i++) {
 203       bl.append("small");
 204       int orig_len = fj.prepare_entry(tls, &bl);
 205       fj.reserve_throttle_and_backoff(bl.length());
 206       fj.submit_entry(seq++, bl, orig_len, gb.new_sub());
 207     }
 208     gb.activate();
 209
 210     wait();
 211
 212     fj.close();
 213   }
 214 }
 215
 216 TEST(TestFileJournal, WriteManyVecs) {
 217   g_ceph_context->_conf.set_val("journal_ignore_corruption", "false");
 218   g_ceph_context->_conf.set_val("journal_write_header_frequency", "0");
 219   g_ceph_context->_conf.apply_changes(nullptr);
 220
 221   for (unsigned i = 0 ; i < 3; ++i) {
 222     SCOPED_TRACE(subtests[i].description);
 223     fsid.generate_random();
 224     FileJournal fj(g_ceph_context, fsid, finisher, &sync_cond, path,
 225                   subtests[i].directio, subtests[i].aio, subtests[i].faio);
 226     ASSERT_EQ(0, fj.create());
 227     ASSERT_EQ(0, fj.make_writeable());
 228
 229     C_GatherBuilder gb(g_ceph_context, new C_SafeCond(&wait_lock, &cond, &done));
 230
 231     bufferlist first;
 232     first.append("small");
 233     vector<ObjectStore::Transaction> tls;
 234     int orig_len = fj.prepare_entry(tls, &first);
 235     fj.reserve_throttle_and_backoff(first.length());
 236     fj.submit_entry(1, first, orig_len, gb.new_sub());
 237
 238     bufferlist bl;
 239     for (int i=0; i<IOV_MAX * 2; i++) {
 240       bufferptr bp = buffer::create_page_aligned(4096);
 241       memset(bp.c_str(), (char)i, 4096);
 242       bl.append(bp);
 243     }
 244     bufferlist origbl = bl;
 245     orig_len = fj.prepare_entry(tls, &bl);
 246     fj.reserve_throttle_and_backoff(bl.length());
 247     fj.submit_entry(2, bl, orig_len, gb.new_sub());
 248     gb.activate();
 249     wait();
 250
 251     fj.close();
 252
 253     fj.open(1);
 254     bufferlist inbl;
 255     string v;
 256     uint64_t seq = 0;
 257     ASSERT_EQ(true, fj.read_entry(inbl, seq));
 258     ASSERT_EQ(seq, 2ull);
 259     ASSERT_TRUE(inbl.contents_equal(origbl));
 260     ASSERT_EQ(0, fj.make_writeable());
 261     fj.close();
 262
 263   }
 264 }
 265
 266 TEST(TestFileJournal, ReplaySmall) {
 267   g_ceph_context->_conf.set_val("journal_ignore_corruption", "false");
 268   g_ceph_context->_conf.set_val("journal_write_header_frequency", "0");
 269   g_ceph_context->_conf.apply_changes(nullptr);
 270
 271   vector<ObjectStore::Transaction> tls;
 272
 273   for (unsigned i = 0 ; i < 3; ++i) {
 274     SCOPED_TRACE(subtests[i].description);
 275     fsid.generate_random();
 276     FileJournal fj(g_ceph_context, fsid, finisher, &sync_cond, path,
 277                   subtests[i].directio, subtests[i].aio, subtests[i].faio);
 278     ASSERT_EQ(0, fj.create());
 279     ASSERT_EQ(0, fj.make_writeable());
 280
 281     C_GatherBuilder gb(g_ceph_context, new C_SafeCond(&wait_lock, &cond, &done));
 282
 283     bufferlist bl;
 284     bl.append("small");
 285     int orig_len = fj.prepare_entry(tls, &bl);
 286     fj.reserve_throttle_and_backoff(bl.length());
 287     fj.submit_entry(1, bl, orig_len, gb.new_sub());
 288     bl.append("small");
 289     orig_len = fj.prepare_entry(tls, &bl);
 290     fj.reserve_throttle_and_backoff(bl.length());
 291     fj.submit_entry(2, bl, orig_len, gb.new_sub());
 292     bl.append("small");
 293     orig_len = fj.prepare_entry(tls, &bl);
 294     fj.reserve_throttle_and_backoff(bl.length());
 295     fj.submit_entry(3, bl, orig_len, gb.new_sub());
 296     gb.activate();
 297     wait();
 298
 299     fj.close();
 300
 301     fj.open(1);
 302
 303     bufferlist inbl;
 304     string v;
 305     uint64_t seq = 0;
 306     ASSERT_EQ(true, fj.read_entry(inbl, seq));
 307     ASSERT_EQ(seq, 2ull);
 308     inbl.copy(0, inbl.length(), v);
 309     ASSERT_EQ("small", v);
 310     inbl.clear();
 311     v.clear();
 312
 313     ASSERT_EQ(true, fj.read_entry(inbl, seq));
 314     ASSERT_EQ(seq, 3ull);
 315     inbl.copy(0, inbl.length(), v);
 316     ASSERT_EQ("small", v);
 317     inbl.clear();
 318     v.clear();
 319
 320     ASSERT_TRUE(!fj.read_entry(inbl, seq));
 321
 322     ASSERT_EQ(0, fj.make_writeable());
 323     fj.close();
 324   }
 325 }
 326
 327 TEST(TestFileJournal, ReplayCorrupt) {
 328   g_ceph_context->_conf.set_val("journal_ignore_corruption", "true");
 329   g_ceph_context->_conf.set_val("journal_write_header_frequency", "0");
 330   g_ceph_context->_conf.apply_changes(nullptr);
 331
 332   vector<ObjectStore::Transaction> tls;
 333   for (unsigned i = 0 ; i < 3; ++i) {
 334     SCOPED_TRACE(subtests[i].description);
 335     fsid.generate_random();
 336     FileJournal fj(g_ceph_context, fsid, finisher, &sync_cond, path,
 337                   subtests[i].directio, subtests[i].aio, subtests[i].faio);
 338     ASSERT_EQ(0, fj.create());
 339     ASSERT_EQ(0, fj.make_writeable());
 340
 341     C_GatherBuilder gb(g_ceph_context, new C_SafeCond(&wait_lock, &cond, &done));
 342
 343     const char *needle =    "i am a needle";
 344     const char *newneedle = "in a haystack";
 345     bufferlist bl;
 346     bl.append(needle);
 347     int orig_len = fj.prepare_entry(tls, &bl);
 348     fj.reserve_throttle_and_backoff(bl.length());
 349     fj.submit_entry(1, bl, orig_len, gb.new_sub());
 350     bl.append(needle);
 351     orig_len = fj.prepare_entry(tls, &bl);
 352     fj.reserve_throttle_and_backoff(bl.length());
 353     fj.submit_entry(2, bl, orig_len, gb.new_sub());
 354     bl.append(needle);
 355     orig_len = fj.prepare_entry(tls, &bl);
 356     fj.reserve_throttle_and_backoff(bl.length());
 357     fj.submit_entry(3, bl, orig_len, gb.new_sub());
 358     bl.append(needle);
 359     orig_len = fj.prepare_entry(tls, &bl);
 360     fj.reserve_throttle_and_backoff(bl.length());
 361     fj.submit_entry(4, bl, orig_len, gb.new_sub());
 362     gb.activate();
 363     wait();
 364
 365     fj.close();
 366
 367     cout << "corrupting journal" << std::endl;
 368     char buf[1024*128];
 369     int fd = open(path, O_RDONLY);
 370     ASSERT_GE(fd, 0);
 371     int r = safe_read_exact(fd, buf, sizeof(buf));
 372     ASSERT_EQ(0, r);
 373     int n = 0;
 374     for (unsigned o=0; o < sizeof(buf) - strlen(needle); o++) {
 375       if (memcmp(buf+o, needle, strlen(needle)) == 0) {
 376         if (n >= 2) {
 377           cout << "replacing at offset " << o << std::endl;
 378           memcpy(buf+o, newneedle, strlen(newneedle));
 379         } else {
 380           cout << "leaving at offset " << o << std::endl;
 381         }
 382         n++;
 383       }
 384     }
 385     ASSERT_EQ(n, 4);
 386     close(fd);
 387     fd = open(path, O_WRONLY);
 388     ASSERT_GE(fd, 0);
 389     r = safe_write(fd, buf, sizeof(buf));
 390     ASSERT_EQ(r, 0);
 391     close(fd);
 392
 393     fj.open(1);
 394
 395     bufferlist inbl;
 396     string v;
 397     uint64_t seq = 0;
 398     ASSERT_EQ(true, fj.read_entry(inbl, seq));
 399     ASSERT_EQ(seq, 2ull);
 400     inbl.copy(0, inbl.length(), v);
 401     ASSERT_EQ(needle, v);
 402     inbl.clear();
 403     v.clear();
 404     bool corrupt;
 405     ASSERT_FALSE(fj.read_entry(inbl, seq, &corrupt));
 406     ASSERT_TRUE(corrupt);
 407
 408     ASSERT_EQ(0, fj.make_writeable());
 409     fj.close();
 410   }
 411 }
 412
 413 TEST(TestFileJournal, WriteTrim) {
 414   g_ceph_context->_conf.set_val("journal_ignore_corruption", "false");
 415   g_ceph_context->_conf.set_val("journal_write_header_frequency", "0");
 416   g_ceph_context->_conf.apply_changes(nullptr);
 417
 418   for (unsigned i = 0 ; i < 3; ++i) {
 419     SCOPED_TRACE(subtests[i].description);
 420     fsid.generate_random();
 421     FileJournal fj(g_ceph_context, fsid, finisher, &sync_cond, path,
 422                   subtests[i].directio, subtests[i].aio, subtests[i].faio);
 423     ASSERT_EQ(0, fj.create());
 424     ASSERT_EQ(0, fj.make_writeable());
 425
 426     list<C_Sync*> ls;
 427
 428     bufferlist bl;
 429     char foo[1024*1024];
 430     memset(foo, 1, sizeof(foo));
 431
 432     uint64_t seq = 1, committed = 0;
 433     vector<ObjectStore::Transaction> tls;
 434
 435     for (unsigned i=0; i<size_mb*2; i++) {
 436       bl.clear();
 437       bl.push_back(buffer::copy(foo, sizeof(foo)));
 438       bl.zero();
 439       ls.push_back(new C_Sync);
 440       int orig_len = fj.prepare_entry(tls, &bl);
 441       fj.reserve_throttle_and_backoff(bl.length());
 442       fj.submit_entry(seq++, bl, orig_len, ls.back()->c);
 443
 444       while (ls.size() > size_mb/2) {
 445         delete ls.front();
 446         ls.pop_front();
 447         committed++;
 448         fj.committed_thru(committed);
 449       }
 450     }
 451
 452     while (ls.size()) {
 453       delete ls.front();
 454       ls.pop_front();
 455       fj.committed_thru(++committed);
 456     }
 457
 458     ASSERT_TRUE(fj.journalq_empty());
 459
 460     fj.close();
 461   }
 462 }
 463
 464 TEST(TestFileJournal, WriteTrimSmall) {
 465   g_ceph_context->_conf.set_val("journal_ignore_corruption", "false");
 466   g_ceph_context->_conf.set_val("journal_write_header_frequency", "0");
 467   g_ceph_context->_conf.apply_changes(nullptr);
 468   vector<ObjectStore::Transaction> tls;
 469
 470   for (unsigned i = 0 ; i < 3; ++i) {
 471     SCOPED_TRACE(subtests[i].description);
 472     fsid.generate_random();
 473     FileJournal fj(g_ceph_context, fsid, finisher, &sync_cond, path,
 474                   subtests[i].directio, subtests[i].aio, subtests[i].faio);
 475     ASSERT_EQ(0, fj.create());
 476     ASSERT_EQ(0, fj.make_writeable());
 477
 478     list<C_Sync*> ls;
 479
 480     bufferlist bl;
 481     char foo[1024*1024];
 482     memset(foo, 1, sizeof(foo));
 483
 484     uint64_t seq = 1, committed = 0;
 485
 486     for (unsigned i=0; i<size_mb*2; i++) {
 487       bl.clear();
 488       for (int k=0; k<128; k++)
 489         bl.push_back(buffer::copy(foo, sizeof(foo) / 128));
 490       bl.zero();
 491       ls.push_back(new C_Sync);
 492       int orig_len = fj.prepare_entry(tls, &bl);
 493       fj.reserve_throttle_and_backoff(bl.length());
 494       fj.submit_entry(seq++, bl, orig_len, ls.back()->c);
 495
 496       while (ls.size() > size_mb/2) {
 497         delete ls.front();
 498         ls.pop_front();
 499         committed++;
 500         fj.committed_thru(committed);
 501       }
 502     }
 503
 504     while (ls.size()) {
 505       delete ls.front();
 506       ls.pop_front();
 507       fj.committed_thru(committed);
 508     }
 509
 510     fj.close();
 511   }
 512 }
 513
 514 TEST(TestFileJournal, ReplayDetectCorruptFooterMagic) {
 515   g_ceph_context->_conf.set_val("journal_ignore_corruption", "true");
 516   g_ceph_context->_conf.set_val("journal_write_header_frequency", "1");
 517   g_ceph_context->_conf.apply_changes(nullptr);
 518
 519   vector<ObjectStore::Transaction> tls;
 520   for (unsigned i = 0 ; i < 3; ++i) {
 521     SCOPED_TRACE(subtests[i].description);
 522     fsid.generate_random();
 523     FileJournal fj(g_ceph_context, fsid, finisher, &sync_cond, path,
 524                   subtests[i].directio, subtests[i].aio, subtests[i].faio);
 525     ASSERT_EQ(0, fj.create());
 526     ASSERT_EQ(0, fj.make_writeable());
 527
 528     C_GatherBuilder gb(g_ceph_context, new C_SafeCond(&wait_lock, &cond, &done));
 529
 530     const char *needle =    "i am a needle";
 531     for (unsigned i = 1; i <= 4; ++i) {
 532       bufferlist bl;
 533       bl.append(needle);
 534       int orig_len = fj.prepare_entry(tls, &bl);
 535       fj.reserve_throttle_and_backoff(bl.length());
 536       fj.submit_entry(i, bl, orig_len, gb.new_sub());
 537     }
 538     gb.activate();
 539     wait();
 540
 541     bufferlist bl;
 542     bl.append("needle");
 543     int orig_len = fj.prepare_entry(tls, &bl);
 544     fj.reserve_throttle_and_backoff(bl.length());
 545     fj.submit_entry(5, bl, orig_len, new C_SafeCond(&wait_lock, &cond, &done));
 546     wait();
 547
 548     fj.close();
 549     int fd = open(path, O_WRONLY);
 550
 551     cout << "corrupting journal" << std::endl;
 552     fj.open(0);
 553     fj.corrupt_footer_magic(fd, 2);
 554
 555     uint64_t seq = 0;
 556     bl.clear();
 557     bool corrupt = false;
 558     bool result = fj.read_entry(bl, seq, &corrupt);
 559     ASSERT_TRUE(result);
 560     ASSERT_EQ(seq, 1UL);
 561     ASSERT_FALSE(corrupt);
 562
 563     result = fj.read_entry(bl, seq, &corrupt);
 564     ASSERT_FALSE(result);
 565     ASSERT_TRUE(corrupt);
 566
 567     ASSERT_EQ(0, fj.make_writeable());
 568     fj.close();
 569     ::close(fd);
 570   }
 571 }
 572
 573 TEST(TestFileJournal, ReplayDetectCorruptPayload) {
 574   g_ceph_context->_conf.set_val("journal_ignore_corruption", "true");
 575   g_ceph_context->_conf.set_val("journal_write_header_frequency", "1");
 576   g_ceph_context->_conf.apply_changes(nullptr);
 577
 578   vector<ObjectStore::Transaction> tls;
 579   for (unsigned i = 0 ; i < 3; ++i) {
 580     SCOPED_TRACE(subtests[i].description);
 581     fsid.generate_random();
 582     FileJournal fj(g_ceph_context, fsid, finisher, &sync_cond, path,
 583                   subtests[i].directio, subtests[i].aio, subtests[i].faio);
 584     ASSERT_EQ(0, fj.create());
 585     ASSERT_EQ(0, fj.make_writeable());
 586
 587     C_GatherBuilder gb(g_ceph_context, new C_SafeCond(&wait_lock, &cond, &done));
 588
 589     const char *needle =    "i am a needle";
 590     for (unsigned i = 1; i <= 4; ++i) {
 591       bufferlist bl;
 592       bl.append(needle);
 593       int orig_len = fj.prepare_entry(tls, &bl);
 594       fj.reserve_throttle_and_backoff(bl.length());
 595       fj.submit_entry(i, bl, orig_len, gb.new_sub());
 596     }
 597     gb.activate();
 598     wait();
 599
 600     bufferlist bl;
 601     bl.append("needle");
 602     int orig_len = fj.prepare_entry(tls, &bl);
 603     fj.reserve_throttle_and_backoff(bl.length());
 604     fj.submit_entry(5, bl, orig_len, new C_SafeCond(&wait_lock, &cond, &done));
 605     wait();
 606
 607     fj.close();
 608     int fd = open(path, O_WRONLY);
 609
 610     cout << "corrupting journal" << std::endl;
 611     fj.open(0);
 612     fj.corrupt_payload(fd, 2);
 613
 614     uint64_t seq = 0;
 615     bl.clear();
 616     bool corrupt = false;
 617     bool result = fj.read_entry(bl, seq, &corrupt);
 618     ASSERT_TRUE(result);
 619     ASSERT_EQ(seq, 1UL);
 620     ASSERT_FALSE(corrupt);
 621
 622     result = fj.read_entry(bl, seq, &corrupt);
 623     ASSERT_FALSE(result);
 624     ASSERT_TRUE(corrupt);
 625
 626     ASSERT_EQ(0, fj.make_writeable());
 627     fj.close();
 628     ::close(fd);
 629   }
 630 }
 631
 632 TEST(TestFileJournal, ReplayDetectCorruptHeader) {
 633   g_ceph_context->_conf.set_val("journal_ignore_corruption", "true");
 634   g_ceph_context->_conf.set_val("journal_write_header_frequency", "1");
 635   g_ceph_context->_conf.apply_changes(nullptr);
 636
 637   vector<ObjectStore::Transaction> tls;
 638   for (unsigned i = 0 ; i < 3; ++i) {
 639     SCOPED_TRACE(subtests[i].description);
 640     fsid.generate_random();
 641     FileJournal fj(g_ceph_context, fsid, finisher, &sync_cond, path,
 642                   subtests[i].directio, subtests[i].aio, subtests[i].faio);
 643     ASSERT_EQ(0, fj.create());
 644     ASSERT_EQ(0, fj.make_writeable());
 645
 646     C_GatherBuilder gb(g_ceph_context, new C_SafeCond(&wait_lock, &cond, &done));
 647
 648     const char *needle =    "i am a needle";
 649     for (unsigned i = 1; i <= 4; ++i) {
 650       bufferlist bl;
 651       bl.append(needle);
 652       int orig_len = fj.prepare_entry(tls, &bl);
 653       fj.reserve_throttle_and_backoff(bl.length());
 654       fj.submit_entry(i, bl, orig_len, gb.new_sub());
 655     }
 656     gb.activate();
 657     wait();
 658
 659     bufferlist bl;
 660     bl.append("needle");
 661     int orig_len = fj.prepare_entry(tls, &bl);
 662     fj.reserve_throttle_and_backoff(bl.length());
 663     fj.submit_entry(5, bl, orig_len, new C_SafeCond(&wait_lock, &cond, &done));
 664     wait();
 665
 666     fj.close();
 667     int fd = open(path, O_WRONLY);
 668
 669     cout << "corrupting journal" << std::endl;
 670     fj.open(0);
 671     fj.corrupt_header_magic(fd, 2);
 672
 673     uint64_t seq = 0;
 674     bl.clear();
 675     bool corrupt = false;
 676     bool result = fj.read_entry(bl, seq, &corrupt);
 677     ASSERT_TRUE(result);
 678     ASSERT_EQ(seq, 1UL);
 679     ASSERT_FALSE(corrupt);
 680
 681     result = fj.read_entry(bl, seq, &corrupt);
 682     ASSERT_FALSE(result);
 683     ASSERT_TRUE(corrupt);
 684
 685     ASSERT_EQ(0, fj.make_writeable());
 686     fj.close();
 687     ::close(fd);
 688   }
 689 }