New config option "initial_repl_sleep_delay_usec"

cybozu/config_parser.hpp

@@ -7,6 +7,7 @@
 #include <unordered_map>
 #include <string>
 #include <stdexcept>
+#include <cstdint>
 
 namespace cybozu {
 
@@ -103,6 +104,23 @@ class config_parser {
         }
     }
 
+    // Get an uint64_t integer converted from the value associated with `key`.
+    // @key   A configuration key.
+    //
+    // Get an uint64_t integer converted from the value associated with `key`.
+    // Raise <not_found> or <illegal_value>.
+    //
+    // @return An uint64_t integer converted from the associated value.
+    std::uint64_t get_as_uint64(const std::string& key) const {
+        try {
+            return std::stoull(get(key));
+        } catch(const std::invalid_argument& e) {
+            throw illegal_value(key);
+        } catch(const std::out_of_range& e) {
+            throw illegal_value(key);
+        }
+    }
+
     // Get a boolean converted from the value associated with `key`.
     // @key   A configuration key.
     //

docs/usage.md

@@ -51,6 +51,8 @@ These options are to configure memcache protocol:
     Objects larger than this will be stored in temporary files.
 * `repl_buffer_size` (Default: 30)  
     The replication buffer size.  Unit is MiB.
+* `initial_repl_sleep_delay` (Default: 0)  
+    Slow down the scan of the entire hash by the GC thread to prevent errors with the message "Replication buffer is full." during the initial replication. The GC thread sleeps for the time specified here for each scan of the hash bucket. Unit is microseconds.
 * `secure_erase` (Default: false)  
     If `true`, object memory will be cleared as soon as the object is removed.
 * `lock_memory` (Default: false)  

etc/yrmcds.conf

@@ -42,6 +42,12 @@ heap_data_limit = 256K
 # The value must be an integer > 0.  Default is 30 (MiB).
 repl_buffer_size = 30
 
+# Slow down the scan of the entire hash by the GC thread to prevent
+# errors with the message "Replication buffer is full." during the initial
+# replication. The GC thread sleeps for the time specified here for each
+# scan of the hash bucket. Unit is microseconds.
+initial_repl_sleep_delay = 0
+
 # Clear memory used by deleted or expired objects securely.
 # This ensures confidential data such as crypto keys will not be
 # leaked after expiration.

src/config.cpp

@@ -24,6 +24,7 @@ const char MAX_DATA_SIZE[] = "max_data_size";
 const char HEAP_DATA_LIMIT[] = "heap_data_limit";
 const char MEMORY_LIMIT[] = "memory_limit";
 const char REPL_BUFSIZE[] = "repl_buffer_size";
+const char INITIAL_REPL_SLEEP_DELAY[] = "initial_repl_sleep_delay";
 const char SECURE_ERASE[] = "secure_erase";
 const char LOCK_MEMORY[] = "lock_memory";
 const char WORKERS[] = "workers";
@@ -206,6 +207,11 @@ void config::load(const std::string& path) {
         m_repl_bufsize = bufs;
     }
 
+    if( cp.exists(INITIAL_REPL_SLEEP_DELAY) ) {
+        std::uint64_t n = cp.get_as_uint64(INITIAL_REPL_SLEEP_DELAY);
+        m_initial_repl_sleep_delay = n;
+    }
+
     if( cp.exists(SECURE_ERASE) ) {
         m_secure_erase = cp.get_as_bool(SECURE_ERASE);
     }

src/config.hpp

@@ -111,6 +111,9 @@ class config {
     unsigned int repl_bufsize() const noexcept {
         return m_repl_bufsize;
     }
+    std::uint64_t initial_repl_sleep_delay() const noexcept {
+        return m_initial_repl_sleep_delay;
+    }
     bool secure_erase() const noexcept {
         return m_secure_erase;
     }
@@ -152,6 +155,7 @@ class config {
     std::size_t m_heap_data_limit = DEFAULT_HEAP_DATA_LIMIT;
     std::size_t m_memory_limit = DEFAULT_MEMORY_LIMIT;
     unsigned int m_repl_bufsize = DEFAULT_REPL_BUFSIZE;
+    uint64_t m_initial_repl_sleep_delay = DEFAULT_INITIAL_REPL_SLEEP_DELAY;
     bool m_secure_erase = false;
     bool m_lock_memory = false;
     unsigned int m_workers = DEFAULT_WORKER_THREADS;

src/constants.hpp

@@ -17,6 +17,7 @@ const std::size_t   DEFAULT_MAX_DATA_SIZE  = static_cast<std::size_t>(1) << 20;
 const std::size_t   DEFAULT_HEAP_DATA_LIMIT= 256 << 10;
 const std::size_t   DEFAULT_MEMORY_LIMIT   = static_cast<std::size_t>(1) << 30;
 const unsigned int  DEFAULT_REPL_BUFSIZE   = 30;
+const std::uint64_t DEFAULT_INITIAL_REPL_SLEEP_DELAY = 0;
 const int           DEFAULT_WORKER_THREADS = 8;
 const unsigned int  DEFAULT_GC_INTERVAL    = 10;
 const unsigned int  DEFAULT_SLAVE_TIMEOUT  = 10;

src/memcache/gc.cpp

@@ -122,10 +122,25 @@ void gc_thread::gc() {
         return false;
     };
 
+    // Putting the thread to sleep tens of microseconds with each loop is desired, but due to the precision of the timer,
+    // it's not appropriate to do sleep with each loop. The `initial_repl_sleep_delay` is accumulated until it
+    // exceeds 10000 microseconds (10 milliseconds), and then the thread is put to sleep all at once when this limit is exceeded.
+    static const std::uint64_t SLEEP_THRESHOLD = 10000;
+    std::uint64_t sleep_sum = 0;
+
     for( auto it = m_hash.begin(); it != m_hash.end(); ++it ) {
         m_objects_in_bucket = 0;
         it->gc(pred);
         m_flushers.clear();
+
+        if( ! m_new_slaves.empty() ) {
+            sleep_sum += g_config.initial_repl_sleep_delay();
+            if( sleep_sum >= SLEEP_THRESHOLD ) {
+                std::this_thread::sleep_for(
+                    std::chrono::microseconds(sleep_sum));
+                sleep_sum = 0;
+            }
+        }
     }
 
     if( flush )

test/config.cpp

@@ -15,6 +15,7 @@ AUTOTEST(config) {
     cybozu_assert(g_config.group() == "nogroup");
     cybozu_assert(g_config.memory_limit() == (1024 << 20));
     cybozu_assert(g_config.repl_bufsize() == 100);
+    cybozu_assert(g_config.initial_repl_sleep_delay() == 40);
     cybozu_assert(g_config.secure_erase() == true);
     cybozu_assert(g_config.lock_memory() == true);
     cybozu_assert(g_config.threshold() == cybozu::severity::warning);

test/test.conf

@@ -15,6 +15,7 @@ max_data_size	= 5M
 heap_data_limit	= 16K
 memory_limit	= 1024M
 repl_buffer_size= 100
+initial_repl_sleep_delay = 40
 secure_erase	= true
 lock_memory	= true
 workers		= 10