test/actoraccept/main.cpp - main - Ant010ff/ffpp

test/actoraccept/main.cpp
379 строк16 KB
Ant010ff
Demo code update.
27 июн 2026, 18:24
27 июн 2026, 18:24f820ccd
#include "../common/common.hpp"

//By default broker is asynchronous using asynchronous task dispatch method.
struct Policy : CommonPolicy { 
	//using LockableT = ffpp::Lockable<std::shared_mutex>;
	//using LockableT = ffpp::Lockable<ffpp::TicketMutex<ffpp::MutexFlags::Sleep>>;
	//using LockableT = ffpp::Lockable<ffpp::StripedTicketMutex<ffpp::MutexFlags::Sleep | ffpp::MutexFlags::CheckRecursion>, true>;
	//using LockableT = ffpp::Lockable<ffpp::StripedTicketMutex<ffpp::MutexFlags::CheckRecursion>, true>;
	//using LockableT = ffpp::Lockable<ffpp::StripedTicketMutex<ffpp::MutexFlags::CheckRecursion>>;
	//using LockableT = ffpp::Lockable<ffpp::StripedTicketMutex<ffpp::MutexFlags::Sleep | ffpp::MutexFlags::CheckRecursion>>;
	//using LockableT = ffpp::Lockable<ffpp::StripedTicketMutex<ffpp::MutexFlags::Sleep>>;
	using LockableT = ffpp::Lockable<ffpp::StripedTicketMutex<>>;
	using ThreadLockableT = LockableT;
	using TableLockableT = LockableT;

	struct QueuePolicy : ffpp::RingQueuePolicy {
		//Using scalable r/w mutex to protect dynamic ring-queues. Exclusive lock is required only when internal buffer reallocation
		//is necessary. In all the rest cases enqueue/dequeue operations are performed concurrently under shared lock.
		struct MutexPolicy : ffpp::StripedTicketMutexPolicy {
			static size_t SharedSlots() { 
				//According to AI-provided theoretical estimation 8 - 24 concurrent threads can be considered theoretical scalability 
				//limit of a Vyukov Bounded MPMC Queue as of 2025.
				return std::clamp(size_t(std::thread::hardware_concurrency() / 4), size_t(1), size_t(16)); 
			}
		};
		using MutexT = ffpp::StripedTicketMutex<ffpp::MutexFlags::Default, MutexPolicy>;
		//using MutexT = ffpp::StripedTicketMutex<ffpp::MutexFlags::Sleep | ffpp::MutexFlags::CheckRecursion, MutexPolicy>;
		//using MutexT = ffpp::StripedTicketMutex<ffpp::MutexFlags::Sleep, MutexPolicy>;
		using LockableT = ffpp::Lockable<MutexT>;

		//static constexpr auto QueueMode() { return ffpp::QueueFlags::Dynamic; }

		// static bool OnCapacity(size_t nCurrent, size_t nRequested) {
		// 	Trace::Put("Ring-queue capacity changing:", nCurrent, "->", nRequested);
		// 	return true;
		// }
	};
	using QueuePolicyT = QueuePolicy;

	//DoubleQueueing enables internal swappable queues for any ffpp::FunctionalQueue-derived class. This means that each internal 
	//subqueue (identified according to corresponding task priority) contains two atomically swappable segments (instancies of a
	//given queue type QueueT). Swapping is controlled by task processing logic of consumer. It is performed only when currently 
	//processing subqueue back-segment becomes empty while providers continues to enqueue tasks to subqueues' front-segments. 
	//Double-queueing can lower overall task processing latency but doubles memory consumption for the same application task.
	//static constexpr bool DoubleQueueing() { return true; }

	//Contention tracking enables contention level estimation which is the maximum registered number of concurrent Enqueue calls
	//(any task/message emission) on ffpp::FunctionalSequence-derived object (by instance). That is any actor (ffpp::Actor), 
	//thread (ffpp::FunctionalQueueThread) or thread inside thread pool (ffpp::FunctionalQueuePool) either user defined or internal 
	//broker's pool (ffpp::FunctionalBroker). If enabled this can be used by thread pool for more precise scheduling in case of
	//ffpp::DispatchFlags::Optimal mode. This is also used to provide metric of pool's internal contention pressure as a sum 
	//of contention levels of its processing threads (see ffpp::FunctionalQueuePool::ForPoolMetrics).
	static constexpr bool TrackContention() { return true; }

	//Synchronous broker can be used as actors' environment and generic event bus because it provides minimal dispatch latency 
	//(FunctionalBroker::Accepts API by default uses direct synchronous message emission). Note that function chaining and asynchronous 
	//finalization are unavailable with synchronous broker. However synchronous broker still contains valid thread pool (initially 
	//inactive) for convenience though doesn't use it for own needs. If it is necessary thread pool can be disabled with EnablePool()
	//policy method.
	static constexpr bool Asynchronous() { return false; }
	
	static constexpr ffpp::DispatchFlags DispatchMethod() { 
		//return ffpp::DispatchFlags::Elastic; 
		//return ffpp::DispatchFlags::Adaptive; 
		return 
			ffpp::DispatchFlags::Undefined 
			| ffpp::DispatchFlags::Static 
			| ffpp::DispatchFlags::Affinity
			| ffpp::DispatchFlags::Locality
			//| ffpp::DispatchFlags::Sparse 
			//| ffpp::DispatchFlags::Hypercube 
			//| ffpp::DispatchFlags::Reshuffle
			//| ffpp::DispatchFlags::RandomL2
			| ffpp::DispatchFlags::Optimal 
			| ffpp::DispatchFlags::IdleQueue
			| ffpp::DispatchFlags::Resteal 
			//| ffpp::DispatchFlags::WideSteal
		; 
	}

	static constexpr ffpp::SchedulingFlags SchedulingMethod() { 
		//return ffpp::SchedulingFlags::Immediate; 
		return ffpp::SchedulingFlags::Deferred; 
		//return ffpp::SchedulingFlags::Deferred | ffpp::SchedulingFlags::Bind; 
		//return ffpp::SchedulingFlags::Offload; 
	}

	//When priority (sub-queue) list is known in design-time and not supposed to change at runtime (usually the case, but depends 
	//on actual task) it can be defined in the policy to enable FunctionalQueue's fixed priority mode (all sub-queues are created 
	//on instance construction and can not be changed later) which increases performance by avoiding FunctionalQueue's internal 
	//mutex lock protecting sub-queue collection. Zero is a default priority (sub-queue id) value.
	static constexpr auto Queues() { return std::array { 0, 1 }; }

	// static size_t ThreadGroupSize(size_t nThreads) noexcept { 
	// 	return 0;
	// 	//return ffpp::math::Sqrt(nThreads);
	// }

	static bool SetThreadAffinity(ffpp::ThreadHandleT hThread, size_t iThread) {
		//Default: bind each pool's thread to a single hardware thread selected with a round-robin approach (if number of pool 
		//threads is greater than number of hardware threads available).
		return ffpp::SetThreadHardwareAffinity(hThread, iThread);

		//size_t ieThread = ffpp::MapThreadIndex(iThread) & ~size_t(1), ioThread = ieThread + 1;
	
		//Example 1: allow pool's threads to run only on the lower half of hardware threads.
		//return ffpp::SetThreadHardwareAffinity(hThread, ieThread >> 1);
		
		//Example 2: allow pool's threads to run only on the lower half of even hardware threads.
		//return ffpp::SetThreadHardwareAffinity(hThread, (ieThread >> 1) & ~size_t(1));

		//Example 3: allow each pool's thread to run on two adjacent hardware threads (SMT2?).
		//return ffpp::SetThreadHardwareAffinity(hThread, { ieThread, ioThread });
	}
};

using BrokerT = ffpp::FunctionalBroker<Policy>;
using ClockT = std::chrono::steady_clock;

namespace msg {

	struct Init { 
		size_t nPingees = std::thread::hardware_concurrency(); 
	};
	
	struct Complete { 
		ffpp::SharedGuardT<> sgComplete; 
	};

	struct Ping { 
		BrokerT::Id 
			idPinger;
		ClockT::time_point 
			tpTimestamp = ClockT::now(); 
	};

	struct Pong { 
		BrokerT::Id 
			idPingee;
		ClockT::time_point 
			tpPing = { },
			tpTimestamp = ClockT::now(); 
	};

	struct Stats {
		ClockT::time_point
			tpStart = { };
		uint64_t 
			nPings = 0, 
			nPPS = 0;
	};

}//msg

class Pingee : public ffpp::Actor<Pingee, true, Policy> {
public:
	Pingee(BrokerT::Id idGroup) 
	: Actor(BrokerT::GetPool()) //It is not necessary here to use broker's pool, any one can be used.
	, c_idGroup_(idGroup)
	{ }

	~Pingee() { }

	static BrokerT::Id Create(BrokerT::Id idGroup) {
		auto spInstance = std::make_shared<Pingee>(idGroup);
		//By default when Accept<...> is used to register subscriber broker will keep strong reference to instance been registered.
		//Returned finalizer function is ignored here for demo purposes. To stop processing and release all actor instances the 
		//whole group idGroup will be finalized.
		BrokerT::Accept<msg::Ping>(spInstance->GetInstanceId(), idGroup, spInstance);
		//Registering msg::Complete with higher priority of (1), default is (0).
		BrokerT::Accept<msg::Complete>(spInstance->GetInstanceId(), idGroup, 1, spInstance);
		return spInstance->GetInstanceId();
	}

	BrokerT::Id GetGroupId() const { return c_idGroup_; }
	BrokerT::Id GetInstanceId() const { return c_idInstance_; }

	void On(msg::Ping const& msg) { 
		//Responding to host Pinger instance with msg::Pong message sending it via BrokerT using target instance id.
		BrokerT::Emit(msg.idPinger, msg::Pong { GetInstanceId(), msg.tpTimestamp });
		if constexpr(
			((BrokerT::PolicyT::DispatchMethod() & ffpp::DispatchFlags::Bind) == ffpp::DispatchFlags::Bind)
			|| ((BrokerT::PolicyT::SchedulingMethod() & ffpp::SchedulingFlags::Bind) == ffpp::SchedulingFlags::Bind)
		) {
			if(idProcessor_ == std::thread::id { }) {
				idProcessor_ = std::this_thread::get_id();
			}
			else if(idProcessor_ != std::this_thread::get_id()) {
				ffpp::Except<std::logic_error>::Throw("Invalid processing thread id!");
			}
		}
	}

	void On(msg::Complete const& msg) { 
		Finalize(); 
	}

private:
	BrokerT::Id const
		c_idGroup_ = BrokerT::InvalidId(),
		c_idInstance_ = BrokerT::GenerateId();
	std::thread::id
		idProcessor_;
};//Pingee

class Pinger : public ffpp::Actor<Pinger, true, Policy> {
public:
	Pinger() 
	//It is not necessary here to use broker's pool, any one can be used. As well as it is not necessary to create Pinger and 
	//Pingee instances bound to one and the same pool.
	: Actor(BrokerT::GetPool()) 
	{ }

	~Pinger() { }

	static BrokerT::Id Create() {
		auto spInstance = std::make_shared<Pinger>();
		//By default when Accept<...> is used to register subscriber broker will keep strong reference to instance been registered.
		//Returned finalizer function is ignored here for demo purposes. To stop processing and release all actor instances the 
		//whole group will be finalized.
		BrokerT::Accept<msg::Init, msg::Pong>(spInstance->GetInstanceId(), spInstance->GetGroupId(), spInstance);
		//Registering msg::Complete with higher priority of (1), default is (0).
		BrokerT::Accept<msg::Complete>(spInstance->GetInstanceId(), spInstance->GetGroupId(), 1, spInstance);
		return spInstance->GetInstanceId();
	}

	BrokerT::Id GetGroupId() const { return GetInstanceId(); }
	BrokerT::Id GetInstanceId() const { return c_idInstance_; }

	void On(msg::Init& msg) { 
		Trace::Put("Pinger", GetInstanceId(), "creating", msg.nPingees, "pingee actors.");
		vPingees_.reserve(msg.nPingees);
		while(msg.nPingees-- > 0) vPingees_.push_back(Pingee::Create(GetGroupId()));
		tpStart_ = ClockT::now();
		for(auto id : vPingees_) BrokerT::Emit(id, msg::Ping { GetInstanceId() });
	}

	void On(msg::Complete const& msg) {
		auto const tpStop = ClockT::now();
		Finalize();
		//Sending completion  message to pingees...
		//for(auto id : vPingees_) BrokerT::Emit(id, msg);
		//Or broadcasting msg::Complete as event. This must be done after current Pinger instance finalization because Pinger and 
		//its Pingees belong to the same group. Otherwise Pinger will receive msg::Complete once again and main thread will be blocked
		//because of msg::Complete::sgComplete stalled in its message queue.
		BrokerT::Emit(msg, GetGroupId());
		msg::Stats stats {
			tpStart_,
			nPings_,
			size_t(1000.0 * nPings_ / std::chrono::duration_cast<std::chrono::milliseconds>(tpStop - tpStart_).count())
		};
		Trace::Put(
			"Pinger", GetInstanceId(), "completed. Total pings:", nPings_
			, "(", stats.nPPS, "pps )"
			, ", average latency:", IntervalToString(sumLatency_ / std::max<uint64_t>(nPings_, 1))
			, ", min latency:", IntervalToString(minLatency_)
			, ", max latency:", IntervalToString(maxLatency_)
			, ", average full latency:", IntervalToString(sumFullLatency_ / std::max<uint64_t>(nPings_, 1))
			, ", min full latency:", IntervalToString(minFullLatency_)
			, ", max full latency:", IntervalToString(maxFullLatency_), "."
		);
		BrokerT::Emit(stats);
	}

	void On(msg::Pong const& msg) {
		nPings_++;
		uint64_t const 
			usFullLatency = std::chrono::duration_cast<std::chrono::nanoseconds>(ClockT::now() - msg.tpPing).count(),
			usLatency = std::chrono::duration_cast<std::chrono::nanoseconds>(msg.tpTimestamp - msg.tpPing).count();
		sumLatency_ += usLatency;
		if(usLatency < minLatency_) minLatency_ = usLatency;
		else if(usLatency > maxLatency_) maxLatency_ = usLatency;
		sumFullLatency_ += usFullLatency;
		if(usFullLatency < minFullLatency_) minFullLatency_ = usFullLatency;
		else if(usFullLatency > maxFullLatency_) maxFullLatency_ = usFullLatency;
		BrokerT::Emit(msg.idPingee, msg::Ping { GetInstanceId() });
	}

private:
	BrokerT::Id const
		c_idInstance_ = BrokerT::GenerateId();
	std::vector<BrokerT::Id> 
		vPingees_;
	uint64_t 
		nPings_ = 0,
		sumLatency_ = 0,
		minLatency_ = std::numeric_limits<size_t>::max(),
		maxLatency_ = 0,
		sumFullLatency_ = 0,
		minFullLatency_ = std::numeric_limits<size_t>::max(),
		maxFullLatency_ = 0;
	ClockT::time_point
		tpStart_;
};//Pinger

int main([[maybe_unused]] int argc, [[maybe_unused]] char** argv) {
	FFPP_TRY {
		using namespace std::literals;

		Log<>::Init(
			algy::FilesystemOptions { algy::debug, algy::c_bitDefaultCaps, std::filesystem::path(argv[0]).parent_path() / "log" , "actoraccept" },
			algy::ConsoleOptions { algy::debug, algy::c_bitDefaultCaps }
		);
		Accent::Put("Starting actor accept test...");
		PutRuntimeInfo();
		Info::Put("Actor thread pool direct dispatch:", !bool(Policy::DispatchMethod() & ffpp::DispatchFlags::Indirect));
		Info::Put("Actor fixed thread pool:", bool(Policy::DispatchMethod() & ffpp::DispatchFlags::Static));
		Info::Put("Actor double queueing (actors and thread pool):", Policy::DoubleQueueing());

		size_t
			#if FFPP_ENABLE_DEBUG
				secDuration = 0 
			#else
				secDuration = argc > 1 ? std::stoull(argv[1]) : 30 
			#endif
			, nPingers = argc > 2 ? std::stoull(argv[2]) : std::thread::hardware_concurrency()
			, nPingees = argc > 3 ? std::stoull(argv[3]) : std::thread::hardware_concurrency()
		;
		if(argc > 4) Policy::nPoolLimit = std::stoull(argv[4]);

		BrokerT::Id const gidStats = BrokerT::GenerateId();
		BrokerT::On<msg::Stats>(gidStats, [
			tpStart = ClockT::time_point::max(),
			nPings = uint64_t(0),
			mnPPS = std::numeric_limits<uint64_t>::max(), 
			mxPPS = uint64_t(0), 
			smPPS = uint64_t(0), 
			nPingers, 
			nComplete = 0
		] (auto const& stats) mutable {
			tpStart = std::min(tpStart, stats.tpStart);
			nPings += stats.nPings;
			mnPPS = std::min(mnPPS, stats.nPPS);
			mxPPS = std::max(mxPPS, stats.nPPS);
			smPPS += stats.nPPS;
			if(++nComplete == nPingers) {
				auto const secDuration = std::max<uint64_t>(std::chrono::duration_cast<std::chrono::milliseconds>(ClockT::now() - tpStart).count(), 1) / 1000.0;
				Info::Put("Overall stats:");
				Info::Put("Pinger PPS:", size_t(double(smPPS) / nPingers), "(avg),", mnPPS, "(min),", mxPPS, "(max)");
				Info::Put("Total PPS:", size_t(double(nPings) / secDuration));
			}
		});
		
		//Thread pool state monitoring.
		auto fnl = StartPoolMonitoring<BrokerT>(1s, 1);

		std::vector<BrokerT::Id> vPingers; vPingers.reserve(nPingers);
		Info::Put("Creating", nPingers, "pinger actor(s) with", nPingees, "pingee actor(s) each, running total of", nPingers + nPingers * nPingees, "actors.");
		if(secDuration == 0) Info::Put("Press Enter to exit...");
		else Info::Put("Running test for", secDuration, "sec...");
		while(nPingers-- > 0) vPingers.push_back(Pinger::Create());
		//Initializing pingers - sending msg::Init message:
		for(auto id : vPingers) BrokerT::Emit(id, msg::Init { nPingees });
		//Or broadcasting msg::Init for all instancies accepting it:
		//BrokerT::Emit(msg::Init { nPingees });
		if(secDuration == 0) std::cin.get();
		else std::this_thread::sleep_for(std::chrono::seconds(secDuration));
		BrokerT::GetTimer()->Complete();
		Accent::Put("Finalizing...");
		{
			auto [_, sgComplete] = ffpp::BeginJoin();
			for(auto id : vPingers) BrokerT::Emit(id, msg::Complete { sgComplete });
		}
		//Finalizing groups, including actors' instances and their subscriptions. Pinger's group id equals its instance id.
		for(auto id : vPingers) BrokerT::Finalize(id); 
		//BrokerT::Finalize();
		Accent::Put("Finished.");

		return 0;
	}
	FFPP_CATCH(ffpp::Base::Exception, ex) {
		if(Log<>::IsValid()) Fatal::Put("Top level FFPP exception:", ex);
		return -2;
	}
	FFPP_CATCH(std::exception, ex) {
		if(Log<>::IsValid()) Fatal::Put("Top level exception:", ex);
		return -1;
	}
}//main