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

test/actorping/main.cpp
246 строк9 KB
Ant010ff
Demo code update.
20 июн 2026, 15:13
20 июн 2026, 15:138c3aab0

//#define FFPP_NO_EXCEPTIONS (1)

#include <exception>

/*
namespace FFPP {

	template<typename Tag>
	struct ExceptPolicy;

    template<>
	struct ExceptPolicy<void> {
		template<typename TException, bool t_bSourceLocation>
		[[noreturn]] static void OnExcept(auto const& sl, auto const&... args) {
            //Log error info, perform reasonable cleanup, and **terminate the process**.
			std::terminate();
		}
	};

}
*/

/*
struct GlobalExceptPolicy {
	template<typename TException, bool t_bSourceLocation>
	[[noreturn]] static void OnExcept(auto const& sl, auto const&... args) {
		//Log error info, perform reasonable cleanup, and **terminate the process**.
		std::terminate();
	}
};

#define FFPP_EXCEPT_POLICY_CLASS GlobalExceptPolicy
*/

#include "../common/common.hpp"

struct PoolPolicy : CommonPolicy { 
	//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::vector { 0 }; }

	static constexpr ffpp::DispatchFlags DispatchMethod() { 
		//return ffpp::DispatchFlags::Single; //fastest in this test due to overall simplicity of the task and better cache utilization
		//return ffpp::DispatchFlags::Optimal; 
		//return ffpp::DispatchFlags::Fill; 
		//return ffpp::DispatchFlags::Indirect | ffpp::DispatchFlags::Optimal; 
		//return ffpp::DispatchFlags::Bind | ffpp::DispatchFlags::Optimal; 
		//return ffpp::DispatchFlags::Static | ffpp::DispatchFlags::Optimal | ffpp::DispatchFlags::Resteal; 
		//return ffpp::DispatchFlags::Static | ffpp::DispatchFlags::Random2 | ffpp::DispatchFlags::Resteal; 
		//return ffpp::DispatchFlags::Random2 | ffpp::DispatchFlags::Resteal; 
		return ffpp::DispatchFlags::Optimal | ffpp::DispatchFlags::Resteal; 
	}

	//By instance processing exceptions handling (optional). Can be completely omitted or implemented for exact actor types. Here 
	//a generic handler is used.
	static void OnProcessingException(auto* pActor, std::exception const& ex) { 
		#if !(defined(_GLIBCXX_RELEASE) && (_GLIBCXX_RELEASE <= 12))
			Warn::Put(
				"Actor processing exception:", ex, ", instance:", std::format("{:#018X}", size_t(pActor))
				#if FFPP_TRACK_ORIGIN
				, "@", pActor->GetOrigin()
				#endif
			);
			//Warn::Fmt("Actor processing exception: {}, instance: {:#018X}", std::string(ex.what()), size_t(pActor));
			//pActor->Finalize(); //demo, not necessary
		#else
			Warn::Put(
				"Actor processing exception:", ex
				#if FFPP_TRACK_ORIGIN
				, "@", pActor->GetOrigin()
				#endif
			);
		#endif
	}
};

struct ActorPolicy : CommonPolicy { 
	static constexpr auto Queues() { return std::vector { 0 }; }
	static constexpr ffpp::SchedulingFlags SchedulingMethod() { 
		//return ffpp::SchedulingFlags::Immediate; 
		//return ffpp::SchedulingFlags::Deferred; 
		return ffpp::SchedulingFlags::Offload; 
	}
};

using ClockT = std::chrono::steady_clock;

class Pingee : public ffpp::Actor<Pingee, true, ActorPolicy> {
public:
	using ffpp::Actor<Pingee, true, ActorPolicy>::Actor;

	template<typename TPinger>
	struct MsgPing { 
		std::shared_ptr<TPinger> 
			spPinger;
		size_t 
			iPingee = 0;
		std::chrono::steady_clock::time_point 
			tpTimestamp = std::chrono::steady_clock::now(); 
	};
	
	struct MsgPong { 
		size_t 
			iPingee = 0;
		std::chrono::steady_clock::time_point 
			tpPing = { },
			tpTimestamp = std::chrono::steady_clock::now(); 
	};

	template<typename TPinger>
	void On(MsgPing<TPinger> const& msg) { 
		//Log<>::Put("Actor", msg.iPingee, "got ping.");
		msg.spPinger->Emit(MsgPong { msg.iPingee, msg.tpTimestamp });
		SimulateException("Random pingee exception!");
	}
};//Pingee

class Pinger : public ffpp::Actor<Pinger, true, ActorPolicy> {
public:
	using ffpp::Actor<Pinger, true, ActorPolicy>::Actor;

	template<typename TActorPool>
	struct MsgInit { 
		std::shared_ptr<TActorPool> spActorPool;
		size_t nPingees; 
	};
	
	struct MsgComplete { 
		ffpp::SharedGuardT<> sgComplete; 
	};

	template<typename TActorPool>
	void On(MsgInit<TActorPool>& msg) { 
		Trace::Put("Creating", msg.nPingees, "pingee actors.");
		vPingees_.reserve(msg.nPingees);
		//while(msg.nPingees-- > 0) vPingees_.push_back(std::make_shared<Pingee>(msg.spActorPool));
		while(msg.nPingees-- > 0) vPingees_.push_back(ffpp::Factory<Pingee>{}.MakeShared(msg.spActorPool));
		for(size_t i = 0; i < vPingees_.size(); ++i) {
			vPingees_[i]->Emit(Pingee::MsgPing { Self(), i });
		}
		tpStart_ = ClockT::now();
	}

	void On(MsgComplete const&) { 
		auto const tpStop = ClockT::now();
		for(auto const& spPingee : vPingees_) spPingee->Finalize();
		Finalize(); 
		Info::Put(
			"Pinger completed. Total pings:", nPings_
			, "(", size_t(1000.0 * nPings_ / std::chrono::duration_cast<std::chrono::milliseconds>(tpStop - tpStart_).count()), "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_), "."
		);
	}

	void On(Pingee::MsgPong 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;
		// Log<>::Put("Pong got from actor:", msg.iPingee, ", latency:", usLatency, "us");
		// Log<>::Put("Pinging actor:", msg.iPingee);
		//SleepUpTo(200ms); //Simulating processing latency inside threadpool, comment this line to evaluate actual latency.
		vPingees_[msg.iPingee]->Emit(Pingee::MsgPing { Self(), msg.iPingee });
	}

private:
	std::vector<Pingee::SP> 
		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::verbose, algy::c_bitDefaultCaps, std::filesystem::path(argv[0]).parent_path() / "log" , "actorping" },
			algy::ConsoleOptions { algy::debug, algy::c_bitDedicated }
		);

		size_t
			secDuration = argc > 1 ? std::stoull(argv[1]) : 30 
			, nPingees = argc > 2 ? std::stoull(argv[2]) : std::thread::hardware_concurrency()
		;

		using ActorPoolT = ffpp::FunctionalQueuePool<PoolPolicy>;
		//auto spActorPool = std::make_shared<ActorPoolT>();
		//auto spActorPool = ActorPoolT::MakeShared();
		auto spActorPool = ffpp::Factory<ActorPoolT>{}.MakeShared();
		using TimerT = ffpp::FunctionalTimer<CommonPolicy>;
		auto spTimer = ffpp::Factory<TimerT>{}.MakeShared(500ms);

		Info::Put("Starting actor ping test...");
		Info::Put("Meta :", ffpp::Metadata());
		Info::Put("Creating pinger actor with", nPingees, "pingee actor(s).");
		//Thread pool state monitoring.
		auto fnlMonitoring = StartPoolMonitoring(spActorPool, spTimer, 1s);
		//auto spPinger = std::make_shared<Pinger>(spActorPool, 0);
		//auto spPinger = Pinger::MakeShared(spActorPool, 0);
		auto spPinger = ffpp::Factory<Pinger>{}.MakeShared(spActorPool, 0);
		spPinger->Emit(Pinger::MsgInit { spActorPool, nPingees });
		if(secDuration == 0) Info::Put("Press Enter to exit...");
		else Info::Put("Running test for", secDuration, "sec...");
		if(secDuration == 0) std::cin.get();
		else std::this_thread::sleep_for(std::chrono::seconds(secDuration));
		Info::Put("Finalizing...");
		auto [ugJoin, sgComplete] = ffpp::BeginJoin();
		spPinger->Emit(Pinger::MsgComplete { sgComplete });
		ffpp::FinishJoin(sgComplete, ugJoin);
		spTimer->Complete();
		spActorPool->Complete();
		Info::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