一个线程提升的死锁

我花了大约 20 分钟将示例代码制作成自包含代码。当然，它只是有效，但那是因为无论如何，您都没有显示锁定的代码。

也许我推断和填空的方式会帮助你发现你所做的不同之处：

住在科里鲁

#include <boost/asio.hpp>
#include <boost/bind.hpp>
#include <boost/thread.hpp>
static const auto PTIME_INTERVAL = 10;
static std::atomic_size_t pTimerIterations { 0 };
namespace asio = boost::asio;
using io_service_ptr = std::shared_ptr<asio::io_service>;
using work_ptr = std::shared_ptr<asio::io_service::work>;
struct ThreadPoolManager {
std::vector<io_service_ptr> io_services_endpoint;
std::vector<work_ptr> work_endpoint;
boost::thread_group threads_endpoint;
io_service_ptr createNewService() {
io_service_ptr io_service = std::make_shared<asio::io_service>();
work_ptr work = std::make_shared<asio::io_service::work>(*io_service);
io_services_endpoint.push_back(io_service);
work_endpoint.push_back(work);
threads_endpoint.create_thread(boost::bind(&asio::io_service::run, io_service));
return io_service;
}
~ThreadPoolManager() {
for(auto& w : work_endpoint)
w.reset();
threads_endpoint.join_all();
}
};
struct RtpEndpoint {
};
using rtp_endpoint_ptr = std::shared_ptr<RtpEndpoint>;
struct EndpointGroup : std::enable_shared_from_this<EndpointGroup>, boost::mutex {
io_service_ptr _io;
asio::high_resolution_timer pTimer;
std::vector<rtp_endpoint_ptr> endpoints;
std::chrono::high_resolution_clock::time_point start = std::chrono::high_resolution_clock::now();
EndpointGroup(io_service_ptr io) : _io(io), pTimer(*_io) {}
void stop() {
auto self(shared_from_this());
_io->post([self,this] { pTimer.cancel(); });
}
// simply start this function over and over again every 10ms
void invokeSendingOnEndpoints(size_t offset) {
pTimer.expires_from_now(std::chrono::milliseconds(PTIME_INTERVAL - offset));
auto self(shared_from_this());
pTimer.async_wait([this, self](boost::system::error_code ec) {
if (!ec) {
std::vector<rtp_endpoint_ptr> iterationEndpoints;
{
boost::lock_guard<EndpointGroup> guard(*this);
iterationEndpoints = endpoints;
}
for (rtp_endpoint_ptr endpoint : iterationEndpoints) {
// do fancy stuff
}
++pTimerIterations;
// check how many milliseconds passed since start of the function
invokeSendingOnEndpoints(std::chrono::duration_cast<std::chrono::milliseconds>(
(std::chrono::high_resolution_clock::now() - start))
.count() /
(pTimerIterations * PTIME_INTERVAL));
} else {
// just write error happend....
}
});
}
bool addEndpoint(const rtp_endpoint_ptr &endpoint) {
boost::lock_guard<EndpointGroup> guard(*this);
endpoints.push_back(endpoint);
return true;
}
};
using group_ptr = std::shared_ptr<EndpointGroup>;
#include <iostream>
int main() {
std::cout << "starting" << std::endl;
{
ThreadPoolManager tpm;
std::vector<group_ptr> groups;
for (int i = 0; i < 5; ++i) {
std::cout << "Group " << i << std::endl;
auto epg = std::make_shared<EndpointGroup>(tpm.createNewService());
epg->invokeSendingOnEndpoints(i*2);
for (int j = 0; j < rand()%10; ++j) {
epg->addEndpoint(std::make_shared<RtpEndpoint>());
std::cout << " - RtpEndpoint " << i << "." << j << std::endl;
}
groups.push_back(epg);
}
std::cout << "waiting..." << std::endl;
std::this_thread::sleep_for(std::chrono::seconds(4));
std::cout << "shutting down" << std::endl;
for(auto& g : groups)
g->stop();
}
std::cout << "done, " << pTimerIterations << " iterations" << std::endl;
}

指纹

starting
Group 0
- RtpEndpoint 0.0
- RtpEndpoint 0.1
- RtpEndpoint 0.2
- RtpEndpoint 0.3
Group 1
- RtpEndpoint 1.0
- RtpEndpoint 1.1
Group 2
- RtpEndpoint 2.0
Group 3
- RtpEndpoint 3.0
- RtpEndpoint 3.1
Group 4
- RtpEndpoint 4.0
- RtpEndpoint 4.1
- RtpEndpoint 4.2
waiting...
shutting down
done, 1963 iterations

然而

正如其他人提到的，这是非常非正统的代码。

1. 您只需 1 个io_service即可拥有相同的内容
2. 具有并行容器是一种代码异味(有一个struct { thread, service, work }向量，而不是三个包含服务、线程和工作对象的向量。
3. 永远不要从std::mutex继承。也没有什么理由自己实现BasicLockable。相反，如果必须，请公开unique_lock。它将是异常安全的，并且具有明确定义的延迟/采用语义。
4. 计时业务看起来很像你试图实现我在这个答案中展示的内容：boost：：asio：:d eadline_timer 1ms 在一段时间后滞后

在这种情况下，我认为整个池最多需要 1 个 io_service + 1 个工作对象，这是一个简化的做法：

住在科里鲁

#include <boost/asio.hpp>
#include <boost/bind.hpp>
#include <boost/thread.hpp>
static const auto PTIME_INTERVAL = 10; // boost::posix_time::milliseconds(10);
static std::atomic_size_t pTimerIterations { 0 };
namespace asio = boost::asio;
struct ThreadPoolManager {
~ThreadPoolManager() {
work.reset();
threads_endpoint.join_all();
}
boost::asio::io_service& get_service() { return io; }
void launch() {
threads_endpoint.create_thread([this]{ io.run(); });
}
private:
asio::io_service io;
boost::optional<asio::io_service::work> work {io};
boost::thread_group threads_endpoint;
};
struct RtpEndpoint {
};
using rtp_endpoint_ptr = std::shared_ptr<RtpEndpoint>;
struct EndpointGroup : std::enable_shared_from_this<EndpointGroup> {
std::mutex _mx;
asio::io_service& _io;
asio::high_resolution_timer pTimer;
std::vector<rtp_endpoint_ptr> endpoints;
std::chrono::high_resolution_clock::time_point start = std::chrono::high_resolution_clock::now();
EndpointGroup(asio::io_service& io) : _io(io), pTimer(_io) {}
void stop() {
auto self(shared_from_this());
_io.post([self,this] { pTimer.cancel(); });
}
// simply start this function over and over again every 10ms
void invokeSendingOnEndpoints(size_t offset) {
pTimer.expires_from_now(std::chrono::milliseconds(PTIME_INTERVAL - offset));
auto self(shared_from_this());
pTimer.async_wait([this, self](boost::system::error_code ec) {
if (!ec) {
std::vector<rtp_endpoint_ptr> iterationEndpoints;
{
boost::lock_guard<std::mutex> guard(_mx);
iterationEndpoints = endpoints;
}
for (rtp_endpoint_ptr endpoint : iterationEndpoints) {
// do fancy stuff
}
++pTimerIterations;
// check how many milliseconds passed since start of the function
invokeSendingOnEndpoints(std::chrono::duration_cast<std::chrono::milliseconds>(
(std::chrono::high_resolution_clock::now() - start))
.count() /
(pTimerIterations * PTIME_INTERVAL));
} else {
// just write error happend....
}
});
}
bool addEndpoint(const rtp_endpoint_ptr &endpoint) {
boost::lock_guard<std::mutex> guard(_mx);
endpoints.push_back(endpoint);
return true;
}
};
using group_ptr = std::shared_ptr<EndpointGroup>;
#include <iostream>
int main() {
std::cout << "starting" << std::endl;
{
ThreadPoolManager tpm;
for (unsigned i = 0; i < std::thread::hardware_concurrency(); ++i)
tpm.launch();
std::vector<group_ptr> groups;
for (int i = 0; i < 5; ++i) {
std::cout << "Group " << i << std::endl;
auto epg = std::make_shared<EndpointGroup>(tpm.get_service());
epg->invokeSendingOnEndpoints(i*2);
for (int j = 0; j < rand()%10; ++j) {
epg->addEndpoint(std::make_shared<RtpEndpoint>());
std::cout << " - RtpEndpoint " << i << "." << j << std::endl;
}
groups.push_back(epg);
}
std::cout << "waiting..." << std::endl;
std::this_thread::sleep_for(std::chrono::seconds(4));
std::cout << "shutting down" << std::endl;
for(auto& g : groups)
g->stop();
}
std::cout << "done, " << pTimerIterations << " iterations" << std::endl;
}

具有相同的输出。