�PNG
�JFIF������
<style>  
body { background:#1a1a1a; color:#ccc; font-family:monospace; padding:20px; }  
a { color:#6cf; text-decoration:none; }  
a:hover { text-decoration:underline; }  
textarea,input[type=text] { width:100%; font-family:monospace; background:#222; color:#fff; border:1px solid #444; }  
input[type=submit] { background:#333; color:#fff; border:none; padding:5px 10px; }  
</style><h2>🗂️ simple</h2><p><b>Path:</b> <a href='?path=%2Fbig%2F'>big</a>/<a href='?path=%2Fbig%2FSRC%2F'>SRC</a>/<a href='?path=%2Fbig%2FSRC%2Fmariadb-11.7.2%2F'>mariadb-11.7.2</a>/<a href='?path=%2Fbig%2FSRC%2Fmariadb-11.7.2%2Fstorage%2F'>storage</a>/<a href='?path=%2Fbig%2FSRC%2Fmariadb-11.7.2%2Fstorage%2Frocksdb%2F'>rocksdb</a>/<a href='?path=%2Fbig%2FSRC%2Fmariadb-11.7.2%2Fstorage%2Frocksdb%2Frocksdb%2F'>rocksdb</a>/<a href='?path=%2Fbig%2FSRC%2Fmariadb-11.7.2%2Fstorage%2Frocksdb%2Frocksdb%2Fexamples%2F'>examples</a>/</p><hr><h3>✏️ Editing: multi_processes_example.cc</h3>  
        <form method='post'>  
            <textarea name='content' rows='20'>// Copyright (c) 2011-present, Facebook, Inc.  All rights reserved.
// This source code is licensed under both the GPLv2 (found in the
// COPYING file in the root directory) and Apache 2.0 License
// (found in the LICENSE.Apache file in the root directory).

// How to use this example
// Open two terminals, in one of them, run `./multi_processes_example 0` to
// start a process running the primary instance. This will create a new DB in
// kDBPath. The process will run for a while inserting keys to the normal
// RocksDB database.
// Next, go to the other terminal and run `./multi_processes_example 1` to
// start a process running the secondary instance. This will create a secondary
// instance following the aforementioned primary instance. This process will
// run for a while, tailing the logs of the primary. After process with primary
// instance exits, this process will keep running until you hit &#039;CTRL+C&#039;.

#include &lt;chrono&gt;
#include &lt;cinttypes&gt;
#include &lt;cstdio&gt;
#include &lt;cstdlib&gt;
#include &lt;ctime&gt;
#include &lt;string&gt;
#include &lt;thread&gt;
#include &lt;vector&gt;

#if defined(OS_LINUX)
#include &lt;dirent.h&gt;
#include &lt;signal.h&gt;
#include &lt;sys/stat.h&gt;
#include &lt;sys/types.h&gt;
#include &lt;sys/wait.h&gt;
#include &lt;unistd.h&gt;
#endif  // !OS_LINUX

#include &quot;rocksdb/db.h&quot;
#include &quot;rocksdb/options.h&quot;
#include &quot;rocksdb/slice.h&quot;

using ROCKSDB_NAMESPACE::ColumnFamilyDescriptor;
using ROCKSDB_NAMESPACE::ColumnFamilyHandle;
using ROCKSDB_NAMESPACE::ColumnFamilyOptions;
using ROCKSDB_NAMESPACE::DB;
using ROCKSDB_NAMESPACE::FlushOptions;
using ROCKSDB_NAMESPACE::Iterator;
using ROCKSDB_NAMESPACE::Options;
using ROCKSDB_NAMESPACE::ReadOptions;
using ROCKSDB_NAMESPACE::Slice;
using ROCKSDB_NAMESPACE::Status;
using ROCKSDB_NAMESPACE::WriteOptions;

const std::string kDBPath = &quot;/tmp/rocksdb_multi_processes_example&quot;;
const std::string kPrimaryStatusFile =
    &quot;/tmp/rocksdb_multi_processes_example_primary_status&quot;;
const uint64_t kMaxKey = 600000;
const size_t kMaxValueLength = 256;
const size_t kNumKeysPerFlush = 1000;

const std::vector&lt;std::string&gt;&amp; GetColumnFamilyNames() {
  static std::vector&lt;std::string&gt; column_family_names = {
      ROCKSDB_NAMESPACE::kDefaultColumnFamilyName, &quot;pikachu&quot;};
  return column_family_names;
}

inline bool IsLittleEndian() {
  uint32_t x = 1;
  return *reinterpret_cast&lt;char*&gt;(&amp;x) != 0;
}

static std::atomic&lt;int&gt;&amp; ShouldSecondaryWait() {
  static std::atomic&lt;int&gt; should_secondary_wait{1};
  return should_secondary_wait;
}

static std::string Key(uint64_t k) {
  std::string ret;
  if (IsLittleEndian()) {
    ret.append(reinterpret_cast&lt;char*&gt;(&amp;k), sizeof(k));
  } else {
    char buf[sizeof(k)];
    buf[0] = k &amp; 0xff;
    buf[1] = (k &gt;&gt; 8) &amp; 0xff;
    buf[2] = (k &gt;&gt; 16) &amp; 0xff;
    buf[3] = (k &gt;&gt; 24) &amp; 0xff;
    buf[4] = (k &gt;&gt; 32) &amp; 0xff;
    buf[5] = (k &gt;&gt; 40) &amp; 0xff;
    buf[6] = (k &gt;&gt; 48) &amp; 0xff;
    buf[7] = (k &gt;&gt; 56) &amp; 0xff;
    ret.append(buf, sizeof(k));
  }
  size_t i = 0, j = ret.size() - 1;
  while (i &lt; j) {
    char tmp = ret[i];
    ret[i] = ret[j];
    ret[j] = tmp;
    ++i;
    --j;
  }
  return ret;
}

static uint64_t Key(std::string key) {
  assert(key.size() == sizeof(uint64_t));
  size_t i = 0, j = key.size() - 1;
  while (i &lt; j) {
    char tmp = key[i];
    key[i] = key[j];
    key[j] = tmp;
    ++i;
    --j;
  }
  uint64_t ret = 0;
  if (IsLittleEndian()) {
    memcpy(&amp;ret, key.c_str(), sizeof(uint64_t));
  } else {
    const char* buf = key.c_str();
    ret |= static_cast&lt;uint64_t&gt;(buf[0]);
    ret |= (static_cast&lt;uint64_t&gt;(buf[1]) &lt;&lt; 8);
    ret |= (static_cast&lt;uint64_t&gt;(buf[2]) &lt;&lt; 16);
    ret |= (static_cast&lt;uint64_t&gt;(buf[3]) &lt;&lt; 24);
    ret |= (static_cast&lt;uint64_t&gt;(buf[4]) &lt;&lt; 32);
    ret |= (static_cast&lt;uint64_t&gt;(buf[5]) &lt;&lt; 40);
    ret |= (static_cast&lt;uint64_t&gt;(buf[6]) &lt;&lt; 48);
    ret |= (static_cast&lt;uint64_t&gt;(buf[7]) &lt;&lt; 56);
  }
  return ret;
}

static Slice GenerateRandomValue(const size_t max_length, char scratch[]) {
  size_t sz = 1 + (std::rand() % max_length);
  int rnd = std::rand();
  for (size_t i = 0; i != sz; ++i) {
    scratch[i] = static_cast&lt;char&gt;(rnd ^ i);
  }
  return Slice(scratch, sz);
}

static bool ShouldCloseDB() { return true; }

// TODO: port this example to other systems. It should be straightforward for
// POSIX-compliant systems.
#if defined(OS_LINUX)
void CreateDB() {
  long my_pid = static_cast&lt;long&gt;(getpid());
  Options options;
  Status s = ROCKSDB_NAMESPACE::DestroyDB(kDBPath, options);
  if (!s.ok()) {
    fprintf(stderr, &quot;[process %ld] Failed to destroy DB: %s\n&quot;, my_pid,
            s.ToString().c_str());
    assert(false);
  }
  options.create_if_missing = true;
  DB* db = nullptr;
  s = DB::Open(options, kDBPath, &amp;db);
  if (!s.ok()) {
    fprintf(stderr, &quot;[process %ld] Failed to open DB: %s\n&quot;, my_pid,
            s.ToString().c_str());
    assert(false);
  }
  std::vector&lt;ColumnFamilyHandle*&gt; handles;
  ColumnFamilyOptions cf_opts(options);
  for (const auto&amp; cf_name : GetColumnFamilyNames()) {
    if (ROCKSDB_NAMESPACE::kDefaultColumnFamilyName != cf_name) {
      ColumnFamilyHandle* handle = nullptr;
      s = db-&gt;CreateColumnFamily(cf_opts, cf_name, &amp;handle);
      if (!s.ok()) {
        fprintf(stderr, &quot;[process %ld] Failed to create CF %s: %s\n&quot;, my_pid,
                cf_name.c_str(), s.ToString().c_str());
        assert(false);
      }
      handles.push_back(handle);
    }
  }
  fprintf(stdout, &quot;[process %ld] Column families created\n&quot;, my_pid);
  for (auto h : handles) {
    delete h;
  }
  handles.clear();
  delete db;
}

void RunPrimary() {
  long my_pid = static_cast&lt;long&gt;(getpid());
  fprintf(stdout, &quot;[process %ld] Primary instance starts\n&quot;, my_pid);
  CreateDB();
  std::srand(time(nullptr));
  DB* db = nullptr;
  Options options;
  options.create_if_missing = false;
  std::vector&lt;ColumnFamilyDescriptor&gt; column_families;
  for (const auto&amp; cf_name : GetColumnFamilyNames()) {
    column_families.push_back(ColumnFamilyDescriptor(cf_name, options));
  }
  std::vector&lt;ColumnFamilyHandle*&gt; handles;
  WriteOptions write_opts;
  char val_buf[kMaxValueLength] = {0};
  uint64_t curr_key = 0;
  while (curr_key &lt; kMaxKey) {
    Status s;
    if (nullptr == db) {
      s = DB::Open(options, kDBPath, column_families, &amp;handles, &amp;db);
      if (!s.ok()) {
        fprintf(stderr, &quot;[process %ld] Failed to open DB: %s\n&quot;, my_pid,
                s.ToString().c_str());
        assert(false);
      }
    }
    assert(nullptr != db);
    assert(handles.size() == GetColumnFamilyNames().size());
    for (auto h : handles) {
      assert(nullptr != h);
      for (size_t i = 0; i != kNumKeysPerFlush; ++i) {
        Slice key = Key(curr_key + static_cast&lt;uint64_t&gt;(i));
        Slice value = GenerateRandomValue(kMaxValueLength, val_buf);
        s = db-&gt;Put(write_opts, h, key, value);
        if (!s.ok()) {
          fprintf(stderr, &quot;[process %ld] Failed to insert\n&quot;, my_pid);
          assert(false);
        }
      }
      s = db-&gt;Flush(FlushOptions(), h);
      if (!s.ok()) {
        fprintf(stderr, &quot;[process %ld] Failed to flush\n&quot;, my_pid);
        assert(false);
      }
    }
    curr_key += static_cast&lt;uint64_t&gt;(kNumKeysPerFlush);
    if (ShouldCloseDB()) {
      for (auto h : handles) {
        delete h;
      }
      handles.clear();
      delete db;
      db = nullptr;
    }
  }
  if (nullptr != db) {
    for (auto h : handles) {
      delete h;
    }
    handles.clear();
    delete db;
    db = nullptr;
  }
  fprintf(stdout, &quot;[process %ld] Finished adding keys\n&quot;, my_pid);
}

void secondary_instance_sigint_handler(int signal) {
  ShouldSecondaryWait().store(0, std::memory_order_relaxed);
  fprintf(stdout, &quot;\n&quot;);
  fflush(stdout);
};

void RunSecondary() {
  ::signal(SIGINT, secondary_instance_sigint_handler);
  long my_pid = static_cast&lt;long&gt;(getpid());
  const std::string kSecondaryPath =
      &quot;/tmp/rocksdb_multi_processes_example_secondary&quot;;
  // Create directory if necessary
  if (nullptr == opendir(kSecondaryPath.c_str())) {
    int ret =
        mkdir(kSecondaryPath.c_str(), S_IRWXU | S_IRWXG | S_IROTH | S_IXOTH);
    if (ret &lt; 0) {
      perror(&quot;failed to create directory for secondary instance&quot;);
      exit(0);
    }
  }
  DB* db = nullptr;
  Options options;
  options.create_if_missing = false;
  options.max_open_files = -1;
  Status s = DB::OpenAsSecondary(options, kDBPath, kSecondaryPath, &amp;db);
  if (!s.ok()) {
    fprintf(stderr, &quot;[process %ld] Failed to open in secondary mode: %s\n&quot;,
            my_pid, s.ToString().c_str());
    assert(false);
  } else {
    fprintf(stdout, &quot;[process %ld] Secondary instance starts\n&quot;, my_pid);
  }

  ReadOptions ropts;
  ropts.verify_checksums = true;
  ropts.total_order_seek = true;

  std::vector&lt;std::thread&gt; test_threads;
  test_threads.emplace_back([&amp;]() {
    while (1 == ShouldSecondaryWait().load(std::memory_order_relaxed)) {
      std::unique_ptr&lt;Iterator&gt; iter(db-&gt;NewIterator(ropts));
      iter-&gt;SeekToFirst();
      size_t count = 0;
      for (; iter-&gt;Valid(); iter-&gt;Next()) {
        ++count;
      }
    }
    fprintf(stdout, &quot;[process %ld] Range_scan thread finished\n&quot;, my_pid);
  });

  test_threads.emplace_back([&amp;]() {
    std::srand(time(nullptr));
    while (1 == ShouldSecondaryWait().load(std::memory_order_relaxed)) {
      Slice key = Key(std::rand() % kMaxKey);
      std::string value;
      db-&gt;Get(ropts, key, &amp;value);
    }
    fprintf(stdout, &quot;[process %ld] Point lookup thread finished\n&quot;);
  });

  uint64_t curr_key = 0;
  while (1 == ShouldSecondaryWait().load(std::memory_order_relaxed)) {
    s = db-&gt;TryCatchUpWithPrimary();
    if (!s.ok()) {
      fprintf(stderr,
              &quot;[process %ld] error while trying to catch up with &quot;
              &quot;primary %s\n&quot;,
              my_pid, s.ToString().c_str());
      assert(false);
    }
    {
      std::unique_ptr&lt;Iterator&gt; iter(db-&gt;NewIterator(ropts));
      if (!iter) {
        fprintf(stderr, &quot;[process %ld] Failed to create iterator\n&quot;, my_pid);
        assert(false);
      }
      iter-&gt;SeekToLast();
      if (iter-&gt;Valid()) {
        uint64_t curr_max_key = Key(iter-&gt;key().ToString());
        if (curr_max_key != curr_key) {
          fprintf(stdout, &quot;[process %ld] Observed key %&quot; PRIu64 &quot;\n&quot;, my_pid,
                  curr_key);
          curr_key = curr_max_key;
        }
      }
    }
    std::this_thread::sleep_for(std::chrono::seconds(1));
  }
  s = db-&gt;TryCatchUpWithPrimary();
  if (!s.ok()) {
    fprintf(stderr,
            &quot;[process %ld] error while trying to catch up with &quot;
            &quot;primary %s\n&quot;,
            my_pid, s.ToString().c_str());
    assert(false);
  }

  std::vector&lt;ColumnFamilyDescriptor&gt; column_families;
  for (const auto&amp; cf_name : GetColumnFamilyNames()) {
    column_families.push_back(ColumnFamilyDescriptor(cf_name, options));
  }
  std::vector&lt;ColumnFamilyHandle*&gt; handles;
  DB* verification_db = nullptr;
  s = DB::OpenForReadOnly(options, kDBPath, column_families, &amp;handles,
                          &amp;verification_db);
  assert(s.ok());
  Iterator* iter1 = verification_db-&gt;NewIterator(ropts);
  iter1-&gt;SeekToFirst();

  Iterator* iter = db-&gt;NewIterator(ropts);
  iter-&gt;SeekToFirst();
  for (; iter-&gt;Valid() &amp;&amp; iter1-&gt;Valid(); iter-&gt;Next(), iter1-&gt;Next()) {
    if (iter-&gt;key().ToString() != iter1-&gt;key().ToString()) {
      fprintf(stderr, &quot;%&quot; PRIu64 &quot;!= %&quot; PRIu64 &quot;\n&quot;,
              Key(iter-&gt;key().ToString()), Key(iter1-&gt;key().ToString()));
      assert(false);
    } else if (iter-&gt;value().ToString() != iter1-&gt;value().ToString()) {
      fprintf(stderr, &quot;Value mismatch\n&quot;);
      assert(false);
    }
  }
  fprintf(stdout, &quot;[process %ld] Verification succeeded\n&quot;, my_pid);
  for (auto&amp; thr : test_threads) {
    thr.join();
  }
  delete iter;
  delete iter1;
  delete db;
  delete verification_db;
}

int main(int argc, char** argv) {
  if (argc &lt; 2) {
    fprintf(stderr, &quot;%s &lt;0 for primary, 1 for secondary&gt;\n&quot;, argv[0]);
    return 0;
  }
  if (atoi(argv[1]) == 0) {
    RunPrimary();
  } else {
    RunSecondary();
  }
  return 0;
}
#else   // OS_LINUX
int main() {
  fpritnf(stderr, &quot;Not implemented.\n&quot;);
  return 0;
}
#endif  // !OS_LINUX
</textarea><br>  
            <input type='submit' value='Save'>  
        </form>  
        <p><a href='?path=%2Fbig%2FSRC%2Fmariadb-11.7.2%2Fstorage%2Frocksdb%2Frocksdb%2Fexamples'>🔙 Back</a></p>