Protecting few deadline timers with mutex

WARNING: ThreadSanitizer: data race (pid=18054)
  Read of size 1 at 0x7b1000014d90 by thread T11:
    #8 opflexagent::InterfaceStatsManager::on_timer(boost::system::error_code const&) ovs/InterfaceStatsManager.cpp:105 (libopflex_agent_renderer_openvswitch.so+0x1e74e1)
    #14 operator() lib/Agent.cpp:605 (libopflex_agent.so.0+0x275439)
    #15 __invoke_impl<void, opflexagent::Agent::start()::<lambda()> > /usr/include/c++/9/bits/invoke.h:60 (libopflex_agent.so.0+0x275439)
    #16 __invoke<opflexagent::Agent::start()::<lambda()> > /usr/include/c++/9/bits/invoke.h:95 (libopflex_agent.so.0+0x275439)
    #17 _M_invoke<0> /usr/include/c++/9/thread:244 (libopflex_agent.so.0+0x275439)
    #18 operator() /usr/include/c++/9/thread:251 (libopflex_agent.so.0+0x275439)
    #19 _M_run /usr/include/c++/9/thread:195 (libopflex_agent.so.0+0x275439)
    #20 <null> <null> (libstdc++.so.6+0xd086f)

  Previous write of size 1 at 0x7b1000014d90 by main thread (mutexes: write M180):
    #4 opflexagent::InterfaceStatsManager::stop() ovs/InterfaceStatsManager.cpp:91 (libopflex_agent_renderer_openvswitch.so+0x1e81af)
    #5 opflexagent::OVSRenderer::stop() ovs/OVSRenderer.cpp:259 (libopflex_agent_renderer_openvswitch.so+0xcb92b)
    #6 opflexagent::Agent::stop() lib/Agent.cpp:718 (libopflex_agent.so.0+0x273368)
    #7 AgentLauncher::run() cmd/opflex_agent.cpp:128 (opflex_agent+0x3a4a7)
    #8 main cmd/opflex_agent.cpp:322 (opflex_agent+0x16dc2)

  Location is heap block of size 64 at 0x7b1000014d80 allocated by main thread:
    #0 operator new(unsigned long) <null> (libtsan.so.0+0x77cf2)
    #1 opflexagent::InterfaceStatsManager::start() ovs/InterfaceStatsManager.cpp:73 (libopflex_agent_renderer_openvswitch.so+0x1e9646)
    #2 opflexagent::OVSRenderer::start() ovs/OVSRenderer.cpp:187 (libopflex_agent_renderer_openvswitch.so+0xd5508)
    #3 opflexagent::Agent::start() lib/Agent.cpp:601 (libopflex_agent.so.0+0x27937a)
    #4 AgentLauncher::run() cmd/opflex_agent.cpp:120 (opflex_agent+0x3a438)
    #5 main cmd/opflex_agent.cpp:322 (opflex_agent+0x16dc2)

  Mutex M180 (0x7ffc4bcb7548) created at:
    #0 pthread_mutex_lock <null> (libtsan.so.0+0x41d5b)
    #1 __gthread_mutex_lock /usr/include/x86_64-linux-gnu/c++/9/bits/gthr-default.h:749 (opflex_agent+0x3a3f0)
    #2 std::mutex::lock() /usr/include/c++/9/bits/std_mutex.h:100 (opflex_agent+0x3a3f0)
    #3 std::unique_lock<std::mutex>::lock() /usr/include/c++/9/bits/unique_lock.h:141 (opflex_agent+0x3a3f0)
    #4 std::unique_lock<std::mutex>::unique_lock(std::mutex&) /usr/include/c++/9/bits/unique_lock.h:71 (opflex_agent+0x3a3f0)
    #5 AgentLauncher::run() cmd/opflex_agent.cpp:115 (opflex_agent+0x3a3f0)
    #6 main cmd/opflex_agent.cpp:322 (opflex_agent+0x16dc2)

  Thread T11 (tid=18066, running) created by main thread at:
    #0 pthread_create <null> (libtsan.so.0+0x2d3be)
    #1 std::thread::_M_start_thread(std::unique_ptr<std::thread::_State, std::default_delete<std::thread::_State> >, void (*)()) <null> (libstdc++.so.6+0xd0b04)
    #2 AgentLauncher::run() cmd/opflex_agent.cpp:120 (opflex_agent+0x3a438)
    #3 main cmd/opflex_agent.cpp:322 (opflex_agent+0x16dc2)

Signed-off-by: Gautam Venkataramanan <gautam.chennai@gmail.com>

agent-ovs/cmd/test/policy_repo_stress.cpp

@@ -178,19 +178,25 @@ class StatsSimulator : public opflexagent::PolicyListener {
 
         LOG(INFO) << "Starting stats simulator";
 
-        timer.reset(new deadline_timer(io, milliseconds(timer_interval)));
-        timer->async_wait([this](const boost::system::error_code& ec) {
-                on_timer(ec);
-            });
+        {
+            const std::lock_guard<std::mutex> guard(timer_mutex);
+            timer.reset(new deadline_timer(io, milliseconds(timer_interval)));
+            timer->async_wait([this](const boost::system::error_code& ec) {
+                    on_timer(ec);
+                });
+        }
 
         io_service_thread.reset(new std::thread([this]() { io.run(); }));
     }
 
     void stop() {
         stopping = true;
 
-        if (timer) {
-            timer->cancel();
+        {
+            const std::lock_guard<std::mutex> guard(timer_mutex);
+            if (timer) {
+                timer->cancel();
+            }
         }
         if (io_service_thread) {
             io_service_thread->join();
@@ -201,6 +207,7 @@ class StatsSimulator : public opflexagent::PolicyListener {
     void on_timer(const boost::system::error_code& ec) {
         if (ec) {
             // shut down the timer when we get a cancellation
+            const std::lock_guard<std::mutex> guard(timer_mutex);
             timer.reset();
             return;
         }
@@ -214,6 +221,7 @@ class StatsSimulator : public opflexagent::PolicyListener {
         }
 
         if (!stopping) {
+            const std::lock_guard<std::mutex> guard(timer_mutex);
             timer->expires_at(timer->expires_at() +
                               milliseconds(timer_interval));
             timer->async_wait([this](const boost::system::error_code& ec) {
@@ -238,6 +246,7 @@ class StatsSimulator : public opflexagent::PolicyListener {
     boost::asio::io_service io;
     std::atomic_bool stopping;
     std::unique_ptr<std::thread> io_service_thread;
+    std::mutex timer_mutex;
     std::unique_ptr<boost::asio::deadline_timer> timer;
 
     uint32_t timer_interval;

agent-ovs/lib/SimStats.cpp

@@ -156,20 +156,23 @@ void SimStats::start() {
     LOG(INFO) << "Starting stats simulator";
 
     if (agent.getContractInterval() != 0) {
+        const std::lock_guard<std::mutex> guard(timer_mutex);
         contract_timer.reset(new deadline_timer(io, milliseconds(agent.getContractInterval())));
         contract_timer->async_wait([this](const boost::system::error_code& ec) {
                                           on_timer_contract(ec);
         });
     }
 
     if (agent.getSecurityGroupInterval() != 0) {
+        const std::lock_guard<std::mutex> guard(timer_mutex);
         security_group_timer.reset(new deadline_timer(io, milliseconds(agent.getSecurityGroupInterval())));
         security_group_timer->async_wait([this](const boost::system::error_code& ec) {
                                                 on_timer_security_group(ec);
         });
     }
 
     if (agent.getInterfaceInterval() != 0) {
+        const std::lock_guard<std::mutex> guard(timer_mutex);
         interface_timer.reset(new deadline_timer(io, milliseconds(agent.getInterfaceInterval())));
         interface_timer->async_wait([this](const boost::system::error_code& ec) {
                                            on_timer_interface(ec);
@@ -183,6 +186,7 @@ void SimStats::stop() {
     stopping = true;
 
     try {
+        const std::lock_guard<std::mutex> guard(timer_mutex);
         if (contract_timer) {
             contract_timer->cancel();
         }
@@ -215,6 +219,7 @@ inline bool SimStats::on_timer_check(std::shared_ptr<boost::asio::deadline_timer
                            const boost::system::error_code& ec) {
     if (ec) {
         // shut down the timer when we get a cancellation
+        const std::lock_guard<std::mutex> guard(timer_mutex);
         timer.reset();
         return false;
     }
@@ -224,6 +229,7 @@ inline void SimStats::timer_restart(uint32_t interval,
                       std::function<void(const boost::system::error_code& ec)> on_timer,
                       std::shared_ptr<boost::asio::deadline_timer> timer) {
     if (!this->stopping) {
+        const std::lock_guard<std::mutex> guard(timer_mutex);
         timer->expires_at(timer->expires_at() +
                           milliseconds(interval));
         timer->async_wait(on_timer);

agent-ovs/lib/include/opflexagent/SimStats.h

@@ -100,6 +100,7 @@ class SimStats : public opflexagent::PolicyListener {
     boost::asio::io_service io;
     std::atomic_bool stopping;
     std::shared_ptr<std::thread> io_service_thread;
+    std::mutex timer_mutex;
     std::shared_ptr<boost::asio::deadline_timer> contract_timer;
     std::shared_ptr<boost::asio::deadline_timer> security_group_timer;
     std::shared_ptr<boost::asio::deadline_timer> interface_timer;

agent-ovs/ovs/AdvertManager.cpp

@@ -66,6 +66,7 @@ AdvertManager::AdvertManager(Agent& agent_,
       started(false), stopping(false) {
     //TunnelEpTimer has no dependencies and needs to be
     //initialized early
+    lock_guard<recursive_mutex> guard(timer_mutex);
     tunnelEpAdvTimer.reset(new deadline_timer(*ioService));
 
 }
@@ -779,6 +780,7 @@ void AdvertManager::onTunnelEpAdvTimer(const boost::system::error_code& ec) {
 }
 
 void AdvertManager::doScheduleTunnelEpAdv(uint64_t time) {
+    lock_guard<recursive_mutex> guard(timer_mutex);
     tunnelEpAdvTimer->expires_from_now(seconds(time));
     tunnelEpAdvTimer->
             async_wait(bind(&AdvertManager::onTunnelEpAdvTimer,

agent-ovs/ovs/InterfaceStatsManager.cpp

@@ -70,6 +70,7 @@ void InterfaceStatsManager::start() {
         agent->getEndpointManager().registerListener(this);
     }
 
+    const std::lock_guard<std::mutex> guard(timer_mutex);
     timer.reset(new deadline_timer(agent_io, milliseconds(timer_interval)));
     timer->async_wait(bind(&InterfaceStatsManager::on_timer, this, error));
 }
@@ -87,6 +88,7 @@ void InterfaceStatsManager::stop() {
         agent->getEndpointManager().unregisterListener(this);
     }
 
+    const std::lock_guard<std::mutex> guard(timer_mutex);
     if (timer) {
         timer->cancel();
     }
@@ -102,6 +104,7 @@ void InterfaceStatsManager::endpointUpdated(const std::string& uuid) {
 void InterfaceStatsManager::on_timer(const error_code& ec) {
     if (ec) {
         // shut down the timer when we get a cancellation
+        const std::lock_guard<std::mutex> guard(timer_mutex);
         timer.reset();
         return;
     }
@@ -129,6 +132,7 @@ void InterfaceStatsManager::on_timer(const error_code& ec) {
     }
 
     if (!stopping) {
+        const std::lock_guard<std::mutex> guard(timer_mutex);
         timer->expires_at(timer->expires_at() + milliseconds(timer_interval));
         timer->async_wait(bind(&InterfaceStatsManager::on_timer, this, error));
     }

agent-ovs/ovs/include/InterfaceStatsManager.h

@@ -107,6 +107,7 @@ class InterfaceStatsManager : private boost::noncopyable,
     SwitchConnection* accessConnection;
     boost::asio::io_service& agent_io;
     long timer_interval;
+    std::mutex timer_mutex;
     std::unique_ptr<boost::asio::deadline_timer> timer;
 
     /**