Fix insertion rate limit

Author: miguelportilla

CHANGELOG.md

@@ -1,3 +1,16 @@
+2.0.1
+
+* Fixed a bug introduced in 2.0 where flush time was incorrectly updated 
+* Improved thread management in context class
+
+---
+
+2.0.0
+
+* Add thread management context class
+
+---
+
 1.0.2
 
 * Better CMakeLists.txt for finding Boost

include/nudb/impl/basic_store.ipp

@@ -344,6 +344,10 @@ insert(
     auto const sleep =
         s_->rate && rate > s_->rate;
     m.unlock();
+
+    // The caller of insert must be blocked when the rate of insertion
+    // (measured in approximate bytes per second) exceeds the maximum rate
+    // that can be flushed. The precise sleep duration is not important.
     if(sleep)
         std::this_thread::sleep_for(milliseconds{25});
 }
@@ -750,6 +754,7 @@ flush()
 #endif
     {
         unique_lock_type m{m_};
+        s_->when = clock_type::now();
         if(! s_->p1.empty())
         {
             std::size_t work;
@@ -774,7 +779,6 @@ flush()
         #endif
         }
         s_->p1.periodic_activity();
-        s_->when = clock_type::now();
     }
     if(ec_)
         ecb_.store(true);

include/nudb/impl/context.ipp

@@ -73,11 +73,10 @@ run()
                 break;
             if(first)
             {
-                auto const status = cv_f_.wait_until(
-                    lock, when + std::chrono::seconds{1});
+                cv_f_.wait_until(lock, when + std::chrono::seconds{1});
                 if(stop_)
                     break;
-                if(status == std::cv_status::timeout && ! waiting_.empty())
+                if(! waiting_.empty())
                 {
                     // Move everything in waiting_ to flushing_
                     for(auto store = waiting_.head_; store;
@@ -89,7 +88,13 @@ run()
                 when = clock_type::now();
                 if(flushing_.empty())
                     continue;
-                cv_f_.notify_all();
+
+                if (num_threads_ > 1)
+                {
+                    // Let other threads flush while this one splices
+                    cv_f_.notify_all();
+                    continue;
+                }
             }
             else
             {
@@ -103,7 +108,7 @@ run()
             }
         }
 
-        // process everything in flushing_
+        // Process everything in flushing_
         for(;;)
             if(! flush_one())
                 break;