limit the number of buffers that are in the copy queue

If PostgreSQL is much slower than the data processing, the
queue can become very large and lead to out-of-memory
errors.

Author: lonvia

db-copy.cpp

@@ -28,9 +28,18 @@ 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_worker_queue.push_back(std::move(buffer));
-    m_queue_cond.notify_one();
+
+    for (;;) {
+        std::unique_lock<std::mutex> lock(m_queue_mutex);
+        if (m_worker_queue.size() >= db_cmd_copy_t::Max_buffers) {
+            m_queue_full_cond.wait(lock);
+            continue;
+        }
+
+        m_worker_queue.push_back(std::move(buffer));
+        m_queue_cond.notify_one();
+        break;
+    }
 }
 
 void db_copy_thread_t::sync_and_wait()
@@ -67,6 +76,7 @@ void db_copy_thread_t::worker_thread()
 
             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,7 @@ class db_cmd_t
 
 struct db_cmd_copy_t : public db_cmd_t
 {
-    enum { Max_buf_size = 10 * 1024 * 1024 };
+    enum { Max_buf_size = 10 * 1024 * 1024, 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 +141,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.