Inside Bitcoin

1751    fprintf(stderr, "iter %d; errors %d\n", i, errors);
1752    if (!Put("foo", big).ok()) {
1753      errors++;
1754      DelayMilliseconds(100);
1755    }
1756  }
1757  ASSERT_GT(errors, 0);
1758  env_->non_writable_.store(false, std::memory_order_release);
1759}
1760
1761TEST(DBTest, WriteSyncError) {
1762  // Check that log sync errors cause the DB to disallow future writes.
1763
1764  // (a) Cause log sync calls to fail
1765  Options options = CurrentOptions();
1766  options.env = env_;
1767  Reopen(&options);
1768  env_->data_sync_error_.store(true, std::memory_order_release);
1769
1770  // (b) Normal write should succeed
1771  WriteOptions w;
1772  ASSERT_OK(db_->Put(w, "k1", "v1"));
1773  ASSERT_EQ("v1", Get("k1"));
1774
1775  // (c) Do a sync write; should fail
1776  w.sync = true;
1777  ASSERT_TRUE(!db_->Put(w, "k2", "v2").ok());
1778  ASSERT_EQ("v1", Get("k1"));
1779  ASSERT_EQ("NOT_FOUND", Get("k2"));
1780
1781  // (d) make sync behave normally
1782  env_->data_sync_error_.store(false, std::memory_order_release);
1783
1784  // (e) Do a non-sync write; should fail
1785  w.sync = false;
1786  ASSERT_TRUE(!db_->Put(w, "k3", "v3").ok());
1787  ASSERT_EQ("v1", Get("k1"));
1788  ASSERT_EQ("NOT_FOUND", Get("k2"));
1789  ASSERT_EQ("NOT_FOUND", Get("k3"));
1790}
1791
1792TEST(DBTest, ManifestWriteError) {
1793  // Test for the following problem:
1794  // (a) Compaction produces file F
1795  // (b) Log record containing F is written to MANIFEST file, but Sync() fails
1796  // (c) GC deletes F
1797  // (d) After reopening DB, reads fail since deleted F is named in log record
1798
1799  // We iterate twice.  In the second iteration, everything is the
1800  // same except the log record never makes it to the MANIFEST file.
1801  for (int iter = 0; iter < 2; iter++) {
1802    std::atomic<bool>* error_type = (iter == 0) ? &env_->manifest_sync_error_
1803                                                : &env_->manifest_write_error_;
1804
1805    // Insert foo=>bar mapping
1806    Options options = CurrentOptions();
1807    options.env = env_;
1808    options.create_if_missing = true;
1809    options.error_if_exists = false;
1810    DestroyAndReopen(&options);
1811    ASSERT_OK(Put("foo", "bar"));
1812    ASSERT_EQ("bar", Get("foo"));
1813
1814    // Memtable compaction (will succeed)
1815    dbfull()->TEST_CompactMemTable();
1816    ASSERT_EQ("bar", Get("foo"));
1817    const int last = config::kMaxMemCompactLevel;
1818    ASSERT_EQ(NumTableFilesAtLevel(last), 1);  // foo=>bar is now in last level
1819
1820    // Merging compaction (will fail)
1821    error_type->store(true, std::memory_order_release);
1822    dbfull()->TEST_CompactRange(last, nullptr, nullptr);  // Should fail
1823    ASSERT_EQ("bar", Get("foo"));
1824
1825    // Recovery: should not lose data
1826    error_type->store(false, std::memory_order_release);
1827    Reopen(&options);
1828    ASSERT_EQ("bar", Get("foo"));
1829  }
1830}
1831
1832TEST(DBTest, MissingSSTFile) {
1833  ASSERT_OK(Put("foo", "bar"));
1834  ASSERT_EQ("bar", Get("foo"));
1835
1836  // Dump the memtable to disk.
1837  dbfull()->TEST_CompactMemTable();
1838  ASSERT_EQ("bar", Get("foo"));
1839
1840  Close();
1841  ASSERT_TRUE(DeleteAnSSTFile());
1842  Options options = CurrentOptions();
1843  options.paranoid_checks = true;
1844  Status s = TryReopen(&options);
1845  ASSERT_TRUE(!s.ok());
1846  ASSERT_TRUE(s.ToString().find("issing") != std::string::npos) << s.ToString();
1847}
1848
1849TEST(DBTest, StillReadSST) {
1850  ASSERT_OK(Put("foo", "bar"));
1851  ASSERT_EQ("bar", Get("foo"));
1852
1853  // Dump the memtable to disk.
1854  dbfull()->TEST_CompactMemTable();
1855  ASSERT_EQ("bar", Get("foo"));
1856  Close();
1857  ASSERT_GT(RenameLDBToSST(), 0);
1858  Options options = CurrentOptions();
1859  options.paranoid_checks = true;
1860  Status s = TryReopen(&options);
1861  ASSERT_TRUE(s.ok());
1862  ASSERT_EQ("bar", Get("foo"));
1863}
1864
1865TEST(DBTest, FilesDeletedAfterCompaction) {
1866  ASSERT_OK(Put("foo", "v2"));
1867  Compact("a", "z");
1868  const int num_files = CountFiles();
1869  for (int i = 0; i < 10; i++) {
1870    ASSERT_OK(Put("foo", "v2"));
1871    Compact("a", "z");
1872  }
1873  ASSERT_EQ(CountFiles(), num_files);
1874}
1875
1876TEST(DBTest, BloomFilter) {
1877  env_->count_random_reads_ = true;
1878  Options options = CurrentOptions();
1879  options.env = env_;
1880  options.block_cache = NewLRUCache(0);  // Prevent cache hits
1881  options.filter_policy = NewBloomFilterPolicy(10);
1882  Reopen(&options);
1883
1884  // Populate multiple layers
1885  const int N = 10000;
1886  for (int i = 0; i < N; i++) {
1887    ASSERT_OK(Put(Key(i), Key(i)));
1888  }
1889  Compact("a", "z");
1890  for (int i = 0; i < N; i += 100) {
1891    ASSERT_OK(Put(Key(i), Key(i)));
1892  }
1893  dbfull()->TEST_CompactMemTable();
1894
1895  // Prevent auto compactions triggered by seeks
1896  env_->delay_data_sync_.store(true, std::memory_order_release);
1897
1898  // Lookup present keys.  Should rarely read from small sstable.
1899  env_->random_read_counter_.Reset();
1900  for (int i = 0; i < N; i++) {
1901    ASSERT_EQ(Key(i), Get(Key(i)));
1902  }
1903  int reads = env_->random_read_counter_.Read();
1904  fprintf(stderr, "%d present => %d reads\n", N, reads);
1905  ASSERT_GE(reads, N);
1906  ASSERT_LE(reads, N + 2 * N / 100);
1907
1908  // Lookup present keys.  Should rarely read from either sstable.
1909  env_->random_read_counter_.Reset();
1910  for (int i = 0; i < N; i++) {
1911    ASSERT_EQ("NOT_FOUND", Get(Key(i) + ".missing"));
1912  }
1913  reads = env_->random_read_counter_.Read();
1914  fprintf(stderr, "%d missing => %d reads\n", N, reads);
1915  ASSERT_LE(reads, 3 * N / 100);
1916
1917  env_->delay_data_sync_.store(false, std::memory_order_release);
1918  Close();
1919  delete options.block_cache;
1920  delete options.filter_policy;
1921}
1922
1923// Multi-threaded test:
1924namespace {
1925
1926static const int kNumThreads = 4;
1927static const int kTestSeconds = 10;
1928static const int kNumKeys = 1000;
1929
1930struct MTState {
1931  DBTest* test;
1932  std::atomic<bool> stop;
1933  std::atomic<int> counter[kNumThreads];
1934  std::atomic<bool> thread_done[kNumThreads];
1935};
1936
1937struct MTThread {
1938  MTState* state;
1939  int id;
1940};
1941
1942static void MTThreadBody(void* arg) {
1943  MTThread* t = reinterpret_cast<MTThread*>(arg);
1944  int id = t->id;
1945  DB* db = t->state->test->db_;
1946  int counter = 0;
1947  fprintf(stderr, "... starting thread %d\n", id);
1948  Random rnd(1000 + id);
1949  std::string value;
1950  char valbuf[1500];
1951  while (!t->state->stop.load(std::memory_order_acquire)) {
1952    t->state->counter[id].store(counter, std::memory_order_release);
1953
1954    int key = rnd.Uniform(kNumKeys);
1955    char keybuf[20];
1956    snprintf(keybuf, sizeof(keybuf), "%016d", key);
1957
1958    if (rnd.OneIn(2)) {
1959      // Write values of the form <key, my id, counter>.
1960      // We add some padding for force compactions.
1961      snprintf(valbuf, sizeof(valbuf), "%d.%d.%-1000d", key, id,
1962               static_cast<int>(counter));
1963      ASSERT_OK(db->Put(WriteOptions(), Slice(keybuf), Slice(valbuf)));
1964    } else {
1965      // Read a value and verify that it matches the pattern written above.
1966      Status s = db->Get(ReadOptions(), Slice(keybuf), &value);
1967      if (s.IsNotFound()) {
1968        // Key has not yet been written
1969      } else {
1970        // Check that the writer thread counter is >= the counter in the value
1971        ASSERT_OK(s);
1972        int k, w, c;
1973        ASSERT_EQ(3, sscanf(value.c_str(), "%d.%d.%d", &k, &w, &c)) << value;
1974        ASSERT_EQ(k, key);
1975        ASSERT_GE(w, 0);
1976        ASSERT_LT(w, kNumThreads);
1977        ASSERT_LE(c, t->state->counter[w].load(std::memory_order_acquire));
1978      }
1979    }
1980    counter++;
1981  }
1982  t->state->thread_done[id].store(true, std::memory_order_release);
1983  fprintf(stderr, "... stopping thread %d after %d ops\n", id, counter);
1984}
1985
1986}  // namespace
1987
1988TEST(DBTest, MultiThreaded) {
1989  do {
1990    // Initialize state
1991    MTState mt;
1992    mt.test = this;
1993    mt.stop.store(false, std::memory_order_release);
1994    for (int id = 0; id < kNumThreads; id++) {
1995      mt.counter[id].store(false, std::memory_order_release);
1996      mt.thread_done[id].store(false, std::memory_order_release);
1997    }
1998
1999    // Start threads
2000    MTThread thread[kNumThreads];