Merge pull request #960 from lonvia/make-copy-queue-bounded

limit the number of buffers that are in the copy queue

Author: lonvia

db-copy.cpp

@@ -28,7 +28,11 @@ db_copy_thread_t::~db_copy_thread_t() { finish(); }
 void db_copy_thread_t::add_buffer(std::unique_ptr<db_cmd_t> &&buffer)
 {
     assert(m_worker.joinable()); // thread must not have been finished
+
     std::unique_lock<std::mutex> lock(m_queue_mutex);
+    m_queue_full_cond.wait(lock,
+            [&]{ return m_worker_queue.size() < db_cmd_copy_t::Max_buffers; });
+
     m_worker_queue.push_back(std::move(buffer));
     m_queue_cond.notify_one();
 }
@@ -60,13 +64,11 @@ void db_copy_thread_t::worker_thread()
         std::unique_ptr<db_cmd_t> item;
         {
             std::unique_lock<std::mutex> lock(m_queue_mutex);
-            if (m_worker_queue.empty()) {
-                m_queue_cond.wait(lock);
-                continue;
-            }
+            m_queue_cond.wait(lock, [&]{ return !m_worker_queue.empty(); });
 
             item = std::move(m_worker_queue.front());
             m_worker_queue.pop_front();
+            m_queue_full_cond.notify_one();
         }
 
         switch (item->type) {

db-copy.hpp

@@ -67,7 +67,22 @@ class db_cmd_t
 
 struct db_cmd_copy_t : public db_cmd_t
 {
-    enum { Max_buf_size = 10 * 1024 * 1024 };
+    enum {
+        /** Size of a single buffer with COPY data for Postgresql.
+         *  This is a trade-off between memory usage and sending large chunks
+         *  to speed up processing. Currently a one-size fits all value.
+         *  Needs more testing and individual values per queue.
+         */
+        Max_buf_size = 10 * 1024 * 1024,
+        /** Maximum length of the queue with COPY data.
+         *  In the usual case, PostgreSQL should be faster processing the
+         *  data than it can be produced and there should only be one element
+         *  in the queue. If PostgreSQL is slower, then the queue will always
+         *  be full and it is better to keep the queue smaller to reduce memory
+         *  usage. Current value is just assumed to be a reasonable trade off.
+         */
+        Max_buffers = 10
+    };
     /// Name of the target table for the copy operation
     std::shared_ptr<db_target_descr_t> target;
     /// Vector with object to delete before copying
@@ -141,6 +156,7 @@ class db_copy_thread_t
     std::thread m_worker;
     std::mutex m_queue_mutex;
     std::condition_variable m_queue_cond;
+    std::condition_variable m_queue_full_cond;
     std::deque<std::unique_ptr<db_cmd_t>> m_worker_queue;
 
     // Target for copy operation currently ongoing.