From 04d980c08c7d175ceee0e089fadf81d8145fd87c Mon Sep 17 00:00:00 2001 From: superwhiskers Date: Mon, 2 Nov 2026 14:44:31 -0500 Subject: [PATCH] implement a generic co-simulation interface - add client/server transport and message handling - add co-simulation interface tests --- GridKit/CMakeLists.txt | 5 + GridKit/CoSim/CMakeLists.txt | 6 + GridKit/CoSim/CoSim.cpp | 649 ++++++++++++++++++++++ GridKit/CoSim/CoSim.hpp | 93 ++++ GridKit/CoSim/CoSimInternal.hpp | 106 ++++ GridKit/CoSim/Message.hpp | 144 +++++ GridKit/Definitions.hpp.in | 1 + cmake/Config.cmake.in | 3 + tests/UnitTests/Network/CMakeLists.txt | 11 + tests/UnitTests/Network/CoSimTests.hpp | 290 ++++++++++ tests/UnitTests/Network/runCoSimTests.cpp | 16 + 11 files changed, 1324 insertions(+) create mode 100644 GridKit/CoSim/CMakeLists.txt create mode 100644 GridKit/CoSim/CoSim.cpp create mode 100644 GridKit/CoSim/CoSim.hpp create mode 100644 GridKit/CoSim/CoSimInternal.hpp create mode 100644 GridKit/CoSim/Message.hpp create mode 100644 tests/UnitTests/Network/CoSimTests.hpp create mode 100644 tests/UnitTests/Network/runCoSimTests.cpp diff --git a/GridKit/CMakeLists.txt b/GridKit/CMakeLists.txt index 35e1e75f03..4553e63ff9 100644 --- a/GridKit/CMakeLists.txt +++ b/GridKit/CMakeLists.txt @@ -25,6 +25,11 @@ install(FILES ${GRIDKIT_THIRD_PARTY_DIR}/magic-enum/LICENSE # General Utilities and File IO add_subdirectory(Utilities) +if(GRIDKIT_ENABLE_ZMQ) + # Co-simulation support + add_subdirectory(CoSim) +endif() + # Memory management utilities add_subdirectory(MemoryUtilities) diff --git a/GridKit/CoSim/CMakeLists.txt b/GridKit/CoSim/CMakeLists.txt new file mode 100644 index 0000000000..4359a22f9a --- /dev/null +++ b/GridKit/CoSim/CMakeLists.txt @@ -0,0 +1,6 @@ +gridkit_add_library( + co_sim + SOURCES CoSim.cpp + HEADERS CoSim.hpp Message.hpp + LINK_LIBRARIES PRIVATE cppzmq + INCLUDE_DIRECTORIES PRIVATE ${GRIDKIT_THIRD_PARTY_DIR}/nlohmann-json/include) diff --git a/GridKit/CoSim/CoSim.cpp b/GridKit/CoSim/CoSim.cpp new file mode 100644 index 0000000000..ee3e097c75 --- /dev/null +++ b/GridKit/CoSim/CoSim.cpp @@ -0,0 +1,649 @@ +#include +#include +#include +#include +#include +#include + +#include + +#include + +namespace GridKit +{ + namespace CoSim + { + namespace + { + using json = nlohmann::json; + + /// Convert a message type to a string. + /// + /// @throw runtime_error if the message type was unknown + std::string typeName(MessageType type) + { + switch (type) + { + case MessageType::INIT: + return "init"; + case MessageType::STEP: + return "step"; + case MessageType::UPDATE: + return "update"; + case MessageType::COMMIT: + return "commit"; + case MessageType::END: + return "end"; + case MessageType::SHUTDOWN: + return "shutdown"; + case MessageType::CLOSE: + return "close"; + case MessageType::FAIL: + return "fail"; + } + + throw std::runtime_error("invalid co-simulation message type"); + } + + /// Convert a string to a message type. + /// + /// @throw runtime_error if the message type was unknown + MessageType parseType(const std::string& name) + { + for (auto type : + {MessageType::INIT, + MessageType::STEP, + MessageType::UPDATE, + MessageType::COMMIT, + MessageType::END, + MessageType::SHUTDOWN, + MessageType::CLOSE, + MessageType::FAIL}) + { + if (typeName(type) == name) + { + return type; + } + } + + throw std::runtime_error("unknown co-simulation message type: " + name); + } + + /// Validate a co-simulation message. + /// + /// @throws runtime_error if the co simulation signals are invalid or if a + /// time value was not finite + template + void validate(const MessageData& message) + { + if (!std::isfinite(message.time) || !std::isfinite(message.final_time) + || !std::isfinite(message.interval)) + { + throw std::runtime_error("non-finite co-simulation time"); + } + + for (const auto& [name, value] : message.signals) + { + if (name.empty() || !std::isfinite(value)) + { + throw std::runtime_error("invalid co-simulation signal: " + name); + } + } + + for (auto time : message.event_times) + { + if (!std::isfinite(time)) + { + throw std::runtime_error("non-finite co-simulation event time"); + } + } + } + } // namespace + + template + detail::Channel::Channel( + Disposition disposition, + const std::string& endpoint, + int timeout_ms, + const std::atomic* state) + : context_(), + socket_( + context_, + disposition == Disposition::SERVER ? zmq::socket_type::rep + : zmq::socket_type::req), + timeout_ms_(timeout_ms), + state_(state), + can_send_(disposition == Disposition::CLIENT) + { + if (timeout_ms <= 0 || endpoint.empty()) + { + throw std::invalid_argument( + "co-simulation requires an endpoint and a positive timeout"); + } + + // bound socket cleanup + socket_.set(zmq::sockopt::linger, std::min(timeout_ms, 1000)); + socket_.set(zmq::sockopt::immediate, 1); + + if (disposition == Disposition::SERVER) + { + socket_.bind(endpoint); + } + else + { + socket_.connect(endpoint); + } + } + + template + void detail::Channel::checkStop() const + { + if (state_ + && state_->load(std::memory_order_relaxed) + == RunState::STOP_REQUESTED) + { + throw Stopped{}; + } + } + + template + void detail::Channel::wait( + short events, + const char* operation, + std::chrono::steady_clock::time_point deadline) + { + using Clock = std::chrono::steady_clock; + + while (true) + { + checkStop(); + const auto remaining = + std::chrono::duration_cast( + deadline - Clock::now()); + if (remaining.count() <= 0) + { + throw std::runtime_error( + std::format("Co-simulation {} timed out", operation)); + } + + zmq::pollitem_t item{socket_.handle(), 0, events, 0}; + try + { + zmq::poll( + &item, 1, std::min(remaining, std::chrono::milliseconds(25))); + } + catch (const zmq::error_t& error) + { + if (error.num() == EINTR) + { + continue; + } + throw; + } + checkStop(); + if (item.revents & events) + { + return; + } + } + } + + template + void detail::Channel::send( + const MessageData& message) + { + checkStop(); + if (!can_send_) + { + throw std::logic_error("Co-simulation socket is waiting for a reply"); + } + + validate(message); + json wire = { + {"type", typeName(message.type)}, + {"session", message.session}, + {"time", message.time}, + {"signals", message.signals}}; + + if (message.type == MessageType::INIT) + { + wire["final_time"] = message.final_time; + wire["interval"] = message.interval; + wire["event_times"] = message.event_times; + } + + if (message.type == MessageType::FAIL) + { + wire["error"] = message.error; + } + + const auto data = wire.dump(); + const auto deadline = std::chrono::steady_clock::now() + + std::chrono::milliseconds(timeout_ms_); + + do + { + wait(ZMQ_POLLOUT, "send", deadline); + } while (!socket_.send(zmq::buffer(data), zmq::send_flags::dontwait)); + + can_send_ = false; + } + + template + MessageData detail::Channel::receive() + { + checkStop(); + + if (can_send_) + { + throw std::logic_error( + "Co-simulation socket must send before receiving"); + } + + zmq::message_t raw; + const auto deadline = std::chrono::steady_clock::now() + + std::chrono::milliseconds(timeout_ms_); + do + { + wait(ZMQ_POLLIN, "receive", deadline); + } while (!socket_.recv(raw, zmq::recv_flags::dontwait)); + + can_send_ = true; + const auto wire = json::parse(raw.to_string_view()); + + MessageData message; + message.type = parseType(wire.at("type").get()); + if (!wire.at("session").is_number_unsigned()) + { + throw std::runtime_error("Invalid co-simulation session ID"); + } + + message.session = wire.at("session").get(); + message.time = wire.at("time").get(); + message.signals = wire.at("signals").get>(); + + if (message.type == MessageType::INIT) + { + message.final_time = wire.at("final_time").get(); + message.interval = wire.at("interval").get(); + message.event_times = wire.at("event_times").get>(); + } + + if (message.type == MessageType::FAIL) + { + message.error = wire.at("error").get(); + } + + validate(message); + return message; + } + + template + MessageData detail::Channel::exchange( + const MessageData& request) + { + send(request); + auto reply = receive(); + if (reply.type == MessageType::FAIL) + { + closePeer(reply); + throw PeerFailure("co-simulation peer failed: " + reply.error); + } + + const bool final_signals = + request.type == MessageType::STEP && reply.type == MessageType::END; + + if ((!final_signals && reply.type != request.type) + || reply.session != request.session || reply.time != request.time) + { + throw std::runtime_error( + "co-simulation acknowledgement has the wrong type, session or time"); + } + + return reply; + } + + template + class detail::Channel::Teardown + { + Channel& channel_; + const std::atomic* state_; + int timeout_ms_; + + public: + explicit Teardown(Channel& channel) noexcept + : channel_(channel), + state_(channel.state_), + timeout_ms_(channel.timeout_ms_) + { + channel_.state_ = nullptr; + channel_.timeout_ms_ = 100; + } + + ~Teardown() + { + channel_.state_ = state_; + channel_.timeout_ms_ = timeout_ms_; + } + }; + + template + void detail::Channel::finishCancellation() noexcept + { + try + { + socket_.set(zmq::sockopt::linger, 100); + } + catch (...) + { + } + } + + template + void detail::Channel::fail( + const MessageData& message) noexcept + { + Teardown teardown(*this); + + try + { + if (can_send_) + { + send(message); + if (socket_.get(zmq::sockopt::type) == ZMQ_REQ) + { + const auto reply = receive(); + closePeer(reply); + } + else + { + awaitClose(); + } + } + } + catch (...) + { + } + } + + template + void detail::Channel::closePeer( + const MessageMetadata& last) noexcept + { + Teardown teardown(*this); + + try + { + MessageData close; + close.type = MessageType::CLOSE; + close.session = last.session; + close.time = last.time; + send(close); + receive(); + } + catch (...) + { + } + } + + template + void detail::Channel::awaitClose() + { + auto close = receive(); + if (close.type != MessageType::CLOSE) + { + close.type = MessageType::FAIL; + close.error = "Co-simulation already completed"; + close.signals.clear(); + } + send(close); + } + + template + void detail::Channel::finish( + const MessageData& reply) + { + send(reply); + Teardown teardown(*this); + + try + { + awaitClose(); + } + catch (...) + { + } + } + + template + CoSim::CoSim( + const std::string& address, + MessageStepFunction step, + MessageExtractFunction extract, + int timeout_ms) + : address_(address), + step_(std::move(step)), + extract_(std::move(extract)), + timeout_ms_(timeout_ms) + { + if (address.empty() || timeout_ms <= 0 || !step_ || !extract_) + { + throw std::invalid_argument( + "co-simulation requires an endpoint, callbacks and a positive timeout"); + } + } + + template + CoSim::CoSim( + const std::string& address, + StepFunction step, + ExtractFunction extract, + int timeout_ms) + : address_(address), timeout_ms_(timeout_ms) + { + if (address.empty() || timeout_ms <= 0 || !step || !extract) + { + throw std::invalid_argument( + "co-simulation requires an endpoint, callbacks and a positive timeout"); + } + + step_ = [step = std::move(step)](const auto& message, const auto&) + { return step(message.signals); }; + extract_ = [extract = std::move(extract)](const auto&) + { + MessageData message; + message.type = MessageType::STEP; + message.signals = extract(); + return message; + }; + } + + template + RunResult CoSim::run() + { + auto expected = detail::RunState::IDLE; + if (!state_.compare_exchange_strong( + expected, + detail::RunState::RUNNING, + std::memory_order_acquire, + std::memory_order_relaxed)) + { + throw std::logic_error( + "attempted to start a co-simulation while it was already running"); + } + + // ensure the state is reset on scope exit + struct Guard + { + std::atomic& state; + + ~Guard() + { + state.store(detail::RunState::IDLE, std::memory_order_release); + } + } guard{state_}; + + detail::Channel channel( + disposition, address_, timeout_ms_, &state_); + MessageMetadata current; + + struct Completed + { + }; + + try + { + MessageMetadata last_reply; + bool peer_finished = false; + const auto receive = [&] + { + auto message = channel.receive(); + current = message; + + if (message.type == MessageType::FAIL) + { + throw std::runtime_error( + "co-simulation client failed: " + message.error); + } + + if (message.type == MessageType::CLOSE) + { + channel.send(message); + throw Completed{}; + } + return message; + }; + const MessageExchangeFunction exchange = + [&](const auto& message) + { + if constexpr (disposition == Disposition::SERVER) + { + if (current.type == MessageType::STEP + && message.type == MessageType::END) + { + channel.finish(message); + throw Completed{}; + } + channel.send(message); + return receive(); + } + else + { + if (peer_finished) + { + throw std::runtime_error( + "co-simulation peer ended before the callback completed"); + } + current = message; + auto reply = channel.exchange(message); + last_reply = reply; + peer_finished = message.type == MessageType::STEP + && reply.type == MessageType::END; + return reply; + } + }; + while (true) + { + if (state_.load(std::memory_order_relaxed) + == detail::RunState::STOP_REQUESTED) + { + throw detail::Stopped{}; + } + + if constexpr (disposition == Disposition::SERVER) + { + const auto incoming = receive(); + const bool proceed = step_(incoming, exchange); + + if (state_.load(std::memory_order_relaxed) + == detail::RunState::STOP_REQUESTED) + { + throw detail::Stopped{}; + } + + auto reply = extract_(exchange); + + // terminate the step exchange + if (!proceed && current.type == MessageType::STEP + && reply.type == MessageType::STEP) + { + reply.type = MessageType::END; + } + + if (!proceed) + { + channel.finish(reply); + return RunResult::COMPLETED; + } + + channel.send(reply); + } + else + { + auto outgoing = extract_(exchange); + bool proceed = false; + if (!peer_finished) + { + const auto incoming = exchange(outgoing); + proceed = step_(incoming, exchange); + } + if (!proceed || peer_finished) + { + channel.closePeer(last_reply); + return state_.load(std::memory_order_relaxed) + == detail::RunState::STOP_REQUESTED + ? RunResult::STOPPED + : RunResult::COMPLETED; + } + } + } + } + catch (const Completed&) + { + return RunResult::COMPLETED; + } + catch (const detail::Stopped&) + { + MessageData failure; + failure.type = MessageType::FAIL; + failure.session = current.session; + failure.time = std::isfinite(current.time) ? current.time : 0.0; + failure.error = "co-simulation stopped"; + channel.fail(failure); + channel.finishCancellation(); + return RunResult::STOPPED; + } + catch (const detail::PeerFailure&) + { + throw; + } + catch (const std::exception& error) + { + MessageData failure; + failure.type = MessageType::FAIL; + failure.session = current.session; + failure.time = std::isfinite(current.time) ? current.time : 0.0; + failure.error = error.what(); + channel.fail(failure); + throw; + } + catch (...) + { + MessageData failure; + failure.type = MessageType::FAIL; + failure.session = current.session; + failure.time = std::isfinite(current.time) ? current.time : 0.0; + failure.error = "non-standard exception in co-simulation callback"; + channel.fail(failure); + throw; + } + } + + template class CoSim; + template class CoSim; + template class CoSim; + template class CoSim; + template class detail::Channel; + template class detail::Channel; + } // namespace CoSim +} // namespace GridKit diff --git a/GridKit/CoSim/CoSim.hpp b/GridKit/CoSim/CoSim.hpp new file mode 100644 index 0000000000..ee025bb1df --- /dev/null +++ b/GridKit/CoSim/CoSim.hpp @@ -0,0 +1,93 @@ +#pragma once + +#include +#include + +#include + +namespace GridKit +{ + namespace CoSim + { + namespace detail + { + /// State of a co-simulation run. + enum class RunState : std::uint8_t + { + /// Simulation is not running. + IDLE, + + /// Simulation is running. + RUNNING, + + /// Simulation stop has been requested. + STOP_REQUESTED + }; + } // namespace detail + + /// Co-simulation adapter. + template + class CoSim + { + static_assert( + std::atomic::is_always_lock_free, + "lock-free atomics for std::uint8_t are required for co-simulation"); + + public: + CoSim( + const std::string& address, + StepFunction step, + ExtractFunction extract, + int timeout_ms = 30000); + CoSim( + const std::string& address, + MessageStepFunction step, + MessageExtractFunction extract, + int timeout_ms = 30000); + CoSim(const CoSim&) = delete; + CoSim& operator=(const CoSim&) = delete; + + /// Run the co-simulation. + RunResult run(); + + /// Request the co-simulation to stop. + void stop() noexcept + { + auto expected = detail::RunState::RUNNING; + state_.compare_exchange_strong( + expected, detail::RunState::STOP_REQUESTED, std::memory_order_relaxed); + } + + /// Check if the co-simulation is running. + bool running() const noexcept + { + return state_.load(std::memory_order_acquire) != detail::RunState::IDLE; + } + + private: + /// Endpoint address of the co-simulation. + /// + /// If the adapter disposition is `Disposition::SERVER`, this + /// represents the address the adapter will listen on. Otherwise, + /// this is the address of a co-simulation server to connect to. + std::string address_; + + /// Step function. + MessageStepFunction step_; + + /// Extraction function. + MessageExtractFunction extract_; + + /// Timeout of ZeroMQ operations in milliseconds. + int timeout_ms_; + + /// Current state of the co-simulation adapter. + std::atomic state_{detail::RunState::IDLE}; + }; + + extern template class CoSim; + extern template class CoSim; + extern template class CoSim; + extern template class CoSim; + } // namespace CoSim +} // namespace GridKit diff --git a/GridKit/CoSim/CoSimInternal.hpp b/GridKit/CoSim/CoSimInternal.hpp new file mode 100644 index 0000000000..d19104a506 --- /dev/null +++ b/GridKit/CoSim/CoSimInternal.hpp @@ -0,0 +1,106 @@ +#pragma once + +#include +#include +#include + +#include + +#include + +namespace GridKit +{ + namespace CoSim + { + namespace detail + { + /// Co-simulation was stopped. + class Stopped : public std::exception + { + public: + const char* what() const noexcept override + { + return "co-simulation stopped"; + } + }; + + /// Co-simulation peer failed. + class PeerFailure : public std::runtime_error + { + public: + using std::runtime_error::runtime_error; + }; + + /// Private transport implementation. + template + class Channel + { + public: + Channel( + Disposition disposition, + const std::string& endpoint, + int timeout_ms = 30000, + const std::atomic* state = nullptr); + Channel(const Channel&) = delete; + Channel& operator=(const Channel&) = delete; + + /// Send a co-simulation message. + void send(const MessageData& message); + + /// Receive a co-simulation message. + MessageData receive(); + + /// Exchange co-simulation messages with the peer. + MessageData exchange( + const MessageData& request); + + /// Notify the peer of failure. + void fail(const MessageData& message) noexcept; + + /// Acknowledge the last peer response prior to closing the connection. + void closePeer(const MessageMetadata& last) noexcept; + + /// Send the final response and briefly wait for the peer to consume it. + void finish(const MessageData& reply); + + /// Bound cleanup without discarding a queued failure reply. + void finishCancellation() noexcept; + + private: + /// Scope guard to update the timeout to a shorter teardown timeout and + /// restore it when the scope is exited. + class Teardown; + + /// Wait for a ZeroMQ event. + void wait( + short events, + const char* operation, + std::chrono::steady_clock::time_point deadline); + + /// Check if the simulation has been stopped. + void checkStop() const; + + /// Wait for a close reply. + void awaitClose(); + + /// ZeroMQ context. + zmq::context_t context_; + + /// ZeroMQ socket. + zmq::socket_t socket_; + + /// Timeout of ZeroMQ operations in milliseconds. + int timeout_ms_; + + /// Reference to the co-simulation state. + const std::atomic* state_; + + /// Whether or not we are currently able to send a message. + bool can_send_; + }; + + extern template class Channel; + extern template class Channel; + } // namespace detail + } // namespace CoSim +} // namespace GridKit diff --git a/GridKit/CoSim/Message.hpp b/GridKit/CoSim/Message.hpp new file mode 100644 index 0000000000..585e220e2a --- /dev/null +++ b/GridKit/CoSim/Message.hpp @@ -0,0 +1,144 @@ +#pragma once + +#include +#include +#include +#include +#include + +namespace GridKit +{ + namespace CoSim + { + /// Disposition of the co-simulation adapter. + enum class Disposition + { + /// Adapter is a server. + SERVER, + + /// Adapter is a client. + CLIENT, + }; + + /// Kinds of messages. + enum class MessageType + { + /// Start a new simulation session. + INIT, + + /// Advance simulation time to a target using inputs, then return outputs. + STEP, + + /// Recalculate the interface at the current time without advancing it. + UPDATE, + + /// Accept the interface point and write pending monitor samples. + COMMIT, + + /// Finish the current study and release its model. + END, + + /// Request shutting down a server after completing a batch. + SHUTDOWN, + + /// Acknowledge receipt of the final response before closing a connection. + CLOSE, + + /// Report an error or cancellation. Delivery is best effort. + FAIL, + }; + + /// Signals exchanged between co-simulation participants. + template + using SignalValues = std::map; + + /// Metadata within a co-simulation message. + /// + /// This includes all information aside from signal values. + struct MessageMetadata + { + /// Message kind. + MessageType type{MessageType::INIT}; + + /// Session this message belongs to. + std::uint64_t session{0}; + + /// Simulation time associated with the message. + double time{0.0}; + + /// Final simulation time, exchanged during INIT. + double final_time{0.0}; + + /// Communication interval, exchanged during INIT. + double interval{0.0}; + + /// Event times, exchanged during INIT to coordinate communication points. + std::vector event_times; + + /// Explanation of an error reported by a peer. + std::string error; + + MessageMetadata() = default; + + MessageMetadata( + MessageType type, std::uint64_t session, double time = 0.0) + : type(type), session(session), time(time) + { + } + }; + + /// Full co-simulation message. + template + struct MessageData : MessageMetadata + { + /// Signals contained within this message. + SignalValues signals; + + MessageData() = default; + + MessageData(MessageType type, std::uint64_t session, double time = 0.0) + : MessageMetadata(type, session, time) + { + } + }; + + /// Callback form to apply received signals and control simulation + /// continuity. + template + using StepFunction = std::function&)>; + + /// Callback form to extract signal data from a simulation. + template + using ExtractFunction = std::function()>; + + /// Function to send a message and wait for the next incoming message. + /// + /// With the client disposition, this is a request then a reply. For a + /// server, this is a reply to the current request and the next request. + template + using MessageExchangeFunction = + std::function(const MessageData&)>; + + /// Apply an incoming message and control simulation continuity. + template + using MessageStepFunction = std::function&, + const MessageExchangeFunction&)>; + + /// Produce the next outgoing message, optionally exchanging intermediate + /// messages first. + template + using MessageExtractFunction = std::function< + MessageData(const MessageExchangeFunction&)>; + + /// Result of a run. + enum class RunResult + { + /// Run completed. + COMPLETED, + + /// Run was stopped. + STOPPED + }; + } // namespace CoSim +} // namespace GridKit diff --git a/GridKit/Definitions.hpp.in b/GridKit/Definitions.hpp.in index 77075c9dc8..9452e59cd8 100644 --- a/GridKit/Definitions.hpp.in +++ b/GridKit/Definitions.hpp.in @@ -3,6 +3,7 @@ #cmakedefine GRIDKIT_ENABLE_ENZYME #cmakedefine GRIDKIT_ENABLE_OPENMP #cmakedefine GRIDKIT_ENABLE_THREADS +#cmakedefine GRIDKIT_ENABLE_ZMQ #cmakedefine GRIDKIT_ENABLE_SUNDIALS_SPARSE #cmakedefine GRIDKIT_ENABLE_DEVELOPER_MODE diff --git a/cmake/Config.cmake.in b/cmake/Config.cmake.in index 3ecd9cb5e6..34c1936dde 100644 --- a/cmake/Config.cmake.in +++ b/cmake/Config.cmake.in @@ -8,6 +8,9 @@ include ( "${CMAKE_CURRENT_LIST_DIR}/GridKitTargets.cmake" ) include(CMakeFindDependencyMacro) +if(@GRIDKIT_ENABLE_ZMQ@) + find_dependency(cppzmq) +endif() set(GridKit_ENABLE_SUNDIALS @GRIDKIT_ENABLE_SUNDIALS@) if(GridKit_ENABLE_SUNDIALS) list(APPEND CMAKE_PREFIX_PATH @SUNDIALS_DIR@) diff --git a/tests/UnitTests/Network/CMakeLists.txt b/tests/UnitTests/Network/CMakeLists.txt index b73f80c1ee..4e2e0044ed 100644 --- a/tests/UnitTests/Network/CMakeLists.txt +++ b/tests/UnitTests/Network/CMakeLists.txt @@ -4,4 +4,15 @@ if(GRIDKIT_ENABLE_ZMQ) add_test(NAME PhasorDynamicsZMQTest COMMAND test_zmq) install(TARGETS test_zmq RUNTIME DESTINATION bin) + + add_executable(test_cosim runCoSimTests.cpp) + target_link_libraries( + test_cosim + GridKit::co_sim + GridKit::testing + cppzmq) + target_include_directories(test_cosim PRIVATE ${GRIDKIT_THIRD_PARTY_DIR}/nlohmann-json/include) + + add_test(NAME CoSimulationChannelTest COMMAND test_cosim) + set_tests_properties(CoSimulationChannelTest PROPERTIES TIMEOUT 30) endif() diff --git a/tests/UnitTests/Network/CoSimTests.hpp b/tests/UnitTests/Network/CoSimTests.hpp new file mode 100644 index 0000000000..5487b732b7 --- /dev/null +++ b/tests/UnitTests/Network/CoSimTests.hpp @@ -0,0 +1,290 @@ +#include +#include +#include +#include +#include +#include +#include + +#include + +#include +#include +#include + +namespace GridKit +{ + namespace Testing + { + using namespace GridKit::CoSim; + using nlohmann::json; + + template + class CoSimTests + { + using Channel = detail::Channel; + using Message = MessageData; + using Signals = SignalValues; + using PeerFailure = detail::PeerFailure; + + /// Retrieve an endpoint to use for the co-simulation. + static std::string endpoint() + { + static unsigned serial = 0; + + const auto timestamp = + std::chrono::steady_clock::now().time_since_epoch().count(); + const auto name = + std::format("gridkit-unit-{}-{}", timestamp, ++serial); + + return std::format( + "ipc://{}", (std::filesystem::temp_directory_path() / name).string()); + } + + public: + TestOutcome sessions() + { + TestStatus outcome; + + const auto address = endpoint(); + auto server = std::async( + std::launch::async, + [&]() -> auto + { + bool outcome = true; + Channel channel(Disposition::SERVER, address, 5000); + + for (std::uint64_t session = 1; session <= 2; session++) + { + auto init = channel.receive(); + + outcome &= init.type == MessageType::INIT; + outcome &= init.session == session; + outcome &= init.signals.at("voltage real") == 1.03; + + init.signals = {{"current from generator", 2.5}}; + init.event_times = {0.025, 0.035}; + channel.send(init); + + auto step = channel.receive(); + + outcome &= step.time == 0.05; + outcome &= step.type == MessageType::STEP; + + channel.send(step); + + auto end = channel.receive(); + + outcome &= end.type == MessageType::END; + + channel.send(end); + } + + const auto stop = channel.receive(); + + outcome &= stop.type == MessageType::SHUTDOWN; + + channel.finish(stop); + + return outcome; + }); + + Channel client(Disposition::CLIENT, address, 5000); + + for (std::uint64_t session = 1; session <= 2; session++) + { + Message init(MessageType::INIT, session); + init.final_time = 0.05; + init.interval = 0.05; + init.signals = {{"voltage real", 1.03}}; + + auto reply = client.exchange(init); + + outcome *= reply.signals.at("current from generator") == 2.5; + outcome *= reply.event_times.size() == 2; + + Message step(MessageType::STEP, session); + step.time = 0.05; + + client.exchange(step); + + Message end(MessageType::END, session); + + client.exchange(end); + } + + Message stop(MessageType::SHUTDOWN, 2); + + client.exchange(stop); + client.closePeer(stop); + + outcome *= server.get(); + + return outcome.report(__func__); + } + + TestOutcome rejectedReplies() + { + TestStatus outcome; + + for (int kind = 0; kind < 4; kind++) + { + const auto address = endpoint(); + auto server = std::async( + std::launch::async, + [&] + { + Channel channel(Disposition::SERVER, address, 5000); + auto message = channel.receive(); + + switch (kind) + { + case 0: + message.type = MessageType::FAIL; + message.error = "solver failure"; + break; + case 1: + message.session++; + break; + case 2: + message.time = 0.1; + break; + case 3: + message.type = MessageType::END; + break; + } + + channel.finish(message); + }); + + Channel client(Disposition::CLIENT, address, 5000); + + if (kind == 0) + { + outcome *= throws( + [&] + { client.exchange(Message()); }); + } + else + { + outcome *= + throws( + [&] + { client.exchange(Message()); }); + client.closePeer(Message()); + } + + server.get(); + } + + return outcome.report(__func__); + } + + TestOutcome malformedMessages() + { + TestStatus outcome; + + for (const std::string wire : + {"{", + "{}", + R"({"type":"unknown","session":1,"time":0,"signals":{}})", + R"({"type":"step","session":-1,"time":0,"signals":{}})", + R"({"type":"step","session":1,"time":0,"signals":{"v":null}})"}) + { + const auto address = endpoint(); + auto server = std::async( + std::launch::async, + [&]() -> auto + { + bool outcome = true; + Channel channel(Disposition::SERVER, address, 5000); + + try + { + channel.receive(); + } + catch (const json::exception& _) + { + } + catch (const std::runtime_error& _) + { + } + catch (...) + { + throw; + } + + Message failure(MessageType::FAIL, 0); + failure.error = "malformed request"; + + channel.finish(failure); + + return outcome; + }); + + zmq::context_t context; + zmq::socket_t socket(context, zmq::socket_type::req); + socket.set(zmq::sockopt::linger, 0); + socket.set(zmq::sockopt::rcvtimeo, 5000); + socket.connect(address); + + outcome *= + socket.send(zmq::buffer(wire), zmq::send_flags::none).has_value(); + + zmq::message_t reply; + + auto ok = socket.recv(reply, zmq::recv_flags::none).has_value(); + if (!ok) + { + outcome *= false; + } + else + { + outcome *= + reply.to_string().find("malformed request") != std::string::npos; + + const std::string close = + R"({"type":"close","session":0,"time":0,"signals":{}})"; + + outcome *= socket.send(zmq::buffer(close), zmq::send_flags::none) + .has_value(); + + socket.set(zmq::sockopt::rcvtimeo, 100); + + static_cast(socket.recv(reply, zmq::recv_flags::none)); + } + + outcome *= server.get(); + } + + return outcome.report(__func__); + } + + TestOutcome boundedWaits() + { + TestStatus outcome; + + Channel client(Disposition::CLIENT, endpoint(), 100); + + outcome *= throws( + [&] + { client.exchange(Message()); }); + + Message invalid; + invalid.signals = {{"bad", std::numeric_limits::infinity()}}; + + Channel other(Disposition::CLIENT, endpoint(), 100); + + outcome *= throws( + [&] + { other.send(invalid); }); + + outcome *= throws( + [&] + { Channel invalid_channel(Disposition::CLIENT, endpoint(), 0); }); + + return outcome.report(__func__); + } + }; + } // namespace Testing +} // namespace GridKit diff --git a/tests/UnitTests/Network/runCoSimTests.cpp b/tests/UnitTests/Network/runCoSimTests.cpp new file mode 100644 index 0000000000..944d197f3f --- /dev/null +++ b/tests/UnitTests/Network/runCoSimTests.cpp @@ -0,0 +1,16 @@ +#include "CoSimTests.hpp" + +int main() +{ + using namespace GridKit::Testing; + + TestingResults result; + CoSimTests test; + + result += test.sessions(); + result += test.rejectedReplies(); + result += test.malformedMessages(); + result += test.boundedWaits(); + + return result.summary(); +}