Можно ли параллельно запускать io_service::run?

Question

[HamsterGamer]

Имею довольно большой код, в котором санитайзер выявляет data race. В этом коде я использую мьютексы в каждом методе моего класса с сетевым интерфейсом, который хранит/создает для себя свой сокет, примерно так:

struct NetRequster {
//...
   void some_work_if_recive();  
   void some_work_if_timeout();
// ...
private:
   ba::ip::tcp::socket socket_;
   std::mutex m;
// ...
};

Если данные придут, то вызывается метод some_work_if_recive, который что-то сделает и в дальнейшем вырубит все операции со своим классом с помощью булевого флажка, если данные не придут за определенное время, то вызывается some_work_if_timeout, который вызовет у сокета метод close, и все эти данные всегда будут под мьютексом, параллелить же планируется работу отдельных NetRequster. С другой стороны, создав много таких реквестеров я просто запускаю io_service::run в N потоках.
Однако io_service где-то у себя проверяет открыт ли сокет методом is_open, что в свою очередь не закрывается моим мьютексом и приводит к дата рейсу. Я не думаю что это нормально, но выходит что параллелить так вообще не получится, подскажите что я делаю не так или как оно должно выглядеть на самом деле? Возможно даже есть пример того что я хочу!
Вот что выдает санитайзер:

Comments

[HamsterGamer]

переписал со strand и все равно санитайзер плачет на это место

[HamsterGamer]

Я смог повторить ошибку, вот код:

#include <boost/asio.hpp>
#include <boost/asio/deadline_timer.hpp>
#include <boost/asio/io_service.hpp>
#include <boost/asio/ip/tcp.hpp>
#include <boost/asio/read_until.hpp>
#include <boost/asio/streambuf.hpp>
#include <boost/asio/strand.hpp>
#include <boost/thread.hpp>
#include <boost/asio/write.hpp>
#include <boost/bind.hpp>
#include <iostream>
#include <thread>

using namespace boost::asio;
io_service service;
struct D {
    explicit D(io_service & ios) : socket_(ios), timeout_(ios), strand_(ios), resolver_(ios) {
        do_response();
    }
private:
    void do_response(){
        resolver_.async_resolve("tracker.dler.com",  "6969",
                                strand_.wrap( [this](boost::system::error_code const & ec,
                                                      ip::tcp::resolver::iterator endpoints){
                                    if (!ec) {
                                        do_connect(std::move(endpoints));
                                        timeout_.async_wait([this](boost::system::error_code const & ec) {
                                            if (!ec) {
                                                deadline();
                                            }
                                        });
                                    } else {
                                        socket_.close();
                                        std::cerr << "ec == true" << std::endl;
                                    }
                                }));
    }
    void do_connect(ip::tcp::resolver::iterator endpoint){
        async_connect(socket_, std::move(endpoint),
                          strand_.wrap([this](boost::system::error_code const & ec, [[maybe_unused]] const ip::tcp::resolver::iterator&){
                              std::this_thread::sleep_for(std::chrono::milliseconds(100));
                              if (!ec) {
                                  std::cout << "OK!" << std::endl;
                                  timeout_.cancel();
                              }
                          }));
        timeout_.expires_from_now(boost::posix_time::seconds(1));
    }
    void deadline(){
        if (timeout_.expires_at() <=  deadline_timer::traits_type::now()) {
            socket_.close();
            std::cerr << "TIMEOUT" << std::endl;
        } else {
            timeout_.async_wait(strand_.wrap([this](boost::system::error_code const &ec) {
                if (!ec) {
                    deadline();
                }
            }));
        }
    }

    ip::tcp::socket socket_;
    deadline_timer timeout_;
    io_service::strand strand_;
    ip::tcp::resolver resolver_;
};
void func(int i)
{
    std::cout << "func called, i= " << i << "/" << boost::this_thread::get_id() << std::endl;
}
void worker_thread()
{
    service.run();
}
int main(int argc, char* argv[])
{
    std::list<D> ds;
    for ( int i = 0; i < 15; ++i) {
        ds.emplace_back(service);
    }
    boost::thread_group threads;
    for ( int i = 0; i < 3; ++i)
        threads.create_thread(worker_thread);
    // wait for all threads to be created
    boost::this_thread::sleep( boost::posix_time::millisec(500));
    threads.join_all();
}