Improve FifoEvent, ActiveMethod, ActiveResult (#4211)

Co-authored-by: Alexander B <[email protected]>

Foundation/Foundation_vs140.vcxproj

@@ -1385,6 +1385,7 @@
     <ClCompile Include="src\Thread.cpp" />
     <ClCompile Include="src\ThreadLocal.cpp" />
     <ClCompile Include="src\ThreadPool.cpp" />
+    <ClCompile Include="src\ActiveThreadPool.cpp" />
     <ClCompile Include="src\ThreadTarget.cpp" />
     <ClCompile Include="src\Thread_POSIX.cpp">
       <ExcludedFromBuild Condition="'$(Configuration)|$(Platform)'=='debug_shared|Win32'">true</ExcludedFromBuild>

Foundation/Foundation_vs150.vcxproj

@@ -1385,6 +1385,7 @@
     <ClCompile Include="src\Thread.cpp" />
     <ClCompile Include="src\ThreadLocal.cpp" />
     <ClCompile Include="src\ThreadPool.cpp" />
+    <ClCompile Include="src\ActiveThreadPool.cpp" />
     <ClCompile Include="src\ThreadTarget.cpp" />
     <ClCompile Include="src\Thread_POSIX.cpp">
       <ExcludedFromBuild Condition="'$(Configuration)|$(Platform)'=='debug_shared|Win32'">true</ExcludedFromBuild>

Foundation/Foundation_vs160.vcxproj

@@ -1391,6 +1391,7 @@
     <ClCompile Include="src\Thread.cpp" />
     <ClCompile Include="src\ThreadLocal.cpp" />
     <ClCompile Include="src\ThreadPool.cpp" />
+    <ClCompile Include="src\ActiveThreadPool.cpp" />
     <ClCompile Include="src\ThreadTarget.cpp" />
     <ClCompile Include="src\Thread_POSIX.cpp">
       <ExcludedFromBuild Condition="'$(Configuration)|$(Platform)'=='debug_shared|Win32'">true</ExcludedFromBuild>

Foundation/Foundation_vs170.vcxproj

@@ -1950,6 +1950,7 @@
     <ClCompile Include="src\Thread.cpp" />
     <ClCompile Include="src\ThreadLocal.cpp" />
     <ClCompile Include="src\ThreadPool.cpp" />
+    <ClCompile Include="src\ActiveThreadPool.cpp" />
     <ClCompile Include="src\ThreadTarget.cpp" />
     <ClCompile Include="src\Thread_POSIX.cpp">
       <ExcludedFromBuild Condition="'$(Configuration)|$(Platform)'=='debug_shared|Win32'">true</ExcludedFromBuild>

Foundation/Makefile

@@ -6,7 +6,7 @@
 
 include $(POCO_BASE)/build/rules/global
 
-objects = ArchiveStrategy Ascii ASCIIEncoding AsyncChannel \
+objects = ArchiveStrategy Ascii ASCIIEncoding AsyncChannel ActiveThreadPool\
 	Base32Decoder Base32Encoder Base64Decoder Base64Encoder \
 	BinaryReader BinaryWriter Bugcheck ByteOrder Channel Checksum Clock Configurable ConsoleChannel \
 	Condition CountingStream DateTime LocalDateTime DateTimeFormat DateTimeFormatter DateTimeParser \

Foundation/include/Poco/AbstractEvent.h

@@ -339,9 +339,8 @@ class AbstractEvent
 		}
 
 		NotifyAsyncParams params = par;
-		TArgs retArgs(params.args);
-		params.ptrStrat->notify(params.pSender, retArgs);
-		return retArgs;
+		params.ptrStrat->notify(params.pSender, params.args);
+		return params.args;
 	}
 
 	TStrategy _strategy; /// The strategy used to notify observers.

Foundation/include/Poco/ActiveStarter.h

@@ -19,7 +19,7 @@
 
 
 #include "Poco/Foundation.h"
-#include "Poco/ThreadPool.h"
+#include "Poco/ActiveThreadPool.h"
 #include "Poco/ActiveRunnable.h"
 
 
@@ -36,7 +36,7 @@ class ActiveStarter
 public:
 	static void start(OwnerType* /*pOwner*/, ActiveRunnableBase::Ptr pRunnable)
 	{
-		ThreadPool::defaultPool().start(*pRunnable);
+		ActiveThreadPool::defaultPool().start(*pRunnable);
 		pRunnable->duplicate(); // The runnable will release itself.
 	}
 };

Foundation/include/Poco/ActiveThreadPool.h

@@ -0,0 +1,145 @@
+//
+// ActiveThreadPool.h
+//
+// Library: Foundation
+// Package: Threading
+// Module:  ActiveThreadPool
+//
+// Definition of the ActiveThreadPool class.
+//
+// Copyright (c) 2004-2006, Applied Informatics Software Engineering GmbH.
+// and Contributors.
+//
+// SPDX-License-Identifier:	BSL-1.0
+//
+
+
+#ifndef Foundation_ActiveThreadPool_INCLUDED
+#define Foundation_ActiveThreadPool_INCLUDED
+
+
+#include "Poco/Foundation.h"
+#include "Poco/Thread.h"
+#include "Poco/Mutex.h"
+#include "Poco/Environment.h"
+#include <vector>
+
+
+namespace Poco {
+
+
+class Runnable;
+class ActiveThread;
+
+
+class Foundation_API ActiveThreadPool
+	/// A thread pool always keeps a number of threads running, ready
+	/// to accept work.
+	/// Threads in an active thread pool are re-used
+	/// Every thread in the pool has own notification-queue with Runnable
+	/// Every Runnable executes on next thread (round-robin model)
+	/// The thread pool always keeps fixed number of threads running.
+	/// Use case for this pool is running many (more than os-max-thread-count) short live tasks
+	/// Round-robin model allow efficiently utilize cpu cores
+{
+public:
+	ActiveThreadPool(int capacity = static_cast<int>(Environment::processorCount()) + 1,
+		int stackSize = POCO_THREAD_STACK_SIZE);
+		/// Creates a thread pool with fixed capacity threads.
+		/// Threads are created with given stack size.
+
+	ActiveThreadPool(std::string  name,
+		int capacity = static_cast<int>(Environment::processorCount()) + 1,
+		int stackSize = POCO_THREAD_STACK_SIZE);
+		/// Creates a thread pool with the given name and fixed capacity threads.
+		/// Threads are created with given stack size.
+
+	~ActiveThreadPool();
+		/// Currently running threads will remain active
+		/// until they complete.
+
+	int capacity() const;
+		/// Returns the capacity of threads.
+
+	int getStackSize() const;
+		/// Returns the stack size used to create new threads.
+
+	void start(Runnable& target);
+		/// Obtains a thread and starts the target.
+
+	void start(Runnable& target, const std::string& name);
+		/// Obtains a thread and starts the target.
+		/// Assigns the given name to the thread.
+
+	void startWithPriority(Thread::Priority priority, Runnable& target);
+		/// Obtains a thread, adjusts the thread's priority, and starts the target.
+
+	void startWithPriority(Thread::Priority priority, Runnable& target, const std::string& name);
+		/// Obtains a thread, adjusts the thread's priority, and starts the target.
+		/// Assigns the given name to the thread.
+
+	void stopAll();
+		/// Stops all running threads and waits for their completion.
+		///
+		/// Will also delete all thread objects.
+		/// If used, this method should be the last action before
+		/// the thread pool is deleted.
+		///
+		/// Note: If a thread fails to stop within 10 seconds
+		/// (due to a programming error, for example), the
+		/// underlying thread object will not be deleted and
+		/// this method will return anyway. This allows for a
+		/// more or less graceful shutdown in case of a misbehaving
+		/// thread.
+
+	void joinAll();
+		/// Waits for all threads to complete.
+		///
+		/// Note that this will join() underlying
+		/// threads and restart them for next tasks.
+
+	const std::string& name() const;
+		/// Returns the name of the thread pool,
+		/// or an empty string if no name has been
+		/// specified in the constructor.
+
+	static ActiveThreadPool& defaultPool();
+		/// Returns a reference to the default
+		/// thread pool.
+
+protected:
+	ActiveThread* getThread();
+	ActiveThread* createThread();
+
+private:
+	ActiveThreadPool(const ActiveThreadPool& pool);
+	ActiveThreadPool& operator = (const ActiveThreadPool& pool);
+
+	typedef std::vector<ActiveThread*> ThreadVec;
+
+	std::string _name;
+	int _capacity;
+	int _serial;
+	int _stackSize;
+	ThreadVec _threads;
+	mutable FastMutex _mutex;
+	std::atomic<size_t> _lastThreadIndex{0};
+};
+
+
+inline int ActiveThreadPool::getStackSize() const
+{
+	return _stackSize;
+}
+
+
+inline const std::string& ActiveThreadPool::name() const
+{
+	return _name;
+}
+
+
+} // namespace Poco
+
+
+#endif // Foundation_ActiveThreadPool_INCLUDED

Foundation/include/Poco/DefaultStrategy.h

@@ -147,6 +147,11 @@ class DefaultStrategy<void,TDelegate>: public NotificationStrategy<void, TDelega
 	{
 	}
 
+	DefaultStrategy(DefaultStrategy&& s):
+		_delegates(std::move(s._delegates))
+	{
+	}
+
 	~DefaultStrategy()
 	{
 	}
@@ -201,6 +206,15 @@ class DefaultStrategy<void,TDelegate>: public NotificationStrategy<void, TDelega
 		return *this;
 	}
 
+	DefaultStrategy& operator = (DefaultStrategy&& s)
+	{
+		if (this != &s)
+		{
+			_delegates = std::move(s._delegates);
+		}
+		return *this;
+	}
+
 	void clear()
 	{
 		for (Iterator it = _delegates.begin(); it != _delegates.end(); ++it)

Foundation/include/Poco/FIFOStrategy.h

@@ -41,6 +41,11 @@ class FIFOStrategy: public DefaultStrategy<TArgs, TDelegate>
 	{
 	}
 
+	FIFOStrategy(FIFOStrategy&& s):
+		DefaultStrategy<TArgs, TDelegate>(std::move(s))
+	{
+	}
+
 	~FIFOStrategy()
 	{
 	}
@@ -50,6 +55,12 @@ class FIFOStrategy: public DefaultStrategy<TArgs, TDelegate>
 		DefaultStrategy<TArgs, TDelegate>::operator = (s);
 		return *this;
 	}
+
+	FIFOStrategy& operator = (FIFOStrategy&& s)
+	{
+		DefaultStrategy<TArgs, TDelegate>::operator = (s);
+		return *this;
+	}
 };