Database buffer class

Question

I designed the following DataBaseBuffer class, to be used by different threads that generate queries and need to send them to the database. It was inspired by producer–consumer problem solution.

#include <nanodbc/nanodbc.h>
#include <queue>
#include <string>
#include <mutex>
#include <thread>
#include <algorithm>

class DataBaseBuffer {
   public:
    DataBaseBuffer(std::string& _connection_string_);
    void run();  // thread to send queries to the database
    void push(std::string& _query);
    nanodbc::connection conn_;  // connection to db

   private:
    void send_queries();
    nanodbc::connection conn_;        // connection to db
    std::string& connection_string_;  // save connection incase of reconnection
    std::queue<std::string> buffer_;
    std::queue<std::string> buffer_secondary_;
    std::mutex mutex_;
};

DataBaseBuffer::DataBasBuffer(std::string& _connection_string) : connection_string_(_connection_string) {}

void DataBaseBuffer::run() {
    conn_.connect(connection_string_);
    while (true) {
        send_queries();
        std::this_thread::sleep_for(std::chrono::milliseconds(1));
    }
}

void DataBaseBuffer::send_queries() {
    {
        std::lock_guard<std::mutex> lock(mutex_);
        std::swap(buffer_, buffer_secondary_); // move queries from main buffer to the secondary 
    }
    while (!buffer_secondary_.empty()) {
        nanodbc::execute(conn_, buffer_secondary_.front());
        buffer_secondary_.pop();
    }
}

void DataBaseBuffer::push(std::string& _query) {
    {
        std::lock_guard<std::mutex> lock(mutex_);
        buffer_.push(_query);
    }
}

Are there any flaws in this design that could affect the smooth running of the program?

Answer 1

Separate concerns

Your class does two things: it interacts with the database and it implements a thread-safe queue. It would be much better to separate those two things. Create a class ThreadSafeQueue that implements thread-safe access to the queue. Then DatabaseBuffer can in turn use that, so it's implementation is simplified:

class ThreadSafeQueue {
public:
    void push(const std::string& item);
    std::string pop();

private:
    std::queue<std::string> buffer_;
    std::mutex mutex_;
};

class DatabaseBuffer {
public:
    /* same public functions as before */
    ...

private:
    void send_queries();
    nanodbc::connection conn_;
    std::string& connection_string_;
    ThreadSafeQueue queue_;
};

Pass strings by const reference where appropriate

If you pass a parameter by pointer or reference, but the parameter should not be changed, then make sure it is const as well. This allows the compiler to produce an error if you do accidentily write to the parameter, and it also allows the compiler to generate more optimal code.

Use a condition variable to sleep until the queue is non-empty

Repeatedly checking if the queue is non-empty in a loop is problematic, even if you call sleep_for(). Sleeping for too long obviously causes delays in the processing of queries, but sleeping too short will still use a bit of CPU time, and more importantly it will prevent the CPU from going into a deeper sleep mode, and thus it will keep using more power than necessary. The proper way to handle this is to use a condition variable that you can wait for to be notified by another thread. Here is an example of how ThreadSafeQueue could be implemented:

class ThreadSafeQueue {
public:
    void push(const std::string& item) {
        {
            std::lock_guard lock(mutex_);
            queue_.push(item);
        }

        cv_.notify_one();
    }

    std::string pop() {
        std::lock_guard lock(mutex_);
        cv_.wait(mutex_, []{return !queue_.empty();});
        auto item = queue_.front();
        queue.pop();
        return item;
    }

private:
    std::queue<std::string> buffer_;
    std::mutex mutex_;
    std::condition_variable cv_;
};

This will cause the thread calling pop() to check if the queue is empty, and if so it will sleep until it gets notified by another thread.

Some things to think about:

• You can change it to return a whole std::queue<std::string> instead of just one item, if you think that is important for efficiency.
• One issue is that if the producer stops, pop() might then wait indefinitely for an item to be added that never comes. Think of a way to safely terminate the consumer.
• You can make ThreadSafeQueue a template so it can store things other than std::strings.

What about query results?

Your code only executes queries, but it doesn't look at the result at all. Even an INSERT statement could fail, so at least you'd want some status result to be returned, but if you have a SELECT query then you also want the data returned by the database to be stored somewhere.

There are various ways to deal with this. I'll present one here that uses std::async() instead of a queue runner:

class DatabaseBuffer {
public:
    ...
    std::future<nanodbc::result> push(const std::string& query);

private:
    nanodbc::connection conn_;
    std::string& connection_string_; 
    std::mutex mutex_;
};

std::future<nanodbc::result> DatabaseBuffer::push(const std::string& query) {
    return std::async(std::launch::async, [&](std::string query){
        std::lock_guard lock(mutex_);
        return nanodbc::execute(conn_, query);
    }, query);
}

The push() function calls std::async(), which in turn will launch a thread that runs a function that locks the mutex and executes the query. While this happens, a std::future<> is immediately returned to the caller. The caller can then do other things, like launcher more queries, before checking the future. Getting the result from the future will block until the thread running the async function has finished. So to use it:

DatabaseBuffer db(...);

auto future = db.push("SELECT name from users;");
// Do something else
...
// Show the result
for (auto result = future.get(); result.next();) {
    std::cout << result.get<int>(0) << '\n';
}

You don't have to use std::async() to be able to use std::futures though; you can still have your own queue runner, but use std::promise objects to return a std::future to the caller that you can later then set to the result of a query.