4
\$\begingroup\$

I am currently having a problem with too many database connections. Therefore, I am trying to implement a database connection timeout. In the code below the changing of value of the shared resource will be replaced with connecting and disconnecting to the database. I am new to programming with threads and I kindly wonder if I have made any logical errors or if there are any better approaches to this problem? There is no built-in timeout in OCCI which I use.

#include <thread>
#include <chrono>
#include <iostream>
#include <mutex>
#include <atomic>

namespace chrono = std::chrono;

class scoped_thread{
  std::thread t;
public:
  explicit scoped_thread(std::thread t_): t(std::move(t_)){
    if ( !t.joinable()) throw std::logic_error("No thread");
  }
  ~scoped_thread(){
    t.join();
    std::cout << "Thread " << std::this_thread::get_id() << " joined.\n";
  }
  scoped_thread(scoped_thread&)= delete;
  scoped_thread& operator=(scoped_thread const &)= delete;
};

template<typename T>
class LockWrapper {
    T* t;
    std::mutex mut;

public:
    class Proxy {
        T* t;
        std::unique_lock<std::mutex> lock;
        Proxy(LockWrapper& wr) : t(wr.t), lock(wr.mut) { }
        friend LockWrapper;
    public:
        T& operator*() { return *t; }
        T* operator->() { return t; }
    };
    LockWrapper(T* t) : t(t) {}

    Proxy aquire() { return {*this}; }
};

using clock_type = chrono::system_clock;
using shared_deadline = std::atomic<clock_type::time_point>;
constexpr auto tlim=100;
template<typename T> void timeout(T& shared_resource,const shared_deadline& deadline) {
    while(clock_type::now()<deadline.load())
    {
        constexpr auto sleeping_time=1;
        std::cout << "Sleeping for another  " << 1 << " seconds\n";
        std::this_thread::sleep_until(clock_type::now()+chrono::seconds(1));
    }
    *shared_resource.aquire()=0; // will replace by database disconnect
    std::cout << "Setting resource to " << (*shared_resource.aquire()) << " after timeout occurred in thread " << std::this_thread::get_id() << ".\n";
}
int main()
{
    int int_value=1;
    LockWrapper<int> shared_resource(&int_value);
    constexpr auto timeout_length=10;
    auto get_new_deadline=[timeout_length](){return clock_type::now()+chrono::seconds(timeout_length);};
    shared_deadline deadline(get_new_deadline());
    auto th = std::make_unique<scoped_thread>(std::thread(timeout<LockWrapper<int>>, std::ref(shared_resource),std::cref(deadline)));
    constexpr int sleeping_time=20;
    for(auto t=0;t<tlim;t+=sleeping_time) // This loop simulates repeated database access
    {
        std::cout << "Slept for " << t << " seconds. Sleeping for another " << sleeping_time << " seconds\n";
        std::this_thread::sleep_for(chrono::seconds(sleeping_time));
        if((*shared_resource.aquire())==0)
        {
            *shared_resource.aquire()=1; // will replace by database connect
            std::cout << "Setting resource to " << (*shared_resource.aquire()) << "\n";
            deadline=get_new_deadline();
            th = std::make_unique<scoped_thread>(std::thread(timeout<LockWrapper<int>>, std::ref(shared_resource),std::cref(deadline)));
        }
        else
        {
            deadline=get_new_deadline();
        }
    }
}
\$\endgroup\$
3
\$\begingroup\$

Code

  • Acquire not aquire.
  • The scoped_thread move constructor is missing a &, and should have the signature scoped_thread(scoped_thread&&) = delete.
  • system_clock is the wrong clock type to use, as it's not monotonic (the user can adjust it whenever they choose). steady_clock would probably be the right choice here.
  • Since scoped_thread stores the thread object as a member, the scoped_thread constructor could use variable template arguments and perfect forwarding, rather than passing in a std::thread object. This makes the user code slightly shorter.
  • When declaring time constants (tlim, sleeping_time), it would be better to use std::chrono types at the point of declaration, rather than later on in the code, so that the constant has units associated with it directly.
  • (Style/Opinion:) Vertical spacing makes code easier to read:

    using clock_type = chrono::system_clock;
    using shared_deadline = std::atomic<clock_type::time_point>;
    
    constexpr auto tlim=100;
    
    template<typename T> void timeout(T& shared_resource,const shared_deadline& deadline)
    

    With a little extra spacing, it's a lot easier to parse this visually as two using declarations, a constant and the start of a function.


Design - locking

There are a couple of places where the resource is locked and then unlocked again rapidly, e.g.:

    if((*shared_resource.aquire())==0)
    {
        *shared_resource.aquire()=1; // will replace by database connect
        std::cout << "Setting resource to " << (*shared_resource.aquire()) << "\n";

Maintaining the lock would be better, otherwise what's printed to std::cout may not be what you expect:

    auto proxy = shared_resource.aquire();

    if (*proxy == 0)
    {
        *proxy = 1; // will replace by database connect
        std::cout << "Setting resource to " << *proxy << "\n";

Design - cancelling / timeout behaviour

  • If you comment out every line after creating the timeout object:

    auto th = std::make_unique<scoped_thread>(std::thread(timeout<LockWrapper<int>>, std::ref(shared_resource), std::cref(deadline)));
    

    main() won't complete until the timeout finishes. This is probably not desirable behaviour.

  • There is no way to cancel the timeout without manually setting a deadline zero seconds in the future (i.e. dodgy hacks). Furthermore, the fixed sleep of 1 second inside timeout() means you may have to wait up to a second anyway.

  • Sleeping for one second in a loop seems like strange behaviour, since we know exactly how long we actually need to sleep for before waking up.

  • It is error-prone to manually increase the deadline every time an operation that requires the connection is used.


Changes

Some of the problems above can be solved by encapsulating the timeout functionality in a class, and using a std::condition_variable to do the timed waiting. This also allows the callback to be cancelled (or called immediately) if desired.

class ThreadedCallback
{
public:

    using clock_t = chrono::steady_clock;
    using duration_t = clock_t::duration;
    using action_t = std::function<void()>;

    ThreadedCallback(duration_t timeout, action_t const& action)
    {
        start(timeout, action);
    }

    ~ThreadedCallback()
    {
        cancel();
    }

    ThreadedCallback(ThreadedCallback&&) = delete;
    ThreadedCallback(ThreadedCallback const&) = delete;

    void restart(duration_t timeout, action_t const& action)
    {
        cancel();

        start(timeout, action);
    }

    void cancel()
    {
        {
            std::lock_guard<std::mutex> lock(m_mutex);

            if (!m_cancelled)
            {
                m_cancelled = true;
                m_cv.notify_one();
            }
        }

        assert(m_thread.joinable());
        m_thread.join();
    }

private:

    void start(duration_t timeout, action_t const& action)
    {
        assert(!m_thread.joinable());
        m_thread = std::thread([=] () { run(timeout, action); });
    }

    void run(duration_t timeout, action_t const& action)
    {
        std::unique_lock<std::mutex> lock(m_mutex);

        m_cancelled = false;
        m_cv.wait_for(lock, timeout, [&] () { return m_cancelled; });

        // note: this check can be removed (and cancel() renamed to call_now()) 
        // or a boolean flag added to cancel() to indicate whether the action should still be called anyway
        if (!m_cancelled)
            action();
    }

    std::thread m_thread;
    std::mutex m_mutex;
    std::condition_variable m_cv;
    bool m_cancelled;
};

int main()
{
    int int_value = 1;
    LockWrapper<int> shared_resource(&int_value);

    auto disconnect = [&] ()
    {
        auto proxy = shared_resource.acquire();
        *proxy = 0; // will replace by database disconnect
        std::cout << "Setting resource to " << *proxy << " after timeout occurred in thread " << std::this_thread::get_id() << ".\n";
    };

    auto timeout = std::chrono::seconds(5);
    ThreadedCallback callback(timeout, disconnect);

    constexpr auto tlim = std::chrono::seconds(50);
    constexpr auto sleeping_time = std::chrono::seconds(2);
    for (auto t = std::chrono::seconds(0); t < tlim; t += sleeping_time) // This loop simulates repeated database access
    {
        std::cout << "Slept for " << t.count() << " seconds. Sleeping for another " << sleeping_time.count() << " seconds\n";
        std::this_thread::sleep_for(sleeping_time);

        auto proxy = shared_resource.acquire();

        if (*proxy == 0)
        {
            *proxy = 1; // will replace by database connect
            std::cout << "Setting resource to " << *proxy << "\n";

            callback.restart(timeout, disconnect);
        }
        else
        {
            callback.restart(timeout, disconnect);
        }
    }

    std::cout << "done!" << std::endl;
}

This still doesn't solve the issue of manually restarting the timer with every access operation. Perhaps a timeout class like this would be better kept in the resource (or LockWrapper) itself.

For example, since we want the time after the last access, the Proxy class destructor could start the timeout (e.g. call a function in the LockWrapper to create a ThreadedCallback in a unique_pointer), and the Proxy constructor could call a function to clear that pointer (cancelling the disconnect operation).

Then the timeout would be directly tied to the access itself, and not have to be reset manually by user code.

\$\endgroup\$
3
  • \$\begingroup\$ Thanks a lot for your valuable review. I agree with what I have read and understood so far. In particular I like your advice to declare std::chrono types instead of ints. I will look into your proposal in detail later this week. One comment I have is that the access operations will usually be much more frequent than the timeouts, i.e. I will have quite long timeout times. Thanks again for your expert help! \$\endgroup\$ Feb 21 '18 at 11:35
  • 1
    \$\begingroup\$ Hi again. Sorry for the long delay. Hope you haven't timed out? :) I have read your proposal. Your solution seems to restart the timeout thread for every access. It seems expensive compared to updating the timeout time. What do you think? \$\endgroup\$ Mar 14 '18 at 18:03
  • \$\begingroup\$ Yep, that's true. This way seemed easiest at the time, but it would probably be better not to restart the thread. \$\endgroup\$
    – user673679
    Mar 16 '18 at 20:53
0
\$\begingroup\$

Based on the answer from @user673679 (using condition_variable which I was unaware of) I changed to a solution in which the same timeout thread continuously performs the following:

  1. Sets the status of the resource to not accessed.
  2. Sleeps a certain time.
  3. Checks if the resource has been accessed.
  4. If not accessed then the timeout action is performed (database disconnection).
  5. Start over again.

Source code

#include <thread>
#include <chrono>
#include <iostream>
#include <mutex>
#include <condition_variable>

namespace chrono = std::chrono;
using namespace std::chrono_literals;
enum class state {
    accessed, untouched
};
inline std::string to_string(state s)
{
    switch(s)
    {
        case state::accessed:
            return "accessed";
        case state::untouched:
            return "untouched";
    }
    return std::string();
}
template<typename T>
class LockWrapper {
    T* t;
    state s;
    std::mutex mut;
public:

    class Proxy {
        T* t;
        state& s;
        std::unique_lock<std::mutex> lock;

        Proxy(LockWrapper& wr) : t(wr.t), s(wr.s), lock(wr.mut) {
        }
        friend LockWrapper;
    public:

        T& operator*() {
            s = state::accessed;
            return *t;
        }

        T* operator->() {
            s = state::accessed;
            return t;
        }

        void reset() {
            s = state::untouched;
        }

        state get_state() const {
            return s;
        }
    };

    explicit LockWrapper(T* t) : t(t), s(state::untouched) {
    }

    Proxy acquire() {
        return {*this};
    }
};

template<typename T> class timeout_thread {
    const chrono::seconds timeout_duration;
    bool is_cancelled;
    std::mutex cv_m;
    std::condition_variable cv;
    LockWrapper<T>& shared_resource;
    std::thread t;// I think the thread needs to be constructed last as dependent upon the other members
    void run() {
        std::unique_lock<std::mutex> lk(cv_m);
        while(true)
        {
            {
                auto p = shared_resource.acquire();
                p.reset();
            }
            if(cv.wait_for(lk,timeout_duration, [this](){std::cout << "Checking condition..."<< std::endl; return is_cancelled;}))
            {
                break;
            }
            else
            {
                auto p=shared_resource.acquire();
                std::cout << "Checking state:" << to_string(p.get_state()) << std::endl;
                if(p.get_state()==state::untouched)
                {
                    p->timeout_action();
                }
            }
        }
    }
public:

    timeout_thread(LockWrapper<T>& sr,chrono::seconds to_dur)
    : timeout_duration(to_dur),is_cancelled(false),shared_resource(sr), 
      t(std::thread([ = ](){run();}))
    {
        if (!t.joinable()) throw std::logic_error("No thread");
    }

    ~timeout_thread() {
        {
            std::lock_guard<std::mutex> lk(cv_m);
            is_cancelled = true;
        }
        cv.notify_one();
        t.join();
        std::cout << "Thread " << std::this_thread::get_id() << " joined.\n";
    }
    timeout_thread(timeout_thread&) = delete;
    timeout_thread& operator=(timeout_thread const &) = delete;
    timeout_thread(timeout_thread&&) = delete;
};

class shared_resource {
    int& value;
public:
    shared_resource(int& val) : value(val) {
        ;
    }
    void timeout_action()
    {
        std::cout << __func__ << std::endl;
        value=0;
    }
    int& get_value()
    {
        return value;
    }
};
int main() {
    int int_value = 1;
    shared_resource sr(int_value);
    LockWrapper<shared_resource> lw(&sr);
    timeout_thread<shared_resource> t(lw,2s);
    for (int i = 4; i < 5; ++i) {
        const auto st=i * 1s;
        std::cout << "Sleeping for " << st.count() << "\n";
        std::this_thread::sleep_for(st);
        auto p = lw.acquire();
        p->get_value() = 1;
    }
}
\$\endgroup\$

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service, privacy policy and cookie policy

Not the answer you're looking for? Browse other questions tagged or ask your own question.