Unverified Commit 5f629561 authored by Dennis Jenkins's avatar Dennis Jenkins Committed by GitHub

Merge pull request #353 from knowledge4igor/fix_data_race

Fix data race, maybe fix SEGFAULT in issue #318
parents 51226002 28f4c437
...@@ -16,6 +16,7 @@ ...@@ -16,6 +16,7 @@
#include <deque> #include <deque>
#include <memory> #include <memory>
#include <unordered_map> #include <unordered_map>
#include <mutex>
namespace Pistache { namespace Pistache {
namespace Tcp { namespace Tcp {
...@@ -199,6 +200,7 @@ private: ...@@ -199,6 +200,7 @@ private:
PollableQueue<WriteEntry> writesQueue; PollableQueue<WriteEntry> writesQueue;
std::unordered_map<Fd, std::deque<WriteEntry> > toWrite; std::unordered_map<Fd, std::deque<WriteEntry> > toWrite;
std::mutex toWriteLock;
PollableQueue<TimerEntry> timersQueue; PollableQueue<TimerEntry> timersQueue;
std::unordered_map<Fd, TimerEntry> timers; std::unordered_map<Fd, TimerEntry> timers;
......
...@@ -55,7 +55,10 @@ Transport::handleNewPeer(const std::shared_ptr<Tcp::Peer>& peer) { ...@@ -55,7 +55,10 @@ Transport::handleNewPeer(const std::shared_ptr<Tcp::Peer>& peer) {
handlePeer(peer); handlePeer(peer);
} }
int fd = peer->fd(); int fd = peer->fd();
{
std::lock_guard<std::mutex> lock(toWriteLock);
toWrite.emplace(fd, std::deque<WriteEntry>{}); toWrite.emplace(fd, std::deque<WriteEntry>{});
}
} }
void void
...@@ -94,10 +97,13 @@ Transport::onReady(const Aio::FdSet& fds) { ...@@ -94,10 +97,13 @@ Transport::onReady(const Aio::FdSet& fds) {
auto tag = entry.getTag(); auto tag = entry.getTag();
auto fd = tag.value(); auto fd = tag.value();
{
std::lock_guard<std::mutex> lock(toWriteLock);
auto it = toWrite.find(fd); auto it = toWrite.find(fd);
if (it == std::end(toWrite)) { if (it == std::end(toWrite)) {
throw std::runtime_error("Assertion Error: could not find write data"); throw std::runtime_error("Assertion Error: could not find write data");
} }
}
reactor()->modifyFd(key(), fd, NotifyOn::Read, Polling::Mode::Edge); reactor()->modifyFd(key(), fd, NotifyOn::Read, Polling::Mode::Edge);
...@@ -163,7 +169,9 @@ Transport::handlePeerDisconnection(const std::shared_ptr<Peer>& peer) { ...@@ -163,7 +169,9 @@ Transport::handlePeerDisconnection(const std::shared_ptr<Peer>& peer) {
peers.erase(it); peers.erase(it);
{
// Clean up buffers // Clean up buffers
std::lock_guard<std::mutex> lock(toWriteLock);
auto & wq = toWrite[fd]; auto & wq = toWrite[fd];
while (wq.size() > 0) { while (wq.size() > 0) {
auto & entry = wq.front(); auto & entry = wq.front();
...@@ -175,6 +183,7 @@ Transport::handlePeerDisconnection(const std::shared_ptr<Peer>& peer) { ...@@ -175,6 +183,7 @@ Transport::handlePeerDisconnection(const std::shared_ptr<Peer>& peer) {
wq.pop_front(); wq.pop_front();
} }
toWrite.erase(fd); toWrite.erase(fd);
}
close(fd); close(fd);
} }
...@@ -182,13 +191,19 @@ Transport::handlePeerDisconnection(const std::shared_ptr<Peer>& peer) { ...@@ -182,13 +191,19 @@ Transport::handlePeerDisconnection(const std::shared_ptr<Peer>& peer) {
void void
Transport::asyncWriteImpl(Fd fd) Transport::asyncWriteImpl(Fd fd)
{ {
bool stop = false;
while (!stop) {
std::lock_guard<std::mutex> lock(toWriteLock);
auto it = toWrite.find(fd); auto it = toWrite.find(fd);
// cleanup will have been handled by handlePeerDisconnection // cleanup will have been handled by handlePeerDisconnection
if (it == std::end(toWrite)) { return; } if (it == std::end(toWrite)) { return; }
auto & wq = it->second; auto & wq = it->second;
bool stop = false; if (wq.size() == 0) {
while (!stop && wq.size() > 0) { break;
}
auto & entry = wq.front(); auto & entry = wq.front();
int flags = entry.flags; int flags = entry.flags;
const BufferHolder &buffer = entry.buffer; const BufferHolder &buffer = entry.buffer;
...@@ -314,7 +329,10 @@ Transport::handleWriteQueue() { ...@@ -314,7 +329,10 @@ Transport::handleWriteQueue() {
auto fd = write.peerFd; auto fd = write.peerFd;
if (!isPeerFd(fd)) continue; if (!isPeerFd(fd)) continue;
{
std::lock_guard<std::mutex> lock(toWriteLock);
toWrite[fd].push_back(std::move(write)); toWrite[fd].push_back(std::move(write));
}
reactor()->modifyFd(key(), fd, NotifyOn::Read | NotifyOn::Write, Polling::Mode::Edge); reactor()->modifyFd(key(), fd, NotifyOn::Read | NotifyOn::Write, Polling::Mode::Edge);
} }
......
Markdown is supported
0%
or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment