diff --git a/SolutionManager.hpp b/SolutionManager.hpp index 170a005..4d01f10 100644 --- a/SolutionManager.hpp +++ b/SolutionManager.hpp @@ -40,8 +40,260 @@ License: MPL2.0 (https://www.mozilla.org/en-US/MPL/2.0/) #ifndef NEBULOUS_SOLUTION_MANAGER #define NEBULOUS_SOLUTION_MANAGER +// Standard headers + +#include // Constant strings +#include // Normal strings +#include // Multimap for the work queue +#include // Solver ready status +#include // Pool of local solvers +#include // Range based views +#include // Standard algorithms +#include // For nice error messages +#include // Standard exceptions +#include // Error location reporting + +// Other packages + +#include // JSON object definition +using JSON = nlohmann::json; // Short form name space +#include // To print readable types + +// Theron++ headers + +#include "Actor.hpp" // Actor base class +#include "Utility/StandardFallbackHandler.hpp" // Exception unhanded messages +#include "Communication/NetworkingActor.hpp" // Networking actors + +// AMQ communication headers + +#include "Communication/AMQ/AMQjson.hpp" // For JSON metric messages +#include "Communication/AMQ/AMQEndpoint.hpp" // For AMQ related things + +// NebulOuS headers + +#include "Solver.hpp" // The basic solver class + namespace NebulOuS { - +/*============================================================================== + + Solution Manager + +==============================================================================*/ + +template< SolverAlgorithm SolverType > +class SolverManager +: virtual public Theron::Actor, + virtual public Theron::StandardFallbackHandler, + virtual public Theron::NetworkingActor< + typename Theron::AMQ::Message::PayloadType > +{ + // There is a topic name used to publish solutions found by the solvers. This + // topic is given to the constructor and kept as a constant during the class + // execution. + +private: + + const Address SolutionReceiver; + + // -------------------------------------------------------------------------- + // Solver management + // -------------------------------------------------------------------------- + // + // The solution manager dispatches the application execution contexts as + // requests for solutions to a pool of solvers. + +private: + + std::list< SolverType > SolverPool; + std::unordered_set< Address > ActiveSolvers, PassiveSolvers; + + // -------------------------------------------------------------------------- + // Application Execution Context management + // -------------------------------------------------------------------------- + // + // The contexts are dispatched in time sorted order. However, the time + // to solve a problem depends on the complexity of the the context and the + // results may therefore become available out-of-order. Each application + // execution context should carry a unique identifier, and this is used as + // the index key for quickly finding the right execution context. There is + // a second view of the queue of application context where the identifiers + // are sorted based on their time stamp. + + std::unordered_map< Solver::ContextIdentifierType, + Solver:: ApplicationExecutionContext > Contexts; + + std::multimap< Solver::TimePointType, Solver::ContextIdentifierType > + ContextExecutionQueue; + + // When the new applicaton execution context message arrives, it will be + // queued, and its time point recoreded. If there are passive solvers, + // the handler will immediately dispatch the contexts to each of these in + // time order. Essentially, it implements a 'riffle' for the passive solvers + // and the pending contexts.The issue is that there are likely different + // cardinalities of the two sets, and the solvers should be marked as + // active after the dispatch and the context identifiers should be + // removed from the queue after the dispatch. + + void DispatchToSolvers( void ) + { + if( !PassiveSolvers.empty() && !ContextExecutionQueue.empty() ) + { + for( const auto & [ SolverAddress, ContextElement ] : + ranges::views::zip( PassiveSolvers, ContextExecutionQueue ) ) + Send( Contexts.at( ContextElement.second ), SolverAddress ); + + // The number of contexts dispatched must equal the minimum of the + // available solvers and the available contexts. + + std::size_t DispatchedContexts + = std::min( PassiveSolvers.size(), ContextExecutionQueue.size() ); + + // Then move the passive solver addresses used to active solver addresses + + std::ranges::move( + std::ranges::subrange( PassiveSolvers.begin(), + PassiveSolvers.begin() + DispatchedContexts ), + std::inserter( ActiveSolvers ) ); + + // Then the dispatched context identifiers are removed from queue + + ContextExecutionQueue.erase( ContextExecutionQueue.begin(), + ContextExecutionQueue.begin() + DispatchedContexts ); + } +} + +// The handler function simply enqueues the received context, records its +// timesamp and dispatch as many contexts as possible to the solvers. Note +// that the context identifiers must be unique and there is a logic error +// if there is already a context with the same identifier. Then an invalid +// arguemtn exception will be thrown. This strategy should be reconsidered +// if there will be multiple entities firing execution contexts. + +void HandleApplicationExecutionContext( + const Solver:: ApplicationExecutionContext & TheContext, + const Address TheRequester ) +{ + auto [_, Success] = Contexts.try_emplace( + TheContext[ Solver::ContextIdentifier.data() ], TheContext ); + + if( Success ) + { + ContextExecutionQueue.emplace( + TheContext[ Solver::TimeStamp.data() ], + TheContext[ Solver::ContextIdentifier.data() ] ); + + DispatchToSolvers(); + } + else + { + std::source_location Location = std::source_location::current(); + std::ostringstream ErrorMessage; + + ErrorMessage << "[" << Location.file_name() << " at line " + << Location.line() + << "in function " << Location.function_name() <<"] " + << "An Application Execution Context with identifier " + << TheContext[ Solver::ContextIdentifier.data() ] + << " was received while there is already one with the same " + << "identifer. The identifiers must be unique!"; + + throw std::invalid_argument( ErrorMessage.str() ); + } +} + +// -------------------------------------------------------------------------- +// Solutions +// -------------------------------------------------------------------------- +// +// When a solution is received from a solver, it will be dispatched to all +// entities subscribing to the solution topic, and the solver will be returned +// to the pool of passive solvers. The dispatch function will be called at the +// end to ensure that the solver starts working on queued application execution +// contexts, if any. + +void PublishSolution( const Solver::Solution & TheSolution, + const Addres TheSolver ) +{ + Send( TheSolution, SolutionReceiver ); + PassiveSolvers.insert( ActiveSolvers.extract( TheSolver ) ); + DispatchToSolvers(); +} + +// -------------------------------------------------------------------------- +// Constructor and destructor +// -------------------------------------------------------------------------- +// +// The constructor takes the name of the Solution Mnager Actor, the name of +// the topic where the solutions should be published, and the topic where the +// application execution contexts will be published. If the latter is empty, +// the manager will not listen to any externally generated requests, only those +// being sent from the Metric Updater supposed to exist on the same Actor +// system node as the manager.The final arguments to the constructor is a +// set of arguments to the solver type in the order expected by the solver +// type and repeated for the number of (local) solvers that should be created. +// +// Currently this manager does not support dispatching configurations to +// remote solvers and collect responses from these. However, this can be +// circumvented by creating a local "solver" transferring the requests to +// a remote solvers and collecting results from the remote solver. + +SolverManager( const std::string & TheActorName, + const Theron::AMQ::TopicName & SolutionTopic, + const Theron::AMQ::TopicName & ContextPublisherTopic, + const auto & ...SolverArguments ) +: Actor( TheActorName ), + StandardFallbackHandler( Actor::GetAddress().AsString() ), + NetworkingActor( Actor::GetAddress().AsString() ), + SolutionReceiver( SolutionTopic ), + SolverPool(), ActiveSolvers(), PassiveSolvers(), + Contexts(), ContextExecutionQueue() +{ + // The solvers are created by the expanding the arguments for the solvers + // one by one creating new elements in the solver pool + + ( SolverPool.emplace_back( SolverArguments ), ... ); + + // If the solvers were successfully created, their addresses are recorded as + // passive servers, and a publisher is made for the solution channel, and + // optionally, a subscritpion is made for the alternative context publisher + // topic. If the solvers could not be created, then an invalid argument + // exception will be thrown. + + if( !SolverPool.empty() ) + { + std::ranges::transform( ServerPool, std::inserter( PassiveSolvers ), + [](const SolverType & TheSolver){ return TheSolver.GetAddress(); } ); + + Send( Theron::AMQ::NetworkLayer::TopicSubscription( + Theron::AMQ::NetworkLayer::TopicSubscription::Action::Publisher, + SolutionTopic ), GetSessionLayerAddress() ); + + if( !ContextPublisherTopic.empty() ) + Send( Theron::AMQ::NetworkLayer::TopicSubscription( + Theron::AMQ::NetworkLayer::TopicSubscription::Action::Subscription, + ContextPublisherTopic ), GetSessionLayerAddress() ); + } + else + { + std::source_location Location = std::source_location::current(); + std::ostringstream ErrorMessage; + + ErrorMessage << "[" << Location.file_name() << " at line " + << Location.line() + << "in function " << Location.function_name() <<"] " + << "It was not possible to construct any solver of type " + << boost::core::demangle( typeid( SolverType ).name() ) + << " from the given constructor argument types: "; + + (( ErrorMessage << boost::core::demangle( typeid( SolverArguments ).name() ) << " " ), ... ); + + throw std::invalid_argument( ErrorMessage.str() ); + } +} + +}; + } // namespace NebulOuS #endif // NEBULOUS_SOLUTION_MANAGER \ No newline at end of file diff --git a/Solver.hpp b/Solver.hpp index 2cd3509..dcaefa3 100644 --- a/Solver.hpp +++ b/Solver.hpp @@ -36,6 +36,9 @@ License: MPL2.0 (https://www.mozilla.org/en-US/MPL/2.0/) // Standard headers #include // Constant strings +#include // Normal strings +#include // To store metric-value maps +#include // To test template parameters // Other packages @@ -58,14 +61,13 @@ namespace NebulOuS Solver Actor ==============================================================================*/ -// class Solver : virtual public Theron::Actor, virtual public Theron::StandardFallbackHandler { -private: +public: // -------------------------------------------------------------------------- // Application Execution Context @@ -103,8 +105,178 @@ private: static constexpr std::string_view ExecutionContext = "ExecutionContext"; + // To ensure that the execution context is correctly provided by the senders + // The expected metric value structure is defined as a type based on the + // standard unsorted map based on a JSON value object since this can hold + // various value types. + using MetricValueType = std::unordered_map< std::string, JSON >; + + // The identification type for the application execution context is defined + // so that other classes may use it, but also so that it can be easily + // changed if needed. It is assumed that the type must have a hash function + // so that the type can be used in ordered data structures. + + using ContextIdentifierType = std::string; + + // The same goes for the time point type. This is defined as the number of + // microseconds since the POSIX time epoch (1 January 1970) and stored as a + // long integral value. + + using TimePointType = unsigned long long; + + // The message is a simple JSON object where the various fields of the + // message struct are set by the constructor to ensure that all fields are + // given when the message is constructed. + + class ApplicationExecutionContext + : public Theron::AMQ::JSONMessage + { + public: + + static constexpr + std::string_view MessageIdentifier = "Solver::ApplicationExecutionContext"; + + ApplicationExecutionContext( const ContextIdentifierType & TheIdentifier, + const TimePointType MicroSecondTimePoint, + const std::string ObjectiveFunctionID, + const MetricValueType & TheContext ) + : JSONMessage( MessageIdentifier.data(), + { { ContextIdentifier.data(), TheIdentifier }, + { TimeStamp.data(), MicroSecondTimePoint }, + { ObjectiveFunctionLabel.data(), ObjectiveFunctionID }, + { ExecutionContext.data(), TheContext } } + ) {} + + ApplicationExecutionContext( const ApplicationExecutionContext & Other ) + : JSONMessage( Other ) + {} + + ApplicationExecutionContext() = delete; + virtual ~ApplicationExecutionContext() = default; + }; + + // The handler for this message is virtual as it where the real action will + // happen and the search for the optimal solution will hopefully lead to a + // feasible soltuion that can be returned to the sender of the applicaton + // context. + +protected: + + virtual void SolveProblem( const ApplicationExecutionContext & TheContext, + const Address TheRequester ) = 0; + + // -------------------------------------------------------------------------- + // Solution + // -------------------------------------------------------------------------- + // + // When a solution is found to a given problem, the solver should return the + // found optimal value for the given objective function, It should return + // this value together with the values assigned to the feasible variables + // leading to this optimal objective value. Additionally, the message will + // contain the time point for which this solution is valid, and the + // application execution context as the optimal solution is conditioned + // on this solution. + // + // Since the probelm being resolved can be multi-objective, the values of all + // objective values will be returned as a JSON map where the attributes are + // the names of the objective functions in the optimisation problem, and the + // values are the ones assigned by the optimiser. This JSON map object is + // passed under the global attribute "ObjectiveValues" + +public: + + using ObjectiveValuesType = MetricValueType; + static constexpr std::string_view ObjectiveValues = "ObjectiveValues"; + + class Solution + : public Theron::AMQ::JSONMessage + { + public: + + static constexpr std::string_view MessageIdentifier = "Solver::Solution"; + + Solution( const ContextIdentifierType & TheIdentifier, + const TimePointType MicroSecondTimePoint, + const std::string ObjectiveFunctionID, + const ObjectiveValuesType & TheObjectiveValues, + const MetricValueType & TheContext ) + : JSONMessage( MessageIdentifier.data() , + { { ContextIdentifier.data(), TheIdentifier }, + { TimeStamp.data(), MicroSecondTimePoint }, + { ObjectiveFunctionLabel.data(), ObjectiveFunctionID }, + { ObjectiveValues.data(), TheObjectiveValues }, + { ExecutionContext.data(), TheContext } } ) + {} + + Solution() = delete; + virtual ~Solution() = default; + }; + + // -------------------------------------------------------------------------- + // Optimisation problem definition + // -------------------------------------------------------------------------- + // + // There are many ways the optimisation problem can be passed to the solver, + // and it is therefore not possible to give an exact format for the message + // to define or update the optimisation problem. The message is basically + // left as a JSON message and it will be up to the actual solver algorithm + // to implement this in a way appropriate for the algorithm. + + class OptimisationProblem + : public Theron::AMQ::JSONMessage + { + public: + + static constexpr + std::string_view MessageIdentifier = "Solver::OptimisationProblem"; + + OptimisationProblem( const JSON & TheProblem ) + : JSONMessage( MessageIdentifier.data(), TheProblem ) + {} + + OptimisationProblem() = delete; + virtual ~OptimisationProblem() = default; + }; + + // The handler for this message must also be defined by the algorithm that + // implements the solver. + + virtual void DefineProblem( const OptimisationProblem & TheProblem, + const Address TheOracle ) = 0; + + // -------------------------------------------------------------------------- + // Constructor and destructor + // -------------------------------------------------------------------------- + // + // The constructor defines the message handlers so that the derived soler + // classes will not need to deal with the Actor specific details, and to + // ensure that the handlers are called when the Actor receives the various + // messages. The constructor requires an actor name as the only parameter. + + Solver( const std::string & TheSolverName ) + : Actor( TheSolverName ), + StandardFallbackHandler( Actor::GetAddress().AsString() ) + { + RegisterHandler( this, &Solver::SolveProblem ); + RegisterHandler( this, &Solver::DefineProblem ); + } + + Solver() = delete; + virtual ~Solver() = default; }; +/*============================================================================== + + Solver concept + +==============================================================================*/ +// +// A concept is defined to validate that solvers used inherits this standard +// base class and that they implement the virtual methods. + +template< class TheSolverType > +concept SolverAlgorithm = std::derived_from< TheSolverType, Solver >; + } // namespace NebulOuS #endif // NEBULOUS_SOLVER \ No newline at end of file