fix(UDPHandler): data race #3613; clean up all Net tsan warnings

.vscode/settings.json

@@ -89,12 +89,15 @@
 		"ranges": "cpp",
 		"cfenv": "cpp",
 		"__bits": "cpp",
-		"variant": "cpp"
+		"variant": "cpp",
+		"condition_variable": "cpp",
+		"valarray": "cpp"
 	},
 	"files.exclude": {
         "**/.dep": true,
         "**/bin": true,
         "**/obj": true
 	},
-	"git.ignoreLimitWarning": true
+	"git.ignoreLimitWarning": true,
+	"cmake.configureOnOpen": false
 }

Foundation/include/Poco/Activity.h

@@ -143,6 +143,7 @@ public:
 		if (_running)
 		{
 			_done.wait();
+			_running = false;
 		}
 	}
 
@@ -154,6 +155,7 @@ public:
 		if (_running)
 		{
 			_done.wait(milliseconds);
+			_running = false;
 		}
 	}
 	
@@ -178,11 +180,9 @@ protected:
 		}
 		catch (...)
 		{
-			_running = false;
 			_done.set();
 			throw;
 		}
-		_running = false;
 		_done.set();
 	}
 	

Foundation/testsuite/src/ThreadTest.cpp

@@ -108,7 +108,7 @@ public:
 	}
 
 private:
-	bool _finished;
+	std::atomic<bool> _finished;
 };
 
 
@@ -140,8 +140,8 @@ public:
 	}
 
 private:
-	int _counter;
-	bool _sleepy;
+	std::atomic<int> _counter;
+	std::atomic<bool> _sleepy;
 };
 
 

Net/include/Poco/Net/SocketProactor.h

@@ -319,6 +319,7 @@ private:
 			/// Runs the next I/O completion handler in the queue.
 		{
 			IONotification* pNf = dynamic_cast<IONotification*>(_nq.waitDequeueNotification());
+			if (_activity.isStopped()) return false;
 			if (pNf)
 			{
 				try

Net/include/Poco/Net/TCPServer.h

@@ -27,6 +27,7 @@
 #include "Poco/Runnable.h"
 #include "Poco/Thread.h"
 #include "Poco/ThreadPool.h"
+#include <atomic>
 
 
 namespace Poco {
@@ -227,7 +228,7 @@ private:
 	TCPServerDispatcher* _pDispatcher;
 	TCPServerConnectionFilter::Ptr _pConnectionFilter;
 	Poco::Thread _thread;
-	bool _stopped;
+	std::atomic<bool> _stopped;
 };
 
 

Net/include/Poco/Net/UDPClient.h

@@ -86,7 +86,7 @@ private:
 	DatagramSocket      _socket;
 	SocketAddress       _address;
 	Thread*             _pThread;
-	bool                _stop;
+	std::atomic<bool>   _stop;
 	Poco::AtomicCounter _dataBacklog;
 	Poco::AtomicCounter _errorBacklog;
 };

Net/include/Poco/Net/UDPHandler.h

@@ -30,6 +30,7 @@
 #include "Poco/StringTokenizer.h"
 #include <deque>
 #include <cstring>
+#include <atomic>
 
 
 namespace Poco {
@@ -44,11 +45,12 @@ template <std::size_t S = POCO_UDP_BUF_SIZE>
 class UDPHandlerImpl: public Runnable, public RefCountedObject
 	/// UDP handler handles the data that arrives to the UDP server.
 	/// The class is thread-safe and runs in its own thread, so many handlers
-	/// can be used in parallel.Handler manages and provides the storage
+	/// can be used in parallel. Handler manages and provides the storage
 	/// (fixed-size memory blocks of S size) to the reader, which signals back
 	/// to the handler when there is data or error ready for processing.
 	/// Typically, user will inherit from this class and override processData()
-	/// and processError() members to do the actual work.
+	/// and processError() members to do the actual work. To auto-start the handler,
+	/// the inheriting class can call start() in the constructor.
 {
 public:
 	typedef UDPMsgSizeT             MsgSizeT;
@@ -76,7 +78,6 @@ public:
 		_pErr(pErr)
 		/// Creates the UDPHandlerImpl.
 	{
-		_thread.start(*this);
 	}
 
 	~UDPHandlerImpl()
@@ -143,9 +144,9 @@ public:
 	}
 
 	void notify()
-		/// Sets the ready event.
+		/// Sets the data ready event.
 	{
-		_ready.set();
+		_dataReady.set();
 	}
 
 	void run()
@@ -153,28 +154,31 @@ public:
 	{
 		while (!_stop)
 		{
-			_ready.wait();
+			_dataReady.wait();
 			if (_stop) break;
 			if (_mutex.tryLock(10))
 			{
-				BufMap::iterator it = _buffers.begin();
-				BufMap::iterator end = _buffers.end();
-				for (; it != end; ++it)
+				if (!_stop)
 				{
-					BufList::iterator lIt = it->second.begin();
-					BufList::iterator lEnd = it->second.end();
-					for (; lIt != lEnd; ++lIt)
+					BufMap::iterator it = _buffers.begin();
+					BufMap::iterator end = _buffers.end();
+					for (; it != end; ++it)
 					{
-						if (hasData(*lIt))
+						BufList::iterator lIt = it->second.begin();
+						BufList::iterator lEnd = it->second.end();
+						for (; lIt != lEnd; ++lIt)
 						{
-							processData(*lIt);
-							--_dataBacklog;
-							setIdle(*lIt);
-						}
-						else if (isError(*lIt))
-						{
-							processError(*lIt);
-							++_errorBacklog;
+							if (hasData(*lIt))
+							{
+								processData(*lIt);
+								--_dataBacklog;
+								setIdle(*lIt);
+							}
+							else if (isError(*lIt))
+							{
+								processError(*lIt);
+								++_errorBacklog;
+							}
 						}
 					}
 				}
@@ -188,7 +192,7 @@ public:
 		/// Signals the handler to stop.
 	{
 		_stop = true;
-		_ready.set();
+		_dataReady.set();
 	}
 
 	bool stopped() const
@@ -329,6 +333,12 @@ public:
 		setIdle(buf);
 	}
 
+	void start()
+		/// Stars the handler run in thread.
+	{
+		_thread.start(*this);
+	}
+
 private:
 	typedef std::deque<char*>                BufList;
 	typedef std::map<poco_socket_t, BufList> BufMap;
@@ -355,10 +365,10 @@ private:
 		*ret = _buffers[sock].back();
 	}
 
-	Poco::Event       _ready;
+	Poco::Event       _dataReady;
 	Poco::Thread      _thread;
-	bool              _stop;
-	bool              _done;
+	std::atomic<bool> _stop;
+	std::atomic<bool> _done;
 	BufMap            _buffers;
 	BufIt             _bufIt;
 	std::size_t       _bufListSize;

Net/include/Poco/Net/UDPServer.h

@@ -90,9 +90,9 @@ public:
 	}
 
 private:
-	P            _poller;
+	P _poller;
 	Poco::Thread _thread;
-	bool         _stop;
+	std::atomic<bool> _stop;
 };
 
 

Net/src/RemoteSyslogListener.cpp

@@ -96,7 +96,7 @@ public:
 private:
 	Poco::NotificationQueue& _queue;
 	DatagramSocket           _socket;
-	bool                     _stopped;
+	std::atomic<bool>        _stopped;
 };
 
 
@@ -190,7 +190,7 @@ private:
 
 private:
 	Poco::NotificationQueue& _queue;
-	bool                     _stopped;
+	std::atomic<bool>        _stopped;
 	RemoteSyslogListener*    _pListener;
 };
 

Net/testsuite/src/DialogServer.h

@@ -70,11 +70,11 @@ private:
 	Poco::Thread             _thread;
 	Poco::Event              _ready;
 	mutable Poco::FastMutex  _mutex;
-	bool                     _stop;
+	std::atomic<bool>        _stop;
 	std::vector<std::string> _nextResponses;
 	std::vector<std::string> _lastCommands;
-	bool                     _acceptCommands;
-	bool                     _log;
+	std::atomic<bool>        _acceptCommands;
+	std::atomic<bool>        _log;
 };
 
 

Net/testsuite/src/EchoServer.h

@@ -18,6 +18,7 @@
 #include "Poco/Net/ServerSocket.h"
 #include "Poco/Thread.h"
 #include "Poco/Event.h"
+#include <atomic>
 
 
 class EchoServer: public Poco::Runnable
@@ -50,8 +51,8 @@ private:
 	Poco::Net::ServerSocket _socket;
 	Poco::Thread _thread;
 	Poco::Event  _ready;
-	bool         _stop;
-	bool         _done;
+	std::atomic<bool> _stop;
+	std::atomic<bool> _done;
 };
 
 

Net/testsuite/src/HTTPTestServer.h

@@ -51,7 +51,7 @@ private:
 	Poco::Net::ServerSocket  _socket;
 	Poco::Thread _thread;
 	Poco::Event  _ready;
-	bool         _stop;
+	std::atomic<bool> _stop;
 	std::string  _lastRequest;
 };
 

Net/testsuite/src/MulticastEchoServer.h

@@ -61,7 +61,7 @@ private:
 	Poco::Net::NetworkInterface _if;
 	Poco::Thread _thread;
 	Poco::Event  _ready;
-	bool         _stop;
+	std::atomic<bool> _stop;
 };
 
 

Net/testsuite/src/PollSetTest.cpp

@@ -38,7 +38,8 @@ class Poller : public Poco::Runnable
 {
 public:
 	Poller(PollSet& pollSet, const Timespan& timeout): _pollSet(pollSet),
-		_timeout(timeout)
+		_timeout(timeout),
+		_running(false)
 	{
 	}
 
@@ -57,7 +58,7 @@ public:
 private:
 	PollSet& _pollSet;
 	Timespan _timeout;
-	bool _running = false;
+	std::atomic<bool> _running;
 };
 
 

Net/testsuite/src/SocketProactorTest.cpp

@@ -56,7 +56,7 @@ void SocketProactorTest::testTCPSocketProactor()
 	};
 	proactor.addSend(s, SocketProactor::Buffer(hello.begin(), hello.end()), onSendCompletion);
 	SocketProactor::Buffer buf(hello.size(), 0);
-	bool received = false, receivePassed = false;
+	std::atomic<bool> received(false), receivePassed(false);
 	auto onRecvCompletion = [&](std::error_code err, int bytes)
 	{
 		receivePassed = (err.value() == 0) &&
@@ -107,8 +107,8 @@ void SocketProactorTest::testTCPSocketProactor()
 	assertFalse (received);
 	assertFalse (receivePassed);
 
-	bool error = false;
-	bool errorPassed = false;
+	std::atomic<bool> error(false);
+	std::atomic<bool> errorPassed(false);
 	auto onError = [&](std::error_code err, int bytes)
 	{
 		errorPassed = (err.value() != 0) && (bytes == 0);
@@ -151,7 +151,7 @@ void SocketProactorTest::testUDPSocketProactor()
 		SocketAddress("127.0.0.1", echoServer.port()),
 		onSendCompletion);
 	Poco::Net::SocketProactor::Buffer buf(hello.size(), 0);
-	bool received = false, receivePassed = false;
+	std::atomic<bool> received(false), receivePassed(false);
 	SocketAddress sa;
 	auto onRecvCompletion = [&](std::error_code err, int bytes)
 	{
@@ -229,7 +229,7 @@ void SocketProactorTest::testSocketProactorStartStop()
 		SocketAddress("127.0.0.1", echoServer.port()),
 		onSendCompletion);
 	Poco::Net::SocketProactor::Buffer buf(hello.size(), 0);
-	bool received = false, receivePassed = false;
+	std::atomic<bool> received(false), receivePassed(false);
 	SocketAddress sa;
 	auto onRecvCompletion = [&](std::error_code err, int bytes)
 	{

Net/testsuite/src/SyslogTest.cpp

@@ -63,6 +63,7 @@ std::size_t CachingChannel::getMaxSize() const
 
 std::size_t CachingChannel::getCurrentSize() const
 {
+	Poco::FastMutex::ScopedLock lock(_mutex);
 	return _size;
 }
 

Net/testsuite/src/UDPEchoServer.h

@@ -49,7 +49,7 @@ private:
 	Poco::Net::DatagramSocket _socket;
 	Poco::Thread _thread;
 	Poco::Event  _ready;
-	bool         _stop;
+	std::atomic<bool> _stop;
 };
 
 

Net/testsuite/src/UDPServerTest.cpp

@@ -45,7 +45,10 @@ namespace
 {
 	struct TestUDPHandler : public Poco::Net::UDPHandler
 	{
-		TestUDPHandler() : counter(0), errCounter(0) {}
+		TestUDPHandler() : counter(0), errCounter(0)
+		{
+			start();
+		}
 
 		void processData(char *buf)
 		{