5150 reactor sock remove (#5151)

* fix(NotificationCenter): use RWLock to prevent lock-order-inversion #5150

Change NotificationCenter from Mutex to RWLock to fix TSAN lock-order-inversion
warnings when notification handlers call back into the notification center
(e.g., hasEventHandler() from within a SocketReactor handler).

Key changes:
- Replace Mutex with RWLock for better read concurrency
- Use read locks for hasObserver(), observersToNotify(), hasObservers(),
  countObservers(), backlog()
- Use write locks for addObserver(), removeObserver(), clear()
- observersToNotify() now copies observer list and releases lock before
  calling accepts() to avoid lock cycle with NObserver's internal mutex
- Add clear() method to atomically disable and remove all observers
- Update AsyncNotificationCenter to use RWLock::ScopedLock

Also fixes AsyncNotificationCenter shutdown to properly signal the dequeue
thread via ShutdownNotification instead of relying only on wakeUpAll() #5058

* fix(Net): add SocketReactor::remove() for safe socket cleanup #5150

Add SocketReactor::remove() method that atomically removes a socket
from the poll set and disables all its event handlers. This prevents
race conditions when removing handlers in destructors, where events
could still be dispatched to handlers while other handlers for the
same socket are being removed.

Changes:
- Add SocketReactor::remove(socket) for atomic socket removal
- Add SocketNotifier::disableObservers() to disable all handlers
- Add SocketNotifier::socket() getter
- Skip dispatch for removed sockets in dispatch() methods
- Update tests to use remove() before removeEventHandler()
- Add testSocketReactorRemove() test

* fix(Foundation): improve ProcessRunner robustness #5054

- Fix race condition: set _pPH after _pid to ensure pid() returns
  valid value when running() returns true
- Add upfront invalid PID validation before termination attempt
- Escalate to Process::kill() if requestTermination times out,
  with its own timeout before throwing
- Try to remove stale PID file on timeout instead of throwing
- Wrap start() wait logic in try/catch to kill orphan process
  and reset state if startup fails
- Clear _error under lock after successful stop to prevent
  stale error from affecting next start()

* Update Net/include/Poco/Net/SocketReactor.h

Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com>

* Update Foundation/src/AsyncNotificationCenter.cpp

Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com>

* Update Foundation/include/Poco/AsyncNotificationCenter.h

Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com>

* Update Foundation/src/AsyncNotificationCenter.cpp

Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com>

* chore: add TaskManager::testCancel to win ignore list

* chore: few improvements in NotificationCenter and ODBC cmake.

---------

Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com>
Co-authored-by: Matej Kenda <matejken@gmail.com>

Author: 3 people

Data/ODBC/CMakeLists.txt

@@ -54,7 +54,7 @@ find_file(
 if (_msodbc_h)
 	get_filename_component(MSODBC_DIR ${_msodbc_h} DIRECTORY CACHE)
 	message(STATUS "ODBC: Found msodbcsql.h in: ${MSODBC_DIR}")
-	target_include_directories(DataODBC PUBLIC "${MSODBC_DIR}")
+	target_include_directories(DataODBC SYSTEM PRIVATE "${MSODBC_DIR}")
 
 	if (ENABLE_DATA_SQL_SERVER_BIG_STRINGS)
 		target_compile_definitions(DataODBC PUBLIC POCO_DATA_SQL_SERVER_BIG_STRINGS=1)

Foundation/include/Poco/AsyncNotificationCenter.h

@@ -151,6 +151,11 @@ class Foundation_API AsyncNotificationCenter: public NotificationCenter
 
 	using Adapter = RunnableAdapter<AsyncNotificationCenter>;
 
+	class ShutdownNotification: public Notification
+		/// Internal notification used to signal the dequeue loop to stop.
+	{
+	};
+
 	const AsyncMode _mode { AsyncMode::ENQUEUE };
 
 	// Async enqueue for notifications

Foundation/include/Poco/NotificationCenter.h

@@ -20,7 +20,7 @@
 
 #include "Poco/Foundation.h"
 #include "Poco/Notification.h"
-#include "Poco/Mutex.h"
+#include "Poco/RWLock.h"
 #include "Poco/SharedPtr.h"
 #include "Poco/Observer.h"
 #include "Poco/NObserver.h"
@@ -186,11 +186,18 @@ class Foundation_API NotificationCenter
 		/// Returns a reference to the default
 		/// NotificationCenter.
 
+	void clear();
+		/// Disables and removes all observers.
+		///
+		/// This can be called to ensure that no more
+		/// notifications will be dispatched to observers,
+		/// even if they are currently being dispatched.
+
 protected:
 	using AbstractObserverPtr = SharedPtr<AbstractObserver>;
 	using ObserverList = std::vector<AbstractObserverPtr>;
 
-	Mutex& mutex()
+	RWLock& mutex()
 	{
 		return _mutex;
 	}
@@ -200,8 +207,8 @@ class Foundation_API NotificationCenter
 
 private:
 
-	ObserverList  _observers;
-	mutable Mutex _mutex;
+	ObserverList   _observers;
+	mutable RWLock _mutex;
 };
 } // namespace Poco
 

Foundation/src/AsyncNotificationCenter.cpp

@@ -16,6 +16,7 @@
 #include "Poco/AsyncNotificationCenter.h"
 
 #include "Poco/AbstractObserver.h"
+#include "Poco/RWLock.h"
 #include "Poco/Stopwatch.h"
 #include "Poco/Debugger.h"
 #include "Poco/ErrorHandler.h"
@@ -155,7 +156,7 @@ void AsyncNotificationCenter::notifyObservers(Notification::Ptr& pNotification)
 
 void AsyncNotificationCenter::start()
 {
-	Poco::ScopedLock l(mutex());
+	RWLock::ScopedLock l(mutex());
 
 	if (_mode == AsyncMode::ENQUEUE || _mode == AsyncMode::BOTH)
 	{
@@ -180,8 +181,8 @@ void AsyncNotificationCenter::start()
 
 	if (_mode == AsyncMode::NOTIFY || _mode == AsyncMode::BOTH)
 	{
-		auto dispatch = [this](std::stop_token stopToken, std::size_t id) {
-			this->dispatchNotifications(stopToken, id);
+		auto dispatch = [this](std::stop_token stopToken, std::size_t workerId) {
+			this->dispatchNotifications(stopToken, workerId);
 		};
 
 		for (std::size_t i {0}; i < _workersCount; ++i)
@@ -200,6 +201,7 @@ void AsyncNotificationCenter::stop()
 	{
 		if (_enqueueThreadStarted.exchange(false))
 		{
+			_nq.enqueueUrgentNotification(new ShutdownNotification);
 			_nq.wakeUpAll();
 			while (!_enqueueThreadDone) Thread::sleep(100);
 			_enqueueThread.join();
@@ -224,8 +226,11 @@ void AsyncNotificationCenter::dequeue()
 	Notification::Ptr pNf;
 	_enqueueThreadStarted = true;
 	_enqueueThreadDone = false;
-	while ((pNf = _nq.waitDequeueNotification()))
+	while (true)
 	{
+		pNf = _nq.waitDequeueNotification();
+		if (!pNf) break;
+		if (pNf.cast<ShutdownNotification>()) break;
 		try
 		{
 			notifyObservers(pNf);

Foundation/src/NotificationCenter.cpp

@@ -14,29 +14,20 @@
 
 #include "Poco/NotificationCenter.h"
 #include "Poco/Notification.h"
-#include "Poco/Observer.h"
-#include "Poco/AutoPtr.h"
+#include "Poco/RWLock.h"
 
 
 namespace Poco {
 
 
-NotificationCenter::NotificationCenter()
-{
-}
+NotificationCenter::NotificationCenter() = default;
 
 
 NotificationCenter::~NotificationCenter()
 {
 	try
 	{
-		ObserverList observersToDisable;
-		{
-			Mutex::ScopedLock lock(_mutex);
-			observersToDisable = std::move(_observers);
-		}
-		for (auto& o: observersToDisable)
-			o->disable();
+		clear();
 	}
 	catch(...)
 	{
@@ -47,35 +38,30 @@ NotificationCenter::~NotificationCenter()
 
 void NotificationCenter::addObserver(const AbstractObserver& observer)
 {
-	Mutex::ScopedLock lock(_mutex);
-	_observers.push_back(observer.clone());
+	RWLock::ScopedLock lock(_mutex);
+	_observers.emplace_back(observer.clone());
 	_observers.back()->start();
 }
 
 
 void NotificationCenter::removeObserver(const AbstractObserver& observer)
 {
-	AbstractObserverPtr pObserver;
+	RWLock::ScopedLock lock(_mutex);
+	for (auto it = _observers.begin(); it != _observers.end(); ++it)
 	{
-		Mutex::ScopedLock lock(_mutex);
-		for (auto it = _observers.begin(); it != _observers.end(); ++it)
+		if (observer.equals(**it))
 		{
-			if (observer.equals(**it))
-			{
-				pObserver = *it;
-				_observers.erase(it);
-				break;
-			}
+			(*it)->disable();
+			_observers.erase(it);
+			return;
 		}
 	}
-	if (pObserver)
-		pObserver->disable();
 }
 
 
 bool NotificationCenter::hasObserver(const AbstractObserver& observer) const
 {
-	Mutex::ScopedLock lock(_mutex);
+	RWLock::ScopedReadLock lock(_mutex);
 	for (const auto& p: _observers)
 		if (observer.equals(*p)) return true;
 
@@ -85,9 +71,14 @@ bool NotificationCenter::hasObserver(const AbstractObserver& observer) const
 
 NotificationCenter::ObserverList NotificationCenter::observersToNotify(const Notification::Ptr& pNotification) const
 {
+	ObserverList observers;
+	{
+		RWLock::ScopedReadLock lock(_mutex);
+		observers = _observers;
+	}
+	// Filter outside the lock to avoid lock-order-inversion with NObserver mutex
 	ObserverList ret;
-	ScopedLock<Mutex> lock(_mutex);
-	for (auto& o : _observers)
+	for (auto& o : observers)
 	{
 		if (o->accepts(pNotification))
 			ret.push_back(o);
@@ -116,15 +107,15 @@ void NotificationCenter::notifyObservers(Notification::Ptr& pNotification)
 
 bool NotificationCenter::hasObservers() const
 {
-	Mutex::ScopedLock lock(_mutex);
+	RWLock::ScopedReadLock lock(_mutex);
 
 	return !_observers.empty();
 }
 
 
 std::size_t NotificationCenter::countObservers() const
 {
-	Mutex::ScopedLock lock(_mutex);
+	RWLock::ScopedReadLock lock(_mutex);
 
 	return _observers.size();
 }
@@ -134,10 +125,10 @@ int NotificationCenter::backlog() const
 {
 	int cnt = 0;
 
-	ScopedLockWithUnlock<Mutex> lock(_mutex);
+	_mutex.readLock();
 	ObserverList observersToCount(_observers);
-	lock.unlock();
-	for (auto& p : observersToCount)
+	_mutex.unlock();
+	for (const auto& p : observersToCount)
 		cnt += p->backlog();
 
 	return cnt;
@@ -151,4 +142,13 @@ NotificationCenter& NotificationCenter::defaultCenter()
 }
 
 
+void NotificationCenter::clear()
+{
+	RWLock::ScopedLock lock(_mutex);
+	for (auto& o: _observers)
+		o->disable();
+	_observers.clear();
+}
+
+
 } // namespace Poco