🎮 Implemented node exec cmd client. (#7)

Author: serges147

include/ocvsmd/platform/bsd/kqueue_single_threaded_executor.hpp

@@ -40,7 +40,7 @@ namespace platform
 namespace bsd
 {
 
-/// @brief Defines BSD Linux platform specific single-threaded executor based on `kqueue` mechanism.
+/// @brief Defines BSD Linux platform-specific single-threaded executor based on `kqueue` mechanism.
 ///
 class KqueueSingleThreadedExecutor final : public libcyphal::platform::SingleThreadedExecutor,
                                            public IPosixExecutorExtension

include/ocvsmd/platform/defines.hpp

@@ -31,6 +31,8 @@ using SingleThreadedExecutor = bsd::KqueueSingleThreadedExecutor;
 using SingleThreadedExecutor = Linux::EpollSingleThreadedExecutor;
 #endif
 
+/// Waits for the predicate to be fulfilled by spinning the executor and its awaitable resources.
+///
 template <typename Executor, typename Predicate>
 void waitPollingUntil(Executor& executor, Predicate predicate)
 {
@@ -55,13 +57,14 @@ void waitPollingUntil(Executor& executor, Predicate predicate)
             timeout = std::min(timeout, spin_result.next_exec_time.value() - executor.now());
         }
 
-        if (const auto maybe_poll_failure = executor.pollAwaitableResourcesFor(cetl::make_optional(timeout)))
+        if (const auto poll_failure = executor.pollAwaitableResourcesFor(cetl::make_optional(timeout)))
         {
-            spdlog::warn("Failed to poll awaitable resources.");
+            (void) poll_failure;
+            spdlog::warn("Failed to poll awaitable resources.");  // TODO: Log the error.
         }
     }
 
-    spdlog::debug("Predicate is fulfilled (worst_lateness={}us).",
+    spdlog::trace("Predicate is fulfilled (worst_lateness={}us).",
                   std::chrono::duration_cast<std::chrono::microseconds>(worst_lateness).count());
 }
 

include/ocvsmd/platform/linux/epoll_single_threaded_executor.hpp

@@ -38,7 +38,7 @@ namespace platform
 namespace Linux
 {
 
-/// @brief Defines Linux platform specific single-threaded executor based on `epoll` mechanism.
+/// @brief Defines Linux platform-specific single-threaded executor based on `epoll` mechanism.
 ///
 class EpollSingleThreadedExecutor final : public libcyphal::platform::SingleThreadedExecutor,
                                           public IPosixExecutorExtension

include/ocvsmd/platform/posix_utils.hpp

@@ -13,6 +13,8 @@ namespace ocvsmd
 namespace platform
 {
 
+/// Wraps a POSIX syscall and retries it if it was interrupted by a signal.
+///
 template <typename Call>
 int posixSyscallError(const Call& call)
 {

include/ocvsmd/sdk/daemon.hpp

@@ -6,7 +6,7 @@
 #ifndef OCVSMD_SDK_DAEMON_HPP_INCLUDED
 #define OCVSMD_SDK_DAEMON_HPP_INCLUDED
 
-// ➕ #include "node_command_client.hpp"
+#include "node_command_client.hpp"
 
 #include <cetl/cetl.hpp>
 #include <cetl/pf17/cetlpf.hpp>
@@ -25,20 +25,39 @@ namespace sdk
 class Daemon
 {
 public:
+    /// Defines the shared pointer type for the factory.
+    ///
     using Ptr = std::shared_ptr<Daemon>;
 
+    /// Creates a new instance of the factory, and establishes a connection to the daemon.
+    ///
+    /// @param memory The memory resource to use for the factory and its subcomponents.
+    ///               The memory resource must outlive the factory.
+    ///               In use for IPC (de)serialization only; other functionality uses usual c++ heap.
+    /// @param executor The executor to use for the factory and its subcomponents.
+    ///                 Instance of the executor must outlive the factory.
+    ///                 Should support `IPosixExecutorExtension` interface (via `cetl::rtti`).
+    /// @return Shared pointer to the successfully created factory.
+    ///         `nullptr` on failure (see logs for the reason of failure).
+    ///
     CETL_NODISCARD static Ptr make(cetl::pmr::memory_resource& memory,
                                    libcyphal::IExecutor&       executor,
                                    const std::string&          connection);
 
+    // No copy/move semantics.
     Daemon(Daemon&&)                 = delete;
     Daemon(const Daemon&)            = delete;
     Daemon& operator=(Daemon&&)      = delete;
     Daemon& operator=(const Daemon&) = delete;
 
     virtual ~Daemon() = default;
 
-    // ➕ virtual NodeCommandClient::Ptr getNodeCommandClient() = 0;
+    /// Gets a pointer to the shared entity which represents the Node Exec Command component of the OCVSMD engine.
+    ///
+    /// @return Shared pointer to the client side of the Node Exec Command component.
+    ///         The component is always present in the OCVSMD engine, so the result is never `nullptr`.
+    ///
+    virtual NodeCommandClient::Ptr getNodeCommandClient() const = 0;
 
 protected:
     Daemon() = default;

include/ocvsmd/sdk/execution.hpp

@@ -24,9 +24,11 @@ namespace sdk
 namespace detail
 {
 
+// Defines various internal types for `sync_wait` implementation.
+//
 template <typename Result>
 class StateOf;
-
+//
 template <typename Result>
 class ReceiverOf final
 {
@@ -46,7 +48,7 @@ class ReceiverOf final
     std::shared_ptr<StateOf<Result>> state_;
 
 };  // ReceiverOf
-
+//
 template <typename Result>
 class StateOf final : public std::enable_shared_from_this<StateOf<Result>>
 {
@@ -78,13 +80,23 @@ class StateOf final : public std::enable_shared_from_this<StateOf<Result>>
 
 /// Abstract interface of a result sender.
 ///
+/// There is no (at least for now) support of async failures via C++ exceptions in the SDK.
+/// So, if an async sender might fail, then the `Result` subtype
+/// is expected to cover both success and failure cases (f.e. using `variant`).
+///
 template <typename Result_>
 class SenderOf
 {
 public:
-    using Ptr    = std::unique_ptr<SenderOf>;
+    /// Defines the shared pointer type for the interface.
+    ///
+    using Ptr = std::unique_ptr<SenderOf>;
+
+    /// Defines the result type of the sender.
+    ///
     using Result = Result_;
 
+    // No copy/move semantics.
     SenderOf(SenderOf&&)                 = delete;
     SenderOf(const SenderOf&)            = delete;
     SenderOf& operator=(SenderOf&&)      = delete;
@@ -94,7 +106,9 @@ class SenderOf
 
     /// Initiates an operation execution by submitting a given receiver to this sender.
     ///
-    /// The submit "consumes" the receiver (no longer usable after this call).
+    /// @tparam Receiver The type of the receiver functor. Should be callable with the result of the sender.
+    /// @param receiver The receiver of the sender's result.
+    ///                 Method "consumes" the receiver (no longer usable after this call).
     ///
     template <typename Receiver>
     void submit(Receiver&& receiver)
@@ -108,13 +122,15 @@ class SenderOf
 protected:
     SenderOf() = default;
 
+    /// Implementation extension point for the derived classes.
+    ///
     virtual void submitImpl(std::function<void(Result&&)>&& receiver) = 0;
 
 };  // SenderOf
 
 /// Initiates an operation execution by submitting a given receiver to the sender.
 ///
-/// The submit "consumes" the receiver (no longer usable after this call).
+/// Submit "consumes" the receiver (no longer usable after this call).
 ///
 template <typename Sender, typename Receiver>
 void submit(Sender& sender, Receiver&& receiver)

include/ocvsmd/sdk/node_command_client.hpp

@@ -0,0 +1,106 @@
+//
+// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+// SPDX-License-Identifier: MIT
+//
+
+#ifndef OCVSMD_SDK_NODE_COMMAND_CLIENT_HPP_INCLUDED
+#define OCVSMD_SDK_NODE_COMMAND_CLIENT_HPP_INCLUDED
+
+#include "execution.hpp"
+
+#include <uavcan/node/ExecuteCommand_1_3.hpp>
+
+#include <cetl/pf17/cetlpf.hpp>
+#include <cetl/pf20/cetlpf.hpp>
+
+#include <chrono>
+#include <cstdint>
+#include <memory>
+#include <unordered_map>
+
+namespace ocvsmd
+{
+namespace sdk
+{
+
+/// Defines client side interface of the OCVSMD Node Exec Command component.
+///
+class NodeCommandClient
+{
+public:
+    /// Defines the shared pointer type for the interface.
+    ///
+    using Ptr = std::shared_ptr<NodeCommandClient>;
+
+    NodeCommandClient(NodeCommandClient&&)                 = delete;
+    NodeCommandClient(const NodeCommandClient&)            = delete;
+    NodeCommandClient& operator=(NodeCommandClient&&)      = delete;
+    NodeCommandClient& operator=(const NodeCommandClient&) = delete;
+
+    virtual ~NodeCommandClient() = default;
+
+    /// Defines the result type of the command execution.
+    ///
+    /// On success, the result is a map of node IDs to their responses (`status` and `output` params).
+    /// Missing Cyphal nodes (or failed to respond in a given timeout) are not included in the map.
+    ///
+    struct Command final
+    {
+        using NodeRequest  = uavcan::node::ExecuteCommand_1_3::Request;
+        using NodeResponse = uavcan::node::ExecuteCommand_1_3::Response;
+
+        using Success = std::unordered_map<std::uint16_t, NodeResponse>;
+        using Failure = int;  // `errno`-like error code.
+        using Result  = cetl::variant<Success, Failure>;
+
+    };  // Command
+
+    /// Sends a Cyphal command to the specified Cyphal network nodes.
+    ///
+    /// On the OCVSMD engine side, the `node_request` is sent concurrently to all specified Cyphal nodes.
+    /// Responses are sent back to the client side as they arrive.
+    /// Result will be available when the last response has arrived, or the timeout has expired.
+    ///
+    /// @param node_ids The list of Cyphal node IDs to send the command to. Duplicates are ignored.
+    /// @param node_request The Cyphal command request to send (aka broadcast) to the `node_ids`.
+    /// @param timeout The maximum time to wait for all Cyphal node responses to arrive.
+    /// @return An execution sender which emits the async overall result of the operation.
+    ///
+    virtual SenderOf<Command::Result>::Ptr sendCommand(const cetl::span<const std::uint16_t> node_ids,
+                                                       const Command::NodeRequest&           node_request,
+                                                       const std::chrono::microseconds       timeout) = 0;
+
+    /// A convenience method for invoking `sendCommand` with COMMAND_RESTART.
+    ///
+    /// @param node_ids The list of Cyphal node IDs to send the command to. Duplicates are ignored.
+    /// @param timeout The maximum time to wait for all Cyphal node responses to arrive. Default is 1 second.
+    /// @return An execution sender which emits the async result of the operation.
+    ///
+    SenderOf<Command::Result>::Ptr restart(  //
+        const cetl::span<const std::uint16_t> node_ids,
+        const std::chrono::microseconds       timeout = std::chrono::seconds{1});
+
+    /// A convenience method for invoking `sendCommand` with COMMAND_BEGIN_SOFTWARE_UPDATE.
+    ///
+    /// @param node_ids The list of Cyphal node IDs to send the command to. Duplicates are ignored.
+    /// @param file_path The path to the software update file. Limited to 255 characters.
+    ///                  Relative to one of the roots configured in the file server.
+    /// @param timeout The maximum time to wait for all Cyphal node responses to arrive. Default is 1 second.
+    /// @return An execution sender which emits the async result of the operation.
+    ///
+    SenderOf<Command::Result>::Ptr beginSoftwareUpdate(  //
+        const cetl::span<const std::uint16_t> node_ids,
+        const cetl::string_view               file_path,
+        const std::chrono::microseconds       timeout = std::chrono::seconds{1});
+
+protected:
+    NodeCommandClient() = default;
+
+    virtual cetl::pmr::memory_resource& getMemoryResource() const noexcept = 0;
+
+};  // NodeCommandClient
+
+}  // namespace sdk
+}  // namespace ocvsmd
+
+#endif  // OCVSMD_SDK_NODE_COMMAND_CLIENT_HPP_INCLUDED

src/cli/main.cpp

@@ -8,7 +8,7 @@
 #include <ocvsmd/platform/defines.hpp>
 #include <ocvsmd/sdk/daemon.hpp>
 #include <ocvsmd/sdk/execution.hpp>
-//➕ #include <ocvsmd/sdk/node_command_client.hpp>
+#include <ocvsmd/sdk/node_command_client.hpp>
 
 #include <cetl/pf17/cetlpf.hpp>
 
@@ -82,18 +82,19 @@ int main(const int argc, const char** const argv)
             std::cerr << "Failed to create daemon.";
             return EXIT_FAILURE;
         }
-/*➕
-        // Demo of daemon's node command client, sending a command to node 42, 43 & 44.
+
+#if 1  // NOLINT
+
+        // Demo of daemon's node command client - sending a command to node 42, 43 & 44.
         {
             using Command = ocvsmd::sdk::NodeCommandClient::Command;
 
             auto node_cmd_client = daemon->getNodeCommandClient();
 
-            const std::vector<std::uint16_t> node_ids = {42};
+            const std::vector<std::uint16_t> node_ids = {42, 43, 44};
             // auto sender     = node_cmd_client->restart({node_ids.data(), node_ids.size()});
             auto sender     = node_cmd_client->beginSoftwareUpdate({node_ids.data(), node_ids.size()}, "firmware.bin");
             auto cmd_result = ocvsmd::sdk::sync_wait<Command::Result>(executor, std::move(sender));
-
             if (const auto* const err = cetl::get_if<Command::Failure>(&cmd_result))
             {
                 spdlog::error("Failed to send command: {}", std::strerror(*err));
@@ -109,7 +110,8 @@ int main(const int argc, const char** const argv)
                 }
             }
         }
-➕*/
+#endif
+
         if (g_running == 0)
         {
             spdlog::debug("Received termination signal.");

src/common/ipc/channel.hpp

@@ -59,6 +59,20 @@ class AnyChannel
 
 };  // AnyChannel
 
+/// Defines an abstract bidirectional communication channel.
+///
+/// Supports sending of messages (plural!) of type `Output`,
+/// and receiving messages of type `Input` via `EventHandler` callback.
+///
+/// Channel could be opened permanently (f.e. for receiving some kind of notifications, like status updates),
+/// but usually it represents one RPC session, which could be completed (finished) with an optional error code.
+/// Such completion normally done at server side (when RPC request has been fulfilled),
+/// but client could also complete the channel. Any unexpected IPC communication error (like f.e. sudden death of
+/// either client or server process) also leads to channel completion (with `ipc::ErrorCode::Disconnected` error).
+///
+/// Channel could be moved, but not copied.
+/// Channel lifetime is managed by its owner - an IPC service client or server.
+///
 template <typename Input_, typename Output_>
 class Channel final : public AnyChannel
 {
@@ -69,10 +83,12 @@ class Channel final : public AnyChannel
     using EventVar     = cetl::variant<Connected, Input, Completed>;
     using EventHandler = std::function<void(const EventVar&)>;
 
+    // Move-only.
     ~Channel()                                   = default;
     Channel(Channel&& other) noexcept            = default;
     Channel& operator=(Channel&& other) noexcept = default;
 
+    // No copy.
     Channel(const Channel&)            = delete;
     Channel& operator=(const Channel&) = delete;
 
@@ -89,7 +105,7 @@ class Channel final : public AnyChannel
             });
     }
 
-    void complete(const int error_code)
+    void complete(const int error_code = 0)
     {
         return gateway_->complete(error_code);
     }