optimistic_transaction_test.cc 40 KB

12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455565758596061626364656667686970717273747576777879808182838485868788899091929394959697989910010110210310410510610710810911011111211311411511611711811912012112212312412512612712812913013113213313413513613713813914014114214314414514614714814915015115215315415515615715815916016116216316416516616716816917017117217317417517617717817918018118218318418518618718818919019119219319419519619719819920020120220320420520620720820921021121221321421521621721821922022122222322422522622722822923023123223323423523623723823924024124224324424524624724824925025125225325425525625725825926026126226326426526626726826927027127227327427527627727827928028128228328428528628728828929029129229329429529629729829930030130230330430530630730830931031131231331431531631731831932032132232332432532632732832933033133233333433533633733833934034134234334434534634734834935035135235335435535635735835936036136236336436536636736836937037137237337437537637737837938038138238338438538638738838939039139239339439539639739839940040140240340440540640740840941041141241341441541641741841942042142242342442542642742842943043143243343443543643743843944044144244344444544644744844945045145245345445545645745845946046146246346446546646746846947047147247347447547647747847948048148248348448548648748848949049149249349449549649749849950050150250350450550650750850951051151251351451551651751851952052152252352452552652752852953053153253353453553653753853954054154254354454554654754854955055155255355455555655755855956056156256356456556656756856957057157257357457557657757857958058158258358458558658758858959059159259359459559659759859960060160260360460560660760860961061161261361461561661761861962062162262362462562662762862963063163263363463563663763863964064164264364464564664764864965065165265365465565665765865966066166266366466566666766866967067167267367467567667767867968068168268368468568668768868969069169269369469569669769869970070170270370470570670770870971071171271371471571671771871972072172272372472572672772872973073173273373473573673773873974074174274374474574674774874975075175275375475575675775875976076176276376476576676776876977077177277377477577677777877978078178278378478578678778878979079179279379479579679779879980080180280380480580680780880981081181281381481581681781881982082182282382482582682782882983083183283383483583683783883984084184284384484584684784884985085185285385485585685785885986086186286386486586686786886987087187287387487587687787887988088188288388488588688788888989089189289389489589689789889990090190290390490590690790890991091191291391491591691791891992092192292392492592692792892993093193293393493593693793893994094194294394494594694794894995095195295395495595695795895996096196296396496596696796896997097197297397497597697797897998098198298398498598698798898999099199299399499599699799899910001001100210031004100510061007100810091010101110121013101410151016101710181019102010211022102310241025102610271028102910301031103210331034103510361037103810391040104110421043104410451046104710481049105010511052105310541055105610571058105910601061106210631064106510661067106810691070107110721073107410751076107710781079108010811082108310841085108610871088108910901091109210931094109510961097109810991100110111021103110411051106110711081109111011111112111311141115111611171118111911201121112211231124112511261127112811291130113111321133113411351136113711381139114011411142114311441145114611471148114911501151115211531154115511561157115811591160116111621163116411651166116711681169117011711172117311741175117611771178117911801181118211831184118511861187118811891190119111921193119411951196119711981199120012011202120312041205120612071208120912101211121212131214121512161217121812191220122112221223122412251226122712281229123012311232123312341235123612371238123912401241124212431244124512461247124812491250125112521253125412551256125712581259126012611262126312641265126612671268126912701271127212731274127512761277127812791280128112821283128412851286128712881289129012911292129312941295129612971298129913001301130213031304130513061307130813091310131113121313131413151316131713181319132013211322132313241325132613271328132913301331133213331334133513361337133813391340134113421343134413451346134713481349135013511352135313541355135613571358135913601361136213631364136513661367136813691370137113721373137413751376137713781379138013811382138313841385138613871388138913901391139213931394139513961397139813991400140114021403140414051406140714081409141014111412141314141415141614171418141914201421142214231424142514261427142814291430143114321433143414351436143714381439144014411442144314441445144614471448144914501451145214531454145514561457145814591460146114621463146414651466146714681469147014711472147314741475147614771478147914801481148214831484148514861487148814891490149114921493149414951496149714981499150015011502150315041505150615071508150915101511151215131514151515161517151815191520152115221523152415251526152715281529153015311532153315341535
  1. // Copyright (c) 2011-present, Facebook, Inc. All rights reserved.
  2. // This source code is licensed under both the GPLv2 (found in the
  3. // COPYING file in the root directory) and Apache 2.0 License
  4. // (found in the LICENSE.Apache file in the root directory).
  5. #ifndef ROCKSDB_LITE
  6. #include <functional>
  7. #include <string>
  8. #include <thread>
  9. #include "db/db_impl/db_impl.h"
  10. #include "logging/logging.h"
  11. #include "port/port.h"
  12. #include "rocksdb/db.h"
  13. #include "rocksdb/perf_context.h"
  14. #include "rocksdb/utilities/optimistic_transaction_db.h"
  15. #include "rocksdb/utilities/transaction.h"
  16. #include "test_util/sync_point.h"
  17. #include "test_util/testharness.h"
  18. #include "test_util/transaction_test_util.h"
  19. #include "util/crc32c.h"
  20. #include "util/random.h"
  21. using std::string;
  22. namespace ROCKSDB_NAMESPACE {
  23. class OptimisticTransactionTest
  24. : public testing::Test,
  25. public testing::WithParamInterface<OccValidationPolicy> {
  26. public:
  27. OptimisticTransactionDB* txn_db;
  28. string dbname;
  29. Options options;
  30. OptimisticTransactionTest() {
  31. options.create_if_missing = true;
  32. options.max_write_buffer_number = 2;
  33. options.max_write_buffer_size_to_maintain = 1600;
  34. dbname = test::PerThreadDBPath("optimistic_transaction_testdb");
  35. DestroyDB(dbname, options);
  36. Open();
  37. }
  38. ~OptimisticTransactionTest() override {
  39. delete txn_db;
  40. DestroyDB(dbname, options);
  41. }
  42. void Reopen() {
  43. delete txn_db;
  44. txn_db = nullptr;
  45. Open();
  46. }
  47. private:
  48. void Open() {
  49. ColumnFamilyOptions cf_options(options);
  50. OptimisticTransactionDBOptions occ_opts;
  51. occ_opts.validate_policy = GetParam();
  52. std::vector<ColumnFamilyDescriptor> column_families;
  53. std::vector<ColumnFamilyHandle*> handles;
  54. column_families.push_back(
  55. ColumnFamilyDescriptor(kDefaultColumnFamilyName, cf_options));
  56. Status s =
  57. OptimisticTransactionDB::Open(DBOptions(options), occ_opts, dbname,
  58. column_families, &handles, &txn_db);
  59. assert(s.ok());
  60. assert(txn_db != nullptr);
  61. assert(handles.size() == 1);
  62. delete handles[0];
  63. }
  64. };
  65. TEST_P(OptimisticTransactionTest, SuccessTest) {
  66. WriteOptions write_options;
  67. ReadOptions read_options;
  68. string value;
  69. Status s;
  70. txn_db->Put(write_options, Slice("foo"), Slice("bar"));
  71. txn_db->Put(write_options, Slice("foo2"), Slice("bar"));
  72. Transaction* txn = txn_db->BeginTransaction(write_options);
  73. ASSERT_TRUE(txn);
  74. txn->GetForUpdate(read_options, "foo", &value);
  75. ASSERT_EQ(value, "bar");
  76. txn->Put(Slice("foo"), Slice("bar2"));
  77. txn->GetForUpdate(read_options, "foo", &value);
  78. ASSERT_EQ(value, "bar2");
  79. s = txn->Commit();
  80. ASSERT_OK(s);
  81. txn_db->Get(read_options, "foo", &value);
  82. ASSERT_EQ(value, "bar2");
  83. delete txn;
  84. }
  85. TEST_P(OptimisticTransactionTest, WriteConflictTest) {
  86. WriteOptions write_options;
  87. ReadOptions read_options;
  88. string value;
  89. Status s;
  90. txn_db->Put(write_options, "foo", "bar");
  91. txn_db->Put(write_options, "foo2", "bar");
  92. Transaction* txn = txn_db->BeginTransaction(write_options);
  93. ASSERT_TRUE(txn);
  94. txn->Put("foo", "bar2");
  95. // This Put outside of a transaction will conflict with the previous write
  96. s = txn_db->Put(write_options, "foo", "barz");
  97. ASSERT_OK(s);
  98. s = txn_db->Get(read_options, "foo", &value);
  99. ASSERT_EQ(value, "barz");
  100. ASSERT_EQ(1, txn->GetNumKeys());
  101. s = txn->Commit();
  102. ASSERT_TRUE(s.IsBusy()); // Txn should not commit
  103. // Verify that transaction did not write anything
  104. txn_db->Get(read_options, "foo", &value);
  105. ASSERT_EQ(value, "barz");
  106. txn_db->Get(read_options, "foo2", &value);
  107. ASSERT_EQ(value, "bar");
  108. delete txn;
  109. }
  110. TEST_P(OptimisticTransactionTest, WriteConflictTest2) {
  111. WriteOptions write_options;
  112. ReadOptions read_options;
  113. OptimisticTransactionOptions txn_options;
  114. string value;
  115. Status s;
  116. txn_db->Put(write_options, "foo", "bar");
  117. txn_db->Put(write_options, "foo2", "bar");
  118. txn_options.set_snapshot = true;
  119. Transaction* txn = txn_db->BeginTransaction(write_options, txn_options);
  120. ASSERT_TRUE(txn);
  121. // This Put outside of a transaction will conflict with a later write
  122. s = txn_db->Put(write_options, "foo", "barz");
  123. ASSERT_OK(s);
  124. txn->Put("foo", "bar2"); // Conflicts with write done after snapshot taken
  125. s = txn_db->Get(read_options, "foo", &value);
  126. ASSERT_EQ(value, "barz");
  127. s = txn->Commit();
  128. ASSERT_TRUE(s.IsBusy()); // Txn should not commit
  129. // Verify that transaction did not write anything
  130. txn_db->Get(read_options, "foo", &value);
  131. ASSERT_EQ(value, "barz");
  132. txn_db->Get(read_options, "foo2", &value);
  133. ASSERT_EQ(value, "bar");
  134. delete txn;
  135. }
  136. TEST_P(OptimisticTransactionTest, ReadConflictTest) {
  137. WriteOptions write_options;
  138. ReadOptions read_options, snapshot_read_options;
  139. OptimisticTransactionOptions txn_options;
  140. string value;
  141. Status s;
  142. txn_db->Put(write_options, "foo", "bar");
  143. txn_db->Put(write_options, "foo2", "bar");
  144. txn_options.set_snapshot = true;
  145. Transaction* txn = txn_db->BeginTransaction(write_options, txn_options);
  146. ASSERT_TRUE(txn);
  147. txn->SetSnapshot();
  148. snapshot_read_options.snapshot = txn->GetSnapshot();
  149. txn->GetForUpdate(snapshot_read_options, "foo", &value);
  150. ASSERT_EQ(value, "bar");
  151. // This Put outside of a transaction will conflict with the previous read
  152. s = txn_db->Put(write_options, "foo", "barz");
  153. ASSERT_OK(s);
  154. s = txn_db->Get(read_options, "foo", &value);
  155. ASSERT_EQ(value, "barz");
  156. s = txn->Commit();
  157. ASSERT_TRUE(s.IsBusy()); // Txn should not commit
  158. // Verify that transaction did not write anything
  159. txn->GetForUpdate(read_options, "foo", &value);
  160. ASSERT_EQ(value, "barz");
  161. txn->GetForUpdate(read_options, "foo2", &value);
  162. ASSERT_EQ(value, "bar");
  163. delete txn;
  164. }
  165. TEST_P(OptimisticTransactionTest, TxnOnlyTest) {
  166. // Test to make sure transactions work when there are no other writes in an
  167. // empty db.
  168. WriteOptions write_options;
  169. ReadOptions read_options;
  170. string value;
  171. Status s;
  172. Transaction* txn = txn_db->BeginTransaction(write_options);
  173. ASSERT_TRUE(txn);
  174. txn->Put("x", "y");
  175. s = txn->Commit();
  176. ASSERT_OK(s);
  177. delete txn;
  178. }
  179. TEST_P(OptimisticTransactionTest, FlushTest) {
  180. WriteOptions write_options;
  181. ReadOptions read_options, snapshot_read_options;
  182. string value;
  183. Status s;
  184. txn_db->Put(write_options, Slice("foo"), Slice("bar"));
  185. txn_db->Put(write_options, Slice("foo2"), Slice("bar"));
  186. Transaction* txn = txn_db->BeginTransaction(write_options);
  187. ASSERT_TRUE(txn);
  188. snapshot_read_options.snapshot = txn->GetSnapshot();
  189. txn->GetForUpdate(snapshot_read_options, "foo", &value);
  190. ASSERT_EQ(value, "bar");
  191. txn->Put(Slice("foo"), Slice("bar2"));
  192. txn->GetForUpdate(snapshot_read_options, "foo", &value);
  193. ASSERT_EQ(value, "bar2");
  194. // Put a random key so we have a memtable to flush
  195. s = txn_db->Put(write_options, "dummy", "dummy");
  196. ASSERT_OK(s);
  197. // force a memtable flush
  198. FlushOptions flush_ops;
  199. txn_db->Flush(flush_ops);
  200. s = txn->Commit();
  201. // txn should commit since the flushed table is still in MemtableList History
  202. ASSERT_OK(s);
  203. txn_db->Get(read_options, "foo", &value);
  204. ASSERT_EQ(value, "bar2");
  205. delete txn;
  206. }
  207. TEST_P(OptimisticTransactionTest, FlushTest2) {
  208. WriteOptions write_options;
  209. ReadOptions read_options, snapshot_read_options;
  210. string value;
  211. Status s;
  212. txn_db->Put(write_options, Slice("foo"), Slice("bar"));
  213. txn_db->Put(write_options, Slice("foo2"), Slice("bar"));
  214. Transaction* txn = txn_db->BeginTransaction(write_options);
  215. ASSERT_TRUE(txn);
  216. snapshot_read_options.snapshot = txn->GetSnapshot();
  217. txn->GetForUpdate(snapshot_read_options, "foo", &value);
  218. ASSERT_EQ(value, "bar");
  219. txn->Put(Slice("foo"), Slice("bar2"));
  220. txn->GetForUpdate(snapshot_read_options, "foo", &value);
  221. ASSERT_EQ(value, "bar2");
  222. // Put a random key so we have a MemTable to flush
  223. s = txn_db->Put(write_options, "dummy", "dummy");
  224. ASSERT_OK(s);
  225. // force a memtable flush
  226. FlushOptions flush_ops;
  227. txn_db->Flush(flush_ops);
  228. // Put a random key so we have a MemTable to flush
  229. s = txn_db->Put(write_options, "dummy", "dummy2");
  230. ASSERT_OK(s);
  231. // force a memtable flush
  232. txn_db->Flush(flush_ops);
  233. s = txn_db->Put(write_options, "dummy", "dummy3");
  234. ASSERT_OK(s);
  235. // force a memtable flush
  236. // Since our test db has max_write_buffer_number=2, this flush will cause
  237. // the first memtable to get purged from the MemtableList history.
  238. txn_db->Flush(flush_ops);
  239. s = txn->Commit();
  240. // txn should not commit since MemTableList History is not large enough
  241. ASSERT_TRUE(s.IsTryAgain());
  242. txn_db->Get(read_options, "foo", &value);
  243. ASSERT_EQ(value, "bar");
  244. delete txn;
  245. }
  246. // Trigger the condition where some old memtables are skipped when doing
  247. // TransactionUtil::CheckKey(), and make sure the result is still correct.
  248. TEST_P(OptimisticTransactionTest, CheckKeySkipOldMemtable) {
  249. const int kAttemptHistoryMemtable = 0;
  250. const int kAttemptImmMemTable = 1;
  251. for (int attempt = kAttemptHistoryMemtable; attempt <= kAttemptImmMemTable;
  252. attempt++) {
  253. options.max_write_buffer_number_to_maintain = 3;
  254. Reopen();
  255. WriteOptions write_options;
  256. ReadOptions read_options;
  257. ReadOptions snapshot_read_options;
  258. ReadOptions snapshot_read_options2;
  259. string value;
  260. Status s;
  261. ASSERT_OK(txn_db->Put(write_options, Slice("foo"), Slice("bar")));
  262. ASSERT_OK(txn_db->Put(write_options, Slice("foo2"), Slice("bar")));
  263. Transaction* txn = txn_db->BeginTransaction(write_options);
  264. ASSERT_TRUE(txn != nullptr);
  265. Transaction* txn2 = txn_db->BeginTransaction(write_options);
  266. ASSERT_TRUE(txn2 != nullptr);
  267. snapshot_read_options.snapshot = txn->GetSnapshot();
  268. ASSERT_OK(txn->GetForUpdate(snapshot_read_options, "foo", &value));
  269. ASSERT_EQ(value, "bar");
  270. ASSERT_OK(txn->Put(Slice("foo"), Slice("bar2")));
  271. snapshot_read_options2.snapshot = txn2->GetSnapshot();
  272. ASSERT_OK(txn2->GetForUpdate(snapshot_read_options2, "foo2", &value));
  273. ASSERT_EQ(value, "bar");
  274. ASSERT_OK(txn2->Put(Slice("foo2"), Slice("bar2")));
  275. // txn updates "foo" and txn2 updates "foo2", and now a write is
  276. // issued for "foo", which conflicts with txn but not txn2
  277. ASSERT_OK(txn_db->Put(write_options, "foo", "bar"));
  278. if (attempt == kAttemptImmMemTable) {
  279. // For the second attempt, hold flush from beginning. The memtable
  280. // will be switched to immutable after calling TEST_SwitchMemtable()
  281. // while CheckKey() is called.
  282. ROCKSDB_NAMESPACE::SyncPoint::GetInstance()->LoadDependency(
  283. {{"OptimisticTransactionTest.CheckKeySkipOldMemtable",
  284. "FlushJob::Start"}});
  285. ROCKSDB_NAMESPACE::SyncPoint::GetInstance()->EnableProcessing();
  286. }
  287. // force a memtable flush. The memtable should still be kept
  288. FlushOptions flush_ops;
  289. if (attempt == kAttemptHistoryMemtable) {
  290. ASSERT_OK(txn_db->Flush(flush_ops));
  291. } else {
  292. assert(attempt == kAttemptImmMemTable);
  293. DBImpl* db_impl = static_cast<DBImpl*>(txn_db->GetRootDB());
  294. db_impl->TEST_SwitchMemtable();
  295. }
  296. uint64_t num_imm_mems;
  297. ASSERT_TRUE(txn_db->GetIntProperty(DB::Properties::kNumImmutableMemTable,
  298. &num_imm_mems));
  299. if (attempt == kAttemptHistoryMemtable) {
  300. ASSERT_EQ(0, num_imm_mems);
  301. } else {
  302. assert(attempt == kAttemptImmMemTable);
  303. ASSERT_EQ(1, num_imm_mems);
  304. }
  305. // Put something in active memtable
  306. ASSERT_OK(txn_db->Put(write_options, Slice("foo3"), Slice("bar")));
  307. // Create txn3 after flushing, when this transaction is commited,
  308. // only need to check the active memtable
  309. Transaction* txn3 = txn_db->BeginTransaction(write_options);
  310. ASSERT_TRUE(txn3 != nullptr);
  311. // Commit both of txn and txn2. txn will conflict but txn2 will
  312. // pass. In both ways, both memtables are queried.
  313. SetPerfLevel(PerfLevel::kEnableCount);
  314. get_perf_context()->Reset();
  315. s = txn->Commit();
  316. // We should have checked two memtables
  317. ASSERT_EQ(2, get_perf_context()->get_from_memtable_count);
  318. // txn should fail because of conflict, even if the memtable
  319. // has flushed, because it is still preserved in history.
  320. ASSERT_TRUE(s.IsBusy());
  321. get_perf_context()->Reset();
  322. s = txn2->Commit();
  323. // We should have checked two memtables
  324. ASSERT_EQ(2, get_perf_context()->get_from_memtable_count);
  325. ASSERT_TRUE(s.ok());
  326. txn3->Put(Slice("foo2"), Slice("bar2"));
  327. get_perf_context()->Reset();
  328. s = txn3->Commit();
  329. // txn3 is created after the active memtable is created, so that is the only
  330. // memtable to check.
  331. ASSERT_EQ(1, get_perf_context()->get_from_memtable_count);
  332. ASSERT_TRUE(s.ok());
  333. TEST_SYNC_POINT("OptimisticTransactionTest.CheckKeySkipOldMemtable");
  334. ROCKSDB_NAMESPACE::SyncPoint::GetInstance()->DisableProcessing();
  335. SetPerfLevel(PerfLevel::kDisable);
  336. delete txn;
  337. delete txn2;
  338. delete txn3;
  339. }
  340. }
  341. TEST_P(OptimisticTransactionTest, NoSnapshotTest) {
  342. WriteOptions write_options;
  343. ReadOptions read_options;
  344. string value;
  345. Status s;
  346. txn_db->Put(write_options, "AAA", "bar");
  347. Transaction* txn = txn_db->BeginTransaction(write_options);
  348. ASSERT_TRUE(txn);
  349. // Modify key after transaction start
  350. txn_db->Put(write_options, "AAA", "bar1");
  351. // Read and write without a snapshot
  352. txn->GetForUpdate(read_options, "AAA", &value);
  353. ASSERT_EQ(value, "bar1");
  354. txn->Put("AAA", "bar2");
  355. // Should commit since read/write was done after data changed
  356. s = txn->Commit();
  357. ASSERT_OK(s);
  358. txn->GetForUpdate(read_options, "AAA", &value);
  359. ASSERT_EQ(value, "bar2");
  360. delete txn;
  361. }
  362. TEST_P(OptimisticTransactionTest, MultipleSnapshotTest) {
  363. WriteOptions write_options;
  364. ReadOptions read_options, snapshot_read_options;
  365. string value;
  366. Status s;
  367. txn_db->Put(write_options, "AAA", "bar");
  368. txn_db->Put(write_options, "BBB", "bar");
  369. txn_db->Put(write_options, "CCC", "bar");
  370. Transaction* txn = txn_db->BeginTransaction(write_options);
  371. ASSERT_TRUE(txn);
  372. txn_db->Put(write_options, "AAA", "bar1");
  373. // Read and write without a snapshot
  374. txn->GetForUpdate(read_options, "AAA", &value);
  375. ASSERT_EQ(value, "bar1");
  376. txn->Put("AAA", "bar2");
  377. // Modify BBB before snapshot is taken
  378. txn_db->Put(write_options, "BBB", "bar1");
  379. txn->SetSnapshot();
  380. snapshot_read_options.snapshot = txn->GetSnapshot();
  381. // Read and write with snapshot
  382. txn->GetForUpdate(snapshot_read_options, "BBB", &value);
  383. ASSERT_EQ(value, "bar1");
  384. txn->Put("BBB", "bar2");
  385. txn_db->Put(write_options, "CCC", "bar1");
  386. // Set a new snapshot
  387. txn->SetSnapshot();
  388. snapshot_read_options.snapshot = txn->GetSnapshot();
  389. // Read and write with snapshot
  390. txn->GetForUpdate(snapshot_read_options, "CCC", &value);
  391. ASSERT_EQ(value, "bar1");
  392. txn->Put("CCC", "bar2");
  393. s = txn->GetForUpdate(read_options, "AAA", &value);
  394. ASSERT_OK(s);
  395. ASSERT_EQ(value, "bar2");
  396. s = txn->GetForUpdate(read_options, "BBB", &value);
  397. ASSERT_OK(s);
  398. ASSERT_EQ(value, "bar2");
  399. s = txn->GetForUpdate(read_options, "CCC", &value);
  400. ASSERT_OK(s);
  401. ASSERT_EQ(value, "bar2");
  402. s = txn_db->Get(read_options, "AAA", &value);
  403. ASSERT_OK(s);
  404. ASSERT_EQ(value, "bar1");
  405. s = txn_db->Get(read_options, "BBB", &value);
  406. ASSERT_OK(s);
  407. ASSERT_EQ(value, "bar1");
  408. s = txn_db->Get(read_options, "CCC", &value);
  409. ASSERT_OK(s);
  410. ASSERT_EQ(value, "bar1");
  411. s = txn->Commit();
  412. ASSERT_OK(s);
  413. s = txn_db->Get(read_options, "AAA", &value);
  414. ASSERT_OK(s);
  415. ASSERT_EQ(value, "bar2");
  416. s = txn_db->Get(read_options, "BBB", &value);
  417. ASSERT_OK(s);
  418. ASSERT_EQ(value, "bar2");
  419. s = txn_db->Get(read_options, "CCC", &value);
  420. ASSERT_OK(s);
  421. ASSERT_EQ(value, "bar2");
  422. // verify that we track multiple writes to the same key at different snapshots
  423. delete txn;
  424. txn = txn_db->BeginTransaction(write_options);
  425. // Potentially conflicting writes
  426. txn_db->Put(write_options, "ZZZ", "zzz");
  427. txn_db->Put(write_options, "XXX", "xxx");
  428. txn->SetSnapshot();
  429. OptimisticTransactionOptions txn_options;
  430. txn_options.set_snapshot = true;
  431. Transaction* txn2 = txn_db->BeginTransaction(write_options, txn_options);
  432. txn2->SetSnapshot();
  433. // This should not conflict in txn since the snapshot is later than the
  434. // previous write (spoiler alert: it will later conflict with txn2).
  435. txn->Put("ZZZ", "zzzz");
  436. s = txn->Commit();
  437. ASSERT_OK(s);
  438. delete txn;
  439. // This will conflict since the snapshot is earlier than another write to ZZZ
  440. txn2->Put("ZZZ", "xxxxx");
  441. s = txn2->Commit();
  442. ASSERT_TRUE(s.IsBusy());
  443. delete txn2;
  444. }
  445. TEST_P(OptimisticTransactionTest, ColumnFamiliesTest) {
  446. WriteOptions write_options;
  447. ReadOptions read_options, snapshot_read_options;
  448. OptimisticTransactionOptions txn_options;
  449. string value;
  450. Status s;
  451. ColumnFamilyHandle *cfa, *cfb;
  452. ColumnFamilyOptions cf_options;
  453. // Create 2 new column families
  454. s = txn_db->CreateColumnFamily(cf_options, "CFA", &cfa);
  455. ASSERT_OK(s);
  456. s = txn_db->CreateColumnFamily(cf_options, "CFB", &cfb);
  457. ASSERT_OK(s);
  458. delete cfa;
  459. delete cfb;
  460. delete txn_db;
  461. txn_db = nullptr;
  462. // open DB with three column families
  463. std::vector<ColumnFamilyDescriptor> column_families;
  464. // have to open default column family
  465. column_families.push_back(
  466. ColumnFamilyDescriptor(kDefaultColumnFamilyName, ColumnFamilyOptions()));
  467. // open the new column families
  468. column_families.push_back(
  469. ColumnFamilyDescriptor("CFA", ColumnFamilyOptions()));
  470. column_families.push_back(
  471. ColumnFamilyDescriptor("CFB", ColumnFamilyOptions()));
  472. std::vector<ColumnFamilyHandle*> handles;
  473. s = OptimisticTransactionDB::Open(options, dbname, column_families, &handles,
  474. &txn_db);
  475. ASSERT_OK(s);
  476. assert(txn_db != nullptr);
  477. Transaction* txn = txn_db->BeginTransaction(write_options);
  478. ASSERT_TRUE(txn);
  479. txn->SetSnapshot();
  480. snapshot_read_options.snapshot = txn->GetSnapshot();
  481. txn_options.set_snapshot = true;
  482. Transaction* txn2 = txn_db->BeginTransaction(write_options, txn_options);
  483. ASSERT_TRUE(txn2);
  484. // Write some data to the db
  485. WriteBatch batch;
  486. batch.Put("foo", "foo");
  487. batch.Put(handles[1], "AAA", "bar");
  488. batch.Put(handles[1], "AAAZZZ", "bar");
  489. s = txn_db->Write(write_options, &batch);
  490. ASSERT_OK(s);
  491. txn_db->Delete(write_options, handles[1], "AAAZZZ");
  492. // These keys do no conflict with existing writes since they're in
  493. // different column families
  494. txn->Delete("AAA");
  495. txn->GetForUpdate(snapshot_read_options, handles[1], "foo", &value);
  496. Slice key_slice("AAAZZZ");
  497. Slice value_slices[2] = {Slice("bar"), Slice("bar")};
  498. txn->Put(handles[2], SliceParts(&key_slice, 1), SliceParts(value_slices, 2));
  499. ASSERT_EQ(3, txn->GetNumKeys());
  500. // Txn should commit
  501. s = txn->Commit();
  502. ASSERT_OK(s);
  503. s = txn_db->Get(read_options, "AAA", &value);
  504. ASSERT_TRUE(s.IsNotFound());
  505. s = txn_db->Get(read_options, handles[2], "AAAZZZ", &value);
  506. ASSERT_EQ(value, "barbar");
  507. Slice key_slices[3] = {Slice("AAA"), Slice("ZZ"), Slice("Z")};
  508. Slice value_slice("barbarbar");
  509. // This write will cause a conflict with the earlier batch write
  510. txn2->Put(handles[1], SliceParts(key_slices, 3), SliceParts(&value_slice, 1));
  511. txn2->Delete(handles[2], "XXX");
  512. txn2->Delete(handles[1], "XXX");
  513. s = txn2->GetForUpdate(snapshot_read_options, handles[1], "AAA", &value);
  514. ASSERT_TRUE(s.IsNotFound());
  515. // Verify txn did not commit
  516. s = txn2->Commit();
  517. ASSERT_TRUE(s.IsBusy());
  518. s = txn_db->Get(read_options, handles[1], "AAAZZZ", &value);
  519. ASSERT_EQ(value, "barbar");
  520. delete txn;
  521. delete txn2;
  522. txn = txn_db->BeginTransaction(write_options, txn_options);
  523. snapshot_read_options.snapshot = txn->GetSnapshot();
  524. txn2 = txn_db->BeginTransaction(write_options, txn_options);
  525. ASSERT_TRUE(txn);
  526. std::vector<ColumnFamilyHandle*> multiget_cfh = {handles[1], handles[2],
  527. handles[0], handles[2]};
  528. std::vector<Slice> multiget_keys = {"AAA", "AAAZZZ", "foo", "foo"};
  529. std::vector<std::string> values(4);
  530. std::vector<Status> results = txn->MultiGetForUpdate(
  531. snapshot_read_options, multiget_cfh, multiget_keys, &values);
  532. ASSERT_OK(results[0]);
  533. ASSERT_OK(results[1]);
  534. ASSERT_OK(results[2]);
  535. ASSERT_TRUE(results[3].IsNotFound());
  536. ASSERT_EQ(values[0], "bar");
  537. ASSERT_EQ(values[1], "barbar");
  538. ASSERT_EQ(values[2], "foo");
  539. txn->Delete(handles[2], "ZZZ");
  540. txn->Put(handles[2], "ZZZ", "YYY");
  541. txn->Put(handles[2], "ZZZ", "YYYY");
  542. txn->Delete(handles[2], "ZZZ");
  543. txn->Put(handles[2], "AAAZZZ", "barbarbar");
  544. ASSERT_EQ(5, txn->GetNumKeys());
  545. // Txn should commit
  546. s = txn->Commit();
  547. ASSERT_OK(s);
  548. s = txn_db->Get(read_options, handles[2], "ZZZ", &value);
  549. ASSERT_TRUE(s.IsNotFound());
  550. // Put a key which will conflict with the next txn using the previous snapshot
  551. txn_db->Put(write_options, handles[2], "foo", "000");
  552. results = txn2->MultiGetForUpdate(snapshot_read_options, multiget_cfh,
  553. multiget_keys, &values);
  554. ASSERT_OK(results[0]);
  555. ASSERT_OK(results[1]);
  556. ASSERT_OK(results[2]);
  557. ASSERT_TRUE(results[3].IsNotFound());
  558. ASSERT_EQ(values[0], "bar");
  559. ASSERT_EQ(values[1], "barbar");
  560. ASSERT_EQ(values[2], "foo");
  561. // Verify Txn Did not Commit
  562. s = txn2->Commit();
  563. ASSERT_TRUE(s.IsBusy());
  564. s = txn_db->DropColumnFamily(handles[1]);
  565. ASSERT_OK(s);
  566. s = txn_db->DropColumnFamily(handles[2]);
  567. ASSERT_OK(s);
  568. delete txn;
  569. delete txn2;
  570. for (auto handle : handles) {
  571. delete handle;
  572. }
  573. }
  574. TEST_P(OptimisticTransactionTest, EmptyTest) {
  575. WriteOptions write_options;
  576. ReadOptions read_options;
  577. string value;
  578. Status s;
  579. s = txn_db->Put(write_options, "aaa", "aaa");
  580. ASSERT_OK(s);
  581. Transaction* txn = txn_db->BeginTransaction(write_options);
  582. s = txn->Commit();
  583. ASSERT_OK(s);
  584. delete txn;
  585. txn = txn_db->BeginTransaction(write_options);
  586. txn->Rollback();
  587. delete txn;
  588. txn = txn_db->BeginTransaction(write_options);
  589. s = txn->GetForUpdate(read_options, "aaa", &value);
  590. ASSERT_EQ(value, "aaa");
  591. s = txn->Commit();
  592. ASSERT_OK(s);
  593. delete txn;
  594. txn = txn_db->BeginTransaction(write_options);
  595. txn->SetSnapshot();
  596. s = txn->GetForUpdate(read_options, "aaa", &value);
  597. ASSERT_EQ(value, "aaa");
  598. s = txn_db->Put(write_options, "aaa", "xxx");
  599. s = txn->Commit();
  600. ASSERT_TRUE(s.IsBusy());
  601. delete txn;
  602. }
  603. TEST_P(OptimisticTransactionTest, PredicateManyPreceders) {
  604. WriteOptions write_options;
  605. ReadOptions read_options1, read_options2;
  606. OptimisticTransactionOptions txn_options;
  607. string value;
  608. Status s;
  609. txn_options.set_snapshot = true;
  610. Transaction* txn1 = txn_db->BeginTransaction(write_options, txn_options);
  611. read_options1.snapshot = txn1->GetSnapshot();
  612. Transaction* txn2 = txn_db->BeginTransaction(write_options);
  613. txn2->SetSnapshot();
  614. read_options2.snapshot = txn2->GetSnapshot();
  615. std::vector<Slice> multiget_keys = {"1", "2", "3"};
  616. std::vector<std::string> multiget_values;
  617. std::vector<Status> results =
  618. txn1->MultiGetForUpdate(read_options1, multiget_keys, &multiget_values);
  619. ASSERT_TRUE(results[1].IsNotFound());
  620. txn2->Put("2", "x");
  621. s = txn2->Commit();
  622. ASSERT_OK(s);
  623. multiget_values.clear();
  624. results =
  625. txn1->MultiGetForUpdate(read_options1, multiget_keys, &multiget_values);
  626. ASSERT_TRUE(results[1].IsNotFound());
  627. // should not commit since txn2 wrote a key txn has read
  628. s = txn1->Commit();
  629. ASSERT_TRUE(s.IsBusy());
  630. delete txn1;
  631. delete txn2;
  632. txn1 = txn_db->BeginTransaction(write_options, txn_options);
  633. read_options1.snapshot = txn1->GetSnapshot();
  634. txn2 = txn_db->BeginTransaction(write_options, txn_options);
  635. read_options2.snapshot = txn2->GetSnapshot();
  636. txn1->Put("4", "x");
  637. txn2->Delete("4");
  638. // txn1 can commit since txn2's delete hasn't happened yet (it's just batched)
  639. s = txn1->Commit();
  640. ASSERT_OK(s);
  641. s = txn2->GetForUpdate(read_options2, "4", &value);
  642. ASSERT_TRUE(s.IsNotFound());
  643. // txn2 cannot commit since txn1 changed "4"
  644. s = txn2->Commit();
  645. ASSERT_TRUE(s.IsBusy());
  646. delete txn1;
  647. delete txn2;
  648. }
  649. TEST_P(OptimisticTransactionTest, LostUpdate) {
  650. WriteOptions write_options;
  651. ReadOptions read_options, read_options1, read_options2;
  652. OptimisticTransactionOptions txn_options;
  653. string value;
  654. Status s;
  655. // Test 2 transactions writing to the same key in multiple orders and
  656. // with/without snapshots
  657. Transaction* txn1 = txn_db->BeginTransaction(write_options);
  658. Transaction* txn2 = txn_db->BeginTransaction(write_options);
  659. txn1->Put("1", "1");
  660. txn2->Put("1", "2");
  661. s = txn1->Commit();
  662. ASSERT_OK(s);
  663. s = txn2->Commit();
  664. ASSERT_TRUE(s.IsBusy());
  665. delete txn1;
  666. delete txn2;
  667. txn_options.set_snapshot = true;
  668. txn1 = txn_db->BeginTransaction(write_options, txn_options);
  669. read_options1.snapshot = txn1->GetSnapshot();
  670. txn2 = txn_db->BeginTransaction(write_options, txn_options);
  671. read_options2.snapshot = txn2->GetSnapshot();
  672. txn1->Put("1", "3");
  673. txn2->Put("1", "4");
  674. s = txn1->Commit();
  675. ASSERT_OK(s);
  676. s = txn2->Commit();
  677. ASSERT_TRUE(s.IsBusy());
  678. delete txn1;
  679. delete txn2;
  680. txn1 = txn_db->BeginTransaction(write_options, txn_options);
  681. read_options1.snapshot = txn1->GetSnapshot();
  682. txn2 = txn_db->BeginTransaction(write_options, txn_options);
  683. read_options2.snapshot = txn2->GetSnapshot();
  684. txn1->Put("1", "5");
  685. s = txn1->Commit();
  686. ASSERT_OK(s);
  687. txn2->Put("1", "6");
  688. s = txn2->Commit();
  689. ASSERT_TRUE(s.IsBusy());
  690. delete txn1;
  691. delete txn2;
  692. txn1 = txn_db->BeginTransaction(write_options, txn_options);
  693. read_options1.snapshot = txn1->GetSnapshot();
  694. txn2 = txn_db->BeginTransaction(write_options, txn_options);
  695. read_options2.snapshot = txn2->GetSnapshot();
  696. txn1->Put("1", "5");
  697. s = txn1->Commit();
  698. ASSERT_OK(s);
  699. txn2->SetSnapshot();
  700. txn2->Put("1", "6");
  701. s = txn2->Commit();
  702. ASSERT_OK(s);
  703. delete txn1;
  704. delete txn2;
  705. txn1 = txn_db->BeginTransaction(write_options);
  706. txn2 = txn_db->BeginTransaction(write_options);
  707. txn1->Put("1", "7");
  708. s = txn1->Commit();
  709. ASSERT_OK(s);
  710. txn2->Put("1", "8");
  711. s = txn2->Commit();
  712. ASSERT_OK(s);
  713. delete txn1;
  714. delete txn2;
  715. s = txn_db->Get(read_options, "1", &value);
  716. ASSERT_OK(s);
  717. ASSERT_EQ(value, "8");
  718. }
  719. TEST_P(OptimisticTransactionTest, UntrackedWrites) {
  720. WriteOptions write_options;
  721. ReadOptions read_options;
  722. string value;
  723. Status s;
  724. // Verify transaction rollback works for untracked keys.
  725. Transaction* txn = txn_db->BeginTransaction(write_options);
  726. txn->PutUntracked("untracked", "0");
  727. txn->Rollback();
  728. s = txn_db->Get(read_options, "untracked", &value);
  729. ASSERT_TRUE(s.IsNotFound());
  730. delete txn;
  731. txn = txn_db->BeginTransaction(write_options);
  732. txn->Put("tracked", "1");
  733. txn->PutUntracked("untracked", "1");
  734. txn->MergeUntracked("untracked", "2");
  735. txn->DeleteUntracked("untracked");
  736. // Write to the untracked key outside of the transaction and verify
  737. // it doesn't prevent the transaction from committing.
  738. s = txn_db->Put(write_options, "untracked", "x");
  739. ASSERT_OK(s);
  740. s = txn->Commit();
  741. ASSERT_OK(s);
  742. s = txn_db->Get(read_options, "untracked", &value);
  743. ASSERT_TRUE(s.IsNotFound());
  744. delete txn;
  745. txn = txn_db->BeginTransaction(write_options);
  746. txn->Put("tracked", "10");
  747. txn->PutUntracked("untracked", "A");
  748. // Write to tracked key outside of the transaction and verify that the
  749. // untracked keys are not written when the commit fails.
  750. s = txn_db->Delete(write_options, "tracked");
  751. s = txn->Commit();
  752. ASSERT_TRUE(s.IsBusy());
  753. s = txn_db->Get(read_options, "untracked", &value);
  754. ASSERT_TRUE(s.IsNotFound());
  755. delete txn;
  756. }
  757. TEST_P(OptimisticTransactionTest, IteratorTest) {
  758. WriteOptions write_options;
  759. ReadOptions read_options, snapshot_read_options;
  760. OptimisticTransactionOptions txn_options;
  761. string value;
  762. Status s;
  763. // Write some keys to the db
  764. s = txn_db->Put(write_options, "A", "a");
  765. ASSERT_OK(s);
  766. s = txn_db->Put(write_options, "G", "g");
  767. ASSERT_OK(s);
  768. s = txn_db->Put(write_options, "F", "f");
  769. ASSERT_OK(s);
  770. s = txn_db->Put(write_options, "C", "c");
  771. ASSERT_OK(s);
  772. s = txn_db->Put(write_options, "D", "d");
  773. ASSERT_OK(s);
  774. Transaction* txn = txn_db->BeginTransaction(write_options);
  775. ASSERT_TRUE(txn);
  776. // Write some keys in a txn
  777. s = txn->Put("B", "b");
  778. ASSERT_OK(s);
  779. s = txn->Put("H", "h");
  780. ASSERT_OK(s);
  781. s = txn->Delete("D");
  782. ASSERT_OK(s);
  783. s = txn->Put("E", "e");
  784. ASSERT_OK(s);
  785. txn->SetSnapshot();
  786. const Snapshot* snapshot = txn->GetSnapshot();
  787. // Write some keys to the db after the snapshot
  788. s = txn_db->Put(write_options, "BB", "xx");
  789. ASSERT_OK(s);
  790. s = txn_db->Put(write_options, "C", "xx");
  791. ASSERT_OK(s);
  792. read_options.snapshot = snapshot;
  793. Iterator* iter = txn->GetIterator(read_options);
  794. ASSERT_OK(iter->status());
  795. iter->SeekToFirst();
  796. // Read all keys via iter and lock them all
  797. std::string results[] = {"a", "b", "c", "e", "f", "g", "h"};
  798. for (int i = 0; i < 7; i++) {
  799. ASSERT_OK(iter->status());
  800. ASSERT_TRUE(iter->Valid());
  801. ASSERT_EQ(results[i], iter->value().ToString());
  802. s = txn->GetForUpdate(read_options, iter->key(), nullptr);
  803. ASSERT_OK(s);
  804. iter->Next();
  805. }
  806. ASSERT_FALSE(iter->Valid());
  807. iter->Seek("G");
  808. ASSERT_OK(iter->status());
  809. ASSERT_TRUE(iter->Valid());
  810. ASSERT_EQ("g", iter->value().ToString());
  811. iter->Prev();
  812. ASSERT_OK(iter->status());
  813. ASSERT_TRUE(iter->Valid());
  814. ASSERT_EQ("f", iter->value().ToString());
  815. iter->Seek("D");
  816. ASSERT_OK(iter->status());
  817. ASSERT_TRUE(iter->Valid());
  818. ASSERT_EQ("e", iter->value().ToString());
  819. iter->Seek("C");
  820. ASSERT_OK(iter->status());
  821. ASSERT_TRUE(iter->Valid());
  822. ASSERT_EQ("c", iter->value().ToString());
  823. iter->Next();
  824. ASSERT_OK(iter->status());
  825. ASSERT_TRUE(iter->Valid());
  826. ASSERT_EQ("e", iter->value().ToString());
  827. iter->Seek("");
  828. ASSERT_OK(iter->status());
  829. ASSERT_TRUE(iter->Valid());
  830. ASSERT_EQ("a", iter->value().ToString());
  831. iter->Seek("X");
  832. ASSERT_OK(iter->status());
  833. ASSERT_FALSE(iter->Valid());
  834. iter->SeekToLast();
  835. ASSERT_OK(iter->status());
  836. ASSERT_TRUE(iter->Valid());
  837. ASSERT_EQ("h", iter->value().ToString());
  838. // key "C" was modified in the db after txn's snapshot. txn will not commit.
  839. s = txn->Commit();
  840. ASSERT_TRUE(s.IsBusy());
  841. delete iter;
  842. delete txn;
  843. }
  844. TEST_P(OptimisticTransactionTest, SavepointTest) {
  845. WriteOptions write_options;
  846. ReadOptions read_options, snapshot_read_options;
  847. OptimisticTransactionOptions txn_options;
  848. string value;
  849. Status s;
  850. Transaction* txn = txn_db->BeginTransaction(write_options);
  851. ASSERT_TRUE(txn);
  852. s = txn->RollbackToSavePoint();
  853. ASSERT_TRUE(s.IsNotFound());
  854. txn->SetSavePoint(); // 1
  855. ASSERT_OK(txn->RollbackToSavePoint()); // Rollback to beginning of txn
  856. s = txn->RollbackToSavePoint();
  857. ASSERT_TRUE(s.IsNotFound());
  858. s = txn->Put("B", "b");
  859. ASSERT_OK(s);
  860. s = txn->Commit();
  861. ASSERT_OK(s);
  862. s = txn_db->Get(read_options, "B", &value);
  863. ASSERT_OK(s);
  864. ASSERT_EQ("b", value);
  865. delete txn;
  866. txn = txn_db->BeginTransaction(write_options);
  867. ASSERT_TRUE(txn);
  868. s = txn->Put("A", "a");
  869. ASSERT_OK(s);
  870. s = txn->Put("B", "bb");
  871. ASSERT_OK(s);
  872. s = txn->Put("C", "c");
  873. ASSERT_OK(s);
  874. txn->SetSavePoint(); // 2
  875. s = txn->Delete("B");
  876. ASSERT_OK(s);
  877. s = txn->Put("C", "cc");
  878. ASSERT_OK(s);
  879. s = txn->Put("D", "d");
  880. ASSERT_OK(s);
  881. ASSERT_OK(txn->RollbackToSavePoint()); // Rollback to 2
  882. s = txn->Get(read_options, "A", &value);
  883. ASSERT_OK(s);
  884. ASSERT_EQ("a", value);
  885. s = txn->Get(read_options, "B", &value);
  886. ASSERT_OK(s);
  887. ASSERT_EQ("bb", value);
  888. s = txn->Get(read_options, "C", &value);
  889. ASSERT_OK(s);
  890. ASSERT_EQ("c", value);
  891. s = txn->Get(read_options, "D", &value);
  892. ASSERT_TRUE(s.IsNotFound());
  893. s = txn->Put("A", "a");
  894. ASSERT_OK(s);
  895. s = txn->Put("E", "e");
  896. ASSERT_OK(s);
  897. // Rollback to beginning of txn
  898. s = txn->RollbackToSavePoint();
  899. ASSERT_TRUE(s.IsNotFound());
  900. txn->Rollback();
  901. s = txn->Get(read_options, "A", &value);
  902. ASSERT_TRUE(s.IsNotFound());
  903. s = txn->Get(read_options, "B", &value);
  904. ASSERT_OK(s);
  905. ASSERT_EQ("b", value);
  906. s = txn->Get(read_options, "D", &value);
  907. ASSERT_TRUE(s.IsNotFound());
  908. s = txn->Get(read_options, "D", &value);
  909. ASSERT_TRUE(s.IsNotFound());
  910. s = txn->Get(read_options, "E", &value);
  911. ASSERT_TRUE(s.IsNotFound());
  912. s = txn->Put("A", "aa");
  913. ASSERT_OK(s);
  914. s = txn->Put("F", "f");
  915. ASSERT_OK(s);
  916. txn->SetSavePoint(); // 3
  917. txn->SetSavePoint(); // 4
  918. s = txn->Put("G", "g");
  919. ASSERT_OK(s);
  920. s = txn->Delete("F");
  921. ASSERT_OK(s);
  922. s = txn->Delete("B");
  923. ASSERT_OK(s);
  924. s = txn->Get(read_options, "A", &value);
  925. ASSERT_OK(s);
  926. ASSERT_EQ("aa", value);
  927. s = txn->Get(read_options, "F", &value);
  928. ASSERT_TRUE(s.IsNotFound());
  929. s = txn->Get(read_options, "B", &value);
  930. ASSERT_TRUE(s.IsNotFound());
  931. ASSERT_OK(txn->RollbackToSavePoint()); // Rollback to 3
  932. s = txn->Get(read_options, "F", &value);
  933. ASSERT_OK(s);
  934. ASSERT_EQ("f", value);
  935. s = txn->Get(read_options, "G", &value);
  936. ASSERT_TRUE(s.IsNotFound());
  937. s = txn->Commit();
  938. ASSERT_OK(s);
  939. s = txn_db->Get(read_options, "F", &value);
  940. ASSERT_OK(s);
  941. ASSERT_EQ("f", value);
  942. s = txn_db->Get(read_options, "G", &value);
  943. ASSERT_TRUE(s.IsNotFound());
  944. s = txn_db->Get(read_options, "A", &value);
  945. ASSERT_OK(s);
  946. ASSERT_EQ("aa", value);
  947. s = txn_db->Get(read_options, "B", &value);
  948. ASSERT_OK(s);
  949. ASSERT_EQ("b", value);
  950. s = txn_db->Get(read_options, "C", &value);
  951. ASSERT_TRUE(s.IsNotFound());
  952. s = txn_db->Get(read_options, "D", &value);
  953. ASSERT_TRUE(s.IsNotFound());
  954. s = txn_db->Get(read_options, "E", &value);
  955. ASSERT_TRUE(s.IsNotFound());
  956. delete txn;
  957. }
  958. TEST_P(OptimisticTransactionTest, UndoGetForUpdateTest) {
  959. WriteOptions write_options;
  960. ReadOptions read_options, snapshot_read_options;
  961. OptimisticTransactionOptions txn_options;
  962. string value;
  963. Status s;
  964. txn_db->Put(write_options, "A", "");
  965. Transaction* txn1 = txn_db->BeginTransaction(write_options);
  966. ASSERT_TRUE(txn1);
  967. s = txn1->GetForUpdate(read_options, "A", &value);
  968. ASSERT_OK(s);
  969. txn1->UndoGetForUpdate("A");
  970. Transaction* txn2 = txn_db->BeginTransaction(write_options);
  971. txn2->Put("A", "x");
  972. s = txn2->Commit();
  973. ASSERT_OK(s);
  974. delete txn2;
  975. // Verify that txn1 can commit since A isn't conflict checked
  976. s = txn1->Commit();
  977. ASSERT_OK(s);
  978. delete txn1;
  979. txn1 = txn_db->BeginTransaction(write_options);
  980. txn1->Put("A", "a");
  981. s = txn1->GetForUpdate(read_options, "A", &value);
  982. ASSERT_OK(s);
  983. txn1->UndoGetForUpdate("A");
  984. txn2 = txn_db->BeginTransaction(write_options);
  985. txn2->Put("A", "x");
  986. s = txn2->Commit();
  987. ASSERT_OK(s);
  988. delete txn2;
  989. // Verify that txn1 cannot commit since A will still be conflict checked
  990. s = txn1->Commit();
  991. ASSERT_TRUE(s.IsBusy());
  992. delete txn1;
  993. txn1 = txn_db->BeginTransaction(write_options);
  994. s = txn1->GetForUpdate(read_options, "A", &value);
  995. ASSERT_OK(s);
  996. s = txn1->GetForUpdate(read_options, "A", &value);
  997. ASSERT_OK(s);
  998. txn1->UndoGetForUpdate("A");
  999. txn2 = txn_db->BeginTransaction(write_options);
  1000. txn2->Put("A", "x");
  1001. s = txn2->Commit();
  1002. ASSERT_OK(s);
  1003. delete txn2;
  1004. // Verify that txn1 cannot commit since A will still be conflict checked
  1005. s = txn1->Commit();
  1006. ASSERT_TRUE(s.IsBusy());
  1007. delete txn1;
  1008. txn1 = txn_db->BeginTransaction(write_options);
  1009. s = txn1->GetForUpdate(read_options, "A", &value);
  1010. ASSERT_OK(s);
  1011. s = txn1->GetForUpdate(read_options, "A", &value);
  1012. ASSERT_OK(s);
  1013. txn1->UndoGetForUpdate("A");
  1014. txn1->UndoGetForUpdate("A");
  1015. txn2 = txn_db->BeginTransaction(write_options);
  1016. txn2->Put("A", "x");
  1017. s = txn2->Commit();
  1018. ASSERT_OK(s);
  1019. delete txn2;
  1020. // Verify that txn1 can commit since A isn't conflict checked
  1021. s = txn1->Commit();
  1022. ASSERT_OK(s);
  1023. delete txn1;
  1024. txn1 = txn_db->BeginTransaction(write_options);
  1025. s = txn1->GetForUpdate(read_options, "A", &value);
  1026. ASSERT_OK(s);
  1027. txn1->SetSavePoint();
  1028. txn1->UndoGetForUpdate("A");
  1029. txn2 = txn_db->BeginTransaction(write_options);
  1030. txn2->Put("A", "x");
  1031. s = txn2->Commit();
  1032. ASSERT_OK(s);
  1033. delete txn2;
  1034. // Verify that txn1 cannot commit since A will still be conflict checked
  1035. s = txn1->Commit();
  1036. ASSERT_TRUE(s.IsBusy());
  1037. delete txn1;
  1038. txn1 = txn_db->BeginTransaction(write_options);
  1039. s = txn1->GetForUpdate(read_options, "A", &value);
  1040. ASSERT_OK(s);
  1041. txn1->SetSavePoint();
  1042. s = txn1->GetForUpdate(read_options, "A", &value);
  1043. ASSERT_OK(s);
  1044. txn1->UndoGetForUpdate("A");
  1045. txn2 = txn_db->BeginTransaction(write_options);
  1046. txn2->Put("A", "x");
  1047. s = txn2->Commit();
  1048. ASSERT_OK(s);
  1049. delete txn2;
  1050. // Verify that txn1 cannot commit since A will still be conflict checked
  1051. s = txn1->Commit();
  1052. ASSERT_TRUE(s.IsBusy());
  1053. delete txn1;
  1054. txn1 = txn_db->BeginTransaction(write_options);
  1055. s = txn1->GetForUpdate(read_options, "A", &value);
  1056. ASSERT_OK(s);
  1057. txn1->SetSavePoint();
  1058. s = txn1->GetForUpdate(read_options, "A", &value);
  1059. ASSERT_OK(s);
  1060. txn1->UndoGetForUpdate("A");
  1061. txn1->RollbackToSavePoint();
  1062. txn1->UndoGetForUpdate("A");
  1063. txn2 = txn_db->BeginTransaction(write_options);
  1064. txn2->Put("A", "x");
  1065. s = txn2->Commit();
  1066. ASSERT_OK(s);
  1067. delete txn2;
  1068. // Verify that txn1 can commit since A isn't conflict checked
  1069. s = txn1->Commit();
  1070. ASSERT_OK(s);
  1071. delete txn1;
  1072. }
  1073. namespace {
  1074. Status OptimisticTransactionStressTestInserter(OptimisticTransactionDB* db,
  1075. const size_t num_transactions,
  1076. const size_t num_sets,
  1077. const size_t num_keys_per_set) {
  1078. size_t seed = std::hash<std::thread::id>()(std::this_thread::get_id());
  1079. Random64 _rand(seed);
  1080. WriteOptions write_options;
  1081. ReadOptions read_options;
  1082. OptimisticTransactionOptions txn_options;
  1083. txn_options.set_snapshot = true;
  1084. RandomTransactionInserter inserter(&_rand, write_options, read_options,
  1085. num_keys_per_set,
  1086. static_cast<uint16_t>(num_sets));
  1087. for (size_t t = 0; t < num_transactions; t++) {
  1088. bool success = inserter.OptimisticTransactionDBInsert(db, txn_options);
  1089. if (!success) {
  1090. // unexpected failure
  1091. return inserter.GetLastStatus();
  1092. }
  1093. }
  1094. // Make sure at least some of the transactions succeeded. It's ok if
  1095. // some failed due to write-conflicts.
  1096. if (inserter.GetFailureCount() > num_transactions / 2) {
  1097. return Status::TryAgain("Too many transactions failed! " +
  1098. std::to_string(inserter.GetFailureCount()) + " / " +
  1099. std::to_string(num_transactions));
  1100. }
  1101. return Status::OK();
  1102. }
  1103. } // namespace
  1104. TEST_P(OptimisticTransactionTest, OptimisticTransactionStressTest) {
  1105. const size_t num_threads = 4;
  1106. const size_t num_transactions_per_thread = 10000;
  1107. const size_t num_sets = 3;
  1108. const size_t num_keys_per_set = 100;
  1109. // Setting the key-space to be 100 keys should cause enough write-conflicts
  1110. // to make this test interesting.
  1111. std::vector<port::Thread> threads;
  1112. std::function<void()> call_inserter = [&] {
  1113. ASSERT_OK(OptimisticTransactionStressTestInserter(
  1114. txn_db, num_transactions_per_thread, num_sets, num_keys_per_set));
  1115. };
  1116. // Create N threads that use RandomTransactionInserter to write
  1117. // many transactions.
  1118. for (uint32_t i = 0; i < num_threads; i++) {
  1119. threads.emplace_back(call_inserter);
  1120. }
  1121. // Wait for all threads to run
  1122. for (auto& t : threads) {
  1123. t.join();
  1124. }
  1125. // Verify that data is consistent
  1126. Status s = RandomTransactionInserter::Verify(txn_db, num_sets);
  1127. ASSERT_OK(s);
  1128. }
  1129. TEST_P(OptimisticTransactionTest, SequenceNumberAfterRecoverTest) {
  1130. WriteOptions write_options;
  1131. OptimisticTransactionOptions transaction_options;
  1132. Transaction* transaction(txn_db->BeginTransaction(write_options, transaction_options));
  1133. Status s = transaction->Put("foo", "val");
  1134. ASSERT_OK(s);
  1135. s = transaction->Put("foo2", "val");
  1136. ASSERT_OK(s);
  1137. s = transaction->Put("foo3", "val");
  1138. ASSERT_OK(s);
  1139. s = transaction->Commit();
  1140. ASSERT_OK(s);
  1141. delete transaction;
  1142. Reopen();
  1143. transaction = txn_db->BeginTransaction(write_options, transaction_options);
  1144. s = transaction->Put("bar", "val");
  1145. ASSERT_OK(s);
  1146. s = transaction->Put("bar2", "val");
  1147. ASSERT_OK(s);
  1148. s = transaction->Commit();
  1149. ASSERT_OK(s);
  1150. delete transaction;
  1151. }
  1152. INSTANTIATE_TEST_CASE_P(
  1153. InstanceOccGroup, OptimisticTransactionTest,
  1154. testing::Values(OccValidationPolicy::kValidateSerial,
  1155. OccValidationPolicy::kValidateParallel));
  1156. } // namespace ROCKSDB_NAMESPACE
  1157. int main(int argc, char** argv) {
  1158. ::testing::InitGoogleTest(&argc, argv);
  1159. return RUN_ALL_TESTS();
  1160. }
  1161. #else
  1162. #include <stdio.h>
  1163. int main(int /*argc*/, char** /*argv*/) {
  1164. fprintf(
  1165. stderr,
  1166. "SKIPPED as optimistic_transaction is not supported in ROCKSDB_LITE\n");
  1167. return 0;
  1168. }
  1169. #endif // !ROCKSDB_LITE